diff --git a/.agents/skills/lib_py/delegation/engine.py b/.agents/skills/lib_py/delegation/engine.py index 65c14f2..9952f60 100644 --- a/.agents/skills/lib_py/delegation/engine.py +++ b/.agents/skills/lib_py/delegation/engine.py @@ -48,13 +48,26 @@ REQUEST_TO_STATUS = { "cancelled": "cancelled", "deadline_timeout": "error", "merge": None, + # Operator-authority requests. The wrapper drives multi-iteration delegation + # (`submit --type loop|discuss`) by re-dispatching a finished job to the next + # session, and by finalising it as an error when the iteration budget is + # exhausted. Both legitimately move a terminal record, which is why they are + # separate requests gated on authority rather than an exception in the table. + "redispatch": "pending", + "finalize": "error", } +# Only these sources may move a record out of a terminal state. An agent can +# never walk back its own terminal; that is the whole point of the CAS table. +OPERATOR_SOURCES = ("registry_cli", "operator") +OPERATOR_REQUESTS = ("redispatch", "finalize") + # A caller that merges `status=` is making the request that owns that # value; mapping it here is what stops a raw field write from sidestepping the -# CAS table. `pending` has no owning request (nothing transitions *to* pending), -# so it stays an ordinary metadata merge, exactly as before. +# CAS table. Every status has an owning request, `pending` included: the wrapper +# resets a finished job to pending to hand it to the next iteration. STATUS_TO_REQUEST = { + "pending": "redispatch", "running": "started", "completed": "completed", "error": "error", @@ -93,17 +106,37 @@ class _Decision(object): self.conflict = conflict -def _decide(record: Dict[str, Any], request: str, expected_status, expected_attempt) -> _Decision: +def _decide(record: Dict[str, Any], request: str, expected_status, expected_attempt, + source="unknown", fields=None) -> _Decision: """Pure CAS decision. No I/O, so it is safe to call under the lock. Implements the target table in docs/contracts/issue-5/state-transitions.md ยง3. """ current = record.get("status") + fields = fields or {} - # Stale attempt or generation loses before anything else is considered. + # An unrecognised request must never fall through to the permissive branch. + if request not in REQUEST_TO_STATUS: + return _Decision(False, current, "unknown request: {!r}".format(request)) + + # A status may only be set by the request that owns it. Letting one ride in + # as an ordinary field would let `progress` commit a terminal, which is the + # decision table being bypassed from inside its own boundary. + if "status" in fields: + return _Decision(False, current, + "status must be requested, not merged as a field " + "(got status={!r} on request {!r})".format(fields["status"], request)) + + # A supplied precondition must be *proved*, not merely un-contradicted. If a + # caller names an expected attempt and the record cannot show one, the + # compare-and-swap has nothing to compare and must fail closed. if expected_attempt is not None: actual = record.get("attempt_id", record.get("iteration")) - if actual is not None and actual != expected_attempt: + if actual is None: + return _Decision(False, current, + "expected attempt {!r} but the record carries none".format( + expected_attempt)) + if actual != expected_attempt: return _Decision(False, current, "stale attempt: expected {!r}, record has {!r}".format( expected_attempt, actual)) @@ -115,10 +148,34 @@ def _decide(record: Dict[str, Any], request: str, expected_status, expected_atte requested = REQUEST_TO_STATUS.get(request, None) + # Authority first, so a refusal says *why* rather than blaming the state. + if request in OPERATOR_REQUESTS and source not in OPERATOR_SOURCES: + return _Decision(False, current, + "{} requires operator authority (source={!r})".format(request, source)) + if current in TERMINAL_STATUSES: # Idempotent re-receipt of the same terminal is a no-op ack. if requested == current: return _Decision(False, current, "idempotent: already {}".format(current)) + + operator = source in OPERATOR_SOURCES + if operator and request == "redispatch": + # Re-dispatching a finished job to the next iteration. The attempt + # must move forward, so a replayed redispatch cannot resurrect it. + new_iteration = fields.get("iteration") + old_iteration = record.get("iteration") + if new_iteration is None: + return _Decision(False, current, + "redispatch requires an iteration", conflict=True) + if old_iteration is not None and int(new_iteration) <= int(old_iteration): + return _Decision(False, current, + "redispatch iteration {!r} does not advance past {!r}".format( + new_iteration, old_iteration), conflict=True) + return _Decision(True, "pending", "{} -> pending (redispatch)".format(current)) + + if operator and request in ("finalize", "error"): + return _Decision(True, "error", "{} -> error (finalize)".format(current)) + if requested is None: return _Decision(False, current, "late {} after terminal {}; refusing".format(request, current), @@ -128,8 +185,8 @@ def _decide(record: Dict[str, Any], request: str, expected_status, expected_atte conflict=True) if requested is None: - # progress / permission_required / metadata-only merge: the engine does - # not set a status, so any status in `fields` is applied as-is. + # progress / permission_required / metadata-only merge. No status is set + # here: a status-bearing field was already refused above. return _Decision(True, None, "metadata updated ({})".format(request)) return _Decision(True, requested, "{} -> {}".format(current, requested)) @@ -155,7 +212,8 @@ def transition(job_id, request="merge", *, registry_dir=None, expected_status=No with mqtt_common.registry_lock(registry_dir): record = mqtt_common.load_job(job_id, registry_dir) from_status = record.get("status") - decision = _decide(record, request, expected_status, expected_attempt) + decision = _decide(record, request, expected_status, expected_attempt, + source=source, fields=fields) notifications: List[Dict[str, Any]] = [] if decision.commit: @@ -206,7 +264,8 @@ def _flush(notifications, registry_dir) -> None: try: if kind == "status_changed": mqtt_common.update_logged_status( - job_id, note["to"], updated_at=note.get("timestamp")) + job_id, note["to"], updated_at=note.get("timestamp"), + transition_seq=note.get("transition_seq")) mqtt_common.append_event(job_id, { "event": "status_changed", "from": note.get("from"), diff --git a/.agents/skills/lib_py/delegation/mqtt_common.py b/.agents/skills/lib_py/delegation/mqtt_common.py index 901d946..61830e9 100644 --- a/.agents/skills/lib_py/delegation/mqtt_common.py +++ b/.agents/skills/lib_py/delegation/mqtt_common.py @@ -392,7 +392,8 @@ def _atomic_write_record(job_id: str, registry_dir: str, record: Dict[str, Any]) raise -def update_job_status(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR, **fields: Any) -> Dict[str, Any]: +def update_job_status(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR, + _source: str = "unknown", **fields: Any) -> Dict[str, Any]: """Merge ``fields`` into a job record through the single transition boundary. Signature, return value and exceptions are unchanged: phase 1's G-1..G-4 @@ -408,17 +409,31 @@ def update_job_status(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR, **f if not engine.cas_enabled(): return _update_job_status_legacy(job_id, registry_dir, **fields) + # The status-to-request mapping lives in transition_job_status so it happens + # exactly once; doing it here as well would strip the status before the + # mapping could see it. + result = transition_job_status(job_id, registry_dir, _source=_source, **fields) + return result.record + + +def transition_job_status(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR, + _source: str = "unknown", **fields: Any): + """Same write, but returns the full TransitionResult. + + ``update_job_status`` must keep returning the record because phase 1's + contracts and the shell helpers depend on that shape. A caller that needs to + know whether the transition was actually committed -- the operator CLI, which + used to exit 0 on a silently refused write -- uses this instead. + """ + from lib_py.delegation import engine # same package: no sibling hop (see __init__) + request = "merge" if "status" in fields: - # A merged status value is really a transition request; routing it to the - # owning request is what subjects it to the CAS table instead of letting - # a raw field write slip past it. request = engine.STATUS_TO_REQUEST.get(fields["status"], "merge") if request != "merge": fields = {k: v for k, v in fields.items() if k != "status"} - result = engine.transition(job_id, request, registry_dir=registry_dir, - fields=fields, source="update_job_status") - return result.record + return engine.transition(job_id, request, registry_dir=registry_dir, + fields=fields, source=_source) def _update_job_status_legacy(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR, **fields: Any) -> Dict[str, Any]: @@ -549,10 +564,31 @@ def append_event(job_id: str, event_dict: Dict[str, Any], logs_dir: Optional[str def update_logged_status(job_id: str, status: str, logs_dir: Optional[str] = None, **extras: Any) -> None: """Rewrite ``//status.json`` (current status for fast point - queries) atomically. Best-effort; merges any ``extras``.""" + queries) atomically. Best-effort; merges any ``extras``. + + Ordering: this used to run under the registry lock, which serialised it for + free. Phase 2a moved the mirror outside the lock to keep the transaction + short, so a slow flush could otherwise land after a newer one and leave this + file behind the registry. When the caller supplies ``transition_seq`` the + write is skipped if the file already records a newer transition, which + restores the sequential consistency the lock used to provide. + """ try: path = job_log_path(job_id, STATUS_FILENAME, logs_dir) path.parent.mkdir(parents=True, exist_ok=True) + + incoming_seq = extras.get("transition_seq") + if incoming_seq is not None and path.exists(): + try: + existing = json.loads(path.read_text(encoding="utf-8")) + existing_seq = existing.get("transition_seq") + if existing_seq is not None and int(existing_seq) >= int(incoming_seq): + logger.debug("skipping stale status mirror for %s (seq %s <= %s)", + job_id, incoming_seq, existing_seq) + return + except Exception: + pass # unreadable existing file: fall through and rewrite it + record: Dict[str, Any] = {"job_id": job_id, "status": status, "updated_at": _utcnow()} record.update(extras) tmp = path.with_name(path.name + ".tmp") diff --git a/.agents/skills/lib_py/delegation/publish_event.py b/.agents/skills/lib_py/delegation/publish_event.py index 61c7997..512b635 100755 --- a/.agents/skills/lib_py/delegation/publish_event.py +++ b/.agents/skills/lib_py/delegation/publish_event.py @@ -176,7 +176,10 @@ def record_outcome(prepared: PreparedEvent, registry_dir: str, new_status = EVENT_TO_STATUS.get(prepared.event) if new_status: try: - mqtt_common.update_job_status(prepared.job_id, registry_dir, status=new_status) + # Agent authority on purpose: an agent may never move a terminal + # record, which is what the operator-only requests are gated for. + mqtt_common.update_job_status(prepared.job_id, registry_dir, + _source="publish_event", status=new_status) except Exception as exc: # pragma: no cover - best effort logger.warning("status sync failed: %s", exc) diff --git a/.agents/skills/lib_py/delegation/registry.py b/.agents/skills/lib_py/delegation/registry.py index ecb0577..cb69f8e 100755 --- a/.agents/skills/lib_py/delegation/registry.py +++ b/.agents/skills/lib_py/delegation/registry.py @@ -326,6 +326,26 @@ def _build_parser() -> argparse.ArgumentParser: return parser +def _operator_write(job_id: str, registry_dir: str, **fields) -> int: + """Apply an operator-authority write and report refusal on the exit status. + + Before phase 2a this path could not be refused, so it always exited 0. Once + the CAS boundary can decline a write, exiting 0 anyway would hide it -- which + is exactly how a refused loop re-dispatch looked like success to the wrapper. + """ + from lib_py.delegation import engine # same package: no sibling hop (see __init__) + + if not engine.cas_enabled(): + mqtt_common.update_job_status(job_id, registry_dir, **fields) + return 0 + result = mqtt_common.transition_job_status( + job_id, registry_dir, _source="registry_cli", **fields) + if not result.committed: + print("refused: {}".format(result.reason), file=sys.stderr) + return 4 + return 0 + + def main(argv: Optional[List[str]] = None) -> int: mqtt_common.ensure_env_loaded() mqtt_common.setup_logging(logging.INFO) @@ -374,11 +394,11 @@ def main(argv: Optional[List[str]] = None) -> int: if args.command == "status": try: - update_status(args.job, rd, args.status) + rc = _operator_write(args.job, rd, status=args.status) except (FileNotFoundError, ValueError) as exc: print(str(exc), file=sys.stderr) return 1 - return 0 + return rc if args.command == "update": fields = {} @@ -393,11 +413,11 @@ def main(argv: Optional[List[str]] = None) -> int: if args.role is not None: fields["role"] = args.role try: - mqtt_common.update_job_status(args.job, rd, **fields) + rc = _operator_write(args.job, rd, **fields) except FileNotFoundError as exc: print(str(exc), file=sys.stderr) return 1 - return 0 + return rc if args.command == "get-feedback": print(get_feedback(args.job, rd)) diff --git a/.agents/skills/lib_py/loop/policy.py b/.agents/skills/lib_py/loop/policy.py index 76639a1..62bbe98 100644 --- a/.agents/skills/lib_py/loop/policy.py +++ b/.agents/skills/lib_py/loop/policy.py @@ -25,6 +25,31 @@ def _timeline_by_kind(scenario): return out +def collect_observed_identity(scenario) -> dict: + """Gather identifiers the *evidence* carries, for comparison with dispatch. + + Without this, attribution only proves the dispatch record is internally + well-formed -- it never confronts what was actually seen. Association of a + job or attempt to an observation is the external collector's business, so + these are read from the timeline items and the report manifest, never + invented as native Herdr fields. + """ + observed = {} + for item in scenario.get("timeline") or []: + payload = item.get("payload") or {} + for field in ("job_id", "attempt_id", "agent_session_uuid", + "pane_process_generation"): + if payload.get(field) is not None: + # Every occurrence is kept. Taking only the first would let a + # later, contradicting observation pass unnoticed -- which is the + # opposite of what this collection is for. + observed.setdefault(field, []).append(payload[field]) + manifest = payload.get("report_manifest") + if isinstance(manifest, dict) and manifest.get("attempt_id") is not None: + observed.setdefault("manifest_attempt_id", []).append(manifest["attempt_id"]) + return observed + + def decide(scenario) -> Decision: """Map one composite scenario to a decision. @@ -62,11 +87,31 @@ def decide(scenario) -> Decision: if latest.get("agent_status") == "blocked" or latest.get("matched_rule"): return Decision("hold_blocked", "agent is blocked on an interactive prompt") - # 4. Attribution, including the execution anchor. + # 4. Attribution: the dispatch record must be well formed AND agree with the + # identifiers the evidence itself carries. Flags on the dispatch record + # alone are self-assertion, not correspondence. + observed = collect_observed_identity(scenario) + manifest_attempts = observed.pop("manifest_attempt_id", []) + attributed = attribution.check(dispatch) if not attributed.ok: return Decision("hold", attributed.reason) + for field, values in observed.items(): + for value in values: + if value != dispatch.get(field): + return Decision("hold", "attribution mismatch on {}: observed {!r}, " + "dispatched {!r}".format(field, value, + dispatch.get(field))) + + # The finalized report must belong to the attempt being judged; a manifest + # from another attempt is evidence of a different run finishing. + for manifest_attempt in manifest_attempts: + if manifest_attempt != dispatch.get("attempt_id"): + return Decision("hold", + "report manifest belongs to attempt {!r}, not {!r}".format( + manifest_attempt, dispatch.get("attempt_id"))) + # 5. Still working, or a tool is running underneath a ready prompt. if latest.get("agent_status") not in QUIESCENT: return Decision("hold", "agent is not quiescent (status={!r})".format( diff --git a/tests/test_issue5_phase2a_loop.py b/tests/test_issue5_phase2a_loop.py index 61d0bb1..7da8dca 100644 --- a/tests/test_issue5_phase2a_loop.py +++ b/tests/test_issue5_phase2a_loop.py @@ -695,3 +695,195 @@ def test_g9_phase2a_spawns_no_daemon_and_calls_no_frozen_helper(): src = path.read_text(encoding="utf-8") for forbidden in ("ensure_subscriber_daemon", "preexec_fn", "lib.sh"): assert forbidden not in src, "{} references {}".format(path.name, forbidden) + + +# ============================================================================== +# 8. Cross-review regressions (job 9d1e0077) +# +# F-A (claude): the CAS table omitted two legitimate wrapper transitions, so +# `submit --type loop|discuss` silently stopped working by default +# and the CLI still exited 0. +# F-B (claude): the audit status mirror moved outside the lock and became +# last-writer-wins, so a slow flush could leave it behind. +# R1 (codex): the policy trusted the dispatch record's own flags and never +# confronted the identifiers the evidence carried. +# R2 (codex): a supplied expected_attempt was ignored when the record had none. +# R3 (codex): a raw status field rode in on a metadata merge. +# ============================================================================== + +@pytest.mark.parametrize("mode", CAS_MODES) +def test_fa_wrapper_loop_redispatch_and_finalize_work(mam_sandbox, tmp_path, monkeypatch, mode): + """F-A: the exact CLI calls the wrapper makes for --type loop|discuss. + + Iteration reset hands a finished job to the next session; exhausting the + budget finalises it as an error. Both must work in either mode.""" + monkeypatch.setenv("MAM_ENGINE_CAS", mode) + monkeypatch.chdir(mam_sandbox) + from lib_py.delegation import mqtt_common, registry + + reg = _reg(tmp_path) + job_id = _make_job(reg, agent_session="herdr:claude", job_type="loop") + mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed") + + rc = registry.main(["--registry-dir", reg, "update", "--job", job_id, + "--agent-session", "herdr:rev", "--prompt", "review", + "--iteration", "2", "--role", "Reviewer", "--status", "pending"]) + record = _load(reg, job_id) + assert rc == 0, "the wrapper's iteration reset must succeed" + assert record["status"] == "pending" + assert record["iteration"] == 2, "metadata must not be discarded" + assert record["agent_session"] == "herdr:rev" + + mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed") + rc2 = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "error"]) + assert rc2 == 0 + assert _load(reg, job_id)["status"] == "error" + + +def test_fa_redispatch_requires_operator_authority_and_a_forward_attempt( + mam_sandbox, tmp_path, monkeypatch +): + """The new authority must not become a way for an agent to undo a terminal.""" + 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) + + agent = engine.transition(job_id, "redispatch", registry_dir=reg, + fields={"iteration": 2}, source="publish_event") + assert agent.committed is False and "operator authority" in agent.reason + + stale = engine.transition(job_id, "redispatch", registry_dir=reg, + fields={"iteration": 1}, source="registry_cli") + assert stale.committed is False and "does not advance" in stale.reason + + ok = engine.transition(job_id, "redispatch", registry_dir=reg, + fields={"iteration": 2}, source="registry_cli") + assert ok.committed and ok.to_status == "pending" + + +def test_fa_agent_still_cannot_move_a_terminal(mam_sandbox, tmp_path, monkeypatch): + """The protection the CAS table exists for must survive the new authority.""" + 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="completed") + mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="error") + assert _load(reg, job_id)["status"] == "completed" + + +def test_fa_a_refused_operator_write_exits_non_zero(mam_sandbox, tmp_path, monkeypatch): + """The silence is what hid F-A: a refused write used to exit 0.""" + monkeypatch.chdir(mam_sandbox) + from lib_py.delegation import engine, registry + 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) + # cancelled after completed is a genuine conflict, and must be reported + rc = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "cancelled"]) + assert rc != 0, "a refused operator write must not look like success" + assert _load(reg, job_id)["status"] == "completed" + + +def test_fb_a_late_status_mirror_cannot_overwrite_a_newer_one(mam_sandbox, tmp_path, monkeypatch): + """F-B: the mirror runs outside the lock, so ordering rides on transition_seq.""" + monkeypatch.chdir(mam_sandbox) + from lib_py.delegation import mqtt_common + reg = _reg(tmp_path) + job_id = _make_job(reg) + logs = mqtt_common.get_logs_dir() + + mqtt_common.update_logged_status(job_id, "completed", logs, transition_seq=2) + mqtt_common.update_logged_status(job_id, "running", logs, transition_seq=1) # late flush + assert mqtt_common.read_logged_status(job_id, logs)["status"] == "completed" + # a genuinely newer transition still lands + mqtt_common.update_logged_status(job_id, "error", logs, transition_seq=3) + assert mqtt_common.read_logged_status(job_id, logs)["status"] == "error" + + +@pytest.mark.parametrize("field,value", [ + ("agent_session_uuid", "someone-else"), + ("attempt_id", 999), + ("job_id", "another-job"), +]) +def test_r1_policy_rejects_evidence_that_contradicts_the_dispatch(field, value): + """R1: attribution must confront what was observed, not just the record.""" + from lib_py.loop import policy + import copy + scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8")) + assert policy.decide(scenario).decision == "candidate_approved" + mutated = copy.deepcopy(scenario) + mutated["timeline"][2]["payload"][field] = value + decision = policy.decide(mutated) + assert decision.decision == "hold", "{}={!r} must not be approved".format(field, value) + assert field in decision.reason + + +def test_r1_policy_rejects_a_manifest_from_another_attempt(): + """A finalized report proves *an* attempt finished, not necessarily this one.""" + from lib_py.loop import policy + import copy + scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8")) + mutated = copy.deepcopy(scenario) + mutated["timeline"][1]["payload"]["report_manifest"]["attempt_id"] = 999 + decision = policy.decide(mutated) + assert decision.decision == "hold" and "manifest" in decision.reason + + +def test_r2_expected_attempt_must_be_proved_not_merely_uncontradicted( + mam_sandbox, tmp_path, monkeypatch +): + """R2: a precondition the record cannot answer must fail closed.""" + monkeypatch.chdir(mam_sandbox) + from lib_py.delegation import engine, mqtt_common + reg = _reg(tmp_path) + job_id = _make_job(reg) + engine.transition(job_id, "claim", registry_dir=reg) + + record = _load(reg, job_id) + record.pop("iteration", None) + record.pop("attempt_id", None) + mqtt_common._atomic_write_record(job_id, reg, record) + before = _dir_fingerprint(Path(reg)) + + res = engine.transition(job_id, "completed", registry_dir=reg, expected_attempt=999) + assert res.committed is False and "carries none" in res.reason + assert _dir_fingerprint(Path(reg)) == before, "a rejection must not write" + + # with an attempt present the precondition behaves normally + engine.transition(job_id, "progress", registry_dir=reg, fields={"attempt_id": 7}) + assert engine.transition(job_id, "completed", registry_dir=reg, + expected_attempt=8).committed is False + assert engine.transition(job_id, "completed", registry_dir=reg, + expected_attempt=7).committed is True + + +@pytest.mark.parametrize("request_", ["progress", "permission_required", "merge"]) +def test_r3_a_status_cannot_ride_in_as_a_metadata_field( + mam_sandbox, tmp_path, monkeypatch, request_ +): + """R3: the boundary itself must refuse it, not just one wrapper.""" + 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) + res = engine.transition(job_id, request_, registry_dir=reg, + fields={"status": "completed"}) + assert res.committed is False and "must be requested" in res.reason + assert _load(reg, job_id)["status"] == "running" + + +def test_r3_an_unknown_request_is_refused(mam_sandbox, tmp_path, monkeypatch): + 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) + res = engine.transition(job_id, "not_a_request", registry_dir=reg, fields={"x": 1}) + assert res.committed is False and "unknown request" in res.reason + assert "x" not in _load(reg, job_id)