""" Synthetic stop/return state model. This is a local, in-process demonstration of the state logic described in docs/drafts/stop-return-2026-10-11/demo-plan.md. It does not call any network, billing, printer or storage system. Every "external" effect here is a plain Python value inside this process, standing in for a third-party system's status. State model: local: running -> stopping -> stopped Once stopped, the local producer cannot schedule new work. external: pending -> confirmed | canceled | unknown unknown cannot be promoted to confirmed or failed merely by a timeout. A changed version/target/amount invalidates any prior approval tied to the old one. Reversal is only offered against a confirmed effect that is itself marked reversible. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Optional LOCAL_STATES = ("running", "stopping", "stopped") EXTERNAL_STATES = ("pending", "confirmed", "canceled", "unknown", "none") class SchedulingBlocked(Exception): """Raised when new work is scheduled against a stopped local producer.""" class ReversalNotAllowed(Exception): """Raised when reversal is attempted against anything but a confirmed, reversible effect.""" class InvalidTransition(Exception): """Raised for a transition the model does not define.""" @dataclass class ApprovalToken: """An approval tied to a specific version/target/amount triple. Any change to version, target or amount invalidates this token. A new approval is a new decision, not a continuation of the old one. """ version: str target: str amount: Optional[float] valid: bool = True def matches(self, version: str, target: str, amount: Optional[float]) -> bool: return self.version == version and self.target == target and self.amount == amount @dataclass class Task: """One synthetic unit of work the demo tracks through stop and return.""" id: str description: str local_state: str = "running" external_state: str = "none" # "none" = no external request was ever sent reversible: bool = False approval: Optional[ApprovalToken] = None evidence_note: str = "" next_evidence_step: str = "" history: list = field(default_factory=list) def _log(self, event: str) -> None: self.history.append(event) # --- local lifecycle ------------------------------------------------- def stop(self) -> None: """Move local -> stopping -> stopped. Idempotent once stopped.""" if self.local_state == "running": self.local_state = "stopping" self._log("local: running -> stopping") self.local_state = "stopped" self._log("local: stopping -> stopped") elif self.local_state == "stopping": self.local_state = "stopped" self._log("local: stopping -> stopped") elif self.local_state == "stopped": self._log("local: stop() called again, already stopped (no-op)") else: raise InvalidTransition(f"unknown local_state {self.local_state!r}") def schedule_new_work(self) -> None: """A stopped local producer may never schedule new work. This is the rule the demo exists to show on screen, not just enforce in the backend: a disabled control needs this as its reason. """ if self.local_state == "stopped": raise SchedulingBlocked( f"task {self.id}: local producer is stopped; " "it cannot schedule new work. Resume only safe remaining " "work, or start a new, separately approved action." ) self._log("local: scheduled new work") # --- external effect --------------------------------------------------- def resolve_external(self, outcome: str, evidence_note: str = "") -> None: """Move external pending -> confirmed | canceled | unknown. This may be called more than once in sequence (e.g. pending -> unknown -> confirmed) as real evidence arrives later; it is not a one-shot terminal call. """ if outcome not in ("confirmed", "canceled", "unknown", "pending"): raise InvalidTransition(f"unknown external outcome {outcome!r}") prior = self.external_state self.external_state = outcome if evidence_note: self.evidence_note = evidence_note self._log(f"external: {prior} -> {outcome}") def timeout_tick(self) -> None: """Represents elapsed time passing with no new evidence. This must be a no-op for state transitions. unknown must never be promoted to confirmed or failed just because a timeout fired; the model has no "failed" state at all for this reason, and timeout_tick() never changes external_state. """ self._log("external: timeout elapsed (no state change)") # --- approval ------------------------------------------------------ def grant_approval(self, version: str, target: str, amount: Optional[float] = None) -> None: self.approval = ApprovalToken(version=version, target=target, amount=amount) self._log(f"approval: granted for version={version} target={target} amount={amount}") def change_request( self, version: Optional[str] = None, target: Optional[str] = None, amount: Optional[float] = None, ) -> None: """Change the action's version, target or amount after approval. Any one of these changing invalidates the prior approval token. Renewed approval is then required before the task may proceed; this function never silently re-approves. """ if self.approval is None: self._log("approval: change_request() called with no prior approval to invalidate") return new_version = version if version is not None else self.approval.version new_target = target if target is not None else self.approval.target new_amount = amount if amount is not None else self.approval.amount if not self.approval.matches(new_version, new_target, new_amount): self.approval.valid = False self._log( "approval: invalidated because version/target/amount changed " f"(was version={self.approval.version} target={self.approval.target} " f"amount={self.approval.amount}; now version={new_version} " f"target={new_target} amount={new_amount})" ) def has_valid_approval(self) -> bool: return self.approval is not None and self.approval.valid # --- reversal / compensation ---------------------------------------- def reverse(self) -> None: """Reverse this task's external effect. Only permitted when the external effect is confirmed and marked reversible. Anything else (pending, unknown, canceled, none, or confirmed-but-not-reversible) is rejected with an explanation rather than silently ignored or silently allowed. A caller that wants to undo a non-reversible confirmed effect must file that as compensation: a new, separately approved action, not this call. """ if self.external_state != "confirmed": raise ReversalNotAllowed( f"task {self.id}: external effect is {self.external_state!r}, " "not confirmed. There is nothing settled yet to reverse. " f"{self.next_evidence_step or 'Wait for resolving evidence.'}" ) if not self.reversible: raise ReversalNotAllowed( f"task {self.id}: external effect is confirmed but marked " "not reversible. This cannot be undone as a true reversal; " "if you need the effect gone, that is compensation, a new " "action with its own approval and consequences." ) self._log("external: confirmed -> reversed") self.external_state = "canceled" self.reversible = False def to_dict(self) -> dict: return { "id": self.id, "description": self.description, "local_state": self.local_state, "external_state": self.external_state, "reversible": self.reversible, "approval_valid": self.has_valid_approval(), "evidence_note": self.evidence_note, "next_evidence_step": self.next_evidence_step, "history": list(self.history), }