984 lines
36 KiB
Python
984 lines
36 KiB
Python
#!/usr/bin/env python3
|
|
"""Run durable Codex and Claude workers from Hermes' authoritative Kanban."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import concurrent.futures
|
|
import json
|
|
import os
|
|
import re
|
|
import selectors
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import uuid
|
|
import urllib.error
|
|
import urllib.request
|
|
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"))
|
|
SWITCHYARD_URL = os.environ.get(
|
|
"HERMES_SWITCHYARD_URL",
|
|
"http://hermes-switchyard.hermes.svc.cluster.local:9005/v1/chat/completions",
|
|
)
|
|
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
|
|
PROVIDER_HEALTH_MAX_AGE_SECONDS = 5 * 60
|
|
PROVIDER_HEALTH_PATHS = {
|
|
"codex": DATA_ROOT / "provider-health/codex.json",
|
|
"claude": DATA_ROOT / "provider-health/claude.json",
|
|
}
|
|
WORKTREE_LOCK = threading.Lock()
|
|
BOARD_CORRUPTION_ERRORS: dict[str, str] = {}
|
|
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"
|
|
NO_CODEX_THREAD = "no rollout found for thread id"
|
|
|
|
|
|
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 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 fresh_unavailable_provider(now: float | None = None) -> str | None:
|
|
"""Return one recently proven-down provider for automatic route exclusion."""
|
|
current = time.time() if now is None else now
|
|
unavailable: list[str] = []
|
|
for provider, path in PROVIDER_HEALTH_PATHS.items():
|
|
health = load_json(path)
|
|
try:
|
|
age = current - path.stat().st_mtime
|
|
except OSError:
|
|
continue
|
|
if (
|
|
0 <= age <= PROVIDER_HEALTH_MAX_AGE_SECONDS
|
|
and health.get("state") == "unavailable"
|
|
):
|
|
unavailable.append(provider)
|
|
# If both providers are down, keep the normal attempt/fallback path so the
|
|
# card records authoritative current errors instead of trusting snapshots.
|
|
return unavailable[0] if len(unavailable) == 1 else None
|
|
|
|
|
|
def _decode_worker_target(value: str) -> tuple[str, str, str]:
|
|
"""Decode the selected model header emitted by a worker decision route."""
|
|
parts = value.split("/", 3)
|
|
if len(parts) != 4 or parts[0] != "worker":
|
|
raise RuntimeError(f"invalid Switchyard worker target: {value}")
|
|
provider, model, effort = parts[1:]
|
|
if provider not in {"codex", "claude"} or effort not in EFFORTS:
|
|
raise RuntimeError(f"unsupported Switchyard worker target: {value}")
|
|
return provider, model, effort
|
|
|
|
|
|
def select_route(
|
|
prompt: str,
|
|
assignee: str,
|
|
*,
|
|
exclude_provider: str | None = None,
|
|
exclude_reason: str | None = None,
|
|
switchyard_url: str = SWITCHYARD_URL,
|
|
open_request: Callable[..., Any] = urllib.request.urlopen,
|
|
) -> Route:
|
|
"""Ask Switchyard to select one native CLI worker at this boundary."""
|
|
started = time.monotonic()
|
|
manual_provider, manual_effort = parse_assignee(assignee)
|
|
if manual_provider and manual_effort:
|
|
route_id = f"atlas/worker/manual/{manual_provider}/{manual_effort}"
|
|
source = "switchyard-manual"
|
|
else:
|
|
route_id = "atlas/worker/auto/maximum"
|
|
source = "switchyard-classifier"
|
|
context = prompt
|
|
if exclude_provider:
|
|
reason = exclude_reason or "failed or exhausted capacity at this boundary"
|
|
context += (
|
|
f"\n\nRouting constraint: the {exclude_provider} provider {reason}. "
|
|
"Do not select it."
|
|
)
|
|
payload = json.dumps(
|
|
{
|
|
"model": route_id,
|
|
"messages": [{"role": "user", "content": context}],
|
|
"stream": False,
|
|
"max_tokens": 1,
|
|
}
|
|
).encode("utf-8")
|
|
request = urllib.request.Request(
|
|
switchyard_url,
|
|
data=payload,
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
try:
|
|
with open_request(request, timeout=60) as response:
|
|
selected = str(response.headers.get("x-model-router-selected-model") or "")
|
|
rationale = str(response.headers.get("x-model-router-rationale") or "")
|
|
response_body = response.read()
|
|
except (OSError, urllib.error.URLError) as exc:
|
|
raise RuntimeError(f"Switchyard worker routing failed: {exc}") from exc
|
|
provider, model, effort = _decode_worker_target(selected)
|
|
# Switchyard preserves the stable tier target in the selection header and
|
|
# top-level response model. The worker broker's assistant content contains
|
|
# the steward-resolved provider model required by the native CLI.
|
|
try:
|
|
response_document = json.loads(response_body)
|
|
resolved_target = str(response_document["choices"][0]["message"]["content"] or "")
|
|
resolved_provider, resolved_model, resolved_effort = _decode_worker_target(resolved_target)
|
|
required_prefix = "gpt-" if provider == "codex" else "claude-"
|
|
if (
|
|
resolved_provider == provider
|
|
and resolved_effort == effort
|
|
and resolved_model.startswith(required_prefix)
|
|
):
|
|
model = resolved_model
|
|
except (AttributeError, IndexError, KeyError, RuntimeError, TypeError, ValueError, json.JSONDecodeError):
|
|
pass
|
|
if exclude_provider and provider == exclude_provider:
|
|
alternate = "claude" if provider == "codex" else "codex"
|
|
guarded = select_route(
|
|
prompt,
|
|
f"cli-{alternate}-{effort}",
|
|
switchyard_url=switchyard_url,
|
|
open_request=open_request,
|
|
)
|
|
return Route(
|
|
provider=guarded.provider,
|
|
model=guarded.model,
|
|
effort=guarded.effort,
|
|
profile=guarded.profile,
|
|
classifier=f"{source}-health-guard",
|
|
reason=(
|
|
f"Switchyard selected excluded {exclude_provider}; preserved "
|
|
f"its {effort} effort on healthy-provider route. {guarded.reason}"
|
|
),
|
|
latency_ms=int((time.monotonic() - started) * 1000),
|
|
fallback_chain=(),
|
|
)
|
|
return Route(
|
|
provider=provider,
|
|
model=model,
|
|
effort=effort,
|
|
profile=f"{provider}-{effort}",
|
|
classifier=source,
|
|
reason=rationale or f"Switchyard selected {selected}",
|
|
latency_ms=int((time.monotonic() - started) * 1000),
|
|
fallback_chain=(),
|
|
)
|
|
|
|
|
|
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,
|
|
start_new_session=True,
|
|
)
|
|
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:
|
|
_terminate_worker_process(process)
|
|
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"):
|
|
_terminate_worker_process(process)
|
|
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
|
|
_terminate_worker_process(process)
|
|
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 _terminate_worker_process(process: subprocess.Popen[str]) -> None:
|
|
"""Reap a worker and every subprocess in its isolated process group."""
|
|
if process.poll() is None:
|
|
try:
|
|
os.killpg(process.pid, signal.SIGTERM)
|
|
except ProcessLookupError:
|
|
pass
|
|
try:
|
|
process.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
try:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
process.wait(timeout=10)
|
|
|
|
# A provider can exit before one of its terminal or language-server children.
|
|
# The process group remains addressable after its leader exits, so reap those
|
|
# children as well before a retry is allowed to own the same worktree.
|
|
try:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
|
|
def _base_env() -> dict[str, str]:
|
|
"""Build a worker environment while preserving Vault-backed CLI homes."""
|
|
env = os.environ.copy()
|
|
env.update(
|
|
{
|
|
"HOME": str(DATA_ROOT / "home"),
|
|
"CODEX_HOME": os.environ.get("CODEX_HOME", "/runtime-access/codex"),
|
|
"CLAUDE_CONFIG_DIR": os.environ.get(
|
|
"CLAUDE_CONFIG_DIR", "/runtime-access/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,
|
|
)
|
|
if (
|
|
state.get("codex_thread_id")
|
|
and result.returncode != 0
|
|
and NO_CODEX_THREAD in result.output.lower()
|
|
):
|
|
state.pop("codex_thread_id", None)
|
|
atomic_json(state_file, state)
|
|
result_file.unlink(missing_ok=True)
|
|
result = stream_process(
|
|
_codex_command(route, prompt, workspace, state, result_file),
|
|
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")
|
|
excluded_provider = (
|
|
fresh_unavailable_provider() if assignee == "cli-auto" else None
|
|
)
|
|
route = select_route(
|
|
context,
|
|
assignee,
|
|
exclude_provider=excluded_provider,
|
|
exclude_reason="is unavailable according to fresh native health"
|
|
if excluded_provider
|
|
else None,
|
|
)
|
|
if excluded_provider:
|
|
kanban_db.add_comment(
|
|
conn,
|
|
task_id,
|
|
"cli-lane-runner",
|
|
f"Provider health guard excluded {excluded_provider} before automatic routing.",
|
|
)
|
|
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 _connect_healthy_board(kanban_db: Any, board: str) -> Any | None:
|
|
"""Open one board without letting localized corruption stop other lanes."""
|
|
try:
|
|
conn = kanban_db.connect(board=board)
|
|
except getattr(kanban_db, "KanbanDbCorruptError", ()) as error:
|
|
detail = str(error)
|
|
if BOARD_CORRUPTION_ERRORS.get(board) != detail:
|
|
print(
|
|
f"quarantining corrupt Kanban board {board!r}: {detail}",
|
|
file=sys.stderr,
|
|
flush=True,
|
|
)
|
|
BOARD_CORRUPTION_ERRORS[board] = detail
|
|
return None
|
|
BOARD_CORRUPTION_ERRORS.pop(board, None)
|
|
return conn
|
|
|
|
|
|
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 = _connect_healthy_board(kanban_db, board)
|
|
if conn is None:
|
|
continue
|
|
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 = _connect_healthy_board(kanban_db, board)
|
|
if conn is None:
|
|
continue
|
|
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())
|