titan/atlas-iac
3
0
Fork 0
Code Issues Pull Requests 9 Actions Packages Projects Releases Wiki Activity

hermes: harden replay and decomposition races

Browse Source
This commit is contained in:
jenkins 2026-08-16 20:27:12 -03:00
parent 93b8e42d51
commit 18be595700
4 changed files with 1019 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
dockerfiles/hermes-execution-safety-regression.py
View File

@ -167,22 +167,45 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase):
self.assertEqual(completions.calls, 1)
self.assertEqual(self._status(task_id), "ready")
def test_execution_history_gained_during_llm_call_blocks_commit(self) -> None:
def _insert_run(self, task_id: str) -> None:
with kanban_db.connect_closing() as connection:
kanban_db._synthesize_ended_run(
connection,
task_id,
outcome="reclaimed",
summary="concurrent execution evidence",
)
def test_execution_history_gained_during_single_llm_call_blocks_commit(self) -> None:
task_id = kanban_db.create_task(
self.connection,
title="concurrent single objective",
triage=True,
)
completions = self._client(
{
"fanout": False,
"title": "redundant rewrite",
"body": "must not be committed",
"assignee": "default",
},
before_response=lambda: self._insert_run(task_id),
)
outcome = kanban_decompose.decompose_task(task_id, automatic=True)
self.assertFalse(outcome.ok)
self.assertIn("gained execution history", outcome.reason)
self.assertEqual(completions.calls, 1)
self.assertEqual(self._status(task_id), "triage")
def test_execution_history_gained_during_fanout_llm_call_blocks_commit(self) -> None:
task_id = kanban_db.create_task(
self.connection,
title="concurrent triage objective",
triage=True,
)
def add_run() -> None:
with kanban_db.connect_closing() as connection:
kanban_db._synthesize_ended_run(
connection,
task_id,
outcome="reclaimed",
summary="concurrent execution evidence",
)
completions = self._client(
{
"fanout": True,
@ -195,7 +218,7 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase):
}
],
},
before_response=add_run,
before_response=lambda: self._insert_run(task_id),
)
before = len(kanban_db.list_tasks(self.connection))

167
dockerfiles/patch-hermes-execution-safety.py
View File

@ -15,6 +15,106 @@ def replace_once(source: str, before: str, after: str, label: str) -> str:
source_root = Path(os.environ.get("HERMES_SOURCE_ROOT", "/opt/hermes"))
db_path = source_root / "hermes_cli/kanban_db.py"
db = db_path.read_text(encoding="utf-8")
specify_signature_before = '''def specify_triage_task(
conn: sqlite3.Connection,
task_id: str,
*,
title: Optional[str] = None,
body: Optional[str] = None,
assignee: Optional[str] = None,
author: Optional[str] = None,
) -> bool:
'''
specify_signature_after = '''def specify_triage_task(
conn: sqlite3.Connection,
task_id: str,
*,
title: Optional[str] = None,
body: Optional[str] = None,
assignee: Optional[str] = None,
author: Optional[str] = None,
require_no_runs: bool = False,
) -> bool:
'''
db = replace_once(
db,
specify_signature_before,
specify_signature_after,
"transactional specify signature",
)
specify_guard_before = ''' if existing is None:
return False
sets: list[str] = ["status = 'todo'"]
'''
specify_guard_after = ''' if existing is None:
return False
if require_no_runs and conn.execute(
"SELECT 1 FROM task_runs WHERE task_id = ? LIMIT 1",
(task_id,),
).fetchone() is not None:
return False
sets: list[str] = ["status = 'todo'"]
'''
db = replace_once(
db,
specify_guard_before,
specify_guard_after,
"transactional specify execution-history guard",
)
decompose_signature_before = '''def decompose_triage_task(
conn: sqlite3.Connection,
task_id: str,
*,
root_assignee: Optional[str],
children: list[dict],
author: Optional[str] = None,
auto_promote: bool = True,
) -> Optional[list[str]]:
'''
decompose_signature_after = '''def decompose_triage_task(
conn: sqlite3.Connection,
task_id: str,
*,
root_assignee: Optional[str],
children: list[dict],
author: Optional[str] = None,
auto_promote: bool = True,
require_no_runs: bool = False,
) -> Optional[list[str]]:
'''
db = replace_once(
db,
decompose_signature_before,
decompose_signature_after,
"transactional decomposition signature",
)
decompose_guard_before = ''' if root_row["status"] != "triage":
return None
tenant = root_row["tenant"]
'''
decompose_guard_after = ''' if root_row["status"] != "triage":
return None
if require_no_runs and conn.execute(
"SELECT 1 FROM task_runs WHERE task_id = ? LIMIT 1",
(task_id,),
).fetchone() is not None:
return None
tenant = root_row["tenant"]
'''
db = replace_once(
db,
decompose_guard_before,
decompose_guard_after,
"transactional decomposition execution-history guard",
)
db_path.write_text(db, encoding="utf-8")
decompose_path = source_root / "hermes_cli/kanban_decompose.py"
decompose = decompose_path.read_text(encoding="utf-8")
@ -87,17 +187,20 @@ decompose = replace_once(
"auto-decompose preflight guard",
)
single_before = ''' with kb.connect_closing() as conn:
ok = kb.specify_triage_task(
'''
single_after = ''' if automatic and _has_execution_history(task_id):
return DecomposeOutcome(
task_id,
False,
"task gained execution history and requires deliberate manual triage",
single_before = ''' author=audit_author,
)
with kb.connect_closing() as conn:
ok = kb.specify_triage_task(
if not ok:
'''
single_after = ''' author=audit_author,
require_no_runs=automatic,
)
if not ok:
if automatic and _has_execution_history(task_id):
return DecomposeOutcome(
task_id,
False,
"task gained execution history and requires deliberate manual triage",
)
'''
decompose = replace_once(
decompose,
@ -106,25 +209,43 @@ decompose = replace_once(
"auto-decompose single-task commit guard",
)
fanout_before = ''' try:
with kb.connect_closing() as conn:
child_ids = kb.decompose_triage_task(
fanout_before = ''' author=audit_author,
auto_promote=auto_promote,
)
'''
fanout_after = ''' if automatic and _has_execution_history(task_id):
return DecomposeOutcome(
task_id,
False,
"task gained execution history and requires deliberate manual triage",
)
try:
with kb.connect_closing() as conn:
child_ids = kb.decompose_triage_task(
fanout_after = ''' author=audit_author,
auto_promote=auto_promote,
require_no_runs=automatic,
)
'''
decompose = replace_once(
decompose,
fanout_before,
fanout_after,
"auto-decompose fanout commit guard",
"auto-decompose transactional fanout guard",
)
fanout_outcome_before = ''' if child_ids is None:
return DecomposeOutcome(
task_id, False, "task moved out of triage before decomposition",
)
'''
fanout_outcome_after = ''' if child_ids is None:
if automatic and _has_execution_history(task_id):
return DecomposeOutcome(
task_id,
False,
"task gained execution history and requires deliberate manual triage",
)
return DecomposeOutcome(
task_id, False, "task moved out of triage before decomposition",
)
'''
decompose = replace_once(
decompose,
fanout_outcome_before,
fanout_outcome_after,
"auto-decompose fanout rejection reason",
)
decompose_path.write_text(decompose, encoding="utf-8")

472
services/hermes/scripts/cli_lane_runner.py
View File

@ -4,12 +4,14 @@
from __future__ import annotations
import concurrent.futures
import hashlib
import json
import os
import re
import selectors
import signal
import sqlite3
import stat
import subprocess
import sys
import threading
@ -43,12 +45,17 @@ HEARTBEAT_SECONDS = 20
PROVIDER_HEALTH_MAX_AGE_SECONDS = 5 * 60
PROVIDER_AUTH_FAILURE_MAX_AGE_SECONDS = 12 * 60 * 60
KANBAN_STORAGE_ATTEMPTS = 5
ARTIFACT_GC_INTERVAL_SECONDS = 5 * 60
ARTIFACT_RETENTION_AGE_SECONDS = 30 * 24 * 60 * 60
ARTIFACT_RETENTION_COUNT = 256
ARTIFACT_RETENTION_BYTES = 128 * 1024 * 1024
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] = {}
LAST_ARTIFACT_GC = 0.0
CAPACITY_PATTERN = re.compile(
r"(?:rate.?limit|capacity|overload|usage.?limit|quota|credit|exhaust|429|529|authentication|oauth|token.*expired)",
re.I,
@ -127,22 +134,65 @@ class TerminalFinalizationPending(RuntimeError):
"""An accepted worker result is durable but not committed to Kanban yet."""
@dataclass(frozen=True)
class TerminalIdentity:
"""Replay authority derived only from a journal's directory and filename."""
board: str
task_id: str
run_id: int
state: str
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)
"""Durably replace a small state document without following temp symlinks."""
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
os.chmod(path.parent, 0o700, follow_symlinks=False)
temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp")
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
flags |= getattr(os, "O_NOFOLLOW", 0)
descriptor = None
try:
descriptor = os.open(temporary, flags, mode)
with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
descriptor = None
json.dump(value, stream, indent=2, sort_keys=True)
stream.write("\n")
stream.flush()
os.fsync(stream.fileno())
os.chmod(temporary, mode, follow_symlinks=False)
os.replace(temporary, path)
_fsync_directory(path.parent)
finally:
if descriptor is not None:
os.close(descriptor)
try:
temporary.unlink()
except FileNotFoundError:
pass
def _fsync_directory(directory: Path) -> None:
"""Persist directory-entry changes after an atomic rename or quarantine."""
flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
flags |= getattr(os, "O_NOFOLLOW", 0)
descriptor = os.open(directory, flags)
try:
os.fsync(descriptor)
finally:
os.close(descriptor)
def load_json(path: Path) -> dict[str, Any]:
try:
value = json.loads(path.read_text(encoding="utf-8"))
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
descriptor = os.open(path, flags)
with os.fdopen(descriptor, "r", encoding="utf-8") as stream:
value = json.load(stream)
except (OSError, json.JSONDecodeError):
return {}
return value if isinstance(value, dict) else {}
@ -313,10 +363,80 @@ def _candidate_path(state_file: Path, run_id: Any, sequence: int) -> Path:
)
def _terminal_path(state_file: Path, run_id: Any) -> Path:
"""Return the replay journal path for an accepted terminal response."""
def _terminal_path(
state_file: Path,
run_id: Any,
state: str = "pending",
) -> Path:
"""Return the identity-bound replay path for an accepted terminal response."""
if state not in {"pending", "committed"}:
raise ValueError(f"invalid terminal journal state: {state}")
safe_run = re.sub(r"[^a-zA-Z0-9_.-]+", "-", str(run_id or "unknown"))
return state_file.with_name(f"{state_file.stem}.run-{safe_run}.terminal.json")
return state_file.with_name(
f"{state_file.stem}.run-{safe_run}.terminal.{state}.json"
)
_TERMINAL_NAME = re.compile(
r"^(?P<task>[a-zA-Z0-9_.-]+)\.run-(?P<run>[0-9]+)\."
r"terminal\.(?P<state>pending|committed)\.json$"
)
_SAFE_BOARD = re.compile(r"^[a-zA-Z0-9_.-]+$")
def _terminal_identity(path: Path) -> TerminalIdentity | None:
"""Parse replay authority lexically before opening a journal or board."""
try:
relative = path.relative_to(STATE_ROOT)
except ValueError:
return None
if len(relative.parts) != 2:
return None
board, filename = relative.parts
match = _TERMINAL_NAME.fullmatch(filename)
try:
board_stat = path.parent.stat(follow_symlinks=False)
except OSError:
return None
if (
not match
or not stat.S_ISDIR(board_stat.st_mode)
or not _SAFE_BOARD.fullmatch(board)
or board in {".", ".."}
or match.group("task") in {".", ".."}
):
return None
return TerminalIdentity(
board=board,
task_id=match.group("task"),
run_id=int(match.group("run")),
state=match.group("state"),
)
def _load_terminal_json(path: Path, identity: TerminalIdentity) -> dict[str, Any]:
"""Read through the already-validated board directory without symlink hops."""
if _terminal_identity(path) != identity:
return {}
directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
directory_flags |= getattr(os, "O_NOFOLLOW", 0)
file_flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
directory = None
descriptor = None
try:
directory = os.open(path.parent, directory_flags)
descriptor = os.open(path.name, file_flags, dir_fd=directory)
with os.fdopen(descriptor, "r", encoding="utf-8") as stream:
descriptor = None
value = json.load(stream)
except (OSError, json.JSONDecodeError):
return {}
finally:
if descriptor is not None:
os.close(descriptor)
if directory is not None:
os.close(directory)
return value if isinstance(value, dict) else {}
def _persist_candidate(
@ -381,7 +501,10 @@ def _write_terminal_record(
return path, record
def _terminal_record_valid(record: dict[str, Any]) -> bool:
def _terminal_record_valid(
record: dict[str, Any],
identity: TerminalIdentity | None = None,
) -> bool:
"""Reject malformed or non-terminal replay journals without side effects."""
if not isinstance(record, dict):
return False
@ -399,10 +522,18 @@ def _terminal_record_valid(record: dict[str, Any]) -> bool:
structured = json.loads(record["result"])
except (TypeError, json.JSONDecodeError):
return False
return (
valid = (
isinstance(structured, dict)
and structured.get("status") == "completed"
and not structured.get("blockers")
and record.get("kanban_state") in {"pending", "committed"}
)
if not valid or identity is None:
return valid
return (
record.get("board") == identity.board
and record.get("task_id") == identity.task_id
and record.get("expected_run_id") == identity.run_id
)
@ -945,21 +1076,142 @@ def _board_call(
raise last_storage_error
def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: str) -> Path:
"""Move an unusable journal aside with a deterministic, private name."""
board_dir = path.parent
try:
board_stat = board_dir.stat(follow_symlinks=False)
except OSError:
board_stat = None
if board_stat is None or not stat.S_ISDIR(board_stat.st_mode):
print(
f"refused terminal journal path through unsafe board directory: {path}",
file=sys.stderr,
flush=True,
)
return path
quarantine = board_dir / "quarantine"
quarantine.mkdir(parents=True, exist_ok=True, mode=0o700)
os.chmod(quarantine, 0o700, follow_symlinks=False)
try:
source_stat = path.stat(follow_symlinks=False)
regular = stat.S_ISREG(source_stat.st_mode)
except OSError:
regular = False
try:
if not regular:
raise OSError("journal is not a regular file")
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
descriptor = os.open(path, flags)
with os.fdopen(descriptor, "rb") as stream:
fingerprint = hashlib.sha256(stream.read()).hexdigest()[:16]
except OSError:
fingerprint = hashlib.sha256(path.name.encode("utf-8")).hexdigest()[:16]
safe_reason = re.sub(r"[^a-zA-Z0-9_.-]+", "-", reason).strip("-") or "invalid"
destination = quarantine / f"{path.name}.{safe_reason}.{fingerprint}.quarantine"
source_exists = path.exists() or path.is_symlink()
if source_exists and regular and not destination.exists():
os.chmod(path, 0o600, follow_symlinks=False)
os.replace(path, destination)
os.chmod(destination, 0o600, follow_symlinks=False)
_fsync_directory(board_dir)
_fsync_directory(quarantine)
elif source_exists:
path.unlink()
_fsync_directory(board_dir)
if not destination.exists():
atomic_json(
destination,
{
"original_name": path.name,
"reason": reason,
"recorded_at": utc_now(),
},
)
print(
"quarantined terminal journal "
f"{path.name}: {reason}; identity={identity or 'unparseable'}",
file=sys.stderr,
flush=True,
)
return destination
def _recover_quarantined_run(
kanban_db: Any,
identity: TerminalIdentity | None,
reason: str,
) -> bool:
"""Make an exact external run retryable after its journal is quarantined."""
if identity is None:
return False
def operation(conn: Any) -> bool:
task = kanban_db.get_task(conn, identity.task_id)
if task is None:
return False
if (
str(_task_value(task, "status", "")) != "running"
or _task_value(task, "current_run_id", None) != identity.run_id
or not _external(task)
):
return False
return bool(
kanban_db.reclaim_task(
conn,
identity.task_id,
reason=f"terminal journal quarantined ({reason}); exact run may retry",
)
)
try:
recovered = bool(_board_call(kanban_db, identity.board, operation))
except Exception as error:
_record_board_access_error(identity.board, error)
return False
if recovered:
print(
f"reclaimed {identity.board}/{identity.task_id} run {identity.run_id} "
f"after terminal journal quarantine: {reason}",
file=sys.stderr,
flush=True,
)
return recovered
def _commit_terminal_file(path: Path, identity: TerminalIdentity, document: dict[str, Any]) -> Path:
"""Persist committed state, then atomically retire a pending journal."""
committed = _terminal_path(
state_path(identity.board, identity.task_id),
identity.run_id,
"committed",
)
document["kanban_state"] = "committed"
document["committed_at"] = utc_now()
atomic_json(path, document)
os.replace(path, committed)
os.chmod(committed, 0o600, follow_symlinks=False)
_fsync_directory(committed.parent)
return committed
def _finalize_terminal_record(
kanban_db: Any,
path: Path,
record: dict[str, Any] | None = None,
_record: dict[str, Any] | None = None,
) -> str:
"""Commit one exact-run terminal journal, or leave it safely pending."""
document = dict(record or load_json(path))
identity = _terminal_identity(path)
if identity is None or identity.state != "pending":
return "invalid"
document = _load_terminal_json(path, identity)
if not _terminal_record_valid(document):
return "invalid"
board = str(document["board"])
task_id = str(document["task_id"])
expected_run_id = int(document["expected_run_id"])
if not _terminal_record_valid(document, identity):
return "foreign"
def operation(conn: Any) -> str:
task = kanban_db.get_task(conn, task_id)
task = kanban_db.get_task(conn, identity.task_id)
if task is None:
return "stale"
status = str(_task_value(task, "status", ""))
@ -971,24 +1223,22 @@ def _finalize_terminal_record(
)
if (
status != "running"
or _task_value(task, "current_run_id", None) != expected_run_id
or _task_value(task, "current_run_id", None) != identity.run_id
):
return "stale"
completed = kanban_db.complete_task(
conn,
task_id,
identity.task_id,
result=document["result"],
summary=document["summary"],
metadata=document["metadata"],
expected_run_id=expected_run_id,
expected_run_id=identity.run_id,
)
return "committed" if completed else "pending"
outcome = str(_board_call(kanban_db, board, operation))
outcome = str(_board_call(kanban_db, identity.board, operation))
if outcome == "committed":
document["kanban_state"] = "committed"
document["committed_at"] = utc_now()
atomic_json(path, document)
_commit_terminal_file(path, identity, document)
return outcome
@ -997,18 +1247,30 @@ def recover_pending_finalizations() -> int:
from hermes_cli import kanban_db
recovered = 0
for path in sorted(STATE_ROOT.glob("*/*.terminal.json")):
record = load_json(path)
if record.get("kanban_state") == "committed":
for path in sorted(STATE_ROOT.glob("*/*.terminal.pending.json")):
identity = _terminal_identity(path)
record = _load_terminal_json(path, identity) if identity is not None else {}
if identity is None:
_quarantine_terminal(path, None, "malformed-name")
continue
if not _terminal_record_valid(record):
_quarantine_terminal(path, identity, "malformed-payload")
_recover_quarantined_run(kanban_db, identity, "malformed-payload")
continue
if not _terminal_record_valid(record, identity):
_quarantine_terminal(path, identity, "foreign-identity")
_recover_quarantined_run(kanban_db, identity, "foreign-identity")
continue
try:
outcome = _finalize_terminal_record(kanban_db, path, record)
except Exception as error:
board = str(record.get("board") or "unknown")
_record_board_access_error(board, error)
_record_board_access_error(identity.board, error)
continue
if outcome == "committed":
recovered += 1
elif outcome in {"invalid", "foreign", "stale"}:
_quarantine_terminal(path, identity, outcome)
_recover_quarantined_run(kanban_db, identity, outcome)
return recovered
@ -1016,22 +1278,101 @@ def _has_pending_finalization(board: str, task_id: str, run_id: Any) -> bool:
"""Keep an exact run claimed while its accepted result awaits replay."""
if not isinstance(run_id, int):
return False
path = _terminal_path(state_path(board, task_id), run_id)
path = _terminal_path(state_path(board, task_id), run_id, "pending")
try:
path.stat()
except FileNotFoundError:
return False
except OSError:
return True
record = load_json(path)
if not _terminal_record_valid(record):
return True
identity = (
record.get("board") == board
and record.get("task_id") == task_id
and record.get("expected_run_id") == run_id
return False
identity = _terminal_identity(path)
if identity is None:
return False
record = _load_terminal_json(path, identity)
return _terminal_record_valid(record, identity)
def _artifact_gc_candidates(board_dir: Path) -> list[Path]:
"""List only bounded artifact classes; pending journals are excluded."""
candidates: list[Path] = []
patterns = (
"*.provider-*.result.json",
"*.candidate-*.json",
"*.terminal.committed.json",
)
return not identity or record.get("kanban_state") != "committed"
for pattern in patterns:
candidates.extend(board_dir.glob(pattern))
quarantine = board_dir / "quarantine"
if quarantine.is_dir():
candidates.extend(quarantine.glob("*.quarantine"))
return candidates
def gc_lane_artifacts(
*,
now: float | None = None,
max_age_seconds: int = ARTIFACT_RETENTION_AGE_SECONDS,
max_count: int = ARTIFACT_RETENTION_COUNT,
max_bytes: int = ARTIFACT_RETENTION_BYTES,
) -> int:
"""Bound non-pending lane artifacts by age, count, and bytes per board."""
current = time.time() if now is None else now
removed = 0
for board_dir in STATE_ROOT.iterdir() if STATE_ROOT.is_dir() else ():
if not board_dir.is_dir() or board_dir.is_symlink():
continue
board_removed = 0
entries: list[tuple[Path, os.stat_result]] = []
for path in _artifact_gc_candidates(board_dir):
try:
file_stat = path.stat(follow_symlinks=False)
except OSError:
continue
if not stat.S_ISREG(file_stat.st_mode):
continue
if max_age_seconds >= 0 and current - file_stat.st_mtime > max_age_seconds:
path.unlink()
removed += 1
board_removed += 1
else:
entries.append((path, file_stat))
retained_count = 0
retained_bytes = 0
for path, file_stat in sorted(
entries, key=lambda item: item[1].st_mtime, reverse=True
):
exceeds_count = max_count >= 0 and retained_count >= max_count
exceeds_bytes = (
max_bytes >= 0 and retained_bytes + file_stat.st_size > max_bytes
)
if exceeds_count or exceeds_bytes:
path.unlink()
removed += 1
board_removed += 1
continue
retained_count += 1
retained_bytes += file_stat.st_size
if board_removed:
_fsync_directory(board_dir)
return removed
def maybe_gc_lane_artifacts(*, now: float | None = None) -> int:
"""Run artifact retention on a bounded cadence, not every dispatcher tick."""
global LAST_ARTIFACT_GC
current = time.time() if now is None else now
if current - LAST_ARTIFACT_GC < ARTIFACT_GC_INTERVAL_SECONDS:
return 0
LAST_ARTIFACT_GC = current
try:
return gc_lane_artifacts(now=current)
except OSError as error:
print(
f"lane artifact retention deferred: {type(error).__name__}: {error}",
file=sys.stderr,
flush=True,
)
return 0
def execute_claim(board: str, task_id: str) -> None:
@ -1176,6 +1517,16 @@ def execute_claim(board: str, task_id: str) -> None:
heartbeat,
remaining,
)
candidate_file = None
if result.structured:
candidate_file = _persist_candidate(
state,
state_file,
dict(result.structured),
route=route,
returncode=result.returncode,
goal_turn=goal_turn,
)
if result.capacity_failure:
unavailable_provider = route.provider
retry_context = (
@ -1206,6 +1557,15 @@ def execute_claim(board: str, task_id: str) -> None:
max(1, int(deadline - time.monotonic())),
)
route = fallback
if result.structured:
candidate_file = _persist_candidate(
state,
state_file,
dict(result.structured),
route=route,
returncode=result.returncode,
goal_turn=goal_turn,
)
structured = result.structured
if structured:
structured = dict(structured)
@ -1213,16 +1573,6 @@ def execute_claim(board: str, task_id: str) -> None:
workspace,
structured.get("artifacts"),
)
candidate_file = _persist_candidate(
state,
state_file,
structured,
route=route,
returncode=result.returncode,
goal_turn=goal_turn,
)
else:
candidate_file = None
metadata = {
"executor": "direct-cli-lane",
"provider": route.provider,
@ -1281,8 +1631,9 @@ def execute_claim(board: str, task_id: str) -> None:
and result.returncode == 0
and completion_problem is None
):
terminal_file = _terminal_path(state_file, run_id)
metadata["terminal_record"] = str(terminal_file)
metadata["terminal_record"] = str(
_terminal_path(state_file, run_id, "committed")
)
terminal_file, terminal_record = _write_terminal_record(
state_file,
board=board,
@ -1292,11 +1643,17 @@ def execute_claim(board: str, task_id: str) -> None:
summary=str(structured.get("summary") or "Completed"),
metadata=metadata,
)
outcome = _finalize_terminal_record(
kanban_db,
terminal_file,
terminal_record,
)
try:
outcome = _finalize_terminal_record(
kanban_db,
terminal_file,
terminal_record,
)
except Exception as error:
raise TerminalFinalizationPending(
"accepted worker result is durably journaled; "
f"Kanban finalization raised {type(error).__name__}"
) from error
if outcome != "committed":
raise TerminalFinalizationPending(
"accepted worker result is durably journaled but "
@ -1537,6 +1894,7 @@ def main() -> int:
print(f"worker future failed: {error}", file=sys.stderr, flush=True)
del futures[future]
recover_pending_finalizations()
maybe_gc_lane_artifacts()
active = set(futures.values())
try:
newly_claimed = claim_ready(active, workers - len(futures))

438
testing/tests/test_hermes_cli_lanes.py
View File

@ -727,7 +727,7 @@ def test_accepted_result_survives_failed_finalization_and_replays_exact_run(
lanes.execute_claim("cassandra", "t_terminal")
candidates = list((state_root / "cassandra").glob("*.candidate-*.json"))
terminals = list((state_root / "cassandra").glob("*.terminal.json"))
terminals = list((state_root / "cassandra").glob("*.terminal.pending.json"))
assert len(candidates) == 1
assert json.loads(candidates[0].read_text())["structured"] == structured
assert len(terminals) == 1
@ -739,7 +739,12 @@ def test_accepted_result_survives_failed_finalization_and_replays_exact_run(
assert lanes.recover_pending_finalizations() == 1
assert task.status == "done"
assert len(completions) == 2
terminal = json.loads(terminals[0].read_text())
assert not terminals[0].exists()
committed = list(
(state_root / "cassandra").glob("*.terminal.committed.json")
)
assert len(committed) == 1
terminal = json.loads(committed[0].read_text())
assert terminal["kanban_state"] == "committed"
assert completions[-1]["result"] == terminal["result"]
@ -842,7 +847,334 @@ def test_terminal_replay_never_crosses_into_a_replacement_run(
assert lanes.recover_pending_finalizations() == 0
assert completions == []
assert json.loads(terminal_path.read_text())["kanban_state"] == "pending"
assert not terminal_path.exists()
quarantined = list((state_root / "cassandra/quarantine").glob("*.quarantine"))
assert len(quarantined) == 1
def _completed_result(summary: str = "done") -> dict:
return {
"status": "completed",
"summary": summary,
"changed_files": [],
"tests_run": [],
"artifacts": [],
"findings": [],
"blockers": [],
}
def test_complete_exception_after_journal_is_replayable_and_never_blocks(
tmp_path: Path,
monkeypatch,
):
state_root = tmp_path / "cli-lanes"
monkeypatch.setattr(lanes, "STATE_ROOT", state_root)
task = SimpleNamespace(
id="t_crash",
status="running",
result=None,
current_run_id=51,
assignee="cli-auto",
max_runtime_seconds=60,
)
completion_raises = {"value": True}
blocks = []
comments = []
class Connection:
def close(self):
return None
def complete_task(_conn, _task_id, **kwargs):
if completion_raises["value"]:
raise RuntimeError("crash between journal and commit")
task.status = "done"
task.result = kwargs["result"]
task.current_run_id = None
return True
fake_db = SimpleNamespace(
scoped_current_board=lambda _board: nullcontext(),
connect=lambda board: Connection(),
get_task=lambda _conn, _task_id: task,
worker_log_path=lambda _task_id, board: tmp_path / "worker.log",
_resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_crash"),
set_branch_name=lambda *_args: None,
set_workspace_path=lambda *_args: None,
build_worker_context=lambda *_args: "Finish safely.",
heartbeat_worker=lambda *_args, **_kwargs: True,
add_comment=lambda _conn, _task_id, _author, body: comments.append(body),
complete_task=complete_task,
block_task=lambda *_args, **kwargs: blocks.append(kwargs),
)
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
monkeypatch.setattr(
lanes,
"select_route",
lambda *_args, **_kwargs: lanes.Route(
"codex", "gpt-5.6-sol", "high", "codex-high", "test", "test", 1, ()
),
)
monkeypatch.setattr(
lanes,
"run_provider",
lambda *_args, **_kwargs: lanes.ProcessResult(
0, "", _completed_result("crash-safe result"), False
),
)
lanes.execute_claim("cassandra", "t_crash")
pending = list((state_root / "cassandra").glob("*.terminal.pending.json"))
assert len(pending) == 1
assert blocks == []
assert any("RuntimeError" in body for body in comments)
completion_raises["value"] = False
assert lanes.recover_pending_finalizations() == 1
assert task.status == "done"
assert not pending[0].exists()
def test_terminal_payload_cannot_select_a_different_board(
tmp_path: Path,
monkeypatch,
):
state_root = tmp_path / "cli-lanes"
monkeypatch.setattr(lanes, "STATE_ROOT", state_root)
path = lanes._terminal_path(lanes.state_path("alpha", "t_alpha"), 3)
lanes.atomic_json(
path,
{
"board": "beta",
"task_id": "t_beta",
"expected_run_id": 9,
"result": json.dumps(_completed_result()),
"summary": "forged",
"metadata": {},
"kanban_state": "pending",
},
)
alpha_task = SimpleNamespace(
id="t_alpha", status="running", current_run_id=3, assignee="cli-auto"
)
opened = []
reclaimed = []
class Connection:
def close(self):
return None
fake_db = SimpleNamespace(
scoped_current_board=lambda _board: nullcontext(),
connect=lambda board: (opened.append(board) or Connection()),
get_task=lambda _conn, task_id: alpha_task if task_id == "t_alpha" else None,
reclaim_task=lambda _conn, task_id, **_kwargs: (reclaimed.append(task_id) or True),
)
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
assert lanes.recover_pending_finalizations() == 0
assert opened == ["alpha"]
assert reclaimed == ["t_alpha"]
assert not path.exists()
quarantined = list((state_root / "alpha/quarantine").glob("*.quarantine"))
assert len(quarantined) == 1
assert quarantined[0].stat().st_mode & 0o777 == 0o600
def test_terminal_finalize_rereads_journal_after_a_path_swap(
tmp_path: Path,
monkeypatch,
):
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
state_file = lanes.state_path("alpha", "t_alpha")
path, safe_record = lanes._write_terminal_record(
state_file,
board="alpha",
task_id="t_alpha",
run_id=4,
structured=_completed_result("safe"),
summary="safe",
metadata={},
)
forged = dict(safe_record)
forged.update({"board": "beta", "task_id": "t_beta", "expected_run_id": 7})
lanes.atomic_json(path, forged)
opened = []
fake_db = SimpleNamespace(
scoped_current_board=lambda _board: nullcontext(),
connect=lambda board: (opened.append(board) or pytest.fail("must not open DB")),
)
assert lanes._finalize_terminal_record(fake_db, path, safe_record) == "foreign"
assert opened == []
@pytest.mark.parametrize("payload", [b"", b'{"board":"cassandra"'])
def test_malformed_exact_run_journal_is_quarantined_and_reclaimed(
tmp_path: Path,
monkeypatch,
payload: bytes,
):
state_root = tmp_path / "cli-lanes"
monkeypatch.setattr(lanes, "STATE_ROOT", state_root)
path = lanes._terminal_path(lanes.state_path("cassandra", "t_partial"), 12)
path.parent.mkdir(parents=True)
path.write_bytes(payload)
task = SimpleNamespace(
id="t_partial", status="running", current_run_id=12, assignee="cli-auto"
)
reclaimed = []
class Connection:
def close(self):
return None
fake_db = SimpleNamespace(
scoped_current_board=lambda _board: nullcontext(),
connect=lambda board: Connection(),
get_task=lambda _conn, _task_id: task,
reclaim_task=lambda *_args, **_kwargs: (reclaimed.append(True) or True),
)
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
assert lanes.recover_pending_finalizations() == 0
assert reclaimed == [True]
assert not path.exists()
assert lanes._has_pending_finalization("cassandra", "t_partial", 12) is False
quarantined = list((state_root / "cassandra/quarantine").glob("*.quarantine"))
assert len(quarantined) == 1
assert quarantined[0].stat().st_mode & 0o777 == 0o600
def test_recovery_does_not_parse_committed_journals_every_tick(
tmp_path: Path,
monkeypatch,
):
state_root = tmp_path / "cli-lanes"
monkeypatch.setattr(lanes, "STATE_ROOT", state_root)
committed = lanes._terminal_path(
lanes.state_path("cassandra", "t_done"), 1, "committed"
)
lanes.atomic_json(committed, {"kanban_state": "committed"})
reads = []
monkeypatch.setattr(lanes, "load_json", lambda path: (reads.append(path) or {}))
monkeypatch.setitem(
sys.modules,
"hermes_cli",
SimpleNamespace(kanban_db=SimpleNamespace()),
)
assert lanes.recover_pending_finalizations() == 0
assert reads == []
def test_atomic_json_fsyncs_file_and_directory_and_uses_private_mode(
tmp_path: Path,
monkeypatch,
):
calls = []
real_fsync = lanes.os.fsync
monkeypatch.setattr(
lanes.os,
"fsync",
lambda descriptor: (calls.append(descriptor), real_fsync(descriptor))[1],
)
path = tmp_path / "state.json"
lanes.atomic_json(path, {"safe": True})
assert json.loads(path.read_text()) == {"safe": True}
assert path.stat().st_mode & 0o777 == 0o600
assert len(calls) >= 2
def test_atomic_json_refuses_a_precreated_temp_symlink(
tmp_path: Path,
monkeypatch,
):
path = tmp_path / "state.json"
victim = tmp_path / "credential"
victim.write_text("do not overwrite", encoding="utf-8")
temporary = tmp_path / ".state.json.fixed.tmp"
temporary.symlink_to(victim)
monkeypatch.setattr(lanes.uuid, "uuid4", lambda: SimpleNamespace(hex="fixed"))
with pytest.raises(FileExistsError):
lanes.atomic_json(path, {"unsafe": True})
assert victim.read_text(encoding="utf-8") == "do not overwrite"
assert not path.exists()
def test_artifact_gc_prunes_by_age_without_touching_pending_journals(
tmp_path: Path,
monkeypatch,
):
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
board = lanes.STATE_ROOT / "cassandra"
board.mkdir(parents=True)
quarantine = board / "quarantine"
quarantine.mkdir()
old_artifacts = [
board / "t.run-1.candidate-1.json",
board / "t.run-1.provider-1.result.json",
board / "t.run-1.terminal.committed.json",
quarantine / "t.invalid.1234.quarantine",
]
for artifact in old_artifacts:
artifact.write_text("{}", encoding="utf-8")
pending = board / "t.run-1.terminal.pending.json"
pending.write_text("{}", encoding="utf-8")
old_time = 100.0
for artifact in old_artifacts:
os.utime(artifact, (old_time, old_time))
os.utime(pending, (old_time, old_time))
assert lanes.gc_lane_artifacts(
now=1000.0, max_age_seconds=10, max_count=100, max_bytes=10000
) == len(old_artifacts)
assert not any(artifact.exists() for artifact in old_artifacts)
assert pending.exists()
def test_artifact_gc_prunes_oldest_by_count_and_total_bytes(
tmp_path: Path,
monkeypatch,
):
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
board = lanes.STATE_ROOT / "cassandra"
board.mkdir(parents=True)
files = []
for sequence in range(3):
path = board / f"t.run-1.provider-{sequence}.result.json"
path.write_text("x" * 40, encoding="utf-8")
os.utime(path, (100 + sequence, 100 + sequence))
files.append(path)
assert lanes.gc_lane_artifacts(
now=200.0, max_age_seconds=1000, max_count=2, max_bytes=45
) == 2
assert [path.exists() for path in files] == [False, False, True]
def test_artifact_gc_is_interval_gated(
tmp_path: Path,
monkeypatch,
):
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
monkeypatch.setattr(lanes, "LAST_ARTIFACT_GC", 0.0)
calls = []
monkeypatch.setattr(
lanes, "gc_lane_artifacts", lambda **kwargs: (calls.append(kwargs) or 0)
)
assert lanes.maybe_gc_lane_artifacts(now=1000.0) == 0
assert lanes.maybe_gc_lane_artifacts(now=1001.0) == 0
assert len(calls) == 1
@pytest.mark.parametrize(
@ -936,7 +1268,7 @@ def test_claim_requires_structured_evidence_and_surfaces_artifacts(
block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)),
)
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
monkeypatch.setattr(lanes, "state_path", lambda _board, _task_id: tmp_path / "state.json")
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
monkeypatch.setattr(
lanes,
"select_route",
@ -1004,11 +1336,7 @@ def test_goal_card_continues_after_local_judge_rejects_progress(
block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)),
)
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
monkeypatch.setattr(
lanes,
"state_path",
lambda _board, _task_id: tmp_path / "state.json",
)
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
claude_low = lanes.Route(
"claude", "claude-fable-5", "low", "claude-low", "jetson", "vote", 1, ()
)
@ -1087,7 +1415,9 @@ def test_goal_card_continues_after_local_judge_rejects_progress(
assert route_calls[2][1]["exclude_provider"] == "claude"
assert "prior rejected reports" in judge_contexts[1]
assert "commit, push, and remote verification are missing" in judge_contexts[1]
candidates = sorted(tmp_path.glob("state.run-12.candidate-*.json"))
candidates = sorted(
(lanes.STATE_ROOT / "cassandra").glob("t_goal.run-12.candidate-*.json")
)
assert len(candidates) == 2
assert json.loads(candidates[0].read_text())["structured"]["summary"] == (
"Focused tests passed."
@ -1098,6 +1428,87 @@ def test_goal_card_continues_after_local_judge_rejects_progress(
assert reports == []
def test_capacity_fallback_preserves_first_claude_structured_response(
tmp_path: Path,
monkeypatch,
):
task = SimpleNamespace(
id="t_fallback",
status="running",
result=None,
current_run_id=14,
assignee="cli-auto",
max_runtime_seconds=60,
)
class Connection:
def close(self):
return None
fake_db = SimpleNamespace(
scoped_current_board=lambda _board: nullcontext(),
connect=lambda board: Connection(),
get_task=lambda _conn, _task_id: task,
worker_log_path=lambda _task_id, board: tmp_path / "worker.log",
_resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_fallback"),
set_branch_name=lambda *_args: None,
set_workspace_path=lambda *_args: None,
build_worker_context=lambda *_args: "Complete with a fallback if needed.",
heartbeat_worker=lambda *_args, **_kwargs: True,
add_comment=lambda *_args: None,
complete_task=lambda _conn, _task_id, **kwargs: (
setattr(task, "status", "done") or setattr(task, "result", kwargs["result"]) or True
),
block_task=lambda *_args, **_kwargs: pytest.fail("fallback should complete"),
)
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
claude = lanes.Route(
"claude", "claude-fable-5", "high", "claude-high", "test", "test", 1, ()
)
codex = lanes.Route(
"codex", "gpt-5.6-sol", "high", "codex-high", "test", "test", 1, ()
)
monkeypatch.setattr(
lanes,
"select_route",
lambda _prompt, assignee, **_kwargs: codex
if assignee == "cli-codex-high"
else claude,
)
reports = [
lanes.ProcessResult(
1,
"subscription capacity exhausted",
{
**_completed_result("Claude preserved evidence"),
"status": "incomplete",
"blockers": ["subscription capacity exhausted"],
},
True,
),
lanes.ProcessResult(0, "", _completed_result("Codex completed"), False),
]
monkeypatch.setattr(lanes, "run_provider", lambda *_args, **_kwargs: reports.pop(0))
lanes.execute_claim("cassandra", "t_fallback")
candidates = sorted(
(lanes.STATE_ROOT / "cassandra").glob("t_fallback.run-14.candidate-*.json")
)
assert len(candidates) == 2
first, second = [json.loads(path.read_text()) for path in candidates]
assert (first["provider"], first["structured"]["summary"]) == (
"claude",
"Claude preserved evidence",
)
assert (second["provider"], second["structured"]["summary"]) == (
"codex",
"Codex completed",
)
assert task.status == "done"
def test_workspace_preparation_failure_durably_blocks_the_claim(tmp_path: Path, monkeypatch):
task = SimpleNamespace(id="t_bad_worktree", current_run_id=7, assignee="cli-auto")
calls = []
@ -1143,11 +1554,15 @@ def test_artifacts_cannot_escape_the_task_worktree(tmp_path: Path):
def test_restart_provider_change_includes_explicit_workspace_handoff(tmp_path: Path, monkeypatch):
task = SimpleNamespace(
id="t_resume",
status="running",
result=None,
current_run_id=9,
assignee="cli-auto",
max_runtime_seconds=60,
)
state_file = tmp_path / "state.json"
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
state_file = lanes.state_path("cassandra", "t_resume")
state_file.parent.mkdir(parents=True)
state_file.write_text(
json.dumps({"current_route": {"provider": "claude"}}),
encoding="utf-8",
@ -1173,7 +1588,6 @@ def test_restart_provider_change_includes_explicit_workspace_handoff(tmp_path: P
block_task=lambda *_args, **_kwargs: None,
)
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
monkeypatch.setattr(lanes, "state_path", lambda _board, _task_id: state_file)
monkeypatch.setattr(
lanes,
"select_route",