GodopuandClaude Opus 5
803ba43884
The gate does not re-decide anything. Phase 2a already decides, fail-closed and against contract-pinned vectors; this supplies real evidence to that decision and commits only if three independent gates agree: the global switch is on, the backend has observed evidence that it reaches a quiescent state, and the job itself opted in. Default is off. The backend allowlist is evidence-based rather than optimistic. Inferring "finished" from stillness requires stillness to have been observed, and only three of six backends have that. An operator may narrow the list; a name that was never observed is dropped rather than trusted, so it cannot be widened by configuration. Everything else holds. An unreadable manifest, a report rewritten after finalization, missing evidence for a declared side effect, a policy that raises: all refuse. "Could not decide" is never "done". A late explicit terminal wins, because the boundary refuses the inferred one -- that is the boundary working, not an error in the gate. Notification intent is committed in the same atomic write as the state, so a crash between committing and publishing leaves a record that still says the world has not been told. Recovery re-sends; duplicates are already dropped by sequence. A failed send deliberately leaves the intent pending, because an unsent notification that looks sent is worse than one that is retried. Two fixes this surfaced. Settling used a request that terminal records refuse, which made an owed notification impossible to ever settle; settling is now a status-free bookkeeping request, valid in any state. And the phase 2a guard against shell references was matching hashlib.sha256 as a substring, so it now requires a word boundary. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd
317 lines
14 KiB
Python
317 lines
14 KiB
Python
"""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,
|
|
# 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",
|
|
# Bookkeeping: records that an owed notification was delivered. It cannot
|
|
# change status, and it is *about* a terminal, so refusing it in a terminal
|
|
# state would make an owed notification impossible to ever settle.
|
|
"notification_settled": None,
|
|
}
|
|
|
|
# Requests that carry no status at all and are therefore safe in any state,
|
|
# including a terminal one. Keep this list short and obviously status-free.
|
|
BOOKKEEPING_REQUESTS = ("notification_settled",)
|
|
|
|
# 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=<value>` is making the request that owns that
|
|
# value; mapping it here is what stops a raw field write from sidestepping the
|
|
# 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",
|
|
"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", "idempotent")
|
|
|
|
def __init__(self, committed, from_status, to_status, reason, record, notifications,
|
|
idempotent=False):
|
|
self.committed = committed
|
|
# Not every uncommitted transition is a refusal. Re-receiving a terminal
|
|
# that already holds is a no-op ACK in the contract table, and a caller
|
|
# that treats it as failure turns a success path into an error.
|
|
self.idempotent = idempotent
|
|
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", "idempotent")
|
|
|
|
def __init__(self, commit, to_status, reason, conflict=False, idempotent=False):
|
|
self.commit = commit
|
|
self.to_status = to_status
|
|
self.reason = reason
|
|
self.conflict = conflict
|
|
self.idempotent = idempotent
|
|
|
|
|
|
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 {}
|
|
|
|
# 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 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))
|
|
|
|
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)
|
|
|
|
# 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 request in BOOKKEEPING_REQUESTS:
|
|
return _Decision(True, None, "bookkeeping ({})".format(request))
|
|
|
|
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),
|
|
idempotent=True)
|
|
|
|
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),
|
|
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. 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))
|
|
|
|
|
|
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",
|
|
notify=None) -> 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,
|
|
source=source, fields=fields)
|
|
|
|
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) or 0) + 1
|
|
if notify is not None:
|
|
# One atomic commit records both the new state and the fact that
|
|
# the world has not been told yet. If the process dies before the
|
|
# publish, the record still says a notification is owed.
|
|
from lib_py.delegation import notifier
|
|
record[notifier.FIELD] = notifier.intent_for(
|
|
notify, record["transition_seq"], record["updated_at"],
|
|
detail=str(fields.get("detail", "")))
|
|
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,
|
|
idempotent=decision.idempotent)
|
|
|
|
|
|
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"),
|
|
transition_seq=note.get("transition_seq"))
|
|
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)
|