#!/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. You have owner-level Kubernetes access in every namespace. Prefer Flux-tracked manifests for durable changes, but use kubectl, Flux, exec, port-forwarding, rollout operations, and existing Vault workflows when the objective or incident requires them. Persist any desired-state mutation back to Git. Do not force-push, hard-reset, clean untracked files, 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(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())