Support length-only trace windows

src/aituner/spec.py

@@ -151,10 +151,12 @@ class TraceSpec:
     timestamp_field: str
     max_concurrency: int
     max_requests_per_probe: int | None = None
+    synthetic_prompt_cap_tokens: int | None = None
 
     @classmethod
     def from_dict(cls, data: Mapping[str, Any]) -> "TraceSpec":
         max_requests = data.get("max_requests_per_probe")
+        synthetic_prompt_cap = data.get("synthetic_prompt_cap_tokens")
         return cls(
             windows_path=_require_str(data.get("windows_path"), context="trace.windows_path"),
             window_id=_require_str(data.get("window_id"), context="trace.window_id"),
@@ -167,6 +169,9 @@ class TraceSpec:
                 data.get("max_concurrency", 64), context="trace.max_concurrency"
             ),
             max_requests_per_probe=int(max_requests) if max_requests is not None else None,
+            synthetic_prompt_cap_tokens=(
+                int(synthetic_prompt_cap) if synthetic_prompt_cap is not None else None
+            ),
         )
 
 

src/aituner/trace.py

@@ -81,6 +81,14 @@ def _coerce_messages(row: Mapping[str, Any]) -> list[dict[str, Any]]:
     raise TraceError("trace row is missing chat messages/prompt text")
 
 
+def _synthetic_prompt_from_tokens(token_count: int) -> str:
+    if token_count <= 0:
+        return "hello"
+    # Keep it ASCII and structurally simple so the same trace can be replayed
+    # on any OpenAI-compatible engine without extra tokenizer assets.
+    return " ".join(["token"] * token_count)
+
+
 def _coerce_completion_tokens(row: Mapping[str, Any]) -> int | None:
     for key in ("max_completion_tokens", "max_tokens", "output_length", "completion_tokens"):
         value = row.get(key)
@@ -123,9 +131,24 @@ def load_trace_requests(study: StudySpec, *, study_spec_path: Path) -> tuple[Win
             sampling_u = row.get(study.trace.u_field, 1.0)
             if isinstance(sampling_u, bool) or not isinstance(sampling_u, (int, float)):
                 raise TraceError(f"trace row {idx} is missing numeric {study.trace.u_field}")
+            prompt_tokens_hint = _coerce_prompt_tokens(row)
+            try:
+                messages = _coerce_messages(row)
+            except TraceError:
+                capped_prompt_tokens = prompt_tokens_hint or 0
+                if study.trace.synthetic_prompt_cap_tokens is not None:
+                    capped_prompt_tokens = min(
+                        capped_prompt_tokens, study.trace.synthetic_prompt_cap_tokens
+                    )
+                messages = [
+                    {
+                        "role": "user",
+                        "content": _synthetic_prompt_from_tokens(capped_prompt_tokens),
+                    }
+                ]
             body: dict[str, Any] = {
                 "model": study.model.served_model_name,
-                "messages": _coerce_messages(row),
+                "messages": messages,
                 "stream": True,
                 "stream_options": {"include_usage": True},
             }
@@ -141,7 +164,7 @@ def load_trace_requests(study: StudySpec, *, study_spec_path: Path) -> tuple[Win
                     arrival_s=float(timestamp),
                     sampling_u=float(sampling_u),
                     body=body,
-                    prompt_tokens_hint=_coerce_prompt_tokens(row),
+                    prompt_tokens_hint=prompt_tokens_hint,
                     completion_tokens_hint=completion_tokens,
                 )
             )