From c02e14966c9bb211f9cf7eefba8bddc31e6fd3e0 Mon Sep 17 00:00:00 2001 From: Eric Lee Date: Tue, 21 Jul 2026 10:52:29 -0700 Subject: [PATCH 1/2] feat(eval): emit ATIF trajectory.json from the clawcodex harbor adapter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Harbor's built-in claude-code agent writes a rich step-by-step trajectory.json (ATIF) that its viewer/leaderboard render; the clawcodex adapter emitted none. Add it: - eval/harbor/clawcodex_agent.py: SUPPORTS_ATIF=True; populate_context_ post_run now builds an ATIF-v1.7 Trajectory (Agent header from the system:init event; Steps with per-turn message/reasoning/tool_calls and observations; FinalMetrics from the stream-json result usage) and writes trajectory.json. Two sources, preferred order: * the persisted session conversation (RICH — per-turn narration, reasoning, tool calls AND results), synced under /sessions/; * the stream-json log (always present) — every tool call+result plus the final answer, and the authoritative token usage. Best-effort throughout: a build failure never fails the trial. Token backfill behavior is unchanged. - src/entrypoints/headless.py: persist the session (session.save()) at the end of a headless run. The print path previously left no session on disk, so `-p` runs were un-resumable and carried no structured transcript; this both enables --resume for headless and gives the adapter its rich trajectory source. Best-effort (never changes exit code/output). Verified: rich path yields user→agent(tool+result)→agent steps with narration; fallback path yields 53 tool steps + metrics from a real stream-json log; a live headless run persists the conversation; 669 headless/session tests pass. Co-Authored-By: Claude Fable 5 --- eval/harbor/clawcodex_agent.py | 345 +++++++++++++++++++++++++++++++-- src/entrypoints/headless.py | 15 ++ 2 files changed, 347 insertions(+), 13 deletions(-) diff --git a/eval/harbor/clawcodex_agent.py b/eval/harbor/clawcodex_agent.py index d0e42330..8636349a 100644 --- a/eval/harbor/clawcodex_agent.py +++ b/eval/harbor/clawcodex_agent.py @@ -76,6 +76,15 @@ from harbor.environments.base import BaseEnvironment from harbor.models.agent.context import AgentContext from harbor.models.trial.paths import EnvironmentPaths +from harbor.models.trajectories import ( + Agent, + FinalMetrics, + Observation, + ObservationResult, + Step, + ToolCall, + Trajectory, +) # Host env vars forwardable into the container (clawcodex's builtin provider # key candidates — see src/providers/__init__.py in the clawcodex repo), keyed @@ -220,6 +229,11 @@ def fresh_subscription_credentials() -> dict[str, Any]: class Clawcodex(BaseInstalledAgent): """Run the clawcodex CLI headless inside a Harbor task environment.""" + # Emit an ATIF trajectory.json in populate_context_post_run (below), so + # Harbor's viewer / leaderboard get a step-by-step trajectory like the + # built-in claude-code agent produces. + SUPPORTS_ATIF: bool = True + CLI_FLAGS = [ CliFlag( "max_turns", @@ -483,15 +497,64 @@ async def run( await self.exec_as_agent(environment, command=command, env=self._build_env()) @override + # ------------------------------------------------------------------ # + # Trajectory (ATIF) + token metrics + # ------------------------------------------------------------------ # + def populate_context_post_run(self, context: AgentContext) -> None: - """Backfill token metrics from the final stream-json ``result`` event.""" - stream_path = self.logs_dir / "clawcodex.txt" + """Backfill token metrics and write an ATIF ``trajectory.json``. + + Runs on the host after Harbor syncs the container ``/logs/agent`` + tree back. Two sources, in preference order: + + * the persisted clawcodex session conversation (rich — per-turn + assistant narration, reasoning, tool calls AND results), synced + under ``/sessions/`` when the CLI is new enough to save the + session in headless mode; and + * the stream-json log (``clawcodex.txt``) — always present; carries + every tool call + result and the final answer, and is the sole + source of authoritative token usage. + + Everything here is best-effort: a trajectory-build failure must + never fail the trial (mirrors the built-in claude-code agent). + """ + events = self._parse_stream_events() + result_event = self._last_result_event(events) + + # Token metrics — authoritative, from the stream-json result event. + if result_event: + usage = result_event.get("usage") + if isinstance(usage, dict): + input_tokens = usage.get("input_tokens") or 0 + cache_read = usage.get("cache_read_input_tokens") or 0 + cache_creation = usage.get("cache_creation_input_tokens") or 0 + context.n_input_tokens = input_tokens + cache_read + cache_creation + context.n_cache_tokens = cache_read + context.n_output_tokens = usage.get("output_tokens") or 0 + try: - content = stream_path.read_text(encoding="utf-8") - except OSError: + trajectory = self._build_trajectory(events, result_event) + except Exception as exc: # noqa: BLE001 — never fail a trial over this + self.logger.debug(f"clawcodex trajectory build failed: {exc}") + return + if trajectory is None: return - result_event: dict[str, Any] | None = None + path = self.logs_dir / "trajectory.json" + try: + with open(path, "w", encoding="utf-8") as handle: + json.dump(trajectory.to_json_dict(), handle, indent=2, ensure_ascii=False) + self.logger.debug(f"wrote clawcodex trajectory to {path}") + except OSError as exc: + self.logger.debug(f"failed to write trajectory {path}: {exc}") + + def _parse_stream_events(self) -> list[dict[str, Any]]: + """All JSON objects from the teed stream-json log, in order.""" + try: + content = (self.logs_dir / "clawcodex.txt").read_text(encoding="utf-8") + except OSError: + return [] + events: list[dict[str, Any]] = [] for line in content.splitlines(): line = line.strip() if not line.startswith("{"): @@ -500,18 +563,274 @@ def populate_context_post_run(self, context: AgentContext) -> None: event = json.loads(line) except json.JSONDecodeError: continue - if isinstance(event, dict) and event.get("type") == "result": - result_event = event + if isinstance(event, dict): + events.append(event) + return events - if not result_event: - return + @staticmethod + def _last_result_event(events: list[dict[str, Any]]) -> dict[str, Any] | None: + for event in reversed(events): + if event.get("type") == "result": + return event + return None + def _final_metrics( + self, result_event: dict[str, Any] | None, total_steps: int + ) -> FinalMetrics | None: + if not result_event: + return None usage = result_event.get("usage") if not isinstance(usage, dict): - return + usage = {} input_tokens = usage.get("input_tokens") or 0 cache_read = usage.get("cache_read_input_tokens") or 0 cache_creation = usage.get("cache_creation_input_tokens") or 0 - context.n_input_tokens = input_tokens + cache_read + cache_creation - context.n_cache_tokens = cache_read - context.n_output_tokens = usage.get("output_tokens") or 0 + extra: dict[str, Any] = {} + num_turns = result_event.get("num_turns") + if isinstance(num_turns, int): + extra["num_turns"] = num_turns + duration_ms = result_event.get("duration_ms") + if isinstance(duration_ms, (int, float)): + extra["duration_ms"] = duration_ms + return FinalMetrics( + total_prompt_tokens=input_tokens + cache_read + cache_creation, + total_completion_tokens=usage.get("output_tokens") or 0, + total_cached_tokens=cache_read, + total_cost_usd=result_event.get("total_cost_usd"), + total_steps=total_steps, + extra=extra or None, + ) + + def _agent_meta(self, events: list[dict[str, Any]]) -> tuple[Agent, str | None]: + """Build the ATIF ``Agent`` header from the system:init event.""" + init = next( + (e for e in events if e.get("type") == "system" and e.get("subtype") == "init"), + {}, + ) + session_id = init.get("session_id") + model_name = init.get("model") or self._parsed_model_name + extra: dict[str, Any] = {} + for key in ("provider", "cwd", "permission_mode"): + if init.get(key): + extra[key] = init[key] + tools = init.get("tools") + agent = Agent( + name=self.name(), + version=self.version() or "unknown", + model_name=model_name, + tool_definitions=( + [{"name": t} for t in tools] if isinstance(tools, list) else None + ), + extra=extra or None, + ) + return agent, session_id + + def _load_session_messages(self) -> list[dict[str, Any]] | None: + """The rich persisted conversation (Anthropic-shape messages), if the + CLI saved a session under the synced config dir. Returns the newest + session's ``conversation.messages`` or ``None``.""" + roots = [self.logs_dir / "sessions", self.logs_dir] + candidates: list[Path] = [] + for root in roots: + if root.is_dir(): + candidates.extend(root.rglob("*.json")) + best: list[dict[str, Any]] | None = None + best_len = 0 + for path in candidates: + if path.name == "trajectory.json": + continue + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + continue + conv = data.get("conversation") if isinstance(data, dict) else None + msgs = conv.get("messages") if isinstance(conv, dict) else None + if isinstance(msgs, list) and len(msgs) > best_len: + best, best_len = msgs, len(msgs) + return best + + def _build_trajectory( + self, events: list[dict[str, Any]], result_event: dict[str, Any] | None + ) -> Trajectory | None: + agent, session_id = self._agent_meta(events) + messages = self._load_session_messages() + if messages: + steps = self._steps_from_conversation(messages, agent.model_name) + else: + steps = self._steps_from_stream(events, agent.model_name) + if not steps: + return None + return Trajectory( + schema_version="ATIF-v1.7", + session_id=session_id or "unknown", + agent=agent, + steps=steps, + final_metrics=self._final_metrics(result_event, len(steps)), + ) + + @staticmethod + def _split_content(content: Any) -> tuple[str, str | None, list[dict[str, Any]], list[dict[str, Any]]]: + """Split an Anthropic-shape message ``content`` into (text, reasoning, + tool_use blocks, tool_result blocks).""" + if isinstance(content, str): + return content.strip(), None, [], [] + text_parts: list[str] = [] + reasoning_parts: list[str] = [] + tool_uses: list[dict[str, Any]] = [] + tool_results: list[dict[str, Any]] = [] + if isinstance(content, list): + for block in content: + if not isinstance(block, dict): + continue + btype = block.get("type") + if btype == "text" and isinstance(block.get("text"), str): + text_parts.append(block["text"]) + elif btype in ("thinking", "reasoning") and isinstance( + block.get("thinking") or block.get("text"), str + ): + reasoning_parts.append(block.get("thinking") or block.get("text")) + elif btype == "tool_use": + tool_uses.append(block) + elif btype == "tool_result": + tool_results.append(block) + text = "\n\n".join(p.strip() for p in text_parts if p and p.strip()) + reasoning = "\n\n".join(p.strip() for p in reasoning_parts if p and p.strip()) + return text, (reasoning or None), tool_uses, tool_results + + @staticmethod + def _stringify(value: Any) -> str: + if isinstance(value, str): + return value + try: + return json.dumps(value, ensure_ascii=False) + except TypeError: + return str(value) + + def _steps_from_conversation( + self, messages: list[dict[str, Any]], default_model: str | None + ) -> list[Step]: + """Rich path: one ATIF step per assistant turn, with its tool calls; + tool_result blocks from the following user message attach back to the + matching call as observations (bundled by ``tool_use_id``).""" + steps: list[Step] = [] + pending: dict[str, ToolCall] = {} + pending_step: dict[str, Step] = {} + for msg in messages: + role = msg.get("role") + if msg.get("isMeta"): + continue + text, reasoning, tool_uses, tool_results = self._split_content( + msg.get("content") + ) + timestamp = msg.get("timestamp") + if role == "assistant": + calls: list[ToolCall] = [] + for tu in tool_uses: + call_id = tu.get("id") or tu.get("tool_use_id") + if not call_id: + continue + args = tu.get("input") + call = ToolCall( + tool_call_id=call_id, + function_name=tu.get("name") or "", + arguments=args if isinstance(args, dict) else {"input": args}, + ) + calls.append(call) + pending[call_id] = call + step = Step( + step_id=len(steps) + 1, + timestamp=timestamp, + source="agent", + model_name=default_model, + message=text, + reasoning_content=reasoning, + tool_calls=calls or None, + llm_call_count=1, + ) + steps.append(step) + for c in calls: + pending_step[c.tool_call_id] = step + elif role == "user": + # tool results attach to the prior call's step; a leading + # plain-text user message is the instruction. + for tr in tool_results: + call_id = tr.get("tool_use_id") + step = pending_step.get(call_id) if call_id else None + if step is None: + continue + content = tr.get("content") + obs = ObservationResult( + source_call_id=call_id, + content=self._stringify(content) if content is not None else None, + extra={"is_error": True} if tr.get("is_error") else None, + ) + if step.observation is None: + step.observation = Observation(results=[obs]) + else: + step.observation.results.append(obs) + if text and not tool_results: + steps.append( + Step( + step_id=len(steps) + 1, + timestamp=timestamp, + source="user", + message=text, + ) + ) + return steps + + def _steps_from_stream( + self, events: list[dict[str, Any]], default_model: str | None + ) -> list[Step]: + """Fallback path: build steps from the flat stream-json events — one + step per tool call (with its result), plus the final answer.""" + steps: list[Step] = [] + step_by_call: dict[str, Step] = {} + for event in events: + etype = event.get("type") + if etype == "tool_use": + call_id = event.get("tool_use_id") + if not call_id: + continue + args = event.get("input") + step = Step( + step_id=len(steps) + 1, + source="agent", + model_name=default_model, + message="", + tool_calls=[ + ToolCall( + tool_call_id=call_id, + function_name=event.get("name") or "", + arguments=args if isinstance(args, dict) else {"input": args}, + ) + ], + llm_call_count=1, + ) + steps.append(step) + step_by_call[call_id] = step + elif etype == "tool_result": + call_id = event.get("tool_use_id") + step = step_by_call.get(call_id) if call_id else None + if step is None: + continue + content = event.get("output") + obs = ObservationResult( + source_call_id=call_id, + content=self._stringify(content) if content is not None else None, + extra={"is_error": True} if event.get("is_error") else None, + ) + step.observation = Observation(results=[obs]) + elif etype == "assistant": + text = event.get("text") + if isinstance(text, str) and text.strip(): + steps.append( + Step( + step_id=len(steps) + 1, + source="agent", + model_name=default_model, + message=text, + llm_call_count=1, + ) + ) + return steps diff --git a/src/entrypoints/headless.py b/src/entrypoints/headless.py index 7b971777..a1f32c49 100644 --- a/src/entrypoints/headless.py +++ b/src/entrypoints/headless.py @@ -648,6 +648,21 @@ def _persist(msg: Any) -> None: finally: restore_sigint() + # Persist the full session (conversation + cost) to disk at the end of a + # headless run. Interactive/TUI runs already persist; the print path did + # not, so `-p` sessions could not be `--resume`d and left no structured + # transcript. Best-effort: a save failure must never change the exit + # code or output. Enables downstream trajectory reconstruction (the + # Harbor adapter builds an ATIF trajectory.json from this). + try: + session.save() + except Exception: # noqa: BLE001 — persistence is best-effort + import logging as _logging + + _logging.getLogger(__name__).debug( + "headless session save failed", exc_info=True + ) + # A -p goal run that ends without achieving the condition (budget pause, # evaluator-timeout park, interrupt) exits non-zero so scripts can tell # "loop finished" from "condition met". Achieved goals keep exit 0. From 338fb43b3430009d34cdd320d1e4f0f741eabb45 Mon Sep 17 00:00:00 2001 From: Eric Lee Date: Tue, 21 Jul 2026 11:07:12 -0700 Subject: [PATCH 2/2] feat(eval): trajectory cost from session block, fallback notes, backfill script (critic follow-ups) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Read total_cost_usd from the persisted session's cost block (the --print result event has no cost field) and set both FinalMetrics.total_cost_usd AND context.cost_usd — leaderboard cost parity with the claude-code agent (critic's substantive finding). - Fallback (stream-json-only) trajectories carry a Trajectory.notes marker explaining per-step narration is unavailable. - Dedup _load_session_data's rglob (was scanning the sessions/ subtree twice via nested roots). - Add eval/harbor/backfill_trajectories.py: reproduce trajectory.json for a PAST job from its stream-json logs, no re-run. Idempotent (skips existing unless --force); reports trials that crashed before any event. Backfilled 86/89 of the 2026-07-19 opus job (3 misses = watchdog-crash trials whose 5-line logs have no events to reconstruct). Critic verdict on the branch: APPROVE. Co-Authored-By: Claude Fable 5 --- eval/harbor/backfill_trajectories.py | 103 +++++++++++++++++++++++++++ eval/harbor/clawcodex_agent.py | 84 +++++++++++++++------- 2 files changed, 161 insertions(+), 26 deletions(-) create mode 100644 eval/harbor/backfill_trajectories.py diff --git a/eval/harbor/backfill_trajectories.py b/eval/harbor/backfill_trajectories.py new file mode 100644 index 00000000..6ea46bbb --- /dev/null +++ b/eval/harbor/backfill_trajectories.py @@ -0,0 +1,103 @@ +"""Backfill ATIF ``trajectory.json`` for a past Harbor job — no re-run. + +The ``clawcodex`` adapter now emits a trajectory per trial +(``clawcodex_agent.Clawcodex.populate_context_post_run``), but jobs run +before that landed have none. Every trial still kept its stream-json log +(``agent/clawcodex.txt``), which carries the full tool-call/result +sequence, the final answer, and token usage — enough to reconstruct the +trajectory offline via the adapter's own fallback converter. + +Usage (from the repo root, in Harbor's venv so ``harbor`` imports resolve):: + + PYTHONPATH=eval/harbor \ + ~/.local/share/uv/tools/harbor/bin/python \ + eval/harbor/backfill_trajectories.py eval/harbor/jobs/ [...] + +Writes ``agent/trajectory.json`` into each trial dir (same location the +built-in claude-code agent uses), skipping trials that already have one +unless ``--force`` is passed. Trials whose log is empty/crashed before any +event produce no trajectory (nothing to reconstruct) and are reported. + +Note: reconstructed trajectories use the stream-json FALLBACK path, so +per-step assistant narration is absent (the pre-``session.save()`` print +path persisted no conversation). Fresh runs get the rich, narrated +trajectory automatically. +""" + +import argparse +import json +import logging +import sys +from pathlib import Path + +# The adapter lives next to this file; import it directly. +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from clawcodex_agent import Clawcodex # noqa: E402 + +from harbor.models.agent.context import AgentContext # noqa: E402 + + +def _model_name(trial_dir: Path) -> tuple[str | None, str | None]: + """(model_name, version) from the trial's result.json agent_info.""" + try: + info = json.loads((trial_dir / "result.json").read_text()).get( + "agent_info" + ) or {} + except (OSError, ValueError): + return None, None + model_info = info.get("model_info") or {} + model = model_info.get("name") + provider = model_info.get("provider") + model_name = f"{provider}/{model}" if provider and model else model + return model_name, info.get("version") + + +def backfill_job(job_dir: Path, *, force: bool) -> tuple[int, int, int]: + """Returns (written, skipped_existing, no_events).""" + written = skipped = empty = 0 + for trial_dir in sorted(p for p in job_dir.iterdir() if p.is_dir()): + agent_dir = trial_dir / "agent" + log = agent_dir / "clawcodex.txt" + if not log.exists(): + continue + out = agent_dir / "trajectory.json" + if out.exists() and not force: + skipped += 1 + continue + model_name, version = _model_name(trial_dir) + agent = Clawcodex(logs_dir=agent_dir, model_name=model_name, version=version) + agent.populate_context_post_run(AgentContext()) + if out.exists(): + written += 1 + else: + empty += 1 + print(f" no events to reconstruct: {trial_dir.name}") + return written, skipped, empty + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("job_dirs", nargs="+", type=Path) + parser.add_argument( + "--force", action="store_true", help="overwrite existing trajectory.json" + ) + args = parser.parse_args() + logging.disable(logging.CRITICAL) # the adapter's debug logs are noise here + + total_w = total_s = total_e = 0 + for job_dir in args.job_dirs: + if not job_dir.is_dir(): + print(f"skip (not a dir): {job_dir}", file=sys.stderr) + continue + w, s, e = backfill_job(job_dir, force=args.force) + print(f"{job_dir}: wrote {w}, skipped-existing {s}, no-events {e}") + total_w += w + total_s += s + total_e += e + print(f"total: wrote {total_w}, skipped {total_s}, no-events {total_e}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/eval/harbor/clawcodex_agent.py b/eval/harbor/clawcodex_agent.py index 8636349a..11edcb4b 100644 --- a/eval/harbor/clawcodex_agent.py +++ b/eval/harbor/clawcodex_agent.py @@ -533,10 +533,14 @@ def populate_context_post_run(self, context: AgentContext) -> None: context.n_output_tokens = usage.get("output_tokens") or 0 try: - trajectory = self._build_trajectory(events, result_event) + trajectory, cost_usd = self._build_trajectory(events, result_event) except Exception as exc: # noqa: BLE001 — never fail a trial over this self.logger.debug(f"clawcodex trajectory build failed: {exc}") return + # Leaderboard cost column (matches the claude-code agent, which sets + # context.cost_usd from its final metrics). + if cost_usd is not None: + context.cost_usd = cost_usd if trajectory is None: return @@ -575,28 +579,37 @@ def _last_result_event(events: list[dict[str, Any]]) -> dict[str, Any] | None: return None def _final_metrics( - self, result_event: dict[str, Any] | None, total_steps: int + self, + result_event: dict[str, Any] | None, + total_steps: int, + cost_usd: float | None, ) -> FinalMetrics | None: - if not result_event: + if not result_event and cost_usd is None: return None - usage = result_event.get("usage") + usage = (result_event or {}).get("usage") if not isinstance(usage, dict): usage = {} input_tokens = usage.get("input_tokens") or 0 cache_read = usage.get("cache_read_input_tokens") or 0 cache_creation = usage.get("cache_creation_input_tokens") or 0 extra: dict[str, Any] = {} - num_turns = result_event.get("num_turns") + num_turns = (result_event or {}).get("num_turns") if isinstance(num_turns, int): extra["num_turns"] = num_turns - duration_ms = result_event.get("duration_ms") + duration_ms = (result_event or {}).get("duration_ms") if isinstance(duration_ms, (int, float)): extra["duration_ms"] = duration_ms + # Cost: the --print result event carries no cost field, but the + # persisted session's cost block does (Session.save()). Prefer that. return FinalMetrics( total_prompt_tokens=input_tokens + cache_read + cache_creation, total_completion_tokens=usage.get("output_tokens") or 0, total_cached_tokens=cache_read, - total_cost_usd=result_event.get("total_cost_usd"), + total_cost_usd=( + cost_usd + if cost_usd is not None + else (result_event or {}).get("total_cost_usd") + ), total_steps=total_steps, extra=extra or None, ) @@ -625,16 +638,17 @@ def _agent_meta(self, events: list[dict[str, Any]]) -> tuple[Agent, str | None]: ) return agent, session_id - def _load_session_messages(self) -> list[dict[str, Any]] | None: - """The rich persisted conversation (Anthropic-shape messages), if the - CLI saved a session under the synced config dir. Returns the newest - session's ``conversation.messages`` or ``None``.""" - roots = [self.logs_dir / "sessions", self.logs_dir] - candidates: list[Path] = [] - for root in roots: - if root.is_dir(): - candidates.extend(root.rglob("*.json")) - best: list[dict[str, Any]] | None = None + def _load_session_data(self) -> tuple[list[dict[str, Any]] | None, float | None]: + """The richest persisted session synced under the logs dir, as + ``(conversation.messages, total_cost_usd)``. Picks the session with + the most messages; either element may be ``None`` when absent.""" + # ``self.logs_dir`` is the parent of ``sessions/`` — one root covers + # the subtree; a second nested root would just re-parse each file. + candidates = ( + sorted(self.logs_dir.rglob("*.json")) if self.logs_dir.is_dir() else [] + ) + best_msgs: list[dict[str, Any]] | None = None + best_cost: float | None = None best_len = 0 for path in candidates: if path.name == "trajectory.json": @@ -643,30 +657,48 @@ def _load_session_messages(self) -> list[dict[str, Any]] | None: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError): continue - conv = data.get("conversation") if isinstance(data, dict) else None + if not isinstance(data, dict): + continue + conv = data.get("conversation") msgs = conv.get("messages") if isinstance(conv, dict) else None if isinstance(msgs, list) and len(msgs) > best_len: - best, best_len = msgs, len(msgs) - return best + best_len = len(msgs) + best_msgs = msgs + cost = (data.get("cost") or {}).get("total_cost_usd") + best_cost = float(cost) if isinstance(cost, (int, float)) else None + return best_msgs, best_cost def _build_trajectory( - self, events: list[dict[str, Any]], result_event: dict[str, Any] | None - ) -> Trajectory | None: + self, + events: list[dict[str, Any]], + result_event: dict[str, Any] | None, + ) -> tuple[Trajectory | None, float | None]: + """Returns ``(trajectory, cost_usd)`` — cost is surfaced so the + caller can also set it on the AgentContext.""" agent, session_id = self._agent_meta(events) - messages = self._load_session_messages() + messages, cost_usd = self._load_session_data() + notes = None if messages: steps = self._steps_from_conversation(messages, agent.model_name) else: steps = self._steps_from_stream(events, agent.model_name) + # No persisted conversation → the stream-json has no per-turn + # narration, so agent steps carry only their tool calls. + notes = ( + "Reconstructed from the stream-json log; per-step assistant " + "narration is unavailable (no persisted session conversation)." + ) if not steps: - return None - return Trajectory( + return None, cost_usd + trajectory = Trajectory( schema_version="ATIF-v1.7", session_id=session_id or "unknown", agent=agent, steps=steps, - final_metrics=self._final_metrics(result_event, len(steps)), + notes=notes, + final_metrics=self._final_metrics(result_event, len(steps), cost_usd), ) + return trajectory, cost_usd @staticmethod def _split_content(content: Any) -> tuple[str, str | None, list[dict[str, Any]], list[dict[str, Any]]]: