hermes(agent): replace HERDR with native CLI lanes

services/hermes/agent-configmap.yaml

@@ -75,13 +75,13 @@ data:
         max_concurrent_runs: 4
 
     kanban:
-      # The board is authoritative state; Herdr, invoked by the coordinator,
+      # Hermes Kanban is the only control plane. Native profiles dispatch in
-      # owns worker execution so a task cannot launch twice.
+      # the gateway; cli-* assignees are atomically claimed by the direct lane.
-      dispatch_in_gateway: false
+      dispatch_in_gateway: true
       dispatch_interval_seconds: 15
       failure_limit: 2
       orchestrator_profile: default
-      default_assignee: codex-medium
+      default_assignee: cli-auto
       max_in_progress_per_profile: 1
       auto_decompose: true
       auto_decompose_per_tick: 2
@@ -93,7 +93,6 @@ data:
 
     plugins:
       enabled:
-        - herdr-agent-state
         - auto-router
 
     skills:
@@ -168,10 +167,10 @@ data:
     Keep every project's conversation, objectives, tasks, evidence, and
     blockers in that project's Hermes Project and Kanban board. Cassandra is
     the initial project. Use native Hermes delegation as the normal planning,
-    fan-out, and synthesis path. Use Herdr only as a hidden durability adapter
+    fan-out, and synthesis path. Durable real Codex and Claude Code CLI work is
-    when work specifically benefits from a persistent real Codex or Claude Code
+    claimed directly from the same Kanban board; there is no second scheduler.
-    CLI session. You remain responsible for planning, routing, fallback, review,
+    You remain responsible for planning, routing, fallback, review, and the
-    and the final synthesized answer.
+    final synthesized answer.
 
     Prefer Codex for implementation, debugging, test loops, and focused repo
     changes. Prefer Claude Code for architecture, long-context investigation,
@@ -180,9 +179,9 @@ data:
 
     Use the browser for live or dynamic pages when search/extraction is
     insufficient. Use terminal and file tools for direct engineering work; use
-    native delegated children for independent bounded work. Use Herdr workers
+    native delegated children for independent bounded work. Use `cli-auto` board
-    only when an objective benefits from persistent Codex or Claude Code CLI
+    tasks when an objective benefits from persistent Codex or Claude Code CLI
-    execution or a long-lived CLI session that can be resumed later.
+    execution that survives browser disconnects and can resume after restarts.
 
     Local Jetson inference is the first provider-independent fallback. Use it
     for bounded classification, summaries, and continuity when hosted capacity
@@ -231,23 +230,17 @@ data:
     foreground coordinator. Do not delegate a one-tool mechanical action merely
     to create an agent.
 
-    For persistent real Codex or Claude Code CLI work, launch a worker with:
+    For persistent real Codex or Claude Code CLI work, create a bounded Kanban
-
+    worktree task assigned to `cli-auto`. The direct lane reserves the task atomically,
-    `herdr-dispatch --auto --start --project <path> --task <short-name> --prompt <bounded-objective>`
+    sends every start/retry/continuation boundary through the Jetson classifier,
-
+    records provider/model/effort and session identifiers on the task, streams
-    AUTO sends that bounded worker objective through the same Jetson classifier
+    logs into the task worker log, and resumes the provider session after a pod
-    before selecting the real CLI, model, and effort. Explicit `--shape`,
+    restart. Manual lanes are `cli-codex-{low,medium,high,xhigh}` and
-    `--effort`, and `--provider` remain available for a deliberate manual
+    `cli-claude-{low,medium,high,xhigh}`; they still call the Jetson for the audit
-    override. The command creates a separate Herdr worker space. The visible
+    record, then apply the explicit override. Observe workers through Kanban and
-    project tab remains a single Hermes coordinator pane. Never split Codex,
+    the WebUI worker list, not terminal panes. If Codex reports its first-use
-    Claude, or a second Hermes process into that tab. Worker-space labels include
+    login requirement, run `codex login --device-auth` once in `/terminal/` and
-    the parent project, provider, and task (for example
+    ask Brad to complete the displayed code.
-    `cassandra-claude-review`) so the left rail preserves ownership and purpose.
-    Keep completed workers available as labeled spaces until their evidence has
-    been synthesized; do not tile them over the coordinator. Use `herdr agent
-    list`, `herdr agent wait`, `herdr agent read`, and `herdr agent prompt` to
-    supervise them. If Codex reports its first-use login requirement, run
-    `codex login --device-auth` once and ask Brad to complete the displayed code.
 
     A hosted capacity failure should fall across providers at the same effort
     before dropping to local inference. Do not duplicate a task that is still
@@ -264,12 +257,13 @@ data:
     # Agent Hermes
 
     The authenticated root of agent.hermes.bstein.dev opens the persistent
-    browser terminal with the stock Hermes TUI in the foreground. Give Hermes
+    Hermes WebUI and its Project/Kanban controls. `/terminal/` opens the stock
+    Hermes TUI. Give Hermes
     the outcome you want and it will decompose dependent work, classify every
     delegated leaf on the Jetson, choose Codex or Claude, preserve the task on
     the Cassandra board, and synthesize the evidence. Persistent real Codex and
-    Claude Code CLI sessions run through Herdr behind that interface only when
+    Claude Code CLI sessions run as direct Kanban workers behind that interface.
-    useful. Use `/route status` to inspect the current decision, `/route auto`
+    Use `/route status` to inspect the current decision, `/route auto`
     for automatic routing, or `/route manual <codex|claude>
     <low|medium|high|xhigh> [model]` for a persistent override. The first native
     Codex worker requires one device-code login; subsequent sessions persist on

services/hermes/agent-deployment.yaml

@@ -21,10 +21,10 @@ spec:
         app: hermes-agent
       annotations:
         ai.bstein.dev/role: project-coordinator
-        ai.bstein.dev/execution: Stock Hermes TUI with durable Codex and Claude Code CLI workers
+        ai.bstein.dev/execution: Hermes Kanban with durable direct Codex and Claude Code CLI workers
         ai.bstein.dev/model-policy: Jetson-assisted AUTO routing, low through xhigh, cross-provider fallback
         ai.bstein.dev/placement: rpi5 preferred; Jetson deferred until state storage is available
-        ai.bstein.dev/config-rev: "20260810-stock-tui-per-subtask-routing"
+        ai.bstein.dev/config-rev: "20260810-native-kanban-direct-cli-lanes"
         vault.hashicorp.com/agent-inject: "true"
         vault.hashicorp.com/role: hermes-agent
         vault.hashicorp.com/agent-inject-secret-anthropic-token: kv/data/atlas/hermes/agent-tokens
@@ -87,9 +87,8 @@ spec:
               mkdir -p \
                 /opt/data/home/.claude \
                 /opt/data/home/.codex \
-                /opt/data/home/.config/herdr \
                 /opt/data/home/.kube \
-                /opt/data/herdr \
+                /opt/data/cli-lanes \
                 /opt/data/logs \
                 /opt/data/tools/bin \
                 /opt/data/workspace/coordinator \
@@ -151,14 +150,6 @@ spec:
               set -eu
               tools=/opt/data/tools
               mkdir -p "${tools}/bin"
-              herdr_version="$("${tools}/bin/herdr" --version 2>/dev/null || true)"
-              case "${herdr_version}" in *0.8.0*) herdr_ready=1 ;; *) herdr_ready=0 ;; esac
-              if [ "${herdr_ready}" != "1" ]; then
-                curl -fsSL -o "${tools}/bin/herdr.tmp" https://github.com/herdrdev/herdr/releases/download/v0.8.0/herdr-linux-aarch64
-                printf '%s  %s\n' f647ac66468d9efbc642fe534fb284468f0aea60641606fc008dfc0d82a3ca87 "${tools}/bin/herdr.tmp" | sha256sum -c -
-                chmod 0755 "${tools}/bin/herdr.tmp"
-                mv "${tools}/bin/herdr.tmp" "${tools}/bin/herdr"
-              fi
               ttyd_version="$("${tools}/bin/ttyd" --version 2>/dev/null || true)"
               case "${ttyd_version}" in *1.7.7*) ttyd_ready=1 ;; *) ttyd_ready=0 ;; esac
               if [ "${ttyd_ready}" != "1" ]; then
@@ -242,7 +233,7 @@ spec:
           resources:
             requests: {cpu: 50m, memory: 128Mi}
             limits: {cpu: 500m, memory: 512Mi}
-        - name: install-herdr-integrations
+        - name: configure-agent-clients
           image: registry.bstein.dev/bstein/hermes-agent@sha256:46de11122226a68adad6d902e81ef9b4b53deb0433dcc61f57fa43fae2e3a3a5
           imagePullPolicy: IfNotPresent
           command:
@@ -250,9 +241,7 @@ spec:
             - -ec
             - |
               /opt/hermes/.venv/bin/python /opt/coordinator/configure_agent_clients.py
-              herdr integration install codex
+              /opt/hermes/.venv/bin/python /opt/coordinator/migrate_herdr_state.py
-              herdr integration install claude
-              herdr integration install hermes
           env:
             - {name: HERMES_HOME, value: /opt/data}
             - {name: HOME, value: /opt/data/home}
@@ -306,8 +295,6 @@ spec:
             - {name: HOME, value: /opt/data/home}
             - {name: CODEX_HOME, value: /opt/data/home/.codex}
             - {name: CLAUDE_CONFIG_DIR, value: /opt/data/home/.claude}
-            - {name: HERDR_CONFIG_PATH, value: /opt/data/home/.config/herdr/config.toml}
-            - {name: HERDR_SOCKET_PATH, value: /opt/data/herdr/herdr.sock}
             - {name: KUBECONFIG, value: /opt/data/home/.kube/config}
             - {name: PATH, value: /opt/coordinator:/opt/data/tools/bin:/opt/data/home/.local/bin:/opt/hermes/.venv/bin:/opt/hermes/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin}
             - {name: PLAYWRIGHT_BROWSERS_PATH, value: /opt/hermes/.playwright}
@@ -319,12 +306,13 @@ spec:
             - {name: API_SERVER_HOST, value: 0.0.0.0}
             - {name: API_SERVER_PORT, value: "8642"}
             - {name: API_SERVER_CORS_ORIGINS, value: https://agent.hermes.bstein.dev}
+            - {name: HERMES_MEDIA_DELIVERY_STRICT, value: "1"}
+            - {name: HERMES_MEDIA_ALLOW_DIRS, value: /opt/data/workspace}
           volumeMounts:
             - {name: home, mountPath: /opt/data}
             - {name: provider-auth, mountPath: /shared-auth}
             - {name: coordinator, mountPath: /opt/coordinator, readOnly: true}
             - {name: auth-patch, mountPath: /opt/hermes/hermes_cli/auth.py, subPath: auth.py}
-            - {name: coordinator, mountPath: /opt/data/home/.local/bin/herdr-dispatch, subPath: herdr_dispatch.py, readOnly: true}
             - {name: kubeconfig, mountPath: /opt/data/home/.kube/config, subPath: config, readOnly: true}
             - {name: auto-router-plugin, mountPath: /opt/data/plugins/auto-router, readOnly: true}
           startupProbe:
@@ -408,7 +396,7 @@ spec:
           resources:
             requests: {cpu: 50m, memory: 128Mi}
             limits: {cpu: 750m, memory: 1Gi}
-        - name: herdr-tui
+        - name: terminal
           image: registry.bstein.dev/bstein/hermes-agent@sha256:46de11122226a68adad6d902e81ef9b4b53deb0433dcc61f57fa43fae2e3a3a5
           imagePullPolicy: IfNotPresent
           command: [/bin/sh, -ec]
@@ -420,6 +408,7 @@ spec:
               exec /opt/data/tools/bin/ttyd \
                 --writable \
                 --check-origin \
+                --base-path /terminal \
                 --interface 0.0.0.0 \
                 --port 7681 \
                 --cwd /opt/data/workspace \
@@ -444,15 +433,13 @@ spec:
                     done
                   '
           ports:
-            - {name: herdr-tui, containerPort: 7681, protocol: TCP}
+            - {name: terminal, containerPort: 7681, protocol: TCP}
           env:
             - {name: HOME, value: /opt/data/home}
             - {name: HERMES_HOME, value: /opt/data}
             - {name: HERMES_AUTH_FILE, value: /shared-auth/auth.json}
             - {name: CODEX_HOME, value: /opt/data/home/.codex}
             - {name: CLAUDE_CONFIG_DIR, value: /opt/data/home/.claude}
-            - {name: HERDR_CONFIG_PATH, value: /opt/data/home/.config/herdr/config.toml}
-            - {name: HERDR_SOCKET_PATH, value: /opt/data/herdr/herdr.sock}
             - {name: KUBECONFIG, value: /opt/data/home/.kube/config}
             - {name: PYTHONPATH, value: /opt/hermes}
             - {name: PATH, value: /opt/coordinator:/opt/data/tools/bin:/opt/data/home/.local/bin:/opt/hermes/.venv/bin:/usr/local/bin:/usr/bin:/bin}
@@ -464,22 +451,21 @@ spec:
             - {name: provider-auth, mountPath: /shared-auth}
             - {name: coordinator, mountPath: /opt/coordinator, readOnly: true}
             - {name: auth-patch, mountPath: /opt/hermes/hermes_cli/auth.py, subPath: auth.py}
-            - {name: coordinator, mountPath: /opt/data/home/.local/bin/herdr-dispatch, subPath: herdr_dispatch.py, readOnly: true}
             - {name: kubeconfig, mountPath: /opt/data/home/.kube/config, subPath: config, readOnly: true}
             - {name: auto-router-plugin, mountPath: /opt/data/plugins/auto-router, readOnly: true}
             - {name: tmp, mountPath: /tmp}
             - {name: ttyd-index, mountPath: /ttyd-index, readOnly: true}
           startupProbe:
-            tcpSocket: {port: herdr-tui}
+            tcpSocket: {port: terminal}
             periodSeconds: 5
             timeoutSeconds: 3
             failureThreshold: 60
           readinessProbe:
-            tcpSocket: {port: herdr-tui}
+            tcpSocket: {port: terminal}
             periodSeconds: 10
             timeoutSeconds: 3
           livenessProbe:
-            tcpSocket: {port: herdr-tui}
+            tcpSocket: {port: terminal}
             initialDelaySeconds: 30
             periodSeconds: 30
             timeoutSeconds: 5
@@ -496,61 +482,32 @@ spec:
           resources:
             requests: {cpu: 25m, memory: 64Mi}
             limits: {cpu: 500m, memory: 512Mi}
-        - name: herdr-server
+        - name: cli-lane-runner
           image: registry.bstein.dev/bstein/hermes-agent@sha256:46de11122226a68adad6d902e81ef9b4b53deb0433dcc61f57fa43fae2e3a3a5
           imagePullPolicy: IfNotPresent
-          command:
+          command: [/bin/sh, -ec]
-            - sh
+          args:
-            - -c
             - |
-              set -eu
               set -a
               . /opt/data/.env
               set +a
-              herdr server &
+              exec /opt/hermes/.venv/bin/python /opt/coordinator/cli_lane_runner.py
-              server_pid=$!
-              trap 'kill "${server_pid}" 2>/dev/null || true' TERM INT
-              agent_payload=
-              for attempt in $(seq 1 60); do
-                candidate="$(/opt/data/tools/bin/herdr agent list 2>/dev/null || true)"
-                if printf '%s' "${candidate}" | /opt/hermes/.venv/bin/python -c \
-                  'import json, sys; json.load(sys.stdin)' 2>/dev/null; then
-                  agent_payload="${candidate}"
-                  break
-                fi
-                sleep 1
-              done
-              legacy_hermes_panes="$(printf '%s' "${agent_payload}" | \
-                /opt/hermes/.venv/bin/python -c '
-              import json
-              import sys
-
-              raw = sys.stdin.read().strip()
-              payload = json.loads(raw) if raw else {}
-              for agent in payload.get("result", {}).get("agents", []):
-                  if agent.get("agent") == "hermes" and agent.get("pane_id"):
-                      print(agent["pane_id"])
-              ')"
-              for pane in ${legacy_hermes_panes}; do
-                /opt/data/tools/bin/herdr pane close "${pane}" >/dev/null 2>&1 || true
-              done
-              wait "${server_pid}"
           env:
             - {name: HERMES_HOME, value: /opt/data}
             - {name: HERMES_AUTH_FILE, value: /shared-auth/auth.json}
             - {name: HOME, value: /opt/data/home}
             - {name: CODEX_HOME, value: /opt/data/home/.codex}
             - {name: CLAUDE_CONFIG_DIR, value: /opt/data/home/.claude}
-            - {name: HERDR_CONFIG_PATH, value: /opt/data/home/.config/herdr/config.toml}
-            - {name: HERDR_SOCKET_PATH, value: /opt/data/herdr/herdr.sock}
             - {name: KUBECONFIG, value: /opt/data/home/.kube/config}
             - {name: PYTHONPATH, value: /opt/hermes}
+            - {name: HERMES_CLI_LANE_CONCURRENCY, value: "4"}
             - {name: PATH, value: /opt/coordinator:/opt/data/tools/bin:/opt/data/home/.local/bin:/opt/hermes/.venv/bin:/usr/local/bin:/usr/bin:/bin}
-            - {name: PLAYWRIGHT_BROWSERS_PATH, value: /opt/hermes/.playwright}
-            - {name: AGENT_BROWSER_EXECUTABLE_PATH, value: /opt/hermes/.playwright/chromium_headless_shell-1228/chrome-linux/headless_shell}
-            - {name: AGENT_BROWSER_ARGS, value: "--no-sandbox,--disable-dev-shm-usage"}
           securityContext:
             allowPrivilegeEscalation: false
+            capabilities:
+              drop: [ALL]
+            readOnlyRootFilesystem: true
+            runAsNonRoot: true
             runAsUser: 10000
             runAsGroup: 10000
             seccompProfile:
@@ -560,12 +517,12 @@ spec:
             - {name: provider-auth, mountPath: /shared-auth}
             - {name: coordinator, mountPath: /opt/coordinator, readOnly: true}
             - {name: auth-patch, mountPath: /opt/hermes/hermes_cli/auth.py, subPath: auth.py}
-            - {name: coordinator, mountPath: /opt/data/home/.local/bin/herdr-dispatch, subPath: herdr_dispatch.py, readOnly: true}
             - {name: kubeconfig, mountPath: /opt/data/home/.kube/config, subPath: config, readOnly: true}
             - {name: auto-router-plugin, mountPath: /opt/data/plugins/auto-router, readOnly: true}
+            - {name: tmp, mountPath: /tmp}
           resources:
-            requests: {cpu: 50m, memory: 128Mi}
+            requests: {cpu: 100m, memory: 256Mi}
-            limits: {cpu: "1", memory: 2Gi}
+            limits: {cpu: "3", memory: 6Gi}
         - name: model-steward
           image: registry.bstein.dev/bstein/hermes-agent@sha256:46de11122226a68adad6d902e81ef9b4b53deb0433dcc61f57fa43fae2e3a3a5
           imagePullPolicy: IfNotPresent

services/hermes/agent-ingress.yaml

@@ -1,4 +1,15 @@
 # services/hermes/agent-ingress.yaml
+apiVersion: traefik.io/v1alpha1
+kind: Middleware
+metadata:
+  name: hermes-agent-terminal-slash
+  namespace: hermes
+spec:
+  redirectRegex:
+    regex: ^https://agent\.hermes\.bstein\.dev/terminal$
+    replacement: https://agent.hermes.bstein.dev/terminal/
+    permanent: false
+---
 apiVersion: networking.k8s.io/v1
 kind: Ingress
 metadata:
@@ -7,6 +18,7 @@ metadata:
   annotations:
     cert-manager.io/cluster-issuer: letsencrypt
     traefik.ingress.kubernetes.io/router.entrypoints: websecure
+    traefik.ingress.kubernetes.io/router.middlewares: hermes-hermes-agent-terminal-slash@kubernetescrd
     traefik.ingress.kubernetes.io/router.tls: "true"
 spec:
   ingressClassName: traefik

services/hermes/kustomization.yaml

@@ -46,14 +46,15 @@ configMapGenerator:
     namespace: hermes
     files:
       - claude=scripts/claude
+      - claude_command_policy.py=scripts/claude_command_policy.py
+      - cli_lane_runner.py=scripts/cli_lane_runner.py
       - codex=scripts/codex
       - configure_agent_clients.py=scripts/configure_agent_clients.py
       - gitea_askpass.sh=scripts/gitea_askpass.sh
-      - herdr_dispatch.py=scripts/herdr_dispatch.py
-      - herdr_tab_router.py=scripts/herdr_tab_router.py
       - hermes_coordinator.py=scripts/hermes_coordinator.py
       - hermes_model_routing.py=scripts/hermes_model_routing.py
       - hermes_stt_client.py=scripts/hermes_stt_client.py
+      - migrate_herdr_state.py=scripts/migrate_herdr_state.py
       - patch_hermes_auth.py=scripts/patch_hermes_auth.py
       - patch_ttyd_index.py=scripts/patch_ttyd_index.py
     options:

services/hermes/networkpolicy.yaml

@@ -73,6 +73,7 @@ spec:
               app: oauth2-proxy-hermes-agent
       ports:
         - {protocol: TCP, port: 7681}
+        - {protocol: TCP, port: 8787}
   egress:
     - to:
         - namespaceSelector:

services/hermes/oauth2-proxy.yaml

@@ -82,6 +82,8 @@ spec:
             - --redirect-url=https://agent.hermes.bstein.dev/oauth2/callback
             - --oidc-issuer-url=https://sso.bstein.dev/realms/atlas
             - --user-id-claim=sub
+            - --oidc-groups-claim=groups
+            - --allowed-group=/hermes-owner
             - --code-challenge-method=S256
             - --scope=openid profile email
             - --email-domain=*
@@ -99,7 +101,8 @@ spec:
             - --cookie-csrf-per-request-limit=8
             - --cookie-refresh=1h
             - --cookie-expire=8h
-            - --upstream=http://hermes-agent.hermes.svc.cluster.local:7681
+            - --upstream=http://hermes-agent.hermes.svc.cluster.local:7681/terminal/
+            - --upstream=http://hermes-agent.hermes.svc.cluster.local:8787/
             - --http-address=0.0.0.0:4180
             - --skip-provider-button=true
             - --reverse-proxy=true

services/hermes/plugins/auto-router/__init__.py

@@ -601,7 +601,7 @@ def _record_subagent_plan(
 
 def _runtime_agent(ctx: Any) -> Any | None:
     """Return the active agent without assuming a single CLI lifecycle."""
-    cli = getattr(ctx._manager, "_cli_ref", None)
+    cli = getattr(getattr(ctx, "_manager", None), "_cli_ref", None)
     return getattr(cli, "agent", None) if cli is not None else None
 
 
@@ -656,6 +656,7 @@ def _pre_turn_route(ctx: Any, **kwargs: Any) -> None:
     if agent is None or not text or text.startswith("/"):
         return
     if policy["mode"] == "manual":
+        audit = classify_task(text, kwargs.get("conversation_history"))
         manual = policy.get("manual") or {}
         provider = str(manual.get("provider") or "")
         effort = str(manual.get("effort") or "")
@@ -667,7 +668,12 @@ def _pre_turn_route(ctx: Any, **kwargs: Any) -> None:
             plan = select_route(_load_json(ROUTING_PATH), decision)
         else:
             decision = Decision(
-                "question", effort, provider, "manual", "explicit user override"
+                audit.shape,
+                effort,
+                provider,
+                f"manual-{audit.classifier}",
+                f"explicit user override; Jetson audit suggested {audit.provider}/{audit.effort}",
+                audit.latency_ms,
             )
             plan = select_route(_load_json(ROUTING_PATH), decision, model)
     else:
@@ -677,7 +683,7 @@ def _pre_turn_route(ctx: Any, **kwargs: Any) -> None:
     _record_plan(policy, plan)
     emit = getattr(agent, "_emit_status", None)
     if callable(emit):
-        if plan["classifier"] == "manual":
+        if str(plan["classifier"]).startswith("manual"):
             emit(
                 f"MANUAL target → {plan['provider']}/{plan['model']} · "
                 f"{plan['effort']} · automatic capacity fallback remains enabled"
@@ -696,10 +702,8 @@ def _pre_turn_route(ctx: Any, **kwargs: Any) -> None:
 
 
 def _pre_internal_route(ctx: Any, **kwargs: Any) -> None:
-    """Reclassify every AUTO tool-loop continuation before request building."""
+    """Classify every tool-loop continuation before request building."""
     policy = _current_policy()
-    if policy["mode"] != "auto":
-        return
     agent = kwargs.get("agent") or _runtime_agent(ctx)
     if agent is None:
         return
@@ -710,16 +714,33 @@ def _pre_internal_route(ctx: Any, **kwargs: Any) -> None:
     if not text.strip():
         return
 
-    decision = classify_task(text)
+    audit = classify_task(text)
-    decision = Decision(
+    if policy["mode"] == "manual":
-        decision.shape,
+        manual = policy.get("manual") or {}
-        decision.effort,
+        provider = str(manual.get("provider") or "")
-        decision.provider,
+        effort = str(manual.get("effort") or "")
-        f"{decision.classifier}-internal",
+        model = str(manual.get("model") or "")
-        f"{decision.reason}; reclassified for the next internal prompt",
+        if provider not in PROVIDERS or effort not in EFFORTS:
-        decision.latency_ms,
+            return
-    )
+        decision = Decision(
-    plan = select_route(_load_json(ROUTING_PATH), decision)
+            audit.shape,
+            effort,
+            provider,
+            f"manual-{audit.classifier}-internal",
+            f"explicit user override; Jetson internal audit suggested {audit.provider}/{audit.effort}",
+            audit.latency_ms,
+        )
+    else:
+        model = ""
+        decision = Decision(
+            audit.shape,
+            audit.effort,
+            audit.provider,
+            f"{audit.classifier}-internal",
+            f"{audit.reason}; reclassified for the next internal prompt",
+            audit.latency_ms,
+        )
+    plan = select_route(_load_json(ROUTING_PATH), decision, model)
     previous_effort = str(
         (getattr(agent, "reasoning_config", None) or {}).get("effort") or ""
     )
@@ -735,7 +756,7 @@ def _pre_internal_route(ctx: Any, **kwargs: Any) -> None:
     emit = getattr(agent, "_emit_status", None)
     if changed and callable(emit):
         emit(
-            f"AUTO internal #{api_call_count} → "
+            f"{policy['mode'].upper()} internal #{api_call_count} → "
             f"{plan['provider']}/{plan['model']} · {plan['effort']} (Jetson)"
         )
 
@@ -743,8 +764,6 @@ def _pre_internal_route(ctx: Any, **kwargs: Any) -> None:
 def _pre_subagent_route(ctx: Any, **kwargs: Any) -> None:
     """Classify and route each native Hermes child before it starts work."""
     policy = _current_policy()
-    if policy["mode"] != "auto":
-        return
     child = kwargs.get("agent")
     goal = str(kwargs.get("goal") or "").strip()
     context = str(kwargs.get("context") or "").strip()
@@ -754,16 +773,33 @@ def _pre_subagent_route(ctx: Any, **kwargs: Any) -> None:
     task_text = goal
     if context:
         task_text += f"\n\nDelegated context:\n{context[-6000:]}"
-    decision = classify_task(task_text)
+    audit = classify_task(task_text)
-    decision = Decision(
+    if policy["mode"] == "manual":
-        decision.shape,
+        manual = policy.get("manual") or {}
-        decision.effort,
+        provider = str(manual.get("provider") or "")
-        decision.provider,
+        effort = str(manual.get("effort") or "")
-        f"{decision.classifier}-subagent",
+        model = str(manual.get("model") or "")
-        f"{decision.reason}; independently classified delegated task",
+        if provider not in PROVIDERS or effort not in EFFORTS:
-        decision.latency_ms,
+            return
-    )
+        decision = Decision(
-    plan = select_route(_load_json(ROUTING_PATH), decision)
+            audit.shape,
+            effort,
+            provider,
+            f"manual-{audit.classifier}-subagent",
+            f"explicit user override; Jetson child audit suggested {audit.provider}/{audit.effort}",
+            audit.latency_ms,
+        )
+    else:
+        model = ""
+        decision = Decision(
+            audit.shape,
+            audit.effort,
+            audit.provider,
+            f"{audit.classifier}-subagent",
+            f"{audit.reason}; independently classified delegated task",
+            audit.latency_ms,
+        )
+    plan = select_route(_load_json(ROUTING_PATH), decision, model)
     _apply_route(ctx, child, plan)
     task_index = int(kwargs.get("task_index") or 0)
     _record_subagent_plan(policy, plan, goal, task_index)
@@ -772,7 +808,7 @@ def _pre_subagent_route(ctx: Any, **kwargs: Any) -> None:
     emit = getattr(parent, "_emit_status", None)
     if callable(emit):
         emit(
-            f"AUTO child #{task_index + 1} → "
+            f"{policy['mode'].upper()} child #{task_index + 1} → "
             f"{plan['provider']}/{plan['model']} · {plan['effort']} (Jetson)"
         )
 

services/hermes/scripts/claude

@@ -4,6 +4,6 @@ set -eu
 
 exec /opt/data/tools/bin/claude \
   --dangerously-skip-permissions \
-  --permission-mode bypassPermissions \
   --autocompact auto \
+  --settings /opt/data/home/.claude/settings.json \
   "$@"

services/hermes/scripts/claude_command_policy.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+"""Block known-dangerous shell mutations in unattended Claude workers."""
+
+from __future__ import annotations
+
+import json
+import re
+import sys
+from typing import Any
+
+
+DENIED_COMMANDS: tuple[tuple[str, re.Pattern[str]], ...] = (
+    ("Kubernetes mutations are delivered through Flux", re.compile(r"(?:^|[;&|]\s*)kubectl\s+(?:apply|delete|patch|scale|exec|port-forward)\b", re.I)),
+    ("Flux control operations require an explicit operator action", re.compile(r"(?:^|[;&|]\s*)flux\s+(?:reconcile|suspend|resume)\b", re.I)),
+    ("Vault data access is outside the worker boundary", re.compile(r"(?:^|[;&|]\s*)vault\s+kv\b", re.I)),
+    ("force-push is prohibited", re.compile(r"(?:^|[;&|]\s*)git\s+push\b[^\n]*(?:--force(?:-with-lease)?|(?:^|\s)-f(?:\s|$))", re.I)),
+    ("destructive Git reset is prohibited", re.compile(r"(?:^|[;&|]\s*)git\s+reset\b[^\n]*--hard\b", re.I)),
+    ("destructive Git clean is prohibited", re.compile(r"(?:^|[;&|]\s*)git\s+clean\b[^\n]*(?:-[a-z]*f[a-z]*|--force)\b", re.I)),
+)
+
+
+def command_from_payload(payload: dict[str, Any]) -> str:
+    """Extract the Bash command from a Claude PreToolUse hook payload."""
+    tool_input = payload.get("tool_input")
+    if not isinstance(tool_input, dict):
+        return ""
+    value = tool_input.get("command")
+    return value if isinstance(value, str) else ""
+
+
+def denial_reason(command: str) -> str | None:
+    """Return a concise block reason for a denied command."""
+    for reason, pattern in DENIED_COMMANDS:
+        if pattern.search(command):
+            return reason
+    return None
+
+
+def main() -> int:
+    """Apply the policy to one Claude hook event."""
+    try:
+        payload = json.load(sys.stdin)
+    except (json.JSONDecodeError, OSError):
+        print("Claude command policy received invalid JSON", file=sys.stderr)
+        return 2
+    if not isinstance(payload, dict):
+        print("Claude command policy received an invalid event", file=sys.stderr)
+        return 2
+    command = command_from_payload(payload)
+    reason = denial_reason(command)
+    if reason:
+        print(f"Blocked by the Hermes owner-workspace policy: {reason}.", file=sys.stderr)
+        return 2
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/hermes/scripts/cli_lane_runner.py

@@ -0,0 +1,844 @@
+#!/usr/bin/env python3
+"""Run durable Codex and Claude workers from Hermes' authoritative Kanban."""
+
+from __future__ import annotations
+
+import concurrent.futures
+import importlib.util
+import json
+import os
+import re
+import selectors
+import subprocess
+import sys
+import threading
+import time
+import uuid
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Callable
+
+
+DATA_ROOT = Path(os.environ.get("HERMES_HOME", "/opt/data"))
+ROUTING_PATH = DATA_ROOT / "workspace/coordinator/model-routing.json"
+ROUTER_PATH = DATA_ROOT / "plugins/auto-router/__init__.py"
+STATE_ROOT = DATA_ROOT / "cli-lanes"
+CODEX_BIN = DATA_ROOT / "tools/bin/codex"
+CLAUDE_BIN = DATA_ROOT / "tools/bin/claude"
+CLAUDE_SETTINGS = DATA_ROOT / "home/.claude/settings.json"
+RESULT_SCHEMA_PATH = STATE_ROOT / "worker-result.schema.json"
+EFFORTS = ("low", "medium", "high", "xhigh")
+EXTERNAL_PREFIX = "cli-"
+DEFAULT_CLAIM_TTL = 7 * 24 * 60 * 60
+DEFAULT_MAX_RUNTIME = 12 * 60 * 60
+HEARTBEAT_SECONDS = 20
+WORKTREE_LOCK = threading.Lock()
+CAPACITY_PATTERN = re.compile(
+    r"(?:rate.?limit|capacity|overload|usage.?limit|quota|credit|exhaust|429|529|authentication|oauth|token.*expired)",
+    re.I,
+)
+NO_CLAUDE_SESSION = "No conversation found with session ID:"
+CLAUDE_SESSION_COLLISION = "Session ID already in use"
+
+
+RESULT_SCHEMA: dict[str, Any] = {
+    "type": "object",
+    "additionalProperties": False,
+    "required": ["status", "summary", "changed_files", "tests_run", "artifacts", "blockers"],
+    "properties": {
+        "status": {"type": "string", "enum": ["completed", "blocked"]},
+        "summary": {"type": "string"},
+        "changed_files": {"type": "array", "items": {"type": "string"}},
+        "tests_run": {"type": "array", "items": {"type": "string"}},
+        "artifacts": {"type": "array", "items": {"type": "string"}},
+        "blockers": {"type": "array", "items": {"type": "string"}},
+    },
+}
+
+
+@dataclass(frozen=True)
+class Route:
+    """One provider/model/effort decision for an observable worker boundary."""
+
+    provider: str
+    model: str
+    effort: str
+    profile: str
+    classifier: str
+    reason: str
+    latency_ms: int
+    fallback_chain: tuple[str, ...]
+
+
+@dataclass
+class ProcessResult:
+    """Captured outcome from one provider CLI invocation."""
+
+    returncode: int
+    output: str
+    structured: dict[str, Any] | None
+    capacity_failure: bool
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def atomic_json(path: Path, value: dict[str, Any], mode: int = 0o600) -> None:
+    """Durably replace a small non-secret state document."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
+    temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    temporary.chmod(mode)
+    os.replace(temporary, path)
+
+
+def load_json(path: Path) -> dict[str, Any]:
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return {}
+    return value if isinstance(value, dict) else {}
+
+
+def _load_router(path: Path = ROUTER_PATH) -> Any:
+    """Load the same Jetson-first classifier used by interactive Hermes."""
+    spec = importlib.util.spec_from_file_location("hermes_cli_lane_router", path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"AUTO router could not be loaded: {path}")
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def parse_assignee(assignee: str) -> tuple[str | None, str | None]:
+    """Parse external lane overrides while leaving cli-auto fully automatic."""
+    value = str(assignee or "").strip().lower()
+    if value == "cli-auto":
+        return None, None
+    match = re.fullmatch(r"cli-(codex|claude)-(low|medium|high|xhigh)", value)
+    if not match:
+        raise ValueError(f"unsupported external lane assignee: {assignee}")
+    return match.group(1), match.group(2)
+
+
+def _split_route(value: str) -> tuple[str, str]:
+    provider, separator, model = value.partition("/")
+    if not separator or not provider or not model:
+        raise RuntimeError(f"invalid managed route: {value}")
+    return provider, model
+
+
+def _worker_provider(provider: str) -> str | None:
+    return {"openai-codex": "codex", "anthropic": "claude"}.get(provider)
+
+
+def select_route(
+    prompt: str,
+    assignee: str,
+    *,
+    routing_path: Path = ROUTING_PATH,
+    router_path: Path = ROUTER_PATH,
+    exclude_provider: str | None = None,
+) -> Route:
+    """Always consult the Jetson, then apply a deliberate lane override if set."""
+    router = _load_router(router_path)
+    decision = router.classify_task(prompt)
+    manual_provider, manual_effort = parse_assignee(assignee)
+    selected_provider = manual_provider or str(decision.provider)
+    effort = manual_effort or str(decision.effort)
+    if effort not in EFFORTS:
+        raise RuntimeError(f"classifier returned unsupported effort: {effort}")
+    if exclude_provider == selected_provider:
+        selected_provider = "claude" if selected_provider == "codex" else "codex"
+    status = load_json(routing_path)
+    profile = f"{selected_provider}-{effort}"
+    chain = (status.get("routes") or {}).get(profile)
+    if not isinstance(chain, list) or not chain:
+        raise RuntimeError(f"managed route unavailable: {profile}")
+    hosted = [str(item) for item in chain if _worker_provider(_split_route(str(item))[0])]
+    if not hosted:
+        raise RuntimeError(f"no hosted CLI route available: {profile}")
+    first = next(
+        (item for item in hosted if _worker_provider(_split_route(item)[0]) == selected_provider),
+        hosted[0],
+    )
+    provider_key, model = _split_route(first)
+    actual_provider = _worker_provider(provider_key)
+    if actual_provider is None:
+        raise RuntimeError(f"route is not a CLI provider: {first}")
+    return Route(
+        provider=actual_provider,
+        model=model,
+        effort=effort,
+        profile=f"{actual_provider}-{effort}",
+        classifier=str(decision.classifier),
+        reason=(
+            str(decision.reason)
+            + ("; manual lane override applied after Jetson classification" if manual_provider else "")
+        ),
+        latency_ms=int(decision.latency_ms),
+        fallback_chain=tuple(item for item in hosted if item != first),
+    )
+
+
+def state_path(board: str, task_id: str) -> Path:
+    safe_board = re.sub(r"[^a-zA-Z0-9_.-]+", "-", board)
+    safe_task = re.sub(r"[^a-zA-Z0-9_.-]+", "-", task_id)
+    return STATE_ROOT / safe_board / f"{safe_task}.json"
+
+
+def build_prompt(context: str, workspace: Path, handoff: str = "") -> str:
+    """Create a bounded worker contract with an explicit machine-readable result."""
+    return f"""You are a durable coding worker managed by Hermes Kanban.
+
+Work only on this objective and its acceptance criteria:
+
+{context}
+
+Workspace: {workspace}
+{handoff}
+
+Operate autonomously inside the workspace. Inspect before editing, preserve unrelated user changes, run proportionate tests, and do not claim completion without evidence. Kubernetes mutations must be delivered through Flux-tracked repository manifests; cluster access is read-only. Do not force-push, hard-reset, clean untracked files, read Vault values, or expose credentials.
+
+Return a final JSON object matching the supplied schema. Use status=blocked only for a concrete unresolved blocker. List changed files, tests run, durable artifact paths, and blockers explicitly.
+"""
+
+
+def git_handoff(workspace: Path, prior_output: str) -> str:
+    """Build an explicit cross-provider handoff without transferring hidden state."""
+    def read_git(*args: str) -> str:
+        completed = subprocess.run(
+            ["git", "-C", str(workspace), *args],
+            text=True,
+            capture_output=True,
+            timeout=30,
+            check=False,
+        )
+        return (completed.stdout or completed.stderr).strip()[-12000:]
+
+    return (
+        "\nCross-provider handoff from a failed or exhausted worker:\n"
+        f"Git status:\n{read_git('status', '--short', '--branch')}\n\n"
+        f"Current diff summary:\n{read_git('diff', '--stat')}\n\n"
+        f"Prior worker tail:\n{prior_output[-10000:]}\n"
+        "Reinspect the workspace and verify all inherited claims before continuing."
+    )
+
+
+def workspace_artifacts(workspace: Path, values: Any) -> list[str]:
+    """Return existing regular artifacts contained by the task worktree."""
+    if not isinstance(values, list):
+        return []
+    root = workspace.resolve()
+    artifacts: list[str] = []
+    for value in values:
+        if not isinstance(value, str) or not value.strip():
+            continue
+        candidate = Path(value.strip()).expanduser()
+        if not candidate.is_absolute():
+            candidate = root / candidate
+        try:
+            resolved = candidate.resolve(strict=True)
+            resolved.relative_to(root)
+        except (OSError, RuntimeError, ValueError):
+            continue
+        if resolved.is_file():
+            artifacts.append(str(resolved))
+    return list(dict.fromkeys(artifacts))
+
+
+def _extract_json(value: Any) -> dict[str, Any] | None:
+    if isinstance(value, dict) and value.get("status") in {"completed", "blocked"}:
+        return value
+    if not isinstance(value, str):
+        return None
+    candidates = [value]
+    match = re.search(r"\{.*\}", value, re.DOTALL)
+    if match:
+        candidates.append(match.group(0))
+    for candidate in candidates:
+        try:
+            parsed = json.loads(candidate)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(parsed, dict) and parsed.get("status") in {"completed", "blocked"}:
+            return parsed
+    return None
+
+
+def _event_payload(provider: str, line: str, state: dict[str, Any], state_file: Path) -> dict[str, Any] | None:
+    """Persist provider session identifiers before interpreting final output."""
+    try:
+        event = json.loads(line)
+    except json.JSONDecodeError:
+        return None
+    if not isinstance(event, dict):
+        return None
+    if provider == "codex" and event.get("type") == "thread.started":
+        thread_id = event.get("thread_id") or event.get("thread", {}).get("id")
+        if thread_id:
+            state["codex_thread_id"] = str(thread_id)
+            atomic_json(state_file, state)
+    if provider == "claude" and event.get("session_id"):
+        state["claude_session_id"] = str(event["session_id"])
+        atomic_json(state_file, state)
+    for key in ("structured_output", "result", "output", "text"):
+        parsed = _extract_json(event.get(key))
+        if parsed:
+            return parsed
+    item = event.get("item")
+    if isinstance(item, dict):
+        for key in ("text", "content"):
+            parsed = _extract_json(item.get(key))
+            if parsed:
+                return parsed
+    return None
+
+
+def stream_process(
+    command: list[str],
+    *,
+    provider: str,
+    cwd: Path,
+    env: dict[str, str],
+    log_path: Path,
+    state: dict[str, Any],
+    state_file: Path,
+    heartbeat: Callable[[str], bool],
+    max_runtime: int,
+) -> ProcessResult:
+    """Stream JSONL to Kanban logs while maintaining the authoritative lease."""
+    process = subprocess.Popen(
+        command,
+        cwd=cwd,
+        env=env,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        bufsize=1,
+    )
+    assert process.stdout is not None
+    selector = selectors.DefaultSelector()
+    selector.register(process.stdout, selectors.EVENT_READ)
+    started = time.monotonic()
+    last_heartbeat = 0.0
+    lines: list[str] = []
+    structured: dict[str, Any] | None = None
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+    forced_failure = ""
+    with log_path.open("a", encoding="utf-8") as log:
+        log.write(f"\n[{utc_now()}] starting {provider} worker\n")
+        log.flush()
+        while process.poll() is None:
+            now = time.monotonic()
+            if now - started > max_runtime:
+                process.terminate()
+                forced_failure = "worker exceeded its maximum runtime"
+                lines.append(forced_failure + "\n")
+                break
+            if now - last_heartbeat >= HEARTBEAT_SECONDS:
+                if not heartbeat(f"{provider} worker active for {round(now - started)}s"):
+                    process.terminate()
+                    forced_failure = "Kanban lease was lost; provider process terminated"
+                    lines.append(forced_failure + "\n")
+                    break
+                last_heartbeat = now
+            for key, _ in selector.select(timeout=1.0):
+                line = key.fileobj.readline()
+                if not line:
+                    continue
+                lines.append(line)
+                if len(lines) > 4000:
+                    lines = lines[-4000:]
+                log.write(line)
+                log.flush()
+                parsed = _event_payload(provider, line, state, state_file)
+                structured = parsed or structured
+        if process.poll() is None:
+            try:
+                process.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait(timeout=10)
+        remainder = process.stdout.read()
+        if remainder:
+            lines.append(remainder)
+            log.write(remainder)
+            for line in remainder.splitlines():
+                parsed = _event_payload(provider, line, state, state_file)
+                structured = parsed or structured
+        if forced_failure:
+            log.write(forced_failure + "\n")
+        log.write(f"\n[{utc_now()}] {provider} exit={process.returncode}\n")
+    selector.close()
+    output = "".join(lines)[-100000:]
+    returncode = int(process.returncode if process.returncode is not None else 1)
+    return ProcessResult(
+        returncode=returncode,
+        output=output,
+        structured=structured,
+        capacity_failure=returncode != 0 and bool(CAPACITY_PATTERN.search(output)),
+    )
+
+
+def _base_env() -> dict[str, str]:
+    env = os.environ.copy()
+    env.update(
+        {
+            "HOME": str(DATA_ROOT / "home"),
+            "CODEX_HOME": str(DATA_ROOT / "home/.codex"),
+            "CLAUDE_CONFIG_DIR": str(DATA_ROOT / "home/.claude"),
+            "GIT_TERMINAL_PROMPT": "0",
+        }
+    )
+    env["PATH"] = f"{DATA_ROOT / 'tools/bin'}:/opt/coordinator:" + env.get("PATH", "")
+    return env
+
+
+def _codex_command(route: Route, prompt: str, workspace: Path, state: dict[str, Any], result_file: Path) -> list[str]:
+    base = [
+        str(CODEX_BIN),
+        "--dangerously-bypass-approvals-and-sandbox",
+        "--dangerously-bypass-hook-trust",
+        "exec",
+    ]
+    thread_id = str(state.get("codex_thread_id") or "")
+    common = [
+        "--json",
+        "-m",
+        route.model,
+        "-c",
+        f'model_reasoning_effort="{route.effort}"',
+        "--output-schema",
+        str(RESULT_SCHEMA_PATH),
+        "-o",
+        str(result_file),
+    ]
+    if thread_id:
+        return [*base, "resume", *common, thread_id, prompt]
+    return [
+        *base,
+        *common,
+        "-C",
+        str(workspace),
+        prompt,
+    ]
+
+
+def _claude_command(route: Route, prompt: str, state: dict[str, Any], resume: bool) -> list[str]:
+    session_id = str(state["claude_session_id"])
+    session = ["--resume", session_id] if resume else ["--session-id", session_id]
+    denied = [
+        "Bash(kubectl apply *)",
+        "Bash(kubectl delete *)",
+        "Bash(kubectl patch *)",
+        "Bash(kubectl scale *)",
+        "Bash(kubectl exec *)",
+        "Bash(kubectl port-forward *)",
+        "Bash(flux reconcile *)",
+        "Bash(flux suspend *)",
+        "Bash(flux resume *)",
+        "Bash(vault kv *)",
+        "Bash(git push --force *)",
+        "Bash(git reset --hard *)",
+        "Bash(git clean -f *)",
+    ]
+    return [
+        str(CLAUDE_BIN),
+        "--dangerously-skip-permissions",
+        "--autocompact",
+        "auto",
+        "--settings",
+        str(CLAUDE_SETTINGS),
+        "--disallowedTools",
+        *denied,
+        "--model",
+        route.model,
+        "--effort",
+        route.effort,
+        "--output-format",
+        "stream-json",
+        "--verbose",
+        "--json-schema",
+        json.dumps(RESULT_SCHEMA, separators=(",", ":")),
+        *session,
+        "-p",
+        prompt,
+    ]
+
+
+def run_provider(
+    route: Route,
+    prompt: str,
+    workspace: Path,
+    state: dict[str, Any],
+    state_file: Path,
+    log_path: Path,
+    heartbeat: Callable[[str], bool],
+    max_runtime: int,
+) -> ProcessResult:
+    """Run or resume one provider session using its pinned structured CLI."""
+    state.setdefault("attempts", []).append({"route": asdict(route), "started_at": utc_now()})
+    if route.provider == "claude" and not state.get("claude_session_id"):
+        # The reservation is durable before Claude starts, closing the crash gap.
+        state["claude_session_id"] = str(uuid.uuid4())
+    state["current_route"] = asdict(route)
+    state["updated_at"] = utc_now()
+    atomic_json(state_file, state)
+    env = _base_env()
+    result_file = state_file.with_suffix(".result.json")
+    if route.provider == "codex":
+        result_file.unlink(missing_ok=True)
+        command = _codex_command(route, prompt, workspace, state, result_file)
+        result = stream_process(
+            command,
+            provider="codex",
+            cwd=workspace,
+            env=env,
+            log_path=log_path,
+            state=state,
+            state_file=state_file,
+            heartbeat=heartbeat,
+            max_runtime=max_runtime,
+        )
+        file_result = load_json(result_file)
+        if file_result.get("status") in {"completed", "blocked"}:
+            result.structured = file_result
+        return result
+
+    resume = bool(state.get("claude_started"))
+    result = stream_process(
+        _claude_command(route, prompt, state, resume),
+        provider="claude",
+        cwd=workspace,
+        env=env,
+        log_path=log_path,
+        state=state,
+        state_file=state_file,
+        heartbeat=heartbeat,
+        max_runtime=max_runtime,
+    )
+    if resume and result.returncode == 1 and NO_CLAUDE_SESSION in result.output:
+        result = stream_process(
+            _claude_command(route, prompt, state, False),
+            provider="claude",
+            cwd=workspace,
+            env=env,
+            log_path=log_path,
+            state=state,
+            state_file=state_file,
+            heartbeat=heartbeat,
+            max_runtime=max_runtime,
+        )
+    elif not resume and result.returncode == 1 and CLAUDE_SESSION_COLLISION in result.output:
+        result = stream_process(
+            _claude_command(route, prompt, state, True),
+            provider="claude",
+            cwd=workspace,
+            env=env,
+            log_path=log_path,
+            state=state,
+            state_file=state_file,
+            heartbeat=heartbeat,
+            max_runtime=max_runtime,
+        )
+    if result.returncode == 0 or NO_CLAUDE_SESSION not in result.output:
+        state["claude_started"] = True
+        atomic_json(state_file, state)
+    return result
+
+
+def _task_value(task: Any, name: str, default: Any = None) -> Any:
+    return getattr(task, name, default)
+
+
+def _task_context(kanban_db: Any, conn: Any, task_id: str) -> str:
+    value = kanban_db.build_worker_context(conn, task_id)
+    if isinstance(value, str):
+        return value
+    return json.dumps(value, indent=2, default=str)
+
+
+def _resolve_workspace(kanban_db: Any, conn: Any, task: Any, board: str) -> Path:
+    # External coding lanes always receive their own linked worktree. A task
+    # that does not resolve to a Git repository is blocked rather than sharing
+    # a mutable checkout with another unattended worker.
+    # Git worktree creation updates common-repository metadata. Serialize that
+    # short materialization step while allowing the provider workers themselves
+    # to run concurrently in independent worktrees.
+    with WORKTREE_LOCK:
+        value, branch_name = kanban_db._resolve_worktree_workspace(task, board=board)
+    kanban_db.set_branch_name(conn, str(_task_value(task, "id")), branch_name)
+    return Path(value).resolve()
+
+
+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
+
+    with kanban_db.scoped_current_board(board):
+        conn = kanban_db.connect(board=board)
+        task = kanban_db.get_task(conn, task_id)
+        if task is None:
+            return
+        run_id = _task_value(task, "current_run_id", None)
+        assignee = str(_task_value(task, "assignee", "cli-auto") or "cli-auto")
+        try:
+            state_file = state_path(board, task_id)
+            state = load_json(state_file)
+            state.update(
+                {
+                    "board": board,
+                    "task_id": task_id,
+                    "run_id": run_id,
+                    "assignee": assignee,
+                }
+            )
+            atomic_json(state_file, state)
+            log_path = Path(kanban_db.worker_log_path(task_id, board=board))
+            workspace = _resolve_workspace(kanban_db, conn, task, board)
+            kanban_db.set_workspace_path(conn, task_id, str(workspace))
+            context = _task_context(kanban_db, conn, task_id)
+        except Exception as error:
+            kanban_db.block_task(
+                conn,
+                task_id,
+                reason=f"Direct CLI lane preparation failed: {type(error).__name__}: {error}",
+                kind="capability",
+                expected_run_id=run_id,
+            )
+            conn.close()
+            return
+
+        def heartbeat(note: str) -> bool:
+            return bool(
+                kanban_db.heartbeat_worker(
+                    conn,
+                    task_id,
+                    note=note,
+                    expected_run_id=run_id,
+                )
+            )
+
+        try:
+            previous_route = state.get("current_route")
+            route = select_route(context, assignee)
+            kanban_db.add_comment(
+                conn,
+                task_id,
+                "cli-lane-runner",
+                f"CLI route: {route.provider}/{route.model} at {route.effort}; classifier={route.classifier}; {route.reason}",
+            )
+            resume_handoff = ""
+            if (
+                isinstance(previous_route, dict)
+                and previous_route.get("provider")
+                and previous_route.get("provider") != route.provider
+            ):
+                try:
+                    previous_output = log_path.read_text(encoding="utf-8")[-10000:]
+                except OSError:
+                    previous_output = "Previous provider log was unavailable after restart."
+                resume_handoff = git_handoff(workspace, previous_output)
+                kanban_db.add_comment(
+                    conn,
+                    task_id,
+                    "cli-lane-runner",
+                    f"Restart-time provider change: {previous_route.get('provider')} -> {route.provider}; explicit workspace handoff attached.",
+                )
+            prompt = build_prompt(context, workspace, resume_handoff)
+            max_runtime = int(
+                _task_value(task, "max_runtime_seconds", 0) or DEFAULT_MAX_RUNTIME
+            )
+            result = run_provider(route, prompt, workspace, state, state_file, log_path, heartbeat, max_runtime)
+            if result.capacity_failure:
+                retry_context = (
+                    context
+                    + "\n\nRouting boundary: the first provider failed from capacity/authentication. "
+                    + "Select the alternate hosted provider at an appropriate effort."
+                )
+                alternate = "claude" if route.provider == "codex" else "codex"
+                fallback = select_route(
+                    retry_context,
+                    f"cli-{alternate}-{route.effort}",
+                )
+                kanban_db.add_comment(
+                    conn,
+                    task_id,
+                    "cli-lane-runner",
+                    f"Provider fallback: {route.provider} -> {fallback.provider}; Jetson reclassified the retry boundary.",
+                )
+                result = run_provider(
+                    fallback,
+                    build_prompt(context, workspace, git_handoff(workspace, result.output)),
+                    workspace,
+                    state,
+                    state_file,
+                    log_path,
+                    heartbeat,
+                    max_runtime,
+                )
+                route = fallback
+            structured = result.structured
+            if structured:
+                structured = dict(structured)
+                structured["artifacts"] = workspace_artifacts(
+                    workspace,
+                    structured.get("artifacts"),
+                )
+            metadata = {
+                "executor": "direct-cli-lane",
+                "provider": route.provider,
+                "model": route.model,
+                "effort": route.effort,
+                "classifier": route.classifier,
+                "state_file": str(state_file),
+                "codex_thread_id": state.get("codex_thread_id"),
+                "claude_session_id": state.get("claude_session_id"),
+            }
+            if structured:
+                for key in ("changed_files", "tests_run", "artifacts", "blockers"):
+                    value = structured.get(key)
+                    metadata[key] = value if isinstance(value, list) else []
+            if structured and structured.get("status") == "completed" and result.returncode == 0:
+                kanban_db.complete_task(
+                    conn,
+                    task_id,
+                    result=json.dumps(structured, sort_keys=True),
+                    summary=str(structured.get("summary") or "Completed"),
+                    metadata=metadata,
+                    expected_run_id=run_id,
+                )
+            else:
+                reason = (
+                    "; ".join(str(item) for item in (structured or {}).get("blockers", []))
+                    if structured
+                    else result.output[-4000:]
+                )
+                kanban_db.block_task(
+                    conn,
+                    task_id,
+                    reason=reason or f"{route.provider} worker failed with exit {result.returncode}",
+                    kind="transient" if result.capacity_failure else "capability",
+                    expected_run_id=run_id,
+                )
+        except Exception as error:
+            kanban_db.block_task(
+                conn,
+                task_id,
+                reason=f"Direct CLI lane failed: {type(error).__name__}: {error}",
+                kind="capability",
+                expected_run_id=run_id,
+            )
+        finally:
+            conn.close()
+
+
+def _board_slug(board: Any) -> str:
+    if isinstance(board, dict):
+        return str(board.get("slug") or board.get("id") or "")
+    return str(getattr(board, "slug", None) or getattr(board, "id", None) or board)
+
+
+def _external(task: Any) -> bool:
+    return str(_task_value(task, "assignee", "") or "").startswith(EXTERNAL_PREFIX)
+
+
+def recover_orphans() -> None:
+    """Return external running tasks to ready after a runner/pod restart."""
+    from hermes_cli import kanban_db
+
+    for raw_board in kanban_db.list_boards(include_archived=False):
+        board = _board_slug(raw_board)
+        if not board:
+            continue
+        with kanban_db.scoped_current_board(board):
+            conn = kanban_db.connect(board=board)
+            try:
+                for task in kanban_db.list_tasks(conn):
+                    if _external(task) and str(_task_value(task, "status", "")) == "running":
+                        kanban_db.reclaim_task(
+                            conn,
+                            str(_task_value(task, "id")),
+                            reason="direct CLI lane restarted; provider session will resume",
+                        )
+            finally:
+                conn.close()
+
+
+def claim_ready(active: set[tuple[str, str]], limit: int) -> list[tuple[str, str]]:
+    """Atomically claim external ready tasks across all non-archived boards."""
+    from hermes_cli import kanban_db
+
+    claimed: list[tuple[str, str]] = []
+    if limit <= 0:
+        return claimed
+    for raw_board in kanban_db.list_boards(include_archived=False):
+        board = _board_slug(raw_board)
+        if not board:
+            continue
+        with kanban_db.scoped_current_board(board):
+            conn = kanban_db.connect(board=board)
+            try:
+                kanban_db.recompute_ready(conn)
+                for task in kanban_db.list_tasks(conn):
+                    task_id = str(_task_value(task, "id", ""))
+                    assignee = str(_task_value(task, "assignee", "") or "")
+                    if task_id and not assignee and str(_task_value(task, "status", "")) == "ready":
+                        if kanban_db.assign_task(conn, task_id, "cli-auto"):
+                            task = kanban_db.get_task(conn, task_id)
+                            assignee = "cli-auto"
+                    if (
+                        not task_id
+                        or (board, task_id) in active
+                        or not assignee.startswith(EXTERNAL_PREFIX)
+                        or str(_task_value(task, "status", "")) != "ready"
+                    ):
+                        continue
+                    try:
+                        result = kanban_db.claim_task(
+                            conn,
+                            task_id,
+                            ttl_seconds=DEFAULT_CLAIM_TTL,
+                            claimer="direct-cli-lane",
+                        )
+                    except Exception:
+                        continue
+                    if result is not None:
+                        claimed.append((board, task_id))
+                    if len(claimed) >= limit:
+                        return claimed
+            finally:
+                conn.close()
+    return claimed
+
+
+def main() -> int:
+    """Continuously bridge external Kanban lanes to provider CLIs."""
+    RESULT_SCHEMA_PATH.parent.mkdir(parents=True, exist_ok=True)
+    atomic_json(RESULT_SCHEMA_PATH, RESULT_SCHEMA, 0o644)
+    recover_orphans()
+    workers = max(1, min(int(os.environ.get("HERMES_CLI_LANE_CONCURRENCY", "4")), 8))
+    futures: dict[concurrent.futures.Future[None], tuple[str, str]] = {}
+    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
+        while True:
+            for future in list(futures):
+                if future.done():
+                    try:
+                        future.result()
+                    except Exception as error:
+                        print(f"worker future failed: {error}", file=sys.stderr, flush=True)
+                    del futures[future]
+            active = set(futures.values())
+            for board, task_id in claim_ready(active, workers - len(futures)):
+                future = pool.submit(execute_claim, board, task_id)
+                futures[future] = (board, task_id)
+            time.sleep(5)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/hermes/scripts/configure_agent_clients.py

@@ -26,10 +26,77 @@ def configure_claude_state(path: Path) -> None:
     os.chmod(path, 0o600)
 
 
+def configure_claude_settings(path: Path) -> None:
+    """Layer unattended deny rules and a command hook onto Claude settings."""
+    value: dict[str, Any] = {}
+    if path.is_file():
+        try:
+            loaded = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            loaded = {}
+        if isinstance(loaded, dict):
+            value = loaded
+
+    permissions = value.get("permissions")
+    if not isinstance(permissions, dict):
+        permissions = {}
+    managed_denies = [
+        "Bash(kubectl apply *)",
+        "Bash(kubectl delete *)",
+        "Bash(kubectl patch *)",
+        "Bash(kubectl scale *)",
+        "Bash(kubectl exec *)",
+        "Bash(kubectl port-forward *)",
+        "Bash(flux reconcile *)",
+        "Bash(flux suspend *)",
+        "Bash(flux resume *)",
+        "Bash(vault kv *)",
+        "Bash(git push --force *)",
+        "Bash(git reset --hard *)",
+        "Bash(git clean -f *)",
+    ]
+    existing_denies = permissions.get("deny")
+    if not isinstance(existing_denies, list):
+        existing_denies = []
+    permissions["deny"] = [
+        *[str(item) for item in existing_denies],
+        *[item for item in managed_denies if item not in existing_denies],
+    ]
+    value["permissions"] = permissions
+
+    hooks = value.get("hooks")
+    if not isinstance(hooks, dict):
+        hooks = {}
+    managed_hook = {
+        "matcher": "Bash",
+        "hooks": [
+            {
+                "type": "command",
+                "command": "/opt/hermes/.venv/bin/python /opt/coordinator/claude_command_policy.py",
+                "timeout": 10,
+            }
+        ],
+    }
+    existing_hooks = hooks.get("PreToolUse")
+    if not isinstance(existing_hooks, list):
+        existing_hooks = []
+    existing_hooks = [
+        item
+        for item in existing_hooks
+        if "claude_command_policy.py" not in json.dumps(item, sort_keys=True)
+    ]
+    hooks["PreToolUse"] = [*existing_hooks, managed_hook]
+    value["hooks"] = hooks
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8")
+    os.chmod(path, 0o600)
+
+
 def main() -> None:
     """Configure provider clients under the persistent Hermes home."""
     home = Path(os.environ.get("CLAUDE_CONFIG_DIR", "/opt/data/home/.claude"))
     configure_claude_state(home / ".claude.json")
+    configure_claude_settings(home / "settings.json")
 
 
 if __name__ == "__main__":

services/hermes/scripts/herdr_dispatch.py

@@ -1,291 +0,0 @@
-#!/usr/bin/env python3
-"""Plan and launch difficulty-aware Codex or Claude Code workers through Herdr."""
-
-from __future__ import annotations
-
-import argparse
-import importlib.util
-import json
-import os
-import re
-import subprocess
-import sys
-from pathlib import Path
-from typing import Any
-
-
-ALLOWED_EFFORTS = ("low", "medium", "high", "xhigh")
-ROUTING_PATH = Path("/opt/data/workspace/coordinator/model-routing.json")
-HERDR_BIN = Path("/opt/data/tools/bin/herdr")
-CODEX_AUTH = Path("/opt/data/home/.codex/auth.json")
-AUTO_ROUTER_PATH = Path("/opt/data/plugins/auto-router/__init__.py")
-PROMPT_READY_MARKERS = {
-    "codex": "OpenAI Codex",
-    "claude": "Claude Code",
-}
-
-
-def _load_routes(path: Path) -> dict[str, Any]:
-    """Load the non-secret route status generated by the model steward."""
-    try:
-        value = json.loads(path.read_text(encoding="utf-8"))
-    except (OSError, json.JSONDecodeError) as error:
-        raise RuntimeError(f"routing status unavailable: {path}") from error
-    if not isinstance(value, dict) or not isinstance(value.get("routes"), dict):
-        raise RuntimeError(f"routing status is malformed: {path}")
-    return value
-
-
-def _split_route(route: str) -> tuple[str, str]:
-    """Split a provider/model route without corrupting model punctuation."""
-    provider, separator, model = route.partition("/")
-    if not separator or not provider or not model:
-        raise RuntimeError(f"invalid route: {route}")
-    return provider, model
-
-
-def select_plan(
-    status: dict[str, Any], shape: str, effort: str, provider: str | None = None
-) -> dict[str, Any]:
-    """Select a capped route and retain the complete capacity fallback chain."""
-    if effort not in ALLOWED_EFFORTS:
-        raise ValueError(f"effort must be one of: {', '.join(ALLOWED_EFFORTS)}")
-    default_provider = "codex" if shape == "implementation" else "claude"
-    selected_provider = provider or default_provider
-    if selected_provider not in {"codex", "claude"}:
-        raise ValueError("provider must be codex or claude")
-    profile = f"{selected_provider}-{effort}"
-    chain = status["routes"].get(profile)
-    if not isinstance(chain, list) or not chain:
-        raise RuntimeError(f"route profile unavailable: {profile}")
-    primary_provider, model = _split_route(str(chain[0]))
-    return {
-        "shape": shape,
-        "effort": effort,
-        "profile": profile,
-        "worker": selected_provider,
-        "provider": primary_provider,
-        "model": model,
-        "fallback_chain": [str(route) for route in chain[1:]],
-    }
-
-
-def select_auto_plan(
-    status: dict[str, Any], prompt: str, router_path: Path = AUTO_ROUTER_PATH
-) -> dict[str, Any]:
-    """Classify one bounded CLI objective on the Jetson and resolve its route."""
-    if not prompt.strip():
-        raise ValueError("AUTO routing requires a non-empty worker prompt")
-    if not router_path.is_file():
-        raise RuntimeError(f"AUTO router is unavailable: {router_path}")
-    spec = importlib.util.spec_from_file_location("hermes_auto_router_dispatch", router_path)
-    if spec is None or spec.loader is None:
-        raise RuntimeError(f"AUTO router could not be loaded: {router_path}")
-    router = importlib.util.module_from_spec(spec)
-    sys.modules[spec.name] = router
-    spec.loader.exec_module(router)
-    decision = router.classify_task(prompt)
-    plan = select_plan(
-        status,
-        str(decision.shape),
-        str(decision.effort),
-        str(decision.provider),
-    )
-    plan.update(
-        {
-            "classifier": str(decision.classifier),
-            "classification_reason": str(decision.reason),
-            "classification_latency_ms": int(decision.latency_ms),
-        }
-    )
-    return plan
-
-
-def _slug(value: str, limit: int = 32) -> str:
-    """Return a Herdr-safe stable label."""
-    cleaned = re.sub(r"[^a-z0-9_-]+", "-", value.lower()).strip("-")
-    if not cleaned or not cleaned[0].isalpha():
-        cleaned = f"task-{cleaned}"
-    return cleaned[:limit].rstrip("-")
-
-
-def _run(command: list[str], env: dict[str, str]) -> dict[str, Any]:
-    """Run one Herdr JSON command and surface a concise error."""
-    completed = subprocess.run(
-        command,
-        env=env,
-        text=True,
-        capture_output=True,
-        timeout=320,
-        check=False,
-    )
-    if completed.returncode != 0:
-        detail = completed.stderr.strip() or completed.stdout.strip()
-        raise RuntimeError(detail or f"command failed with {completed.returncode}")
-    try:
-        value = json.loads(completed.stdout)
-    except json.JSONDecodeError as error:
-        raise RuntimeError("Herdr returned non-JSON output") from error
-    if not isinstance(value, dict):
-        raise RuntimeError("Herdr returned an unexpected response")
-    return value
-
-
-def launch_worker(
-    plan: dict[str, Any], project: Path, task: str, prompt: str | None
-) -> dict[str, Any]:
-    """Create an isolated Herdr workspace and launch the selected worker."""
-    if not HERDR_BIN.is_file():
-        raise RuntimeError("Herdr is not installed in the agent tools volume")
-    if not project.is_dir():
-        raise RuntimeError(f"project workspace does not exist: {project}")
-    if plan["worker"] == "codex" and not CODEX_AUTH.is_file():
-        raise RuntimeError(
-            "Codex CLI needs its one-time device login. Ask Hermes to run "
-            "`codex login --device-auth`, complete the code, then retry."
-        )
-
-    env = os.environ.copy()
-    env.update(
-        {
-            "HOME": "/opt/data/home",
-            "CODEX_HOME": "/opt/data/home/.codex",
-            "CLAUDE_CONFIG_DIR": "/opt/data/home/.claude",
-            "HERDR_CONFIG_PATH": "/opt/data/home/.config/herdr/config.toml",
-            "HERDR_SOCKET_PATH": "/opt/data/herdr/herdr.sock",
-            "PATH": "/opt/data/tools/bin:" + env.get("PATH", ""),
-        }
-    )
-    project_label = _slug(project.name, 12)
-    task_label = _slug(task, 14)
-    label = _slug(f"{project_label}-{plan['worker']}-{task_label}")
-    created = _run(
-        [
-            str(HERDR_BIN),
-            "workspace",
-            "create",
-            "--cwd",
-            str(project),
-            "--label",
-            label,
-            "--no-focus",
-        ],
-        env,
-    )
-    try:
-        pane = str(created["result"]["root_pane"]["pane_id"])
-    except (KeyError, TypeError) as error:
-        raise RuntimeError("Herdr workspace response omitted the root pane") from error
-
-    agent_name = label
-    command = [
-        str(HERDR_BIN),
-        "agent",
-        "start",
-        agent_name,
-        "--kind",
-        plan["worker"],
-        "--pane",
-        pane,
-        "--timeout",
-        "120000",
-        "--",
-    ]
-    if plan["worker"] == "codex":
-        command.extend(
-            [
-                "-m",
-                plan["model"],
-                "-c",
-                f'model_reasoning_effort="{plan["effort"]}"',
-            ]
-        )
-    else:
-        command.extend(
-            [
-                "--model",
-                plan["model"],
-                "--effort",
-                plan["effort"],
-            ]
-        )
-    started = _run(command, env)
-    result = {**plan, "agent": agent_name, "pane": pane, "started": started}
-    if prompt:
-        # Herdr can detect the process before the full-screen prompt has finished
-        # drawing. Wait for a pinned CLI marker so the submitted Enter is not lost.
-        _run(
-            [
-                str(HERDR_BIN),
-                "pane",
-                "wait-output",
-                pane,
-                "--match",
-                PROMPT_READY_MARKERS[plan["worker"]],
-                "--source",
-                "recent",
-                "--lines",
-                "120",
-                "--timeout",
-                "120000",
-            ],
-            env,
-        )
-        result["prompted"] = _run(
-            [
-                str(HERDR_BIN),
-                "agent",
-                "prompt",
-                agent_name,
-                prompt,
-                "--wait",
-                "--until",
-                "working",
-                "--until",
-                "done",
-                "--until",
-                "blocked",
-                "--timeout",
-                "15000",
-            ],
-            env,
-        )
-    return result
-
-
-def main() -> int:
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--auto", action="store_true")
-    parser.add_argument("--shape", choices=("implementation", "architecture", "review", "question"))
-    parser.add_argument("--effort", choices=ALLOWED_EFFORTS)
-    parser.add_argument("--provider", choices=("codex", "claude"))
-    parser.add_argument("--routes", type=Path, default=ROUTING_PATH)
-    parser.add_argument("--start", action="store_true")
-    parser.add_argument("--project", type=Path)
-    parser.add_argument("--task", default="objective")
-    parser.add_argument("--prompt")
-    args = parser.parse_args()
-
-    status = _load_routes(args.routes)
-    if args.auto:
-        if args.provider is not None:
-            parser.error("--provider cannot be combined with --auto")
-        if args.shape is not None or args.effort is not None:
-            parser.error("--shape/--effort cannot be combined with --auto")
-        if not args.prompt:
-            parser.error("--prompt is required with --auto")
-        plan = select_auto_plan(status, args.prompt)
-    else:
-        if args.shape is None or args.effort is None:
-            parser.error("--shape and --effort are required unless --auto is used")
-        plan = select_plan(status, args.shape, args.effort, args.provider)
-    if args.start:
-        if args.project is None:
-            parser.error("--project is required with --start")
-        plan = launch_worker(plan, args.project.resolve(), args.task, args.prompt)
-    print(json.dumps(plan, indent=2, sort_keys=True))
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

services/hermes/scripts/herdr_tab_router.py

@@ -1,249 +0,0 @@
-#!/usr/bin/env python3
-"""Turn interactive tabs in the Agent coordinator space into Hermes sessions."""
-
-from __future__ import annotations
-
-import argparse
-import json
-import os
-import re
-import shlex
-import subprocess
-import time
-from pathlib import Path
-from typing import Any, Callable
-
-
-HERDR_BIN = Path("/opt/data/tools/bin/herdr")
-WORKSPACE_ROOT = Path("/opt/data/workspace")
-PROJECT_ROOT = WORKSPACE_ROOT / "projects"
-
-
-def _slug(value: str) -> str:
-    """Return a stable label accepted by Herdr's agent-name argument."""
-    normalized = re.sub(r"[^a-z0-9]+", "-", value.strip().lower()).strip("-")
-    return normalized or "session"
-
-
-def project_for_label(label: str, project_root: Path = PROJECT_ROOT) -> Path | None:
-    """Resolve a tab label to an existing isolated project directory."""
-    wanted = _slug(label)
-    if not project_root.is_dir():
-        return None
-    for candidate in project_root.iterdir():
-        if candidate.is_dir() and _slug(candidate.name) == wanted:
-            return candidate.resolve()
-    return None
-
-
-def _run_json(command: list[str]) -> dict[str, Any]:
-    """Run a Herdr command and return its JSON object response."""
-    completed = subprocess.run(
-        command,
-        text=True,
-        capture_output=True,
-        timeout=130,
-        check=False,
-        env=os.environ.copy(),
-    )
-    if completed.returncode != 0:
-        detail = completed.stderr.strip() or completed.stdout.strip()
-        raise RuntimeError(detail or f"command failed with {completed.returncode}")
-    try:
-        payload = json.loads(completed.stdout)
-    except json.JSONDecodeError as error:
-        raise RuntimeError("Herdr returned non-JSON output") from error
-    if not isinstance(payload, dict):
-        raise RuntimeError("Herdr returned an unexpected response")
-    return payload
-
-
-def _result_list(payload: dict[str, Any], key: str) -> list[dict[str, Any]]:
-    """Extract a typed list from a Herdr response."""
-    result = payload.get("result") or {}
-    values = result.get(key) or []
-    return [item for item in values if isinstance(item, dict)]
-
-
-def find_workspace(
-    label: str,
-    run: Callable[[list[str]], dict[str, Any]] = _run_json,
-) -> str | None:
-    """Return the live workspace id for the named Agent coordinator space."""
-    payload = run([str(HERDR_BIN), "workspace", "list"])
-    matches = [
-        str(item.get("workspace_id") or "")
-        for item in _result_list(payload, "workspaces")
-        if item.get("label") == label
-    ]
-    return next((item for item in matches if item), None)
-
-
-def route_unmanaged_tabs(
-    workspace_id: str,
-    run: Callable[[list[str]], dict[str, Any]] = _run_json,
-    project_root: Path = PROJECT_ROOT,
-) -> list[str]:
-    """Start routed Hermes in every unmanaged pane of the coordinator space."""
-    tabs_payload = run([str(HERDR_BIN), "tab", "list"])
-    panes_payload = run(
-        [str(HERDR_BIN), "pane", "list", "--workspace", workspace_id]
-    )
-    tab_labels = {
-        str(item.get("tab_id") or ""): str(item.get("label") or "session")
-        for item in _result_list(tabs_payload, "tabs")
-        if item.get("workspace_id") == workspace_id
-    }
-    started: list[str] = []
-    for pane in _result_list(panes_payload, "panes"):
-        if pane.get("workspace_id") != workspace_id or pane.get("agent"):
-            continue
-        pane_id = str(pane.get("pane_id") or "")
-        if not pane_id:
-            continue
-        label = tab_labels.get(str(pane.get("tab_id") or ""), "session")
-        project = project_for_label(label, project_root)
-        if project is not None and Path(str(pane.get("foreground_cwd") or "")) != project:
-            run(
-                [
-                    str(HERDR_BIN),
-                    "pane",
-                    "run",
-                    pane_id,
-                    f"cd {shlex.quote(str(project))}",
-                ]
-            )
-        pane_suffix = _slug(pane_id.split(":")[-1])
-        agent_name = f"{_slug(label)}-{pane_suffix}"
-        run(
-            [
-                str(HERDR_BIN),
-                "agent",
-                "start",
-                agent_name,
-                "--kind",
-                "hermes",
-                "--pane",
-                pane_id,
-                "--timeout",
-                "120000",
-            ]
-        )
-        started.append(pane_id)
-    return started
-
-
-def _pane_number(pane_id: str) -> int:
-    """Return a stable numeric sort key for Herdr pane identifiers."""
-    match = re.search(r":p(\d+)$", pane_id)
-    return int(match.group(1)) if match else 2**31 - 1
-
-
-def _worker_label(
-    tab_label: str,
-    pane: dict[str, Any],
-    agent_names: dict[str, str],
-) -> str:
-    """Describe a detached worker using its parent tab, provider, and role."""
-    pane_id = str(pane.get("pane_id") or "")
-    parent = _slug(tab_label)
-    raw_name = agent_names.get(pane_id, "").strip()
-    name = _slug(raw_name) if raw_name else ""
-    provider = _slug(str(pane.get("agent") or "worker"))
-    suffix = _slug(pane_id.split(":")[-1])
-    if name and name not in {parent, f"{parent}-{suffix}"}:
-        label = name if name.startswith(f"{parent}-") else f"{parent}-{name}"
-    else:
-        label = f"{parent}-{provider}-{suffix}"
-    return label[:32].rstrip("-")
-
-
-def isolate_worker_panes(
-    workspace_id: str,
-    run: Callable[[list[str]], dict[str, Any]] = _run_json,
-) -> list[str]:
-    """Keep one coordinator pane per tab and detach every worker workspace."""
-    tabs_payload = run([str(HERDR_BIN), "tab", "list"])
-    panes_payload = run(
-        [str(HERDR_BIN), "pane", "list", "--workspace", workspace_id]
-    )
-    agents_payload = run([str(HERDR_BIN), "agent", "list"])
-    tab_labels = {
-        str(item.get("tab_id") or ""): str(item.get("label") or "session")
-        for item in _result_list(tabs_payload, "tabs")
-        if item.get("workspace_id") == workspace_id
-    }
-    agent_names = {
-        str(item.get("pane_id") or ""): str(item.get("name") or "")
-        for item in _result_list(agents_payload, "agents")
-        if item.get("workspace_id") == workspace_id
-    }
-    panes_by_tab: dict[str, list[dict[str, Any]]] = {}
-    for pane in _result_list(panes_payload, "panes"):
-        if pane.get("workspace_id") != workspace_id:
-            continue
-        tab_id = str(pane.get("tab_id") or "")
-        if tab_id:
-            panes_by_tab.setdefault(tab_id, []).append(pane)
-
-    moved: list[str] = []
-    for tab_id, panes in panes_by_tab.items():
-        if len(panes) <= 1:
-            continue
-        # The oldest Hermes pane is the tab's stable coordinator. Directly
-        # spawned Codex/Claude workers and later Hermes helpers are detached.
-        primary = min(
-            panes,
-            key=lambda pane: (
-                0 if pane.get("agent") == "hermes" else 1,
-                _pane_number(str(pane.get("pane_id") or "")),
-            ),
-        )
-        for pane in panes:
-            if pane is primary:
-                continue
-            pane_id = str(pane.get("pane_id") or "")
-            if not pane_id:
-                continue
-            run(
-                [
-                    str(HERDR_BIN),
-                    "pane",
-                    "move",
-                    pane_id,
-                    "--new-workspace",
-                    "--label",
-                    _worker_label(tab_labels.get(tab_id, "session"), pane, agent_names),
-                    "--no-focus",
-                ]
-            )
-            moved.append(pane_id)
-    return moved
-
-
-def run_loop(label: str, interval: float) -> None:
-    """Continuously reconcile Agent tabs while allowing Herdr to own workers."""
-    while True:
-        try:
-            workspace_id = find_workspace(label)
-            if workspace_id:
-                for pane_id in route_unmanaged_tabs(workspace_id):
-                    print(f"Started routed Hermes session in {pane_id}", flush=True)
-                for pane_id in isolate_worker_panes(workspace_id):
-                    print(f"Detached worker pane {pane_id}", flush=True)
-        except Exception as error:
-            print(f"Agent tab routing retry: {type(error).__name__}: {error}", flush=True)
-        time.sleep(interval)
-
-
-def main() -> None:
-    """Parse controller options and reconcile the Agent coordinator space."""
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--workspace-label", default="coordinator")
-    parser.add_argument("--interval", type=float, default=1.0)
-    args = parser.parse_args()
-    run_loop(args.workspace_label, max(args.interval, 0.25))
-
-
-if __name__ == "__main__":
-    main()

services/hermes/scripts/migrate_herdr_state.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+"""Archive legacy HERDR metadata before the direct CLI lane takes over."""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+DATA_ROOT = Path("/opt/data")
+
+
+def _load_json(path: Path) -> dict[str, Any]:
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return {}
+    return value if isinstance(value, dict) else {}
+
+
+def archive_legacy_state(data_root: Path = DATA_ROOT) -> Path:
+    """Persist non-secret worker references and retire the exact HERDR binary."""
+    source = data_root / "home/.config/herdr/session.json"
+    archive = data_root / "cli-lanes/migration/herdr-retirement.json"
+    archive.parent.mkdir(parents=True, exist_ok=True)
+    previous = _load_json(archive)
+    legacy = _load_json(source)
+    value = {
+        "schema_version": 1,
+        "migrated_at": previous.get("migrated_at")
+        or datetime.now(timezone.utc).isoformat(),
+        "source": str(source),
+        "legacy_session": legacy,
+        "provider_state_preserved": [
+            str(data_root / "home/.claude"),
+            str(data_root / "home/.codex"),
+        ],
+    }
+    temporary = archive.with_name(f".{archive.name}.{os.getpid()}.tmp")
+    temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    temporary.chmod(0o600)
+    os.replace(temporary, archive)
+
+    # Provider transcripts remain in .claude and .codex. HERDR itself is
+    # reproducibly reinstallable from Git history, so only its pinned binary is
+    # retired after the migration record is durable.
+    binary = data_root / "tools/bin/herdr"
+    if binary.is_file():
+        binary.unlink()
+    for legacy_dir in (data_root / "home/.config/herdr", data_root / "herdr"):
+        if legacy_dir.is_dir():
+            shutil.rmtree(legacy_dir)
+    return archive
+
+
+if __name__ == "__main__":
+    print(archive_legacy_state())

services/hermes/service.yaml

@@ -54,9 +54,9 @@ spec:
     - name: dashboard
       port: 8787
       targetPort: dashboard
-    - name: herdr-tui
+    - name: terminal
       port: 7681
-      targetPort: herdr-tui
+      targetPort: terminal
 ---
 apiVersion: v1
 kind: Service

services/keycloak/bootstrap-jobs/hermes-access-oidc-client-job.yaml

@@ -3,7 +3,7 @@
 apiVersion: batch/v1
 kind: Job
 metadata:
-  name: hermes-access-oidc-client-ensure-5
+  name: hermes-access-oidc-client-ensure-6
   namespace: sso
 spec:
   backoffLimit: 3

services/keycloak/scripts/hermes_access_oidc_ensure.sh

@@ -34,6 +34,43 @@ if [ -z "${ACCESS_TOKEN}" ] || [ "${ACCESS_TOKEN}" = "null" ]; then
   exit 1
 fi
 
+ensure_hermes_owner() {
+  group_name="hermes-owner"
+  groups="$(curl -sS -H "Authorization: Bearer ${ACCESS_TOKEN}" \
+    "${KC_URL}/admin/realms/atlas/groups?search=${group_name}" || true)"
+  group_id="$(printf '%s' "${groups}" | jq -r --arg name "${group_name}" \
+    '[.[] | select(.name == $name)][0].id // empty')"
+  if [ -z "${group_id}" ]; then
+    status="$(curl -sS -o /dev/null -w '%{http_code}' -X POST \
+      -H "Authorization: Bearer ${ACCESS_TOKEN}" \
+      -H 'Content-Type: application/json' \
+      -d "$(jq -nc --arg name "${group_name}" '{name:$name}')" \
+      "${KC_URL}/admin/realms/atlas/groups")"
+    if [ "${status}" != "201" ] && [ "${status}" != "204" ] && [ "${status}" != "409" ]; then
+      echo "Keycloak ${group_name} group create failed (status ${status})" >&2
+      exit 1
+    fi
+    groups="$(curl -sS -H "Authorization: Bearer ${ACCESS_TOKEN}" \
+      "${KC_URL}/admin/realms/atlas/groups?search=${group_name}" || true)"
+    group_id="$(printf '%s' "${groups}" | jq -r --arg name "${group_name}" \
+      '[.[] | select(.name == $name)][0].id // empty')"
+  fi
+  users="$(curl -sS -H "Authorization: Bearer ${ACCESS_TOKEN}" \
+    "${KC_URL}/admin/realms/atlas/users?username=bstein&exact=true&max=1" || true)"
+  user_id="$(printf '%s' "${users}" | jq -r '.[0].id // empty')"
+  if [ -z "${group_id}" ] || [ -z "${user_id}" ]; then
+    echo "Unable to resolve the immutable Hermes owner group or bstein user" >&2
+    exit 1
+  fi
+  status="$(curl -sS -o /dev/null -w '%{http_code}' -X PUT \
+    -H "Authorization: Bearer ${ACCESS_TOKEN}" \
+    "${KC_URL}/admin/realms/atlas/users/${user_id}/groups/${group_id}")"
+  if [ "${status}" != "204" ] && [ "${status}" != "200" ]; then
+    echo "Unable to assign bstein to ${group_name} (status ${status})" >&2
+    exit 1
+  fi
+}
+
 jwt="$(cat /var/run/secrets/kubernetes.io/serviceaccount/token)"
 login_payload="$(jq -nc --arg jwt "${jwt}" --arg role "${VAULT_ROLE}" '{jwt:$jwt,role:$role}')"
 vault_token="$(curl -sS --request POST --data "${login_payload}" \
@@ -104,6 +141,34 @@ ensure_proxy_client() {
     exit 1
   fi
 
+  mapper_payload="$(jq -nc \
+    '{name:"groups",protocol:"openid-connect",protocolMapper:"oidc-group-membership-mapper",consentRequired:false,config:{"full.path":"true","id.token.claim":"true","access.token.claim":"true","userinfo.token.claim":"true","claim.name":"groups","jsonType.label":"String"}}')"
+  mappers="$(curl -sS -H "Authorization: Bearer ${ACCESS_TOKEN}" \
+    "${KC_URL}/admin/realms/atlas/clients/${internal_id}/protocol-mappers/models" || true)"
+  mapper_id="$(printf '%s' "${mappers}" | jq -r '[.[] | select(.name == "groups")][0].id // empty')"
+  if [ -n "${mapper_id}" ]; then
+    update_payload="$(printf '%s' "${mapper_payload}" | jq -c --arg id "${mapper_id}" '. + {id:$id}')"
+    mapper_status="$(curl -sS -o /dev/null -w '%{http_code}' -X PUT \
+      -H "Authorization: Bearer ${ACCESS_TOKEN}" \
+      -H 'Content-Type: application/json' \
+      -d "${update_payload}" \
+      "${KC_URL}/admin/realms/atlas/clients/${internal_id}/protocol-mappers/models/${mapper_id}")"
+    [ "${mapper_status}" = "204" ] || {
+      echo "Keycloak ${client_id} groups mapper update failed (status ${mapper_status})" >&2
+      exit 1
+    }
+  else
+    mapper_status="$(curl -sS -o /dev/null -w '%{http_code}' -X POST \
+      -H "Authorization: Bearer ${ACCESS_TOKEN}" \
+      -H 'Content-Type: application/json' \
+      -d "${mapper_payload}" \
+      "${KC_URL}/admin/realms/atlas/clients/${internal_id}/protocol-mappers/models")"
+    [ "${mapper_status}" = "201" ] || [ "${mapper_status}" = "204" ] || {
+      echo "Keycloak ${client_id} groups mapper create failed (status ${mapper_status})" >&2
+      exit 1
+    }
+  fi
+
   client_secret="$(curl -sS -H "Authorization: Bearer ${ACCESS_TOKEN}" \
     "${KC_URL}/admin/realms/atlas/clients/${internal_id}/client-secret" \
     | jq -r '.value' 2>/dev/null || true)"
@@ -178,6 +243,7 @@ ensure_telegram_config() {
   echo "Hermes Telegram transport secret is ready"
 }
 
+ensure_hermes_owner
 ensure_proxy_client "hermes-chat-proxy" "https://chat.hermes.bstein.dev" "hermes/chat-oidc"
 ensure_proxy_client "hermes-agent-proxy" "https://agent.hermes.bstein.dev" "hermes/agent-oidc"
 ensure_proxy_client "hermes-triage-proxy" "https://triage.hermes.bstein.dev" "hermes/triage-oidc"

testing/tests/test_hermes_auto_router.py

@@ -302,13 +302,26 @@ def test_every_internal_auto_prompt_is_reclassified_and_applied(monkeypatch):
     assert agent.message.startswith("AUTO internal #3")
 
 
-def test_manual_route_skips_internal_reclassification(monkeypatch):
+def test_manual_route_audits_internal_prompt_without_overriding_user_choice(monkeypatch):
-    monkeypatch.setattr(router, "_current_policy", lambda: {"mode": "manual"})
+    monkeypatch.setattr(
+        router,
+        "_current_policy",
+        lambda: {
+            "mode": "manual",
+            "manual": {"provider": "claude", "effort": "medium", "model": ""},
+        },
+    )
+    monkeypatch.setattr(router, "_load_json", lambda path: _status())
+    calls = []
     monkeypatch.setattr(
         router,
         "classify_task",
-        lambda text: (_ for _ in ()).throw(AssertionError("should not classify")),
+        lambda text: calls.append(text)
+        or router.Decision("implementation", "xhigh", "codex", "jetson", "audit", 5),
     )
+    plans = []
+    monkeypatch.setattr(router, "_apply_route", lambda ctx, agent, plan: plans.append(plan))
+    monkeypatch.setattr(router, "_record_internal_plan", lambda *args: None)
 
     router._pre_internal_route(
         object(),
@@ -317,6 +330,9 @@ def test_manual_route_skips_internal_reclassification(monkeypatch):
         conversation_history=[{"role": "tool", "content": "done"}],
         api_call_count=2,
     )
+    assert len(calls) == 1
+    assert plans[0]["profile"] == "claude-medium"
+    assert plans[0]["classifier"] == "manual-jetson-internal"
 
 
 def test_every_native_subagent_is_classified_and_routed_independently(monkeypatch):
@@ -366,15 +382,33 @@ def test_every_native_subagent_is_classified_and_routed_independently(monkeypatc
     assert parent.message.startswith("AUTO child #3")
 
 
-def test_manual_route_leaves_native_subagent_on_parent_override(monkeypatch):
+def test_manual_route_audits_and_applies_override_to_native_subagent(monkeypatch):
-    monkeypatch.setattr(router, "_current_policy", lambda: {"mode": "manual"})
+    monkeypatch.setattr(
+        router,
+        "_current_policy",
+        lambda: {
+            "mode": "manual",
+            "manual": {"provider": "claude", "effort": "medium", "model": ""},
+        },
+    )
+    monkeypatch.setattr(router, "_load_json", lambda path: _status())
+    calls = []
     monkeypatch.setattr(
         router,
         "classify_task",
-        lambda text: (_ for _ in ()).throw(AssertionError("should not classify")),
+        lambda text: calls.append(text)
+        or router.Decision("implementation", "high", "codex", "jetson", "audit", 5),
     )
+    plans = []
+    monkeypatch.setattr(router, "_apply_route", lambda ctx, agent, plan: plans.append(plan))
+    monkeypatch.setattr(router, "_record_subagent_plan", lambda *args: None)
 
-    router._pre_subagent_route(object(), agent=object(), goal="Review the diff")
+    router._pre_subagent_route(
+        object(), agent=object(), parent_agent=object(), goal="Review the diff"
+    )
+    assert len(calls) == 1
+    assert plans[0]["profile"] == "claude-medium"
+    assert plans[0]["classifier"] == "manual-jetson-subagent"
 
 
 def test_manual_policy_is_reapplied_on_every_non_command_turn(monkeypatch):
@@ -387,6 +421,13 @@ def test_manual_policy_is_reapplied_on_every_non_command_turn(monkeypatch):
         },
     )
     monkeypatch.setattr(router, "_load_json", lambda path: _status())
+    monkeypatch.setattr(
+        router,
+        "classify_task",
+        lambda text, history=None: router.Decision(
+            "implementation", "high", "codex", "jetson", "audit", 5
+        ),
+    )
     plans = []
     monkeypatch.setattr(router, "_apply_route", lambda ctx, agent, plan: plans.append(plan))
     monkeypatch.setattr(router, "_record_plan", lambda policy, plan: None)
@@ -401,6 +442,7 @@ def test_manual_policy_is_reapplied_on_every_non_command_turn(monkeypatch):
     assert plans[0]["profile"] == "claude-medium"
     assert plans[0]["model"] == "claude-sonnet-5"
     assert agent.message.startswith("MANUAL target")
+    assert plans[0]["classifier"] == "manual-jetson"
 
 
 def test_post_turn_records_and_announces_capacity_fallback(monkeypatch):

testing/tests/test_hermes_cli_lanes.py

@@ -0,0 +1,717 @@
+"""Focused tests for Agent Hermes' direct Codex and Claude Kanban lanes."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import sys
+from contextlib import nullcontext
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+import yaml
+
+
+SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts"
+HERMES = Path(__file__).parents[2] / "services/hermes"
+KEYCLOAK = Path(__file__).parents[2] / "services/keycloak"
+
+
+def _load(name: str):
+    spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py")
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+lanes = _load("cli_lane_runner")
+policy = _load("claude_command_policy")
+migration = _load("migrate_herdr_state")
+auth_patch = _load("patch_hermes_auth")
+ttyd_patch = _load("patch_ttyd_index")
+client_config = _load("configure_agent_clients")
+
+
+def _agent_deployment() -> dict:
+    return yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())
+
+
+def _oauth_deployment(name: str) -> dict:
+    documents = [
+        item
+        for item in yaml.safe_load_all((HERMES / "oauth2-proxy.yaml").read_text())
+        if item
+    ]
+    return next(
+        item
+        for item in documents
+        if item["kind"] == "Deployment" and item["metadata"]["name"] == name
+    )
+
+
+def test_auto_lane_always_uses_the_jetson_decision(tmp_path: Path):
+    router = tmp_path / "router.py"
+    router.write_text(
+        """
+class Decision:
+    shape = "review"
+    effort = "xhigh"
+    provider = "claude"
+    classifier = "jetson"
+    reason = "local vote"
+    latency_ms = 17
+
+def classify_task(prompt):
+    assert "security review" in prompt
+    return Decision()
+""",
+        encoding="utf-8",
+    )
+    routes = tmp_path / "routes.json"
+    routes.write_text(
+        json.dumps(
+            {
+                "routes": {
+                    "claude-xhigh": [
+                        "anthropic/claude-opus-5",
+                        "openai-codex/gpt-5.6-sol",
+                    ]
+                }
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    route = lanes.select_route(
+        "Perform the security review.",
+        "cli-auto",
+        routing_path=routes,
+        router_path=router,
+    )
+
+    assert route.provider == "claude"
+    assert route.model == "claude-opus-5"
+    assert route.effort == "xhigh"
+    assert route.classifier == "jetson"
+    assert route.latency_ms == 17
+
+
+def test_manual_lane_still_calls_jetson_before_applying_override(tmp_path: Path):
+    router = tmp_path / "router.py"
+    router.write_text(
+        """
+called = 0
+class Decision:
+    shape = "question"
+    effort = "low"
+    provider = "codex"
+    classifier = "jetson"
+    reason = "local vote"
+    latency_ms = 8
+def classify_task(prompt):
+    global called
+    called += 1
+    return Decision()
+""",
+        encoding="utf-8",
+    )
+    routes = tmp_path / "routes.json"
+    routes.write_text(
+        json.dumps({"routes": {"claude-high": ["anthropic/claude-opus-5"]}}),
+        encoding="utf-8",
+    )
+
+    route = lanes.select_route(
+        "Implement it.",
+        "cli-claude-high",
+        routing_path=routes,
+        router_path=router,
+    )
+
+    assert route.provider == "claude"
+    assert route.effort == "high"
+    assert "manual lane override" in route.reason
+
+
+def test_cross_provider_retry_excludes_failed_provider(tmp_path: Path):
+    router = tmp_path / "router.py"
+    router.write_text(
+        """
+class Decision:
+    shape = "implementation"
+    effort = "high"
+    provider = "codex"
+    classifier = "jetson"
+    reason = "retry vote"
+    latency_ms = 9
+def classify_task(prompt): return Decision()
+""",
+        encoding="utf-8",
+    )
+    routes = tmp_path / "routes.json"
+    routes.write_text(
+        json.dumps({"routes": {"claude-high": ["anthropic/claude-opus-5"]}}),
+        encoding="utf-8",
+    )
+
+    route = lanes.select_route(
+        "Retry after capacity exhaustion.",
+        "cli-auto",
+        exclude_provider="codex",
+        routing_path=routes,
+        router_path=router,
+    )
+
+    assert route.provider == "claude"
+    assert route.classifier == "jetson"
+
+
+def test_claude_session_is_reserved_before_first_process(tmp_path: Path, monkeypatch):
+    state: dict = {}
+    state_file = tmp_path / "state.json"
+    observed = {}
+
+    def fake_stream(command, **kwargs):
+        observed["command"] = command
+        observed["state"] = json.loads(state_file.read_text())
+        return lanes.ProcessResult(0, "", {"status": "completed"}, False)
+
+    monkeypatch.setattr(lanes, "stream_process", fake_stream)
+    route = lanes.Route("claude", "claude-opus-5", "high", "claude-high", "jetson", "vote", 1, ())
+
+    lanes.run_provider(
+        route,
+        "Do the work.",
+        tmp_path,
+        state,
+        state_file,
+        tmp_path / "worker.log",
+        lambda _note: True,
+        60,
+    )
+
+    reserved = observed["state"]["claude_session_id"]
+    assert reserved
+    assert ["--session-id", reserved] == observed["command"][-4:-2]
+
+
+def test_claude_recovery_uses_verified_error_state_machine(tmp_path: Path, monkeypatch):
+    state = {"claude_session_id": "d4f91f59-88cb-43d0-a64f-f3284688fe9e", "claude_started": True}
+    state_file = tmp_path / "state.json"
+    calls = []
+
+    def fake_stream(command, **kwargs):
+        calls.append(command)
+        if len(calls) == 1:
+            return lanes.ProcessResult(
+                1,
+                "No conversation found with session ID: d4f91f59-88cb-43d0-a64f-f3284688fe9e",
+                None,
+                False,
+            )
+        return lanes.ProcessResult(0, "", {"status": "completed"}, False)
+
+    monkeypatch.setattr(lanes, "stream_process", fake_stream)
+    route = lanes.Route("claude", "claude-opus-5", "high", "claude-high", "jetson", "vote", 1, ())
+    lanes.run_provider(route, "Continue.", tmp_path, state, state_file, tmp_path / "log", lambda _: True, 60)
+
+    assert "--resume" in calls[0]
+    assert "--session-id" in calls[1]
+    assert calls[0][calls[0].index("--resume") + 1] == calls[1][calls[1].index("--session-id") + 1]
+
+
+def test_claude_collision_never_double_starts_a_reserved_session(tmp_path: Path, monkeypatch):
+    state = {"claude_session_id": "24927cee-f3da-4bcf-b624-602c22860155"}
+    state_file = tmp_path / "state.json"
+    calls = []
+
+    def fake_stream(command, **kwargs):
+        calls.append(command)
+        if len(calls) == 1:
+            return lanes.ProcessResult(1, "Session ID already in use", None, False)
+        return lanes.ProcessResult(0, "", {"status": "completed"}, False)
+
+    monkeypatch.setattr(lanes, "stream_process", fake_stream)
+    route = lanes.Route("claude", "claude-opus-5", "high", "claude-high", "jetson", "vote", 1, ())
+    lanes.run_provider(route, "Continue.", tmp_path, state, state_file, tmp_path / "log", lambda _: True, 60)
+
+    assert "--session-id" in calls[0]
+    assert "--resume" in calls[1]
+    assert calls[0][calls[0].index("--session-id") + 1] == calls[1][calls[1].index("--resume") + 1]
+
+
+def test_codex_thread_started_is_persisted_before_completion(tmp_path: Path):
+    state: dict = {}
+    state_file = tmp_path / "state.json"
+
+    lanes._event_payload(
+        "codex",
+        json.dumps({"type": "thread.started", "thread_id": "thread-123"}),
+        state,
+        state_file,
+    )
+
+    assert json.loads(state_file.read_text())["codex_thread_id"] == "thread-123"
+
+
+def test_successful_process_text_cannot_masquerade_as_capacity_failure(tmp_path: Path):
+    result = lanes.stream_process(
+        [sys.executable, "-c", "print('authentication work completed')"],
+        provider="codex",
+        cwd=tmp_path,
+        env=dict(os.environ),
+        log_path=tmp_path / "worker.log",
+        state={},
+        state_file=tmp_path / "state.json",
+        heartbeat=lambda _note: True,
+        max_runtime=60,
+    )
+
+    assert result.returncode == 0
+    assert result.capacity_failure is False
+
+
+def test_unassigned_ready_task_is_persistently_routed_to_auto_lane(monkeypatch):
+    task = SimpleNamespace(id="t_auto", assignee=None, status="ready")
+    assigned = []
+
+    class Connection:
+        def close(self):
+            return None
+
+    def assign_task(_conn, task_id, profile):
+        assigned.append((task_id, profile))
+        task.assignee = profile
+        return True
+
+    fake_db = SimpleNamespace(
+        list_boards=lambda include_archived=False: [{"slug": "cassandra"}],
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        recompute_ready=lambda _conn: None,
+        list_tasks=lambda _conn: [task],
+        assign_task=assign_task,
+        get_task=lambda _conn, _task_id: task,
+        claim_task=lambda _conn, _task_id, **_kwargs: task,
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+
+    assert lanes.claim_ready(set(), 1) == [("cassandra", "t_auto")]
+    assert assigned == [("t_auto", "cli-auto")]
+
+
+@pytest.mark.parametrize(
+    ("result", "expected_action"),
+    [
+        (lanes.ProcessResult(0, "plain text only", None, False), "block"),
+        (
+            lanes.ProcessResult(
+                0,
+                "",
+                {
+                    "status": "completed",
+                    "summary": "done",
+                    "changed_files": ["src/a.py"],
+                    "tests_run": ["pytest -q"],
+                    "artifacts": ["reports/result.json"],
+                    "blockers": [],
+                },
+                False,
+            ),
+            "complete",
+        ),
+    ],
+)
+def test_claim_requires_structured_evidence_and_surfaces_artifacts(
+    tmp_path: Path,
+    monkeypatch,
+    result,
+    expected_action,
+):
+    task = SimpleNamespace(
+        id="t_worker",
+        current_run_id=4,
+        assignee="cli-auto",
+        max_runtime_seconds=60,
+    )
+    calls = []
+    heartbeats = []
+    artifact = tmp_path / "reports/result.json"
+    artifact.parent.mkdir()
+    artifact.write_text("{}\n", encoding="utf-8")
+
+    class Connection:
+        def close(self):
+            return None
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        get_task=lambda _conn, _task_id: task,
+        worker_log_path=lambda _task_id, board: tmp_path / "worker.log",
+        _resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_worker"),
+        set_branch_name=lambda *_args: None,
+        set_workspace_path=lambda *_args: None,
+        build_worker_context=lambda *_args: "bounded objective",
+        heartbeat_worker=lambda _conn, _task_id, *, note, expected_run_id: (
+            heartbeats.append((note, expected_run_id)) or True
+        ),
+        add_comment=lambda *_args: None,
+        complete_task=lambda *_args, **kwargs: calls.append(("complete", kwargs)),
+        block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)),
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+    monkeypatch.setattr(lanes, "state_path", lambda _board, _task_id: tmp_path / "state.json")
+    monkeypatch.setattr(
+        lanes,
+        "select_route",
+        lambda *_args, **_kwargs: lanes.Route(
+            "codex", "gpt-5.6-sol", "high", "codex-high", "jetson", "vote", 1, ()
+        ),
+    )
+    def run_provider(*args, **_kwargs):
+        assert args[6]("working") is True
+        return result
+
+    monkeypatch.setattr(lanes, "run_provider", run_provider)
+
+    lanes.execute_claim("cassandra", "t_worker")
+
+    assert calls[0][0] == expected_action
+    assert heartbeats == [("working", 4)]
+    if expected_action == "complete":
+        assert calls[0][1]["metadata"]["artifacts"] == [str(artifact)]
+        assert calls[0][1]["metadata"]["tests_run"] == ["pytest -q"]
+    else:
+        assert calls[0][1]["kind"] == "capability"
+
+
+def test_workspace_preparation_failure_durably_blocks_the_claim(tmp_path: Path, monkeypatch):
+    task = SimpleNamespace(id="t_bad_worktree", current_run_id=7, assignee="cli-auto")
+    calls = []
+
+    class Connection:
+        def close(self):
+            return None
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        get_task=lambda _conn, _task_id: task,
+        worker_log_path=lambda _task_id, board: tmp_path / "worker.log",
+        _resolve_worktree_workspace=lambda _task, board: (_ for _ in ()).throw(
+            ValueError("not a Git repository")
+        ),
+        block_task=lambda *_args, **kwargs: calls.append(kwargs),
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+    monkeypatch.setattr(lanes, "state_path", lambda _board, _task_id: tmp_path / "state.json")
+
+    lanes.execute_claim("cassandra", "t_bad_worktree")
+
+    assert calls[0]["kind"] == "capability"
+    assert calls[0]["expected_run_id"] == 7
+    assert "not a Git repository" in calls[0]["reason"]
+
+
+def test_artifacts_cannot_escape_the_task_worktree(tmp_path: Path):
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    inside = workspace / "report.json"
+    outside = tmp_path / "auth.json"
+    inside.write_text("{}\n", encoding="utf-8")
+    outside.write_text("secret\n", encoding="utf-8")
+
+    assert lanes.workspace_artifacts(
+        workspace,
+        ["report.json", str(outside), "missing.json"],
+    ) == [str(inside)]
+
+
+def test_restart_provider_change_includes_explicit_workspace_handoff(tmp_path: Path, monkeypatch):
+    task = SimpleNamespace(
+        id="t_resume",
+        current_run_id=9,
+        assignee="cli-auto",
+        max_runtime_seconds=60,
+    )
+    state_file = tmp_path / "state.json"
+    state_file.write_text(
+        json.dumps({"current_route": {"provider": "claude"}}),
+        encoding="utf-8",
+    )
+    (tmp_path / "worker.log").write_text("prior provider evidence", encoding="utf-8")
+    prompts = []
+
+    class Connection:
+        def close(self):
+            return None
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        get_task=lambda _conn, _task_id: task,
+        worker_log_path=lambda _task_id, board: tmp_path / "worker.log",
+        _resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_resume"),
+        set_branch_name=lambda *_args: None,
+        set_workspace_path=lambda *_args: None,
+        build_worker_context=lambda *_args: "resume objective",
+        add_comment=lambda *_args: None,
+        complete_task=lambda *_args, **_kwargs: None,
+        block_task=lambda *_args, **_kwargs: None,
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+    monkeypatch.setattr(lanes, "state_path", lambda _board, _task_id: state_file)
+    monkeypatch.setattr(
+        lanes,
+        "select_route",
+        lambda *_args, **_kwargs: lanes.Route(
+            "codex", "gpt-5.6-sol", "high", "codex-high", "jetson", "vote", 1, ()
+        ),
+    )
+    monkeypatch.setattr(
+        lanes,
+        "git_handoff",
+        lambda _workspace, output: f"HANDOFF:{output}",
+    )
+    monkeypatch.setattr(
+        lanes,
+        "run_provider",
+        lambda _route, prompt, *_args, **_kwargs: (
+            prompts.append(prompt)
+            or lanes.ProcessResult(
+                0,
+                "",
+                {
+                    "status": "completed",
+                    "summary": "done",
+                    "changed_files": [],
+                    "tests_run": [],
+                    "artifacts": [],
+                    "blockers": [],
+                },
+                False,
+            )
+        ),
+    )
+
+    lanes.execute_claim("cassandra", "t_resume")
+
+    assert "HANDOFF:prior provider evidence" in prompts[0]
+
+
+def test_provider_commands_are_structured_unattended_and_capped(tmp_path: Path):
+    route = lanes.Route("codex", "gpt-5.6-sol", "xhigh", "codex-xhigh", "jetson", "vote", 1, ())
+    command = lanes._codex_command(route, "Work.", tmp_path, {}, tmp_path / "result.json")
+    assert "--dangerously-bypass-approvals-and-sandbox" in command
+    assert "--json" in command
+    assert "--output-schema" in command
+    assert 'model_reasoning_effort="xhigh"' in command
+
+    claude_state = {"claude_session_id": "13864642-2985-4f91-bef5-53f145f878e8"}
+    claude = lanes._claude_command(
+        lanes.Route("claude", "claude-opus-5", "xhigh", "claude-xhigh", "jetson", "vote", 1, ()),
+        "Review.",
+        claude_state,
+        False,
+    )
+    assert "--dangerously-skip-permissions" in claude
+    assert "--output-format" in claude and "stream-json" in claude
+    assert "--json-schema" in claude
+    assert "--disallowedTools" in claude
+    assert "max" not in claude
+
+
+@pytest.mark.parametrize(
+    "command",
+    [
+        "kubectl apply -k services/hermes",
+        "git push --force origin main",
+        "git reset --hard HEAD~1",
+        "git clean -fd",
+        "flux reconcile kustomization hermes",
+        "vault kv get kv/atlas/hermes",
+    ],
+)
+def test_claude_pretool_hook_blocks_hard_denies(command: str):
+    assert policy.denial_reason(command)
+
+
+def test_claude_pretool_hook_allows_normal_engineering():
+    assert policy.denial_reason("pytest -q testing/tests") is None
+    assert policy.denial_reason("git push origin feature/hermes") is None
+
+
+def test_claude_settings_preserve_state_and_install_three_guardrail_layers(tmp_path: Path):
+    state = tmp_path / ".claude.json"
+    settings = tmp_path / "settings.json"
+    state.write_text('{"promptQueueUseCount": 4}\n', encoding="utf-8")
+    settings.write_text('{"theme":"dark"}\n', encoding="utf-8")
+
+    client_config.configure_claude_state(state)
+    client_config.configure_claude_settings(settings)
+
+    state_value = json.loads(state.read_text())
+    settings_value = json.loads(settings.read_text())
+    assert state_value["promptQueueUseCount"] == 4
+    assert state_value["bypassPermissionsModeAccepted"] is True
+    assert settings_value["theme"] == "dark"
+    assert "Bash(git reset --hard *)" in settings_value["permissions"]["deny"]
+    hook = settings_value["hooks"]["PreToolUse"][0]["hooks"][0]
+    assert "claude_command_policy.py" in hook["command"]
+
+
+def test_legacy_state_is_archived_without_removing_provider_transcripts(tmp_path: Path):
+    session = tmp_path / "home/.config/herdr/session.json"
+    session.parent.mkdir(parents=True)
+    session.write_text('{"agents":[{"agent":"claude","session_id":"abc"}]}', encoding="utf-8")
+    binary = tmp_path / "tools/bin/herdr"
+    binary.parent.mkdir(parents=True)
+    binary.write_text("legacy", encoding="utf-8")
+    (tmp_path / "home/.claude").mkdir()
+    (tmp_path / "home/.codex").mkdir()
+
+    archive = migration.archive_legacy_state(tmp_path)
+
+    value = json.loads(archive.read_text())
+    assert value["legacy_session"]["agents"][0]["session_id"] == "abc"
+    assert (tmp_path / "home/.claude").is_dir()
+    assert (tmp_path / "home/.codex").is_dir()
+    assert not binary.exists()
+    assert not (tmp_path / "home/.config/herdr").exists()
+
+
+def test_agent_uses_one_native_kanban_control_plane():
+    configmap = yaml.safe_load((HERMES / "agent-configmap.yaml").read_text())
+    config = yaml.safe_load(configmap["data"]["config.yaml"])
+    assert config["kanban"]["dispatch_in_gateway"] is True
+    assert config["kanban"]["default_assignee"] == "cli-auto"
+    assert config["plugins"]["enabled"] == ["auto-router"]
+
+    deployment = _agent_deployment()
+    pod = deployment["spec"]["template"]["spec"]
+    names = {item["name"] for item in pod["containers"]}
+    assert "cli-lane-runner" in names
+    assert "terminal" in names
+    assert not any("herdr" in name for name in names)
+    rendered = (HERMES / "agent-deployment.yaml").read_text()
+    assert "herdr server" not in rendered
+    assert "herdr-dispatch" not in rendered
+
+
+def test_agent_root_is_webui_and_terminal_is_a_separate_path():
+    deployment = _agent_deployment()
+    terminal = next(
+        item for item in deployment["spec"]["template"]["spec"]["containers"]
+        if item["name"] == "terminal"
+    )
+    command = terminal["args"][0]
+    assert "--base-path /terminal" in command
+    assert "--check-origin" in command
+    assert "/usr/bin/tmux new-session -A" in command
+    assert "--continue" in command
+    assert "--yolo" in command
+
+    oauth = _oauth_deployment("oauth2-proxy-hermes-agent")
+    args = oauth["spec"]["template"]["spec"]["containers"][0]["args"]
+    assert "--upstream=http://hermes-agent.hermes.svc.cluster.local:7681/terminal/" in args
+    assert "--upstream=http://hermes-agent.hermes.svc.cluster.local:8787/" in args
+    assert args.index("--upstream=http://hermes-agent.hermes.svc.cluster.local:7681/terminal/") < args.index("--upstream=http://hermes-agent.hermes.svc.cluster.local:8787/")
+    ingress_documents = [
+        item
+        for item in yaml.safe_load_all((HERMES / "agent-ingress.yaml").read_text())
+        if item
+    ]
+    middleware = next(item for item in ingress_documents if item["kind"] == "Middleware")
+    assert middleware["spec"]["redirectRegex"]["replacement"].endswith("/terminal/")
+
+
+def test_agent_auth_is_bstein_group_and_email_bounded():
+    oauth = _oauth_deployment("oauth2-proxy-hermes-agent")
+    args = oauth["spec"]["template"]["spec"]["containers"][0]["args"]
+    assert "--user-id-claim=sub" in args
+    assert "--oidc-groups-claim=groups" in args
+    assert "--allowed-group=/hermes-owner" in args
+    assert "--authenticated-emails-file=/etc/oauth2-proxy/allowed-emails" in args
+    script = (KEYCLOAK / "scripts/hermes_access_oidc_ensure.sh").read_text()
+    assert 'group_name="hermes-owner"' in script
+    assert "username=bstein&exact=true" in script
+    assert '"full.path":"true"' in script
+
+
+def test_agent_network_boundary_allows_only_authenticated_web_surfaces():
+    documents = [
+        item
+        for item in yaml.safe_load_all((HERMES / "networkpolicy.yaml").read_text())
+        if item
+    ]
+    isolation = next(item for item in documents if item.get("metadata", {}).get("name") == "hermes-agent-isolation")
+    assert isolation["spec"]["ingress"] == [
+        {
+            "from": [{"podSelector": {"matchLabels": {"app": "oauth2-proxy-hermes-agent"}}}],
+            "ports": [
+                {"protocol": "TCP", "port": 7681},
+                {"protocol": "TCP", "port": 8787},
+            ],
+        }
+    ]
+
+
+def test_agent_has_cassandra_readonly_kubernetes_context():
+    config = yaml.safe_load((HERMES / "agent-kubeconfig.yaml").read_text())
+    assert config["current-context"] == "cassandra-readonly"
+    assert config["contexts"][0]["context"]["namespace"] == "cassandra"
+    rbac_path = HERMES.parent / "cassandra/hermes-agent-rbac.yaml"
+    documents = [item for item in yaml.safe_load_all(rbac_path.read_text()) if item]
+    role = next(item for item in documents if item["kind"] == "Role")
+    assert "secrets" not in {resource for rule in role["rules"] for resource in rule["resources"]}
+    assert {verb for rule in role["rules"] for verb in rule["verbs"]} == {"get", "list", "watch"}
+
+
+def test_agent_reconnect_retains_complete_history_and_long_tool_budget():
+    configmap = yaml.safe_load((HERMES / "agent-configmap.yaml").read_text())
+    config = yaml.safe_load(configmap["data"]["config.yaml"])
+    display = config["display"]
+    assert display["resume_exchanges"] >= 10000
+    assert display["resume_max_user_chars"] >= 10000000
+    assert display["resume_max_assistant_chars"] >= 10000000
+    assert config["agent"]["max_turns"] == 180
+    assert config["delegation"]["max_iterations"] == 120
+
+
+def test_auth_patch_honors_explicit_shared_store(tmp_path: Path):
+    source = tmp_path / "auth.py"
+    destination = tmp_path / "patched/auth.py"
+    source.write_text(
+        'from pathlib import Path\nimport os\n\ndef _auth_file_path() -> Path:\n    path = get_hermes_home() / "auth.json"\n    return path\n',
+        encoding="utf-8",
+    )
+    auth_patch.patch(source, destination)
+    content = destination.read_text()
+    assert 'os.environ.get("HERMES_AUTH_FILE"' in content
+
+
+def test_auth_patch_fails_closed_on_upstream_drift(tmp_path: Path):
+    source = tmp_path / "auth.py"
+    source.write_text("def changed():\n    pass\n", encoding="utf-8")
+    with pytest.raises(RuntimeError, match="context changed"):
+        auth_patch.patch(source, tmp_path / "patched.py")
+
+
+def test_ttyd_clipboard_and_reconnect_patch_remain_enabled():
+    source = '<html><body><script>document.execCommand("copy")</script></body></html>'
+    content = ttyd_patch.patch_html(source)
+    assert 'id="atlas-ttyd-clipboard"' in content
+    assert "navigator.clipboard.writeText(text)" in content
+    assert "class AtlasRecoveringWebSocket" in content
+    assert "window.location.reload()" in content
+    assert "event.stopImmediatePropagation()" in content
+
+
+def test_ttyd_patch_fails_closed_on_upstream_drift():
+    with pytest.raises(RuntimeError, match="context changed"):
+        ttyd_patch.patch_html("<html><body>changed</body></html>")

testing/tests/test_hermes_herdr.py

@@ -1,705 +0,0 @@
-"""Focused tests for Hermes-to-Herdr routing and the shared auth patch."""
-
-from __future__ import annotations
-
-import importlib.util
-import sys
-from pathlib import Path
-
-import pytest
-import yaml
-
-
-SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts"
-HERMES = Path(__file__).parents[2] / "services/hermes"
-
-
-def _load(name: str):
-    spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py")
-    assert spec and spec.loader
-    module = importlib.util.module_from_spec(spec)
-    sys.modules[spec.name] = module
-    spec.loader.exec_module(module)
-    return module
-
-
-dispatch = _load("herdr_dispatch")
-tab_router = _load("herdr_tab_router")
-auth_patch = _load("patch_hermes_auth")
-ttyd_patch = _load("patch_ttyd_index")
-client_config = _load("configure_agent_clients")
-
-
-def test_herdr_plan_chooses_task_shape_and_caps_effort():
-    status = {
-        "routes": {
-            "codex-high": [
-                "openai-codex/gpt-5.6-sol",
-                "anthropic/claude-opus-5",
-                "custom/qwen2.5:14b-instruct-q4_0",
-            ],
-            "claude-medium": [
-                "anthropic/claude-sonnet-5",
-                "openai-codex/gpt-5.6-terra",
-            ],
-        }
-    }
-    implementation = dispatch.select_plan(status, "implementation", "high")
-    architecture = dispatch.select_plan(status, "architecture", "medium")
-    assert implementation["worker"] == "codex"
-    assert implementation["model"] == "gpt-5.6-sol"
-    assert architecture["worker"] == "claude"
-    assert architecture["model"] == "claude-sonnet-5"
-    with pytest.raises(ValueError, match="effort"):
-        dispatch.select_plan(status, "review", "max")
-
-
-def test_herdr_auto_classifies_each_cli_worker_with_jetson_router(
-    tmp_path: Path,
-):
-    router = tmp_path / "auto_router.py"
-    router.write_text(
-        """
-class Decision:
-    shape = "review"
-    effort = "xhigh"
-    provider = "claude"
-    classifier = "jetson"
-    reason = "bounded local vote"
-    latency_ms = 17
-
-def classify_task(prompt):
-    assert "security review" in prompt
-    return Decision()
-""",
-        encoding="utf-8",
-    )
-    status = {
-        "routes": {
-            "claude-xhigh": [
-                "anthropic/claude-opus-5",
-                "openai-codex/gpt-5.6-sol",
-            ]
-        }
-    }
-
-    plan = dispatch.select_auto_plan(
-        status,
-        "Perform the final security review.",
-        router,
-    )
-
-    assert plan["worker"] == "claude"
-    assert plan["model"] == "claude-opus-5"
-    assert plan["effort"] == "xhigh"
-    assert plan["classifier"] == "jetson"
-    assert plan["classification_latency_ms"] == 17
-
-
-def test_claude_worker_waits_for_prompt_readiness(tmp_path: Path, monkeypatch):
-    herdr = tmp_path / "herdr"
-    herdr.touch()
-    project = tmp_path / "project"
-    project.mkdir()
-    calls = []
-
-    def fake_run(command, env):
-        calls.append(command)
-        if command[1:3] == ["workspace", "create"]:
-            return {"result": {"root_pane": {"pane_id": "w2:p1"}}}
-        return {"result": {"ok": True}}
-
-    monkeypatch.setattr(dispatch, "HERDR_BIN", herdr)
-    monkeypatch.setattr(dispatch, "_run", fake_run)
-    plan = {
-        "worker": "claude",
-        "model": "claude-haiku-4-5-20251001",
-        "effort": "low",
-    }
-
-    dispatch.launch_worker(plan, project, "review", "Check the implementation.")
-
-    ready = calls[-2]
-    assert ready[1:] == [
-        "pane",
-        "wait-output",
-        "w2:p1",
-        "--match",
-        "Claude Code",
-        "--source",
-        "recent",
-        "--lines",
-        "120",
-        "--timeout",
-        "120000",
-    ]
-    prompt = calls[-1]
-    assert prompt[1:5] == [
-        "agent",
-        "prompt",
-        "project-claude-review",
-        "Check the implementation.",
-    ]
-    assert prompt[-9:] == [
-        "--wait",
-        "--until",
-        "working",
-        "--until",
-        "done",
-        "--until",
-        "blocked",
-        "--timeout",
-        "15000",
-    ]
-
-    workspace_create = calls[0]
-    assert workspace_create[-3:] == ["--label", "project-claude-review", "--no-focus"]
-    start = calls[1]
-    assert start[1:4] == ["agent", "start", "project-claude-review"]
-    assert start[-4:] == [
-        "--model",
-        "claude-haiku-4-5-20251001",
-        "--effort",
-        "low",
-    ]
-    assert "--permission-mode" not in start
-
-
-def test_codex_worker_leaves_inner_policy_to_unattended_wrapper(tmp_path: Path, monkeypatch):
-    herdr = tmp_path / "herdr"
-    herdr.touch()
-    project = tmp_path / "cassandra"
-    project.mkdir()
-    auth = tmp_path / "auth.json"
-    auth.write_text("{}", encoding="utf-8")
-    calls = []
-
-    def fake_run(command, env):
-        calls.append(command)
-        if command[1:3] == ["workspace", "create"]:
-            return {"result": {"root_pane": {"pane_id": "w3:p1"}}}
-        return {"result": {"ok": True}}
-
-    monkeypatch.setattr(dispatch, "HERDR_BIN", herdr)
-    monkeypatch.setattr(dispatch, "CODEX_AUTH", auth)
-    monkeypatch.setattr(dispatch, "_run", fake_run)
-    plan = {"worker": "codex", "model": "gpt-5.6-sol", "effort": "high"}
-
-    dispatch.launch_worker(plan, project, "api-fix", None)
-
-    assert calls[0][-3:] == ["--label", "cassandra-codex-api-fix", "--no-focus"]
-    start = calls[1]
-    assert start[1:4] == ["agent", "start", "cassandra-codex-api-fix"]
-    assert "--dangerously-bypass-approvals-and-sandbox" not in start
-    assert "--dangerously-bypass-hook-trust" not in start
-    assert "--approve-for-me" not in start
-    assert "--sandbox" not in start
-    assert not any("on-request" in value for value in start)
-
-
-def test_provider_wrappers_force_unattended_modes():
-    codex = (HERMES / "scripts/codex").read_text(encoding="utf-8")
-    claude = (HERMES / "scripts/claude").read_text(encoding="utf-8")
-
-    assert "/opt/data/tools/bin/codex" in codex
-    assert "--dangerously-bypass-approvals-and-sandbox" in codex
-    assert "--dangerously-bypass-hook-trust" in codex
-    assert "/opt/data/tools/bin/claude" in claude
-    assert "--dangerously-skip-permissions" in claude
-    assert "--permission-mode bypassPermissions" in claude
-    assert "--autocompact auto" in claude
-
-
-def test_claude_resume_prompt_is_persistently_suppressed(tmp_path: Path):
-    state = tmp_path / ".claude.json"
-    state.write_text('{"promptQueueUseCount": 4}\n', encoding="utf-8")
-
-    client_config.configure_claude_state(state)
-
-    value = yaml.safe_load(state.read_text(encoding="utf-8"))
-    assert value["promptQueueUseCount"] == 4
-    assert value["bypassPermissionsModeAccepted"] is True
-    assert value["resumeReturnDismissed"] is True
-    assert state.stat().st_mode & 0o777 == 0o600
-
-
-def test_claude_unattended_state_is_created_for_a_new_home(tmp_path: Path):
-    state = tmp_path / ".claude" / ".claude.json"
-
-    client_config.configure_claude_state(state)
-
-    value = yaml.safe_load(state.read_text(encoding="utf-8"))
-    assert value == {
-        "bypassPermissionsModeAccepted": True,
-        "resumeReturnDismissed": True,
-    }
-
-
-def test_agent_config_uses_bounded_noninteractive_approvals():
-    configmap = yaml.safe_load((HERMES / "agent-configmap.yaml").read_text())
-    config = yaml.safe_load(configmap["data"]["config.yaml"])
-
-    assert config["approvals"]["mode"] == "off"
-    denied = config["approvals"]["deny"]
-    assert "*kubectl apply*" in denied
-    assert "*git push --force*" in denied
-    assert "*git reset --hard*" in denied
-
-
-def test_auth_patch_honors_explicit_shared_store(tmp_path: Path):
-    source = tmp_path / "auth.py"
-    destination = tmp_path / "patched" / "auth.py"
-    source.write_text(
-        "from pathlib import Path\nimport os\n\n"
-        "def _auth_file_path() -> Path:\n"
-        "    path = get_hermes_home() / \"auth.json\"\n"
-        "    return path\n",
-        encoding="utf-8",
-    )
-    auth_patch.patch(source, destination)
-    content = destination.read_text(encoding="utf-8")
-    assert 'os.environ.get("HERMES_AUTH_FILE"' in content
-    assert "Path(configured)" in content
-
-
-def test_auth_patch_fails_closed_on_upstream_drift(tmp_path: Path):
-    source = tmp_path / "auth.py"
-    source.write_text("def changed():\n    pass\n", encoding="utf-8")
-    with pytest.raises(RuntimeError, match="context changed"):
-        auth_patch.patch(source, tmp_path / "patched.py")
-
-
-def test_ttyd_clipboard_patch_uses_system_clipboard_and_preserves_interrupt():
-    source = (
-        '<html><body><script>document.execCommand("copy")</script></body></html>'
-    )
-    content = ttyd_patch.patch_html(source)
-
-    assert 'id="atlas-ttyd-clipboard"' in content
-    assert "navigator.clipboard.writeText(text)" in content
-    assert "term.hasSelection()" in content
-    assert "event.stopImmediatePropagation()" in content
-    assert 'document.execCommand("copy")' not in content
-    assert "document.execCommand('copy')" in content
-
-
-def test_ttyd_client_patch_recovers_with_a_fresh_authenticated_page():
-    source = (
-        '<html><body><script>document.execCommand("copy")</script></body></html>'
-    )
-    content = ttyd_patch.patch_html(source)
-
-    assert "class AtlasRecoveringWebSocket" in content
-    assert "new URL('token', window.location.href)" in content
-    assert "credentials: 'same-origin'" in content
-    assert "response.type === 'opaqueredirect'" in content
-    assert "typeof payload.token === 'string'" in content
-    assert "window.location.reload()" in content
-    assert "Reconnecting terminal automatically..." in content
-    assert "if (!recoveryStarted || event.key !== 'Enter') return" in content
-    assert content.index('id="atlas-ttyd-clipboard"') < content.index(
-        '<script>void 0</script>'
-    )
-
-
-def test_ttyd_clipboard_patch_fails_closed_on_upstream_drift():
-    with pytest.raises(RuntimeError, match="context changed"):
-        ttyd_patch.patch_html("<html><body>changed</body></html>")
-
-
-def test_agent_tab_router_starts_hermes_in_matching_project(tmp_path: Path):
-    projects = tmp_path / "projects"
-    cassandra = projects / "cassandra"
-    cassandra.mkdir(parents=True)
-    calls = []
-
-    def fake_run(command):
-        calls.append(command)
-        if command[1:3] == ["tab", "list"]:
-            return {
-                "result": {
-                    "tabs": [
-                        {
-                            "tab_id": "w2:t2",
-                            "workspace_id": "w2",
-                            "label": "Cassandra",
-                        }
-                    ]
-                }
-            }
-        if command[1:3] == ["pane", "list"]:
-            return {
-                "result": {
-                    "panes": [
-                        {
-                            "pane_id": "w2:p1",
-                            "tab_id": "w2:t1",
-                            "workspace_id": "w2",
-                            "agent": "hermes",
-                        },
-                        {
-                            "pane_id": "w2:p2",
-                            "tab_id": "w2:t2",
-                            "workspace_id": "w2",
-                            "foreground_cwd": str(tmp_path),
-                        },
-                    ]
-                }
-            }
-        return {"result": {"ok": True}}
-
-    started = tab_router.route_unmanaged_tabs("w2", fake_run, projects)
-
-    assert started == ["w2:p2"]
-    assert calls[-2][1:4] == ["pane", "run", "w2:p2"]
-    assert calls[-2][-1] == f"cd {cassandra.resolve()}"
-    assert calls[-1][1:] == [
-        "agent",
-        "start",
-        "cassandra-p2",
-        "--kind",
-        "hermes",
-        "--pane",
-        "w2:p2",
-        "--timeout",
-        "120000",
-    ]
-
-
-def test_agent_tab_router_only_targets_coordinator_workspace():
-    calls = []
-
-    def fake_run(command):
-        calls.append(command)
-        return {
-            "result": {
-                "workspaces": [
-                    {"workspace_id": "w2", "label": "coordinator"},
-                    {"workspace_id": "w3", "label": "codex-worker"},
-                ]
-            }
-        }
-
-    assert tab_router.find_workspace("coordinator", fake_run) == "w2"
-    assert calls == [[str(tab_router.HERDR_BIN), "workspace", "list"]]
-
-
-def test_agent_tab_router_detaches_workers_without_closing_sessions():
-    calls = []
-
-    def fake_run(command):
-        calls.append(command)
-        if command[1:3] == ["tab", "list"]:
-            return {
-                "result": {
-                    "tabs": [
-                        {
-                            "workspace_id": "w2",
-                            "tab_id": "w2:t2",
-                            "label": "cassandra",
-                        }
-                    ]
-                }
-            }
-        if command[1:3] == ["pane", "list"]:
-            return {
-                "result": {
-                    "panes": [
-                        {
-                            "workspace_id": "w2",
-                            "tab_id": "w2:t2",
-                            "pane_id": "w2:p2",
-                            "agent": "hermes",
-                        },
-                        {
-                            "workspace_id": "w2",
-                            "tab_id": "w2:t2",
-                            "pane_id": "w2:p8",
-                            "agent": "claude",
-                        },
-                        {
-                            "workspace_id": "w2",
-                            "tab_id": "w2:t2",
-                            "pane_id": "w2:p6",
-                            "agent": "codex",
-                        },
-                        {
-                            "workspace_id": "w2",
-                            "tab_id": "w2:t2",
-                            "pane_id": "w2:p7",
-                            "agent": "hermes",
-                        },
-                    ]
-                }
-            }
-        if command[1:3] == ["agent", "list"]:
-            return {
-                "result": {
-                    "agents": [
-                        {"workspace_id": "w2", "pane_id": "w2:p2", "name": "cassandra-p2"},
-                        {"workspace_id": "w2", "pane_id": "w2:p8", "name": "cassandra-claude-review"},
-                        {"workspace_id": "w2", "pane_id": "w2:p6", "name": "cassandra-codex-review"},
-                        {"workspace_id": "w2", "pane_id": "w2:p7", "name": "cassandra-p7"},
-                    ]
-                }
-            }
-        return {"result": {"ok": True}}
-
-    moved = tab_router.isolate_worker_panes("w2", fake_run)
-
-    assert moved == ["w2:p8", "w2:p6", "w2:p7"]
-    move_calls = calls[-3:]
-    assert move_calls[0][1:] == [
-        "pane",
-        "move",
-        "w2:p8",
-        "--new-workspace",
-        "--label",
-        "cassandra-claude-review",
-        "--no-focus",
-    ]
-    assert move_calls[1][-2:] == ["cassandra-codex-review", "--no-focus"]
-    assert move_calls[2][-2:] == ["cassandra-hermes-p7", "--no-focus"]
-    assert not any("close" in command for command in move_calls)
-
-
-def test_detached_worker_label_falls_back_to_parent_provider_and_pane():
-    pane = {"pane_id": "w4:p9", "agent": "codex"}
-
-    assert tab_router._worker_label("Work", pane, {}) == "work-codex-p9"
-
-
-def test_agent_ttyd_defers_identity_to_owner_only_oauth_boundary():
-    deployment = yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())
-    containers = deployment["spec"]["template"]["spec"]["containers"]
-    ttyd = next(container for container in containers if container["name"] == "herdr-tui")
-    command = ttyd["args"][0]
-
-    assert "--check-origin" in command
-    assert "--auth-header" not in command
-    assert "--index /ttyd-index/index.html" in command
-    assert "--client-option disableReconnect=true" in command
-
-    init_names = {
-        container["name"]
-        for container in deployment["spec"]["template"]["spec"]["initContainers"]
-    }
-    assert "prepare-ttyd-index" in init_names
-
-    oauth_documents = [
-        document
-        for document in yaml.safe_load_all((HERMES / "oauth2-proxy.yaml").read_text())
-        if document
-    ]
-    oauth = next(
-        document
-        for document in oauth_documents
-        if document["kind"] == "Deployment"
-        and document["metadata"]["name"] == "oauth2-proxy-hermes-agent"
-    )
-    oauth_args = oauth["spec"]["template"]["spec"]["containers"][0]["args"]
-    assert "--authenticated-emails-file=/etc/oauth2-proxy/allowed-emails" in oauth_args
-    assert "--proxy-websockets=true" in oauth_args
-
-    network_documents = [
-        document
-        for document in yaml.safe_load_all((HERMES / "networkpolicy.yaml").read_text())
-        if document
-    ]
-    isolation = next(
-        document
-        for document in network_documents
-        if document["kind"] == "NetworkPolicy"
-        and document["metadata"]["name"] == "hermes-agent-isolation"
-    )
-    ingress = isolation["spec"]["ingress"]
-    assert ingress == [
-        {
-            "from": [{"podSelector": {"matchLabels": {"app": "oauth2-proxy-hermes-agent"}}}],
-            "ports": [{"protocol": "TCP", "port": 7681}],
-        }
-    ]
-
-
-def test_hermes_oauth_boundaries_use_current_hardened_proxy():
-    documents = [
-        document
-        for document in yaml.safe_load_all((HERMES / "oauth2-proxy.yaml").read_text())
-        if document
-    ]
-    deployments = [
-        document
-        for document in documents
-        if document["kind"] == "Deployment"
-        and document["metadata"]["name"].startswith("oauth2-proxy-hermes-")
-    ]
-
-    assert len(deployments) == 3
-    for deployment in deployments:
-        container = deployment["spec"]["template"]["spec"]["containers"][0]
-        args = container["args"]
-        assert "v7.15.3@sha256:10a1165743a192e" in container["image"]
-        assert "--cookie-csrf-expire=10m" in args
-        assert "--cookie-csrf-per-request=true" in args
-        assert "--cookie-csrf-per-request-limit=8" in args
-        assert "--trusted-proxy-ip=10.42.0.0/16" in args
-
-
-def test_agent_installs_hermes_integration_before_startup():
-    deployment = yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())
-    pod = deployment["spec"]["template"]["spec"]
-    init_config = next(
-        container for container in pod["initContainers"] if container["name"] == "init-config"
-    )
-    assert "ln -s /opt/data /opt/data/home/.hermes" in init_config["command"][-1]
-
-    installer = next(
-        container
-        for container in pod["initContainers"]
-        if container["name"] == "install-herdr-integrations"
-    )
-    command = installer["command"][-1]
-    assert "herdr integration install codex" in command
-    assert "herdr integration install claude" in command
-    assert "herdr integration install hermes" in command
-    assert "|| true" not in command
-
-    containers = {container["name"]: container for container in pod["containers"]}
-    for name in ("herdr-tui", "herdr-server"):
-        env = {item["name"]: item["value"] for item in containers[name]["env"]}
-        assert "/opt/hermes/.venv/bin" in env["PATH"].split(":")
-
-    server_env = {
-        item["name"]: item["value"] for item in containers["herdr-server"]["env"]
-    }
-    assert "/opt/data/home/.local/bin" in server_env["PATH"].split(":")
-
-    for name in ("hermes", "herdr-server"):
-        env = {item["name"]: item["value"] for item in containers[name]["env"]}
-        assert env["AGENT_BROWSER_EXECUTABLE_PATH"].endswith("/chrome-linux/headless_shell")
-        assert "--no-sandbox" in env["AGENT_BROWSER_ARGS"]
-
-    server_command = containers["herdr-server"]["command"][-1]
-    assert "herdr server" in server_command
-    assert 'agent.get("agent") == "hermes"' in server_command
-    assert 'herdr pane close "${pane}"' in server_command
-    assert "herdr agent start coordinator" not in server_command
-    assert "/opt/coordinator/herdr_tab_router.py" not in server_command
-
-    tui_command = containers["herdr-tui"]["args"][0]
-    assert "/usr/bin/tmux new-session -A" in tui_command
-    assert "-s hermes-coordinator" in tui_command
-    assert "/opt/hermes/.venv/bin/hermes" in tui_command
-    assert "--continue" in tui_command
-    assert "--tui" in tui_command
-    assert "--yolo" in tui_command
-    assert "--accept-hooks" in tui_command
-    assert "--checkpoints" not in tui_command
-
-
-def test_agent_mounts_auto_router_into_both_hermes_runtimes():
-    deployment = yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())
-    pod = deployment["spec"]["template"]["spec"]
-    containers = {container["name"]: container for container in pod["containers"]}
-
-    for name in ("hermes", "herdr-server"):
-        mounts = {
-            mount["mountPath"]: mount["name"]
-            for mount in containers[name]["volumeMounts"]
-        }
-        assert mounts["/opt/data/plugins/auto-router"] == "auto-router-plugin"
-
-    volume = next(
-        item for item in pod["volumes"] if item["name"] == "auto-router-plugin"
-    )
-    assert volume["configMap"]["name"] == "hermes-auto-router-plugin"
-
-
-def test_agent_has_cassandra_readonly_kubernetes_context():
-    deployment = yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())
-    pod = deployment["spec"]["template"]["spec"]
-    containers = {container["name"]: container for container in pod["containers"]}
-
-    installer = next(
-        container
-        for container in pod["initContainers"]
-        if container["name"] == "install-agent-tools"
-    )
-    install_command = installer["command"][-1]
-    assert "v1.33.3/bin/linux/arm64/kubectl" in install_command
-    assert "3d514dbae5dc8c09f773df0ef0f5d449" in install_command
-
-    for name in ("hermes", "herdr-server"):
-        env = {item["name"]: item["value"] for item in containers[name]["env"]}
-        assert env["KUBECONFIG"] == "/opt/data/home/.kube/config"
-        mounts = {
-            mount["mountPath"]: mount for mount in containers[name]["volumeMounts"]
-        }
-        assert mounts["/opt/data/home/.kube/config"]["name"] == "kubeconfig"
-        assert mounts["/opt/data/home/.kube/config"]["readOnly"] is True
-
-    config = yaml.safe_load((HERMES / "agent-kubeconfig.yaml").read_text())
-    assert config["current-context"] == "cassandra-readonly"
-    assert config["contexts"][0]["context"]["namespace"] == "cassandra"
-    assert config["users"][0]["user"] == {
-        "tokenFile": "/var/run/secrets/kubernetes.io/serviceaccount/token"
-    }
-
-    rbac_path = HERMES.parent / "cassandra" / "hermes-agent-rbac.yaml"
-    documents = [document for document in yaml.safe_load_all(rbac_path.read_text()) if document]
-    role = next(document for document in documents if document["kind"] == "Role")
-    binding = next(document for document in documents if document["kind"] == "RoleBinding")
-    assert role["metadata"]["namespace"] == "cassandra"
-    assert binding["metadata"]["namespace"] == "cassandra"
-    assert binding["subjects"] == [
-        {"kind": "ServiceAccount", "name": "hermes-agent", "namespace": "hermes"}
-    ]
-    assert "secrets" not in {
-        resource for rule in role["rules"] for resource in rule["resources"]
-    }
-    assert {
-        verb for rule in role["rules"] for verb in rule["verbs"]
-    } == {"get", "list", "watch"}
-
-
-def test_agent_reconnect_renders_complete_conversation_history():
-    documents = [
-        document
-        for document in yaml.safe_load_all((HERMES / "agent-configmap.yaml").read_text())
-        if document
-    ]
-    config_map = next(
-        document
-        for document in documents
-        if document["kind"] == "ConfigMap"
-        and document["metadata"]["name"] == "hermes-agent-config"
-    )
-    config = yaml.safe_load(config_map["data"]["config.yaml"])
-    display = config["display"]
-
-    assert display["resume_exchanges"] >= 10000
-    assert display["resume_max_user_chars"] >= 10000000
-    assert display["resume_max_assistant_chars"] >= 10000000
-    assert display["resume_max_assistant_lines"] >= 1000000
-
-
-def test_agent_coordinator_has_a_long_running_tool_budget():
-    configmap = yaml.safe_load((HERMES / "agent-configmap.yaml").read_text())
-    config = yaml.safe_load(configmap["data"]["config.yaml"])
-
-    assert config["agent"]["max_turns"] == 180
-    assert config["delegation"] == {
-        "max_concurrent_children": 4,
-        "max_iterations": 120,
-        "max_spawn_depth": 2,
-        "orchestrator_enabled": True,
-        "subagent_auto_approve": True,
-    }
-    assert config["tool_loop_guardrails"]["hard_stop_enabled"] is True
-    for platform in ("cli", "api_server"):
-        tools = config["platform_toolsets"][platform]
-        assert "browser" in tools
-        assert "delegation" in tools
-        assert "memory" in tools
-        assert "terminal" in tools
-        assert "vision" in tools