agentplane 0.45.0

//! Replay: reading a run's history back instead of re-doing it.
//!
//! Replay re-executes the *deterministic* zone — plan traversal, guards, retry
//! decisions — and satisfies every effect from the journal. Nothing external is
//! touched: no tool is called twice, no invoice is issued twice, no clock is
//! read again.
//!
//! Two things can happen that are not "read the recorded output":
//!
//! * **Divergence.** The recomputed effect key differs from the journaled one,
//!   which means this code takes a different path than the code that produced
//!   the record. The run is quarantined. It is never allowed to continue, because
//!   a diverging replay silently rewrites history.
//! * **An orphan.** An `EffectStarted` with no terminal record — a crash landed
//!   between "sent the request" and "recorded the answer". Whether that is safe
//!   to re-run is undecidable from the journal, so the effect's declared
//!   [`Recovery`] decides, and the conservative default escalates to a human.
//!
//! # Order is verified per step, not globally
//!
//! Effect keys are derived from `(step, ordinal)`, so *ordinal* order is what
//! carries meaning — and ordinals restart in every step. A single globally
//! ordered cursor coincides with that only while a plan has one step; the moment
//! two steps interleave in the journal, a globally ordered cursor rejects a
//! perfectly faithful replay.
//!
//! Grouping by step is therefore not an optimisation but the correct
//! granularity. It also makes concurrent steps safe to replay: their journal
//! interleaving may differ run to run, and nothing downstream depends on it.

use std::collections::BTreeMap;

use crate::core::{
    Disposition, Doubt, EffectDescriptor, EffectKey, Phase, Recovery, Seq, StepError, StepId,
    Undecided,
};

use super::{Record, RecordKind};

/// What the journal has to say about one effect.
#[derive(Debug, Clone, PartialEq)]
pub enum EffectReplay {
    /// It completed; here is what it returned, and what it cost.
    ///
    /// The figure is recorded rather than recomputed, so a replayed run reaches
    /// the same budget verdict at the same point as the original.
    Done {
        output: serde_json::Value,
        /// Who sent it, when the effect was an awaited inbound event.
        ///
        /// Carried so a replayed run rebuilds the same provenance the live one
        /// had. Without it the two label the same value differently, and every
        /// taint gate downstream may reach a different verdict.
        source: Option<String>,
        /// The operator this plane minted the awaited event for, when it did.
        ///
        /// Read back so a wait can hold an answer to the act it names — a
        /// worklist decision's decider must be the operator the decision was
        /// recorded under — on replay exactly as live.
        by: Option<crate::core::Operator>,
        spend: crate::core::Spend,
        /// The trust and sensitivity the effect declared when it landed.
        ///
        /// Read back for the same reason the spend is, and with a sharper
        /// consequence: these two are sourced from operator configuration
        /// rather than from code, so re-deriving them lets a catalogue edit
        /// relabel a value the run read months ago — with nothing diverging,
        /// because nothing about the call changed.
        declared: crate::core::DeclaredOutput,
    },
    /// It failed, and the failure is part of history — including what that
    /// failure said about whether the call reached the outside world.
    Failed {
        error: String,
        disposition: Disposition,
        /// What the attempt consumed before failing.
        ///
        /// Read back rather than recomputed so a replayed run reaches the same
        /// budget verdict at the same point — the same reason `Done` carries
        /// its spend.
        spend: crate::core::Spend,
        /// Whether the refusal was an answer no retry can change.
        ///
        /// Read back rather than recomputed for the same reason the
        /// disposition is: the retry decision is replayed, and a replay that
        /// could not see this bit would expect a retry the live run never
        /// made.
        permanent: bool,
    },
    /// A limit refused it before it started.
    ///
    /// Distinct from `Failed`: nothing was attempted, and the run stopped
    /// because it was told to rather than because something broke.
    Refused { limit: String, used: String },
    /// Policy refused it before it was attempted.
    ///
    /// Distinct from `Refused`, which is a *limit*. Both stop the run without
    /// attempting anything, and an operator's response to each is different: a
    /// budget is raised, a policy is argued with.
    Denied {
        reason: String,
        action: String,
        resource: String,
    },
    /// It started and we do not know whether it landed.
    ///
    /// The recorded [`Recovery`] is what decides whether re-performing it is
    /// safe. What history asked for is not here: it is on every entry rather
    /// than on this one variant, because the question *what did the run do at
    /// this position* is what a divergence has to answer, and a divergence can
    /// land on any of them.
    Orphan { recovery: Recovery },
}

/// The clear half of a journaled attempt.
///
/// **Clear is the operative word.** An effect's arguments are a sealed field on
/// its record, so a message composed from them would write caller data into a
/// run's conclusion, which is not sealed — the plaintext copy an erasure could
/// not reach. Its `kind` and its attempt number are clear on the record by the
/// same decision, and they are enough to say which *kind* of divergence this
/// is. Which argument moved is a question for a reader holding the key.
#[derive(Debug, Clone, PartialEq, Eq)]
struct Attempted {
    kind: String,
    attempt: u32,
}

impl EffectReplay {
    /// What this state says the attempt cost so far.
    ///
    /// Zero for the states that describe an attempt nobody made: a refusal and
    /// a denial both stop a call before it starts, and an announcement with no
    /// terminal record has reported no figure yet.
    #[must_use]
    pub const fn spend(&self) -> crate::core::Spend {
        match self {
            Self::Done { spend, .. } | Self::Failed { spend, .. } => *spend,
            Self::Refused { .. } | Self::Denied { .. } | Self::Orphan { .. } => {
                crate::core::Spend::ZERO
            }
        }
    }

    /// Fold an earlier record's figure into this one.
    ///
    /// Named over the states that carry a figure rather than over all of them,
    /// so a state that gains one later cannot silently keep discarding what it
    /// supersedes.
    fn add_spend(&mut self, extra: crate::core::Spend) {
        match self {
            Self::Done { spend, .. } | Self::Failed { spend, .. } => *spend += extra,
            Self::Refused { .. } | Self::Denied { .. } | Self::Orphan { .. } => {}
        }
    }
}

/// One step's effects, in the order that step performed them.
///
/// Owned by the step that is replaying it rather than borrowed from a shared
/// map. That is what lets steps run concurrently: a step touches only its own
/// history, so handing each one its slice removes the last shared mutable state
/// between them.
#[derive(Debug, Clone, Default)]
pub struct StepCursor {
    effects: Vec<Journaled>,
    pos: usize,
    /// Where this step parted from its record, once it has.
    diverged: Option<Divergence>,
}

/// Where a strict replay first parted from the record.
///
/// The structure a quarantine message flattens: an author asking *which
/// effect did my edit change* needs the step and both keys as values, not a
/// sentence to parse. Exactly one of three shapes — both keys (this build
/// asked for a different effect), only [`recomputed`](Self::recomputed) (it
/// asked for more than the record holds), only [`recorded`](Self::recorded)
/// (it never asked for an effect the record holds).
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct Divergence {
    pub step: StepId,
    pub phase: Phase,
    /// The effect history holds at this position.
    pub recorded: Option<EffectKey>,
    /// The effect this build asked for there.
    pub recomputed: Option<EffectKey>,
    /// What the effect is, where either side names it.
    pub kind: Option<String>,
    pub detail: String,
}

impl Divergence {
    /// The divergence a step error describes, if it describes one.
    ///
    /// `None` for every other failure: a refusal, a budget stop or a tool
    /// error is the run's own verdict, which a faithful replay reproduces.
    #[must_use]
    pub fn of(step: StepId, phase: Phase, error: &StepError, history: &StepCursor) -> Option<Self> {
        match error {
            StepError::NonDeterminism {
                seq,
                expected,
                actual,
                detail,
            } => Some(Self {
                step,
                phase,
                recorded: Some(*expected),
                recomputed: Some(*actual),
                kind: history.kind_at(*seq),
                detail: detail.clone(),
            }),
            StepError::ReplayOverrun { actual, kind } => Some(Self {
                step,
                phase,
                recorded: None,
                recomputed: Some(*actual),
                kind: Some(kind.clone()),
                detail: format!(
                    "this build asks for `{kind}`, past the end of the recorded history"
                ),
            }),
            _ => None,
        }
    }

    /// A journaled effect this build never asked for.
    #[must_use]
    pub fn unconsumed(step: StepId, phase: Phase, missing: &Unconsumed) -> Self {
        Self {
            step,
            phase,
            recorded: Some(missing.key),
            recomputed: None,
            kind: missing.kind.clone(),
            detail: format!("the record holds {missing}, which this build never asks for"),
        }
    }
}

/// A journaled effect this build never asked for.
///
/// Carries the call rather than only its key, for the reason a divergence does:
/// the finding is *an effect went missing*, and an operator hunting one needs to
/// know which. `kind` is absent exactly where history announced nothing — a
/// refusal has no descriptor to name.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Unconsumed {
    pub key: EffectKey,
    pub kind: Option<String>,
}

impl std::fmt::Display for Unconsumed {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match &self.kind {
            Some(kind) => write!(f, "`{kind}` ({})", self.key),
            None => write!(f, "a refused call ({})", self.key),
        }
    }
}

/// One position in a step's recorded history.
#[derive(Debug, Clone)]
struct Journaled {
    key: EffectKey,
    seq: Seq,
    /// What the record says was attempted here, where anything was.
    ///
    /// `None` for a position nothing announced: a budget refusal and a policy
    /// denial both stop a call before `EffectStarted`, so history has no
    /// descriptor to name. Saying that is better than inventing one.
    what: Option<Attempted>,
    replay: EffectReplay,
}

impl StepCursor {
    /// Overwrite the most recent slot for `key` with a terminal replay state,
    /// **carrying forward what the state it supersedes already cost**.
    ///
    /// Newest-first because a retried effect has one slot per attempt, and a
    /// terminal record always describes the latest one. One implementation
    /// rather than the same `iter_mut().rev().find(..)` at three call sites,
    /// where the fourth copy would be the one written subtly differently.
    ///
    /// The accumulation is what makes a reconciled attempt bill what it
    /// actually cost. An attempt that spent before failing and was then
    /// resolved by a probe wrote two records carrying two figures; the live
    /// pass added both, and a slot that simply took the later record's figure
    /// would replay the same run for less. A slot is one announced attempt,
    /// and its spend is everything that attempt's records report.
    fn settle(&mut self, key: EffectKey, state: EffectReplay) {
        if let Some(slot) = self.effects.iter_mut().rev().find(|e| e.key == key) {
            let carried = slot.replay.spend();
            slot.replay = state;
            slot.replay.add_spend(carried);
        }
    }

    /// Open a slot for an announced attempt.
    ///
    /// An `EffectStarted` with no terminal record after it is an orphan, so the
    /// slot starts in that state and a later record settles it. What the slot
    /// keeps besides the outcome is the *clear* half of what was asked for —
    /// which is what a divergence at this position quotes back.
    fn announce(
        &mut self,
        key: EffectKey,
        seq: Seq,
        descriptor: &EffectDescriptor,
        recovery: &Recovery,
        attempt: u32,
    ) {
        self.effects.push(Journaled {
            key,
            seq,
            what: Some(Attempted {
                kind: descriptor.kind.clone(),
                attempt,
            }),
            replay: EffectReplay::Orphan {
                recovery: recovery.clone(),
            },
        });
    }

    /// Fold one record into this step's replay sequence.
    fn apply(&mut self, key: EffectKey, seq: Seq, kind: &RecordKind) {
        match kind {
            RecordKind::EffectStarted {
                descriptor,
                recovery,
                attempt,
                ..
            } => self.announce(key, seq, descriptor, recovery, *attempt),
            RecordKind::EffectDone {
                output,
                source,
                by,
                spend,
                declared,
            } => {
                self.settle(
                    key,
                    EffectReplay::Done {
                        output: output.clone(),
                        source: source.clone(),
                        by: by.clone(),
                        spend: *spend,
                        declared: *declared,
                    },
                );
            }
            RecordKind::EffectReconciled {
                disposition,
                output,
                spend,
                declared,
                ..
            } => {
                self.settle(
                    key,
                    Self::reconciled(*disposition, output.as_ref(), *spend, *declared),
                );
            }
            // A refusal has no preceding `EffectStarted` — the whole point
            // is that nothing was announced — so it pushes its own entry
            // rather than updating one.
            RecordKind::BudgetRefused { limit, used } => {
                self.effects.push(Journaled {
                    key,
                    seq,
                    what: None,
                    replay: EffectReplay::Refused {
                        limit: limit.clone(),
                        used: used.clone(),
                    },
                });
            }
            // A re-admission supersedes the refusal it was journaled beside: a
            // resume asked the ledger then in force and was answered yes, so a
            // later replay must read refusal-then-continuation as a decision on
            // the record — serving the stale refusal would stop a strict pass
            // at a verdict the run's own later records show being overturned.
            // Step-level re-admissions carry no effect key and never reach this
            // fold; the records both stay in the chain, superseded rather than
            // erased.
            RecordKind::BudgetReadmitted { .. } => {
                if let Some(pos) = self
                    .effects
                    .iter()
                    .rposition(|e| e.key == key && matches!(e.replay, EffectReplay::Refused { .. }))
                {
                    self.effects.remove(pos);
                }
            }
            RecordKind::PolicyDenied {
                reason,
                action,
                resource,
            } => {
                self.effects.push(Journaled {
                    key,
                    seq,
                    what: None,
                    replay: EffectReplay::Denied {
                        reason: reason.clone(),
                        action: action.clone(),
                        resource: resource.clone(),
                    },
                });
            }
            RecordKind::Released { .. } => self.record_release(key, seq),
            RecordKind::EffectFailed {
                error,
                disposition,
                spend,
                permanent,
            } => {
                self.settle(
                    key,
                    EffectReplay::Failed {
                        error: error.clone(),
                        disposition: *disposition,
                        spend: *spend,
                        permanent: *permanent,
                    },
                );
            }
            _ => {}
        }
    }

    /// A reconciliation verdict, in the vocabulary the replay loop already
    /// speaks — so nothing downstream needs a separate path for it:
    ///
    ///   `Landed`       -> the effect is done, with the recovered output
    ///   `DidNotHappen` -> a failure that is safe to repeat
    ///   `InDoubt`      -> a failure that is not
    ///
    /// It overwrites whatever the attempt's earlier record said, because the
    /// probe is the later and better-informed answer.
    fn reconciled(
        disposition: Disposition,
        output: Option<&serde_json::Value>,
        spend: crate::core::Spend,
        declared: Option<crate::core::DeclaredOutput>,
    ) -> EffectReplay {
        match (disposition, output) {
            (Disposition::Landed, Some(output)) => EffectReplay::Done {
                output: output.clone(),
                source: None,
                by: None,
                spend,
                // A `Landed` verdict without a declaration is a record this
                // runtime does not write. Reading it as trusted would be the
                // relabel the field exists to stop, so the conservative point
                // stands in.
                declared: declared.unwrap_or_else(crate::core::DeclaredOutput::untrusted),
            },
            // Zero, because what the failed attempt spent is on the slot
            // already: `settle` carries it forward, and stating a figure here
            // would add the attempt's cost to itself.
            (disposition, _) => EffectReplay::Failed {
                error: "resolved by reconciliation".to_owned(),
                disposition,
                spend: crate::core::Spend::default(),
                permanent: false,
            },
        }
    }

    fn record_release(&mut self, key: EffectKey, seq: Seq) {
        self.effects.push(Journaled {
            key,
            seq,
            what: None,
            replay: EffectReplay::Done {
                output: serde_json::Value::Null,
                source: None,
                by: None,
                spend: crate::core::Spend::default(),
                // A release records no value of its own to label: the released
                // value belongs to the caller and its new label is in the
                // `Released` record. The conservative point keeps this from
                // being the one synthesized `Done` that means *trusted*.
                declared: crate::core::DeclaredOutput::untrusted(),
            },
        });
    }

    /// Whether this step's history is used up.
    ///
    /// Once it is, the step continues live — which is exactly how a crashed run
    /// resumes mid-step rather than restarting it.
    #[must_use]
    pub fn exhausted(&self) -> bool {
        self.pos >= self.effects.len()
    }

    /// The first journaled effect nothing has consumed, if any.
    ///
    /// The strict verifier's witness: a build that performs *fewer* effects
    /// than the record leaves this non-empty at the end, and the finding must
    /// name the key rather than merely count — an operator hunting a missing
    /// effect needs to know which one went missing.
    #[must_use]
    pub fn first_unconsumed(&self) -> Option<Unconsumed> {
        self.effects.get(self.pos).map(|e| Unconsumed {
            key: e.key,
            kind: e.what.as_ref().map(|w| w.kind.clone()),
        })
    }

    /// What the record says was attempted at `seq`, if anything was.
    #[must_use]
    pub fn kind_at(&self, seq: Seq) -> Option<String> {
        self.effects
            .iter()
            .find(|e| e.seq == seq)
            .and_then(|e| e.what.as_ref().map(|w| w.kind.clone()))
    }

    /// Record where this step parted from its history; the first one stands.
    pub fn note_divergence(&mut self, divergence: Divergence) {
        self.diverged.get_or_insert(divergence);
    }

    /// Whether the next journaled effect is `key`, without consuming it.
    ///
    /// Used to ask history a question it is the only authority on: after a
    /// recorded failure, did the run go on to retry? The answer is "the next
    /// effect is that attempt", and inferring it from the current retry policy
    /// instead would let a policy edit rewrite what happened.
    #[must_use]
    pub fn peek_is(&self, key: EffectKey) -> bool {
        self.effects.get(self.pos).is_some_and(|e| e.key == key)
    }

    /// Consume the next journaled effect, verifying it is the one being asked
    /// for.
    ///
    /// Returns `None` once history is exhausted, meaning "perform this one
    /// live".
    ///
    /// `asked` and `attempt` describe the call this build is making, and they
    /// are here for the failure path: two effect keys are two digests, and a
    /// developer who changed a code path needs to know *which* call moved.
    ///
    /// # Errors
    ///
    /// [`StepError::NonDeterminism`] if the recomputed key is not the one
    /// history records at this position.
    pub fn next(
        &mut self,
        recomputed: EffectKey,
        asked: &EffectDescriptor,
        attempt: u32,
    ) -> Result<Option<EffectReplay>, StepError> {
        let Some(entry) = self.effects.get(self.pos) else {
            return Ok(None);
        };
        if entry.key != recomputed {
            return Err(StepError::NonDeterminism {
                seq: entry.seq,
                expected: entry.key,
                actual: recomputed,
                detail: entry.diverged_from(asked, attempt),
            });
        }
        self.pos += 1;
        Ok(Some(entry.replay.clone()))
    }
}

impl Journaled {
    /// Say how the call this build is making differs from the one recorded
    /// here, in terms a reader can act on.
    ///
    /// **Bounded by what is clear on the record.** An effect's arguments are a
    /// sealed field, and this sentence is composed into a run's conclusion,
    /// which is not — so quoting an argument would leave a plaintext copy of
    /// caller data that an erasure cannot reach. The kind and the attempt are
    /// clear by the same decision, and between them they separate the three
    /// ways a position can diverge.
    ///
    /// **Why "the arguments differ" is sound rather than a guess.** The key is
    /// derived from the step, the phase, the ordinal, the attempt, the kind and
    /// the canonical arguments. Step and phase are fixed — this cursor is one
    /// step's — and the ordinal matches by construction at the *first*
    /// divergence, because both sides advance one position per effect and an
    /// earlier mismatch would have stopped the run already. With kind and
    /// attempt equal too, the arguments are the only input left.
    fn diverged_from(&self, asked: &EffectDescriptor, attempt: u32) -> String {
        let Some(what) = &self.what else {
            return format!(
                "history announced no call at this position — it recorded a refusal, which \
                 stops a call before it starts — and this build asks for `{}`",
                asked.kind
            );
        };
        if what.kind != asked.kind {
            return format!(
                "history performed `{}` here and this build asks for `{}` — the code takes a \
                 different path than the code that wrote this journal",
                what.kind, asked.kind
            );
        }
        if what.attempt != attempt {
            return format!(
                "both perform `{}`, at different attempts — history recorded attempt {} and \
                 this build is on attempt {}, so the retry decisions differ rather than the \
                 call",
                what.kind, what.attempt, attempt
            );
        }
        format!(
            "both perform `{}` at attempt {}, so the arguments differ — they are sealed with \
             the record, so comparing them needs a reader holding the key",
            what.kind, what.attempt
        )
    }
}

/// A read cursor over one run's journaled effects.
///
/// Order is checked, not just membership. A step that performs the same effects
/// in a different sequence has diverged just as surely as one that performs
/// different effects, and only ordered verification catches it.
#[derive(Debug, Clone, Default)]
pub struct ReplayCursor {
    /// Keyed by step **and phase**.
    ///
    /// Phase is not decoration here. A step's forward pass and its compensation
    /// are different work with different effect keys, and they can be replayed
    /// in either order — a run that is cancelled before its steps are
    /// re-dispatched compensates a step whose forward history is still
    /// unconsumed. Sharing one cursor between the two makes the compensating
    /// effect read the forward record and report non-determinism against
    /// history that is perfectly sound.
    by_step: BTreeMap<(StepId, Phase), StepCursor>,
    /// A divergence the executor found between steps rather than inside one.
    noted: Option<Divergence>,
}

impl ReplayCursor {
    /// Build from a run's records.
    #[must_use]
    pub fn from_records(records: &[Record]) -> Self {
        let mut by_step: BTreeMap<(StepId, Phase), StepCursor> = BTreeMap::new();

        for r in records {
            let Some(key) = r.effect_key() else { continue };
            // An effect without a step cannot be attributed, so it cannot be
            // replayed in order. Skipping it would silently drop history, so it
            // is treated as belonging to step 0 — which is where the runtime
            // puts effects it writes on a run's behalf.
            let step = r.body.step.unwrap_or(StepId(0));
            by_step
                .entry((step, r.body.phase))
                .or_default()
                .apply(key, r.seq(), r.kind());
        }

        Self {
            by_step,
            noted: None,
        }
    }

    /// Detach a step's history so the step can own it.
    ///
    /// Returns an empty cursor for a step with no recorded effects, which is
    /// the normal case for a live run.
    #[must_use]
    pub fn take(&mut self, step: StepId, phase: Phase) -> StepCursor {
        self.by_step.remove(&(step, phase)).unwrap_or_default()
    }

    /// Return a step's history after it has finished with it.
    pub fn restore(&mut self, step: StepId, phase: Phase, cursor: StepCursor) {
        self.by_step.insert((step, phase), cursor);
    }

    /// Whether this step's history is used up.
    #[must_use]
    pub fn exhausted(&self, step: StepId, phase: Phase) -> bool {
        self.by_step
            .get(&(step, phase))
            .is_none_or(StepCursor::exhausted)
    }

    /// Whether *any* step still has unread history.
    #[must_use]
    pub fn fully_exhausted(&self) -> bool {
        self.by_step.values().all(StepCursor::exhausted)
    }

    /// The first journaled effect nothing has consumed, with the step and
    /// phase it belongs to — the strict verifier's witness that this build
    /// performs fewer effects than the record.
    #[must_use]
    pub fn first_unconsumed(&self) -> Option<(StepId, Phase, Unconsumed)> {
        self.by_step.iter().find_map(|((step, phase), cursor)| {
            cursor.first_unconsumed().map(|u| (*step, *phase, u))
        })
    }

    /// Record a divergence found at a step boundary; the first one stands.
    pub fn note_divergence(&mut self, divergence: Divergence) {
        self.noted.get_or_insert(divergence);
    }

    /// Where this replay first parted from the record, if it did.
    ///
    /// One noted at a boundary wins, because that is the finding the run's
    /// conclusion was written from; otherwise the lowest step's own.
    #[must_use]
    pub fn divergence(&self) -> Option<&Divergence> {
        self.noted
            .as_ref()
            .or_else(|| self.by_step.values().find_map(|c| c.diverged.as_ref()))
    }

    /// The first unconsumed effect in one `(step, phase)` slice, if any.
    #[must_use]
    pub fn unconsumed_in(&self, step: StepId, phase: Phase) -> Option<Unconsumed> {
        self.by_step
            .get(&(step, phase))
            .and_then(StepCursor::first_unconsumed)
    }
}

/// Every mutating effect this history cannot decide the outcome of.
///
/// The one implementation of *what is in doubt*, and it has three readers that
/// would otherwise each have written their own: the executor deciding whether
/// an unwind is safe, the operator API answering *what do I have to go and look
/// up*, and an offline audit of a sealed history nothing may resume. A second
/// copy of this rule would be a copy that is only as strong as its laxest arm.
///
/// Two shapes of doubt, and both count. An **orphan** — a mutating
/// `EffectStarted` with no terminal record — is the crash shape: the request
/// left and the process died before the answer could be written. A terminal
/// **`InDoubt`** is the call concluding with a record that says the runtime
/// does not know whether it reached the world. Tracking only the first is the
/// easy mistake and the expensive one: a run whose last attempt failed in doubt
/// would be cancellable, and the unwind would compensate every step around a
/// call that may have landed.
///
/// A doubt is resolved by **evidence, not by time**: a later
/// `EffectReconciled` that lands or clears it — whether a probe asked the
/// provider or a person answered out of band — or, for an effect whose
/// declaration made repeating safe, a later attempt of the same dispatch (same
/// step, same descriptor) that completed, which under that declaration is the
/// same single performance finally landing.
///
/// Non-mutating effects are absent by construction. A read that never came back
/// is safe to repeat, so there is nothing for anyone to adjudicate; the question
/// here is exclusively *did this change the outside world*.
///
/// Ordered by step, then by the order the doubts arose, so two readers of one
/// history list them the same way.
#[must_use]
pub fn undecided_effects(records: &[Record]) -> Vec<Undecided> {
    // `open` holds announcements not yet terminated; `doubts` holds attempts
    // that terminated saying nothing was established. The descriptor rides
    // along because a later landing of the *same dispatch* is what clears a
    // doubt for a repeatable effect, and identity of dispatch is
    // `(step, descriptor)` rather than the key, which folds in the attempt.
    let mut open: BTreeMap<EffectKey, Pending> = BTreeMap::new();
    let mut doubts: BTreeMap<EffectKey, Pending> = BTreeMap::new();
    for r in records {
        let Some(key) = r.effect_key() else { continue };
        match r.kind() {
            RecordKind::EffectStarted {
                mutates: true,
                descriptor,
                ..
            } => {
                if let Some(step) = r.body.step {
                    open.insert(
                        key,
                        Pending {
                            step,
                            phase: r.body.phase,
                            descriptor: descriptor.clone(),
                            seq: r.body.seq,
                        },
                    );
                }
            }
            RecordKind::EffectDone { .. } => {
                if let Some(settled) = open.remove(&key) {
                    doubts.retain(|_, p| {
                        !(p.step == settled.step && p.descriptor == settled.descriptor)
                    });
                }
            }
            RecordKind::EffectFailed { disposition, .. } => {
                if let Some(pending) = open.remove(&key)
                    && *disposition == Disposition::InDoubt
                {
                    doubts.insert(key, pending);
                }
            }
            RecordKind::EffectReconciled { disposition, .. } => {
                let settled = open.remove(&key);
                match disposition {
                    // Answered: it landed, or it never happened. Either way the
                    // doubt is gone.
                    Disposition::Landed | Disposition::DidNotHappen => {
                        doubts.remove(&key);
                    }
                    // Asked, and still unknown.
                    Disposition::InDoubt => {
                        if let Some(pending) = settled {
                            doubts.insert(key, pending);
                        }
                    }
                }
            }
            _ => {}
        }
    }

    let mut out: Vec<(StepId, Seq, Undecided)> = open
        .into_iter()
        .map(|(effect, p)| p.into_undecided(effect, Doubt::Announced))
        .chain(
            doubts
                .into_iter()
                .map(|(effect, p)| p.into_undecided(effect, Doubt::Inconclusive)),
        )
        .collect();
    out.sort_by_key(|(step, seq, _)| (*step, *seq));
    out.into_iter().map(|(_, _, u)| u).collect()
}

/// An announcement awaiting a verdict, while the fold is still running.
struct Pending {
    step: StepId,
    phase: Phase,
    descriptor: EffectDescriptor,
    seq: Seq,
}

impl Pending {
    /// Paired with the position that orders it, which is deliberately not a
    /// field of [`Undecided`]: a caller acting on a doubt names the effect by
    /// key, and a sequence number beside it would be a second identity for one
    /// thing.
    fn into_undecided(self, effect: EffectKey, doubt: Doubt) -> (StepId, Seq, Undecided) {
        (
            self.step,
            self.seq,
            Undecided {
                effect,
                step: self.step,
                phase: self.phase,
                kind: self.descriptor.kind,
                doubt,
            },
        )
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::{Digest, RunId};
    use crate::journal::Append;
    use serde_json::json;

    fn key(n: u8) -> EffectKey {
        EffectKey::from_hex(&Digest::of(&[n]).to_hex()).unwrap()
    }

    fn desc() -> EffectDescriptor {
        EffectDescriptor::nullary("test.effect")
    }

    fn records(entries: Vec<(StepId, EffectKey, RecordKind)>) -> Vec<Record> {
        let run = RunId::generate();
        let mut prev = Digest::ZERO;
        let mut out = Vec::new();
        for (i, (step, k, kind)) in entries.into_iter().enumerate() {
            let a = Append::new(run, kind).effect(k).step(step);
            let r = Record::seal(a.into_body(i as u64 + 1, 1), prev).unwrap();
            prev = r.hash;
            out.push(r);
        }
        out
    }

    fn started() -> RecordKind {
        RecordKind::EffectStarted {
            descriptor: desc(),
            recovery: Recovery::Retry,
            mutates: false,
            attempt: 1,
            backoff_ms: 0,
            outbound_label: None,
            outbound_bytes: None,
        }
    }

    const S0: StepId = StepId(0);
    const S1: StepId = StepId(1);

    /// Consume one step's next effect the way the runtime does: detach the
    /// step's slice, ask it, hand it back.
    fn next(
        cur: &mut ReplayCursor,
        step: StepId,
        key: EffectKey,
    ) -> Result<Option<EffectReplay>, StepError> {
        next_asked(cur, step, key, &desc())
    }

    /// The same, naming the call this build is making — which is what a
    /// divergence quotes back.
    fn next_asked(
        cur: &mut ReplayCursor,
        step: StepId,
        key: EffectKey,
        asked: &EffectDescriptor,
    ) -> Result<Option<EffectReplay>, StepError> {
        let mut slice = cur.take(step, Phase::Forward);
        let out = slice.next(key, asked, 1);
        cur.restore(step, Phase::Forward, slice);
        out
    }

    #[test]
    fn replays_a_completed_effect_without_performing_it() {
        let recs = records(vec![
            (S0, key(1), started()),
            (
                S0,
                key(1),
                RecordKind::EffectDone {
                    declared: crate::core::DeclaredOutput::untrusted(),
                    output: json!("recorded"),
                    source: None,
                    by: None,
                    spend: crate::core::Spend::default(),
                },
            ),
        ]);
        let mut cur = ReplayCursor::from_records(&recs);
        assert_eq!(
            next(&mut cur, S0, key(1)).unwrap(),
            Some(EffectReplay::Done {
                declared: crate::core::DeclaredOutput::untrusted(),
                output: json!("recorded"),
                source: None,
                by: None,
                spend: crate::core::Spend::default()
            })
        );
        assert!(cur.exhausted(S0, Phase::Forward));
    }

    /// The critical safety property: a replay that would take a different path
    /// stops rather than quietly rewriting history.
    #[test]
    fn divergent_key_is_rejected() {
        let recs = records(vec![
            (S0, key(1), started()),
            (
                S0,
                key(1),
                RecordKind::EffectDone {
                    declared: crate::core::DeclaredOutput::untrusted(),
                    output: json!(1),
                    source: None,
                    by: None,
                    spend: crate::core::Spend::default(),
                },
            ),
        ]);
        let mut cur = ReplayCursor::from_records(&recs);
        assert!(matches!(
            next(&mut cur, S0, key(99)).unwrap_err(),
            StepError::NonDeterminism { .. }
        ));
    }

    /// **The three shapes a position can diverge in, told apart.**
    ///
    /// Two effect keys are two digests, and the party who has to act is whoever
    /// changed the code. What separates *a different call*, *a different
    /// attempt* and *the same call with different arguments* is worth saying,
    /// because each sends a reader somewhere else — and the sentence is
    /// journaled into the run's conclusion, so the unhelpful version would be
    /// the permanent one.
    ///
    /// The third arm is bounded by what may be said: an effect's arguments are
    /// a sealed field and a conclusion is not, so quoting one would leave a
    /// plaintext copy of caller data that an erasure cannot reach.
    #[test]
    fn a_divergence_says_which_of_the_three_things_moved() {
        let recs = records(vec![(S0, key(1), started())]);

        let elsewhere = EffectDescriptor::nullary("test.somewhere-else");
        let msg = next_asked(
            &mut ReplayCursor::from_records(&recs),
            S0,
            key(99),
            &elsewhere,
        )
        .expect_err("diverges")
        .to_string();
        assert!(
            msg.contains("test.effect") && msg.contains("test.somewhere-else"),
            "a changed call must name both sides: {msg}"
        );

        // Same call, and the key differs anyway: nothing else is left.
        let msg = next_asked(&mut ReplayCursor::from_records(&recs), S0, key(99), &desc())
            .expect_err("diverges")
            .to_string();
        assert!(
            msg.contains("the arguments differ") && msg.contains("sealed"),
            "the remaining input has to be named, and named as unquotable: {msg}"
        );

        // A retry this build takes and history did not.
        let mut slice = ReplayCursor::from_records(&recs).take(S0, Phase::Forward);
        let msg = slice
            .next(key(99), &desc(), 2)
            .expect_err("diverges")
            .to_string();
        assert!(
            msg.contains("attempt 1") && msg.contains("attempt 2"),
            "a differing attempt is a different finding from a differing argument: {msg}"
        );

        // A position history refused announced no call at all, and saying so
        // beats inventing a descriptor for it.
        let refused = records(vec![(
            S0,
            key(1),
            RecordKind::BudgetRefused {
                limit: "spend".into(),
                used: "1".into(),
            },
        )]);
        let msg = next_asked(
            &mut ReplayCursor::from_records(&refused),
            S0,
            key(99),
            &desc(),
        )
        .expect_err("diverges")
        .to_string();
        assert!(
            msg.contains("announced no call"),
            "a refusal has no descriptor and the reason must say so: {msg}"
        );
    }

    /// Same effects, wrong order within a step, is still divergence.
    #[test]
    fn reordered_effects_within_a_step_are_rejected() {
        let recs = records(vec![
            (S0, key(1), started()),
            (
                S0,
                key(1),
                RecordKind::EffectDone {
                    declared: crate::core::DeclaredOutput::untrusted(),
                    output: json!(1),
                    source: None,
                    by: None,
                    spend: crate::core::Spend::default(),
                },
            ),
            (S0, key(2), started()),
            (
                S0,
                key(2),
                RecordKind::EffectDone {
                    declared: crate::core::DeclaredOutput::untrusted(),
                    output: json!(2),
                    source: None,
                    by: None,
                    spend: crate::core::Spend::default(),
                },
            ),
        ]);
        let mut cur = ReplayCursor::from_records(&recs);
        assert!(
            next(&mut cur, S0, key(2)).is_err(),
            "order within a step is verified"
        );
    }

    /// **The property a global cursor gets wrong.**
    ///
    /// Two steps interleaved in the journal must each replay against their own
    /// history. A single global cursor would see step 1's effect where step 0's
    /// was expected and reject a perfectly faithful replay.
    #[test]
    fn steps_replay_independently_of_journal_interleaving() {
        let recs = records(vec![
            (S0, key(1), started()),
            (S1, key(10), started()),
            (
                S1,
                key(10),
                RecordKind::EffectDone {
                    declared: crate::core::DeclaredOutput::untrusted(),
                    output: json!("b"),
                    source: None,
                    by: None,
                    spend: crate::core::Spend::default(),
                },
            ),
            (
                S0,
                key(1),
                RecordKind::EffectDone {
                    declared: crate::core::DeclaredOutput::untrusted(),
                    output: json!("a"),
                    source: None,
                    by: None,
                    spend: crate::core::Spend::default(),
                },
            ),
        ]);
        let mut cur = ReplayCursor::from_records(&recs);

        // Step 0 asks first even though step 1 finished first in the journal.
        assert_eq!(
            next(&mut cur, S0, key(1)).unwrap(),
            Some(EffectReplay::Done {
                declared: crate::core::DeclaredOutput::untrusted(),
                output: json!("a"),
                source: None,
                by: None,
                spend: crate::core::Spend::default()
            })
        );
        assert_eq!(
            next(&mut cur, S1, key(10)).unwrap(),
            Some(EffectReplay::Done {
                declared: crate::core::DeclaredOutput::untrusted(),
                output: json!("b"),
                source: None,
                by: None,
                spend: crate::core::Spend::default()
            })
        );
        assert!(cur.fully_exhausted());
    }

    /// One step running out of history does not make another step's history
    /// unavailable.
    #[test]
    fn exhaustion_is_per_step() {
        let recs = records(vec![
            (S0, key(1), started()),
            (
                S0,
                key(1),
                RecordKind::EffectDone {
                    declared: crate::core::DeclaredOutput::untrusted(),
                    output: json!(1),
                    source: None,
                    by: None,
                    spend: crate::core::Spend::default(),
                },
            ),
            (S1, key(10), started()),
            (
                S1,
                key(10),
                RecordKind::EffectDone {
                    declared: crate::core::DeclaredOutput::untrusted(),
                    output: json!(2),
                    source: None,
                    by: None,
                    spend: crate::core::Spend::default(),
                },
            ),
        ]);
        let mut cur = ReplayCursor::from_records(&recs);
        next(&mut cur, S0, key(1)).unwrap();

        assert!(cur.exhausted(S0, Phase::Forward));
        assert!(
            !cur.exhausted(S1, Phase::Forward),
            "step 1 still has history to replay"
        );
        assert!(!cur.fully_exhausted());
    }

    #[test]
    fn an_unknown_step_has_no_history() {
        let cur = ReplayCursor::from_records(&[]);
        assert!(cur.exhausted(StepId(7), Phase::Forward));
        assert!(cur.fully_exhausted());
        assert!(cur.first_unconsumed().is_none());
    }

    #[test]
    fn exhausted_cursor_yields_none_so_execution_continues_live() {
        let recs = records(vec![
            (S0, key(1), started()),
            (
                S0,
                key(1),
                RecordKind::EffectDone {
                    declared: crate::core::DeclaredOutput::untrusted(),
                    output: json!(1),
                    source: None,
                    by: None,
                    spend: crate::core::Spend::default(),
                },
            ),
        ]);
        let mut cur = ReplayCursor::from_records(&recs);
        next(&mut cur, S0, key(1)).unwrap();
        assert_eq!(
            next(&mut cur, S0, key(2)).unwrap(),
            None,
            "resume runs the rest live"
        );
    }

    /// A crash between "sent" and "recorded" leaves this shape, and it must be
    /// recognisable — the runtime decides what to do from the declared recovery.
    #[test]
    fn orphaned_start_is_surfaced_with_its_recovery_mode() {
        let recs = records(vec![(S0, key(1), started())]);
        let mut cur = ReplayCursor::from_records(&recs);
        match next(&mut cur, S0, key(1)).unwrap() {
            Some(EffectReplay::Orphan { recovery, .. }) => {
                assert!(matches!(recovery, Recovery::Retry));
            }
            other => panic!("expected orphan, got {other:?}"),
        }
    }

    #[test]
    fn failure_is_part_of_history() {
        let recs = records(vec![
            (S0, key(1), started()),
            (
                S0,
                key(1),
                RecordKind::EffectFailed {
                    error: "boom".into(),
                    disposition: Disposition::DidNotHappen,
                    spend: crate::core::Spend::default(),
                    permanent: false,
                },
            ),
        ]);
        let mut cur = ReplayCursor::from_records(&recs);
        assert_eq!(
            next(&mut cur, S0, key(1)).unwrap(),
            Some(EffectReplay::Failed {
                error: "boom".into(),
                disposition: Disposition::DidNotHappen,
                spend: crate::core::Spend::default(),
                permanent: false,
            })
        );
    }
}