feat: run-integrity leak detection over finished transcripts

.env.example

@@ -80,6 +80,16 @@ LOG_TO_FILE=false  # Set to true to enable file logging
 # Off by default. Do NOT enable on developer workstations.
 # CODER_EVAL_REMEDIATE_HOME_PLUGINS=0
 
+# Run-integrity gate. Every row gets a verdict recording whether its score measured
+# the agent or a leak: the agent reading graded material (reference solution, checker
+# script, its own task YAML).
+#   detect (default) -- record the verdict + findings on the row, change no outcome
+#   void             -- also downgrade a tainted SUCCESS to FAILURE
+#   off              -- skip the pass entirely
+# Read a real run's `detect` findings before switching to `void`. See
+# docs/REPORT_SCHEMA.md "Run integrity".
+# INTEGRITY_MODE=detect
+
 # Usage Telemetry (anonymous; on by default).
 # Disable entirely:
 # TELEMETRY_ENABLED=false

docs/REPORT_SCHEMA.md

@@ -79,10 +79,43 @@ including: `task_id`, `replicate_index`, `variant_id`, `status`
 `expected_commands`,
 `actual_commands`, `commands_efficiency`, `agent_config`, `sdk_options`,
 `installed_tools`, turn accounting (`total_turns`, `visible_turns`, `expected_turns`,
-`max_turns_exhausted`, `has_final_reply`), and early-stop fields (`stopped_early`,
-`early_stop_reason`, `turns_remaining_at_stop`). `iterations` here is a **reduced**
-turn digest (`{iteration, duration_seconds, command_count, assistant_turn_count,
-crashed, crash_reason}`) — the full transcript is in `task.json`.
+`max_turns_exhausted`, `has_final_reply`), early-stop fields (`stopped_early`,
+`early_stop_reason`, `turns_remaining_at_stop`), and run-integrity fields
+(`integrity_verdict`, `integrity_voided`, `integrity_findings`). `iterations` here is
+a **reduced** turn digest (`{iteration, duration_seconds, command_count,
+assistant_turn_count, crashed, crash_reason}`) — the full transcript is in `task.json`.
+
+The row key set is pinned by `tests/test_reports_experiment.py::TestRowKeySet`.
+`eval_result_to_task_dict` is a hand-maintained dict literal, so a new
+`EvaluationResult` field reaches `task.json` for free but **not** `run.json` unless it
+is added there too.
+
+### Run integrity
+
+Every row carries an integrity verdict: was the score a measurement of the agent, or an
+artifact of the agent reaching graded material — a reference solution, a checker script,
+its own task definition?
+
+| `integrity_verdict` | Meaning |
+| --- | --- |
+| `clean` | The scan saw the whole transcript and found nothing. |
+| `tainted` | A command read graded material. |
+| `inconclusive` | The scan could not see the whole transcript — unrecovered sub-agent tool calls, too many commands with no parameters, or an unrecognised tool touching graded material. Never voids. |
+| `skipped` | The pass did not run (`INTEGRITY_MODE=off`, or no graded material was derivable for the task). |
+
+`INTEGRITY_MODE` decides what a `tainted` verdict does: `detect` (the default) records
+the verdict and findings and changes no outcome; `void` additionally downgrades a
+tainted `SUCCESS` to `FAILURE` and sets `integrity_voided`; `off` skips the pass. Roll
+out with `detect` and read a real run's findings before switching to `void`.
+
+`weighted_score` is left as computed on a voided row: the high score IS the finding (the
+row passed *because* it cheated), so erasing it would destroy the evidence. Taint is
+never a `FinalStatus` member — the terminal-status set is closed, and integrity rides
+alongside the status as orthogonal telemetry, the way early-stop does.
+
+`task.json` carries the full `integrity` object: every finding with its coordinates
+(iteration, command index, tool, evidence excerpt) plus the blind-spot counters
+(`commands_scanned`, `commands_without_parameters`, `subagent_recovery_incomplete`).
 
 ### Missing cost is never fatal
 

src/coder_eval/agents/codex_agent.py

@@ -297,6 +297,10 @@ def __init__(
         self.spawned_children: list[tuple[str, str, str | None]] = []
         # child thread id -> returned message (fallback when the rollout is absent).
         self.collab_results: dict[str, str] = {}
+        # Sub-agents whose inner tool calls never made it into the transcript
+        # (rollout missing, or recovery raised). Surfaced on the TurnRecord so
+        # consumers know the transcript is incomplete rather than empty.
+        self.unrecovered_subagent_threads = 0
 
         # Assistant-transcript reconstruction buffers (one AssistantMessage per gen).
         self.open_blocks: list[ContentBlock] = []
@@ -624,6 +628,7 @@ def finalize(self, status: AgentEndStatus, *, crashed: bool = False, crash_reaso
                 crashed=crashed,
                 crash_reason=crash_reason,
                 duration_seconds=time.monotonic() - self.turn_start_time,
+                unrecovered_subagent_threads=self.unrecovered_subagent_threads,
             )
         )
 
@@ -1462,7 +1467,7 @@ async def _run_turn_with_streaming(
         # Skipped on a cooperative stop: children may have no rollout yet and the
         # run is already decided — recovery adds nothing the armed gate uses.
         if state.spawned_children and not state.stopped_early_hit:
-            await self._recover_subagent_tool_calls(
+            state.unrecovered_subagent_threads = await self._recover_subagent_tool_calls(
                 state.spawned_children,
                 state.collab_results,
                 state.messages,
@@ -1555,7 +1560,11 @@ def _tool_parameters(self, root: Any, root_type: str | None) -> dict[str, Any]:
         if root_type == "fileChange":
             changes = getattr(root, "changes", None) or []
             path = str(changes[0].path) if changes and hasattr(changes[0], "path") else "?"
-            return {"path": path}
+            params: dict[str, Any] = {"path": path}
+            kind = _status_value(getattr(changes[0], "kind", None)) if changes else ""
+            if kind:
+                params["kind"] = kind
+            return params
         if root_type == "collabAgentToolCall":
             params: dict[str, Any] = {"operation": _status_value(getattr(root, "tool", ""))}
             if model := getattr(root, "model", None):
@@ -1700,7 +1709,7 @@ async def _recover_subagent_tool_calls(
         emit: StreamCallback,
         task_id: str,
         turn_id: str,
-    ) -> None:
+    ) -> int:
         """Recover each spawned sub-agent's INNER tool calls AND token usage.
 
         Codex runs every sub-agent on its own child thread whose events never
@@ -1726,14 +1735,23 @@ async def _recover_subagent_tool_calls(
 
         Best-effort: any failure (missing file, parse error) is swallowed so a
         recovery hiccup never fails the turn.
+
+        Returns:
+            Number of sub-agents whose inner tool calls could NOT be recovered.
+            Those calls are absent from ``commands`` and ``messages`` altogether,
+            so a caller analysing the transcript is looking at less than the agent
+            actually did. Swallowing the failure keeps the turn alive; returning
+            the count keeps it from being mistaken for "the sub-agent ran nothing".
         """
         home = self._codex_home()
+        unrecovered = 0
         for thread_id, parent_tool_id, model in spawned_children:
             try:
                 path = await self._await_rollout_file(home, thread_id)
                 if path is None:
                     # No rollout to mine: nest just the returned message (if any) so
                     # the sub-agent's answer still shows, tokenless.
+                    unrecovered += 1
                     self._log.debug("CodexAgent: no rollout found for sub-agent thread %s", thread_id)
                     result = collab_results.get(thread_id)
                     if result:
@@ -1763,9 +1781,20 @@ async def _recover_subagent_tool_calls(
                         )
                     messages.append(self._subagent_generation_message(blocks, gen, parent_tool_id, model, turn_id, gi))
             except Exception as exc:
-                # Best-effort: a recovery hiccup must never fail the turn.
+                # Best-effort: a recovery hiccup must never fail the turn -- but it
+                # leaves this child's commands out of the transcript, so count it.
+                unrecovered += 1
                 self._log.debug("CodexAgent: sub-agent recovery failed for %s: %s", thread_id, exc)
 
+        if unrecovered:
+            self._log.warning(
+                "CodexAgent: %d of %d sub-agent(s) contributed no recovered tool calls; "
+                + "this turn's transcript is incomplete",
+                unrecovered,
+                len(spawned_children),
+            )
+        return unrecovered
+
     @staticmethod
     def _codex_home() -> Path:
         """Codex data directory (rollouts live under ``<home>/sessions``)."""
@@ -2099,17 +2128,28 @@ def _extract_file_change_telemetry(
         apply_patch is recorded as an ``error`` (the old ``status != "error"``
         test never matched the real PatchApplyStatus values, so failed patches
         were scored as successful writes).
+
+        The per-change kinds ride along in ``parameters["kinds"]`` (positionally
+        aligned with ``paths``) when the SDK provides them: unlike Claude's Write,
+        an apply_patch can UPDATE an existing file -- a read of its current
+        content -- and the integrity scan needs the kind to tell an update from
+        an add. Omitted entirely when no change carries a kind.
         """
         try:
-            paths = [str(c.path) for c in changes if hasattr(c, "path")] if changes else []
+            with_paths = [c for c in changes if hasattr(c, "path")] if changes else []
+            paths = [str(c.path) for c in with_paths]
+            kinds = [_status_value(getattr(c, "kind", None)) for c in with_paths]
+            parameters: dict[str, Any] = {"paths": paths}
+            if any(kinds):
+                parameters["kinds"] = kinds
             status_str = _status_value(status)
             failed = status_str in _FILE_CHANGE_FAILURE_STATUSES
             return CommandTelemetry(
                 tool_name="Write",
                 tool_id=change_id,
                 timestamp=datetime.now(),
                 duration_ms=None,
-                parameters={"paths": paths},
+                parameters=parameters,
                 result_status="error" if failed else "success",
                 result_summary=(
                     f"{len(paths)} file(s) changed"

src/coder_eval/cli/run_task_internal_command.py

@@ -159,9 +159,11 @@ def _watch_host_heartbeat() -> None:
     workspace_dir_raw = context.get("workspace_dir")
     workspace_dir = Path(workspace_dir_raw) if workspace_dir_raw else None
     config_lineage = {k: ConfigLineageEntry.model_validate(v) for k, v in (context.get("config_lineage") or {}).items()}
-    # Prefer the host's raw source_yaml so task.json's audit trail matches
-    # the in-process driver. Fall back to the staged (post-override) YAML
-    # for older host versions that didn't forward it.
+    # The host no longer stages its raw `source_yaml` (a second verbatim copy of the
+    # task's success_criteria inside the sandbox, read by nothing here). Record the
+    # staged post-override YAML instead; the host restores its own raw text onto
+    # task.json.task_config.source_yaml when it parses the result back, so the audit
+    # trail is unchanged. An older host that still sends the key is honored.
     host_source_yaml: str | None = context.get("source_yaml")
 
     # Load the post-override spec from the staged YAML. We then point

src/coder_eval/config.py

@@ -13,7 +13,7 @@
 from pydantic import AliasChoices, Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-from coder_eval.models import AgentKind, ApiBackend
+from coder_eval.models import AgentKind, ApiBackend, IntegrityMode
 
 
 # Application Insights connection string baked into the application so a fresh
@@ -137,6 +137,17 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
     gemini_api_key: str | None = None
     antigravity_model: str | None = None
 
+    # Run-integrity gate (see models.enums.IntegrityMode). Kill switch for the
+    # graded-material read pass in coder_eval.integrity. DETECT
+    # (default) records a verdict + findings on every row and changes no score;
+    # VOID additionally downgrades a tainted SUCCESS to FAILURE; OFF skips the
+    # pass. Flip to VOID once a nightly's DETECT findings have been reviewed.
+    # INTEGRITY_MODE is also in the docker env allowlist (models/sandbox.py) —
+    # without that the in-container Settings silently defaults and the gate
+    # differs between the tempdir and docker drivers (same bug class as the
+    # API_BACKEND note in isolation/docker_runner.py).
+    integrity_mode: IntegrityMode = IntegrityMode.DETECT
+
     # Logging
     log_level: str = "INFO"  # Default log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
     log_to_file: bool = False  # Whether to enable file logging