Report stale proposals on read (S-001)

changelog.d/97-stale-proposals.md

@@ -0,0 +1,3 @@
+### Added
+
+- An open proposal whose basis hosts have been observed again since is now reported as stale: `decisions list`, `decisions show` and `cloudfall why` add `stale` and `staleHosts` to each `proposed` decision, `--stale` on `decisions list` and `why` keeps only those, and `decisions approve` names each stale host and its newer observation time before asking for the id. Worked out on read from the observation files; the record and its schema do not change (#26)

docs/specs/001-stale-proposals.md

@@ -1,6 +1,6 @@
 # S-001: Stale proposals
 
-status: approved
+status: built
 
 ## Problem
 
@@ -66,4 +66,16 @@ them.
 
 ## Issues
 
+- cloudfall-dev/cloudfall#97: S-001-1, S-001-2, S-001-3, S-001-4, S-001-5, S-001-6, S-001-7, S-001-8, S-001-9
+
 ## Verification
+
+- S-001-1: proposed `deploy` on a tmp project with `hz1` observed at 09:00, rewrote `hz1.json` to 09:30, ran `cloudfall decisions list`; and `test_s001_1_a_newer_observation_makes_the_proposal_stale`, the proposal read `"stale": true, "staleHosts": ["hz1"]`
+- S-001-2: `cloudfall decisions list` on the same project before the rewrite, plus `test_s001_2_the_same_missing_or_no_observation_is_not_stale`, `test_s001_2_a_basis_that_names_no_host_is_never_stale` and `test_s001_2_times_compare_as_moments_not_as_text`, each read `"stale": false` with no `staleHosts`
+- S-001-3: `decisions list`, `decisions show` and `cloudfall why` on the tmp project's `ran` facts record, plus `test_s001_3_a_decision_that_is_not_proposed_carries_no_staleness` over verified, ran and superseded records, carried neither key
+- S-001-4: `cloudfall decisions list --stale` and `--stale --status proposed` on the tmp project, plus `test_s001_4_the_stale_filter_keeps_only_the_stale_and_combines_with_status`, returned only the stale proposal, the same set both ways
+- S-001-5: `cloudfall why --stale` on the tmp project, plus `test_s001_5_why_stale_keeps_only_the_stale_and_reports_them`, returned only the stale proposal with `stale` and `staleHosts` identical to `decisions list`
+- S-001-6: `cloudfall decisions show deploy-20261020094012` on the tmp project, plus `test_s001_6_show_reports_a_stale_proposal`, reported `"stale": true, "staleHosts": ["hz1"]`
+- S-001-7: `git hash-object` of both tmp records before and after `decisions list`, `decisions show` and `why`, plus `test_s001_7_working_out_staleness_writes_nothing` (byte and mtime compare), the bytes were unchanged
+- S-001-8: `test_s001_8_approve_names_the_stale_hosts_and_still_runs` runs `decisions approve` at a simulated terminal against a real `ansible-playbook`; the prompt named `web-1 was observed again at 2026-09-21T16:02:10Z` and the decision ended `verified` (no terminal was available to run it by hand)
+- S-001-9: wrote invalid JSON to the tmp project's `hz1.json` and ran `cloudfall decisions list`, plus `test_s001_9_an_observation_file_that_is_not_json_is_record_invalid`, it exited `RECORD_INVALID` (86) naming `tmp/s001/tmp/observed/hz1.json`

sdk/src/cloudfall/app.py

@@ -23,7 +23,7 @@
 # treaty reads a handler's annotations at registration, so a streaming
 # handler's Iterator and a provided MCP tool's Mapping must exist at runtime.
 from collections.abc import Iterator, Mapping
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from functools import partial
 from http import HTTPStatus
 from importlib.metadata import version
@@ -105,11 +105,14 @@
     ApprovalChannel,
     ApprovalRequest,
     CheckRunner,
+    Decision,
     DecisionError,
     DecisionStatus,
     DecisionStore,
+    ObservationClock,
     ProposalRequest,
     Requirement,
+    Staleness,
     Targets,
     approve,
     propose,
@@ -745,6 +748,25 @@ def store(self) -> DecisionStore:
             directory=self.root / self.decisions, catalog=SchemaCatalog(self.schemas)
         )
 
+    def clock(self) -> ObservationClock:
+        """Read the observations the records' bases name, to tell what is stale."""
+        return ObservationClock(repository=self.root)
+
+
+_STALE_DESCRIPTION = (
+    "Only proposals a host in their basis has been observed again since"
+)
+
+
+def _with_staleness(
+    decision: Decision, staleness: Staleness | None
+) -> dict[str, object]:
+    """Return the record document, with whether an open proposal is stale."""
+    document = decision.as_document()
+    if staleness is not None:
+        document.update(staleness.as_dict())
+    return document
+
 
 class CatalogPayload(Payload):
     """``operations list``: every declared operation and the risk counts."""
@@ -778,6 +800,7 @@ class ListDecisionsArgs(DecisionsArgs):
         default=(),
         description="Only decisions with this status (repeatable)",
     )
+    stale: bool = Flag(default=False, description=_STALE_DESCRIPTION)
 
 
 @dataclass(frozen=True, slots=True, kw_only=True)
@@ -830,17 +853,25 @@ def operations_show(args: ShowOperationArgs, _ctx: Ctx) -> OperationPayload:
             "List what still waits for an approval",
             "cloudfall decisions list --status proposed",
         ),
+        (
+            "List the proposals made on evidence observed again since",
+            "cloudfall decisions list --stale",
+        ),
     ],
 )
 def decisions_list(args: ListDecisionsArgs, _ctx: Ctx) -> DecisionsPayload:
     """Answer every decision record: the record outlives the catalog."""
     store = args.store()
+    clock = args.clock()
     try:
-        documents = [
-            decision.as_document()
-            for decision in store.list()
-            if not args.status or decision.status in args.status
-        ]
+        documents: list[dict[str, object]] = []
+        for decision in store.list():
+            if args.status and decision.status not in args.status:
+                continue
+            staleness = clock.staleness(decision)
+            if args.stale and (staleness is None or not staleness.stale):
+                continue
+            documents.append(_with_staleness(decision, staleness))
     except DecisionError as error:
         raise _record_invalid(error) from error
     return DecisionsPayload(
@@ -859,11 +890,14 @@ def decisions_show(args: ShowDecisionArgs, _ctx: Ctx) -> DecisionPayload:
     """Answer one decision record, or say there is none."""
     try:
         decision = args.store().load(args.decision)
+        staleness = args.clock().staleness(decision)
     except DecisionError as error:
         if error.code == ERROR_DECISION_MISSING:
             raise Exit.NOT_FOUND(error.detail, context={"code": error.code}) from error
         raise _record_invalid(error) from error
-    return DecisionPayload({"status": "ok", "decision": decision.as_document()})
+    return DecisionPayload(
+        {"status": "ok", "decision": _with_staleness(decision, staleness)}
+    )
 
 
 # operator list, show
@@ -2013,6 +2047,25 @@ def _decision_failed(error: DecisionError) -> Exception:
     return Exit.PRECONDITION(error.detail, context={"code": error.code})
 
 
+def _stale_warning(decision: Decision, staleness: Staleness | None) -> str:
+    """Name each host observed again since the proposal, for the person approving.
+
+    A stale proposal is still approved when the person says so: with
+    observations on a timer every proposal goes stale within minutes.
+    """
+    if staleness is None or not staleness.stale:
+        return ""
+    hosts = "; ".join(
+        f"{host} was observed again at {observed} "
+        f"(the proposal saw {decision.basis.observed_at[host]})"
+        for host, observed in sorted(staleness.newer.items())
+    )
+    return (
+        f"This proposal is stale: {hosts}. Check mode ran against the older "
+        "evidence. "
+    )
+
+
 @operations.command(
     "propose",
     description=(
@@ -2140,10 +2193,15 @@ def decisions_approve(args: ApproveDecisionArgs, ctx: Ctx) -> Decided:
         refuse_unapprovable(decision)
     except DecisionError as error:
         raise _decision_failed(error) from error
+    try:
+        staleness = args.clock().staleness(decision)
+    except DecisionError as error:
+        raise _record_invalid(error) from error
     # Off a terminal this ends with PERSON_REQUIRED, and a mistyped id with
     # ATTESTATION_MISMATCH, before anything runs: --yes never approves.
     attestation = ctx.attest(
-        f"The diff is at {decision.check.diff.path}. Type the decision id to "
+        f"{_stale_warning(decision, staleness)}The diff is at "
+        f"{decision.check.diff.path}. Type the decision id to "
         f"approve {decision.decision_id.value} and run it",
         expected=decision.decision_id.value,
     )
@@ -3172,6 +3230,7 @@ class WhyArgs(DecisionsArgs):
         default=(),
         description="Only decisions with this status (repeatable)",
     )
+    stale: bool = Flag(default=False, description=_STALE_DESCRIPTION)
     investigations: Path = Flag(
         default=Path(INVESTIGATION_DIRECTORY),
         description=(
@@ -3191,13 +3250,14 @@ def __post_init__(self) -> None:
 
     def query(self) -> WhyQuery:
         """Return the question, as the record is filtered by it."""
-        return WhyQuery.from_boundary(
+        query = WhyQuery.from_boundary(
             host=self.host,
             operation=self.operation.value if self.operation is not None else None,
             since=self.since,
             until=self.until,
             status=tuple(entry.value for entry in self.status),
         )
+        return replace(query, stale=self.stale)
 
 
 class WhyPayload(Payload):
@@ -3230,7 +3290,7 @@ def why(args: WhyArgs, _ctx: Ctx) -> WhyPayload:
     store = DecisionStore(directory=args.root / args.decisions, catalog=catalog)
     try:
         investigations = Investigations.load(args.root / args.investigations, catalog)
-        result = answer(store, args.query(), investigations)
+        result = answer(store, args.query(), investigations, args.clock())
     except (DecisionError, WhyError) as error:
         raise Exit.RECORD_INVALID(error.detail, context={"code": error.code}) from error
     return WhyPayload(result.as_dict())

sdk/src/cloudfall/decision.py

@@ -82,6 +82,7 @@
 _ERROR_APPROVER_UNKNOWN = "decision_approver_unknown"
 _ERROR_CHANNEL_UNKNOWN = "decision_approval_channel_unknown"
 ERROR_RECORD_INVALID = "decision_record_invalid"
+ERROR_OBSERVATION_INVALID = "decision_observation_invalid"
 
 
 class DecisionError(RuntimeError):
@@ -242,6 +243,113 @@ def as_dict(self) -> dict[str, object]:
         }
 
 
+@dataclass(frozen=True, slots=True)
+class Staleness:
+    """Which hosts a proposal's basis names have been observed again since.
+
+    Worked out when the record is read and never stored: the record stays
+    what it was when the proposal was made, and the observations say what
+    has happened since.
+    """
+
+    newer: Mapping[str, str]
+    """Each stale host and the later ``observedAt`` its observation file holds."""
+
+    @property
+    def stale(self) -> bool:
+        """Return whether any host the basis names was observed again since."""
+        return bool(self.newer)
+
+    def as_dict(self) -> dict[str, object]:
+        """Serialize the staleness beside a decision at system boundaries."""
+        result: dict[str, object] = {"stale": self.stale}
+        if self.stale:
+            result["staleHosts"] = sorted(self.newer)
+        return result
+
+
+@dataclass(frozen=True, slots=True)
+class ObservationClock:
+    """When each host was last observed, read from the folders bases name.
+
+    A relative basis folder is the repository's, as ``propose`` records it.
+    Each folder is read once, however many decisions cite it.
+    """
+
+    repository: Path
+    _folders: dict[Path, dict[str, tuple[str, Path]]] = field(
+        default_factory=dict, init=False, repr=False, compare=False
+    )
+
+    def staleness(self, decision: Decision) -> Staleness | None:
+        """Return whether one decision is stale; None when it is not ``proposed``.
+
+        Only an open proposal can be stale: every other status is final.
+        A host the basis names whose observation file is gone does not count.
+        """
+        if decision.status is not DecisionStatus.PROPOSED:
+            return None
+        if not decision.basis.observed_at:
+            return Staleness(newer={})
+        latest = self._latest(decision.basis.observations)
+        newer: dict[str, str] = {}
+        for host, recorded in sorted(decision.basis.observed_at.items()):
+            current = latest.get(host)
+            if current is None:
+                continue
+            observed, path = current
+            then = _observation_moment(recorded, f"decision {decision.decision_id}")
+            if _observation_moment(observed, str(path)) > then:
+                newer[host] = observed
+        return Staleness(newer=newer)
+
+    def _latest(self, observations: Path) -> dict[str, tuple[str, Path]]:
+        folder = (
+            observations
+            if observations.is_absolute()
+            else self.repository / observations
+        )
+        cached = self._folders.get(folder)
+        if cached is not None:
+            return cached
+        latest: dict[str, tuple[str, Path]] = {}
+        if folder.exists() and not folder.is_dir():
+            message = f"basis observations are not a folder: {folder}"
+            raise DecisionError(ERROR_OBSERVATION_INVALID, message)
+        if folder.is_dir():
+            for path in sorted(folder.glob("*.json")):
+                observed = _observed_at(path)
+                if observed is not None:
+                    latest[path.stem] = (observed, path)
+        self._folders[folder] = latest
+        return latest
+
+
+def _observed_at(path: Path) -> str | None:
+    """Return the ``spec.observedAt`` one observation file holds, as written."""
+    try:
+        document = cast("object", json.loads(path.read_text(encoding="utf-8")))
+    except (UnicodeDecodeError, json.JSONDecodeError) as error:
+        message = f"observation file is not valid JSON: {path}: {error}"
+        raise DecisionError(ERROR_OBSERVATION_INVALID, message) from error
+    if not isinstance(document, dict):
+        return None
+    spec = document.get("spec")
+    if isinstance(spec, dict) and isinstance(spec.get("observedAt"), str):
+        return str(spec["observedAt"])
+    return None
+
+
+def _observation_moment(value: str, source: str) -> datetime:
+    """Parse an observation time; a naive one is UTC, as the record writes it."""
+    try:
+        parsed = datetime.fromisoformat(value)
+    except ValueError as error:
+        message = f"observation time is not ISO 8601: {source}: {value!r}"
+        raise DecisionError(ERROR_OBSERVATION_INVALID, message) from error
+    return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)
+
+
 class ApprovalChannel(StrEnum):
     """How an approval reached the record.