GodopuandClaude Opus 5
2532e78453
engine.transition is now the only writer of job status. It implements the target table in state-transitions.md §3: claim, non-terminal metadata, explicit terminals, idempotent re-receipt, refusal of conflicting or late signals with an audit entry, stale-attempt rejection, and deadline timeout. Two write paths become one. update_job_status merged fields unconditionally; pick_pending hand-rolled a correct pending->running swap plus its own audit mirror. Neither was wrong alone, but a rule added to one silently missed the other, which is the shape that produced the phase 1 transport bugs. Both keep their signatures and their preserved legacy bodies behind MAM_ENGINE_CAS=0. A merged `status` value is routed to the request that owns it, so a raw field write can no longer slip past the table. Getting that wrong first was caught by the characterization tests: a metadata merge was overwriting the caller's status with the pre-merge value. Lock discipline is now enforced, not described. The lock covers a decision and one record write; audit mirroring is queued and flushed after release, carrying a monotonic transition_seq so ordering survives the move. A test watches for subprocess, socket and sleep calls while the lock is held and asserts none. Structural correction to my own plan, caught by a phase 1 guard: the plan put the engine in lib_py/loop, but delegation would then have had to import a sibling, which is exactly what mixes trees under a frozen loop. The engine belongs with the registry writers, so it lives in lib_py/delegation and loop consumes it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd
57 lines
2.2 KiB
Python
57 lines
2.2 KiB
Python
"""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)
|