fix(core): reject unparseable trajectory sentinels in context, selection, and resume

harness2/core/context.py

@@ -216,13 +216,28 @@ def copy_source(adapter: BenchmarkAdapter, task_id: str, root: Path) -> int:
     return src_n
 
 
+_MISSING_TRAJECTORY_PREFIXES = (
+    "(no trajectory",
+    "(no parseable events in trajectory)",
+    "(failed to read trajectory:",
+    "(trajectory unavailable:",
+)
+
+
+def is_missing_trajectory(text: str) -> bool:
+    """True when `text` is empty or one of the substrate/adapter missing-trajectory sentinels."""
+    stripped = (text or "").strip()
+    return not stripped or stripped.startswith(_MISSING_TRAJECTORY_PREFIXES)
+
+
 def _write_rollout(dest: Path, r: RunView, adapter, trunc: int, names_only: bool) -> str:
     """Write one rollout's trajectory and deliverables; return "" or the failure detail."""
     failure = ""
     try:
-        (dest / "rollouts" / f"{r.label}.md").write_text(
-            adapter.trajectory(r.run_dir, obs_trunc=trunc), encoding="utf-8"
-        )
+        traj = adapter.trajectory(r.run_dir, obs_trunc=trunc)
+        (dest / "rollouts" / f"{r.label}.md").write_text(traj, encoding="utf-8")
+        if is_missing_trajectory(traj):
+            failure = f"{r.label}: {traj.strip() or 'empty trajectory'}"
     except Exception as exc:
         (dest / "rollouts" / f"{r.label}.md").write_text(
             f"(trajectory unavailable: {type(exc).__name__}: {exc})", encoding="utf-8"

harness2/core/schedule.py

@@ -32,8 +32,8 @@
 from pathlib import Path
 from typing import Mapping, Sequence
 
+from . import context, recursion, runmark, selection
 from . import library as library_store
-from . import recursion, runmark, selection
 from . import tree as harness_tree
 from .proposer import Proposer
 from .spec import BenchmarkAdapter, RunView
@@ -399,7 +399,7 @@ def _resume(
                 artifact = adapter.trajectory(run_dir, obs_trunc=adapter.spec.obs_trunc)
             else:
                 artifact = adapter.deliverable_text(run_dir) if adapter.delivered(run_dir) else ""
-            if not artifact.strip() or artifact.startswith("(no trajectory"):
+            if context.is_missing_trajectory(artifact):
                 raise ValueError("adopted output is missing")
             artifacts_intact = getattr(adapter, "artifacts_intact", None)
             if callable(artifacts_intact) and not artifacts_intact(run_dir):

harness2/core/selection.py

@@ -140,7 +140,7 @@ def _render_pool(
             except Exception as exc:
                 sel.dropped.append(f"{lbl} (trajectory failed: {type(exc).__name__}: {exc})"[:200])
                 continue
-            if not body.strip() or body.startswith("(no trajectory"):
+            if context.is_missing_trajectory(body):
                 continue
             pool[lbl] = rv
             artifacts[lbl] = (

tests/test_workflow.py

@@ -17,15 +17,17 @@
 import json
 import tempfile
 import unittest
+from dataclasses import replace
 from pathlib import Path
 from unittest.mock import Mock, patch
 
 from harness2 import config
-from harness2.core import context, judge, recursion, schedule, tree
+from harness2.core import context, judge, recursion, schedule, selection, tree
 from harness2.core.library import Library
 from harness2.core.proposer import Proposer, ProposerError, Reply
-from harness2.core.spec import resolve_channel_model
+from harness2.core.spec import RunView, resolve_channel_model
 from harness2.demo import BASE_SCRIPT, DemoAdapter, DemoProposer
+from harness2.substrate import codex, opencode
 
 
 class WorkflowTests(unittest.TestCase):
@@ -224,6 +226,55 @@ def call(self, mission, context_dir, label, **kwargs):
         self.adapter.score.assert_not_called()
         self.assertEqual((rows[0].graded, rows[0].n_passed), (False, None))
 
+    def test_missing_trajectory_sentinels_rejected_in_context_selection_and_resume(self):
+        self.adapter.spec = replace(self.adapter.spec, state_deliverable=True)
+        run_dir = self.adapter.execute("ledger-1", None, "sentinel-check")
+        (run_dir / "answer.txt").write_text("", encoding="utf-8")
+
+        garbled = self.root / "garbled.jsonl"
+        garbled.write_text("not json\n", encoding="utf-8")
+        sentinels = (
+            opencode.render_events(garbled, obs_trunc=10_000, max_chars=10_000),
+            codex.render_events(self.root / "missing.jsonl", obs_trunc=10_000, max_chars=10_000),
+        )
+        for sentinel in sentinels:
+            with self.subTest(sentinel=sentinel):
+                (run_dir / "trajectory.md").write_text(sentinel, encoding="utf-8")
+                rv = RunView(
+                    label="agg1",
+                    run_id="sentinel-check",
+                    run_dir=run_dir,
+                    ok=True,
+                    delivered=False,
+                )
+                with self.assertRaises(RuntimeError):
+                    context.build(
+                        self.root / "ctx_sentinel",
+                        self.adapter,
+                        "ledger-1",
+                        [rv],
+                    )
+                sel = selection.trivial_selection(
+                    "ledger-1",
+                    1,
+                    "agg",
+                    {"agg1": rv},
+                    adapter=self.adapter,
+                    work_dir=self.root / "work_sentinel",
+                )
+                self.assertFalse(sel.ok)
+                self.assertEqual(sel.n_candidates, 0)
+
+        self.run_stream(k=1)
+        summary_path = self.adapter.spec.runs_root / "demo-oc-agg1/ledger-1/summary.json"
+        summary = json.loads(summary_path.read_text())
+        chosen_run_dir = Path(summary["selections"]["agg"]["chosen_run_dir"])
+        summary["selections"]["agg"]["evidence"] = "trajectory"
+        summary_path.write_text(json.dumps(summary, indent=2), encoding="utf-8")
+        (chosen_run_dir / "trajectory.md").write_text(sentinels[0], encoding="utf-8")
+        repaired = self.run_stream(k=1)
+        self.assertEqual([row["status"] for row in repaired["tasks"]], ["ok", "skipped"])
+
 
 if __name__ == "__main__":
     unittest.main()