Mark decisions proposed during a replay and refuse to approve them

CHANGELOG.md

@@ -23,6 +23,16 @@ Notable changes to Cloudfall. The format follows
   unknown status is refused. Records already on disk keep their status. One
   disk-full incident left 47 records for `cloudfall why --host hz1`, most of
   them reads and repeats (#26)
+- A decision proposed during `agent investigate --replay` names the
+  recording its check output was played back from, as `replay` in the
+  decision record (optional in `operation-decision.schema.json`; a record
+  without it was checked live). `decisions list` shows it, `cloudfall why`
+  says so in its story, and `decisions approve` refuses such a decision with
+  `PRECONDITION` (`decision_replayed`) before asking for the decision id or
+  running anything. `--replay` also refuses, before the model is asked, a
+  `--decisions` directory that resolves to the recording itself, which would
+  have written played-back records into the run being played, and a
+  `--replay` path longer than the 512 characters the records keep (#36)
 - Every turn of `agent investigate` keeps its raw trace, in the stream's
   `turn` event (`trace`) and in the investigation record (`spec.trace`): the
   model's reasoning when it returns one (Nemotron's `reasoning_content`),

config/schemas/v1/operation-decision.schema.json

@@ -363,6 +363,12 @@
           "description": "Why the decision ended in its status.",
           "minLength": 1,
           "maxLength": 300
+        },
+        "replay": {
+          "type": "string",
+          "description": "The recording this decision's check output was played back from, as agent investigate --replay named it. Such a decision cannot be approved. A record without it was checked live.",
+          "minLength": 1,
+          "maxLength": 512
         }
       },
       "required": [

examples/disk-full-incident/README.md

@@ -91,7 +91,9 @@ cloudfall agent investigate --replay recordings/disk-full \
 ```
 
 A call the recording does not hold, such as the logs of a unit nobody asked
-for during the evaluation, fails and says so.
+for during the evaluation, fails and says so. The decisions a replay
+proposes name the recording under `replay` and cannot be approved: their
+check never ran on a host.
 
 ## What happened when we ran it
 

sdk/src/cloudfall/app.py

@@ -112,6 +112,7 @@
     Targets,
     approve,
     propose,
+    refuse_replayed,
     refuse_unapprovable,
 )
 from cloudfall.domain import (
@@ -135,6 +136,7 @@
     ERROR_MODEL_REFUSED,
     ERROR_MODEL_UNREACHABLE,
     INVESTIGATION_DIRECTORY,
+    REPLAY_LIMIT,
     AgentTool,
     ChatTransport,
     Investigation,
@@ -2034,8 +2036,14 @@ def operations_propose(args: ProposeArgs, ctx: Ctx) -> Decided:
     return _propose(args, _check_runner(ctx, args.root))
 
 
-def _propose(args: ProposeArgs, run: CheckRunner) -> Decided:
-    """Propose one operation, running check mode through ``run``."""
+def _propose(
+    args: ProposeArgs, run: CheckRunner, replay: Path | None = None
+) -> Decided:
+    """Propose one operation, running check mode through ``run``.
+
+    ``replay`` names the recording ``run`` plays back, so the record says
+    its check output never came from the hosts.
+    """
     catalog = args.catalog()
     try:
         operation = catalog.get(args.operation)
@@ -2051,6 +2059,7 @@ def _propose(args: ProposeArgs, run: CheckRunner) -> Decided:
         inputs=args.inputs(),
         repository=args.root,
         observations=args.root / args.observed,
+        replay=None if replay is None else replay.as_posix(),
     )
     store = DecisionStore(
         directory=args.root / args.decisions, catalog=SchemaCatalog(args.schemas)
@@ -2120,6 +2129,7 @@ def decisions_approve(args: ApproveDecisionArgs, ctx: Ctx) -> Decided:
             raise Exit.NOT_FOUND(error.detail, context={"code": error.code}) from error
         raise _record_invalid(error) from error
     try:
+        refuse_replayed(decision)
         refuse_unapprovable(decision)
     except DecisionError as error:
         raise _decision_failed(error) from error
@@ -2231,7 +2241,8 @@ class InvestigateArgs(CatalogArgs):
         default=None,
         description=(
             "Decisions directory of a real run to play back instead of running "
-            "Ansible: the model is live, the hosts are recorded"
+            "Ansible: the model is live, the hosts are recorded, and the "
+            "decisions it proposes name the recording and cannot be approved"
         ),
     )
     investigations: Path = Flag(
@@ -2254,6 +2265,22 @@ def __post_init__(self) -> None:
         if not self.alert.strip():
             message = "the alert is empty"
             raise ParseError(message, context={"flag": "alert"})
+        if self.replay is not None and len(self.replay.as_posix()) > REPLAY_LIMIT:
+            # The investigation and its decisions keep the path as given.
+            message = f"--replay is longer than {REPLAY_LIMIT} characters"
+            raise ParseError(message, context={"flag": "replay"})
+        if (
+            self.replay is not None
+            and (self.root / self.decisions).resolve()
+            == (self.root / self.replay).resolve()
+        ):
+            # A recording is a decisions directory: written into, the replay
+            # would add played-back records to the run it is playing.
+            message = (
+                f"--decisions {self.decisions} is the recording --replay "
+                f"{self.replay} plays back; pass another --decisions directory"
+            )
+            raise ParseError(message, context={"flag": "decisions"})
         self.endpoint()
 
     def endpoint(self) -> ModelEndpoint:
@@ -2438,7 +2465,7 @@ def call(arguments: Mapping[str, object]) -> ToolResult:
             decisions=args.decisions,
         )
         try:
-            return _tool_result(_propose(proposal, run), ok=True)
+            return _tool_result(_propose(proposal, run, args.replay), ok=True)
         except CliExit as error:
             if isinstance(error.data, Decided):
                 failed = _tool_result(error.data, ok=False, repository=args.root)

sdk/src/cloudfall/decision.py

@@ -69,6 +69,7 @@
 _ERROR_DECISION_EXISTS = "decision_exists"
 ERROR_DECISION_MISSING = "decision_missing"
 _ERROR_NOT_PROPOSED = "decision_not_proposed"
+ERROR_DECISION_REPLAYED = "decision_replayed"
 _ERROR_APPROVER_UNKNOWN = "decision_approver_unknown"
 _ERROR_CHANNEL_UNKNOWN = "decision_approval_channel_unknown"
 ERROR_RECORD_INVALID = "decision_record_invalid"
@@ -294,6 +295,8 @@ class Decision:
     verdict: str | None = None
     ran_exit_code: int | None = None
     superseded_by: ResourceId | None = None
+    replay: str | None = None
+    """The recording its check output was played back from; None when live."""
 
     def same_call(self, other: Decision) -> bool:
         """Return whether both name the same operation, targets and inputs."""
@@ -334,6 +337,8 @@ def as_document(self) -> dict[str, object]:
             spec["ranExitCode"] = self.ran_exit_code
         if self.superseded_by is not None:
             spec["supersededBy"] = self.superseded_by.value
+        if self.replay is not None:
+            spec["replay"] = self.replay
         return {
             "apiVersion": API_VERSION,
             "kind": DECISION_KIND,
@@ -425,6 +430,8 @@ class ProposalRequest:
     inputs: Mapping[str, object]
     repository: Path
     observations: Path
+    replay: str | None = None
+    """The recording check mode is played back from, as the caller named it."""
 
 
 def propose(
@@ -490,6 +497,7 @@ def propose(
             else None
         ),
         ran_exit_code=exit_code if ran else None,
+        replay=request.replay,
     )
     store.save(decision)
     for older in recorded:
@@ -552,6 +560,7 @@ def approve(
     human read and the run that follows it.
     """
     decision = request.decision
+    refuse_replayed(decision)
     refuse_unapprovable(decision)
     if not request.approver.strip():
         message = "an approval records who gave it; pass --approver or set USER"
@@ -622,6 +631,23 @@ def refuse_unapprovable(decision: Decision) -> None:
     raise DecisionError(_ERROR_NOT_PROPOSED, message)
 
 
+def refuse_replayed(decision: Decision) -> None:
+    """Refuse a decision whose check output was played back, before anything runs.
+
+    Its diff came from a recording, not from the hosts, so approving it
+    would run something nobody saw checked against the fleet. Callers that
+    ask a person for the decision id check this first.
+    """
+    if decision.replay is None:
+        return
+    message = (
+        f"decision {decision.decision_id} was proposed during a replay of "
+        f"{decision.replay}: its check output was played back, not run on the "
+        "hosts, so it cannot be approved; propose it again live"
+    )
+    raise DecisionError(ERROR_DECISION_REPLAYED, message)
+
+
 def _outcome(
     execution: RunRecord,
     verification: RunRecord | None,
@@ -977,6 +1003,7 @@ def _decision_from_document(document: Mapping[str, object]) -> Decision:
             if "supersededBy" in spec
             else None
         ),
+        replay=str(spec["replay"]) if "replay" in spec else None,
         execution=_run_from_document(spec.get("execution")),
         verification=_run_from_document(spec.get("verify")),
         approval=_approval_from_document(approval),

sdk/src/cloudfall/investigation.py

@@ -48,6 +48,8 @@
 """Most investigations recorded under one second's id, with their suffixes."""
 ENDPOINT_LIMIT = 512
 MODEL_NAME_LIMIT = 256
+REPLAY_LIMIT = 512
+"""Most characters of the --replay path the records keep, as their schemas allow."""
 
 ERROR_ENDPOINT_INVALID = "agent_endpoint_invalid"
 ERROR_MODEL_ANSWER_MALFORMED = "agent_model_answer_malformed"

sdk/src/cloudfall/why.py

@@ -372,6 +372,11 @@ def _story(decision: Decision) -> Iterable[str]:
     yield _saw(decision)
     yield _proposed(decision)
     yield _checked(decision)
+    if decision.replay is not None:
+        yield (
+            f"That check output was played back from the recording "
+            f"{decision.replay}, not run on the hosts, so it cannot be approved."
+        )
     yield _gated(decision)
     if decision.approval is not None:
         yield _approved(decision.approval)

sdk/tests/test_decision.py

@@ -4,6 +4,7 @@
 
 import io
 import json
+from dataclasses import replace
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import TYPE_CHECKING
@@ -921,6 +922,94 @@ def test_one_proposal_is_approved_once(tmp_path: Path) -> None:
     assert error.value.code == "decision_not_proposed"
 
 
+def test_a_replayed_decision_is_never_approved(tmp_path: Path) -> None:
+    """Its check output came from a recording, not the hosts (#36)."""
+    repository = _repository(tmp_path)
+    store = _store(repository)
+    request = _proposal(repository)
+    replayed = propose(
+        ProposalRequest(
+            operation=request.operation,
+            targets=request.targets,
+            inputs=request.inputs,
+            repository=request.repository,
+            observations=request.observations,
+            replay="recordings/disk-full",
+        ),
+        store,
+        lambda: MOMENT,
+        _Runner(),
+    )
+    run = _Runner()
+
+    with pytest.raises(DecisionError) as error:
+        approve(
+            ApprovalRequest(
+                decision=store.load(replayed.decision_id),
+                approver="roman",
+                repository=repository,
+                via=TERMINAL,
+            ),
+            store,
+            lambda: MOMENT,
+            run,
+        )
+
+    assert replayed.replay == "recordings/disk-full"
+    assert error.value.code == "decision_replayed"
+    assert "recordings/disk-full" in error.value.detail
+    assert run.recorded == []
+    assert store.load(replayed.decision_id) == replayed
+
+
+def test_a_replayed_decision_since_superseded_is_refused_as_replayed(
+    tmp_path: Path,
+) -> None:
+    """Replayed is said first: no later proposal makes it approvable (#36, #26)."""
+    repository = _repository(tmp_path)
+    store = _store(repository)
+    request = _proposal(repository)
+    replayed = propose(
+        replace(request, replay="recordings/disk-full"),
+        store,
+        lambda: MOMENT,
+        _Runner(),
+    )
+    newer = propose(request, store, lambda: LATER, _Runner())
+    superseded = store.load(replayed.decision_id)
+    run = _Runner()
+
+    with pytest.raises(DecisionError) as error:
+        approve(
+            ApprovalRequest(
+                decision=superseded,
+                approver="roman",
+                repository=repository,
+                via=TERMINAL,
+            ),
+            store,
+            lambda: MOMENT,
+            run,
+        )
+
+    assert superseded.status is DecisionStatus.SUPERSEDED
+    assert superseded.superseded_by == newer.decision_id
+    assert error.value.code == "decision_replayed"
+    assert "replay of recordings/disk-full" in error.value.detail
+    assert run.recorded == []
+    assert store.load(replayed.decision_id) == superseded
+
+
+def test_a_record_without_replay_reads_as_live(tmp_path: Path) -> None:
+    repository = _repository(tmp_path)
+    store = _store(repository)
+
+    decision = propose(_proposal(repository), store, lambda: MOMENT, _Runner())
+
+    assert store.load(decision.decision_id).replay is None
+    assert "replay" not in json.dumps(decision.as_document())
+
+
 def test_an_approval_records_who_gave_it(tmp_path: Path) -> None:
     repository = _repository(tmp_path)
     store = _store(repository)