1294 lines
49 KiB
Python
1294 lines
49 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 sqlite3
|
|
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
|
|
|
|
import cli_lane_goal
|
|
|
|
|
|
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_AUTH_FAILURE_MAX_AGE_SECONDS = 12 * 60 * 60
|
|
KANBAN_STORAGE_ATTEMPTS = 5
|
|
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",
|
|
"findings",
|
|
"blockers",
|
|
],
|
|
"properties": {
|
|
"status": {
|
|
"type": "string",
|
|
"enum": sorted(cli_lane_goal.RESULT_STATUSES),
|
|
},
|
|
"summary": {"type": "string"},
|
|
"changed_files": {"type": "array", "items": {"type": "string"}},
|
|
"tests_run": {"type": "array", "items": {"type": "string"}},
|
|
"artifacts": {"type": "array", "items": {"type": "string"}},
|
|
"findings": {
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
"description": (
|
|
"Defects, risks, or observations discovered by a review or diagnosis. "
|
|
"Findings do not prevent the assigned review or diagnosis from completing."
|
|
),
|
|
},
|
|
"blockers": {
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
"description": (
|
|
"Concrete obstacles that prevent completion of the assigned task itself. "
|
|
"This must be empty when status is completed; review findings belong in findings."
|
|
),
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
@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
|
|
max_age = (
|
|
PROVIDER_AUTH_FAILURE_MAX_AGE_SECONDS
|
|
if health.get("authenticated") is False
|
|
else PROVIDER_HEALTH_MAX_AGE_SECONDS
|
|
)
|
|
if (
|
|
0 <= age <= max_age
|
|
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=incomplete when required work, tests, commands, commits, pushes, or verification are still running or remain to be done. Use status=blocked only when an obstacle prevents completion of the assigned task itself. Never use status=completed for a progress report. For review or diagnostic tasks, put defects and risks in findings; those findings can make the reviewed change unfit to ship without blocking completion of the review. The blockers array must be empty whenever status is completed. List changed files, tests run, durable artifact paths, findings, and task 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 cli_lane_goal.RESULT_STATUSES:
|
|
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 cli_lane_goal.RESULT_STATUSES:
|
|
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
|
|
known_descendants: dict[int, tuple[int, int]] = {}
|
|
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:
|
|
known_descendants.update(_descendant_processes(process.pid))
|
|
now = time.monotonic()
|
|
if now - started > max_runtime:
|
|
_terminate_worker_process(process, known_descendants)
|
|
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, known_descendants)
|
|
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, known_descendants)
|
|
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 _process_record(pid: int) -> tuple[int, int, int] | None:
|
|
"""Return one Linux process's parent, group, and start-time identity."""
|
|
try:
|
|
raw = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8")
|
|
fields = raw[raw.rfind(")") + 2 :].split()
|
|
return int(fields[1]), int(fields[2]), int(fields[19])
|
|
except (IndexError, OSError, ValueError):
|
|
return None
|
|
|
|
|
|
def _descendant_processes(root_pid: int) -> dict[int, tuple[int, int]]:
|
|
"""Snapshot descendants as pid -> (process group, start-time identity)."""
|
|
records: dict[int, tuple[int, int, int]] = {}
|
|
for entry in Path("/proc").iterdir():
|
|
if not entry.name.isdigit():
|
|
continue
|
|
pid = int(entry.name)
|
|
record = _process_record(pid)
|
|
if record is not None:
|
|
records[pid] = record
|
|
family = {root_pid}
|
|
changed = True
|
|
while changed:
|
|
changed = False
|
|
for pid, (parent, _, _) in records.items():
|
|
if pid not in family and parent in family:
|
|
family.add(pid)
|
|
changed = True
|
|
return {
|
|
pid: (records[pid][1], records[pid][2])
|
|
for pid in family
|
|
if pid != root_pid and pid in records
|
|
}
|
|
|
|
|
|
def _process_identity_matches(pid: int, start_time: int) -> bool:
|
|
record = _process_record(pid)
|
|
return record is not None and record[2] == start_time
|
|
|
|
|
|
def _signal_worker_tree(
|
|
root_pid: int,
|
|
descendants: dict[int, tuple[int, int]],
|
|
sig: signal.Signals,
|
|
) -> None:
|
|
"""Signal captured descendants across terminal-created process groups."""
|
|
groups = {root_pid}
|
|
for pid, (process_group, start_time) in descendants.items():
|
|
if _process_identity_matches(pid, start_time):
|
|
groups.add(process_group)
|
|
for process_group in groups:
|
|
try:
|
|
os.killpg(process_group, sig)
|
|
except ProcessLookupError:
|
|
pass
|
|
for pid, (_, start_time) in descendants.items():
|
|
if not _process_identity_matches(pid, start_time):
|
|
continue
|
|
try:
|
|
os.kill(pid, sig)
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
|
|
def _terminate_worker_process(
|
|
process: subprocess.Popen[str],
|
|
known_descendants: dict[int, tuple[int, int]] | None = None,
|
|
) -> None:
|
|
"""Reap a worker and terminal descendants that created their own groups."""
|
|
descendants = dict(known_descendants or {})
|
|
descendants.update(_descendant_processes(process.pid))
|
|
if process.poll() is None:
|
|
_signal_worker_tree(process.pid, descendants, signal.SIGTERM)
|
|
try:
|
|
process.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
|
|
# Provider terminal tools create their own sessions, so killing only the
|
|
# native CLI's process group leaves test/build subprocesses orphaned. The
|
|
# identity check prevents a reused PID from targeting an unrelated worker.
|
|
_signal_worker_tree(process.pid, descendants, signal.SIGKILL)
|
|
if process.poll() is None:
|
|
process.wait(timeout=10)
|
|
|
|
|
|
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 cli_lane_goal.RESULT_STATUSES:
|
|
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 _board_call(
|
|
kanban_db: Any,
|
|
board: str,
|
|
operation: Callable[[Any], Any],
|
|
) -> Any:
|
|
"""Run one bounded Kanban operation on a fresh, promptly closed connection."""
|
|
last_storage_error: Exception | None = None
|
|
for attempt in range(KANBAN_STORAGE_ATTEMPTS):
|
|
conn = None
|
|
try:
|
|
with kanban_db.scoped_current_board(board):
|
|
conn = kanban_db.connect(board=board)
|
|
result = operation(conn)
|
|
BOARD_CORRUPTION_ERRORS.pop(board, None)
|
|
return result
|
|
except (OSError, sqlite3.Error) as error:
|
|
last_storage_error = error
|
|
_record_board_access_error(board, error)
|
|
if attempt + 1 < KANBAN_STORAGE_ATTEMPTS:
|
|
time.sleep(0.2 * (attempt + 1))
|
|
finally:
|
|
if conn is not None:
|
|
conn.close()
|
|
assert last_storage_error is not None
|
|
raise last_storage_error
|
|
|
|
|
|
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
|
|
|
|
run_id = None
|
|
with kanban_db.scoped_current_board(board):
|
|
conn = kanban_db.connect(board=board)
|
|
try:
|
|
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")
|
|
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:
|
|
failure_reason = (
|
|
f"Direct CLI lane preparation failed: {type(error).__name__}: {error}"
|
|
)
|
|
conn.close()
|
|
_board_call(
|
|
kanban_db,
|
|
board,
|
|
lambda fresh: kanban_db.block_task(
|
|
fresh,
|
|
task_id,
|
|
reason=failure_reason,
|
|
kind="capability",
|
|
expected_run_id=run_id,
|
|
),
|
|
)
|
|
return
|
|
finally:
|
|
conn.close()
|
|
|
|
def heartbeat(note: str) -> bool:
|
|
try:
|
|
return bool(
|
|
_board_call(
|
|
kanban_db,
|
|
board,
|
|
lambda fresh: kanban_db.heartbeat_worker(
|
|
fresh,
|
|
task_id,
|
|
note=note,
|
|
expected_run_id=run_id,
|
|
),
|
|
)
|
|
)
|
|
except (OSError, sqlite3.Error):
|
|
# The claim TTL is deliberately long. Keep the expensive
|
|
# provider process alive during a transient volume stall and
|
|
# retry on the next heartbeat instead of losing its work.
|
|
return True
|
|
|
|
def comment(body: str) -> None:
|
|
try:
|
|
_board_call(
|
|
kanban_db,
|
|
board,
|
|
lambda fresh: kanban_db.add_comment(
|
|
fresh,
|
|
task_id,
|
|
"cli-lane-runner",
|
|
body,
|
|
),
|
|
)
|
|
except (OSError, sqlite3.Error):
|
|
# Route state is also written to the durable lane-state file;
|
|
# a later heartbeat or terminal result remains authoritative.
|
|
pass
|
|
|
|
try:
|
|
previous_route = state.get("current_route")
|
|
excluded_provider = (
|
|
fresh_unavailable_provider() if assignee == "cli-auto" else None
|
|
)
|
|
unavailable_provider = excluded_provider
|
|
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:
|
|
comment(
|
|
f"Provider health guard excluded {excluded_provider} before automatic routing.",
|
|
)
|
|
comment(
|
|
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)
|
|
comment(
|
|
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
|
|
)
|
|
goal_mode = bool(_task_value(task, "goal_mode", False))
|
|
goal_max_turns = max(1, int(_task_value(task, "goal_max_turns", 1) or 1))
|
|
goal_turn = max(1, int(state.get("goal_turn", 0) or 0) + 1)
|
|
deadline = time.monotonic() + max_runtime
|
|
while True:
|
|
state["goal_turn"] = goal_turn
|
|
atomic_json(state_file, state)
|
|
remaining = max(1, int(deadline - time.monotonic()))
|
|
result = run_provider(
|
|
route,
|
|
prompt,
|
|
workspace,
|
|
state,
|
|
state_file,
|
|
log_path,
|
|
heartbeat,
|
|
remaining,
|
|
)
|
|
if result.capacity_failure:
|
|
unavailable_provider = route.provider
|
|
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}",
|
|
)
|
|
comment(
|
|
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(1, int(deadline - time.monotonic())),
|
|
)
|
|
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"),
|
|
"goal_mode": goal_mode,
|
|
"goal_turn": goal_turn,
|
|
}
|
|
if structured:
|
|
for key in (
|
|
"changed_files",
|
|
"tests_run",
|
|
"artifacts",
|
|
"findings",
|
|
"blockers",
|
|
):
|
|
value = structured.get(key)
|
|
metadata[key] = value if isinstance(value, list) else []
|
|
|
|
completion_problem = None
|
|
if structured and result.returncode == 0:
|
|
completion_problem = cli_lane_goal.unfinished_result_reason(structured)
|
|
if (
|
|
structured.get("status") == "completed"
|
|
and completion_problem is None
|
|
and goal_mode
|
|
):
|
|
heartbeat("local goal-completion judge active")
|
|
rejection_history = state.get("goal_rejections", [])
|
|
if not isinstance(rejection_history, list):
|
|
rejection_history = []
|
|
judge_context = context
|
|
if goal_turn > 1 or rejection_history:
|
|
judge_context += (
|
|
"\n\nAuthoritative Hermes goal-controller evidence: "
|
|
f"current turn {goal_turn}/{goal_max_turns}; prior rejected "
|
|
f"reports: {json.dumps(rejection_history[-5:])}."
|
|
)
|
|
accepted, judge_reason = cli_lane_goal.judge_goal_completion(
|
|
judge_context,
|
|
structured,
|
|
)
|
|
metadata["goal_judge_reason"] = judge_reason
|
|
if not accepted:
|
|
completion_problem = f"local goal judge requested continuation: {judge_reason}"
|
|
|
|
if (
|
|
structured
|
|
and structured.get("status") == "completed"
|
|
and result.returncode == 0
|
|
and completion_problem is None
|
|
):
|
|
_board_call(
|
|
kanban_db,
|
|
board,
|
|
lambda fresh: kanban_db.complete_task(
|
|
fresh,
|
|
task_id,
|
|
result=json.dumps(structured, sort_keys=True),
|
|
summary=str(structured.get("summary") or "Completed"),
|
|
metadata=metadata,
|
|
expected_run_id=run_id,
|
|
),
|
|
)
|
|
break
|
|
|
|
can_continue = (
|
|
goal_mode
|
|
and completion_problem is not None
|
|
and goal_turn < goal_max_turns
|
|
and deadline - time.monotonic() > 30
|
|
)
|
|
if can_continue:
|
|
rejection_history = state.get("goal_rejections", [])
|
|
if not isinstance(rejection_history, list):
|
|
rejection_history = []
|
|
state["goal_rejections"] = [
|
|
*rejection_history[-4:],
|
|
completion_problem,
|
|
]
|
|
goal_turn += 1
|
|
comment(
|
|
f"Goal completion rejected; continuing turn {goal_turn}/{goal_max_turns}: {completion_problem}",
|
|
)
|
|
escalation_context = (
|
|
context
|
|
+ "\n\nThe previous worker turn missed its completion quality mark: "
|
|
+ completion_problem
|
|
+ "\nSelect a route that can finish and verify the remaining work."
|
|
)
|
|
excluded = unavailable_provider
|
|
if excluded is None and assignee == "cli-auto":
|
|
excluded = fresh_unavailable_provider()
|
|
next_route = select_route(
|
|
escalation_context,
|
|
assignee,
|
|
exclude_provider=excluded,
|
|
exclude_reason="is unavailable according to fresh native health"
|
|
if excluded
|
|
else None,
|
|
)
|
|
comment(
|
|
f"Goal route {goal_turn}/{goal_max_turns}: "
|
|
f"{next_route.provider}/{next_route.model} at {next_route.effort}; "
|
|
f"classifier={next_route.classifier}; {next_route.reason}",
|
|
)
|
|
handoff = (
|
|
git_handoff(workspace, result.output)
|
|
if next_route.provider != route.provider
|
|
else ""
|
|
)
|
|
route = next_route
|
|
prompt = build_prompt(
|
|
context,
|
|
workspace,
|
|
handoff
|
|
+ "\nGoal-loop continuation: the previous final report was rejected because "
|
|
+ completion_problem
|
|
+ ". Reinspect live state, finish the outstanding work, and return new final evidence.",
|
|
)
|
|
continue
|
|
|
|
reason = completion_problem
|
|
if reason is None and structured:
|
|
blockers = structured.get("blockers", [])
|
|
reason = "; ".join(str(item) for item in blockers)
|
|
reason = reason or str(structured.get("summary") or "")
|
|
if reason is None:
|
|
reason = result.output[-4000:]
|
|
_board_call(
|
|
kanban_db,
|
|
board,
|
|
lambda fresh: kanban_db.block_task(
|
|
fresh,
|
|
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,
|
|
),
|
|
)
|
|
break
|
|
except Exception as error:
|
|
failure_reason = f"Direct CLI lane failed: {type(error).__name__}: {error}"
|
|
_board_call(
|
|
kanban_db,
|
|
board,
|
|
lambda fresh: kanban_db.block_task(
|
|
fresh,
|
|
task_id,
|
|
reason=failure_reason,
|
|
kind="capability",
|
|
expected_run_id=run_id,
|
|
),
|
|
)
|
|
|
|
|
|
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 storage faults stop other lanes."""
|
|
try:
|
|
return kanban_db.connect(board=board)
|
|
except Exception as error:
|
|
_record_board_access_error(board, error)
|
|
return None
|
|
|
|
|
|
def _record_board_access_error(board: str, error: Exception) -> None:
|
|
"""Report one board access failure once while allowing other lanes to run."""
|
|
failure_kind = "storage" if isinstance(error, (OSError, sqlite3.Error)) else "access"
|
|
detail = f"{failure_kind} {type(error).__name__}: {error}"
|
|
if BOARD_CORRUPTION_ERRORS.get(board) == detail:
|
|
return
|
|
print(
|
|
f"temporarily skipping Kanban board {board!r}: {detail}",
|
|
file=sys.stderr,
|
|
flush=True,
|
|
)
|
|
BOARD_CORRUPTION_ERRORS[board] = detail
|
|
|
|
|
|
def recover_orphans() -> None:
|
|
"""Return external running tasks to ready after a runner/pod restart."""
|
|
from hermes_cli import kanban_db
|
|
|
|
try:
|
|
boards = kanban_db.list_boards(include_archived=False)
|
|
except Exception as error:
|
|
_record_board_access_error("board-registry", error)
|
|
return
|
|
for raw_board in boards:
|
|
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",
|
|
)
|
|
BOARD_CORRUPTION_ERRORS.pop(board, None)
|
|
except Exception as error:
|
|
_record_board_access_error(board, error)
|
|
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)
|
|
tasks = kanban_db.list_tasks(conn)
|
|
BOARD_CORRUPTION_ERRORS.pop(board, None)
|
|
for task in tasks:
|
|
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
|
|
except Exception as error:
|
|
_record_board_access_error(board, error)
|
|
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())
|
|
try:
|
|
newly_claimed = claim_ready(active, workers - len(futures))
|
|
BOARD_CORRUPTION_ERRORS.pop("board-registry", None)
|
|
except Exception as error:
|
|
_record_board_access_error("board-registry", error)
|
|
newly_claimed = []
|
|
for board, task_id in newly_claimed:
|
|
future = pool.submit(execute_claim, board, task_id)
|
|
futures[future] = (board, task_id)
|
|
time.sleep(5)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|