diff --git a/services/hermes/scripts/cli_lane_evidence.py b/services/hermes/scripts/cli_lane_evidence.py
index d85abb0e..529c1fd7 100644
--- a/services/hermes/scripts/cli_lane_evidence.py
+++ b/services/hermes/scripts/cli_lane_evidence.py
@@ -56,8 +56,12 @@ def _retire_terminal_entry(
     board_descriptor: int,
     quarantine_descriptor: int,
     authority_name: str | None = None,
+    source_descriptor: int | None = None,
 ) -> str:
     """Retire only the inode previously inspected; preserve any replacement."""
+    pinned = os.fstat(source_descriptor) if source_descriptor is not None else source_stat
+    if pinned.st_dev != source_stat.st_dev or pinned.st_ino != source_stat.st_ino:
+        return "replacement"
     try:
         current = os.stat(
             path.name,
@@ -66,7 +70,7 @@ def _retire_terminal_entry(
         )
     except FileNotFoundError:
         return "missing"
-    if current.st_dev != source_stat.st_dev or current.st_ino != source_stat.st_ino:
+    if current.st_dev != pinned.st_dev or current.st_ino != pinned.st_ino:
         return "replacement"
     # A replacement can itself be staged repeatedly while recovery races a
     # writer. Keep every generation bound to the original canonical pending
@@ -76,7 +80,7 @@ def _retire_terminal_entry(
         (authority_name or path.name).encode("utf-8")
     ).hexdigest()[:16]
     source_digest = hashlib.sha256(
-        f"{source_stat.st_dev:x}\0{source_stat.st_ino:x}".encode()
+        f"{pinned.st_dev:x}\0{pinned.st_ino:x}".encode()
     ).hexdigest()[:16]
     for sequence in range(32):
         staging = f".retire.{path_digest}.{source_digest}.{sequence}"
@@ -97,7 +101,7 @@ def _retire_terminal_entry(
         dir_fd=quarantine_descriptor,
         follow_symlinks=False,
     )
-    if staged.st_dev == source_stat.st_dev and staged.st_ino == source_stat.st_ino:
+    if staged.st_dev == pinned.st_dev and staged.st_ino == pinned.st_ino:
         os.unlink(staging, dir_fd=quarantine_descriptor)
         os.fsync(quarantine_descriptor)
         os.fsync(board_descriptor)
@@ -132,6 +136,7 @@ def _write_json_noreplace(path: Path, value: dict[str, Any]) -> bool:
     temporary = f".{path.name}.{uuid.uuid4().hex}.tmp"
     temporary_stat = None
     descriptor = None
+    cleanup_descriptor = None
     try:
         descriptor = os.open(temporary, create_flags, 0o600, dir_fd=directory)
         temporary_stat = os.fstat(descriptor)
@@ -143,6 +148,7 @@ def _write_json_noreplace(path: Path, value: dict[str, Any]) -> bool:
             written = os.fstat(stream.fileno())
             if stat.S_IMODE(written.st_mode) != 0o600 or written.st_nlink != 1:
                 raise OSError("terminal evidence is not private and singly linked")
+            cleanup_descriptor = os.dup(stream.fileno())
         try:
             _rename_noreplace(
                 temporary,
@@ -164,7 +170,10 @@ def _write_json_noreplace(path: Path, value: dict[str, Any]) -> bool:
                     temporary_stat,
                     board_descriptor=directory,
                     quarantine_descriptor=directory,
+                    source_descriptor=cleanup_descriptor,
                 )
+        if cleanup_descriptor is not None:
+            os.close(cleanup_descriptor)
         os.close(directory)
 
 def _terminal_document_digest(document: dict[str, Any]) -> str:
diff --git a/services/hermes/scripts/cli_lane_execution.py b/services/hermes/scripts/cli_lane_execution.py
index 46a3d7b0..59344a4d 100644
--- a/services/hermes/scripts/cli_lane_execution.py
+++ b/services/hermes/scripts/cli_lane_execution.py
@@ -30,6 +30,12 @@ from cli_lane_recovery import _has_pending_finalization
 from cli_lane_routing import fresh_unavailable_provider, select_route
 
 
+def _goal_rejections(state: dict[str, object]) -> list[object]:
+    """Return only bounded list-shaped prior judge evidence."""
+    history = state.get("goal_rejections", [])
+    return history if isinstance(history, list) else []
+
+
 def execute_claim(board: str, task_id: str) -> None:
     """Execute one already-claimed task and commit its outcome to Kanban."""
     from hermes_cli import kanban_db
@@ -260,16 +266,12 @@ def execute_claim(board: str, task_id: str) -> None:
                         and goal_mode
                     ):
                         heartbeat("local goal-completion judge active")
-                        rejection_history = state.get("goal_rejections", [])
-                        if not isinstance(rejection_history, list):
-                            rejection_history = []
-                        judge_context = context
-                        if goal_turn > 1 or rejection_history:
-                            judge_context += (
-                                "\n\nAuthoritative Hermes goal-controller evidence: "
-                                f"current turn {goal_turn}/{goal_max_turns}; prior rejected "
-                                f"reports: {json.dumps(rejection_history[-5:])}."
-                            )
+                        rejection_history = _goal_rejections(state)
+                        judge_context = context + (
+                            "\n\nAuthoritative Hermes goal-controller evidence: "
+                            f"current turn {goal_turn}/{goal_max_turns}; prior rejected "
+                            f"reports: {json.dumps(rejection_history[-5:])}."
+                        )
                         accepted, judge_reason = cli_lane_goal.judge_goal_completion(
                             judge_context,
                             structured,
@@ -338,9 +340,7 @@ def execute_claim(board: str, task_id: str) -> None:
                     and deadline - time.monotonic() > 30
                 )
                 if can_continue:
-                    rejection_history = state.get("goal_rejections", [])
-                    if not isinstance(rejection_history, list):
-                        rejection_history = []
+                    rejection_history = _goal_rejections(state)
                     state["goal_rejections"] = [
                         *rejection_history[-4:],
                         completion_problem,
diff --git a/services/hermes/scripts/cli_lane_finalization.py b/services/hermes/scripts/cli_lane_finalization.py
index ef39ffc5..b5468296 100644
--- a/services/hermes/scripts/cli_lane_finalization.py
+++ b/services/hermes/scripts/cli_lane_finalization.py
@@ -89,6 +89,7 @@ def _retire_snapshot(
         board_descriptor=snapshot.directory_descriptor,
         quarantine_descriptor=snapshot.directory_descriptor,
         authority_name=authority_name,
+        source_descriptor=snapshot.descriptor,
     )
 
 def _retire_snapshot_after_db(path: Path, snapshot: TerminalSnapshot) -> str:
diff --git a/services/hermes/scripts/cli_lane_quarantine.py b/services/hermes/scripts/cli_lane_quarantine.py
index 8c599184..8de6d3ea 100644
--- a/services/hermes/scripts/cli_lane_quarantine.py
+++ b/services/hermes/scripts/cli_lane_quarantine.py
@@ -180,6 +180,7 @@ def _quarantine_terminal(
             source_stat,
             board_descriptor=board_descriptor,
             quarantine_descriptor=quarantine_descriptor,
+            source_descriptor=source_descriptor,
         )
     except OSError as error:
         # A bad quarantine target must not make an attacker-controlled pending
@@ -196,6 +197,7 @@ def _quarantine_terminal(
                         if quarantine_descriptor is not None
                         else board_descriptor
                     ),
+                    source_descriptor=source_descriptor,
                 )
             except OSError:
                 retirement = "deferred"
diff --git a/services/hermes/scripts/cli_lane_retention.py b/services/hermes/scripts/cli_lane_retention.py
index c88df653..9d1b9245 100644
--- a/services/hermes/scripts/cli_lane_retention.py
+++ b/services/hermes/scripts/cli_lane_retention.py
@@ -50,18 +50,27 @@ def _unlink_artifact_if_same(path: Path, observed: os.stat_result) -> bool:
     directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
     directory_flags |= getattr(os, "O_NOFOLLOW", 0)
     descriptor = None
+    source_descriptor = None
     try:
         descriptor = os.open(path.parent, directory_flags)
+        file_flags = getattr(os, "O_PATH", os.O_RDONLY) | getattr(os, "O_NOFOLLOW", 0)
+        source_descriptor = os.open(path.name, file_flags, dir_fd=descriptor)
+        pinned = os.fstat(source_descriptor)
+        if pinned.st_dev != observed.st_dev or pinned.st_ino != observed.st_ino:
+            return False
         outcome = _retire_terminal_entry(
             path,
             observed,
             board_descriptor=descriptor,
             quarantine_descriptor=descriptor,
+            source_descriptor=source_descriptor,
         )
         return outcome == "retired"
     except OSError:
         return False
     finally:
+        if source_descriptor is not None:
+            os.close(source_descriptor)
         if descriptor is not None:
             os.close(descriptor)
 
diff --git a/testing/tests/test_hermes_cli_capabilities.py b/testing/tests/test_hermes_cli_capabilities.py
index adc6e3ca..ddb67492 100644
--- a/testing/tests/test_hermes_cli_capabilities.py
+++ b/testing/tests/test_hermes_cli_capabilities.py
@@ -93,6 +93,9 @@ def test_capability_detection_requires_explicit_safety_keywords():
     assert lanes.detect_kanban_capabilities(variadic).ready is False
     assert lanes._explicit_keyword(object(), "expected_run_id") is False
 
+    missing_completion = lanes.KanbanCapabilities(False, True, True)
+    assert missing_completion.deferred_features == ("exact-run-completion",)
+
 
 def test_startup_health_moves_from_bounded_deferred_to_ready(
     tmp_path: Path,
@@ -117,9 +120,10 @@ def test_startup_health_moves_from_bounded_deferred_to_ready(
     assert "compatibility deferred" in capsys.readouterr().err
 
     new_db = _db(patched=True)
-    new = lanes.kanban_capabilities(new_db)
-    lanes.initialize_kanban_capabilities(new_db, health_path=health)
+    lanes.kanban_capabilities(new_db)
+    new = lanes.initialize_kanban_capabilities(new_db, health_path=health)
     assert new.ready is True
+    assert lanes.kanban_capabilities(new_db) is new
     assert lanes.runtime_health(health)["state"] == "ready"
 
 
diff --git a/testing/tests/test_hermes_cli_dispatch_runtime.py b/testing/tests/test_hermes_cli_dispatch_runtime.py
index 74684105..8355b8c0 100644
--- a/testing/tests/test_hermes_cli_dispatch_runtime.py
+++ b/testing/tests/test_hermes_cli_dispatch_runtime.py
@@ -13,6 +13,8 @@ from testing.tests.test_hermes_cli_support import lanes
 
 
 def test_board_slug_and_connection_failure_paths(monkeypatch):
+    assert lanes._owns_local_workspace(SimpleNamespace(workspace_path=" /workspace "))
+    assert not lanes._owns_local_workspace(SimpleNamespace(workspace_path=""))
     assert lanes._board_slug(SimpleNamespace(slug="cassandra")) == "cassandra"
     assert lanes._board_slug(SimpleNamespace(id="fallback")) == "fallback"
     failures = []
@@ -100,6 +102,41 @@ def test_claim_ready_handles_zero_limit_empty_boards_and_claim_errors(monkeypatc
     assert lanes.claim_ready(set(), 1) == []
 
 
+def test_dispatch_skips_unhealthy_ineligible_and_unclaimed_entries(monkeypatch):
+    class Connection:
+        def __init__(self, board):
+            self.board = board
+
+        def close(self):
+            return None
+
+    tasks = [
+        SimpleNamespace(id="t_ineligible", status="ready", assignee="cli-auto"),
+        SimpleNamespace(id="t_unclaimed", status="ready", assignee="cli-auto"),
+    ]
+    db = SimpleNamespace(
+        list_boards=lambda **_kwargs: [
+            {"slug": "offline"},
+            {"slug": "cassandra"},
+        ],
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: (_ for _ in ()).throw(OSError("offline"))
+        if board == "offline"
+        else Connection(board),
+        recompute_ready=lambda _conn: None,
+        list_tasks=lambda _conn: tasks,
+        claim_task=lambda *_args, **_kwargs: None,
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db))
+    assert lanes.claim_ready(
+        set(),
+        2,
+        eligible=lambda _board, task: task.id == "t_unclaimed",
+    ) == []
+    monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0)
+    lanes.recover_orphans()
+
+
 class _Future:
     def __init__(self):
         self.polls = 0
diff --git a/testing/tests/test_hermes_cli_evidence_edges.py b/testing/tests/test_hermes_cli_evidence_edges.py
index c28ff43f..6a954864 100644
--- a/testing/tests/test_hermes_cli_evidence_edges.py
+++ b/testing/tests/test_hermes_cli_evidence_edges.py
@@ -44,6 +44,7 @@ def test_retirement_reports_missing_replacement_collision_and_restoration(
     path = tmp_path / "pending"
     path.write_text("old", encoding="utf-8")
     source = path.stat()
+    source_descriptor = os.open(path, os.O_RDONLY)
     directory = os.open(tmp_path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
     try:
         path.unlink()
@@ -52,6 +53,7 @@ def test_retirement_reports_missing_replacement_collision_and_restoration(
             source,
             board_descriptor=directory,
             quarantine_descriptor=directory,
+            source_descriptor=source_descriptor,
         ) == "missing"
 
         path.write_text("new", encoding="utf-8")
@@ -60,8 +62,11 @@ def test_retirement_reports_missing_replacement_collision_and_restoration(
             source,
             board_descriptor=directory,
             quarantine_descriptor=directory,
+            source_descriptor=source_descriptor,
         ) == "replacement"
 
+        os.close(source_descriptor)
+        source_descriptor = os.open(path, os.O_RDONLY)
         replacement = path.stat()
         monkeypatch.setattr(
             lanes,
@@ -73,8 +78,10 @@ def test_retirement_reports_missing_replacement_collision_and_restoration(
             replacement,
             board_descriptor=directory,
             quarantine_descriptor=directory,
+            source_descriptor=source_descriptor,
         ) == "collision"
     finally:
+        os.close(source_descriptor)
         os.close(directory)
 
 
@@ -82,6 +89,7 @@ def test_retirement_preserves_a_postcheck_replacement(tmp_path: Path, monkeypatc
     path = tmp_path / "pending"
     path.write_text("old", encoding="utf-8")
     source = path.stat()
+    source_descriptor = os.open(path, os.O_RDONLY)
     directory = os.open(tmp_path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
     calls = []
 
@@ -102,8 +110,10 @@ def test_retirement_preserves_a_postcheck_replacement(tmp_path: Path, monkeypatc
             source,
             board_descriptor=directory,
             quarantine_descriptor=directory,
+            source_descriptor=source_descriptor,
         ) == "replacement"
     finally:
+        os.close(source_descriptor)
         os.close(directory)
     assert path.read_text(encoding="utf-8") == "replacement"
 
@@ -133,6 +143,31 @@ def test_evidence_writer_bounds_payload_and_cleans_failed_temporary(
     monkeypatch.setattr(lanes, "_retire_terminal_entry", real_retire)
 
 
+def test_evidence_writer_closes_unwrapped_and_rejects_unsafe_temporary_files(
+    tmp_path: Path,
+    monkeypatch,
+):
+    destination = tmp_path / "evidence.json"
+    real_fdopen = lanes.os.fdopen
+    opened = []
+
+    def fail_fdopen(descriptor, *_args, **_kwargs):
+        opened.append(descriptor)
+        raise OSError("fdopen failed")
+
+    monkeypatch.setattr(lanes.os, "fdopen", fail_fdopen)
+    with pytest.raises(OSError, match="fdopen failed"):
+        lanes._write_json_noreplace(destination, {"value": "bounded"})
+    with pytest.raises(OSError):
+        os.fstat(opened[0])
+
+    monkeypatch.setattr(lanes.os, "fdopen", real_fdopen)
+    with monkeypatch.context() as unsafe:
+        unsafe.setattr(lanes.stat, "S_IMODE", lambda _mode: 0o644)
+        with pytest.raises(OSError, match="not private"):
+            lanes._write_json_noreplace(destination, {"value": "bounded"})
+
+
 def test_evidence_paths_and_collision_validation_fail_closed(
     tmp_path: Path,
     monkeypatch,
diff --git a/testing/tests/test_hermes_cli_execution_edges.py b/testing/tests/test_hermes_cli_execution_edges.py
index 07b9035d..889218db 100644
--- a/testing/tests/test_hermes_cli_execution_edges.py
+++ b/testing/tests/test_hermes_cli_execution_edges.py
@@ -16,6 +16,13 @@ class _Connection:
         return None
 
 
+def test_goal_rejection_history_is_list_bounded():
+    execution = __import__("cli_lane_execution")
+    history = ["first"]
+    assert execution._goal_rejections({"goal_rejections": history}) is history
+    assert execution._goal_rejections({"goal_rejections": "invalid"}) == []
+
+
 def test_missing_claim_is_a_noop(tmp_path: Path, monkeypatch):
     db = SimpleNamespace(
         scoped_current_board=lambda _board: nullcontext(),
@@ -64,12 +71,18 @@ def test_transient_callback_storage_errors_do_not_kill_provider(
 
     def provider(*args, **_kwargs):
         assert args[6]("still working") is True
-        return lanes.ProcessResult(1, "failed", None, False)
+        return lanes.ProcessResult(
+            1,
+            "failed",
+            {"status": "incomplete", "blockers": ["synthetic provider failure"]},
+            False,
+        )
 
     monkeypatch.setattr(lanes, "run_provider", provider)
     lanes.execute_claim("cassandra", "t_callbacks")
 
     assert blocks and blocks[0]["kind"] == "capability"
+    assert blocks[0]["reason"] == "synthetic provider failure"
 
 
 def test_restart_handoff_survives_missing_prior_log(tmp_path: Path, monkeypatch):
diff --git a/testing/tests/test_hermes_cli_fallback.py b/testing/tests/test_hermes_cli_fallback.py
index e2a8141e..728c8fe7 100644
--- a/testing/tests/test_hermes_cli_fallback.py
+++ b/testing/tests/test_hermes_cli_fallback.py
@@ -53,6 +53,10 @@ def test_goal_card_continues_after_local_judge_rejects_progress(
     )
     monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
     monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    lanes.atomic_json(
+        lanes.state_path("cassandra", "t_goal"),
+        {"goal_rejections": ["prior incomplete report"]},
+    )
     claude_low = lanes.Route(
         "claude", "claude-fable-5", "low", "claude-low", "jetson", "vote", 1, ()
     )
diff --git a/testing/tests/test_hermes_cli_finalization_edges.py b/testing/tests/test_hermes_cli_finalization_edges.py
index 0f2c2a2c..f8210243 100644
--- a/testing/tests/test_hermes_cli_finalization_edges.py
+++ b/testing/tests/test_hermes_cli_finalization_edges.py
@@ -159,6 +159,14 @@ def test_terminal_finalizer_closes_supplied_invalid_snapshots(
 
     monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
     path = lanes._terminal_path(lanes.state_path("board", "task"), 8)
+    lanes.atomic_json(path, {})
     empty = _Snapshot(None)
     assert lanes._finalize_terminal_record(_db(None), path, snapshot=empty) == "invalid"
     assert empty.closed == 1
+
+    finalization = __import__("cli_lane_finalization")
+    direct = _Snapshot(None)
+    assert finalization._finalize_terminal_record(
+        _db(None), path, snapshot=direct
+    ) == "invalid"
+    assert direct.closed == 1
diff --git a/testing/tests/test_hermes_cli_foundation_coverage.py b/testing/tests/test_hermes_cli_foundation_coverage.py
index 71f0c44a..83fda691 100644
--- a/testing/tests/test_hermes_cli_foundation_coverage.py
+++ b/testing/tests/test_hermes_cli_foundation_coverage.py
@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import json
+import os
+import runpy
 import sqlite3
 from contextlib import nullcontext
 from pathlib import Path
@@ -11,7 +13,7 @@ from urllib.error import URLError
 
 import pytest
 
-from testing.tests.test_hermes_cli_support import _completed_result, lanes
+from testing.tests.test_hermes_cli_support import SCRIPTS, _completed_result, lanes
 
 
 def test_board_context_serializes_objects_and_storage_failure_is_bounded(
@@ -93,6 +95,26 @@ def test_json_and_terminal_path_helpers_fail_closed(tmp_path: Path, monkeypatch)
     )
     assert lanes._terminal_evidence_identity(missing_evidence) is None
 
+    with pytest.raises(ValueError, match="positive SQLite"):
+        lanes.TerminalIdentity("board", "task", 0, "pending")
+
+
+def test_recovery_snapshot_close_is_idempotent_after_descriptors_are_released(
+    tmp_path: Path,
+):
+    directory = os.open(tmp_path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+    snapshot = lanes.TerminalRecoverySnapshot(
+        None,
+        tmp_path.stat(),
+        None,
+        directory,
+        b"",
+        "synthetic",
+    )
+    snapshot.close()
+    snapshot.close()
+    assert snapshot.directory_descriptor == -1
+
 
 def test_goal_helpers_cover_compaction_and_deterministic_failures():
     compacted = lanes.cli_lane_goal._bounded("a" * 100, 40)
@@ -122,6 +144,9 @@ def test_goal_helpers_cover_compaction_and_deterministic_failures():
     )
     assert accepted is False
     assert reason
+    assert lanes.cli_lane_goal.unfinished_result_reason(
+        {"status": "completed", "summary": "done", "tests_run": "not-a-list"}
+    ) is None
 
 
 def test_prompt_artifact_and_json_helpers_reject_unsafe_inputs(
@@ -147,6 +172,31 @@ def test_prompt_artifact_and_json_helpers_reject_unsafe_inputs(
     assert lanes._extract_json('prefix {"status":"blocked"} suffix') == {
         "status": "blocked"
     }
+    assert lanes._extract_json('{"status":"unknown"}') is None
+
+    state_file = workspace / "state.json"
+    state = {}
+    assert lanes._event_payload(
+        "codex",
+        json.dumps({"type": "thread.started", "result": {"status": "blocked"}}),
+        state,
+        state_file,
+    ) == {"status": "blocked"}
+    assert lanes._event_payload(
+        "claude",
+        json.dumps({"session_id": "session-2", "text": "not-json"}),
+        state,
+        state_file,
+    ) is None
+    assert state["claude_session_id"] == "session-2"
+
+    prompt_module = __import__("cli_lane_prompt")
+    assert prompt_module.workspace_artifacts(
+        workspace, [str(directory), str(artifact)]
+    ) == [str(artifact.resolve())]
+    assert prompt_module._event_payload(
+        "other", '{"item": {}}', {}, state_file
+    ) is None
 
 
 def test_provider_event_parsing_persists_sessions_and_nested_results(
@@ -205,3 +255,11 @@ def test_switchyard_network_failure_is_explicit():
                 URLError("offline")
             ),
         )
+
+
+def test_runner_main_guard_delegates_to_dispatch(monkeypatch):
+    dispatch = __import__("cli_lane_dispatch")
+    monkeypatch.setattr(dispatch, "main", lambda: 17)
+    with pytest.raises(SystemExit) as raised:
+        runpy.run_path(str(SCRIPTS / "cli_lane_runner.py"), run_name="__main__")
+    assert raised.value.code == 17
diff --git a/testing/tests/test_hermes_cli_provider_edges.py b/testing/tests/test_hermes_cli_provider_edges.py
index 21e57453..6f6f88a3 100644
--- a/testing/tests/test_hermes_cli_provider_edges.py
+++ b/testing/tests/test_hermes_cli_provider_edges.py
@@ -44,6 +44,26 @@ def test_stream_process_enforces_runtime_and_lease_boundaries(
     assert lease_lost.returncode != 0
 
 
+def test_stream_process_bounds_retained_line_history(tmp_path: Path):
+    result = lanes.stream_process(
+        [
+            sys.executable,
+            "-c",
+            "import sys; [sys.stdout.write(f'{n}\\n') for n in range(4010)]",
+        ],
+        provider="codex",
+        cwd=tmp_path,
+        env=dict(os.environ),
+        log_path=tmp_path / "bounded.log",
+        state={},
+        state_file=tmp_path / "state.json",
+        heartbeat=lambda _note: True,
+        max_runtime=60,
+    )
+    assert result.returncode == 0
+    assert "4009" in result.output
+
+
 def test_process_identity_and_signal_failures_are_bounded(monkeypatch):
     assert lanes._process_record(999999999) is None
     monkeypatch.setattr(
@@ -63,9 +83,26 @@ def test_process_identity_and_signal_failures_are_bounded(monkeypatch):
     )
     lanes._signal_worker_tree(10, {11: (12, 13)}, signal.SIGTERM)
 
+    signals = []
+    monkeypatch.setattr(lanes, "_process_identity_matches", lambda *_args: False)
+    monkeypatch.setattr(lanes.os, "killpg", lambda pid, sig: signals.append((pid, sig)))
+    monkeypatch.setattr(
+        lanes.os,
+        "kill",
+        lambda *_args: (_ for _ in ()).throw(AssertionError("stale PID signaled")),
+    )
+    lanes._signal_worker_tree(20, {21: (22, 23)}, signal.SIGKILL)
+    assert signals == [(20, signal.SIGKILL)]
+
 
 def test_descendant_snapshot_follows_multiple_generations(monkeypatch):
-    entries = [Path("/proc/self"), Path("/proc/10"), Path("/proc/11"), Path("/proc/12")]
+    entries = [
+        Path("/proc/self"),
+        Path("/proc/10"),
+        Path("/proc/11"),
+        Path("/proc/12"),
+        Path("/proc/13"),
+    ]
     monkeypatch.setattr(lanes.Path, "iterdir", lambda _path: iter(entries))
     records = {
         10: (1, 10, 100),
@@ -76,6 +113,34 @@ def test_descendant_snapshot_follows_multiple_generations(monkeypatch):
     assert lanes._descendant_processes(10) == {11: (11, 101), 12: (12, 102)}
 
 
+def test_claude_missing_unstarted_session_does_not_mark_started(
+    tmp_path: Path,
+    monkeypatch,
+):
+    state = {"claude_session_id": "synthetic-session"}
+    state_file = tmp_path / "state.json"
+    monkeypatch.setattr(
+        lanes,
+        "stream_process",
+        lambda *_args, **_kwargs: lanes.ProcessResult(
+            1, lanes.NO_CLAUDE_SESSION, None, False
+        ),
+    )
+    route = lanes.Route("claude", "model", "high", "profile", "test", "test", 1, ())
+    result = lanes.run_provider(
+        route,
+        "work",
+        tmp_path,
+        state,
+        state_file,
+        tmp_path / "log",
+        lambda _note: True,
+        60,
+    )
+    assert result.returncode == 1
+    assert "claude_started" not in state
+
+
 def test_terminate_escalates_after_term_timeout(monkeypatch):
     class Process:
         pid = 321
diff --git a/testing/tests/test_hermes_cli_records_edges.py b/testing/tests/test_hermes_cli_records_edges.py
index 3362466a..fed9e872 100644
--- a/testing/tests/test_hermes_cli_records_edges.py
+++ b/testing/tests/test_hermes_cli_records_edges.py
@@ -78,6 +78,17 @@ def test_small_snapshot_rejects_path_replaced_after_read(
     assert lanes._open_small_json_snapshot(path) is None
 
 
+def test_small_snapshot_rejects_a_reader_that_exceeds_its_bound(
+    tmp_path: Path,
+    monkeypatch,
+):
+    path = tmp_path / "record"
+    path.write_text("{}", encoding="utf-8")
+    monkeypatch.setattr(lanes, "MAX_TERMINAL_RECORD_BYTES", 8)
+    monkeypatch.setattr(lanes, "_read_bounded", lambda *_args: b"x" * 9)
+    assert lanes._open_small_json_snapshot(path) is None
+
+
 def test_recovery_snapshot_classifies_nonregular_and_open_failure(
     tmp_path: Path,
     monkeypatch,
diff --git a/testing/tests/test_hermes_cli_recovery_edges.py b/testing/tests/test_hermes_cli_recovery_edges.py
index 27ba6773..46687826 100644
--- a/testing/tests/test_hermes_cli_recovery_edges.py
+++ b/testing/tests/test_hermes_cli_recovery_edges.py
@@ -87,6 +87,9 @@ def test_staged_authority_rejects_outside_nested_and_unknown_state(
     document["kanban_state"] = "unknown"
     assert lanes._staged_terminal_authority(staged, _Snapshot(document)) is None
 
+    valid = _pending_terminal_record("board", "task", 5, "accepted")
+    assert lanes._staged_terminal_authority(staged, _Snapshot(valid)) is None
+
 
 def test_restore_staged_authority_handles_every_first_writer_state(
     tmp_path: Path,
@@ -296,3 +299,34 @@ def test_pending_guard_rejects_invalid_run_and_inaccessible_or_bad_artifacts(
 
     monkeypatch.setattr(Path, "stat", denied)
     assert lanes._has_pending_finalization("board", "task", 12) is False
+
+
+def test_pending_guard_scans_invalid_prepared_and_staged_candidates(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
+    base = lanes.state_path("board", "task")
+    pending = lanes._terminal_path(base, 13)
+    pending.parent.mkdir(parents=True)
+    pending.write_text("{}", encoding="utf-8")
+    (pending.parent / "task.run-13.terminal.prepared-bad.json").write_text(
+        "{}", encoding="utf-8"
+    )
+    (pending.parent / f"task.run-13.terminal.prepared-{'a' * 32}.json").write_text(
+        "{}", encoding="utf-8"
+    )
+    staged = [
+        pending.parent / f".retire.{'b' * 16}.{'c' * 16}.0",
+        pending.parent / f".retire.{'d' * 16}.{'e' * 16}.0",
+    ]
+    for path in staged:
+        path.write_text("{}", encoding="utf-8")
+    snapshots = iter((None, _Snapshot({})))
+    monkeypatch.setattr(
+        lanes,
+        "_open_terminal_recovery_snapshot",
+        lambda _path: next(snapshots),
+    )
+    monkeypatch.setattr(lanes, "_staged_terminal_authority", lambda *_args: None)
+    assert lanes._has_pending_finalization("board", "task", 13) is False
diff --git a/testing/tests/test_hermes_cli_retention_edges.py b/testing/tests/test_hermes_cli_retention_edges.py
index f7deb6a2..50f3cb19 100644
--- a/testing/tests/test_hermes_cli_retention_edges.py
+++ b/testing/tests/test_hermes_cli_retention_edges.py
@@ -87,3 +87,49 @@ def test_gc_skips_symlink_board_and_disappearing_candidate(
 
     monkeypatch.setattr(Path, "stat", disappear)
     assert lanes.gc_lane_artifacts() == 0
+
+
+def test_gc_counts_classified_quarantine_and_bounds_unlink_deferrals(
+    tmp_path: Path,
+    monkeypatch,
+):
+    missing = tmp_path / "missing"
+    assert lanes._unlink_artifact_if_same(missing, tmp_path.stat()) is False
+
+    root = tmp_path / "lanes"
+    board = root / "board"
+    board.mkdir(parents=True)
+    classified = board / "classified.candidate-1.json"
+    classified.write_text("{}", encoding="utf-8")
+    monkeypatch.setattr(lanes, "STATE_ROOT", root)
+    monkeypatch.setattr(
+        lanes,
+        "_artifact_gc_candidates",
+        lambda _board: [classified],
+    )
+    monkeypatch.setattr(
+        lanes,
+        "_quarantine_invalid_retained_terminal",
+        lambda _path: (True, True),
+    )
+    assert lanes.gc_lane_artifacts() == 1
+
+    staged = board / f".retire.{'a' * 16}.{'b' * 16}.0"
+    staged.write_text("{}", encoding="utf-8")
+    monkeypatch.setattr(
+        lanes,
+        "_artifact_gc_candidates",
+        lambda _board: [staged],
+    )
+    monkeypatch.setattr(
+        lanes,
+        "_quarantine_invalid_retained_terminal",
+        lambda _path: (False, False),
+    )
+    monkeypatch.setattr(lanes, "_open_terminal_recovery_snapshot", lambda _path: None)
+    monkeypatch.setattr(lanes, "_unlink_artifact_if_same", lambda *_args: False)
+    assert lanes.gc_lane_artifacts(
+        max_age_seconds=-1,
+        max_count=0,
+        max_bytes=-1,
+    ) == 0
diff --git a/testing/tests/test_hermes_runtime_stage_coverage.py b/testing/tests/test_hermes_runtime_stage_coverage.py
index 004853c2..3ca8e496 100644
--- a/testing/tests/test_hermes_runtime_stage_coverage.py
+++ b/testing/tests/test_hermes_runtime_stage_coverage.py
@@ -39,6 +39,21 @@ def test_private_directory_and_secret_copy_validate_mode_and_content(
     assert (runtime / "value").read_text(encoding="utf-8") == "staged\n"
 
 
+def test_private_directory_and_vault_projection_reject_invalid_entries(
+    tmp_path, monkeypatch
+):
+    module, vault, runtime, _home = _stage(tmp_path, monkeypatch)
+    checks = iter((False, True))
+    with monkeypatch.context() as invalid_directory:
+        invalid_directory.setattr(Path, "is_symlink", lambda _path: next(checks))
+        with pytest.raises(RuntimeError, match="is invalid"):
+            module._owned_directory(runtime)
+
+    (vault / "invalid").write_bytes(b"")
+    with pytest.raises(RuntimeError, match="projection invalid is invalid"):
+        module._read_secret("invalid")
+
+
 @pytest.mark.parametrize(
     ("document", "required", "match"),
     [
@@ -107,6 +122,77 @@ def test_node_ssh_config_rejects_nul_and_removes_partial_file(tmp_path, monkeypa
     assert not (runtime / "node-ssh-config").exists()
 
 
+def test_durable_credentials_reject_links_and_unsafe_existing_files(
+    tmp_path, monkeypatch
+):
+    module, vault, runtime, _home = _stage(tmp_path, monkeypatch)
+    runtime.mkdir()
+    (vault / "credential").write_text(
+        json.dumps({"tokens": {"refresh_token": "synthetic"}}),
+        encoding="utf-8",
+    )
+    destination = runtime / "credential.json"
+    destination.symlink_to(vault / "credential")
+    with pytest.raises(RuntimeError, match="must not be a symlink"):
+        module._bootstrap_json("credential", destination, ("tokens", "refresh_token"))
+
+    destination.unlink()
+    destination.write_text(
+        json.dumps({"tokens": {"refresh_token": "durable"}}),
+        encoding="utf-8",
+    )
+    destination.chmod(0o644)
+    with pytest.raises(RuntimeError, match="is unsafe"):
+        module._bootstrap_json("credential", destination, ("tokens", "refresh_token"))
+
+
+def test_durable_link_creates_repairs_preserves_and_rejects_targets(
+    tmp_path, monkeypatch
+):
+    module, _vault, _runtime, _home = _stage(tmp_path, monkeypatch)
+    monkeypatch.setattr(module.os, "chown", lambda *_args: None)
+    runtime = tmp_path / "worker"
+    runtime.mkdir()
+    durable = tmp_path / "provider"
+
+    module._durable_link(runtime, durable, "state")
+    target = runtime / "state"
+    assert target.is_symlink() and target.resolve() == durable.resolve()
+    module._durable_link(runtime, durable, "state")
+    assert target.is_symlink()
+
+    target.unlink()
+    wrong = tmp_path / "wrong"
+    wrong.mkdir()
+    target.symlink_to(wrong)
+    module._durable_link(runtime, durable, "state")
+    assert target.resolve() == durable.resolve()
+
+    target.unlink()
+    target.write_text("collision", encoding="utf-8")
+    with pytest.raises(RuntimeError, match="refusing to replace"):
+        module._durable_link(runtime, durable, "state")
+
+
+@pytest.mark.parametrize(
+    ("function", "match"),
+    [
+        ("stage_execution_worker", "worker ordinal is invalid"),
+        ("stage_execution_mediator", "mediator ordinal is invalid"),
+    ],
+)
+def test_execution_staging_rejects_invalid_ordinal(
+    tmp_path, monkeypatch, function, match
+):
+    module, _vault, _runtime, _home = _stage(tmp_path, monkeypatch)
+    monkeypatch.setattr(module, "WORKER_ROOT", tmp_path / "worker")
+    monkeypatch.setattr(module, "PROVIDER_ACCESS_ROOT", tmp_path / "provider")
+    monkeypatch.setattr(module, "POOL_ACCESS_ROOT", tmp_path / "pool")
+    monkeypatch.setenv("HERMES_WORKER_ORDINAL", "9")
+    with pytest.raises(RuntimeError, match=match):
+        getattr(module, function)()
+
+
 @pytest.mark.parametrize(
     ("mode", "secret_name"),
     [("chat", "chat-relay-key"), ("triage", "triage-api-key")],
@@ -124,13 +210,21 @@ def test_isolated_tenant_staging_exposes_only_required_secret(
     )
 
 
-@pytest.mark.parametrize("mode", ["agent", "chat", "triage"])
+@pytest.mark.parametrize(
+    "mode", ["agent", "chat", "triage", "execution-worker", "execution-mediator"]
+)
 def test_main_dispatches_each_mode(mode, monkeypatch, capsys):
     module = _load("stage_runtime_access")
     calls = []
     monkeypatch.setattr(module, "stage_agent", lambda: calls.append("agent"))
     monkeypatch.setattr(module, "stage_chat", lambda: calls.append("chat"))
     monkeypatch.setattr(module, "stage_triage", lambda: calls.append("triage"))
+    monkeypatch.setattr(
+        module, "stage_execution_worker", lambda: calls.append("execution-worker")
+    )
+    monkeypatch.setattr(
+        module, "stage_execution_mediator", lambda: calls.append("execution-mediator")
+    )
     monkeypatch.setattr(sys, "argv", ["stage-runtime-access", mode])
     assert module.main() == 0
     assert calls == [mode]