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

hermes: preserve exact CLI lane retirement semantics

Browse Source
This commit is contained in:
Hermes Agent 2026-08-17 13:21:00 +00:00
parent 4d4cf1bd44
commit 56697358ac
17 changed files with 452 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
services/hermes/scripts/cli_lane_evidence.py
View File

@ -56,8 +56,12 @@ def _retire_terminal_entry(
board_descriptor: int,
quarantine_descriptor: int,
authority_name: str | None = None,
source_descriptor: int | None = None,
) -> str:
"""Retire only the inode previously inspected; preserve any replacement."""
pinned = os.fstat(source_descriptor) if source_descriptor is not None else source_stat
if pinned.st_dev != source_stat.st_dev or pinned.st_ino != source_stat.st_ino:
return "replacement"
try:
current = os.stat(
path.name,
@ -66,7 +70,7 @@ def _retire_terminal_entry(
)
except FileNotFoundError:
return "missing"
if current.st_dev != source_stat.st_dev or current.st_ino != source_stat.st_ino:
if current.st_dev != pinned.st_dev or current.st_ino != pinned.st_ino:
return "replacement"
# A replacement can itself be staged repeatedly while recovery races a
# writer. Keep every generation bound to the original canonical pending
@ -76,7 +80,7 @@ def _retire_terminal_entry(
(authority_name or path.name).encode("utf-8")
).hexdigest()[:16]
source_digest = hashlib.sha256(
f"{source_stat.st_dev:x}\0{source_stat.st_ino:x}".encode()
f"{pinned.st_dev:x}\0{pinned.st_ino:x}".encode()
).hexdigest()[:16]
for sequence in range(32):
staging = f".retire.{path_digest}.{source_digest}.{sequence}"
@ -97,7 +101,7 @@ def _retire_terminal_entry(
dir_fd=quarantine_descriptor,
follow_symlinks=False,
)
if staged.st_dev == source_stat.st_dev and staged.st_ino == source_stat.st_ino:
if staged.st_dev == pinned.st_dev and staged.st_ino == pinned.st_ino:
os.unlink(staging, dir_fd=quarantine_descriptor)
os.fsync(quarantine_descriptor)
os.fsync(board_descriptor)
@ -132,6 +136,7 @@ def _write_json_noreplace(path: Path, value: dict[str, Any]) -> bool:
temporary = f".{path.name}.{uuid.uuid4().hex}.tmp"
temporary_stat = None
descriptor = None
cleanup_descriptor = None
try:
descriptor = os.open(temporary, create_flags, 0o600, dir_fd=directory)
temporary_stat = os.fstat(descriptor)
@ -143,6 +148,7 @@ def _write_json_noreplace(path: Path, value: dict[str, Any]) -> bool:
written = os.fstat(stream.fileno())
if stat.S_IMODE(written.st_mode) != 0o600 or written.st_nlink != 1:
raise OSError("terminal evidence is not private and singly linked")
cleanup_descriptor = os.dup(stream.fileno())
try:
_rename_noreplace(
temporary,
@ -164,7 +170,10 @@ def _write_json_noreplace(path: Path, value: dict[str, Any]) -> bool:
temporary_stat,
board_descriptor=directory,
quarantine_descriptor=directory,
source_descriptor=cleanup_descriptor,
)
if cleanup_descriptor is not None:
os.close(cleanup_descriptor)
os.close(directory)
def _terminal_document_digest(document: dict[str, Any]) -> str:

26
services/hermes/scripts/cli_lane_execution.py
View File

@ -30,6 +30,12 @@ from cli_lane_recovery import _has_pending_finalization
from cli_lane_routing import fresh_unavailable_provider, select_route
def _goal_rejections(state: dict[str, object]) -> list[object]:
"""Return only bounded list-shaped prior judge evidence."""
history = state.get("goal_rejections", [])
return history if isinstance(history, list) else []
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
@ -260,16 +266,12 @@ def execute_claim(board: str, task_id: str) -> 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:])}."
)
rejection_history = _goal_rejections(state)
judge_context = 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,
@ -338,9 +340,7 @@ def execute_claim(board: str, task_id: str) -> None:
and deadline - time.monotonic() > 30
)
if can_continue:
rejection_history = state.get("goal_rejections", [])
if not isinstance(rejection_history, list):
rejection_history = []
rejection_history = _goal_rejections(state)
state["goal_rejections"] = [
*rejection_history[-4:],
completion_problem,

1
services/hermes/scripts/cli_lane_finalization.py
View File

@ -89,6 +89,7 @@ def _retire_snapshot(
board_descriptor=snapshot.directory_descriptor,
quarantine_descriptor=snapshot.directory_descriptor,
authority_name=authority_name,
source_descriptor=snapshot.descriptor,
)
def _retire_snapshot_after_db(path: Path, snapshot: TerminalSnapshot) -> str:

2
services/hermes/scripts/cli_lane_quarantine.py
View File

@ -180,6 +180,7 @@ def _quarantine_terminal(
source_stat,
board_descriptor=board_descriptor,
quarantine_descriptor=quarantine_descriptor,
source_descriptor=source_descriptor,
)
except OSError as error:
# A bad quarantine target must not make an attacker-controlled pending
@ -196,6 +197,7 @@ def _quarantine_terminal(
if quarantine_descriptor is not None
else board_descriptor
),
source_descriptor=source_descriptor,
)
except OSError:
retirement = "deferred"

9
services/hermes/scripts/cli_lane_retention.py
View File

@ -50,18 +50,27 @@ def _unlink_artifact_if_same(path: Path, observed: os.stat_result) -> bool:
directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
directory_flags |= getattr(os, "O_NOFOLLOW", 0)
descriptor = None
source_descriptor = None
try:
descriptor = os.open(path.parent, directory_flags)
file_flags = getattr(os, "O_PATH", os.O_RDONLY) | getattr(os, "O_NOFOLLOW", 0)
source_descriptor = os.open(path.name, file_flags, dir_fd=descriptor)
pinned = os.fstat(source_descriptor)
if pinned.st_dev != observed.st_dev or pinned.st_ino != observed.st_ino:
return False
outcome = _retire_terminal_entry(
path,
observed,
board_descriptor=descriptor,
quarantine_descriptor=descriptor,
source_descriptor=source_descriptor,
)
return outcome == "retired"
except OSError:
return False
finally:
if source_descriptor is not None:
os.close(source_descriptor)
if descriptor is not None:
os.close(descriptor)

8
testing/tests/test_hermes_cli_capabilities.py
View File

@ -93,6 +93,9 @@ def test_capability_detection_requires_explicit_safety_keywords():
assert lanes.detect_kanban_capabilities(variadic).ready is False
assert lanes._explicit_keyword(object(), "expected_run_id") is False
missing_completion = lanes.KanbanCapabilities(False, True, True)
assert missing_completion.deferred_features == ("exact-run-completion",)
def test_startup_health_moves_from_bounded_deferred_to_ready(
tmp_path: Path,
@ -117,9 +120,10 @@ def test_startup_health_moves_from_bounded_deferred_to_ready(
assert "compatibility deferred" in capsys.readouterr().err
new_db = _db(patched=True)
new = lanes.kanban_capabilities(new_db)
lanes.initialize_kanban_capabilities(new_db, health_path=health)
lanes.kanban_capabilities(new_db)
new = lanes.initialize_kanban_capabilities(new_db, health_path=health)
assert new.ready is True
assert lanes.kanban_capabilities(new_db) is new
assert lanes.runtime_health(health)["state"] == "ready"

37
testing/tests/test_hermes_cli_dispatch_runtime.py
View File

@ -13,6 +13,8 @@ from testing.tests.test_hermes_cli_support import lanes
def test_board_slug_and_connection_failure_paths(monkeypatch):
assert lanes._owns_local_workspace(SimpleNamespace(workspace_path=" /workspace "))
assert not lanes._owns_local_workspace(SimpleNamespace(workspace_path=""))
assert lanes._board_slug(SimpleNamespace(slug="cassandra")) == "cassandra"
assert lanes._board_slug(SimpleNamespace(id="fallback")) == "fallback"
failures = []
@ -100,6 +102,41 @@ def test_claim_ready_handles_zero_limit_empty_boards_and_claim_errors(monkeypatc
assert lanes.claim_ready(set(), 1) == []
def test_dispatch_skips_unhealthy_ineligible_and_unclaimed_entries(monkeypatch):
class Connection:
def __init__(self, board):
self.board = board
def close(self):
return None
tasks = [
SimpleNamespace(id="t_ineligible", status="ready", assignee="cli-auto"),
SimpleNamespace(id="t_unclaimed", status="ready", assignee="cli-auto"),
]
db = SimpleNamespace(
list_boards=lambda **_kwargs: [
{"slug": "offline"},
{"slug": "cassandra"},
],
scoped_current_board=lambda _board: nullcontext(),
connect=lambda board: (_ for _ in ()).throw(OSError("offline"))
if board == "offline"
else Connection(board),
recompute_ready=lambda _conn: None,
list_tasks=lambda _conn: tasks,
claim_task=lambda *_args, **_kwargs: None,
)
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db))
assert lanes.claim_ready(
set(),
2,
eligible=lambda _board, task: task.id == "t_unclaimed",
) == []
monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0)
lanes.recover_orphans()
class _Future:
def __init__(self):
self.polls = 0

35
testing/tests/test_hermes_cli_evidence_edges.py
View File

@ -44,6 +44,7 @@ def test_retirement_reports_missing_replacement_collision_and_restoration(
path = tmp_path / "pending"
path.write_text("old", encoding="utf-8")
source = path.stat()
source_descriptor = os.open(path, os.O_RDONLY)
directory = os.open(tmp_path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
try:
path.unlink()
@ -52,6 +53,7 @@ def test_retirement_reports_missing_replacement_collision_and_restoration(
source,
board_descriptor=directory,
quarantine_descriptor=directory,
source_descriptor=source_descriptor,
) == "missing"
path.write_text("new", encoding="utf-8")
@ -60,8 +62,11 @@ def test_retirement_reports_missing_replacement_collision_and_restoration(
source,
board_descriptor=directory,
quarantine_descriptor=directory,
source_descriptor=source_descriptor,
) == "replacement"
os.close(source_descriptor)
source_descriptor = os.open(path, os.O_RDONLY)
replacement = path.stat()
monkeypatch.setattr(
lanes,
@ -73,8 +78,10 @@ def test_retirement_reports_missing_replacement_collision_and_restoration(
replacement,
board_descriptor=directory,
quarantine_descriptor=directory,
source_descriptor=source_descriptor,
) == "collision"
finally:
os.close(source_descriptor)
os.close(directory)
@ -82,6 +89,7 @@ def test_retirement_preserves_a_postcheck_replacement(tmp_path: Path, monkeypatc
path = tmp_path / "pending"
path.write_text("old", encoding="utf-8")
source = path.stat()
source_descriptor = os.open(path, os.O_RDONLY)
directory = os.open(tmp_path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
calls = []
@ -102,8 +110,10 @@ def test_retirement_preserves_a_postcheck_replacement(tmp_path: Path, monkeypatc
source,
board_descriptor=directory,
quarantine_descriptor=directory,
source_descriptor=source_descriptor,
) == "replacement"
finally:
os.close(source_descriptor)
os.close(directory)
assert path.read_text(encoding="utf-8") == "replacement"
@ -133,6 +143,31 @@ def test_evidence_writer_bounds_payload_and_cleans_failed_temporary(
monkeypatch.setattr(lanes, "_retire_terminal_entry", real_retire)
def test_evidence_writer_closes_unwrapped_and_rejects_unsafe_temporary_files(
tmp_path: Path,
monkeypatch,
):
destination = tmp_path / "evidence.json"
real_fdopen = lanes.os.fdopen
opened = []
def fail_fdopen(descriptor, *_args, **_kwargs):
opened.append(descriptor)
raise OSError("fdopen failed")
monkeypatch.setattr(lanes.os, "fdopen", fail_fdopen)
with pytest.raises(OSError, match="fdopen failed"):
lanes._write_json_noreplace(destination, {"value": "bounded"})
with pytest.raises(OSError):
os.fstat(opened[0])
monkeypatch.setattr(lanes.os, "fdopen", real_fdopen)
with monkeypatch.context() as unsafe:
unsafe.setattr(lanes.stat, "S_IMODE", lambda _mode: 0o644)
with pytest.raises(OSError, match="not private"):
lanes._write_json_noreplace(destination, {"value": "bounded"})
def test_evidence_paths_and_collision_validation_fail_closed(
tmp_path: Path,
monkeypatch,

15
testing/tests/test_hermes_cli_execution_edges.py
View File

@ -16,6 +16,13 @@ class _Connection:
return None
def test_goal_rejection_history_is_list_bounded():
execution = __import__("cli_lane_execution")
history = ["first"]
assert execution._goal_rejections({"goal_rejections": history}) is history
assert execution._goal_rejections({"goal_rejections": "invalid"}) == []
def test_missing_claim_is_a_noop(tmp_path: Path, monkeypatch):
db = SimpleNamespace(
scoped_current_board=lambda _board: nullcontext(),
@ -64,12 +71,18 @@ def test_transient_callback_storage_errors_do_not_kill_provider(
def provider(*args, **_kwargs):
assert args[6]("still working") is True
return lanes.ProcessResult(1, "failed", None, False)
return lanes.ProcessResult(
1,
"failed",
{"status": "incomplete", "blockers": ["synthetic provider failure"]},
False,
)
monkeypatch.setattr(lanes, "run_provider", provider)
lanes.execute_claim("cassandra", "t_callbacks")
assert blocks and blocks[0]["kind"] == "capability"
assert blocks[0]["reason"] == "synthetic provider failure"
def test_restart_handoff_survives_missing_prior_log(tmp_path: Path, monkeypatch):

4
testing/tests/test_hermes_cli_fallback.py
View File

@ -53,6 +53,10 @@ def test_goal_card_continues_after_local_judge_rejects_progress(
)
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
lanes.atomic_json(
lanes.state_path("cassandra", "t_goal"),
{"goal_rejections": ["prior incomplete report"]},
)
claude_low = lanes.Route(
"claude", "claude-fable-5", "low", "claude-low", "jetson", "vote", 1, ()
)

8
testing/tests/test_hermes_cli_finalization_edges.py
View File

@ -159,6 +159,14 @@ def test_terminal_finalizer_closes_supplied_invalid_snapshots(
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
path = lanes._terminal_path(lanes.state_path("board", "task"), 8)
lanes.atomic_json(path, {})
empty = _Snapshot(None)
assert lanes._finalize_terminal_record(_db(None), path, snapshot=empty) == "invalid"
assert empty.closed == 1
finalization = __import__("cli_lane_finalization")
direct = _Snapshot(None)
assert finalization._finalize_terminal_record(
_db(None), path, snapshot=direct
) == "invalid"
assert direct.closed == 1

60
testing/tests/test_hermes_cli_foundation_coverage.py
View File

@ -3,6 +3,8 @@
from __future__ import annotations
import json
import os
import runpy
import sqlite3
from contextlib import nullcontext
from pathlib import Path
@ -11,7 +13,7 @@ from urllib.error import URLError
import pytest
from testing.tests.test_hermes_cli_support import _completed_result, lanes
from testing.tests.test_hermes_cli_support import SCRIPTS, _completed_result, lanes
def test_board_context_serializes_objects_and_storage_failure_is_bounded(
@ -93,6 +95,26 @@ def test_json_and_terminal_path_helpers_fail_closed(tmp_path: Path, monkeypatch)
)
assert lanes._terminal_evidence_identity(missing_evidence) is None
with pytest.raises(ValueError, match="positive SQLite"):
lanes.TerminalIdentity("board", "task", 0, "pending")
def test_recovery_snapshot_close_is_idempotent_after_descriptors_are_released(
tmp_path: Path,
):
directory = os.open(tmp_path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
snapshot = lanes.TerminalRecoverySnapshot(
None,
tmp_path.stat(),
None,
directory,
b"",
"synthetic",
)
snapshot.close()
snapshot.close()
assert snapshot.directory_descriptor == -1
def test_goal_helpers_cover_compaction_and_deterministic_failures():
compacted = lanes.cli_lane_goal._bounded("a" * 100, 40)
@ -122,6 +144,9 @@ def test_goal_helpers_cover_compaction_and_deterministic_failures():
)
assert accepted is False
assert reason
assert lanes.cli_lane_goal.unfinished_result_reason(
{"status": "completed", "summary": "done", "tests_run": "not-a-list"}
) is None
def test_prompt_artifact_and_json_helpers_reject_unsafe_inputs(
@ -147,6 +172,31 @@ def test_prompt_artifact_and_json_helpers_reject_unsafe_inputs(
assert lanes._extract_json('prefix {"status":"blocked"} suffix') == {
"status": "blocked"
}
assert lanes._extract_json('{"status":"unknown"}') is None
state_file = workspace / "state.json"
state = {}
assert lanes._event_payload(
"codex",
json.dumps({"type": "thread.started", "result": {"status": "blocked"}}),
state,
state_file,
) == {"status": "blocked"}
assert lanes._event_payload(
"claude",
json.dumps({"session_id": "session-2", "text": "not-json"}),
state,
state_file,
) is None
assert state["claude_session_id"] == "session-2"
prompt_module = __import__("cli_lane_prompt")
assert prompt_module.workspace_artifacts(
workspace, [str(directory), str(artifact)]
) == [str(artifact.resolve())]
assert prompt_module._event_payload(
"other", '{"item": {}}', {}, state_file
) is None
def test_provider_event_parsing_persists_sessions_and_nested_results(
@ -205,3 +255,11 @@ def test_switchyard_network_failure_is_explicit():
URLError("offline")
),
)
def test_runner_main_guard_delegates_to_dispatch(monkeypatch):
dispatch = __import__("cli_lane_dispatch")
monkeypatch.setattr(dispatch, "main", lambda: 17)
with pytest.raises(SystemExit) as raised:
runpy.run_path(str(SCRIPTS / "cli_lane_runner.py"), run_name="__main__")
assert raised.value.code == 17

67
testing/tests/test_hermes_cli_provider_edges.py
View File

@ -44,6 +44,26 @@ def test_stream_process_enforces_runtime_and_lease_boundaries(
assert lease_lost.returncode != 0
def test_stream_process_bounds_retained_line_history(tmp_path: Path):
result = lanes.stream_process(
[
sys.executable,
"-c",
"import sys; [sys.stdout.write(f'{n}\\n') for n in range(4010)]",
],
provider="codex",
cwd=tmp_path,
env=dict(os.environ),
log_path=tmp_path / "bounded.log",
state={},
state_file=tmp_path / "state.json",
heartbeat=lambda _note: True,
max_runtime=60,
)
assert result.returncode == 0
assert "4009" in result.output
def test_process_identity_and_signal_failures_are_bounded(monkeypatch):
assert lanes._process_record(999999999) is None
monkeypatch.setattr(
@ -63,9 +83,26 @@ def test_process_identity_and_signal_failures_are_bounded(monkeypatch):
)
lanes._signal_worker_tree(10, {11: (12, 13)}, signal.SIGTERM)
signals = []
monkeypatch.setattr(lanes, "_process_identity_matches", lambda *_args: False)
monkeypatch.setattr(lanes.os, "killpg", lambda pid, sig: signals.append((pid, sig)))
monkeypatch.setattr(
lanes.os,
"kill",
lambda *_args: (_ for _ in ()).throw(AssertionError("stale PID signaled")),
)
lanes._signal_worker_tree(20, {21: (22, 23)}, signal.SIGKILL)
assert signals == [(20, signal.SIGKILL)]
def test_descendant_snapshot_follows_multiple_generations(monkeypatch):
entries = [Path("/proc/self"), Path("/proc/10"), Path("/proc/11"), Path("/proc/12")]
entries = [
Path("/proc/self"),
Path("/proc/10"),
Path("/proc/11"),
Path("/proc/12"),
Path("/proc/13"),
]
monkeypatch.setattr(lanes.Path, "iterdir", lambda _path: iter(entries))
records = {
10: (1, 10, 100),
@ -76,6 +113,34 @@ def test_descendant_snapshot_follows_multiple_generations(monkeypatch):
assert lanes._descendant_processes(10) == {11: (11, 101), 12: (12, 102)}
def test_claude_missing_unstarted_session_does_not_mark_started(
tmp_path: Path,
monkeypatch,
):
state = {"claude_session_id": "synthetic-session"}
state_file = tmp_path / "state.json"
monkeypatch.setattr(
lanes,
"stream_process",
lambda *_args, **_kwargs: lanes.ProcessResult(
1, lanes.NO_CLAUDE_SESSION, None, False
),
)
route = lanes.Route("claude", "model", "high", "profile", "test", "test", 1, ())
result = lanes.run_provider(
route,
"work",
tmp_path,
state,
state_file,
tmp_path / "log",
lambda _note: True,
60,
)
assert result.returncode == 1
assert "claude_started" not in state
def test_terminate_escalates_after_term_timeout(monkeypatch):
class Process:
pid = 321

11
testing/tests/test_hermes_cli_records_edges.py
View File

@ -78,6 +78,17 @@ def test_small_snapshot_rejects_path_replaced_after_read(
assert lanes._open_small_json_snapshot(path) is None
def test_small_snapshot_rejects_a_reader_that_exceeds_its_bound(
tmp_path: Path,
monkeypatch,
):
path = tmp_path / "record"
path.write_text("{}", encoding="utf-8")
monkeypatch.setattr(lanes, "MAX_TERMINAL_RECORD_BYTES", 8)
monkeypatch.setattr(lanes, "_read_bounded", lambda *_args: b"x" * 9)
assert lanes._open_small_json_snapshot(path) is None
def test_recovery_snapshot_classifies_nonregular_and_open_failure(
tmp_path: Path,
monkeypatch,

34
testing/tests/test_hermes_cli_recovery_edges.py
View File

@ -87,6 +87,9 @@ def test_staged_authority_rejects_outside_nested_and_unknown_state(
document["kanban_state"] = "unknown"
assert lanes._staged_terminal_authority(staged, _Snapshot(document)) is None
valid = _pending_terminal_record("board", "task", 5, "accepted")
assert lanes._staged_terminal_authority(staged, _Snapshot(valid)) is None
def test_restore_staged_authority_handles_every_first_writer_state(
tmp_path: Path,
@ -296,3 +299,34 @@ def test_pending_guard_rejects_invalid_run_and_inaccessible_or_bad_artifacts(
monkeypatch.setattr(Path, "stat", denied)
assert lanes._has_pending_finalization("board", "task", 12) is False
def test_pending_guard_scans_invalid_prepared_and_staged_candidates(
tmp_path: Path,
monkeypatch,
):
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
base = lanes.state_path("board", "task")
pending = lanes._terminal_path(base, 13)
pending.parent.mkdir(parents=True)
pending.write_text("{}", encoding="utf-8")
(pending.parent / "task.run-13.terminal.prepared-bad.json").write_text(
"{}", encoding="utf-8"
)
(pending.parent / f"task.run-13.terminal.prepared-{'a' * 32}.json").write_text(
"{}", encoding="utf-8"
)
staged = [
pending.parent / f".retire.{'b' * 16}.{'c' * 16}.0",
pending.parent / f".retire.{'d' * 16}.{'e' * 16}.0",
]
for path in staged:
path.write_text("{}", encoding="utf-8")
snapshots = iter((None, _Snapshot({})))
monkeypatch.setattr(
lanes,
"_open_terminal_recovery_snapshot",
lambda _path: next(snapshots),
)
monkeypatch.setattr(lanes, "_staged_terminal_authority", lambda *_args: None)
assert lanes._has_pending_finalization("board", "task", 13) is False

46
testing/tests/test_hermes_cli_retention_edges.py
View File

@ -87,3 +87,49 @@ def test_gc_skips_symlink_board_and_disappearing_candidate(
monkeypatch.setattr(Path, "stat", disappear)
assert lanes.gc_lane_artifacts() == 0
def test_gc_counts_classified_quarantine_and_bounds_unlink_deferrals(
tmp_path: Path,
monkeypatch,
):
missing = tmp_path / "missing"
assert lanes._unlink_artifact_if_same(missing, tmp_path.stat()) is False
root = tmp_path / "lanes"
board = root / "board"
board.mkdir(parents=True)
classified = board / "classified.candidate-1.json"
classified.write_text("{}", encoding="utf-8")
monkeypatch.setattr(lanes, "STATE_ROOT", root)
monkeypatch.setattr(
lanes,
"_artifact_gc_candidates",
lambda _board: [classified],
)
monkeypatch.setattr(
lanes,
"_quarantine_invalid_retained_terminal",
lambda _path: (True, True),
)
assert lanes.gc_lane_artifacts() == 1
staged = board / f".retire.{'a' * 16}.{'b' * 16}.0"
staged.write_text("{}", encoding="utf-8")
monkeypatch.setattr(
lanes,
"_artifact_gc_candidates",
lambda _board: [staged],
)
monkeypatch.setattr(
lanes,
"_quarantine_invalid_retained_terminal",
lambda _path: (False, False),
)
monkeypatch.setattr(lanes, "_open_terminal_recovery_snapshot", lambda _path: None)
monkeypatch.setattr(lanes, "_unlink_artifact_if_same", lambda *_args: False)
assert lanes.gc_lane_artifacts(
max_age_seconds=-1,
max_count=0,
max_bytes=-1,
) == 0

96
testing/tests/test_hermes_runtime_stage_coverage.py
View File

@ -39,6 +39,21 @@ def test_private_directory_and_secret_copy_validate_mode_and_content(
assert (runtime / "value").read_text(encoding="utf-8") == "staged\n"
def test_private_directory_and_vault_projection_reject_invalid_entries(
tmp_path, monkeypatch
):
module, vault, runtime, _home = _stage(tmp_path, monkeypatch)
checks = iter((False, True))
with monkeypatch.context() as invalid_directory:
invalid_directory.setattr(Path, "is_symlink", lambda _path: next(checks))
with pytest.raises(RuntimeError, match="is invalid"):
module._owned_directory(runtime)
(vault / "invalid").write_bytes(b"")
with pytest.raises(RuntimeError, match="projection invalid is invalid"):
module._read_secret("invalid")
@pytest.mark.parametrize(
("document", "required", "match"),
[
@ -107,6 +122,77 @@ def test_node_ssh_config_rejects_nul_and_removes_partial_file(tmp_path, monkeypa
assert not (runtime / "node-ssh-config").exists()
def test_durable_credentials_reject_links_and_unsafe_existing_files(
tmp_path, monkeypatch
):
module, vault, runtime, _home = _stage(tmp_path, monkeypatch)
runtime.mkdir()
(vault / "credential").write_text(
json.dumps({"tokens": {"refresh_token": "synthetic"}}),
encoding="utf-8",
)
destination = runtime / "credential.json"
destination.symlink_to(vault / "credential")
with pytest.raises(RuntimeError, match="must not be a symlink"):
module._bootstrap_json("credential", destination, ("tokens", "refresh_token"))
destination.unlink()
destination.write_text(
json.dumps({"tokens": {"refresh_token": "durable"}}),
encoding="utf-8",
)
destination.chmod(0o644)
with pytest.raises(RuntimeError, match="is unsafe"):
module._bootstrap_json("credential", destination, ("tokens", "refresh_token"))
def test_durable_link_creates_repairs_preserves_and_rejects_targets(
tmp_path, monkeypatch
):
module, _vault, _runtime, _home = _stage(tmp_path, monkeypatch)
monkeypatch.setattr(module.os, "chown", lambda *_args: None)
runtime = tmp_path / "worker"
runtime.mkdir()
durable = tmp_path / "provider"
module._durable_link(runtime, durable, "state")
target = runtime / "state"
assert target.is_symlink() and target.resolve() == durable.resolve()
module._durable_link(runtime, durable, "state")
assert target.is_symlink()
target.unlink()
wrong = tmp_path / "wrong"
wrong.mkdir()
target.symlink_to(wrong)
module._durable_link(runtime, durable, "state")
assert target.resolve() == durable.resolve()
target.unlink()
target.write_text("collision", encoding="utf-8")
with pytest.raises(RuntimeError, match="refusing to replace"):
module._durable_link(runtime, durable, "state")
@pytest.mark.parametrize(
("function", "match"),
[
("stage_execution_worker", "worker ordinal is invalid"),
("stage_execution_mediator", "mediator ordinal is invalid"),
],
)
def test_execution_staging_rejects_invalid_ordinal(
tmp_path, monkeypatch, function, match
):
module, _vault, _runtime, _home = _stage(tmp_path, monkeypatch)
monkeypatch.setattr(module, "WORKER_ROOT", tmp_path / "worker")
monkeypatch.setattr(module, "PROVIDER_ACCESS_ROOT", tmp_path / "provider")
monkeypatch.setattr(module, "POOL_ACCESS_ROOT", tmp_path / "pool")
monkeypatch.setenv("HERMES_WORKER_ORDINAL", "9")
with pytest.raises(RuntimeError, match=match):
getattr(module, function)()
@pytest.mark.parametrize(
("mode", "secret_name"),
[("chat", "chat-relay-key"), ("triage", "triage-api-key")],
@ -124,13 +210,21 @@ def test_isolated_tenant_staging_exposes_only_required_secret(
)
@pytest.mark.parametrize("mode", ["agent", "chat", "triage"])
@pytest.mark.parametrize(
"mode", ["agent", "chat", "triage", "execution-worker", "execution-mediator"]
)
def test_main_dispatches_each_mode(mode, monkeypatch, capsys):
module = _load("stage_runtime_access")
calls = []
monkeypatch.setattr(module, "stage_agent", lambda: calls.append("agent"))
monkeypatch.setattr(module, "stage_chat", lambda: calls.append("chat"))
monkeypatch.setattr(module, "stage_triage", lambda: calls.append("triage"))
monkeypatch.setattr(
module, "stage_execution_worker", lambda: calls.append("execution-worker")
)
monkeypatch.setattr(
module, "stage_execution_mediator", lambda: calls.append("execution-mediator")
)
monkeypatch.setattr(sys, "argv", ["stage-runtime-access", mode])
assert module.main() == 0
assert calls == [mode]