Capture LLM trajectories across OAuth and API-key runs

docs/agent-quickstart.md

@@ -138,8 +138,9 @@ and what it is:
     agent/                           — agent-side logs
     trajectory/acp_trajectory.jsonl  — the full agent trace (every ACP event:
                                        prompts, tool calls, outputs)
-    trajectory/llm_trajectory.jsonl  — raw provider requests/responses captured
-                                       by the usage-tracking proxy
+    trajectory/llm_trajectory.jsonl  — always-present LLM exchange log
+    trajectory/llm_trajectory.manifest.json
+                                     — source/fidelity/completeness for that log
     trainer/verifiers.jsonl          — trainer-ready scored trajectory record
     trainer/atif.json                — the trajectory in ATIF interchange
                                        format (omitted if the trajectory is empty)

docs/getting-started.md

@@ -202,7 +202,8 @@ Each run writes under `--jobs-dir` (default `jobs/`):
       prompts.json                  # prompts sent to the agent
       trajectory/
         acp_trajectory.jsonl        # full agent trace (ACP events)
-        llm_trajectory.jsonl        # raw provider requests/responses (when the usage-tracking proxy captured exchanges)
+        llm_trajectory.jsonl        # always present; LLM exchanges at the fidelity described below
+        llm_trajectory.manifest.json # capture source, fidelity, completeness, and errors
       trainer/
         verifiers.jsonl             # trainer-ready scored trajectory (Verifiers/ORS record)
         atif.json                   # ATIF trajectory record (omitted if the trajectory is empty)
@@ -213,6 +214,29 @@ Each run writes under `--jobs-dir` (default `jobs/`):
         test-stdout.txt             # verifier stdout
 ```
 
+`llm_trajectory.jsonl` is created when the rollout directory is initialized,
+including for setup failures and tasks that make no model call. Its sidecar is
+the source of truth for interpreting the JSONL:
+
+| Agent/auth path | Primary source | `capture_fidelity` |
+|---|---|---|
+| API key through the BenchFlow gateway (including Azure) | LiteLLM provider request/response capture | `provider_wire` |
+| Claude Code subscription/OAuth | Claude Code raw API-body files correlated by local OTLP logs | `provider_wire` |
+| Claude Code subscription/OAuth fallback | Claude Code native session JSONL | `agent_session` |
+| Codex subscription/OAuth | Codex native session JSONL | `agent_session` |
+
+The manifest status is `complete`, `partial`, `no_model_call`, or
+`capture_failed`. Mixed-role rollouts merge API-key and native-subscription
+exchanges into the same JSONL; `role_captures` records each prepared role's
+scene role, agent, model, auth mode, source, fidelity, completeness, and exchange
+count.
+Missing or ambiguously attributed roles make the rollout-level capture
+`partial`. Reconstructed `agent_session` rows remain useful for audit and viewer
+workflows, but trainer exports fail closed unless the manifest says the capture
+is complete provider-wire data. Claude's own raw-body telemetry can still
+contain provider-redacted extended-thinking blocks; BenchFlow also applies its
+normal secret redaction before publishing the JSONL.
+
 ### Reading results
 
 Exit code 0 means the pipeline completed — it is not a pass/fail signal. A

docs/running-any-benchmark.md

@@ -217,7 +217,8 @@ Every layer terminates at the *same* output contract, written per rollout under
 | `results.jsonl` | Verifiers/Prime-RL shaped rollout row |
 | `rewards.jsonl` | The reward record for the rollout (ORS / OpenReward shape) |
 | `trajectory/acp_trajectory.jsonl` | Full agent trace as ACP events |
-| `trajectory/llm_trajectory.jsonl` | Raw provider requests/responses (when captured) |
+| `trajectory/llm_trajectory.jsonl` | Always-present LLM exchange log (possibly empty for `no_model_call`) |
+| `trajectory/llm_trajectory.manifest.json` | Capture source, fidelity, completeness, and errors |
 | `trainer/verifiers.jsonl` | Trainer-ready scored trajectory (Verifiers record) |
 | `trainer/atif.json` | ATIF trajectory record |
 | `trainer/adp.jsonl` | ADP trajectory record |

src/benchflow/cli/main.py

@@ -286,9 +286,11 @@ def eval_run(
             "--usage-tracking",
             help=(
                 "Telemetry-enforcement policy: auto, required, or off. The "
-                "LiteLLM proxy is always used for routable agents (usage, cost, "
-                "and llm_trajectory.jsonl are always captured); this flag only "
-                "controls whether trusted telemetry is required."
+                "LiteLLM proxy is always used for routable API-key agents; "
+                "native subscription agents use their own telemetry/session "
+                "surface. llm_trajectory.jsonl is always emitted, and its "
+                "manifest records capture fidelity. This flag only controls "
+                "whether trusted usage telemetry is required."
             ),
         ),
     ] = None,

src/benchflow/eval_artifacts.py

@@ -15,6 +15,10 @@
     PrimeSftTrajectoryJsonlError,
     load_llm_trajectory_jsonl,
 )
+from benchflow.trajectories.llm_capture_manifest import (
+    capture_manifest_allows_training,
+    read_llm_trajectory_manifest,
+)
 
 CanonicalizePolicy = Literal["none", "one-healthy-per-task"]
 RetryPolicy = Literal["default", "unscored-only"]
@@ -202,6 +206,18 @@ def _llm_trajectory_status(rollout_dir: Path) -> tuple[bool, bool, int]:
         rows = load_llm_trajectory_jsonl(path, strict=True)
     except PrimeSftTrajectoryJsonlError:
         return True, False, 0
+    manifest = read_llm_trajectory_manifest(rollout_dir)
+    if manifest is not None:
+        try:
+            expected_rows = int(manifest.get("exchange_count") or 0)
+        except (TypeError, ValueError):
+            return True, False, len(rows)
+        if (
+            not capture_manifest_allows_training(manifest)
+            or not rows
+            or expected_rows != len(rows)
+        ):
+            return True, False, len(rows)
     return True, True, len(rows)
 
 

src/benchflow/rollout/__init__.py

@@ -219,6 +219,10 @@
     make_trajectory_sink,
 )
 from benchflow.trajectories._llm_capture import LiveLLMTrajectoryWriter
+from benchflow.trajectories.llm_capture import (
+    LLMTrajectoryCapture,
+    model_call_seen_from_evidence,
+)
 from benchflow.trajectories.tree import RolloutNode, RolloutTree, Step
 from benchflow.usage_tracking import (
     USAGE_SOURCE_AGENT_NATIVE_ACP,
@@ -645,6 +649,7 @@ def __init__(self, config: RolloutConfig) -> None:
         self._started_at: datetime | None = None
         self._job_name: str | None = None
         self._rollout_name: str | None = None
+        self._llm_capture: LLMTrajectoryCapture | None = None
         self._agent_env: dict[str, str] = {}
         self._resolved_prompts: list[str] = []
         self._agent_launch: str = ""
@@ -938,6 +943,17 @@ async def setup(self) -> None:
             self._rollout_name,
         ) = _init_rollout(cfg.task_path, cfg.job_name, cfg.rollout_name, cfg.jobs_dir)
 
+        # Create the artifact contract as soon as the rollout directory exists.
+        # Even setup/start failures therefore leave a valid (possibly empty)
+        # JSONL plus a sidecar explaining why no exchange was captured.
+        self._llm_capture = LLMTrajectoryCapture(
+            self._rollout_dir,
+            agent=cfg.primary_agent,
+            model=cfg.primary_model,
+            session_id=self._rollout_name or "",
+            started_at=self._started_at,
+        )
+
         # C-axis overlay: deep-merge cfg.config_override into the task's resolved
         # config here at the rollout layer (not in the Task constructor), so only
         # this run's tasks are patched and every downstream read sees it. No-op
@@ -958,6 +974,7 @@ async def setup(self) -> None:
             ),
             disallow=self._disallow_web_tools,
         )
+        self._llm_capture.configure(self._agent_env)
         env_config = getattr(getattr(self._task, "config", None), "sandbox", None)
         task_skill_policy = resolve_task_skill_policy(
             task_path=cfg.task_path,
@@ -1225,6 +1242,16 @@ async def install_agent(self) -> None:
             await self._planes.upload_subscription_auth(
                 self._env, agent_name, cred_home
             )
+        llm_capture = getattr(self, "_llm_capture", None)
+        if llm_capture is not None:
+            self._agent_env = await llm_capture.prepare_agent(
+                self._env,
+                agent=agent_name,
+                model=cfg.primary_model,
+                agent_env=self._agent_env,
+                credential_home=cred_home,
+                sandbox_user=cfg.sandbox_user,
+            )
         await self._planes.apply_web_tool_policy(
             self._env,
             agent_name,
@@ -2010,6 +2037,27 @@ async def cleanup(self) -> None:
                 self._usage_runtime = None
 
         self._finalize_usage_metrics()
+        llm_capture = getattr(self, "_llm_capture", None)
+        if llm_capture is not None:
+            acp_events = list(getattr(self, "_trajectory", None) or [])
+            model_call_seen = model_call_seen_from_evidence(
+                getattr(self, "_usage_metrics", None), acp_events
+            )
+            try:
+                await llm_capture.finalize(
+                    self._env,
+                    acp_events=acp_events,
+                    model_call_seen=model_call_seen,
+                )
+            except Exception as e:
+                logger.warning(f"LLM trajectory finalization failed: {e}")
+                try:
+                    llm_capture.record_failure(e, model_call_seen=model_call_seen)
+                except Exception as record_error:
+                    logger.warning(
+                        "LLM trajectory failure manifest write failed: %s",
+                        record_error,
+                    )
         self._enforce_required_usage_tracking()
 
         if self._environment is not None:
@@ -2341,6 +2389,19 @@ async def connect_as(self, role: Role) -> None:
                 disallow=disallow_web_tools,
             )
 
+        llm_capture = getattr(self, "_llm_capture", None)
+        if llm_capture is not None:
+            cred_home = f"/home/{cfg.sandbox_user}" if cfg.sandbox_user else "/root"
+            agent_env = await llm_capture.prepare_agent(
+                self._env,
+                agent=role.agent,
+                model=role.model,
+                agent_env=agent_env,
+                credential_home=cred_home,
+                sandbox_user=cfg.sandbox_user,
+                role_name=role.name,
+            )
+
         self._agent_launch = agent_launch
 
         sf_entrypoint = self._session_factory_entrypoint(role.agent)

src/benchflow/trajectories/__init__.py

@@ -1,7 +1,8 @@
 """Trajectory capture and exchange schemas.
 
-BenchFlow persists ACP-native trajectories plus LiteLLM callback-derived LLM
-request/response exchanges.
+BenchFlow persists ACP-native trajectories plus provider-wire or explicitly
+lower-fidelity native-agent LLM request/response exchanges. The adjacent
+``llm_trajectory.manifest.json`` states the capture source and completeness.
 
 Files
 -----

src/benchflow/trajectories/export_prime_sft.py

@@ -18,6 +18,10 @@
 from typing import Any, Literal, cast
 
 from benchflow._utils.json_safe import dumps_finite, scrub_non_finite
+from benchflow.trajectories.llm_capture_manifest import (
+    capture_manifest_allows_training,
+    read_llm_trajectory_manifest,
+)
 from benchflow.trajectories.types import redact_trajectory_obj
 
 PrimeSftRowMode = Literal["rollout", "exchange"]
@@ -62,6 +66,7 @@ class PrimeSftExportStats:
     skipped_exchanges_provider_error: int = 0
     skipped_no_assistant: int = 0
     skipped_missing_tool_defs: int = 0
+    skipped_insufficient_capture_fidelity: int = 0
     skipped_terminal_error: int = 0
     skipped_invalid: int = 0
     tool_call_ids_rewritten: int = 0
@@ -81,6 +86,9 @@ def as_dict(self) -> dict[str, Any]:
             "skipped_exchanges_provider_error": self.skipped_exchanges_provider_error,
             "skipped_no_assistant": self.skipped_no_assistant,
             "skipped_missing_tool_defs": self.skipped_missing_tool_defs,
+            "skipped_insufficient_capture_fidelity": (
+                self.skipped_insufficient_capture_fidelity
+            ),
             "skipped_terminal_error": self.skipped_terminal_error,
             "skipped_invalid": self.skipped_invalid,
             "tool_call_ids_rewritten": self.tool_call_ids_rewritten,
@@ -1117,6 +1125,15 @@ def normalize_prime_sft_exchange(
     redact: bool = True,
 ) -> tuple[PrimeSftExchangeData | None, str | None]:
     """Normalize one raw LLM exchange through the Prime-SFT validator path."""
+    metadata = exchange.get("metadata")
+    if isinstance(metadata, dict):
+        fidelity = metadata.get("capture_fidelity")
+        if fidelity is not None and fidelity != "provider_wire":
+            return None, "insufficient_capture_fidelity"
+        if metadata.get("request_complete") is False:
+            return None, "insufficient_capture_fidelity"
+        if metadata.get("response_complete") is False:
+            return None, "insufficient_capture_fidelity"
     messages, tool_defs, skip_reason = _exchange_to_messages_and_tools(
         exchange, redact=redact
     )
@@ -1202,6 +1219,13 @@ def convert_benchflow_rollouts_to_prime_sft_rows(
             stats.skipped_reward += 1
             continue
 
+        capture_manifest = read_llm_trajectory_manifest(rollout_dir)
+        if capture_manifest is not None and not capture_manifest_allows_training(
+            capture_manifest
+        ):
+            stats.skipped_insufficient_capture_fidelity += 1
+            continue
+
         trajectory_path = rollout_dir / "trajectory" / "llm_trajectory.jsonl"
         exchanges = load_llm_trajectory_jsonl(trajectory_path, strict=True)
         if not exchanges:
@@ -1234,6 +1258,9 @@ def convert_benchflow_rollouts_to_prime_sft_rows(
             if skip_reason == "missing_tool_defs":
                 stats.skipped_missing_tool_defs += 1
                 continue
+            if skip_reason == "insufficient_capture_fidelity":
+                stats.skipped_insufficient_capture_fidelity += 1
+                continue
             if row is None:
                 stats.skipped_invalid += 1
                 continue

src/benchflow/trajectories/export_trl_sft.py

@@ -24,6 +24,10 @@
     normalize_prime_sft_exchange,
     validate_prime_sft_row,
 )
+from benchflow.trajectories.llm_capture_manifest import (
+    capture_manifest_allows_training,
+    read_llm_trajectory_manifest,
+)
 from benchflow.trajectories.types import redact_trajectory_obj
 
 TrlSftRowMode = Literal["rollout", "exchange"]
@@ -42,6 +46,7 @@ class TrlSftExportStats:
     skipped_provider_error: int = 0
     skipped_no_assistant: int = 0
     skipped_missing_tools: int = 0
+    skipped_insufficient_capture_fidelity: int = 0
     skipped_terminal_error: int = 0
     skipped_helper_calls: int = 0
     skipped_invalid: int = 0
@@ -64,6 +69,9 @@ def as_dict(self) -> dict[str, Any]:
             "skipped_provider_error": self.skipped_provider_error,
             "skipped_no_assistant": self.skipped_no_assistant,
             "skipped_missing_tools": self.skipped_missing_tools,
+            "skipped_insufficient_capture_fidelity": (
+                self.skipped_insufficient_capture_fidelity
+            ),
             "skipped_terminal_error": self.skipped_terminal_error,
             "skipped_helper_calls": self.skipped_helper_calls,
             "skipped_invalid": self.skipped_invalid,
@@ -314,6 +322,12 @@ def convert_benchflow_rollouts_to_trl_sft_rows(
         if min_reward is not None and (reward is None or reward < min_reward):
             stats.skipped_reward += 1
             continue
+        capture_manifest = read_llm_trajectory_manifest(rollout_dir)
+        if capture_manifest is not None and not capture_manifest_allows_training(
+            capture_manifest
+        ):
+            stats.skipped_insufficient_capture_fidelity += 1
+            continue
         trajectory_path = rollout_dir / "trajectory" / "llm_trajectory.jsonl"
         exchanges = load_llm_trajectory_jsonl(trajectory_path, strict=True)
         if not exchanges:
@@ -351,6 +365,9 @@ def convert_benchflow_rollouts_to_trl_sft_rows(
             if skip_reason == "missing_tool_defs":
                 stats.skipped_missing_tools += 1
                 continue
+            if skip_reason == "insufficient_capture_fidelity":
+                stats.skipped_insufficient_capture_fidelity += 1
+                continue
             if row is None:
                 stats.skipped_invalid += 1
                 continue