fix(eval): correct trajectory token/cost accounting + open with the user step

eval/harbor/backfill_trajectories.py

@@ -15,8 +15,9 @@
 
 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.
+unless ``--force`` is passed. Trials whose log is empty/crashed before any event get a one-step
+"attempted" trajectory (the recovered task instruction) when the
+instruction is available, else none; either way they are reported.
 
 Note: reconstructed trajectories use the stream-json FALLBACK path, so
 per-step assistant narration is absent (the pre-``session.save()`` print
@@ -53,6 +54,32 @@ def _model_name(trial_dir: Path) -> tuple[str | None, str | None]:
     return model_name, info.get("version")
 
 
+def _task_instruction(trial_dir: Path) -> str | None:
+    """Recover the task instruction from the Harbor task cache, so a
+    reconstructed trajectory can open with the user step like a live run.
+
+    The trial's ``config.json`` records ``task.name`` + ``task.ref``
+    (``sha256:<hash>``); Harbor caches the compiled task under
+    ``~/.cache/harbor/tasks/packages/<name>/<hash>/instruction.md``.
+    Returns ``None`` when the cache entry isn't present.
+    """
+    try:
+        task = json.loads((trial_dir / "config.json").read_text()).get("task") or {}
+    except (OSError, ValueError):
+        return None
+    name = task.get("name")
+    ref = task.get("ref") or ""
+    if not name or not ref.startswith("sha256:"):
+        return None
+    cache = Path.home() / ".cache" / "harbor" / "tasks" / "packages"
+    instr = cache / name / ref[len("sha256:") :] / "instruction.md"
+    try:
+        text = instr.read_text(encoding="utf-8").strip()
+    except OSError:
+        return None
+    return text or None
+
+
 def backfill_job(job_dir: Path, *, force: bool) -> tuple[int, int, int]:
     """Returns (written, skipped_existing, no_events)."""
     written = skipped = empty = 0
@@ -67,6 +94,11 @@ def backfill_job(job_dir: Path, *, force: bool) -> tuple[int, int, int]:
             continue
         model_name, version = _model_name(trial_dir)
         agent = Clawcodex(logs_dir=agent_dir, model_name=model_name, version=version)
+        # Give the reconstruction the task instruction (opening user step),
+        # recovered from the Harbor task cache; harmless when unavailable.
+        instruction = _task_instruction(trial_dir)
+        if instruction:
+            agent._captured_instruction = instruction
         agent.populate_context_post_run(AgentContext())
         if out.exists():
             written += 1

eval/harbor/clawcodex_agent.py

@@ -445,6 +445,11 @@ async def _seed_container_settings(self, environment: BaseEnvironment) -> None:
     async def run(
         self, instruction: str, environment: BaseEnvironment, context: AgentContext
     ) -> None:
+        # Captured so populate_context_post_run can open a stream-json-only
+        # (reconstructed) trajectory with the task instruction as step 1,
+        # like the built-in claude-code agent does. ``@with_prompt_template``
+        # has already rendered any prompt template into ``instruction``.
+        self._captured_instruction = instruction
         if self._subscription:
             await self._inject_subscription_credentials(environment)
         await self._seed_container_settings(environment)
@@ -521,8 +526,23 @@ def populate_context_post_run(self, context: AgentContext) -> None:
         events = self._parse_stream_events()
         result_event = self._last_result_event(events)
 
-        # Token metrics — authoritative, from the stream-json result event.
-        if result_event:
+        try:
+            trajectory, totals = 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}")
+            trajectory, totals = None, None
+
+        # Leaderboard token/cost columns. Prefer the session's authoritative
+        # BILLING totals (input + cache, summed per turn — matches
+        # claude-code); fall back to the stream-json usage (incomplete: no
+        # cumulative cache) only when no session cost block was synced.
+        if totals is not None:
+            context.n_input_tokens = totals["prompt"]
+            context.n_cache_tokens = totals["cached"]
+            context.n_output_tokens = totals["completion"]
+            if totals["cost"] is not None:
+                context.cost_usd = totals["cost"]
+        elif result_event:
             usage = result_event.get("usage")
             if isinstance(usage, dict):
                 input_tokens = usage.get("input_tokens") or 0
@@ -532,15 +552,6 @@ def populate_context_post_run(self, context: AgentContext) -> None:
                 context.n_cache_tokens = cache_read
                 context.n_output_tokens = usage.get("output_tokens") or 0
 
-        try:
-            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
 
@@ -582,34 +593,41 @@ def _final_metrics(
         self,
         result_event: dict[str, Any] | None,
         total_steps: int,
-        cost_usd: float | None,
+        totals: dict[str, Any] | None,
     ) -> FinalMetrics | None:
-        if not result_event and cost_usd is None:
-            return None
+        """``totals`` = authoritative session billing totals (preferred).
+        Falls back to the stream-json usage (incomplete — no cumulative
+        cache) only when no session cost block was synced."""
         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
+        if totals is not None:
+            prompt = totals["prompt"]
+            completion = totals["completion"]
+            cached = totals["cached"]
+            cost = totals["cost"]
+        elif result_event:
+            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
+            prompt = input_tokens + cache_read + cache_creation
+            completion = usage.get("output_tokens") or 0
+            cached = cache_read
+            cost = result_event.get("total_cost_usd")
+        else:
+            return None
         extra: dict[str, Any] = {}
         num_turns = (result_event or {}).get("num_turns")
         if isinstance(num_turns, int):
             extra["num_turns"] = num_turns
         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=(
-                cost_usd
-                if cost_usd is not None
-                else (result_event or {}).get("total_cost_usd")
-            ),
+            total_prompt_tokens=prompt,
+            total_completion_tokens=completion,
+            total_cached_tokens=cached,
+            total_cost_usd=cost,
             total_steps=total_steps,
             extra=extra or None,
         )
@@ -638,18 +656,62 @@ def _agent_meta(self, events: list[dict[str, Any]]) -> tuple[Agent, str | None]:
         )
         return agent, session_id
 
-    def _load_session_data(self) -> tuple[list[dict[str, Any]] | None, float | None]:
+    @staticmethod
+    def _billing_totals_from_cost_block(
+        cost: dict[str, Any],
+    ) -> dict[str, Any] | None:
+        """Authoritative BILLING token totals from a session ``cost`` block.
+
+        clawcodex's stream-json ``result.usage`` is built for live-CONTEXT
+        measurement, not accounting: ``input_tokens`` is the running sum of
+        NON-cached input and it drops cumulative cache tokens (only a
+        ``last_*`` snapshot survives). For a heavily prompt-cached opus run
+        that under-reports total prompt tokens by orders of magnitude
+        (observed: 6 vs the real ~412 K). The session cost block's
+        ``model_usage`` accumulates the real per-turn billing counters
+        (input + cache_read + cache_creation, summed across turns), which is
+        exactly the convention the built-in claude-code agent reports.
+        Returns ``{prompt, completion, cached, cost}`` or ``None``.
+        """
+        model_usage = cost.get("model_usage")
+        if not isinstance(model_usage, dict) or not model_usage:
+            return None
+        prompt = completion = cached = 0
+        for usage in model_usage.values():
+            if not isinstance(usage, dict):
+                continue
+            inp = usage.get("input_tokens") or 0
+            cread = usage.get("cache_read_input_tokens") or 0
+            ccreate = usage.get("cache_creation_input_tokens") or 0
+            prompt += inp + cread + ccreate
+            cached += cread
+            completion += usage.get("output_tokens") or 0
+        total_cost = cost.get("total_cost_usd")
+        return {
+            "prompt": prompt,
+            "completion": completion,
+            "cached": cached,
+            "cost": float(total_cost) if isinstance(total_cost, (int, float)) else None,
+        }
+
+    def _load_session_data(
+        self,
+    ) -> tuple[list[dict[str, Any]] | None, dict[str, Any] | 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."""
+        ``(conversation.messages, billing_totals)`` where billing_totals is
+        ``{prompt, completion, cached, cost}`` from the cost block (or
+        ``None``). Messages and billing are selected INDEPENDENTLY (each by
+        most messages) so a conversation-bearing session without a cost
+        block still contributes its narration."""
         # ``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
+        best_msgs_len = -1
+        best_totals: dict[str, Any] | None = None
+        best_totals_len = -1
         for path in candidates:
             if path.name == "trajectory.json":
                 continue
@@ -661,44 +723,64 @@ def _load_session_data(self) -> tuple[list[dict[str, Any]] | None, float | None]
                 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_len = len(msgs)
+            n = len(msgs) if isinstance(msgs, list) else 0
+            if isinstance(msgs, list) and n > best_msgs_len:
+                best_msgs_len = n
                 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
+            # Billing from the richest COST-bearing session (== the run's
+            # real session, not a seed) — independent of message selection.
+            cost = data.get("cost")
+            if isinstance(cost, dict) and n > best_totals_len:
+                best_totals_len = n
+                best_totals = self._billing_totals_from_cost_block(cost)
+        return best_msgs, best_totals
 
     def _build_trajectory(
         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."""
+    ) -> tuple[Trajectory | None, dict[str, Any] | None]:
+        """Returns ``(trajectory, billing_totals)`` — the totals are surfaced
+        so the caller can also set the leaderboard token/cost columns."""
         agent, session_id = self._agent_meta(events)
-        messages, cost_usd = self._load_session_data()
+        messages, totals = 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.
+            # narration, so agent steps carry only their tool calls. Prepend
+            # the task instruction as the opening user step (claude-code's
+            # trajectory opens with it) so the trace isn't headless.
+            instruction_step = self._instruction_step()
+            if instruction_step is not None:
+                for s in steps:
+                    s.step_id += 1
+                steps.insert(0, instruction_step)
             notes = (
                 "Reconstructed from the stream-json log; per-step assistant "
                 "narration is unavailable (no persisted session conversation)."
             )
         if not steps:
-            return None, cost_usd
+            return None, totals
         trajectory = Trajectory(
             schema_version="ATIF-v1.7",
             session_id=session_id or "unknown",
             agent=agent,
             steps=steps,
             notes=notes,
-            final_metrics=self._final_metrics(result_event, len(steps), cost_usd),
+            final_metrics=self._final_metrics(result_event, len(steps), totals),
         )
-        return trajectory, cost_usd
+        return trajectory, totals
+
+    def _instruction_step(self) -> Step | None:
+        """The opening user step carrying the task instruction, if ``run()``
+        captured it. ``step_id`` is fixed up by the caller."""
+        instruction = getattr(self, "_captured_instruction", None)
+        if not isinstance(instruction, str) or not instruction.strip():
+            return None
+        return Step(step_id=1, source="user", message=instruction)
 
     @staticmethod
     def _split_content(content: Any) -> tuple[str, str | None, list[dict[str, Any]], list[dict[str, Any]]]: