GodopuandClaude Opus 5
cb7ed1aa38
Seven blocking items across three reviews, every one reproduced here before being accepted. Two themes run through them: a gate whose strictness depended on what the caller remembered to pass, and a boundary that trusted its only current caller instead of enforcing its own rule. Independent verification is no longer optional. Omitting the report directory skipped the disk check entirely and handed the decision back to the collector's own flags -- the self-certification this design exists to prevent. It is now a refusal, as is an empty directory or a job with no recorded dispatch. Evidence is now bound to its target. The policy compares a scenario against the dispatch context carried inside that same scenario, which is self-consistent by construction, so one job's evidence could complete another. The gate now checks the job, attempt, session and pane generation against the recorded dispatch. A reviewer with no verdict token is no longer completed. Phase 0 made this a safety contract and run_loop reads a missing token as NOT PASS, so inferring completion first would start a correction round against a review that was never finished. The requirement applies to reviewers only. Bookkeeping may write one field. A status-free request was an unrestricted merge on a terminal record: it could rewrite the prompt and the session. Settling is also conditional now on the exact intent that was sent, because publishing happens outside the lock and a newer intent can legitimately land mid-flight; clearing unconditionally dropped a notification nobody had sent. The intent's own identity travels in the payload, since the transport sequence cannot identify a retry of the same intent. Side-effect evidence must be evidence. A kind with no reference names something without evidencing it, and the CLI parser could not be the only check because the API and hand-written manifests bypass it. Also honours MAM_ENGINE_CAS=0, which the operations note already promised disables auto-completion, matches backend keys exactly rather than by prefix so a future name cannot inherit evidence never collected for it, records an absolute report path, and imports the notifier at module scope. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd
785 lines
34 KiB
Python
785 lines
34 KiB
Python
#!/usr/bin/env python3
|
|
"""Issue #5 Phase 2b — flag-gated auto-completion.
|
|
|
|
Characterize first, as in 2a: the interesting property here is what the system
|
|
*refuses* to do, so the refusals are pinned before anything is added that could
|
|
weaken them.
|
|
|
|
Sections
|
|
1. X-01..X-05 the fail-closed preconditions as they stand today.
|
|
2. finalize the completion manifest: written by the agent, re-derived here.
|
|
3. autocomplete the gate, and everything that must stop it.
|
|
4. notifier durable pending notification and restart recovery.
|
|
5. flags default off, evidence-backed allowlist, per-job opt-in.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
SKILLS = REPO_ROOT / ".agents" / "skills"
|
|
WATCHER_FIXTURES = REPO_ROOT / "tests" / "fixtures" / "issue_5" / "watcher"
|
|
|
|
|
|
def _reg(tmp_path) -> str:
|
|
d = tmp_path / ".mam" / "jobs"
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
return str(d)
|
|
|
|
|
|
def _make_job(registry_dir, **over):
|
|
from lib_py.delegation import registry
|
|
return registry.register_job(over.pop("prompt", "p"), registry_dir=registry_dir, **over)
|
|
|
|
|
|
def _load(registry_dir, job_id):
|
|
from lib_py.delegation import mqtt_common
|
|
return mqtt_common.load_job(job_id, registry_dir)
|
|
|
|
|
|
def _fingerprint(path: Path) -> dict:
|
|
out = {}
|
|
for p in sorted(Path(path).rglob("*")):
|
|
if p.is_file():
|
|
out[str(p.relative_to(path))] = hashlib.sha256(p.read_bytes()).hexdigest()
|
|
return out
|
|
|
|
|
|
def _scenario(name="finalized_candidate"):
|
|
return json.loads((WATCHER_FIXTURES / (name + ".json")).read_text(encoding="utf-8"))
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _default_flags(monkeypatch):
|
|
"""Auto-completion is off unless a test turns it on. That is the product
|
|
default too, and every test that does not opt in is therefore also a check
|
|
that the default really is off."""
|
|
monkeypatch.delenv("MAM_AUTO_COMPLETE", raising=False)
|
|
monkeypatch.delenv("MAM_AUTO_COMPLETE_BACKENDS", raising=False)
|
|
monkeypatch.setenv("MAM_ENGINE_CAS", "1")
|
|
|
|
|
|
# ==============================================================================
|
|
# 1. X-01..X-05 — the fail-closed preconditions, before anything is added
|
|
# ==============================================================================
|
|
|
|
def test_x01_a_job_without_a_manifest_is_never_approved():
|
|
"""The property that makes unsupported agents safe without any new branch."""
|
|
import copy
|
|
from lib_py.loop import policy
|
|
|
|
scenario = _scenario()
|
|
assert policy.decide(scenario).decision == "candidate_approved"
|
|
stripped = copy.deepcopy(scenario)
|
|
for item in stripped["timeline"]:
|
|
(item.get("payload") or {}).pop("report_manifest", None)
|
|
stripped["filesystem"]["manifest_committed"] = False
|
|
decision = policy.decide(stripped)
|
|
assert decision.decision == "hold" and "manifest" in decision.reason
|
|
|
|
|
|
def test_x02_no_production_code_writes_a_completion_manifest_yet():
|
|
"""Pins the starting point: the protocol is entirely new in phase 2b."""
|
|
hits = []
|
|
for path in sorted((SKILLS / "lib_py").rglob("*.py")):
|
|
if path.name in ("finalize.py", "autocomplete.py"):
|
|
continue
|
|
text = path.read_text(encoding="utf-8")
|
|
if "report-manifest" in text:
|
|
hits.append(path.name)
|
|
assert not hits, "manifest writers already exist: {}".format(hits)
|
|
|
|
|
|
def test_x03_the_record_carries_iteration_not_an_attempt_id(mam_sandbox, tmp_path, monkeypatch):
|
|
"""Why auto-completion does not use the engine's expected_attempt.
|
|
|
|
The engine compares expected_attempt against `attempt_id` falling back to
|
|
`iteration`. A record has only the latter, and the two mean different things,
|
|
so the attempt guard is applied where attempt identity actually lives: the
|
|
dispatch record in the shadow store, compared against observed evidence."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
record = _load(reg, job_id)
|
|
assert "iteration" in record
|
|
assert "attempt_id" not in record, (
|
|
"if this ever changes, revisit the layering decision in the phase 2b report"
|
|
)
|
|
|
|
|
|
def test_x04_policy_rejects_evidence_from_a_superseded_attempt():
|
|
"""The attempt guard that auto-completion actually relies on."""
|
|
import copy
|
|
from lib_py.loop import policy
|
|
|
|
scenario = _scenario()
|
|
superseded = copy.deepcopy(scenario)
|
|
superseded["dispatch_context"]["attempt_id"] = 2 # evidence still says 1
|
|
decision = policy.decide(superseded)
|
|
assert decision.decision == "hold" and "attempt_id" in decision.reason
|
|
|
|
|
|
def test_x05_the_explicit_event_path_is_untouched_by_default(mam_sandbox, tmp_path, monkeypatch):
|
|
"""An explicitly completed job looks exactly as it always did."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import mqtt_common
|
|
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="running")
|
|
mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed")
|
|
record = _load(reg, job_id)
|
|
assert record["status"] == "completed"
|
|
assert "completed_by" not in record, "an explicit completion carries no auto marker"
|
|
|
|
|
|
# ==============================================================================
|
|
# 2. finalize — declared by the agent, re-derived by the reader
|
|
# ==============================================================================
|
|
|
|
def _report(tmp_path, body="done\n", staged=False):
|
|
d = tmp_path / "claude-reports"
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
path = d / "report-final.md"
|
|
(path.with_name(path.name + ".tmp") if staged else path).write_text(body, encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def test_finalize_promotes_a_staged_report_atomically(tmp_path):
|
|
"""A reader must never see a half-written report as finalized."""
|
|
from lib_py.delegation import finalize
|
|
|
|
path = _report(tmp_path, "final body\n", staged=True)
|
|
assert not path.exists() and path.with_name(path.name + ".tmp").exists()
|
|
manifest = finalize.finalize_report(path, "job1", 1)
|
|
assert path.exists() and not path.with_name(path.name + ".tmp").exists()
|
|
assert manifest["sha256"] == finalize.sha256_of(path)
|
|
|
|
|
|
def test_finalize_verifies_against_the_file_not_the_declaration(tmp_path):
|
|
"""The hash is recomputed, so a forged manifest cannot certify itself."""
|
|
from lib_py.delegation import finalize
|
|
|
|
path = _report(tmp_path)
|
|
finalize.finalize_report(path, "job1", 1)
|
|
assert finalize.verify_manifest(path.parent, "job1").ok
|
|
|
|
manifest_file = finalize.manifest_in(path.parent)
|
|
forged = json.loads(manifest_file.read_text(encoding="utf-8"))
|
|
forged["sha256"] = "0" * 64
|
|
manifest_file.write_text(json.dumps(forged), encoding="utf-8")
|
|
verdict = finalize.verify_manifest(path.parent, "job1")
|
|
assert not verdict.ok and "changed after finalization" in verdict.reason
|
|
|
|
|
|
def test_finalize_is_invalidated_by_a_rewritten_report(tmp_path):
|
|
"""A new body needs a new finalization; the old hash is void immediately."""
|
|
from lib_py.delegation import finalize
|
|
|
|
path = _report(tmp_path, "first\n")
|
|
finalize.finalize_report(path, "job1", 1)
|
|
assert finalize.verify_manifest(path.parent, "job1").ok
|
|
path.write_text("rewritten\n", encoding="utf-8")
|
|
verdict = finalize.verify_manifest(path.parent, "job1")
|
|
assert not verdict.ok and "changed after finalization" in verdict.reason
|
|
|
|
|
|
@pytest.mark.parametrize("kwargs,fragment", [
|
|
({"job_id": "other"}, "belongs to job"),
|
|
({"expected_attempt": 9}, "belongs to attempt"),
|
|
])
|
|
def test_finalize_rejects_a_manifest_for_a_different_job_or_attempt(tmp_path, kwargs, fragment):
|
|
from lib_py.delegation import finalize
|
|
path = _report(tmp_path)
|
|
finalize.finalize_report(path, "job1", 1)
|
|
args = {"job_id": "job1"}
|
|
args.update(kwargs)
|
|
verdict = finalize.verify_manifest(path.parent, **args)
|
|
assert not verdict.ok and fragment in verdict.reason
|
|
|
|
|
|
def test_finalize_requires_evidence_for_every_declared_side_effect(tmp_path):
|
|
"""Partial evidence is no evidence: the work is not known to be complete."""
|
|
from lib_py.delegation import finalize
|
|
|
|
path = _report(tmp_path)
|
|
finalize.finalize_report(path, "job1", 1,
|
|
side_effects=finalize.parse_side_effects(["git_push=abc123"]))
|
|
ok = finalize.verify_manifest(path.parent, "job1", expected_artifacts=["git_push"])
|
|
assert ok.ok
|
|
missing = finalize.verify_manifest(path.parent, "job1",
|
|
expected_artifacts=["git_push", "api_call"])
|
|
assert not missing.ok and "api_call" in missing.reason
|
|
|
|
|
|
def test_finalize_rejects_a_malformed_side_effect_pair():
|
|
from lib_py.delegation import finalize
|
|
with pytest.raises(ValueError):
|
|
finalize.parse_side_effects(["nokey"])
|
|
with pytest.raises(ValueError):
|
|
finalize.parse_side_effects(["kind="])
|
|
|
|
|
|
def test_finalize_missing_manifest_is_a_refusal_not_a_crash(tmp_path):
|
|
from lib_py.delegation import finalize
|
|
d = tmp_path / "empty"
|
|
d.mkdir()
|
|
verdict = finalize.verify_manifest(d, "job1")
|
|
assert not verdict.ok and "no completion manifest" in verdict.reason
|
|
|
|
|
|
def test_publish_event_finalizes_before_it_publishes(mam_sandbox, tmp_path, monkeypatch):
|
|
"""The CLI surface an agent actually calls."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import publish_event
|
|
from lib_py.delegation import finalize
|
|
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
record = _load(reg, job_id)
|
|
record["broker"] = {"host": "127.0.0.1", "port": 65432, "tls": False}
|
|
Path(reg, job_id + ".json").write_text(json.dumps(record), encoding="utf-8")
|
|
|
|
path = _report(tmp_path, "the report\n", staged=True)
|
|
rc = publish_event.main([
|
|
"--registry-dir", reg, "--job", job_id, "--event", "completed",
|
|
"--attempts", "1", "--finalize-report", str(path),
|
|
"--side-effect", "git_push=deadbeef",
|
|
])
|
|
assert rc == 2, "the broker is unreachable, but finalization still happened"
|
|
assert path.exists(), "the staged report was promoted"
|
|
verdict = finalize.verify_manifest(path.parent, job_id)
|
|
assert verdict.ok
|
|
assert verdict.manifest["side_effects"][0]["kind"] == "git_push"
|
|
|
|
|
|
def test_publish_event_reports_a_bad_side_effect_instead_of_publishing(mam_sandbox, tmp_path, monkeypatch):
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import publish_event
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
path = _report(tmp_path)
|
|
rc = publish_event.main([
|
|
"--registry-dir", reg, "--job", job_id, "--event", "completed",
|
|
"--finalize-report", str(path), "--side-effect", "malformed",
|
|
])
|
|
assert rc == 1
|
|
|
|
|
|
# ==============================================================================
|
|
# 3. autocomplete — the gate, and everything that must stop it
|
|
# ==============================================================================
|
|
|
|
def _ready_job(tmp_path, mam_sandbox, *, backend="claude-code", opt_in=True,
|
|
artifacts=None, body="report\n"):
|
|
"""A job in exactly the state that should auto-complete, plus its evidence."""
|
|
from lib_py.delegation import finalize
|
|
from lib_py.loop import store
|
|
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg, agent=backend, expected_artifacts=artifacts or [])
|
|
from lib_py.delegation import engine
|
|
engine.transition(job_id, "claim", registry_dir=reg)
|
|
if opt_in:
|
|
engine.transition(job_id, "progress", registry_dir=reg,
|
|
fields={"auto_complete": True})
|
|
|
|
scenario = _scenario()
|
|
dispatch = scenario["dispatch_context"]
|
|
dispatch["job_id"] = job_id
|
|
for item in scenario["timeline"]:
|
|
env = item.get("collector_attribution")
|
|
if env:
|
|
env["job_id"] = job_id
|
|
# A non-herdr payload is a legitimate identifier source, so it has to
|
|
# agree too. Leaving the fixture's original id here is a contradiction,
|
|
# and the policy correctly refuses it.
|
|
payload = item.get("payload") or {}
|
|
if item.get("source_kind") != "herdr" and "job_id" in payload:
|
|
payload["job_id"] = job_id
|
|
store.record_dispatch(job_id, dispatch["attempt_id"], dispatch["agent_session_uuid"],
|
|
dispatch["pane_process_generation"], workspace=str(mam_sandbox))
|
|
store.mark_execution_anchor(job_id, workspace=str(mam_sandbox))
|
|
|
|
report = _report(tmp_path, body)
|
|
finalize.finalize_report(report, job_id, dispatch["attempt_id"],
|
|
side_effects=finalize.parse_side_effects(
|
|
["{}=ref".format(a) for a in (artifacts or [])]))
|
|
return job_id, reg, scenario, report.parent
|
|
|
|
|
|
def test_gate_commits_when_every_condition_holds(mam_sandbox, tmp_path, monkeypatch):
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed, result.reason
|
|
record = _load(reg, job_id)
|
|
assert record["status"] == "completed"
|
|
assert record["completed_by"] == "auto_complete", "an inferred terminal must be marked"
|
|
assert record["pending_notification"]["event"] == "completed"
|
|
|
|
|
|
def test_g4_default_off_commits_nothing_and_leaves_the_registry_untouched(
|
|
mam_sandbox, tmp_path, monkeypatch
|
|
):
|
|
"""G4: with the flag unset, the registry must not change at all."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
|
|
before = _fingerprint(Path(reg))
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed is False
|
|
assert "disabled" in result.reason
|
|
assert _fingerprint(Path(reg)) == before, "the default must write nothing"
|
|
|
|
|
|
@pytest.mark.parametrize("backend", ["agy", "hermes", "opencode", "unknown-agent"])
|
|
def test_g9_a_backend_without_quiescence_evidence_never_auto_completes(
|
|
mam_sandbox, tmp_path, monkeypatch, backend
|
|
):
|
|
"""Inferring 'finished' from stillness needs stillness to have been observed."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox, backend=backend)
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed is False and "quiescent-state evidence" in result.reason
|
|
assert _load(reg, job_id)["status"] == "running"
|
|
|
|
|
|
def test_the_allowlist_can_be_narrowed_but_never_widened(monkeypatch):
|
|
from lib_py.loop import autocomplete
|
|
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE_BACKENDS", "claude")
|
|
assert autocomplete.allowed_backends() == ("claude",)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE_BACKENDS", "claude,hermes,opencode")
|
|
assert autocomplete.allowed_backends() == ("claude",), "unobserved backends are dropped"
|
|
|
|
|
|
def test_a_job_that_did_not_opt_in_is_not_auto_completed(mam_sandbox, tmp_path, monkeypatch):
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox, opt_in=False)
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed is False and "opt in" in result.reason
|
|
|
|
|
|
def test_g5_a_rewritten_report_stops_the_gate(mam_sandbox, tmp_path, monkeypatch):
|
|
"""G5: the manifest is checked against the file, so a late edit disqualifies."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
|
|
(report_dir / "report-final.md").write_text("rewritten after finalization\n", encoding="utf-8")
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed is False and "changed after finalization" in result.reason
|
|
assert _load(reg, job_id)["status"] == "running"
|
|
|
|
|
|
def test_g5_missing_side_effect_evidence_stops_the_gate(mam_sandbox, tmp_path, monkeypatch):
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete
|
|
from lib_py.delegation import finalize
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox,
|
|
artifacts=["git_push"])
|
|
# re-declare without the evidence the job said it would produce
|
|
finalize.finalize_report(report_dir / "report-final.md", job_id, 1, side_effects=[])
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed is False and "git_push" in result.reason
|
|
|
|
|
|
def test_d7_a_policy_that_cannot_decide_holds(mam_sandbox, tmp_path, monkeypatch):
|
|
"""'Could not decide' is never 'done'."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete, policy
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
|
|
|
|
def explode(_scenario):
|
|
raise RuntimeError("evidence is contradictory")
|
|
|
|
monkeypatch.setattr(policy, "decide", explode)
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed is False and "policy error" in result.reason
|
|
assert _load(reg, job_id)["status"] == "running"
|
|
|
|
|
|
def test_a_late_explicit_terminal_beats_the_gate(mam_sandbox, tmp_path, monkeypatch):
|
|
"""If the agent already reported, the boundary refuses and that is correct."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.delegation import mqtt_common
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
|
|
mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="error")
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed is False
|
|
record = _load(reg, job_id)
|
|
assert record["status"] == "error", "the explicit report wins"
|
|
assert "completed_by" not in record
|
|
|
|
|
|
# ==============================================================================
|
|
# 4. notifier — the intent is durable, and recovery re-sends it
|
|
# ==============================================================================
|
|
|
|
def test_the_notification_intent_is_committed_with_the_state(mam_sandbox, tmp_path, monkeypatch):
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import engine, notifier
|
|
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
engine.transition(job_id, "claim", registry_dir=reg)
|
|
engine.transition(job_id, "completed", registry_dir=reg, notify="completed")
|
|
record = _load(reg, job_id)
|
|
assert record["status"] == "completed"
|
|
intent = notifier.pending(job_id, reg)
|
|
assert intent and intent["event"] == "completed" and intent["seq"] == record["transition_seq"]
|
|
|
|
|
|
def test_flush_publishes_then_settles(mam_sandbox, tmp_path, monkeypatch):
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import engine, notifier
|
|
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
engine.transition(job_id, "claim", registry_dir=reg)
|
|
engine.transition(job_id, "completed", registry_dir=reg, notify="completed")
|
|
|
|
sent = []
|
|
assert notifier.flush(job_id, reg, publish=lambda j, i, r: sent.append(i) or 0)
|
|
assert len(sent) == 1 and sent[0]["event"] == "completed"
|
|
assert notifier.pending(job_id, reg) is None
|
|
# settling is idempotent: a second flush has nothing to do
|
|
assert notifier.flush(job_id, reg, publish=lambda *a: 1 / 0)
|
|
|
|
|
|
def test_a_failed_send_leaves_the_intent_pending(mam_sandbox, tmp_path, monkeypatch):
|
|
"""An unsent notification that looks sent is worse than one that is retried."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import engine, notifier
|
|
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
engine.transition(job_id, "claim", registry_dir=reg)
|
|
engine.transition(job_id, "completed", registry_dir=reg, notify="completed")
|
|
|
|
assert notifier.flush(job_id, reg, publish=lambda *a: 2) is False
|
|
assert notifier.pending(job_id, reg) is not None
|
|
assert notifier.flush(job_id, reg, publish=lambda *a: (_ for _ in ()).throw(OSError())) is False
|
|
assert notifier.pending(job_id, reg) is not None
|
|
assert _load(reg, job_id)["status"] == "completed", "the state stays committed regardless"
|
|
|
|
|
|
def test_g7_restart_recovery_resends_every_owed_notification(mam_sandbox, tmp_path, monkeypatch):
|
|
"""G7: the crash window between committing and publishing."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import engine, notifier
|
|
|
|
reg = _reg(tmp_path)
|
|
owed = []
|
|
for _ in range(3):
|
|
job_id = _make_job(reg)
|
|
engine.transition(job_id, "claim", registry_dir=reg)
|
|
engine.transition(job_id, "completed", registry_dir=reg, notify="completed")
|
|
owed.append(job_id)
|
|
settled = _make_job(reg)
|
|
engine.transition(settled, "claim", registry_dir=reg)
|
|
|
|
assert sorted(notifier.jobs_with_pending(reg)) == sorted(owed)
|
|
|
|
sent = []
|
|
results = notifier.recover(reg, publish=lambda j, i, r: sent.append(j) or 0)
|
|
assert all(results.values()) and sorted(sent) == sorted(owed)
|
|
assert notifier.jobs_with_pending(reg) == [], "nothing is owed after recovery"
|
|
|
|
|
|
# ==============================================================================
|
|
# 6. Cross-review regressions (job f63d8279)
|
|
#
|
|
# Three reviewers, seven blocking items, all reproduced before being accepted.
|
|
# The theme: a gate whose strictness depended on the caller, and a boundary that
|
|
# trusted its only current caller instead of enforcing its own rule.
|
|
# ==============================================================================
|
|
|
|
def test_f1_an_omitted_report_directory_is_a_refusal_not_a_shortcut(
|
|
mam_sandbox, tmp_path, monkeypatch
|
|
):
|
|
"""Independent verification cannot be optional: skipping it puts the
|
|
collector's own flags back in charge of the decision."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, _report_dir = _ready_job(tmp_path, mam_sandbox)
|
|
before = _fingerprint(Path(reg))
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox)) # no report_dir
|
|
assert result.committed is False and "independently verify" in result.reason
|
|
assert _fingerprint(Path(reg)) == before, "a refusal must not write"
|
|
|
|
|
|
def test_f1_an_empty_report_directory_is_also_a_refusal(mam_sandbox, tmp_path, monkeypatch):
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, _rd = _ready_job(tmp_path, mam_sandbox)
|
|
empty = tmp_path / "empty-reports"
|
|
empty.mkdir()
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=empty)
|
|
assert result.committed is False and "no completion manifest" in result.reason
|
|
|
|
|
|
def test_r2_evidence_for_another_job_cannot_complete_this_one(
|
|
mam_sandbox, tmp_path, monkeypatch
|
|
):
|
|
"""The policy checks a scenario against the dispatch context inside that same
|
|
scenario, which is self-consistent by construction. The gate must bind the
|
|
evidence to the target."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
|
|
scenario["dispatch_context"]["job_id"] = "some-other-job"
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed is False and "not" in result.reason
|
|
assert _load(reg, job_id)["status"] == "running"
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["attempt_id", "agent_session_uuid", "pane_process_generation"])
|
|
def test_r2_evidence_must_match_the_recorded_dispatch(mam_sandbox, tmp_path, monkeypatch, field):
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
|
|
scenario["dispatch_context"][field] = "not-what-was-dispatched"
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed is False and field in result.reason
|
|
|
|
|
|
def test_r2_a_job_with_no_recorded_dispatch_is_refused(mam_sandbox, tmp_path, monkeypatch):
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
|
|
(Path(mam_sandbox) / ".mam" / "shadow" / "attribution" / (job_id + ".json")).unlink()
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed is False and "no recorded dispatch" in result.reason
|
|
|
|
|
|
def test_the_cas_rollback_switch_also_disables_auto_completion(mam_sandbox, tmp_path, monkeypatch):
|
|
"""Inferred completion is built on the CAS boundary, so committing while
|
|
that boundary is rolled back would bypass it."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
|
|
monkeypatch.setenv("MAM_ENGINE_CAS", "0")
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed is False and "rolled back" in result.reason
|
|
assert _load(reg, job_id)["status"] == "running"
|
|
|
|
|
|
@pytest.mark.parametrize("agent", ["claude-experimental", "codexish", "grok2"])
|
|
def test_o4_a_backend_key_must_match_exactly_not_by_prefix(
|
|
mam_sandbox, tmp_path, monkeypatch, agent
|
|
):
|
|
"""A future name must not inherit evidence that was never collected for it."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox, backend=agent)
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed is False and "quiescent-state evidence" in result.reason
|
|
|
|
|
|
def test_a_reviewer_without_a_verdict_token_is_not_completed(mam_sandbox, tmp_path, monkeypatch):
|
|
"""Phase 0 safety contract: a reviewer that has not written its verdict has
|
|
not finished reviewing, and run_loop reads a missing token as NOT PASS."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.delegation import finalize
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(
|
|
tmp_path, mam_sandbox, body="a thorough review with no verdict line\n")
|
|
from lib_py.delegation import engine
|
|
engine.transition(job_id, "progress", registry_dir=reg, fields={"role": "Reviewer"})
|
|
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed is False and "verdict token" in result.reason
|
|
assert _load(reg, job_id)["status"] == "running"
|
|
|
|
# writing the verdict requires a fresh finalization, and then it passes
|
|
report = report_dir / "report-final.md"
|
|
report.write_text("a thorough review\n\n[VERDICT: PASS]\n", encoding="utf-8")
|
|
finalize.finalize_report(report, job_id, scenario["dispatch_context"]["attempt_id"])
|
|
ok = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert ok.committed, ok.reason
|
|
|
|
|
|
def test_a_worker_role_needs_no_verdict_token(mam_sandbox, tmp_path, monkeypatch):
|
|
"""The token requirement is a reviewer contract, not a general one."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
|
|
from lib_py.loop import autocomplete
|
|
|
|
job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox,
|
|
body="worker output, no verdict\n")
|
|
assert _load(reg, job_id)["role"] == "Worker"
|
|
result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
|
|
workspace=str(mam_sandbox), report_dir=report_dir)
|
|
assert result.committed, result.reason
|
|
|
|
|
|
def test_grok_f2_bookkeeping_cannot_merge_arbitrary_fields(mam_sandbox, tmp_path, monkeypatch):
|
|
"""A status-free request must not become an unrestricted merge on a record
|
|
nothing else is allowed to touch."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import engine
|
|
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
engine.transition(job_id, "claim", registry_dir=reg)
|
|
engine.transition(job_id, "completed", registry_dir=reg, notify="completed")
|
|
before = _load(reg, job_id)
|
|
|
|
result = engine.transition(job_id, "notification_settled", registry_dir=reg,
|
|
fields={"pending_notification": None, "prompt": "HACKED",
|
|
"agent_session": "herdr:attacker"},
|
|
source="not-an-operator")
|
|
after = _load(reg, job_id)
|
|
assert result.committed is False and "may not merge" in result.reason
|
|
assert after["prompt"] == before["prompt"]
|
|
assert after["agent_session"] == before["agent_session"]
|
|
assert after["pending_notification"], "the intent survives a refused settle"
|
|
|
|
# the legitimate settle still works
|
|
from lib_py.delegation import notifier
|
|
assert notifier.settle(job_id, reg, sent_intent=after["pending_notification"])
|
|
assert notifier.pending(job_id, reg) is None
|
|
|
|
|
|
def test_r4_a_newer_intent_is_not_erased_by_a_slow_send(mam_sandbox, tmp_path, monkeypatch):
|
|
"""Publishing happens outside the lock, so a replacement can land mid-flight.
|
|
Settling unconditionally would drop a notification nobody sent."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import engine, notifier
|
|
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
engine.transition(job_id, "claim", registry_dir=reg)
|
|
engine.transition(job_id, "completed", registry_dir=reg, notify="completed")
|
|
|
|
def racing_publisher(job, intent, registry_dir):
|
|
engine.transition(job, "notification_settled", registry_dir=registry_dir,
|
|
fields={"pending_notification":
|
|
notifier.intent_for("error", 999, "later")})
|
|
return 0
|
|
|
|
notifier.flush(job_id, reg, publish=racing_publisher)
|
|
still_owed = notifier.pending(job_id, reg)
|
|
assert still_owed and still_owed["seq"] == 999, "the replacement must survive"
|
|
|
|
|
|
def test_r4_the_published_payload_carries_the_intent_identity(mam_sandbox, tmp_path, monkeypatch):
|
|
"""The transport sequence cannot identify a retry of the same intent, so the
|
|
intent's own identity travels in the payload for consumers to dedupe on."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import engine, notifier
|
|
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
record = _load(reg, job_id)
|
|
record["broker"] = {"host": "127.0.0.1", "port": 65432, "tls": False}
|
|
Path(reg, job_id + ".json").write_text(json.dumps(record), encoding="utf-8")
|
|
engine.transition(job_id, "claim", registry_dir=reg)
|
|
engine.transition(job_id, "completed", registry_dir=reg, notify="completed")
|
|
|
|
intent = notifier.pending(job_id, reg)
|
|
notifier.flush(job_id, reg, attempts=1) # real path; broker is unreachable
|
|
from lib_py.delegation import mqtt_common
|
|
events = list(mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir()))
|
|
published = [e for e in events if e.get("event") == "published"]
|
|
assert published, "the attempt is recorded even though the broker refused"
|
|
carried = published[-1]["payload"]["data"].get("notification_intent")
|
|
assert carried and carried["seq"] == intent["seq"]
|
|
|
|
|
|
@pytest.mark.parametrize("entry", [
|
|
{"kind": "deploy"},
|
|
{"kind": "deploy", "reference": " "},
|
|
{"reference": "abc"},
|
|
"not-an-object",
|
|
])
|
|
def test_r3_a_side_effect_entry_must_carry_real_evidence(tmp_path, entry):
|
|
"""A kind with no reference names something without evidencing it."""
|
|
from lib_py.delegation import finalize
|
|
|
|
path = _report(tmp_path)
|
|
with pytest.raises(ValueError):
|
|
finalize.finalize_report(path, "job1", 1, side_effects=[entry])
|
|
|
|
# and a hand-written manifest is rejected at verification too
|
|
finalize.finalize_report(path, "job1", 1,
|
|
side_effects=[{"kind": "deploy", "reference": "sha"}])
|
|
manifest_file = finalize.manifest_in(path.parent)
|
|
forged = json.loads(manifest_file.read_text(encoding="utf-8"))
|
|
forged["side_effects"] = [entry]
|
|
manifest_file.write_text(json.dumps(forged), encoding="utf-8")
|
|
verdict = finalize.verify_manifest(path.parent, "job1", expected_artifacts=["deploy"])
|
|
assert not verdict.ok
|
|
|
|
|
|
def test_o2_the_manifest_records_an_absolute_report_path(tmp_path, monkeypatch):
|
|
"""A relative path would resolve against whatever directory the reader is in."""
|
|
from lib_py.delegation import finalize
|
|
|
|
path = _report(tmp_path)
|
|
monkeypatch.chdir(tmp_path)
|
|
manifest = finalize.finalize_report(Path("claude-reports/report-final.md"), "job1", 1)
|
|
assert Path(manifest["report_path"]).is_absolute()
|
|
monkeypatch.chdir(REPO_ROOT)
|
|
assert finalize.verify_manifest(path.parent, "job1").ok, "verification survives a chdir"
|