hermes: bind terminal commit to journal inode

dockerfiles/Dockerfile.hermes-agent

@@ -1552,10 +1552,8 @@ COPY dockerfiles/patch_hermes_decomposition_safety.py /tmp/patch_hermes_decompos
 COPY dockerfiles/hermes_execution_regression_support.py /tmp/hermes_execution_regression_support.py
 COPY dockerfiles/hermes_run_safety_regression.py /tmp/hermes_run_safety_regression.py
 COPY dockerfiles/hermes_decomposition_safety_regression.py /tmp/hermes_decomposition_safety_regression.py
-COPY services/hermes/scripts/cli_lane_*.py /tmp/hermes-lane-regression/
 RUN /opt/hermes/.venv/bin/python /tmp/patch-hermes-execution-safety.py \
-    && HERMES_CLI_LANE_SOURCE=/tmp/hermes-lane-regression \
-        /opt/hermes/.venv/bin/python /tmp/hermes-execution-safety-regression.py \
+    && /opt/hermes/.venv/bin/python /tmp/hermes-execution-safety-regression.py \
     && rm /tmp/patch-hermes-execution-safety.py \
         /tmp/hermes-execution-safety-regression.py \
         /tmp/hermes_execution_patch_support.py \
@@ -1563,8 +1561,7 @@ RUN /opt/hermes/.venv/bin/python /tmp/patch-hermes-execution-safety.py \
         /tmp/patch_hermes_decomposition_safety.py \
         /tmp/hermes_execution_regression_support.py \
         /tmp/hermes_run_safety_regression.py \
-        /tmp/hermes_decomposition_safety_regression.py \
-    && find /tmp/hermes-lane-regression -depth -delete
+        /tmp/hermes_decomposition_safety_regression.py
 
 COPY dockerfiles/hermes-session-migrate.py /opt/hermes/bin/hermes-session-migrate
 

dockerfiles/Dockerfile.hermes-agent.dockerignore

@@ -13,7 +13,3 @@
 !dockerfiles/hermes_execution_regression_support.py
 !dockerfiles/hermes_run_safety_regression.py
 !dockerfiles/hermes_decomposition_safety_regression.py
-!services/
-!services/hermes/
-!services/hermes/scripts/
-!services/hermes/scripts/cli_lane_*.py

dockerfiles/hermes_execution_regression_support.py

@@ -4,28 +4,18 @@ from __future__ import annotations
 
 import json
 import os
-import sys
 import tempfile
 import unittest
-from pathlib import Path
 from types import SimpleNamespace
 from unittest import mock
 
 _HERMES_HOME = tempfile.TemporaryDirectory(prefix="hermes-execution-test-home-")
 os.environ["HERMES_HOME"] = _HERMES_HOME.name
-_LANE_SOURCE = Path(
-    os.environ.get("HERMES_CLI_LANE_SOURCE", "/tmp/hermes-lane-regression")
-)
-sys.path.insert(0, str(_LANE_SOURCE))
 
 from agent import auxiliary_client  # noqa: E402
-from cli_lane_config import TerminalIdentity  # noqa: E402
-from cli_lane_finalization import _finalize_document_db  # noqa: E402
 from hermes_cli import kanban_db  # noqa: E402
 from hermes_cli import kanban_decompose  # noqa: E402
 
-__all__ = ("ExecutionSafetyTestCase", "TerminalIdentity", "_finalize_document_db")
-
 
 class _FakeCompletions:
     def __init__(self, payload: dict, before_response=None) -> None:

dockerfiles/hermes_run_safety_regression.py

@@ -6,138 +6,12 @@ import threading
 from unittest import mock
 
 from hermes_cli import kanban_db
-from hermes_execution_regression_support import (
-    ExecutionSafetyTestCase,
-    TerminalIdentity,
-    _finalize_document_db,
-)
-from cli_lane_files import state_path
-from cli_lane_records import _write_terminal_record
-from cli_lane_recovery import recover_pending_finalizations
+from hermes_execution_regression_support import ExecutionSafetyTestCase
 
 
 class ExactRunSafetyTests(ExecutionSafetyTestCase):
     """Verify exact-run transactions reject every stale concurrent actor."""
 
-    def _lane_document(self, summary: str) -> dict:
-        """Return the DB-bearing portion of one accepted lane journal."""
-        return {
-            "result": summary,
-            "summary": summary,
-            "metadata": {"source": "exact-run-regression"},
-        }
-
-    def test_lane_marks_replaced_ended_run_stale_without_mutation(self) -> None:
-        task_id = kanban_db.create_task(self.connection, title="replaced journal")
-        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
-        first_run = kanban_db.get_task(self.connection, task_id).current_run_id
-        self.assertTrue(
-            kanban_db.block_task(
-                self.connection,
-                task_id,
-                reason="first run ended after persistence",
-                expected_run_id=first_run,
-            )
-        )
-        self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
-        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
-        second_run = kanban_db.get_task(self.connection, task_id).current_run_id
-        self.assertTrue(
-            kanban_db.block_task(
-                self.connection,
-                task_id,
-                reason="replacement run also ended",
-                expected_run_id=second_run,
-            )
-        )
-        before = kanban_db.get_task(self.connection, task_id)
-
-        outcome = _finalize_document_db(
-            kanban_db,
-            TerminalIdentity("default", task_id, first_run, "pending"),
-            self._lane_document("stale first result"),
-        )
-
-        after = kanban_db.get_task(self.connection, task_id)
-        self.assertEqual(outcome, "stale")
-        self.assertEqual(after.status, before.status)
-        self.assertEqual(after.current_run_id, before.current_run_id)
-        self.assertEqual(after.result, before.result)
-        self.assertEqual(after.completed_run_id, before.completed_run_id)
-
-    def test_lane_marks_nonexistent_ended_run_stale_without_mutation(self) -> None:
-        task_id = kanban_db.create_task(self.connection, title="no such run")
-        before = kanban_db.get_task(self.connection, task_id)
-
-        outcome = _finalize_document_db(
-            kanban_db,
-            TerminalIdentity("default", task_id, 42, "pending"),
-            self._lane_document("orphan result"),
-        )
-
-        after = kanban_db.get_task(self.connection, task_id)
-        self.assertEqual(outcome, "stale")
-        self.assertEqual(after.status, before.status)
-        self.assertEqual(after.result, before.result)
-        self.assertEqual(kanban_db.list_runs(self.connection, task_id), [])
-
-    def test_full_lane_recovery_conflicts_a_replaced_ended_run(self) -> None:
-        task_id = kanban_db.create_task(self.connection, title="recovery conflict")
-        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
-        first_run = kanban_db.get_task(self.connection, task_id).current_run_id
-        self.assertTrue(
-            kanban_db.block_task(
-                self.connection,
-                task_id,
-                reason="first accepted run ended",
-                expected_run_id=first_run,
-            )
-        )
-        self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
-        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
-        second_run = kanban_db.get_task(self.connection, task_id).current_run_id
-        self.assertTrue(
-            kanban_db.block_task(
-                self.connection,
-                task_id,
-                reason="replacement ended",
-                expected_run_id=second_run,
-            )
-        )
-        before = kanban_db.get_task(self.connection, task_id)
-        state_file = state_path("default", task_id)
-        pending, _record = _write_terminal_record(
-            state_file,
-            board="default",
-            task_id=task_id,
-            run_id=first_run,
-            structured={
-                "status": "completed",
-                "summary": "accepted stale result",
-                "changed_files": [],
-                "tests_run": ["real SQLite recovery regression"],
-                "artifacts": [],
-                "findings": [],
-                "blockers": [],
-            },
-            summary="accepted stale result",
-            metadata={"source": "exact-run-regression"},
-        )
-
-        self.assertEqual(recover_pending_finalizations(), 0)
-
-        after = kanban_db.get_task(self.connection, task_id)
-        self.assertEqual(after.status, before.status)
-        self.assertEqual(after.current_run_id, before.current_run_id)
-        self.assertEqual(after.result, before.result)
-        self.assertFalse(pending.exists())
-        conflicts = list(
-            state_file.parent.glob(
-                f"{state_file.stem}.run-{first_run}.terminal.conflict-*.json"
-            )
-        )
-        self.assertEqual(len(conflicts), 1)
-
     def test_latest_ended_run_can_replay_a_durable_completion(self) -> None:
         task_id = kanban_db.create_task(self.connection, title="journaled result")
         self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))

services/bstein-dev-home/kustomization.yaml

@@ -20,9 +20,9 @@ resources:
   - ingress.yaml
 images:
   - name: registry.bstein.dev/bstein/bstein-dev-home-frontend
-    newTag: 0.1.1-471 # {"$imagepolicy": "bstein-dev-home:bstein-dev-home-frontend:tag"}
+    newTag: 0.1.1-470 # {"$imagepolicy": "bstein-dev-home:bstein-dev-home-frontend:tag"}
   - name: registry.bstein.dev/bstein/bstein-dev-home-backend
-    newTag: 0.1.1-471 # {"$imagepolicy": "bstein-dev-home:bstein-dev-home-backend:tag"}
+    newTag: 0.1.1-470 # {"$imagepolicy": "bstein-dev-home:bstein-dev-home-backend:tag"}
 configMapGenerator:
   - name: chat-ai-gateway
     namespace: bstein-dev-home

services/hermes/scripts/cli_lane_config.py

@@ -38,7 +38,6 @@ ARTIFACT_RETENTION_COUNT = 256
 ARTIFACT_RETENTION_BYTES = 128 * 1024 * 1024
 MAX_TERMINAL_RECORD_BYTES = 1024 * 1024
 QUARANTINE_HASH_BYTES = 64 * 1024
-SQLITE_RUN_ID_MAX = 2**63 - 1
 PROVIDER_HEALTH_PATHS = {
     "codex": DATA_ROOT / "provider-health/codex.json",
     "claude": DATA_ROOT / "provider-health/claude.json",
@@ -124,23 +123,6 @@ class TerminalFinalizationPending(RuntimeError):
     """An accepted worker result is durable but not committed to Kanban yet."""
 
 
-def canonical_run_id(value: object) -> int | None:
-    """Return one positive SQLite-safe run ID from an int or canonical decimal."""
-    if type(value) is int:
-        candidate = value
-    elif type(value) is str:
-        # Bound conversion before ``int`` so an attacker cannot hand Python an
-        # arbitrarily large decimal from an otherwise lexically valid name.
-        if not re.fullmatch(r"[1-9][0-9]{0,18}", value):
-            return None
-        candidate = int(value)
-    else:
-        return None
-    if not 1 <= candidate <= SQLITE_RUN_ID_MAX:
-        return None
-    return candidate
-
-
 @dataclass(frozen=True)
 class TerminalIdentity:
     """Replay authority derived only from a journal's directory and filename."""
@@ -150,11 +132,6 @@ class TerminalIdentity:
     run_id: int
     state: str
 
-    def __post_init__(self) -> None:
-        """Reject noncanonical identities before they can reach SQLite."""
-        if type(self.run_id) is not int or canonical_run_id(self.run_id) is None:
-            raise ValueError("terminal run identity is not a positive SQLite int64")
-
 
 @dataclass
 class TerminalSnapshot:

services/hermes/scripts/cli_lane_files.py

@@ -11,7 +11,7 @@ import uuid
 from pathlib import Path
 from typing import Any
 
-from cli_lane_config import STATE_ROOT, TerminalIdentity, canonical_run_id
+from cli_lane_config import STATE_ROOT, TerminalIdentity
 
 
 def atomic_json(path: Path, value: dict[str, Any], mode: int = 0o600) -> None:
@@ -88,11 +88,9 @@ def _terminal_path(
     """Return the identity-bound replay path for an accepted terminal response."""
     if state not in {"pending", "committed"}:
         raise ValueError(f"invalid terminal journal state: {state}")
-    canonical = canonical_run_id(run_id)
-    if type(run_id) is not int or canonical is None:
-        raise ValueError("terminal run ID must be a positive SQLite int64")
+    safe_run = re.sub(r"[^a-zA-Z0-9_.-]+", "-", str(run_id or "unknown"))
     return state_file.with_name(
-        f"{state_file.stem}.run-{canonical}.terminal.{state}.json"
+        f"{state_file.stem}.run-{safe_run}.terminal.{state}.json"
     )
 
 _TERMINAL_NAME = re.compile(
@@ -121,14 +119,12 @@ def _terminal_identity(path: Path) -> TerminalIdentity | None:
         return None
     board, filename = relative.parts
     match = _TERMINAL_NAME.fullmatch(filename)
-    run_id = canonical_run_id(match.group("run")) if match else None
     try:
         board_stat = path.parent.stat(follow_symlinks=False)
     except OSError:
         return None
     if (
         not match
-        or run_id is None
         or not stat.S_ISDIR(board_stat.st_mode)
         or not _SAFE_BOARD.fullmatch(board)
         or board in {".", ".."}
@@ -138,7 +134,7 @@ def _terminal_identity(path: Path) -> TerminalIdentity | None:
     return TerminalIdentity(
         board=board,
         task_id=match.group("task"),
-        run_id=run_id,
+        run_id=int(match.group("run")),
         state=match.group("state"),
     )
 
@@ -152,14 +148,12 @@ def _terminal_evidence_identity(path: Path) -> TerminalIdentity | None:
         return None
     board, filename = relative.parts
     match = _TERMINAL_EVIDENCE_NAME.fullmatch(filename)
-    run_id = canonical_run_id(match.group("run")) if match else None
     try:
         board_stat = path.parent.stat(follow_symlinks=False)
     except OSError:
         return None
     if (
         not match
-        or run_id is None
         or not stat.S_ISDIR(board_stat.st_mode)
         or not _SAFE_BOARD.fullmatch(board)
         or board in {".", ".."}
@@ -169,6 +163,6 @@ def _terminal_evidence_identity(path: Path) -> TerminalIdentity | None:
     return TerminalIdentity(
         board=board,
         task_id=match.group("task"),
-        run_id=run_id,
+        run_id=int(match.group("run")),
         state=match.group("state"),
     )

services/hermes/scripts/cli_lane_finalization.py

@@ -8,13 +8,7 @@ from pathlib import Path
 from typing import Any
 
 from cli_lane_board import _board_call, _external, _record_board_access_error, _task_value
-from cli_lane_config import (
-    TerminalIdentity,
-    TerminalRecoverySnapshot,
-    TerminalSnapshot,
-    canonical_run_id,
-    utc_now,
-)
+from cli_lane_config import TerminalIdentity, TerminalRecoverySnapshot, TerminalSnapshot, utc_now
 from cli_lane_evidence import (
     _persist_conflict_evidence,
     _persist_prepared_evidence,
@@ -37,7 +31,7 @@ def _recover_exact_run(
     reason: str,
 ) -> bool:
     """Make an exact external run retryable after journal recovery fails."""
-    if identity is None or canonical_run_id(identity.run_id) is None:
+    if identity is None:
         return False
 
     def operation(conn: Any) -> bool:
@@ -157,8 +151,6 @@ def _finalize_document_db(
     document: dict[str, Any],
 ) -> str:
     """Apply one validated exact-run result to Kanban under DB run guards."""
-    if canonical_run_id(identity.run_id) is None:
-        return "stale"
 
     def operation(conn: Any) -> str:
         task = kanban_db.get_task(conn, identity.task_id)
@@ -199,10 +191,7 @@ def _finalize_document_db(
         # A duplicate finalizer can lose the guarded UPDATE to an identical
         # first writer. Re-read rather than reporting a false pending state.
         latest = kanban_db.get_task(conn, identity.task_id)
-        latest_status = (
-            str(_task_value(latest, "status", "")) if latest is not None else ""
-        )
-        if latest_status == "done":
+        if latest is not None and str(_task_value(latest, "status", "")) == "done":
             return (
                 "committed"
                 if (
@@ -212,15 +201,7 @@ def _finalize_document_db(
                 )
                 else "stale"
             )
-        # A guarded completion can be retried only while this exact live run
-        # still owns the task. An ended, replaced, or nonexistent run cannot
-        # become authoritative later and must converge to conflict evidence.
-        if (
-            latest_status == "running"
-            and _task_value(latest, "current_run_id", None) == identity.run_id
-        ):
-            return "pending"
-        return "stale"
+        return "pending"
 
     return str(_board_call(kanban_db, identity.board, operation))
 

services/hermes/scripts/cli_lane_records.py

@@ -18,7 +18,6 @@ from cli_lane_config import (
     TerminalIdentity,
     TerminalRecoverySnapshot,
     TerminalSnapshot,
-    canonical_run_id,
     utc_now,
 )
 from cli_lane_files import (
@@ -282,8 +281,6 @@ def _write_terminal_record(
     metadata: dict[str, Any],
 ) -> tuple[Path, dict[str, Any]]:
     """Journal an accepted result before attempting the Kanban transaction."""
-    if type(run_id) is not int or canonical_run_id(run_id) is None:
-        raise ValueError("terminal run ID must be a positive SQLite int64")
     path = _terminal_path(state_file, run_id)
     record = {
         "board": board,
@@ -311,11 +308,7 @@ def _terminal_record_valid(
         for key in required_strings
     ):
         return False
-    expected_run_id = record.get("expected_run_id")
-    if (
-        type(expected_run_id) is not int
-        or canonical_run_id(expected_run_id) is None
-    ):
+    if not isinstance(record.get("expected_run_id"), int):
         return False
     if not isinstance(record.get("metadata"), dict):
         return False

services/hermes/scripts/cli_lane_recovery.py

@@ -10,12 +10,7 @@ from pathlib import Path
 from typing import Any
 
 from cli_lane_board import _record_board_access_error
-from cli_lane_config import (
-    STATE_ROOT,
-    TerminalIdentity,
-    TerminalRecoverySnapshot,
-    canonical_run_id,
-)
+from cli_lane_config import STATE_ROOT, TerminalIdentity, TerminalRecoverySnapshot
 from cli_lane_evidence import (
     _persist_conflict_evidence,
     _persist_prepared_evidence,
@@ -93,8 +88,7 @@ def _staged_terminal_authority(
         or board in {".", ".."}
         or document.get("board") != board
         or not isinstance(document.get("task_id"), str)
-        or type(document.get("expected_run_id")) is not int
-        or canonical_run_id(document.get("expected_run_id")) is None
+        or not isinstance(document.get("expected_run_id"), int)
     ):
         return None
     state = document.get("kanban_state")
@@ -412,7 +406,7 @@ def recover_pending_finalizations() -> int:
 
 def _has_pending_finalization(board: str, task_id: str, run_id: Any) -> bool:
     """Keep an exact run claimed while its accepted result awaits replay."""
-    if type(run_id) is not int or canonical_run_id(run_id) is None:
+    if not isinstance(run_id, int):
         return False
     path = _terminal_path(state_path(board, task_id), run_id, "pending")
     try:

services/hermes/scripts/cli_lane_retention.py

@@ -16,15 +16,10 @@ from cli_lane_config import (
     ARTIFACT_RETENTION_COUNT,
     STATE_ROOT,
 )
-from cli_lane_evidence import _retire_terminal_entry, _terminal_evidence_valid
-from cli_lane_files import (
-    _fsync_directory,
-    _terminal_evidence_identity,
-    _terminal_identity,
-)
-from cli_lane_quarantine import _quarantine_terminal
+from cli_lane_evidence import _retire_terminal_entry
+from cli_lane_files import _fsync_directory
 from cli_lane_recovery import _staged_terminal_authority
-from cli_lane_records import _open_terminal_recovery_snapshot, _terminal_record_valid
+from cli_lane_records import _open_terminal_recovery_snapshot
 
 
 def _artifact_gc_candidates(board_dir: Path) -> list[Path]:
@@ -65,47 +60,6 @@ def _unlink_artifact_if_same(path: Path, observed: os.stat_result) -> bool:
         if descriptor is not None:
             os.close(descriptor)
 
-
-def _quarantine_invalid_retained_terminal(path: Path) -> tuple[bool, bool]:
-    """Audit retained committed/conflict evidence on the bounded GC cadence."""
-    committed = path.name.endswith(".terminal.committed.json")
-    conflict = ".terminal.conflict-" in path.name and path.name.endswith(".json")
-    if not committed and not conflict:
-        return False, False
-    identity = (
-        _terminal_identity(path) if committed else _terminal_evidence_identity(path)
-    )
-    snapshot = _open_terminal_recovery_snapshot(path)
-    if snapshot is None:
-        return True, False
-    document = snapshot.document
-    valid = False
-    if identity is not None and document is not None:
-        valid = (
-            _terminal_record_valid(document, identity)
-            if committed
-            else _terminal_evidence_valid(document, identity, "conflict")
-        )
-    if valid:
-        snapshot.close()
-        return False, False
-    try:
-        _quarantine_terminal(
-            path,
-            identity,
-            snapshot.invalid_reason or "invalid-retained-terminal-evidence",
-            snapshot=snapshot,
-        )
-    finally:
-        snapshot.close()
-    try:
-        path.stat(follow_symlinks=False)
-    except FileNotFoundError:
-        return True, True
-    except OSError:
-        return True, False
-    return True, False
-
 def gc_lane_artifacts(
     *,
     now: float | None = None,
@@ -122,12 +76,6 @@ def gc_lane_artifacts(
         board_removed = 0
         entries: list[tuple[Path, os.stat_result]] = []
         for path in _artifact_gc_candidates(board_dir):
-            classified, quarantined = _quarantine_invalid_retained_terminal(path)
-            if classified:
-                if quarantined:
-                    removed += 1
-                    board_removed += 1
-                continue
             if path.name.startswith(".retire."):
                 snapshot = _open_terminal_recovery_snapshot(path)
                 if snapshot is not None:

services/hermes/scripts/cli_lane_runner.py

@@ -41,7 +41,6 @@ from cli_lane_config import (
     Route,
     TerminalFinalizationPending,
     TerminalIdentity,
-    canonical_run_id,
     utc_now,
 )
 from cli_lane_dispatch import claim_ready, main, recover_orphans

testing/tests/test_hermes_agent_security.py

@@ -36,9 +36,6 @@ def test_agent_image_runs_execution_safety_patch_and_regressions():
     assert "/opt/hermes/.venv/bin/python /tmp/hermes-execution-safety-regression.py" in (
         dockerfile
     )
-    assert "COPY services/hermes/scripts/cli_lane_*.py" in dockerfile
-    assert "!services/hermes/scripts/cli_lane_*.py" in dockerignore
-    assert "HERMES_CLI_LANE_SOURCE=/tmp/hermes-lane-regression" in dockerfile
 
 
 def test_agent_refreshes_routes_after_restoring_cli_logins():

testing/tests/test_hermes_cli_finalization_authority.py

@@ -1,53 +0,0 @@
-"""Post-transaction authority classification for exact terminal results."""
-
-from __future__ import annotations
-
-from testing.tests.test_hermes_cli_support import (
-    SimpleNamespace,
-    lanes,
-    nullcontext,
-    pytest,
-)
-
-
-@pytest.mark.parametrize(
-    ("status", "current_run_id", "expected"),
-    [
-        ("running", 17, "pending"),
-        ("running", 18, "stale"),
-        ("ready", None, "stale"),
-        ("blocked", None, "stale"),
-        ("triage", None, "stale"),
-    ],
-)
-def test_failed_completion_is_pending_only_for_the_exact_live_run(
-    status: str,
-    current_run_id: int | None,
-    expected: str,
-):
-    task = SimpleNamespace(
-        id="t_authority",
-        status=status,
-        current_run_id=current_run_id,
-        completed_run_id=None,
-        result=None,
-    )
-
-    class Connection:
-        def close(self):
-            return None
-
-    database = SimpleNamespace(
-        scoped_current_board=lambda _board: nullcontext(),
-        connect=lambda board: Connection(),
-        get_task=lambda _conn, _task_id: task,
-        complete_task=lambda *_args, **_kwargs: False,
-    )
-
-    outcome = lanes._finalize_document_db(
-        database,
-        lanes.TerminalIdentity("cassandra", "t_authority", 17, "pending"),
-        {"result": "accepted", "summary": "accepted", "metadata": {}},
-    )
-
-    assert outcome == expected

testing/tests/test_hermes_cli_run_identity.py

@@ -1,217 +0,0 @@
-"""Canonical SQLite run-identity contracts for every terminal artifact class."""
-
-from __future__ import annotations
-
-from testing.tests.test_hermes_cli_support import (
-    Path,
-    _completed_result,
-    _pending_terminal_record,
-    lanes,
-    pytest,
-    sys,
-)
-
-
-class _NoDatabase:
-    """Fail if malformed run authority reaches a Kanban transaction."""
-
-    def __init__(self) -> None:
-        self.accesses = []
-
-    def scoped_current_board(self, *_args, **_kwargs):
-        self.accesses.append("scope")
-        raise AssertionError("noncanonical authority reached the database")
-
-
-@pytest.mark.parametrize(
-    "value",
-    [
-        False,
-        True,
-        0,
-        -1,
-        2**63,
-        "0",
-        "01",
-        "9223372036854775808",
-        "9" * 150,
-        "+1",
-        " 1",
-    ],
-)
-def test_run_identity_rejects_noncanonical_or_non_sqlite_values(value):
-    assert lanes.canonical_run_id(value) is None
-
-
-@pytest.mark.parametrize("value", [1, 2**63 - 1, "1", "9223372036854775807"])
-def test_run_identity_accepts_the_positive_sqlite_int64_domain(value):
-    assert lanes.canonical_run_id(value) == int(value)
-
-
-@pytest.mark.parametrize("run_id", [False, True, 0, -1, 2**63, 10**149])
-def test_terminal_writer_refuses_invalid_runtime_run_ids(tmp_path: Path, run_id):
-    with pytest.raises(ValueError, match="SQLite int64"):
-        lanes._write_terminal_record(
-            lanes.state_path("cassandra", "t_invalid_writer"),
-            board="cassandra",
-            task_id="t_invalid_writer",
-            run_id=run_id,
-            structured=_completed_result(),
-            summary="completed",
-            metadata={},
-        )
-
-
-@pytest.mark.parametrize(
-    ("state", "suffix", "parser"),
-    [
-        ("pending", "terminal.pending.json", "_terminal_identity"),
-        ("committed", "terminal.committed.json", "_terminal_identity"),
-        (
-            "prepared",
-            f"terminal.prepared-{'a' * 32}.json",
-            "_terminal_evidence_identity",
-        ),
-        (
-            "conflict",
-            f"terminal.conflict-{'a' * 32}.json",
-            "_terminal_evidence_identity",
-        ),
-    ],
-)
-@pytest.mark.parametrize("encoded", ["0", "01", "9223372036854775808", "9" * 150])
-def test_every_terminal_filename_class_rejects_noncanonical_run_ids(
-    tmp_path: Path,
-    monkeypatch,
-    state: str,
-    suffix: str,
-    parser: str,
-    encoded: str,
-):
-    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
-    board = lanes.STATE_ROOT / "cassandra"
-    board.mkdir(parents=True)
-    path = board / f"t_filename.run-{encoded}.{suffix}"
-
-    assert getattr(lanes, parser)(path) is None, state
-
-
-@pytest.mark.parametrize("run_id", [False, True, 0, -1, 2**63, 10**149])
-@pytest.mark.parametrize("state", ["pending", "prepared", "committed", "conflict"])
-def test_every_terminal_record_class_rejects_invalid_run_ids(run_id, state):
-    record = _pending_terminal_record(
-        "cassandra", "t_record", run_id, "invalid identity"
-    )
-    record["kanban_state"] = state
-
-    assert lanes._terminal_record_valid(record) is False
-
-
-@pytest.mark.parametrize("run_id", [False, 0, 2**63, 10**149])
-def test_retirement_staging_rejects_invalid_document_run_ids(
-    tmp_path: Path,
-    monkeypatch,
-    run_id,
-):
-    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
-    board = lanes.STATE_ROOT / "cassandra"
-    board.mkdir(parents=True)
-    staged = board / f".retire.{'a' * 16}.{'b' * 16}.0"
-    lanes.atomic_json(
-        staged,
-        _pending_terminal_record(
-            "cassandra", "t_staged", run_id, "invalid staged identity"
-        ),
-    )
-    snapshot = lanes._open_terminal_recovery_snapshot(staged)
-    assert snapshot is not None
-    try:
-        assert lanes._staged_terminal_authority(staged, snapshot) is None
-    finally:
-        snapshot.close()
-    assert lanes._recover_retirement_staging() == 0
-    assert not staged.exists()
-    assert len(list((board / "quarantine").glob("*.quarantine"))) == 1
-
-
-def test_noncanonical_pending_filename_is_quarantined_before_db_access(
-    tmp_path: Path,
-    monkeypatch,
-):
-    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
-    board = lanes.STATE_ROOT / "cassandra"
-    board.mkdir(parents=True)
-    pending = board / "t_noncanonical.run-01.terminal.pending.json"
-    lanes.atomic_json(
-        pending,
-        _pending_terminal_record(
-            "cassandra", "t_noncanonical", 1, "noncanonical filename"
-        ),
-    )
-    database = _NoDatabase()
-    monkeypatch.setitem(
-        sys.modules,
-        "hermes_cli",
-        type("HermesCli", (), {"kanban_db": database})(),
-    )
-
-    assert lanes.recover_pending_finalizations() == 0
-    assert database.accesses == []
-    assert not pending.exists()
-    assert len(list((board / "quarantine").glob("*.quarantine"))) == 1
-
-
-def test_noncanonical_prepared_filename_is_quarantined_before_db_access(
-    tmp_path: Path,
-    monkeypatch,
-):
-    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
-    board = lanes.STATE_ROOT / "cassandra"
-    board.mkdir(parents=True)
-    prepared = (
-        board / f"t_prepared.run-01.terminal.prepared-{'a' * 32}.json"
-    )
-    record = _pending_terminal_record(
-        "cassandra", "t_prepared", 1, "noncanonical prepared filename"
-    )
-    record["kanban_state"] = "prepared"
-    lanes.atomic_json(prepared, record)
-    database = _NoDatabase()
-    monkeypatch.setitem(
-        sys.modules,
-        "hermes_cli",
-        type("HermesCli", (), {"kanban_db": database})(),
-    )
-
-    assert lanes.recover_pending_finalizations() == 0
-    assert database.accesses == []
-    assert not prepared.exists()
-    assert len(list((board / "quarantine").glob("*.quarantine"))) == 1
-
-
-@pytest.mark.parametrize(
-    ("state", "suffix"),
-    [
-        ("committed", "terminal.committed.json"),
-        ("conflict", f"terminal.conflict-{'a' * 32}.json"),
-    ],
-)
-def test_gc_quarantines_noncanonical_retained_evidence(
-    tmp_path: Path,
-    monkeypatch,
-    state: str,
-    suffix: str,
-):
-    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
-    board = lanes.STATE_ROOT / "cassandra"
-    board.mkdir(parents=True)
-    artifact = board / f"t_retained.run-01.{suffix}"
-    record = _pending_terminal_record(
-        "cassandra", "t_retained", 1, "noncanonical retained filename"
-    )
-    record["kanban_state"] = state
-    lanes.atomic_json(artifact, record)
-
-    assert lanes.gc_lane_artifacts() == 1
-    assert not artifact.exists()
-    assert len(list((board / "quarantine").glob("*.quarantine"))) == 1