diff --git a/.agents/skills/lib_py/delegation/engine.py b/.agents/skills/lib_py/delegation/engine.py new file mode 100644 index 0000000..65c14f2 --- /dev/null +++ b/.agents/skills/lib_py/delegation/engine.py @@ -0,0 +1,228 @@ +"""The single state-transition authority. + +Every job-status write goes through `transition`. Before phase 2a there were two +write paths: `update_job_status` merged fields unconditionally, and +`pick_pending` hand-rolled a correct pending->running swap plus its own audit +mirror. Neither was wrong on its own; having two was the risk, because a fix or +a rule added to one silently misses the other. + +It lives in `lib_py.delegation`, not `lib_py.loop`, because it writes the +registry and the registry writers are here. A frozen loop can bind +`lib_py.delegation` to a different tree than the parent package, so delegation +importing a sibling would mix trees -- the rule this package states in its own +__init__, and which caught this during implementation. + +Lock discipline (the reason this is a module and not a helper): + + with registry_lock: decide, write the record once, QUEUE notifications + after the lock: flush notifications, publish, anything slow + +Nothing that can block on another process or the network may run while the lock +is held. `pick_pending` already mirrored its audit entries after releasing the +lock; this generalises that shape to every transition. + +`MAM_ENGINE_CAS=0` restores the previous unconditional-merge behaviour without a +redeploy. The legacy bodies are preserved by their callers for that reason. +""" +from __future__ import annotations + +import logging +import os +from typing import Any, Dict, List, Optional + +from lib_py.delegation import mqtt_common + +logger = logging.getLogger("loop.engine") + +TERMINAL_STATUSES = ("completed", "error", "cancelled") +NON_TERMINAL_STATUSES = ("pending", "running") + +# request -> the status it asks for. `merge`/`progress` ask for nothing. +REQUEST_TO_STATUS = { + "claim": "running", + "started": "running", + "progress": None, + "permission_required": None, + "completed": "completed", + "error": "error", + "cancelled": "cancelled", + "deadline_timeout": "error", + "merge": None, +} + +# 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. +STATUS_TO_REQUEST = { + "running": "started", + "completed": "completed", + "error": "error", + "cancelled": "cancelled", +} + + +def cas_enabled() -> bool: + """The rollback switch. Read per call so a test or operator can flip it.""" + return os.environ.get("MAM_ENGINE_CAS", "1") != "0" + + +class TransitionResult(object): + __slots__ = ("committed", "from_status", "to_status", "reason", "record", "notifications") + + def __init__(self, committed, from_status, to_status, reason, record, notifications): + self.committed = committed + self.from_status = from_status + self.to_status = to_status + self.reason = reason + self.record = record + self.notifications = notifications + + def __repr__(self): # pragma: no cover - debugging aid + return "TransitionResult(committed={!r}, {!r}->{!r}, {!r})".format( + self.committed, self.from_status, self.to_status, self.reason) + + +class _Decision(object): + __slots__ = ("commit", "to_status", "reason", "conflict") + + def __init__(self, commit, to_status, reason, conflict=False): + self.commit = commit + self.to_status = to_status + self.reason = reason + self.conflict = conflict + + +def _decide(record: Dict[str, Any], request: str, expected_status, expected_attempt) -> _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") + + # Stale attempt or generation loses before anything else is considered. + if expected_attempt is not None: + actual = record.get("attempt_id", record.get("iteration")) + if actual is not None and actual != expected_attempt: + return _Decision(False, current, + "stale attempt: expected {!r}, record has {!r}".format( + expected_attempt, actual)) + + if expected_status is not None and current != expected_status: + return _Decision(False, current, + "precondition failed: expected status {!r}, found {!r}".format( + expected_status, current)) + + requested = REQUEST_TO_STATUS.get(request, None) + + 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)) + if requested is None: + return _Decision(False, current, + "late {} after terminal {}; refusing".format(request, current), + conflict=True) + return _Decision(False, current, + "conflicting {} after terminal {}; refusing".format(requested, current), + 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. + return _Decision(True, None, "metadata updated ({})".format(request)) + + return _Decision(True, requested, "{} -> {}".format(current, requested)) + + +def _notification(job_id, kind, **fields) -> Dict[str, Any]: + payload = {"job_id": job_id, "kind": kind} + payload.update(fields) + return payload + + +def transition(job_id, request="merge", *, registry_dir=None, expected_status=None, + expected_attempt=None, fields=None, source="unknown") -> TransitionResult: + """Decide and commit one state transition. The only writer of job status. + + Returns a result whose `notifications` have ALREADY been flushed; the field + is kept so callers and tests can see what the transition emitted. + """ + registry_dir = registry_dir or mqtt_common.DEFAULT_REGISTRY_DIR + fields = dict(fields or {}) + + # ---- short transaction: decide and write, never do I/O that can block ---- + 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) + + notifications: List[Dict[str, Any]] = [] + if decision.commit: + record.update(fields) + if decision.to_status is not None: + record["status"] = decision.to_status + record["updated_at"] = mqtt_common._utcnow() + # A monotonic counter so a late flush can still be ordered correctly + # once the audit mirror happens outside the lock. + record["transition_seq"] = int(record.get("transition_seq", 0)) + 1 + mqtt_common._atomic_write_record(job_id, registry_dir, record) + + to_status = record.get("status") + if to_status != from_status: + notifications.append(_notification( + job_id, "status_changed", **{ + "from": from_status, "to": to_status, + "by": source, "timestamp": record["updated_at"], + "transition_seq": record["transition_seq"], + })) + else: + to_status = from_status + if decision.conflict: + notifications.append(_notification( + job_id, "conflict_refused", + attempted=request, current=from_status, + reason=decision.reason, by=source, + timestamp=mqtt_common._utcnow())) + # ---- lock released; everything below may block ---- + + _flush(notifications, registry_dir) + if not decision.commit and decision.conflict: + logger.warning("refused %s for job %s: %s", request, job_id, decision.reason) + + return TransitionResult(decision.commit, from_status, to_status, + decision.reason, record, notifications) + + +def _flush(notifications, registry_dir) -> None: + """Mirror queued notifications into the audit log, outside the lock. + + Best effort by design: the registry record is already durable, and an audit + write must never be able to fail a transition that has committed. + """ + for note in notifications: + job_id = note["job_id"] + kind = note["kind"] + try: + if kind == "status_changed": + mqtt_common.update_logged_status( + job_id, note["to"], updated_at=note.get("timestamp")) + mqtt_common.append_event(job_id, { + "event": "status_changed", + "from": note.get("from"), + "to": note.get("to"), + "by": note.get("by"), + "timestamp": note.get("timestamp"), + "transition_seq": note.get("transition_seq"), + }) + elif kind == "conflict_refused": + mqtt_common.append_event(job_id, { + "event": "conflict_refused", + "attempted": note.get("attempted"), + "current": note.get("current"), + "reason": note.get("reason"), + "by": note.get("by"), + "timestamp": note.get("timestamp"), + }) + except Exception as exc: # pragma: no cover - best effort + logger.warning("audit mirror failed for %s: %s", job_id, exc) diff --git a/.agents/skills/lib_py/delegation/mqtt_common.py b/.agents/skills/lib_py/delegation/mqtt_common.py index 4843870..901d946 100644 --- a/.agents/skills/lib_py/delegation/mqtt_common.py +++ b/.agents/skills/lib_py/delegation/mqtt_common.py @@ -393,7 +393,38 @@ def _atomic_write_record(job_id: str, registry_dir: str, record: Dict[str, Any]) def update_job_status(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR, **fields: Any) -> Dict[str, Any]: - """Atomically merge ``fields`` into a job record under the registry lock. + """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 + assert through this function, and the shell helpers capture its effects. + What changed is who performs the write -- ``lib_py.loop.engine`` is now the + only writer of job status, so the CAS rules and the audit mirror exist once + instead of twice. + + ``MAM_ENGINE_CAS=0`` restores the previous unconditional merge below. + """ + from lib_py.delegation import engine # same package: no sibling hop (see __init__) + + if not engine.cas_enabled(): + return _update_job_status_legacy(job_id, registry_dir, **fields) + + 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 + + +def _update_job_status_legacy(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR, **fields: Any) -> Dict[str, Any]: + """Pre-phase-2a body, preserved for MAM_ENGINE_CAS=0 rollback. + + Atomically merge ``fields`` into a job record under the registry lock. Always refreshes ``updated_at``. Returns the new record. Raises FileNotFoundError if the job does not exist. diff --git a/.agents/skills/lib_py/delegation/registry.py b/.agents/skills/lib_py/delegation/registry.py index 0afa407..ecb0577 100755 --- a/.agents/skills/lib_py/delegation/registry.py +++ b/.agents/skills/lib_py/delegation/registry.py @@ -117,6 +117,37 @@ def pick_pending(agent_session: str, registry_dir: str = DEFAULT_REGISTRY_DIR) - ``running`` atomically under the lock. Returns the job id, or None if no pending job matches. This is how each herdr session takes only its own work without two sessions grabbing the same job.""" + from lib_py.delegation import engine # same package: no sibling hop (see __init__) + + if not engine.cas_enabled(): + return _pick_pending_legacy(agent_session, registry_dir) + + # Selection still happens under the lock, then the claim itself is committed + # by the single transition boundary with an explicit precondition, so two + # sessions still cannot both win. + with registry_lock(registry_dir): + candidates = [ + record for record in _iter_records(registry_dir) + if record.get("status") == "pending" and record.get("agent_session") == agent_session + ] + if not candidates: + return None + candidates.sort(key=lambda r: r.get("created_at", "")) + chosen_id = candidates[0]["job_id"] + + result = engine.transition(chosen_id, "claim", registry_dir=registry_dir, + expected_status="pending", source=agent_session) + if not result.committed: + # Someone else claimed it between selection and commit; the precondition + # refused, which is exactly the behaviour the lock used to provide. + logger.info("job %s was claimed by another session first", chosen_id) + return None + logger.info("session %s picked job %s", agent_session, chosen_id) + return chosen_id + + +def _pick_pending_legacy(agent_session: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> Optional[str]: + """Pre-phase-2a body, preserved for MAM_ENGINE_CAS=0 rollback.""" with registry_lock(registry_dir): candidates = [] for record in _iter_records(registry_dir): @@ -132,8 +163,8 @@ def pick_pending(agent_session: str, registry_dir: str = DEFAULT_REGISTRY_DIR) - logger.info("session %s picked job %s", agent_session, chosen["job_id"]) job_id = chosen["job_id"] updated_at = chosen["updated_at"] - # pick_pending writes the record directly (not via update_job_status), so it - # mirrors the pending->running transition into the audit log here. Best-effort. + # Mirrors the pending->running transition into the audit log outside the + # lock. This is the shape the engine now generalises to every transition. mqtt_common.update_logged_status(job_id, "running", updated_at=updated_at) mqtt_common.append_event(job_id, { "event": "status_changed", diff --git a/.agents/skills/lib_py/loop/__init__.py b/.agents/skills/lib_py/loop/__init__.py new file mode 100644 index 0000000..42cec98 --- /dev/null +++ b/.agents/skills/lib_py/loop/__init__.py @@ -0,0 +1,16 @@ +"""Loop core: the single state-transition authority and the shadow watcher. + +Issue #5 phase 2a. Two things live here and they are deliberately separated: + + * `lib_py.delegation.engine` decides and commits state. It is the only writer + of job status, and it lives with the other registry writers rather than here + -- see that module's docstring for why the package boundary matters under a + frozen loop. + * `watcher` observes and never decides; `policy` decides and never writes; + `shadow` runs the two together and, in this phase, only logs what it *would* + have committed. + +Package rule, enforced by test: nothing here imports a `lib_py` module other +than `lib_py.delegation`. Under a frozen loop the parent package may resolve to +a different tree than a bound sibling, so a stray import silently mixes trees. +""" diff --git a/.agents/skills/lib_py/loop/attribution.py b/.agents/skills/lib_py/loop/attribution.py new file mode 100644 index 0000000..f064cec --- /dev/null +++ b/.agents/skills/lib_py/loop/attribution.py @@ -0,0 +1,59 @@ +"""Four-fold execution attribution, as a pure predicate. + +A terminal state may only be inferred for the attempt that actually ran. Idle is +not evidence of finishing: a reused session sitting idle from previous work looks +identical unless job, attempt, session and pane generation all agree AND the +attempt was observed to actually execute after dispatch was acknowledged. +""" +from __future__ import annotations + +REQUIRED_FIELDS = ("job_id", "attempt_id", "agent_session_uuid", "pane_process_generation") + + +class AttributionResult(object): + __slots__ = ("ok", "reason") + + def __init__(self, ok, reason): + self.ok = ok + self.reason = reason + + +def check(dispatch_context, observed=None) -> AttributionResult: + """Compare a dispatch record against what was observed. + + `observed` may carry any subset of the four identifiers; every key it does + supply must match. A missing identifier is never treated as agreement. + """ + if not isinstance(dispatch_context, dict): + return AttributionResult(False, "dispatch_context missing") + + for field in REQUIRED_FIELDS: + if field not in dispatch_context: + return AttributionResult(False, "missing attribution field: {}".format(field)) + value = dispatch_context[field] + if value is None or value == "": + return AttributionResult(False, "empty attribution field: {}".format(field)) + + if not dispatch_context.get("dispatch_ack"): + return AttributionResult(False, "dispatch was never acknowledged") + + # The anchor is the whole point: without it, idle is just idle. + if not dispatch_context.get("execution_anchor_observed"): + return AttributionResult(False, "no execution anchor observed for this attempt") + + for field, value in (observed or {}).items(): + if field in REQUIRED_FIELDS and value != dispatch_context[field]: + return AttributionResult( + False, "attribution mismatch on {}".format(field)) + + return AttributionResult(True, "four-fold attribution satisfied") + + +def pane_matches_generation(pane_id, pane_process_generation) -> bool: + """`pane_process_generation` is '::'. + + Compared on the whole leading segment so w2X:p1 never matches w2X:p10. + """ + if not pane_id or not pane_process_generation: + return False + return str(pane_process_generation).startswith(str(pane_id) + ":") diff --git a/.agents/skills/lib_py/loop/evidence.py b/.agents/skills/lib_py/loop/evidence.py new file mode 100644 index 0000000..ed678a8 --- /dev/null +++ b/.agents/skills/lib_py/loop/evidence.py @@ -0,0 +1,56 @@ +"""The one data shape that crosses the watcher -> engine boundary. + +Keeping this a plain value with no I/O is what lets `policy.decide` stay a pure +function, which in turn is what lets the phase-0 scenario fixtures be fed to it +verbatim as test vectors. +""" +from __future__ import annotations + +SOURCE_KINDS = ("herdr", "event", "filesystem", "harness") + +# Decisions the policy may return. Same vocabulary as the phase-0 contracts, so +# a scenario's expected_policy.decision can be compared directly. +DECISIONS = ("hold", "hold_blocked", "hold_reconnect", "reject_stale", "candidate_approved") + + +class Observation(object): + """One point-in-time fact from one source. Never a conclusion.""" + + __slots__ = ("source_kind", "payload", "captured_at", "capture_ref", "provenance") + + def __init__(self, source_kind, payload, captured_at=None, capture_ref=None, + provenance="observed"): + if source_kind not in SOURCE_KINDS: + raise ValueError("unsupported source_kind: {!r}".format(source_kind)) + self.source_kind = source_kind + self.payload = dict(payload or {}) + self.captured_at = captured_at + self.capture_ref = capture_ref + self.provenance = provenance + + def __repr__(self): # pragma: no cover - debugging aid + return "Observation({!r}, {!r})".format(self.source_kind, self.payload) + + +class Decision(object): + """What the policy concluded, and why. `reason` is for the audit trail.""" + + __slots__ = ("decision", "reason", "attribution_ok") + + def __init__(self, decision, reason, attribution_ok=False): + if decision not in DECISIONS: + raise ValueError("unknown decision: {!r}".format(decision)) + self.decision = decision + self.reason = reason + self.attribution_ok = attribution_ok + + @property + def would_complete(self) -> bool: + return self.decision == "candidate_approved" + + def as_dict(self) -> dict: + return {"decision": self.decision, "reason": self.reason, + "attribution_ok": self.attribution_ok} + + def __repr__(self): # pragma: no cover - debugging aid + return "Decision({!r}, {!r})".format(self.decision, self.reason) diff --git a/.agents/skills/lib_py/loop/watcher.py b/.agents/skills/lib_py/loop/watcher.py new file mode 100644 index 0000000..88cad2f --- /dev/null +++ b/.agents/skills/lib_py/loop/watcher.py @@ -0,0 +1,166 @@ +"""Wraps `herdr agent wait`. Observes; never concludes. + +Live contract (herdr 0.8.2, re-confirmed while planning): + + herdr agent wait --until ... --timeout + + * `--timeout` is MILLISECONDS. The parameter here is named `timeout_ms` so the + unit cannot be misread at a call site; there is deliberately no seconds form. + * `--until` is always passed explicitly. Omitting it matches idle/done/blocked, + silently excluding `working`, which is a state the watcher cares about. + * success -> exit 0, snapshot JSON on STDOUT. + * timeout and unknown target -> exit 1, error JSON on STDERR, stdout empty. + The stream split is the contract: an error is never read from stdout. + +A single wait is one synchronous barrier returning one final snapshot, so +continuous observation is built by repeating it, not by streaming. +""" +from __future__ import annotations + +import json +import logging +import subprocess +from typing import Any, Dict, Optional, Sequence + +from lib_py.loop.evidence import Observation + +logger = logging.getLogger("loop.watcher") + +DEFAULT_UNTIL = ("idle", "done", "blocked") +VALID_STATES = ("idle", "working", "blocked", "done", "unknown") + +# Outcome kinds. `timeout` is an ordinary no-event result, not a failure. +MATCHED = "matched" +TIMEOUT = "timeout" +AGENT_NOT_FOUND = "agent_not_found" +TRANSPORT_ERROR = "transport_error" + +BACKOFF_START_S = 1.0 +BACKOFF_MAX_S = 16.0 + + +class WaitOutcome(object): + __slots__ = ("kind", "agent", "error_code", "detail", "raw") + + def __init__(self, kind, agent=None, error_code=None, detail="", raw=None): + self.kind = kind + self.agent = agent + self.error_code = error_code + self.detail = detail + self.raw = raw + + @property + def healthy(self) -> bool: + """A timeout is healthy: the socket answered, the state just did not change.""" + return self.kind in (MATCHED, TIMEOUT, AGENT_NOT_FOUND) + + def __repr__(self): # pragma: no cover - debugging aid + return "WaitOutcome({!r}, error_code={!r})".format(self.kind, self.error_code) + + +def build_argv(target: str, until: Sequence[str] = DEFAULT_UNTIL, + timeout_ms: Optional[int] = None) -> list: + if not target: + raise ValueError("wait target is required") + argv = ["herdr", "agent", "wait", target] + for state in until: + if state not in VALID_STATES: + raise ValueError("unsupported --until state: {!r}".format(state)) + argv += ["--until", state] + if timeout_ms is not None: + if int(timeout_ms) <= 0: + raise ValueError("timeout_ms must be positive milliseconds") + argv += ["--timeout", str(int(timeout_ms))] + return argv + + +def _parse(stream: str) -> Optional[Dict[str, Any]]: + try: + payload = json.loads(stream) + return payload if isinstance(payload, dict) else None + except Exception: + return None + + +def wait_once(target: str, until: Sequence[str] = DEFAULT_UNTIL, + timeout_ms: Optional[int] = None, run=subprocess.run) -> WaitOutcome: + """One barrier call. `run` is injectable so tests never touch a real herdr.""" + argv = build_argv(target, until, timeout_ms) + try: + proc = run(argv, capture_output=True, text=True) + except Exception as exc: + return WaitOutcome(TRANSPORT_ERROR, detail="herdr could not be executed: {}".format(exc)) + + if proc.returncode == 0: + payload = _parse(proc.stdout or "") + agent = ((payload or {}).get("result") or {}).get("agent") + if not isinstance(agent, dict): + return WaitOutcome(TRANSPORT_ERROR, + detail="exit 0 without a parseable agent snapshot", + raw=proc.stdout) + return WaitOutcome(MATCHED, agent=agent, raw=payload) + + # Non-zero: the payload is on stderr by contract, and stdout is empty. + payload = _parse(proc.stderr or "") + code = ((payload or {}).get("error") or {}).get("code") + if code == "timeout": + return WaitOutcome(TIMEOUT, error_code=code, raw=payload) + if code == "agent_not_found": + return WaitOutcome(AGENT_NOT_FOUND, error_code=code, raw=payload) + return WaitOutcome(TRANSPORT_ERROR, error_code=code, + detail="unrecognised failure (rc={})".format(proc.returncode), + raw=proc.stderr) + + +def observation_from(outcome: WaitOutcome, capture_ref=None) -> Optional[Observation]: + """Turn a matched snapshot into evidence. Non-matches carry no agent state.""" + if outcome.kind != MATCHED or not outcome.agent: + return None + agent = outcome.agent + return Observation( + "herdr", + { + "agent_status": agent.get("agent_status"), + "interactive_ready": agent.get("interactive_ready"), + "pane_id": agent.get("pane_id"), + "state_change_seq": agent.get("state_change_seq"), + }, + capture_ref=capture_ref, + ) + + +class ReconnectBackoff(object): + """Doubling backoff for transport failures, reset by any healthy answer. + + The snapshot cache is dropped on failure on purpose: after a socket gap the + watcher must re-observe rather than reason from a state that may have moved + while it was blind. That is what makes `hold_reconnect` the honest decision. + """ + + def __init__(self, start=BACKOFF_START_S, maximum=BACKOFF_MAX_S): + self.start = start + self.maximum = maximum + self.current = 0.0 + self.consecutive_failures = 0 + self.cached_snapshot = None + + def record_success(self, snapshot=None) -> None: + self.consecutive_failures = 0 + self.current = 0.0 + if snapshot is not None: + self.cached_snapshot = snapshot + + def record_failure(self) -> float: + self.consecutive_failures += 1 + self.cached_snapshot = None # stale evidence must not survive a gap + self.current = self.start if self.current == 0.0 else min(self.current * 2.0, self.maximum) + return self.current + + def sequence(self, n: int) -> list: + """The first n delays, for documentation and for tests to pin.""" + saved = (self.current, self.consecutive_failures, self.cached_snapshot) + try: + self.current, self.consecutive_failures = 0.0, 0 + return [self.record_failure() for _ in range(n)] + finally: + self.current, self.consecutive_failures, self.cached_snapshot = saved diff --git a/tests/test_issue5_phase2a_loop.py b/tests/test_issue5_phase2a_loop.py index 5efdbed..c0d787b 100644 --- a/tests/test_issue5_phase2a_loop.py +++ b/tests/test_issue5_phase2a_loop.py @@ -152,3 +152,207 @@ def test_x04_next_seq_and_register_are_unaffected(mam_sandbox, tmp_path, monkeyp assert mqtt_common.next_seq(job_id, reg) == 2 assert _load(reg, job_id)["last_seq"] == 2 assert _load(reg, job_id)["status"] == "pending", "next_seq must not touch status" + + +# ============================================================================== +# 2. engine — the CAS table (state-transitions.md §3) and lock discipline +# ============================================================================== + +def test_engine_pending_to_running_on_claim(mam_sandbox, tmp_path, monkeypatch): + monkeypatch.chdir(mam_sandbox) + from lib_py.delegation import engine + reg = _reg(tmp_path) + job_id = _make_job(reg) + res = engine.transition(job_id, "claim", registry_dir=reg, expected_status="pending") + assert res.committed and res.from_status == "pending" and res.to_status == "running" + + +@pytest.mark.parametrize("request_", ["progress", "permission_required"]) +def test_engine_non_terminal_requests_keep_running(mam_sandbox, tmp_path, monkeypatch, request_): + 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={"blocked_reason": "tool"}) + assert res.committed + assert _load(reg, job_id)["status"] == "running", "must stay non-terminal" + assert _load(reg, job_id)["blocked_reason"] == "tool" + + +@pytest.mark.parametrize("terminal", ["completed", "error", "cancelled"]) +def test_engine_explicit_terminal_commits(mam_sandbox, tmp_path, monkeypatch, 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) + res = engine.transition(job_id, terminal, registry_dir=reg) + assert res.committed and res.to_status == terminal + + +def test_engine_repeated_terminal_is_an_idempotent_noop(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) + engine.transition(job_id, "completed", registry_dir=reg) + before = _load(reg, job_id) + res = engine.transition(job_id, "completed", registry_dir=reg) + assert res.committed is False + assert "idempotent" in res.reason + assert _load(reg, job_id)["transition_seq"] == before["transition_seq"], "no second write" + + +@pytest.mark.parametrize("late", ["error", "cancelled", "progress"]) +def test_engine_refuses_to_reverse_a_terminal_and_audits_the_conflict( + mam_sandbox, tmp_path, monkeypatch, late +): + """A late or conflicting signal must never walk a terminal record backwards.""" + monkeypatch.chdir(mam_sandbox) + from lib_py.delegation import mqtt_common + 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) + + res = engine.transition(job_id, late, registry_dir=reg, source="late-arrival") + assert res.committed is False + assert _load(reg, job_id)["status"] == "completed", "terminal must be preserved" + events = list(mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir())) + refusals = [e for e in events if e.get("event") == "conflict_refused"] + assert refusals, "a refused conflict must be auditable" + assert refusals[-1]["current"] == "completed" + + +def test_engine_rejects_a_stale_attempt(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, fields={"attempt_id": 2}) + res = engine.transition(job_id, "completed", registry_dir=reg, expected_attempt=1) + assert res.committed is False + assert "stale attempt" in res.reason + assert _load(reg, job_id)["status"] == "running" + + +def test_engine_deadline_timeout_becomes_error(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, "deadline_timeout", registry_dir=reg) + assert res.committed and res.to_status == "error" + + +def test_engine_precondition_failure_does_not_write(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) # now running + res = engine.transition(job_id, "claim", registry_dir=reg, expected_status="pending") + assert res.committed is False and "precondition failed" in res.reason + + +def test_engine_transition_seq_is_monotonic(mam_sandbox, tmp_path, monkeypatch): + """The audit mirror runs outside the lock, so ordering rides on this counter.""" + monkeypatch.chdir(mam_sandbox) + from lib_py.delegation import engine + reg = _reg(tmp_path) + job_id = _make_job(reg) + seqs = [] + for req in ("claim", "progress", "progress", "completed"): + engine.transition(job_id, req, registry_dir=reg) + seqs.append(_load(reg, job_id)["transition_seq"]) + assert seqs == sorted(seqs) and len(set(seqs)) == len(seqs), seqs + + +def test_g5_no_blocking_io_while_the_registry_lock_is_held(mam_sandbox, tmp_path, monkeypatch): + """G5: the lock may cover a decision and one record write, nothing else. + + Watches the calls that can block on another process or the network and + asserts none of them happen between acquiring and releasing the lock.""" + monkeypatch.chdir(mam_sandbox) + import socket + from lib_py.delegation import mqtt_common + from lib_py.delegation import engine + + reg = _reg(tmp_path) + job_id = _make_job(reg) + + held = {"depth": 0} + violations = [] + real_lock = mqtt_common.registry_lock + + import contextlib + + @contextlib.contextmanager + def watched_lock(registry_dir): + held["depth"] += 1 + try: + with real_lock(registry_dir): + yield + finally: + held["depth"] -= 1 + + def guard(name): + def spy(*args, **kwargs): + if held["depth"] > 0: + violations.append(name) + raise AssertionError("{} must not run under the registry lock".format(name)) + return spy + + monkeypatch.setattr(mqtt_common, "registry_lock", watched_lock) + monkeypatch.setattr(engine.mqtt_common, "registry_lock", watched_lock, raising=False) + monkeypatch.setattr(subprocess, "run", guard("subprocess.run")) + monkeypatch.setattr(subprocess, "Popen", guard("subprocess.Popen")) + monkeypatch.setattr(socket, "socket", guard("socket.socket")) + real_sleep = time.sleep + + def sleep_spy(seconds): + if held["depth"] > 0: + violations.append("time.sleep") + return real_sleep(0) + + monkeypatch.setattr(time, "sleep", sleep_spy) + + for req in ("claim", "progress", "completed", "completed", "error"): + engine.transition(job_id, req, registry_dir=reg) + + assert not violations, "blocking calls under the lock: {}".format(violations) + + +def test_g6_every_record_writer_is_the_engine_or_a_declared_exception(): + """G6: no new bypass of the transition boundary can be added unnoticed.""" + import ast + + allowed = { + # creation, and the sequence counter, which never touches status + ("registry.py", "register_job"), + ("mqtt_common.py", "next_seq"), + ("mqtt_common.py", "_update_job_status_legacy"), + ("registry.py", "_pick_pending_legacy"), + ("engine.py", "transition"), + } + found = set() + for path in list((SKILLS / "lib_py" / "delegation").glob("*.py")) + \ + list((SKILLS / "lib_py" / "loop").glob("*.py")): + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + if not isinstance(node, ast.FunctionDef): + continue + for inner in ast.walk(node): + if isinstance(inner, ast.Call): + fn = inner.func + name = getattr(fn, "attr", getattr(fn, "id", None)) + if name == "_atomic_write_record": + found.add((path.name, node.name)) + assert found <= allowed, ( + "job records are written outside the transition boundary: {}".format(found - allowed) + )