Supersede a proposal only by one of the same replay value

CHANGELOG.md

@@ -441,6 +441,15 @@ Notable changes to Cloudfall. The format follows
 
 ### Fixed
 
+- A decision proposed during `agent investigate --replay` no longer marks an
+  open live proposal of the same operation, targets and inputs `superseded`,
+  which left the operator a live proposal closed and a replayed one that
+  `decisions approve` refuses. A newer proposal now supersedes an older one
+  only when both carry the same `replay` value: live closes live, a replay
+  closes a replay of the same recording, and nothing crosses between them. A
+  replayed read is still recorded `ran`, but its verdict and `cloudfall why`
+  now say its output was played back from the recording, not run on the
+  hosts (#58)
 - `agent investigate` keeps its investigation record when the reader closes
   stdout mid-run (such as `| head -2`). Since the command streams, such a
   run left the decisions it proposed without the investigation that made

sdk/src/cloudfall/decision.py

@@ -491,17 +491,19 @@ def propose(
         basis=read_basis(request.observations, request.repository),
         check=_preview(exit_code, tree, diff_path, request.repository),
         status=DecisionStatus.RAN if ran else DecisionStatus.PROPOSED,
-        verdict=(
-            f"a read operation runs when it is proposed; it exited {exit_code}"
-            if ran
-            else None
-        ),
+        verdict=_ran_verdict(exit_code, request.replay) if ran else None,
         ran_exit_code=exit_code if ran else None,
         replay=request.replay,
     )
     store.save(decision)
     for older in recorded:
-        if older.status is DecisionStatus.PROPOSED and older.same_call(decision):
+        # A replay never closes a live proposal and a live run never closes
+        # a replayed one: only a record of the same replay value is replaced.
+        if (
+            older.status is DecisionStatus.PROPOSED
+            and older.replay == decision.replay
+            and older.same_call(decision)
+        ):
             store.update(
                 replace(
                     older,
@@ -516,6 +518,15 @@ def propose(
     return decision
 
 
+def _ran_verdict(exit_code: int, replay: str | None) -> str:
+    if replay is None:
+        return f"a read operation runs when it is proposed; it exited {exit_code}"
+    return (
+        f"a read operation's output was played back from the recording {replay}, "
+        f"not run on the hosts; it exited {exit_code}"
+    )
+
+
 @dataclass(frozen=True, slots=True)
 class PlaybookInvocation:
     """One playbook run against one set of targets."""

sdk/src/cloudfall/why.py

@@ -465,6 +465,11 @@ def _ran(subject: str, run: RunRecord, log: object) -> str:
 def _ended(decision: Decision) -> str:
     if decision.status is DecisionStatus.PROPOSED:
         return "It is still proposed: nobody has approved or rejected it."
+    if decision.status is DecisionStatus.RAN and decision.replay is not None:
+        return (
+            f"Its played-back output exited {decision.ran_exit_code}: "
+            "a read operation waits for no approval."
+        )
     if decision.status is DecisionStatus.RAN:
         return (
             f"It ran when it was proposed and exited {decision.ran_exit_code}: "

sdk/tests/test_decision.py

@@ -975,7 +975,12 @@ def test_a_replayed_decision_since_superseded_is_refused_as_replayed(
         lambda: MOMENT,
         _Runner(),
     )
-    newer = propose(request, store, lambda: LATER, _Runner())
+    newer = propose(
+        replace(request, replay="recordings/disk-full"),
+        store,
+        lambda: LATER,
+        _Runner(),
+    )
     superseded = store.load(replayed.decision_id)
     run = _Runner()
 
@@ -1000,6 +1005,89 @@ def test_a_replayed_decision_since_superseded_is_refused_as_replayed(
     assert store.load(replayed.decision_id) == superseded
 
 
+@pytest.mark.parametrize(
+    ("older_replay", "newer_replay", "supersedes"),
+    [
+        (None, None, True),
+        (None, "recordings/disk-full", False),
+        ("recordings/disk-full", None, False),
+        ("recordings/disk-full", "recordings/disk-full", True),
+        ("recordings/disk-full", "recordings/other", False),
+    ],
+    ids=[
+        "live-then-live",
+        "live-then-replay",
+        "replay-then-live",
+        "replay-then-same-replay",
+        "replay-then-other-replay",
+    ],
+)
+def test_only_a_proposal_of_the_same_replay_value_supersedes(
+    tmp_path: Path,
+    *,
+    older_replay: str | None,
+    newer_replay: str | None,
+    supersedes: bool,
+) -> None:
+    """A replay never closes a live proposal, nor a live run a replayed one (#58)."""
+    repository = _repository(tmp_path)
+    store = _store(repository)
+    request = _proposal(repository)
+    older = propose(
+        replace(request, replay=older_replay), store, lambda: MOMENT, _Runner()
+    )
+
+    newer = propose(
+        replace(request, replay=newer_replay), store, lambda: LATER, _Runner()
+    )
+
+    reloaded = store.load(older.decision_id)
+    if supersedes:
+        assert reloaded.status is DecisionStatus.SUPERSEDED
+        assert reloaded.superseded_by == newer.decision_id
+    else:
+        assert reloaded == older
+        assert reloaded.status is DecisionStatus.PROPOSED
+
+
+def test_a_replayed_read_says_its_output_was_played_back(tmp_path: Path) -> None:
+    """It is recorded ran, but it never ran on a host (#58)."""
+    repository = _repository(tmp_path, deploy=DEPLOY, facts=FACTS)
+    store = _store(repository)
+
+    decision = propose(
+        replace(
+            _proposal(repository, "facts", target=None, inputs={}),
+            replay="recordings/disk-full",
+        ),
+        store,
+        lambda: MOMENT,
+        _Runner(exit_code=2),
+    )
+
+    assert decision.status is DecisionStatus.RAN
+    assert decision.ran_exit_code == 2
+    assert decision.verdict is not None
+    assert "played back from the recording recordings/disk-full" in decision.verdict
+    assert "not run on the hosts" in decision.verdict
+    assert "runs when it is proposed" not in decision.verdict
+
+
+def test_a_live_read_keeps_its_verdict(tmp_path: Path) -> None:
+    repository = _repository(tmp_path, deploy=DEPLOY, facts=FACTS)
+
+    decision = propose(
+        _proposal(repository, "facts", target=None, inputs={}),
+        _store(repository),
+        lambda: MOMENT,
+        _Runner(exit_code=0),
+    )
+
+    assert decision.verdict == (
+        "a read operation runs when it is proposed; it exited 0"
+    )
+
+
 def test_a_record_without_replay_reads_as_live(tmp_path: Path) -> None:
     repository = _repository(tmp_path)
     store = _store(repository)

sdk/tests/test_why.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+from dataclasses import replace
 from datetime import UTC, datetime
 from typing import TYPE_CHECKING
 
@@ -178,6 +179,33 @@ def test_every_sentence_of_the_story_cites_the_record(tmp_path: Path) -> None:
     )
 
 
+def test_a_replayed_read_is_not_told_as_run_on_the_hosts(tmp_path: Path) -> None:
+    """Its output was played back from a recording, never run (#58)."""
+    repository = _repository(tmp_path)
+    store = _store(repository)
+    propose(
+        replace(
+            _proposal(repository, "facts", target=None, inputs={}),
+            replay="recordings/disk-full",
+        ),
+        store,
+        lambda: MOMENT,
+        _Runner(),
+    )
+
+    story = answer(store, WhyQuery.from_boundary(operation="facts"))
+    told = story.explanations[0].story
+
+    assert told[3] == (
+        "That check output was played back from the recording "
+        "recordings/disk-full, not run on the hosts, so it cannot be approved."
+    )
+    assert told[-1] == (
+        "Its played-back output exited 0: a read operation waits for no approval."
+    )
+    assert not any("It ran when it was proposed" in line for line in told)
+
+
 def test_a_superseded_proposal_says_what_replaced_it(tmp_path: Path) -> None:
     repository = _repository(tmp_path)
     store = _store(repository)