diff --git a/.agents/skills/lib_py/delegation/job_subscriber.py b/.agents/skills/lib_py/delegation/job_subscriber.py index 838b646..e1dee17 100755 --- a/.agents/skills/lib_py/delegation/job_subscriber.py +++ b/.agents/skills/lib_py/delegation/job_subscriber.py @@ -84,11 +84,13 @@ def _check_disk_fallback( disk_status = None rec_token = None rec_attempt = None + rec_seq = None try: job_rec = load_job(jid, registry_dir) disk_status = job_rec.get("status") rec_token = job_rec.get("dispatch_token") rec_attempt = job_rec.get("attempt_id", job_rec.get("iteration")) + rec_seq = job_rec.get("transition_seq") except Exception: job_rec = {} if not disk_status or disk_status not in ("completed", "error", "cancelled"): @@ -99,7 +101,20 @@ def _check_disk_fallback( except Exception: logged = None if logged: - disk_status = logged.get("status") + # The audit mirror is written after the registry commits, so a + # record that is not terminal may sit next to the previous + # attempt's terminal status. A versioned record (the engine's + # transition_seq) takes the audit status only when the audit is + # at least as new; an audit status without a sequence cannot + # prove which attempt it describes, so a versioned record + # ignores it. An unversioned record (never transitioned, or + # MAM_ENGINE_CAS=0) and a missing record keep the old rule. + aud_seq = logged.get("transition_seq") + if rec_seq is not None and (aud_seq is None or int(aud_seq) < int(rec_seq)): + logger.warning("disk fallback ignored audit status %r for %s: transition_seq %r is older " + "than the record's %r", logged.get("status"), jid, aud_seq, rec_seq) + else: + disk_status = logged.get("status") if disk_status in ("completed", "error", "cancelled"): if expected_dispatch_tokens: diff --git a/tests/test_issue57_logs_follow_registry.py b/tests/test_issue57_logs_follow_registry.py index 6becb04..8db9aa7 100644 --- a/tests/test_issue57_logs_follow_registry.py +++ b/tests/test_issue57_logs_follow_registry.py @@ -35,7 +35,7 @@ if str(SKILLS) not in sys.path: sys.path.insert(0, str(SKILLS)) from lib_py import control_plane # noqa: E402 -from lib_py.delegation import job_subscriber, mqtt_common, registry # noqa: E402 +from lib_py.delegation import engine, job_subscriber, mqtt_common, registry # noqa: E402 from test_issue21_phase2_loop_a import _mutate, _prepare, _rows, _skill, _wt # noqa: E402 from test_issue21_phase2_loop_a import ctl # noqa: E402,F401 @@ -310,3 +310,67 @@ d['herdr_sessions'] = [{{'name': 's57-creator-claude', 'status': 'running', 'del (line,) = [l for l in res.stdout.splitlines() if l.startswith("s57-creator-claude")] assert jid in line and "completed" in line, line assert not (tmp / ".mam" / "delegate_job_logs").exists() + + +# --------------------------------------------------------------------------- +# review (codex P2): conflict refusals and legacy claims follow the registry too +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("path", ["conflict", "legacy-claim"]) +def test_conflict_refusals_and_legacy_claims_follow_the_registry(moved, monkeypatch, path): + ws, reg, logs = moved + jid = registry.register_job("p2", registry_dir=str(reg), agent_session="herdr:t57") + if path == "conflict": + assert engine.transition(jid, "completed", registry_dir=str(reg), source="operator").committed + assert not engine.transition(jid, "error", registry_dir=str(reg), source="late").committed + expected = "conflict_refused" + else: + monkeypatch.setenv("MAM_ENGINE_CAS", "0") + assert registry.pick_pending("herdr:t57", registry_dir=str(reg)) == jid + expected = "status_changed" + assert expected in [e.get("event") for e in mqtt_common.iter_logged_events(jid, str(logs))] + + +# --------------------------------------------------------------------------- +# review (codex P1): an old audit terminal must not close a redispatched attempt +# --------------------------------------------------------------------------- + +def test_old_audit_terminal_does_not_complete_the_redispatched_attempt(moved, monkeypatch): + ws, reg, logs = moved + jid = registry.register_job("p1", registry_dir=str(reg), agent_session="herdr:t57") + assert engine.transition(jid, "completed", registry_dir=str(reg), source="operator").committed + old_audit = mqtt_common.read_logged_status(jid, str(logs)) + flush, observed = engine._flush, {} + + def before_the_mirror(notes, registry_dir): + # the real gap: attempt 2 is committed, the audit still says attempt 1 completed + rec = mqtt_common.load_job(jid, registry_dir) + assert rec["status"] == "pending" and rec["attempt_id"] == 2 + assert old_audit["transition_seq"] < rec["transition_seq"] + pending, terminal = {jid}, {} + job_subscriber._check_disk_fallback(pending, registry_dir, terminal, {jid: rec["dispatch_token"]}, {jid: 2}) + observed.update(pending=pending, terminal=terminal) + flush(notes, registry_dir) + + monkeypatch.setattr(engine, "_flush", before_the_mirror) + assert engine.transition(jid, "redispatch", registry_dir=str(reg), source="operator", + fields={"attempt_id": 2}).committed + assert observed == {"pending": {jid}, "terminal": {}}, "the old completion was attributed to attempt 2" + rec = mqtt_common.load_job(jid, str(reg)) + # a failed mirror leaves the old status.json behind (update_logged_status + # itself refuses to go backwards, so the file is written directly): still ignored + mqtt_common.job_log_path(jid, mqtt_common.STATUS_FILENAME, str(logs)).write_text(json.dumps(old_audit)) + assert mqtt_common.read_logged_status(jid, str(logs))["status"] == "completed" + pending, terminal = {jid}, {} + job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: rec["dispatch_token"]}, {jid: 2}) + assert pending == {jid} and terminal == {} + # an audit status without a sequence cannot prove it is about this attempt + mqtt_common.update_logged_status(jid, "completed", logs_dir=str(logs)) + pending, terminal = {jid}, {} + job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: rec["dispatch_token"]}, {jid: 2}) + assert pending == {jid} and terminal == {} + # a newer audit-only terminal is still taken + mqtt_common.update_logged_status(jid, "completed", logs_dir=str(logs), transition_seq=rec["transition_seq"] + 1) + pending, terminal = {jid}, {} + job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: rec["dispatch_token"]}, {jid: 2}) + assert pending == set() and terminal == {jid: "completed"}