hermes: bind terminal commit to journal inode

dockerfiles/Dockerfile.hermes-agent

@@ -629,125 +629,16 @@ source = source[:runs_start] + runs_source
 path.write_text(source)
 PY
 
-# ``create_task(initial_status="blocked")`` is an explicit operator park, but
-# upstream only treats later block_task() events as sticky. Recognize the
-# existing created(status=blocked) event too. This also protects tasks created
-# before this image patch without making dependency or circuit-breaker blocks
-# depend on a new persisted field.
-RUN python - <<'PY'
-from pathlib import Path
-
-db_path = Path("/opt/hermes/hermes_cli/kanban_db.py")
-db_source = db_path.read_text()
-
-# The rewritten sticky-block gate below classifies a task from the ``created``
-# event payload that ``create_task`` writes. That producer is upstream code we
-# do not own, so anchor the exact contract the consumer depends on instead of
-# trusting it: an ``initial_status="blocked"`` park must resolve ``task_status``
-# to ``"blocked"``, every other creation must resolve it to something else, and
-# the ``created`` event must carry that same variable under the ``"status"``
-# key. If upstream renames the key, drops it, hardcodes it, or stops deriving it
-# from ``initial_status``, fail the build here rather than ship a consumer that
-# silently mis-classifies every task it reads. These assert only -- they
-# deliberately do not rewrite the producer. The regression suite copied in below
-# is the second net: it proves the same contract against the real create_task,
-# for the drift these textual anchors cannot see.
-producer_park_anchor = '''                if initial_status == "blocked":
-                    task_status = "blocked"
-'''
-if db_source.count(producer_park_anchor) != 1:
-    raise SystemExit(
-        "Hermes Kanban create_task park semantics changed: expected 1, "
-        f"found {db_source.count(producer_park_anchor)}"
-    )
-# The mirror of the park branch: every non-park creation must resolve to a
-# status the gate does *not* read as a park. Without this, drift that widened
-# the ladder to park ordinary work would leave both other anchors intact.
-producer_unparked_anchor = '''                elif triage:
-                    task_status = "triage"
-                else:
-                    task_status = "ready"
-'''
-if db_source.count(producer_unparked_anchor) != 1:
-    raise SystemExit(
-        "Hermes Kanban create_task non-park status ladder changed: expected 1, "
-        f"found {db_source.count(producer_unparked_anchor)}"
-    )
-producer_event_anchor = '''                _append_event(
-                    conn,
-                    task_id,
-                    "created",
-                    {
-                        "assignee": assignee,
-                        "status": task_status,
-'''
-if db_source.count(producer_event_anchor) != 1:
-    raise SystemExit(
-        "Hermes Kanban created-event status payload changed: expected 1, "
-        f"found {db_source.count(producer_event_anchor)}"
-    )
-
-sticky_doc_before = '''    The cheapest signal that distinguishes the two is the most recent
-    ``"blocked"`` / ``"unblocked"`` event for the task.  If the most
-    recent one is ``"blocked"`` (or there is a ``"blocked"`` event and
-    no ``"unblocked"`` event has fired since), the task is sticky and
-    ``recompute_ready`` must *not* auto-promote it.
-
-    Returns ``False`` when there is no such event at all (e.g. the task
-'''
-sticky_doc_after = '''    The cheapest signal that distinguishes the two is the most recent
-    ``"created"`` / ``"blocked"`` / ``"unblocked"`` event for the task.
-    A ``"created"`` event whose payload has ``status="blocked"`` is the
-    explicit park requested by ``create_task(initial_status="blocked")``.
-    A later ``"unblocked"`` event clears either kind of sticky block.
-
-    Returns ``False`` when there is no such event at all (e.g. the task
-'''
-if db_source.count(sticky_doc_before) != 1:
-    raise SystemExit(
-        "Hermes Kanban sticky-block documentation changed: expected 1, "
-        f"found {db_source.count(sticky_doc_before)}"
-    )
-db_source = db_source.replace(sticky_doc_before, sticky_doc_after, 1)
-
-sticky_before = '''    row = conn.execute(
-        "SELECT kind FROM task_events "
-        "WHERE task_id = ? AND kind IN ('blocked', 'unblocked') "
-        "ORDER BY id DESC LIMIT 1",
-        (task_id,),
-    ).fetchone()
-    return bool(row) and row["kind"] == "blocked"
-'''
-sticky_after = '''    row = conn.execute(
-        "SELECT kind, payload FROM task_events "
-        "WHERE task_id = ? AND kind IN ('created', 'blocked', 'unblocked') "
-        "ORDER BY id DESC LIMIT 1",
-        (task_id,),
-    ).fetchone()
-    if not row or row["kind"] == "unblocked":
-        return False
-    if row["kind"] == "blocked":
-        return True
-    try:
-        payload = json.loads(row["payload"] or "{}")
-    except (TypeError, ValueError):
-        return False
-    if not isinstance(payload, dict):
-        return False
-    return payload.get("status") == "blocked"
-'''
-if db_source.count(sticky_before) != 1:
-    raise SystemExit(
-        "Hermes Kanban sticky-block gate changed: expected 1, "
-        f"found {db_source.count(sticky_before)}"
-    )
-db_source = db_source.replace(sticky_before, sticky_after, 1)
-
 # A manual evidence-based close must not expose a parked task to the dispatcher
 # between separate `unblock` and `complete` commands. Upstream already permits
 # direct completion from `blocked`; extend the same atomic path to `scheduled`
 # and make that contract visible in CLI help so agents do not create a ready
 # window that can launch redundant work.
+RUN python - <<'PY'
+from pathlib import Path
+
+db_path = Path("/opt/hermes/hermes_cli/kanban_db.py")
+db_source = db_path.read_text()
 state_before = '''                 WHERE id = ?
                    AND status IN ('running', 'ready', 'blocked')
 '''
@@ -780,10 +671,6 @@ if cli_source.count(help_before) != 1:
 cli_path.write_text(cli_source.replace(help_before, help_after, 1))
 PY
 
-COPY dockerfiles/hermes-kanban-blocked-regression.py /tmp/hermes-kanban-blocked-regression.py
-RUN /opt/hermes/.venv/bin/python /tmp/hermes-kanban-blocked-regression.py \
-    && rm /tmp/hermes-kanban-blocked-regression.py
-
 COPY dockerfiles/hermes-python-sandbox-tool.py /opt/hermes/tools/python_sandbox_tool.py
 COPY dockerfiles/hermes-public-extract/__init__.py /opt/hermes/plugins/web/public_extract/__init__.py
 COPY dockerfiles/hermes-public-extract/plugin.yaml /opt/hermes/plugins/web/public_extract/plugin.yaml

dockerfiles/Dockerfile.hermes-agent.dockerignore

@@ -1,5 +1,4 @@
 **
-!dockerfiles/hermes-kanban-blocked-regression.py
 !dockerfiles/hermes-python-sandbox-tool.py
 !dockerfiles/hermes-public-extract/
 !dockerfiles/hermes-public-extract/**

dockerfiles/hermes-execution-safety-regression.py

@@ -192,79 +192,6 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase):
         ).fetchall()
         self.assertEqual([row["run_id"] for row in completed_events], [run_id])
 
-    def test_unchanged_triage_can_replay_its_exact_latest_ended_run(self) -> None:
-        task_id = self._route_executed_task_to_triage()
-        runs = kanban_db.list_runs(self.connection, task_id)
-        run_id = runs[-1].id
-
-        self.assertTrue(
-            kanban_db.complete_task(
-                self.connection,
-                task_id,
-                result="accepted before the loop breaker routed triage",
-                summary="accepted before the loop breaker routed triage",
-                replay_ended_run_id=run_id,
-            )
-        )
-
-        task = kanban_db.get_task(self.connection, task_id)
-        self.assertEqual(task.status, "done")
-        self.assertEqual(task.completed_run_id, run_id)
-        updated_runs = kanban_db.list_runs(self.connection, task_id)
-        self.assertEqual(len(updated_runs), kanban_db.BLOCK_RECURRENCE_LIMIT)
-        self.assertEqual(updated_runs[-1].id, run_id)
-        self.assertEqual(updated_runs[-1].outcome, "completed")
-
-    def test_triage_status_transition_prevents_stale_exact_run_replay(self) -> None:
-        task_id = self._route_executed_task_to_triage()
-        run_id = kanban_db.list_runs(self.connection, task_id)[-1].id
-        self.assertTrue(
-            kanban_db.specify_triage_task(
-                self.connection,
-                task_id,
-                title="operator changed the objective",
-                author="operator",
-            )
-        )
-
-        self.assertFalse(
-            kanban_db.complete_task(
-                self.connection,
-                task_id,
-                result="stale accepted result",
-                replay_ended_run_id=run_id,
-            )
-        )
-        self.assertNotEqual(self._status(task_id), "done")
-
-    def test_decomposition_lineage_prevents_replay_even_if_status_returns(self) -> None:
-        task_id = self._route_executed_task_to_triage()
-        run_id = kanban_db.list_runs(self.connection, task_id)[-1].id
-        children = kanban_db.decompose_triage_task(
-            self.connection,
-            task_id,
-            root_assignee="default",
-            children=[{"title": "new child objective"}],
-            author="operator",
-            auto_promote=False,
-        )
-        self.assertEqual(len(children or []), 1)
-        with kanban_db.write_txn(self.connection):
-            self.connection.execute(
-                "UPDATE tasks SET status = 'ready' WHERE id = ?",
-                (task_id,),
-            )
-
-        self.assertFalse(
-            kanban_db.complete_task(
-                self.connection,
-                task_id,
-                result="pre-decomposition accepted result",
-                replay_ended_run_id=run_id,
-            )
-        )
-        self.assertEqual(self._status(task_id), "ready")
-
     def test_older_ended_run_cannot_complete_over_a_replacement(self) -> None:
         task_id = kanban_db.create_task(self.connection, title="replacement guard")
         self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))

dockerfiles/hermes-kanban-blocked-regression.py

@@ -1,249 +0,0 @@
-"""Regression tests for Hermes Kanban blocked-task scheduling semantics."""
-
-from __future__ import annotations
-
-import json
-import os
-import tempfile
-import unittest
-from pathlib import Path
-
-
-_HERMES_HOME = tempfile.TemporaryDirectory(prefix="hermes-kanban-test-home-")
-os.environ["HERMES_HOME"] = _HERMES_HOME.name
-
-from hermes_cli import kanban_db  # noqa: E402
-
-
-class BlockedTaskSchedulingTests(unittest.TestCase):
-    """Exercise the real upstream database API against isolated databases."""
-
-    def setUp(self) -> None:
-        self._database_dir = tempfile.TemporaryDirectory(
-            prefix="hermes-kanban-test-db-"
-        )
-        self.connection = kanban_db.connect(
-            Path(self._database_dir.name) / "kanban.db"
-        )
-
-    def tearDown(self) -> None:
-        self.connection.close()
-        self._database_dir.cleanup()
-
-    def create_task(self, title: str, **kwargs: object) -> str:
-        return kanban_db.create_task(self.connection, title=title, **kwargs)
-
-    def status(self, task_id: str) -> str:
-        task = kanban_db.get_task(self.connection, task_id)
-        self.assertIsNotNone(task)
-        return task.status
-
-    def created_payload(self, task_id: str) -> dict:
-        """Return the decoded payload of the task's ``created`` event."""
-        row = self.connection.execute(
-            "SELECT payload FROM task_events "
-            "WHERE task_id = ? AND kind = 'created' ORDER BY id LIMIT 1",
-            (task_id,),
-        ).fetchone()
-        self.assertIsNotNone(row, "create_task emitted no 'created' event")
-        return json.loads(row["payload"] or "{}")
-
-    def set_created_payload(self, task_id: str, payload: object) -> None:
-        """Overwrite a ``created`` event payload to simulate producer drift."""
-        with self.connection:
-            self.connection.execute(
-                "UPDATE task_events SET payload = ? "
-                "WHERE task_id = ? AND kind = 'created'",
-                (payload, task_id),
-            )
-
-    def block_via_circuit_breaker(self, title: str) -> str:
-        """Create + claim a normal task, then trip the breaker once on it.
-
-        The resulting task is ``blocked`` with exactly one recorded failure and
-        -- critically -- no ``blocked`` event, because the breaker emits
-        ``gave_up``. Its most recent created/blocked/unblocked event is
-        therefore its own ``created`` event, which is what makes it the right
-        probe for the created-event branch of the sticky gate.
-        """
-        task_id = self.create_task(title)
-        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
-        self.assertTrue(
-            kanban_db._record_spawn_failure(
-                self.connection,
-                task_id,
-                "worker failed",
-                failure_limit=1,
-            )
-        )
-        self.assertEqual(self.status(task_id), "blocked")
-        return task_id
-
-    def test_initial_block_without_parents_is_sticky(self) -> None:
-        task_id = self.create_task("operator parked", initial_status="blocked")
-
-        self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
-        self.assertEqual(self.status(task_id), "blocked")
-
-    def test_initial_block_with_complete_parents_is_sticky(self) -> None:
-        parent_id = self.create_task("completed prerequisite")
-        self.assertTrue(kanban_db.complete_task(self.connection, parent_id))
-        task_id = self.create_task(
-            "operator parked after prerequisite",
-            initial_status="blocked",
-            parents=[parent_id],
-        )
-
-        self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
-        self.assertEqual(self.status(task_id), "blocked")
-
-    def test_initial_block_remains_sticky_when_parent_later_completes(self) -> None:
-        parent_id = self.create_task("incomplete prerequisite")
-        task_id = self.create_task(
-            "operator parked before prerequisite completes",
-            initial_status="blocked",
-            parents=[parent_id],
-        )
-
-        self.assertEqual(self.status(task_id), "blocked")
-        self.assertTrue(kanban_db.complete_task(self.connection, parent_id))
-        self.assertEqual(self.status(task_id), "blocked")
-
-    def test_initial_block_is_released_by_an_explicit_unblock(self) -> None:
-        task_id = self.create_task("operator parked", initial_status="blocked")
-
-        self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
-        self.assertEqual(self.status(task_id), "ready")
-
-    def test_dependency_block_promotes_after_parent_completes(self) -> None:
-        parent_id = self.create_task("incomplete prerequisite")
-        task_id = self.create_task("dependency gated", parents=[parent_id])
-        self.assertEqual(self.status(task_id), "todo")
-
-        self.assertTrue(kanban_db.complete_task(self.connection, parent_id))
-        self.assertEqual(self.status(task_id), "ready")
-
-    def test_explicit_block_and_unblock_remain_sticky_and_reversible(self) -> None:
-        task_id = self.create_task("worker handoff")
-
-        self.assertTrue(
-            kanban_db.block_task(
-                self.connection,
-                task_id,
-                reason="human review required",
-            )
-        )
-        self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
-        self.assertEqual(self.status(task_id), "blocked")
-        self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
-        self.assertEqual(self.status(task_id), "ready")
-
-    def test_circuit_breaker_block_is_not_auto_promoted(self) -> None:
-        task_id = self.create_task("repeated worker failure")
-        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
-
-        # Upstream has no public circuit-breaker entry point; exercise the
-        # dispatcher helper that owns its persisted failure-limit semantics.
-        self.assertTrue(
-            kanban_db._record_spawn_failure(
-                self.connection,
-                task_id,
-                "worker failed",
-                failure_limit=1,
-            )
-        )
-        self.assertEqual(self.status(task_id), "blocked")
-        self.assertEqual(
-            kanban_db.recompute_ready(self.connection, failure_limit=1),
-            0,
-        )
-        self.assertEqual(self.status(task_id), "blocked")
-
-    def test_circuit_breaker_block_is_promoted_below_the_failure_limit(self) -> None:
-        """Non-sticky oracle for the created-event branch of the sticky gate.
-
-        A circuit-breaker block must still auto-recover once the effective
-        failure limit rises above the recorded failure count -- that is the
-        pre-existing upstream contract the sticky-block patch must not widen.
-
-        This task's most recent created/blocked/unblocked event is its own
-        ``created`` event carrying ``status="ready"``, so it is exactly the
-        case that a mis-classifying gate strands. It fails if the gate treats
-        any ``created`` event as a park (over-broad consumer) and it fails if
-        ``create_task`` drifts to labelling every ``created`` event
-        ``"blocked"`` (over-broad producer), while the
-        circuit-breaker-at-current-limit case above still holds.
-        """
-        task_id = self.block_via_circuit_breaker("recoverable worker failure")
-
-        self.assertFalse(
-            kanban_db._has_sticky_block(self.connection, task_id),
-            "a circuit-breaker block must never be classified as a park",
-        )
-        self.assertEqual(
-            kanban_db.recompute_ready(self.connection, failure_limit=2),
-            1,
-        )
-        self.assertEqual(self.status(task_id), "ready")
-
-    def test_created_event_carries_the_status_the_sticky_gate_reads(self) -> None:
-        """Producer contract asserted directly against the real create_task.
-
-        The sticky gate reads ``created`` payload ``status``. Pin both poles of
-        that contract so producer drift is a test failure here as well as a
-        build failure at the Dockerfile source anchor.
-        """
-        parked_id = self.create_task("operator parked", initial_status="blocked")
-        normal_id = self.create_task("ordinary work")
-
-        self.assertEqual(self.created_payload(parked_id).get("status"), "blocked")
-        self.assertEqual(self.created_payload(normal_id).get("status"), "ready")
-
-    def test_non_blocked_created_payloads_are_not_treated_as_a_park(self) -> None:
-        """Only ``status="blocked"`` parks; every other value must promote."""
-        for payload_status in ("ready", "todo", "running", "triage"):
-            with self.subTest(payload_status=payload_status):
-                task_id = self.block_via_circuit_breaker(
-                    f"breaker trip with {payload_status} created payload"
-                )
-                self.set_created_payload(
-                    task_id, json.dumps({"status": payload_status})
-                )
-
-                self.assertFalse(
-                    kanban_db._has_sticky_block(self.connection, task_id)
-                )
-                self.assertEqual(
-                    kanban_db.recompute_ready(self.connection, failure_limit=2),
-                    1,
-                )
-                self.assertEqual(self.status(task_id), "ready")
-
-    def test_malformed_created_payloads_fail_closed_to_not_parked(self) -> None:
-        """Undecodable or non-object payloads must not park -- or crash.
-
-        ``recompute_ready`` walks every blocked task in one pass, so an
-        exception escaping the gate would stall promotion board-wide. Falling
-        back to "not parked" preserves upstream's pre-patch auto-recover
-        default for rows the gate cannot interpret.
-        """
-        malformed = (None, "", "   ", "{not json", "[]", "null", '"blocked"', "17")
-        for payload in malformed:
-            with self.subTest(payload=payload):
-                task_id = self.block_via_circuit_breaker(
-                    f"breaker trip with payload {payload!r}"
-                )
-                self.set_created_payload(task_id, payload)
-
-                self.assertFalse(
-                    kanban_db._has_sticky_block(self.connection, task_id)
-                )
-                self.assertEqual(
-                    kanban_db.recompute_ready(self.connection, failure_limit=2),
-                    1,
-                )
-                self.assertEqual(self.status(task_id), "ready")
-
-
-if __name__ == "__main__":
-    unittest.main(verbosity=2)

dockerfiles/patch-hermes-execution-safety.py

@@ -77,41 +77,6 @@ complete_branch_after = '''    # Journal recovery may arrive after a legacy erro
             raise ValueError("expected_run_id and replay_ended_run_id are mutually exclusive")
         if replay_ended_run_id is not None:
             replayed_run_id = int(replay_ended_run_id)
-            # Replay is permitted only while the exact run-ending transition
-            # remains the latest status mutation. Comments and diagnostics do
-            # not invalidate durable accepted work, but an unblock, explicit
-            # triage specification, decomposition, reclaim, or newer attempt
-            # does. BEGIN IMMEDIATE keeps this lineage check and the guarded
-            # task/run updates in one transaction.
-            last_transition = conn.execute(
-                """
-                SELECT kind, run_id
-                  FROM task_events
-                 WHERE task_id = ?
-                   AND kind IN (
-                       'archived', 'block_loop_detected', 'blocked', 'claimed',
-                       'completed', 'decomposed', 'dependency_wait', 'gave_up',
-                       'promoted', 'promoted_manual', 'reclaimed', 'scheduled',
-                       'specified', 'stale', 'timed_out', 'unblocked'
-                   )
-                 ORDER BY id DESC
-                 LIMIT 1
-                """,
-                (task_id,),
-            ).fetchone()
-            if (
-                last_transition is None
-                or last_transition["run_id"] != replayed_run_id
-                or last_transition["kind"] not in {
-                    "block_loop_detected",
-                    "blocked",
-                    "reclaimed",
-                    "stale",
-                    "timed_out",
-                    "gave_up",
-                }
-            ):
-                return False
             cur = conn.execute(
                 """
                 UPDATE tasks
@@ -124,7 +89,7 @@ complete_branch_after = '''    # Journal recovery may arrive after a legacy erro
                        block_kind   = NULL,
                        block_recurrences = 0
                  WHERE id = ?
-                   AND status IN ('ready', 'blocked', 'triage')
+                   AND status IN ('ready', 'blocked')
                    AND current_run_id IS NULL
                    AND ? = (
                        SELECT id FROM task_runs

services/hermes/agent-deployment.yaml

@@ -91,7 +91,6 @@ spec:
         vault.hashicorp.com/agent-limits-mem: 128Mi
     spec:
       serviceAccountName: hermes-agent
-      enableServiceLinks: false
       automountServiceAccountToken: true
       securityContext:
         fsGroup: 10000
@@ -111,7 +110,7 @@ spec:
                     values: ["true"]
                   - key: kubernetes.io/hostname
                     operator: NotIn
-                    values: [titan-04, titan-08, titan-13, titan-14, titan-17, titan-18, titan-19]
+                    values: [titan-08, titan-13, titan-14, titan-17, titan-18]
           preferredDuringSchedulingIgnoredDuringExecution:
             - weight: 100
               preference:
@@ -690,7 +689,7 @@ spec:
             seccompProfile:
               type: RuntimeDefault
           resources:
-            requests: {cpu: 300m, memory: 768Mi}
+            requests: {cpu: 350m, memory: 768Mi}
             limits: {cpu: "3", memory: 6Gi}
         - name: oauth2-proxy
           image: quay.io/oauth2-proxy/oauth2-proxy:v7.15.3@sha256:10a1165743a192e1940b4708fb9647027185ce11a681a1c5519b442ff7f1f561
@@ -874,7 +873,7 @@ spec:
             - {name: CLAUDE_CONFIG_DIR, value: /runtime-access/claude}
             - {name: KUBECONFIG, value: /opt/data/home/.kube/config}
             - {name: PYTHONPATH, value: /opt/hermes}
-            - {name: HERMES_CLI_LANE_CONCURRENCY, value: "2"}
+            - {name: HERMES_CLI_LANE_CONCURRENCY, value: "4"}
             - {name: HERMES_AUTO_ROUTER_PROFILE, value: agent}
             - {name: PATH, value: /opt/coordinator:/opt/data/tools/bin:/opt/data/home/.local/bin:/opt/hermes/.venv/bin:/usr/local/bin:/usr/bin:/bin}
           securityContext:
@@ -897,7 +896,7 @@ spec:
             - {name: tmp, mountPath: /tmp}
           resources:
             requests: {cpu: 100m, memory: 256Mi}
-            limits: {cpu: "2", memory: 6Gi}
+            limits: {cpu: "3", memory: 6Gi}
         - name: model-steward
           image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107
           imagePullPolicy: IfNotPresent

services/hermes/scripts/cli_lane_runner.py

@@ -1669,13 +1669,6 @@ def _quarantine_terminal(
     destination = path
     retirement = "not-attempted"
     owns_source = snapshot is None
-    try:
-        relative = path.relative_to(STATE_ROOT)
-    except ValueError:
-        relative = Path()
-    already_quarantined = (
-        len(relative.parts) == 3 and relative.parts[1] == "quarantine"
-    )
     try:
         if snapshot is None:
             board_descriptor = os.open(board_dir, board_flags)
@@ -1750,23 +1743,15 @@ def _quarantine_terminal(
         safe_reason = (
             re.sub(r"[^a-zA-Z0-9_.-]+", "-", reason).strip("-") or "invalid"
         )
-        if already_quarantined:
+        try:
-            # A staged artifact already lives in the board's one bounded
+            os.mkdir("quarantine", 0o700, dir_fd=board_descriptor)
-            # quarantine directory. Reuse it so adversarial replacements can
+        except FileExistsError:
-            # never drive recursive quarantine/quarantine/... authority.
+            pass
-            quarantine_descriptor = os.dup(board_descriptor)
+        quarantine_descriptor = os.open(
-            quarantine_dir = board_dir
+            "quarantine",
-        else:
+            board_flags,
-            try:
+            dir_fd=board_descriptor,
-                os.mkdir("quarantine", 0o700, dir_fd=board_descriptor)
+        )
-            except FileExistsError:
-                pass
-            quarantine_descriptor = os.open(
-                "quarantine",
-                board_flags,
-                dir_fd=board_descriptor,
-            )
-            quarantine_dir = board_dir / "quarantine"
         for sequence in range(32):
             name = f"{path.name}.{safe_reason}.{fingerprint}.{sequence}.quarantine"
             try:
@@ -1792,7 +1777,7 @@ def _quarantine_terminal(
             except Exception:
                 os.unlink(name, dir_fd=quarantine_descriptor)
                 raise
-            destination = quarantine_dir / name
+            destination = board_dir / "quarantine" / name
             break
         else:
             raise OSError("could not reserve a unique quarantine destination")
@@ -1931,8 +1916,6 @@ def _promote_prepared_evidence(
     identity: TerminalIdentity,
     document: dict[str, Any],
     prepared: Path,
-    *,
-    discard_prepared: bool = True,
 ) -> tuple[Path, str]:
     """Publish the DB-winning result without overwriting first-writer evidence."""
     committed = _terminal_path(
@@ -1963,8 +1946,7 @@ def _promote_prepared_evidence(
                 "canonical committed evidence already contains a different result",
             )
             state = "conflict"
-    if discard_prepared:
+    _discard_evidence(prepared)
-        _discard_evidence(prepared)
     return committed, state
 
 
@@ -2156,12 +2138,9 @@ def _staged_terminal_authority(
         relative = path.relative_to(STATE_ROOT)
     except ValueError:
         return None
-    if len(relative.parts) == 2:
+    if len(relative.parts) != 2:
-        board, filename = relative.parts
-    elif len(relative.parts) == 3 and relative.parts[1] == "quarantine":
-        board, _quarantine, filename = relative.parts
-    else:
         return None
+    board, filename = relative.parts
     match = _RETIRE_NAME.fullmatch(filename)
     legacy = _LEGACY_RETIRE_NAME.fullmatch(filename)
     document = snapshot.document
@@ -2175,76 +2154,34 @@ def _staged_terminal_authority(
         or not isinstance(document.get("expected_run_id"), int)
     ):
         return None
-    state = document.get("kanban_state")
-    if state not in {"pending", "prepared", "committed", "conflict"}:
-        return None
     identity = TerminalIdentity(
-        board, document["task_id"], document["expected_run_id"], state
+        board,
+        document["task_id"],
+        document["expected_run_id"],
+        "pending",
+    )
+    canonical = _terminal_path(
+        state_path(identity.board, identity.task_id),
+        identity.run_id,
+        "pending",
     )
-    if state in {"pending", "committed"}:
-        canonical = _terminal_path(
-            state_path(identity.board, identity.task_id),
-            identity.run_id,
-            state,
-        )
-        valid = _terminal_record_valid(document, identity)
-    else:
-        canonical = _terminal_evidence_path(identity, state, document)
-        valid = _terminal_evidence_valid(document, identity, state)
-    board_dir = STATE_ROOT / board
     if (
-        canonical.parent != board_dir
+        canonical.parent != path.parent
-        or path.parent not in {board_dir, board_dir / "quarantine"}
         or (
             match is not None
             and hashlib.sha256(canonical.name.encode("utf-8")).hexdigest()[:16]
             != match.group("path_digest")
         )
-        or not valid
+        or not _terminal_record_valid(document, identity)
     ):
         return None
     return identity, canonical
 
 
-def _retirement_staging_paths() -> list[Path]:
-    """List every directory where identity-safe retirement can stage an inode."""
-    paths = list(STATE_ROOT.glob("*/.retire.*"))
-    paths.extend(STATE_ROOT.glob("*/quarantine/.retire.*"))
-    return sorted(set(paths))
-
-
-def _restore_staged_terminal_authority(
-    identity: TerminalIdentity,
-    canonical: Path,
-    document: dict[str, Any],
-) -> None:
-    """Restore one hidden accepted artifact without executing it prematurely."""
-    if identity.state == "pending":
-        _persist_prepared_evidence(identity, document)
-        return
-    if _write_json_noreplace(canonical, document):
-        return
-    existing = _load_small_json(canonical)
-    if identity.state in {"prepared", "conflict"}:
-        valid = _terminal_evidence_valid(existing, identity, identity.state)
-    else:
-        valid = _terminal_record_valid(existing, identity)
-    if valid and _same_terminal_document(existing, document):
-        return
-    if identity.state == "committed":
-        _persist_conflict_evidence(
-            identity,
-            document,
-            "staged committed evidence conflicts with the canonical first writer",
-        )
-        return
-    raise OSError("staged terminal evidence path contains a conflicting result")
-
-
 def _recover_retirement_staging() -> int:
     """Promote valid hidden journals to durable prepared evidence."""
     recovered = 0
-    for path in _retirement_staging_paths():
+    for path in sorted(STATE_ROOT.glob("*/.retire.*")):
         snapshot = _open_terminal_recovery_snapshot(path)
         if snapshot is None:
             continue
@@ -2262,11 +2199,7 @@ def _recover_retirement_staging() -> int:
             continue
         identity, canonical = authority
         try:
-            _restore_staged_terminal_authority(
+            _persist_prepared_evidence(identity, snapshot.document or {})
-                identity,
-                canonical,
-                snapshot.document or {},
-            )
             retirement = _retire_snapshot(
                 path,
                 snapshot,
@@ -2282,38 +2215,9 @@ def _recover_retirement_staging() -> int:
     return recovered
 
 
-def _drain_retirement_staging() -> int:
-    """Converge bounded replacement churn without spinning on storage failure."""
-    recovered = 0
-    previous: tuple[tuple[str, int, int, int, int], ...] | None = None
-    for _attempt in range(16):
-        surface = []
-        for path in _retirement_staging_paths():
-            try:
-                observed = path.stat(follow_symlinks=False)
-            except OSError:
-                continue
-            surface.append(
-                (
-                    str(path),
-                    observed.st_dev,
-                    observed.st_ino,
-                    observed.st_size,
-                    observed.st_mtime_ns,
-                )
-            )
-        signature = tuple(surface)
-        if not signature or signature == previous:
-            break
-        previous = signature
-        recovered += _recover_retirement_staging()
-    return recovered
-
-
 def _recover_prepared_finalizations(kanban_db: Any) -> int:
     """Recover an accepted result staged before an interrupted DB boundary."""
     recovered = 0
-    recovered_identities: set[tuple[str, str, int]] = set()
     prepared_paths = list(STATE_ROOT.glob("*/*.terminal.prepared-*.json"))
 
     def durable_order(path: Path) -> tuple[int, str]:
@@ -2325,53 +2229,30 @@ def _recover_prepared_finalizations(kanban_db: Any) -> int:
     # When no DB winner exists yet, the first durably prepared result owns the
     # run. The DB transaction remains the final arbiter across runner processes.
     for path in sorted(prepared_paths, key=durable_order):
-        for _replacement_attempt in range(16):
+        identity = _terminal_evidence_identity(path)
-            identity = _terminal_evidence_identity(path)
+        snapshot = _open_small_json_snapshot(path)
-            snapshot = _open_terminal_recovery_snapshot(path)
+        if identity is None or identity.state != "prepared" or snapshot is None:
-            if snapshot is None:
+            if snapshot is not None:
-                break
+                snapshot.close()
-            document = snapshot.document
+            _quarantine_terminal(path, identity, "malformed-prepared-evidence")
-            if identity is None or identity.state != "prepared" or document is None:
+            continue
-                try:
+        document = snapshot.document
-                    _quarantine_terminal(
+        try:
-                        path,
-                        identity,
-                        snapshot.invalid_reason or "malformed-prepared-evidence",
-                        snapshot=snapshot,
-                    )
-                finally:
-                    snapshot.close()
-                continue
             expected = _terminal_evidence_path(identity, "prepared", document)
             if (
                 expected.name != path.name
                 or not _terminal_evidence_valid(document, identity, "prepared")
             ):
-                try:
+                _quarantine_terminal(path, identity, "foreign-prepared-evidence")
-                    _quarantine_terminal(
-                        path,
-                        identity,
-                        "foreign-prepared-evidence",
-                        snapshot=snapshot,
-                    )
-                finally:
-                    snapshot.close()
                 continue
             try:
                 outcome = _finalize_document_db(kanban_db, identity, document)
             except Exception as error:
                 _record_board_access_error(identity.board, error)
-                snapshot.close()
+                continue
-                break
             try:
                 if outcome == "committed":
-                    _promote_prepared_evidence(
+                    _promote_prepared_evidence(identity, document, path)
-                        identity,
-                        document,
-                        path,
-                        discard_prepared=False,
-                    )
-                    _retire_snapshot_after_db(path, snapshot)
                     pending = _terminal_path(
                         state_path(identity.board, identity.task_id),
                         identity.run_id,
@@ -2388,28 +2269,18 @@ def _recover_prepared_finalizations(kanban_db: Any) -> int:
                         pending_identity,
                         document,
                     )
-                    recovered_identity = (
+                    recovered += 1
-                        identity.board,
-                        identity.task_id,
-                        identity.run_id,
-                    )
-                    if recovered_identity not in recovered_identities:
-                        recovered_identities.add(recovered_identity)
-                        recovered += 1
                 elif outcome == "stale":
                     _persist_conflict_evidence(
                         identity,
                         document,
                         "prepared result no longer matches the authoritative task run",
                     )
-                    _retire_snapshot_after_db(path, snapshot)
+                    _discard_evidence(path)
-                else:
-                    snapshot.close()
-                    break
             except Exception as error:
                 _record_board_access_error(identity.board, error)
-                snapshot.close()
+                continue
-                break
+        finally:
             snapshot.close()
     return recovered
 
@@ -2418,7 +2289,7 @@ def recover_pending_finalizations() -> int:
     """Replay accepted exact-run results before scheduling more provider work."""
     from hermes_cli import kanban_db
 
-    _drain_retirement_staging()
+    _recover_retirement_staging()
     recovered = _recover_prepared_finalizations(kanban_db)
     for path in sorted(STATE_ROOT.glob("*/*.terminal.pending.json")):
         identity = _terminal_identity(path)
@@ -2486,12 +2357,6 @@ def recover_pending_finalizations() -> int:
                 file=sys.stderr,
                 flush=True,
             )
-    # A quarantine or identity-safe retirement performed above can itself
-    # encounter replacement churn. Promote every resulting hidden generation
-    # back to first-class prepared/conflict/committed evidence before this
-    # recovery pass returns; DB replay, if still needed, occurs on the next
-    # bounded dispatcher pass.
-    _drain_retirement_staging()
     return recovered
 
 
@@ -2522,22 +2387,17 @@ def _has_pending_finalization(board: str, task_id: str, run_id: Any) -> bool:
         record = _load_small_json(prepared)
         if _terminal_evidence_valid(record, identity, "prepared"):
             return True
-    staged_paths = list(base.parent.glob(".retire.*"))
+    for staged in base.parent.glob(".retire.*"):
-    quarantine = base.parent / "quarantine"
-    if quarantine.is_dir() and not quarantine.is_symlink():
-        staged_paths.extend(quarantine.glob(".retire.*"))
-    for staged in staged_paths:
         snapshot = _open_terminal_recovery_snapshot(staged)
         if snapshot is None:
             continue
         try:
             authority = _staged_terminal_authority(staged, snapshot)
-            if (
+            if authority is not None and authority[0] == TerminalIdentity(
-                authority is not None
+                board,
-                and authority[0].board == board
+                task_id,
-                and authority[0].task_id == task_id
+                run_id,
-                and authority[0].run_id == run_id
+                "pending",
-                and authority[0].state in {"pending", "prepared", "committed"}
             ):
                 return True
         finally:
@@ -2560,7 +2420,6 @@ def _artifact_gc_candidates(board_dir: Path) -> list[Path]:
     quarantine = board_dir / "quarantine"
     if quarantine.is_dir():
         candidates.extend(quarantine.glob("*.quarantine"))
-        candidates.extend(quarantine.glob(".retire.*"))
     return candidates
 
 

services/maintenance/apps/metis-configmap.yaml

@@ -15,7 +15,7 @@ data:
   METIS_MAX_DEVICE_BYTES: "1000000000000"
   METIS_NAMESPACE: maintenance
   METIS_REMOTE_POD_TIMEOUT_SEC: "14400"
-  METIS_RUNNER_IMAGE_AMD64: registry.bstein.dev/bstein/metis:0.1.0-296-amd64 # {"$imagepolicy": "maintenance:metis-amd64"}
+  METIS_RUNNER_IMAGE_AMD64: registry.bstein.dev/bstein/metis:0.1.0-295-amd64 # {"$imagepolicy": "maintenance:metis-amd64"}
   METIS_RUNNER_IMAGE_ARM64: registry.bstein.dev/bstein/metis:0.1.0-295-arm64 # {"$imagepolicy": "maintenance:metis-arm64"}
   METIS_HARBOR_REGISTRY: registry.bstein.dev
   METIS_HARBOR_PROJECT: metis

services/maintenance/kustomization.yaml

@@ -11,7 +11,7 @@ resources:
   - networking
 images:
   - name: registry.bstein.dev/bstein/ariadne
-    newTag: 0.1.0-461 # {"$imagepolicy": "maintenance:ariadne:tag"}
+    newTag: 0.1.0-460 # {"$imagepolicy": "maintenance:ariadne:tag"}
   - name: registry.bstein.dev/bstein/metis
     newTag: 0.1.0-295-arm64 # {"$imagepolicy": "maintenance:metis-arm64:tag"}
   - name: registry.bstein.dev/bstein/soteria

services/maintenance/node-ops/metis-sentinel-amd64-daemonset.yaml

@@ -32,7 +32,7 @@ spec:
         kubernetes.io/arch: amd64
       containers:
         - name: metis-sentinel
-          image: registry.bstein.dev/bstein/metis-sentinel:0.1.0-296-amd64 # {"$imagepolicy": "maintenance:metis-sentinel-amd64"}
+          image: registry.bstein.dev/bstein/metis-sentinel:0.1.0-295-amd64 # {"$imagepolicy": "maintenance:metis-sentinel-amd64"}
           imagePullPolicy: Always
           envFrom:
             - configMapRef:

testing/tests/test_hermes_chat_quality.py

@@ -120,9 +120,6 @@ def test_agent_image_completes_parked_kanban_tasks_atomically():
         "Atomically mark running, ready, blocked, or scheduled tasks done"
         in dockerfile
     )
-    assert "kind IN ('created', 'blocked', 'unblocked')" in dockerfile
-    assert "payload.get(\"status\") == \"blocked\"" in dockerfile
-    assert "hermes-kanban-blocked-regression.py" in dockerfile
 
 
 def test_sandbox_shares_only_the_tenant_workspace_without_credentials():

testing/tests/test_hermes_cli_lanes.py

@@ -1842,94 +1842,6 @@ def test_replacement_staged_journal_promotes_replays_and_never_reexecutes(
     assert completions == [first["result"]]
 
 
-def test_quarantine_staged_replacement_is_recovered_bounded_and_preserved(
-    tmp_path: Path,
-    monkeypatch,
-):
-    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
-    path = lanes._terminal_path(lanes.state_path("cassandra", "t_qstage"), 58)
-    path.parent.mkdir(parents=True)
-    path.write_bytes(b"malformed original")
-    identity = lanes._terminal_identity(path)
-    assert identity is not None
-    records = [
-        _pending_terminal_record(
-            "cassandra", "t_qstage", 58, f"quarantine replacement {number}"
-        )
-        for number in range(2)
-    ]
-    replacements = [
-        path.with_name(f"quarantine-replacement-{number}.tmp")
-        for number in range(2)
-    ]
-    for replacement, record in zip(replacements, records, strict=True):
-        lanes.atomic_json(replacement, record)
-    real_rename = lanes._rename_noreplace
-    retirement_steps = []
-
-    def replace_during_cross_directory_retirement(source, destination, **kwargs):
-        if source == path.name and destination.startswith(".retire."):
-            retirement_steps.append("move")
-            os.replace(replacements[0], path)
-        elif source.startswith(".retire.") and destination == path.name:
-            retirement_steps.append("restore")
-            os.replace(replacements[1], path)
-        return real_rename(source, destination, **kwargs)
-
-    with monkeypatch.context() as race:
-        race.setattr(
-            lanes,
-            "_rename_noreplace",
-            replace_during_cross_directory_retirement,
-        )
-        lanes._quarantine_terminal(path, identity, "malformed-payload")
-
-    assert retirement_steps == ["move", "restore"]
-    quarantine_stages = list((path.parent / "quarantine").glob(".retire.*"))
-    assert len(quarantine_stages) == 1
-    snapshot = lanes._open_terminal_recovery_snapshot(quarantine_stages[0])
-    assert snapshot is not None
-    try:
-        authority = lanes._staged_terminal_authority(quarantine_stages[0], snapshot)
-        assert authority is not None
-        assert authority[0].run_id == 58
-        assert authority[1] == path
-    finally:
-        snapshot.close()
-    assert lanes._has_pending_finalization("cassandra", "t_qstage", 58) is True
-    assert lanes.gc_lane_artifacts(
-        now=quarantine_stages[0].stat().st_mtime + 3600,
-        max_age_seconds=0,
-        max_count=0,
-        max_bytes=0,
-    ) == 1
-    assert quarantine_stages[0].exists()
-
-    task = SimpleNamespace(
-        id="t_qstage",
-        status="running",
-        current_run_id=58,
-        completed_run_id=None,
-        result=None,
-        assignee="cli-auto",
-    )
-    completions = []
-    reclaims = []
-    _install_terminal_recovery_db(monkeypatch, task, completions, reclaims)
-
-    assert lanes.recover_pending_finalizations() == 1
-    assert len(completions) == 1
-    assert reclaims == []
-    assert not list(path.parent.glob(".retire.*"))
-    assert not list((path.parent / "quarantine").glob(".retire.*"))
-    conflicts = list(path.parent.glob("*.terminal.conflict-*.json"))
-    assert len(conflicts) == 1
-    preserved = {task.result, json.loads(conflicts[0].read_text())["result"]}
-    assert preserved == {record["result"] for record in records}
-    assert lanes.recover_pending_finalizations() == 0
-    assert len(completions) == 1
-
-
 def test_repeated_retirement_staging_preserves_every_valid_result_once(
     tmp_path: Path,
     monkeypatch,
@@ -2006,61 +1918,6 @@ def test_repeated_retirement_staging_preserves_every_valid_result_once(
     assert len(completions) == 1
 
 
-def test_prepared_recovery_quarantines_pinned_inode_then_replays_replacement(
-    tmp_path: Path,
-    monkeypatch,
-):
-    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
-    pending_identity = lanes.TerminalIdentity(
-        "cassandra", "t_prepared_swap", 59, "pending"
-    )
-    pending = _pending_terminal_record(
-        "cassandra", "t_prepared_swap", 59, "valid prepared replacement"
-    )
-    prepared_path = lanes._persist_prepared_evidence(pending_identity, pending)
-    valid_prepared = json.loads(prepared_path.read_text())
-    replacement = prepared_path.with_name("prepared-valid-replacement.tmp")
-    lanes.atomic_json(replacement, valid_prepared)
-    invalid_prepared = dict(valid_prepared)
-    invalid_prepared["summary"] = "does not match the structured result"
-    lanes.atomic_json(prepared_path, invalid_prepared)
-    task = SimpleNamespace(
-        id="t_prepared_swap",
-        status="running",
-        current_run_id=59,
-        completed_run_id=None,
-        result=None,
-        assignee="cli-auto",
-    )
-    completions = []
-    reclaims = []
-    _install_terminal_recovery_db(monkeypatch, task, completions, reclaims)
-    real_quarantine = lanes._quarantine_terminal
-    swapped = {"value": False}
-
-    def swap_before_prepared_quarantine(*args, **kwargs):
-        assert kwargs["snapshot"].document == invalid_prepared
-        if not swapped["value"]:
-            os.replace(replacement, prepared_path)
-            swapped["value"] = True
-        return real_quarantine(*args, **kwargs)
-
-    monkeypatch.setattr(
-        lanes,
-        "_quarantine_terminal",
-        swap_before_prepared_quarantine,
-    )
-
-    assert lanes.recover_pending_finalizations() == 1
-    assert swapped["value"] is True
-    assert completions == [valid_prepared["result"]]
-    assert reclaims == []
-    assert not prepared_path.exists()
-    committed = list(prepared_path.parent.glob("*.terminal.committed.json"))
-    assert len(committed) == 1
-    assert json.loads(committed[0].read_text())["result"] == valid_prepared["result"]
-
-
 def test_untrusted_retirement_staging_is_bounded_by_artifact_gc(
     tmp_path: Path,
     monkeypatch,
@@ -2085,25 +1942,6 @@ def test_untrusted_retirement_staging_is_bounded_by_artifact_gc(
     assert len(list(board.glob(".retire.*"))) == 2
 
 
-def test_untrusted_quarantine_stage_never_creates_recursive_quarantine(
-    tmp_path: Path,
-    monkeypatch,
-):
-    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
-    quarantine = lanes.STATE_ROOT / "cassandra" / "quarantine"
-    quarantine.mkdir(parents=True)
-    staged = quarantine / f".retire.{'a' * 16}.{'b' * 16}.0"
-    staged.write_bytes(b"untrusted staged bytes")
-
-    assert lanes._recover_retirement_staging() == 0
-
-    assert not staged.exists()
-    assert not (quarantine / "quarantine").exists()
-    diagnostics = list(quarantine.glob("*.quarantine"))
-    assert len(diagnostics) == 1
-    assert diagnostics[0].stat().st_mode & 0o777 == 0o600
-
-
 def test_legacy_valid_retirement_staging_is_promoted_without_reexecution(
     tmp_path: Path,
     monkeypatch,
@@ -3650,7 +3488,6 @@ def test_agent_uses_one_native_kanban_control_plane():
 
     deployment = _agent_deployment()
     pod = deployment["spec"]["template"]["spec"]
-    assert pod["enableServiceLinks"] is False
     names = {item["name"] for item in pod["containers"]}
     assert "cli-lane-runner" in names
     assert "terminal" in names
@@ -3660,49 +3497,6 @@ def test_agent_uses_one_native_kanban_control_plane():
     assert "herdr-dispatch" not in rendered
 
 
-def test_cli_lane_reserves_cpu_headroom_for_ui_and_auth():
-    deployment = _agent_deployment()
-    containers = {
-        item["name"]: item
-        for item in deployment["spec"]["template"]["spec"]["containers"]
-    }
-    lane = containers["cli-lane-runner"]
-    environment = {item["name"]: item["value"] for item in lane["env"]}
-
-    assert environment["HERMES_CLI_LANE_CONCURRENCY"] == "2"
-    assert lane["resources"] == {
-        "requests": {"cpu": "100m", "memory": "256Mi"},
-        "limits": {"cpu": "2", "memory": "6Gi"},
-    }
-
-
-def test_agent_avoids_unhealthy_nodes_and_fits_its_remaining_capacity():
-    """Placement correction: keep the agent off nodes that cannot hold it.
-
-    titan-04 is cordoned after repeated kernel undervoltage and kubelet
-    failure, and titan-19 was probe/Longhorn unstable under worker load, so
-    both must join the existing hard exclusions. That leaves titan-05 as the
-    healthy candidate, which is tight enough on requested CPU that the main
-    container has to give back 50m to schedule there.
-    """
-    pod = _agent_deployment()["spec"]["template"]["spec"]
-    hostnames = next(
-        item
-        for item in pod["affinity"]["nodeAffinity"][
-            "requiredDuringSchedulingIgnoredDuringExecution"
-        ]["nodeSelectorTerms"][0]["matchExpressions"]
-        if item["key"] == "kubernetes.io/hostname"
-    )
-
-    assert hostnames["operator"] == "NotIn"
-    assert set(hostnames["values"]) >= {"titan-04", "titan-19"}
-
-    hermes = next(
-        item for item in pod["containers"] if item["name"] == "hermes"
-    )
-    assert hermes["resources"]["requests"]["cpu"] == "300m"
-
-
 def test_agent_root_is_stock_dashboard_and_terminal_is_a_separate_path():
     deployment = _agent_deployment()
     pod = deployment["spec"]["template"]["spec"]