aion-rs 0.31.0

//! Recorder: single-writer append path over `EventStore`.

use std::collections::HashMap;
use std::sync::Arc;

use aion_core::{
    ActivityError, ActivityId, Event, EventEnvelope, PackageVersion, Payload, RunId,
    ScheduleConfig, ScheduleId, SearchAttributeSchema, SearchAttributeValue, TimerCancelCause,
    TimerId, WithTimeoutOutcome, WorkerAttribution, WorkflowError, WorkflowId,
};
use aion_store::visibility::VisibilityStore;

use crate::lifecycle::visibility::project_visibility;
use aion_store::{EventStore, OutboxRow, WriteToken};
use chrono::{DateTime, Utc};

use crate::durability::{DurabilityError, seq::SequenceHead};

/// Durable fan-out dispatch: the additive atomic events + outbox-rows append.
mod fan_out;
#[cfg(test)]
mod fan_out_tests;
/// Generation boundaries: one batch ends a run and opens its successor.
mod generation;
/// The run-scoped append guard: which generation may still append.
mod run_guard;

pub use fan_out::{FanOutCompletionResult, FanOutItem, FanOutOutcome};
pub use generation::{ContinuationTerminal, ContinuedGeneration, OpeningGeneration};
pub use run_guard::RunAdmission;

/// Identity fields recorded on a `WorkflowStarted` event.
#[derive(Clone, Debug)]
pub struct WorkflowStartRecord {
    /// Logical workflow type started by the caller.
    pub workflow_type: String,
    /// Opaque workflow input payload.
    pub input: Payload,
    /// Concrete run identifier for this execution.
    pub run_id: RunId,
    /// Parent run that continued into this run, when applicable.
    pub parent_run_id: Option<RunId>,
    /// The parent WORKFLOW that spawned this run as a child (aion#77);
    /// `None` for operator-started roots and continue-as-new successors.
    pub parent_workflow_id: Option<WorkflowId>,
    /// Package version this run was resolved against at record time.
    pub package_version: PackageVersion,
}

/// Single append authority for one workflow history.
///
/// A recorder owns the workflow's tracked sequence head and computes every `expected_seq` from that
/// tracker. It never reads the store head after construction; a sequence conflict is surfaced as a
/// hard durability error because it indicates a second writer for the same workflow.
pub struct Recorder {
    workflow_id: WorkflowId,
    store: Arc<dyn EventStore>,
    sequence: SequenceHead,
    write_token: WriteToken,
    visibility: Option<RecorderVisibility>,
    run_id: Option<RunId>,
}

struct RecorderVisibility {
    run_id: RunId,
    store: Arc<dyn VisibilityStore>,
}

impl Recorder {
    /// Creates a recorder for a fresh workflow history starting at sequence head `0`.
    #[must_use]
    pub fn new(workflow_id: WorkflowId, store: Arc<dyn EventStore>) -> Self {
        Self::resume_at(workflow_id, store, 0)
    }

    /// Creates a recorder for a workflow whose existing history head was already derived.
    #[must_use]
    pub fn resume_at(workflow_id: WorkflowId, store: Arc<dyn EventStore>, head: u64) -> Self {
        Self {
            workflow_id,
            store,
            write_token: WriteToken::recorder(),
            sequence: SequenceHead::from_head(head),
            visibility: None,
            run_id: None,
        }
    }

    /// The sequence head this recorder has appended up to.
    ///
    /// Exposed so a caller that appended through the loop's ONE recorder can
    /// hand the resulting head to a successor recorder without a second
    /// whole-history read — the read would answer a question this recorder
    /// already knows the answer to, and on a workloop that read grows without
    /// bound with the loop's age.
    #[must_use]
    pub const fn head(&self) -> u64 {
        self.sequence.current()
    }

    /// Sets the concrete run context used for run-scoped outbox metadata.
    #[must_use]
    pub fn with_run_id(mut self, run_id: RunId) -> Self {
        self.run_id = Some(run_id);
        self
    }

    /// Enables visibility projection upserts after workflow-level state-changing events recorded
    /// directly through this recorder.
    #[must_use]
    pub fn with_visibility(mut self, run_id: RunId, store: Arc<dyn VisibilityStore>) -> Self {
        self.run_id = Some(run_id.clone());
        self.visibility = Some(RecorderVisibility { run_id, store });
        self
    }

    /// Point the visibility projection at `run_id` — the generation this
    /// recorder now writes for, after a boundary started a successor in the
    /// same history. A recorder without visibility has nothing to point.
    pub(crate) fn retarget_visibility(&mut self, run_id: RunId) {
        if let Some(visibility) = &mut self.visibility {
            visibility.run_id = run_id;
        }
    }

    /// Returns the workflow this recorder appends to.
    #[must_use]
    pub const fn workflow_id(&self) -> &WorkflowId {
        &self.workflow_id
    }

    /// Returns the current tracked sequence head.
    #[must_use]
    pub const fn current_head(&self) -> u64 {
        self.sequence.current()
    }

    /// Reconciles the tracked sequence head FORWARD to `durable_head` after an
    /// acknowledgement-lost append is discovered to have landed (aion#145).
    ///
    /// The incident shape: a store call returned an error AFTER its write
    /// landed, so the append never reached `mark_append_success` and this
    /// tracker sat below the durable head — every later append would mint a
    /// stale sequence and die on `SequenceConflict`. Since AD-014 (aion#192)
    /// the ONE append path resolves that outcome itself, under this lock,
    /// before surfacing anything ([`Self::durable_append`]), so a recorder
    /// should no longer be left stale by its own append. This method remains
    /// for the timer bridge's already-recorded-fire branch — the second line,
    /// reached when a fire is redelivered for an event that is already the
    /// timer's last — which has proven UNDER THE RECORDER LOCK that the
    /// active-segment history shows this recorder's own `TimerFired` as the
    /// firing timer's last event (single-writer discipline: an event of that
    /// exact shape can only be ours).
    ///
    /// Forward-only, never backward: a `durable_head` at or below the tracked
    /// head leaves the tracker untouched (the at-head case is the delivery
    /// retry of a fully acknowledged fire, which needs no repair). This method
    /// must NEVER be called in response to a `SequenceConflict` itself — a
    /// genuine double-writer conflict must surface un-resynced (pinned by
    /// `sequence_conflict_surfaces_without_advancing_or_retrying`); resyncing
    /// there would convert the crate's double-writer alarm into silence.
    pub(crate) fn reconcile_head_forward(&mut self, durable_head: u64) {
        if durable_head > self.sequence.current() {
            self.sequence = SequenceHead::from_head(durable_head);
        }
    }

    /// Reads this workflow's recorded history without appending new events.
    ///
    /// This narrow read seam lets replay contexts build a history cursor from the same store the
    /// recorder writes to while preserving the recorder as the only event append authority.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError::Store`] when the backing event store rejects the read.
    pub async fn read_history(&self) -> Result<Vec<Event>, DurabilityError> {
        self.store
            .read_history(&self.workflow_id)
            .await
            .map_err(Into::into)
    }

    /// Records workflow start.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_workflow_started(
        &mut self,
        recorded_at: DateTime<Utc>,
        start: WorkflowStartRecord,
    ) -> Result<(), DurabilityError> {
        let WorkflowStartRecord {
            workflow_type,
            input,
            run_id,
            parent_run_id,
            parent_workflow_id,
            package_version,
        } = start;
        let recorded_run_id = run_id.clone();
        self.append_with(recorded_at, |envelope| Event::WorkflowStarted {
            envelope,
            workflow_type,
            input,
            run_id,
            parent_run_id,
            parent_workflow_id,
            package_version,
        })
        .await?;
        self.run_id = Some(recorded_run_id);
        Ok(())
    }

    /// Records workflow start together with its validated initial search
    /// attributes in one atomic store append.
    ///
    /// Both events land in a single batch so a crash can never leave a durable
    /// `WorkflowStarted` without its initial attributes: recovery would
    /// otherwise resurrect a workflow that is invisible to every visibility
    /// query keyed on those attributes. Empty attribute maps record only the
    /// start event.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] when any attribute is unregistered or has a
    /// type that does not match `schema`, when the event store rejects the
    /// append, or when the sequence tracker cannot advance.
    pub async fn record_workflow_started_with_attributes(
        &mut self,
        recorded_at: DateTime<Utc>,
        start: WorkflowStartRecord,
        attributes: HashMap<String, SearchAttributeValue>,
        schema: &SearchAttributeSchema,
    ) -> Result<(), DurabilityError> {
        if attributes.is_empty() {
            return self.record_workflow_started(recorded_at, start).await;
        }
        for (name, value) in &attributes {
            schema.validate(name, value)?;
        }
        let WorkflowStartRecord {
            workflow_type,
            input,
            run_id,
            parent_run_id,
            parent_workflow_id,
            package_version,
        } = start;
        let recorded_run_id = run_id.clone();

        let started_envelope = self.next_envelope(recorded_at)?;
        let attributes_envelope = self.envelope_after(&started_envelope, recorded_at)?;
        let workflow_id = self.workflow_id.clone();
        let batch = [
            Event::WorkflowStarted {
                envelope: started_envelope,
                workflow_type,
                input,
                run_id,
                parent_run_id,
                parent_workflow_id,
                package_version,
            },
            Event::SearchAttributesUpdated {
                envelope: attributes_envelope,
                workflow_id,
                attributes,
            },
        ];
        self.durable_append(&batch).await?;
        self.run_id = Some(recorded_run_id);
        Ok(())
    }

    /// Records schedule creation in the schedule coordinator history.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_schedule_created(
        &mut self,
        recorded_at: DateTime<Utc>,
        schedule_id: ScheduleId,
        config: ScheduleConfig,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ScheduleCreated {
            envelope,
            schedule_id,
            config,
        })
        .await
    }

    /// Records schedule configuration update in the schedule coordinator history.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_schedule_updated(
        &mut self,
        recorded_at: DateTime<Utc>,
        schedule_id: ScheduleId,
        config: ScheduleConfig,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ScheduleUpdated {
            envelope,
            schedule_id,
            config,
        })
        .await
    }

    /// Records schedule pause in the schedule coordinator history.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_schedule_paused(
        &mut self,
        recorded_at: DateTime<Utc>,
        schedule_id: ScheduleId,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::SchedulePaused {
            envelope,
            schedule_id,
        })
        .await
    }

    /// Records schedule resume in the schedule coordinator history.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_schedule_resumed(
        &mut self,
        recorded_at: DateTime<Utc>,
        schedule_id: ScheduleId,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ScheduleResumed {
            envelope,
            schedule_id,
        })
        .await
    }

    /// Records schedule deletion in the schedule coordinator history.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_schedule_deleted(
        &mut self,
        recorded_at: DateTime<Utc>,
        schedule_id: ScheduleId,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ScheduleDeleted {
            envelope,
            schedule_id,
        })
        .await
    }

    /// Records a schedule-triggered workflow execution in the schedule coordinator history.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_schedule_triggered(
        &mut self,
        recorded_at: DateTime<Utc>,
        schedule_id: ScheduleId,
        workflow_id: WorkflowId,
        run_id: RunId,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ScheduleTriggered {
            envelope,
            schedule_id,
            workflow_id,
            run_id,
        })
        .await
    }

    /// Records workflow completion, then settles the workflow's live outbox
    /// rows (#253).
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append. A settle failure never fails the
    /// recorded terminal (see [`Self::settle_terminal_outbox_rows_nonfatal`]).
    pub async fn record_workflow_completed(
        &mut self,
        recorded_at: DateTime<Utc>,
        result: Payload,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::WorkflowCompleted {
            envelope,
            result,
        })
        .await?;
        self.settle_terminal_outbox_rows_nonfatal().await;
        Ok(())
    }

    /// Records terminal workflow failure, then settles the workflow's live
    /// outbox rows (#253).
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append. A settle failure never fails the
    /// recorded terminal (see [`Self::settle_terminal_outbox_rows_nonfatal`]).
    pub async fn record_workflow_failed(
        &mut self,
        recorded_at: DateTime<Utc>,
        error: WorkflowError,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::WorkflowFailed {
            envelope,
            error,
        })
        .await?;
        self.settle_terminal_outbox_rows_nonfatal().await;
        Ok(())
    }

    /// Records workflow cancellation, then settles the workflow's live outbox
    /// rows (#253).
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append. A settle failure never fails the
    /// recorded terminal (see [`Self::settle_terminal_outbox_rows_nonfatal`]).
    pub async fn record_workflow_cancelled(
        &mut self,
        recorded_at: DateTime<Utc>,
        reason: String,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::WorkflowCancelled {
            envelope,
            reason,
        })
        .await?;
        self.settle_terminal_outbox_rows_nonfatal().await;
        Ok(())
    }

    /// Records a workflow timeout terminal, then settles the workflow's live
    /// outbox rows (#253).
    ///
    /// The `timeout` descriptor is the closed-set kind token carried on
    /// [`Event::WorkflowTimedOut`] (a declared workflow timeout uses
    /// `"workflow"`); it is user-visible in `one_motion` output and replay
    /// terminals. Mirrors [`Self::record_workflow_cancelled`]: a settle failure
    /// never fails the recorded terminal.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the
    /// sequence tracker cannot advance after a successful append.
    pub async fn record_workflow_timed_out(
        &mut self,
        recorded_at: DateTime<Utc>,
        timeout: String,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::WorkflowTimedOut {
            envelope,
            timeout,
        })
        .await?;
        self.settle_terminal_outbox_rows_nonfatal().await;
        Ok(())
    }

    /// Settles the workflow's live (`Pending`/`Claimed`) outbox rows to
    /// `Cancelled` after a durably recorded workflow terminal (#253).
    ///
    /// The Recorder is the single-writer choke point every workflow terminal
    /// flows through, so this hook closes the live window in which a terminal
    /// workflow's staged dispatches could still be claimed and redelivered (the
    /// zombie-round incident). Deliberately NOT run for continue-as-new: the
    /// workflow continues, and the old run's stragglers are already neutralized
    /// by run-scoped completion threading (OBX-011).
    ///
    /// A settle failure is loud but non-fatal: the terminal append already
    /// committed (the workflow genuinely ended), and the boot sweep plus the
    /// stale-claim reconciler gate repair any rows this settle missed.
    async fn settle_terminal_outbox_rows_nonfatal(&self) {
        match self
            .store
            .settle_workflow_outbox_rows_cancelled(&self.workflow_id)
            .await
        {
            Ok(settled) if settled.is_empty() => {}
            Ok(settled) => {
                tracing::info!(
                    workflow_id = %self.workflow_id,
                    settled = settled.len(),
                    dispatch_keys = ?settled,
                    "settled outbox rows for terminal workflow"
                );
            }
            Err(error) => {
                tracing::error!(
                    workflow_id = %self.workflow_id,
                    %error,
                    "failed to settle outbox rows after recording a workflow terminal; \
                     the boot sweep or stale-claim reconciler will repair them"
                );
            }
        }
    }

    /// Records workflow continue-as-new as a terminal event for this run.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_workflow_continued_as_new(
        &mut self,
        recorded_at: DateTime<Utc>,
        input: Payload,
        workflow_type: Option<String>,
        parent_run_id: RunId,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::WorkflowContinuedAsNew {
            envelope,
            input,
            workflow_type,
            parent_run_id,
        })
        .await?;
        self.upsert_visibility_projection_nonfatal().await;
        Ok(())
    }

    /// Records a reopen of a failed run.
    ///
    /// The reopen supersedes the run's prior terminal event in the status
    /// projection (returning it to Running) and names the activities to
    /// re-dispatch on replay. It refreshes the visibility projection so the
    /// reopened workflow is listed as active again. This is the sole append
    /// path for [`Event::WorkflowReopened`]; like every recorded event it lands
    /// through the single-writer sequence discipline.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the
    /// sequence tracker cannot advance after a successful append.
    pub async fn record_workflow_reopened(
        &mut self,
        recorded_at: DateTime<Utc>,
        run_id: RunId,
        reopened: Vec<ActivityId>,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::WorkflowReopened {
            envelope,
            run_id,
            reopened,
        })
        .await?;
        self.upsert_visibility_projection_nonfatal().await;
        Ok(())
    }

    /// Records an operator pause of a running run (#204).
    ///
    /// A NON-terminal marker: it projects the run to `Paused` and refreshes the
    /// visibility projection so `describe`/`list` show Paused. It is the sole
    /// append path for [`Event::WorkflowPaused`] and, like every recorded event,
    /// lands through the single-writer sequence discipline.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the
    /// sequence tracker cannot advance after a successful append.
    pub async fn record_workflow_paused(
        &mut self,
        recorded_at: DateTime<Utc>,
        run_id: RunId,
        reason: Option<String>,
        operator: Option<String>,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::WorkflowPaused {
            envelope,
            run_id,
            reason,
            operator,
        })
        .await?;
        self.upsert_visibility_projection_nonfatal().await;
        Ok(())
    }

    /// Records an operator resume of a paused run (#204).
    ///
    /// Supersedes the run's prior [`Event::WorkflowPaused`] under the status
    /// projection (returning it to Running) and refreshes visibility. Sole append
    /// path for [`Event::WorkflowResumed`].
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the
    /// sequence tracker cannot advance after a successful append.
    pub async fn record_workflow_resumed(
        &mut self,
        recorded_at: DateTime<Utc>,
        run_id: RunId,
        operator: Option<String>,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::WorkflowResumed {
            envelope,
            run_id,
            operator,
        })
        .await?;
        self.upsert_visibility_projection_nonfatal().await;
        Ok(())
    }

    /// Records a validated search-attribute update for this workflow.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] when any attribute is unregistered or has a type that does not
    /// match `schema`, or when the event store rejects the append / sequence advance.
    pub async fn record_search_attributes_updated(
        &mut self,
        recorded_at: DateTime<Utc>,
        attributes: HashMap<String, SearchAttributeValue>,
        schema: &SearchAttributeSchema,
    ) -> Result<(), DurabilityError> {
        for (name, value) in &attributes {
            schema.validate(name, value)?;
        }
        let workflow_id = self.workflow_id.clone();
        self.append_with(recorded_at, |envelope| Event::SearchAttributesUpdated {
            envelope,
            workflow_id,
            attributes,
        })
        .await?;
        self.upsert_visibility_projection_nonfatal().await;
        Ok(())
    }

    /// Records activity scheduling, stamping the `task_queue` and OPTIONAL `node` affinity the
    /// activity dispatches to so reopen/recovery re-targets the **same** pool and node (NSTQ-3 /
    /// NODE-3).
    ///
    /// No SDK-level task-queue or node selection exists yet (NSTQ-4 / NODE-4), so the
    /// single-schedule engine seam passes the named `"default"` task queue and `None` node — the
    /// genuine current values, not a shim.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_activity_scheduled(
        &mut self,
        recorded_at: DateTime<Utc>,
        activity_id: ActivityId,
        activity_type: String,
        input: Payload,
        task_queue: String,
        node: Option<String>,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ActivityScheduled {
            envelope,
            activity_id,
            activity_type,
            input,
            task_queue,
            node,
        })
        .await
    }

    /// Records that the engine DISPATCHED an activity attempt to its task queue.
    ///
    /// Not "a worker started it": every caller invokes this BEFORE the dispatch
    /// reaches the dispatcher, so no worker has been selected when the append
    /// happens. See [`Event::ActivityStarted`] for the full account and for the
    /// pending rename.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_activity_started(
        &mut self,
        recorded_at: DateTime<Utc>,
        activity_id: ActivityId,
        attempt: u32,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ActivityStarted {
            envelope,
            activity_id,
            // NOI-0: the genuine one-based attempt this start belongs to, threaded from the dispatch
            // seam so `ActivityStarted` and its terminal share one `(workflow, activity, attempt)`.
            attempt,
        })
        .await
    }

    /// Records that recovery retained a dangling activity attempt and offered
    /// that same execution identity to the worker adoption path.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the
    /// sequence tracker cannot advance after a successful append.
    pub async fn record_activity_adoption_offered(
        &mut self,
        recorded_at: DateTime<Utc>,
        activity_id: ActivityId,
        attempt: u32,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ActivityAdoptionOffered {
            envelope,
            activity_id,
            attempt,
        })
        .await
    }

    /// Records successful activity completion.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_activity_completed(
        &mut self,
        recorded_at: DateTime<Utc>,
        activity_id: ActivityId,
        result: Payload,
        attempt: u32,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ActivityCompleted {
            envelope,
            activity_id,
            result,
            // NOI-0: the genuine one-based attempt that produced this completion, consistent with the
            // `ActivityStarted` of the same attempt.
            attempt,
        })
        .await
    }

    /// Records failed activity attempt.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_activity_failed(
        &mut self,
        recorded_at: DateTime<Utc>,
        activity_id: ActivityId,
        error: ActivityError,
        attempt: u32,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ActivityFailed {
            envelope,
            activity_id,
            error,
            attempt,
        })
        .await
    }

    /// Records the R5 warning that an ADVISORY activity spent its whole
    /// attempt budget and failed for good.
    ///
    /// This ACCOMPANIES the activity's honest terminal `ActivityFailed`; it
    /// never replaces it. The advisory class suppresses the FAULT, never the
    /// RECORD — the side channel really did fail, and history says so twice:
    /// once as the activity's own failure, once as the visible warning on the
    /// run (RUNTIME-OPERATIONS.md R5).
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the
    /// sequence tracker cannot advance after a successful append.
    pub async fn record_activity_advisory_exhausted(
        &mut self,
        recorded_at: DateTime<Utc>,
        activity_id: ActivityId,
        activity_type: String,
        reason: String,
        attempt: u32,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ActivityAdvisoryExhausted {
            envelope,
            activity_id,
            activity_type,
            reason,
            attempt,
        })
        .await
    }

    /// Records a durable post-refusal fallback queue hop.
    ///
    /// Retry routing calls this only through the workflow recorder held by
    /// `RetryRecorderSeam`; no routing component appends this event directly.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the
    /// sequence tracker cannot advance after a successful append.
    pub async fn record_activity_fallback_routed(
        &mut self,
        recorded_at: DateTime<Utc>,
        activity_id: ActivityId,
        attempt: u32,
        from_task_queue: String,
        to_task_queue: String,
        fallback_index: u32,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ActivityFallbackRouted {
            envelope,
            activity_id,
            attempt,
            from_task_queue,
            to_task_queue,
            fallback_index,
        })
        .await
    }

    /// Records activity cancellation.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_activity_cancelled(
        &mut self,
        recorded_at: DateTime<Utc>,
        activity_id: ActivityId,
        attempt: u32,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ActivityCancelled {
            envelope,
            activity_id,
            // NOI-0: the genuine one-based attempt that was cancelled, consistent with the
            // `ActivityStarted` of the same attempt.
            attempt,
        })
        .await
    }

    /// Records activity cancellation for a fan-out ordinal and settles its outbox row.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event append fails, sequence tracking cannot advance, or
    /// the outbox-aware store rejects the cancellation settlement.
    pub async fn record_activity_cancelled_and_settle_outbox(
        &mut self,
        recorded_at: DateTime<Utc>,
        ordinal: u64,
        attempt: u32,
    ) -> Result<(), DurabilityError> {
        self.record_activity_cancelled(
            recorded_at,
            ActivityId::from_sequence_position(ordinal),
            attempt,
        )
        .await?;
        let dispatch_key = OutboxRow::dispatch_key_for(&self.workflow_id, ordinal);
        self.store
            .settle_outbox_row_cancelled(&dispatch_key)
            .await
            .map_err(Into::into)
    }

    /// Records timer scheduling, returning the appended `TimerStarted`'s
    /// workflow-history sequence — the arming's identity, which the caller
    /// passes to `schedule_timer` so the durable row names exactly this
    /// arming (`TimerEntry::armed_seq`).
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_timer_started(
        &mut self,
        recorded_at: DateTime<Utc>,
        timer_id: TimerId,
        fire_at: DateTime<Utc>,
    ) -> Result<u64, DurabilityError> {
        let envelope = self.next_envelope(recorded_at)?;
        let armed_seq = envelope.seq;
        self.append_one(Event::TimerStarted {
            envelope,
            timer_id,
            fire_at,
        })
        .await?;
        Ok(armed_seq)
    }

    /// Records timer firing.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_timer_fired(
        &mut self,
        recorded_at: DateTime<Utc>,
        timer_id: TimerId,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::TimerFired {
            envelope,
            timer_id,
        })
        .await
    }

    /// Records timer cancellation.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_timer_cancelled(
        &mut self,
        recorded_at: DateTime<Utc>,
        timer_id: TimerId,
        cause: TimerCancelCause,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::TimerCancelled {
            envelope,
            timer_id,
            cause,
        })
        .await
    }

    /// Records a `with_timeout` terminal outcome.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_with_timeout_completed(
        &mut self,
        recorded_at: DateTime<Utc>,
        timer_id: TimerId,
        outcome: WithTimeoutOutcome,
        result: Option<Payload>,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::WithTimeoutCompleted {
            envelope,
            timer_id,
            outcome,
            result,
        })
        .await
    }

    /// Records that a selected worker accepted one activity attempt (WA-010).
    ///
    /// An asynchronous arrival like a signal: the server's handoff seam holds
    /// no Recorder of its own, so the lease reaches this one through
    /// `Engine::record_activity_lease` and serialises with every other append
    /// under the same lock. `worker` is durable names only — see
    /// [`WorkerAttribution`].
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_activity_leased(
        &mut self,
        recorded_at: DateTime<Utc>,
        activity_id: ActivityId,
        attempt: u32,
        worker: WorkerAttribution,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ActivityLeased {
            envelope,
            activity_id,
            attempt,
            worker,
        })
        .await
    }

    /// Records signal delivery.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_signal_received(
        &mut self,
        recorded_at: DateTime<Utc>,
        name: String,
        payload: Payload,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::SignalReceived {
            envelope,
            name,
            payload,
        })
        .await
    }

    /// Records a signal sent by this workflow.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_signal_sent(
        &mut self,
        recorded_at: DateTime<Utc>,
        target_workflow_id: WorkflowId,
        name: String,
        payload: Payload,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::SignalSent {
            envelope,
            target_workflow_id,
            name,
            payload,
        })
        .await
    }

    /// Records child workflow start.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_child_workflow_started(
        &mut self,
        recorded_at: DateTime<Utc>,
        child_workflow_id: WorkflowId,
        workflow_type: String,
        input: Payload,
        package_version: PackageVersion,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ChildWorkflowStarted {
            envelope,
            child_workflow_id,
            workflow_type,
            input,
            package_version,
        })
        .await
    }

    /// Records child workflow completion.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_child_workflow_completed(
        &mut self,
        recorded_at: DateTime<Utc>,
        child_workflow_id: WorkflowId,
        result: Payload,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ChildWorkflowCompleted {
            envelope,
            child_workflow_id,
            result,
        })
        .await
    }

    /// Records child workflow failure.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_child_workflow_failed(
        &mut self,
        recorded_at: DateTime<Utc>,
        child_workflow_id: WorkflowId,
        error: WorkflowError,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::ChildWorkflowFailed {
            envelope,
            child_workflow_id,
            error,
        })
        .await
    }

    /// Records an engine-side cadence fire for a workloop (workloop brief
    /// R1.4/R4.3): the dead-man clock tick, appended through this one
    /// Recorder exactly like any other asynchronous arrival.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_cadence_fired(
        &mut self,
        recorded_at: DateTime<Utc>,
        window_seq: u64,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::CadenceFired {
            envelope,
            window_seq,
        })
        .await
    }

    /// Records a workloop iteration close (R3.1/R3.3): the iteration's routes
    /// and the health samples they land against the loop's invariants.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_iteration_closed(
        &mut self,
        recorded_at: DateTime<Utc>,
        routes: Vec<String>,
        health_samples: Vec<aion_core::HealthSample>,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::IterationClosed {
            envelope,
            routes,
            health_samples,
        })
        .await
    }

    /// Records a workloop retirement (R2.5) ATOMICALLY with its terminal: one
    /// batch appends `LoopRetired { reason }` followed by
    /// `WorkflowCompleted { result }`, so a crash can never leave a retirement
    /// marker without its terminal (an intentional stop reading as a hang) or
    /// a completed loop without its declared reason (an incident reading where
    /// a decommission happened). Terminal outbox rows are settled afterwards,
    /// exactly as [`Self::record_workflow_completed`] settles them.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append. A settle failure never fails the
    /// recorded terminal (see [`Self::settle_terminal_outbox_rows_nonfatal`]).
    pub async fn record_loop_retired(
        &mut self,
        recorded_at: DateTime<Utc>,
        reason: String,
        result: Payload,
    ) -> Result<(), DurabilityError> {
        let retired_envelope = self.next_envelope(recorded_at)?;
        let completed_envelope = self.envelope_after(&retired_envelope, recorded_at)?;
        let batch = [
            Event::LoopRetired {
                envelope: retired_envelope,
                reason,
            },
            Event::WorkflowCompleted {
                envelope: completed_envelope,
                result,
            },
        ];
        self.durable_append(&batch).await?;
        // The retirement is a genuine terminal on the loop's current
        // generation: its one row must read Completed with its end the moment
        // the append is durable, exactly as every other terminal recorder
        // projects — a retired loop that still listed as Running was a row
        // only the next boot reconcile righted.
        self.upsert_visibility_projection_nonfatal().await;
        self.settle_terminal_outbox_rows_nonfatal().await;
        Ok(())
    }

    /// Records a detached top-level hatch (R13.1) in the HATCHING workflow's
    /// history — record-then-start, so replay re-mints the same deterministic
    /// `child_workflow_id` and a crash between record and start leaves a
    /// repairable record instead of an unrecorded orphan.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_workflow_hatched(
        &mut self,
        recorded_at: DateTime<Utc>,
        child_workflow_id: WorkflowId,
        key: String,
    ) -> Result<(), DurabilityError> {
        self.append_with(recorded_at, |envelope| Event::WorkflowHatched {
            envelope,
            child_workflow_id,
            key,
        })
        .await
    }

    /// Records the ONE alarm path (R4.2): the invariant is not confirmed
    /// held, with cause as a field — never a separate alarm channel.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_invariant_unconfirmed(
        &mut self,
        recorded_at: DateTime<Utc>,
        alarm: aion_core::InvariantAlarm,
    ) -> Result<(), DurabilityError> {
        let aion_core::InvariantAlarm {
            invariant,
            cause,
            window_seq,
            last_confirmed_at,
            consecutive_unconfirmed,
        } = alarm;
        self.append_with(recorded_at, |envelope| Event::InvariantUnconfirmed {
            envelope,
            invariant,
            cause,
            window_seq,
            last_confirmed_at,
            consecutive_unconfirmed,
        })
        .await
    }

    fn next_envelope(&self, recorded_at: DateTime<Utc>) -> Result<EventEnvelope, DurabilityError> {
        let seq = self
            .sequence
            .next_seq()
            .ok_or_else(|| DurabilityError::HistoryShape {
                reason: format!(
                    "sequence head overflow advancing {} by 1",
                    self.sequence.current()
                ),
            })?;
        Ok(EventEnvelope {
            seq,
            recorded_at,
            workflow_id: self.workflow_id.clone(),
        })
    }

    fn envelope_after(
        &self,
        previous: &EventEnvelope,
        recorded_at: DateTime<Utc>,
    ) -> Result<EventEnvelope, DurabilityError> {
        let seq = previous
            .seq
            .checked_add(1)
            .ok_or_else(|| DurabilityError::HistoryShape {
                reason: format!("sequence head overflow advancing {} by 1", previous.seq),
            })?;
        Ok(EventEnvelope {
            seq,
            recorded_at,
            workflow_id: self.workflow_id.clone(),
        })
    }

    async fn append_with(
        &mut self,
        recorded_at: DateTime<Utc>,
        build_event: impl FnOnce(EventEnvelope) -> Event,
    ) -> Result<(), DurabilityError> {
        let envelope = self.next_envelope(recorded_at)?;
        self.append_one(build_event(envelope)).await
    }

    async fn append_one(&mut self, event: Event) -> Result<(), DurabilityError> {
        self.durable_append(std::slice::from_ref(&event)).await?;
        if !reprojects_whole_row(&event) {
            self.touch_visibility_nonfatal(&event).await;
        }
        Ok(())
    }

    /// THE one append: every event this recorder writes goes through here.
    ///
    /// A store error other than [`StoreError::SequenceConflict`] is
    /// INDETERMINATE — the write may have landed before the error came back
    /// (aion#192: a shard-actor timeout on a `TimerStarted` that had already
    /// been applied left the tracked head one below the durable head, the
    /// workflow process died on the surfaced error, and every later append —
    /// cancel included — conflicted until a restart rebuilt this recorder from
    /// the store). So the outcome is resolved against the store BEFORE
    /// anything is surfaced, by [`Self::resolve_indeterminate_append`]. A
    /// `SequenceConflict` from the store itself is surfaced as-is, with no
    /// read and no reconciliation: that is the double-writer alarm, and
    /// resyncing on it would silence the alarm (aion#145's other side).
    ///
    /// # Errors
    ///
    /// Returns the store's refusal when the append demonstrably did not land
    /// (a fenced [`StoreError::NotOwner`] is surfaced typed, with no read),
    /// [`StoreError::SequenceConflict`] when something ELSE sits at the
    /// expected sequence, and [`DurabilityError::HistoryShape`] if the tracker
    /// cannot advance.
    async fn durable_append(&mut self, batch: &[Event]) -> Result<(), DurabilityError> {
        let expected_seq = self.sequence.current();
        match self
            .store
            .append(self.write_token, &self.workflow_id, batch, expected_seq)
            .await
        {
            Ok(()) => {}
            Err(conflict @ aion_store::StoreError::SequenceConflict { .. }) => {
                return Err(DurabilityError::Store(conflict));
            }
            // A fenced quorum write is a DEFINITE did-not-land — the store
            // says this node is not the owner, and the typed variant is a
            // retryable routing signal the completion-retry and fence paths
            // match on. It never reaches the resolving read: that read from a
            // just-fenced node is the one most likely to fail (and would fold
            // the variant into `Backend`), and a stale replica could answer
            // with a rogue and mint a spurious `SequenceConflict` out of a
            // routing condition.
            Err(fenced @ aion_store::StoreError::NotOwner { .. }) => {
                return Err(DurabilityError::Store(fenced));
            }
            Err(error) => {
                self.resolve_indeterminate_append(batch, expected_seq, error)
                    .await?;
            }
        }
        self.sequence.mark_append_success(batch.len())
    }

    /// Decide what an errored append actually did, by reading the store.
    ///
    /// Reads everything at or past `expected_seq + 1` (a range read, not the
    /// whole history) and rules:
    ///
    /// - nothing there → the append did not land; the ORIGINAL error is
    ///   returned unchanged and the tracked head is untouched, so the caller's
    ///   own retry path proceeds exactly as before;
    /// - exactly this batch there (same envelopes, same payloads — `Event` is
    ///   compared by value, and under single-writer discipline nothing else
    ///   can produce these events at these sequences) → the append LANDED and
    ///   its acknowledgement was lost; said at WARN so it can be correlated
    ///   with the store's own timeout narration, and reported as success so
    ///   the tracker advances by the batch;
    /// - anything else there → a genuine [`StoreError::SequenceConflict`],
    ///   surfaced un-resynced.
    ///
    /// If the resolving read itself fails, the ORIGINAL append error is
    /// returned with the read failure attached in its message; the head stays
    /// untouched, because nothing was learned.
    ///
    /// Held under the recorder's lock by construction (`&mut self`): no other
    /// append of this workflow can interleave with the read.
    async fn resolve_indeterminate_append(
        &self,
        batch: &[Event],
        expected_seq: u64,
        error: aion_store::StoreError,
    ) -> Result<(), DurabilityError> {
        let Some(from_seq) = expected_seq.checked_add(1) else {
            return Err(DurabilityError::HistoryShape {
                reason: format!("sequence head overflow resolving an append at {expected_seq}"),
            });
        };
        let landed = match self
            .store
            .read_history_from(&self.workflow_id, from_seq)
            .await
        {
            Ok(landed) => landed,
            Err(read_error) => {
                return Err(DurabilityError::Store(aion_store::StoreError::Backend(
                    format!(
                        "{error}; the append's outcome could not be resolved either: {read_error}"
                    ),
                )));
            }
        };
        if landed.is_empty() {
            return Err(DurabilityError::Store(error));
        }
        let durable_head = landed.iter().map(Event::seq).max().unwrap_or(expected_seq);
        if landed.len() == batch.len() && landed == batch {
            tracing::warn!(
                workflow_id = %self.workflow_id,
                run_id = ?self.run_id,
                landed_seq_from = from_seq,
                landed_seq_to = durable_head,
                store_error = %error,
                "append landed but its acknowledgement was lost; resolved against the store and \
                 counted as recorded"
            );
            return Ok(());
        }
        Err(DurabilityError::Store(
            aion_store::StoreError::SequenceConflict {
                expected: expected_seq,
                found: durable_head,
            },
        ))
    }

    /// Projects and upserts THE workflow's row — always for the history's
    /// CURRENT generation, never for the generation this recorder happens to
    /// write for. The two coincide for a live recorder; they differ only for
    /// a straggler (a lease or heartbeat event landing through a superseded
    /// generation's recorder after the boundary replaced its row), and a
    /// write may heal a row forward, never regress it to an older run.
    async fn upsert_visibility_projection(&self) -> Result<(), DurabilityError> {
        let Some(visibility) = &self.visibility else {
            return Ok(());
        };
        let history = self.store.read_history(&self.workflow_id).await?;
        let current = crate::lifecycle::visibility::current_run_id(&history)
            .unwrap_or_else(|| visibility.run_id.clone());
        let record = project_visibility(&history, &current).ok_or_else(|| {
            DurabilityError::HistoryShape {
                reason: String::from(
                    "workflow history has no WorkflowStarted event for visibility projection",
                ),
            }
        })?;
        visibility.store.record_visibility(record).await?;
        Ok(())
    }

    async fn upsert_visibility_projection_nonfatal(&self) {
        if let Err(error) = self.upsert_visibility_projection().await {
            self.warn_visibility_failure(&error, "visibility upsert failed after durable append");
        }
    }

    /// Advances the row's `updated_at` to the event's instant after an append
    /// that re-projects nothing else, and applies the event to the row's
    /// outstanding leases through the SAME transition the whole-history fold
    /// uses ([`aion_core::apply_lease_transition`]) — so a lease or an
    /// attempt terminal moves `current_worker` on the row exactly as it would
    /// move a summary built from history (WA-010 R4). A run whose row is
    /// missing (its start-time upsert failed and reconciliation has not yet
    /// repaired it) gets the whole row projected instead, so the touch heals
    /// the gap it would otherwise widen.
    async fn touch_visibility(&self, event: &Event) -> Result<(), DurabilityError> {
        let Some(visibility) = &self.visibility else {
            return Ok(());
        };
        match visibility.store.get_visibility(&self.workflow_id).await? {
            // A row still carrying another generation's run is stale (the
            // boundary's replace has not landed yet): touching it would
            // advance the wrong generation, so project the whole row for the
            // generation this recorder writes for instead.
            Some(mut record) if record.run_id == visibility.run_id => {
                record.updated_at = *event.recorded_at();
                // The stamp is INDUCTIVE: a touch may move the row to this
                // event only if the row provably folded everything before it
                // (it stood at the previous seq). A projection that failed
                // (tolerated, warned) leaves a gap the touch must never paper
                // over — a row stamped past an unfolded lifecycle event would
                // be trusted by the next boot with a status that is a lie.
                // Left behind the head, the row is re-derived instead.
                if record.head_seq != 0 && event.seq() == record.head_seq.saturating_add(1) {
                    record.head_seq = event.seq();
                }
                aion_core::apply_lease_transition(&mut record.outstanding_leases, event);
                visibility.store.record_visibility(record).await?;
                Ok(())
            }
            Some(_) | None => self.upsert_visibility_projection().await,
        }
    }

    async fn touch_visibility_nonfatal(&self, event: &Event) {
        if let Err(error) = self.touch_visibility(event).await {
            self.warn_visibility_failure(&error, "visibility touch failed after durable append");
        }
    }

    fn warn_visibility_failure(&self, error: &DurabilityError, what: &'static str) {
        let run_id = self
            .visibility
            .as_ref()
            .map(|visibility| &visibility.run_id);
        tracing::warn!(
            workflow_id = %self.workflow_id,
            run_id = run_id.map(ToString::to_string).as_deref().unwrap_or("unknown"),
            error = %error,
            "{what}; crash-consistency window remains until reconciliation repairs visibility"
        );
    }
}

/// Whether `event` is one whose recording path re-projects the whole
/// visibility row afterwards (the lifecycle handlers and this Recorder's own
/// lifecycle record functions do), so the per-append `updated_at` touch would
/// only be overwritten by it.
const fn reprojects_whole_row(event: &Event) -> bool {
    matches!(
        event,
        Event::WorkflowStarted { .. }
            | Event::WorkflowCompleted { .. }
            | Event::WorkflowFailed { .. }
            | Event::WorkflowCancelled { .. }
            | Event::WorkflowTimedOut { .. }
            | Event::WorkflowContinuedAsNew { .. }
            | Event::WorkflowReopened { .. }
            | Event::WorkflowPaused { .. }
            | Event::WorkflowResumed { .. }
            | Event::SearchAttributesUpdated { .. }
    )
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::future::Future;
    use std::sync::Arc;

    use aion_core::RunId;
    use aion_core::WorkflowListRequest;
    use aion_core::{
        Event, Payload, SearchAttributeError, SearchAttributeSchema, SearchAttributeType,
        SearchAttributeValue, TimerId,
    };
    use aion_store::visibility::{VisibilityPage, VisibilityRecord, VisibilityStore};
    use aion_store::{
        InMemoryStore, ReadableEventStore, StoreError, WritableEventStore, WriteToken,
    };
    use async_trait::async_trait;
    use chrono::{DateTime, Utc};
    use serde_json::json;

    use super::Recorder;
    use crate::durability::DurabilityError;

    #[derive(Debug)]
    struct FailingVisibilityStore;

    #[async_trait]
    impl VisibilityStore for FailingVisibilityStore {
        async fn record_visibility(&self, _record: VisibilityRecord) -> Result<(), StoreError> {
            Err(StoreError::Backend(String::from("visibility unavailable")))
        }

        async fn get_visibility(
            &self,
            _workflow_id: &aion_core::WorkflowId,
        ) -> Result<Option<VisibilityRecord>, StoreError> {
            Err(StoreError::Backend(String::from("visibility unavailable")))
        }

        async fn remove_visibility(
            &self,
            _workflow_id: &aion_core::WorkflowId,
            _run_id: &RunId,
        ) -> Result<bool, StoreError> {
            Err(StoreError::Backend(String::from("visibility unavailable")))
        }

        async fn list_workflows(
            &self,
            _request: &WorkflowListRequest,
        ) -> Result<VisibilityPage, StoreError> {
            Err(StoreError::Backend(String::from("visibility unavailable")))
        }
    }

    fn workflow_id(value: u128) -> aion_core::WorkflowId {
        aion_core::WorkflowId::new(uuid::Uuid::from_u128(value))
    }

    fn recorded_at(offset_seconds: i64) -> DateTime<Utc> {
        DateTime::from_timestamp(1_700_000_000 + offset_seconds, 0).unwrap_or_default()
    }

    fn payload(label: &str) -> Result<Payload, Box<dyn std::error::Error>> {
        Ok(Payload::from_json(&json!({ "label": label }))?)
    }

    fn workflow_started(
        seq: u64,
        workflow_id: &aion_core::WorkflowId,
    ) -> Result<Event, Box<dyn std::error::Error>> {
        Ok(Event::WorkflowStarted {
            envelope: aion_core::EventEnvelope {
                seq,
                recorded_at: recorded_at(i64::try_from(seq)?),
                workflow_id: workflow_id.clone(),
            },
            workflow_type: String::from("checkout"),
            input: payload("workflow-input")?,
            run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
            parent_run_id: None,
            parent_workflow_id: None,
            package_version: aion_core::PackageVersion::new("a".repeat(64)),
        })
    }

    #[tokio::test]
    async fn recorder_advances_expected_sequence_between_appends()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(1);
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone());

        recorder
            .record_workflow_started(
                recorded_at(1),
                super::WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("input")?,
                    run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
            )
            .await?;
        recorder
            .record_workflow_completed(recorded_at(2), payload("result")?)
            .await?;

        let history = store.read_history(&workflow_id).await?;
        assert_eq!(history[0].seq(), 1);
        assert_eq!(history[1].seq(), 2);
        assert_eq!(recorder.current_head(), 2);
        Ok(())
    }

    #[tokio::test]
    async fn records_workflow_reopened_and_projects_running()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(1);
        let run_id = aion_core::RunId::new(uuid::Uuid::from_u128(1));
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone());

        recorder
            .record_workflow_started(
                recorded_at(1),
                super::WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("input")?,
                    run_id: run_id.clone(),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
            )
            .await?;
        recorder
            .record_workflow_failed(
                recorded_at(2),
                aion_core::WorkflowError {
                    message: String::from("transient"),
                    details: None,
                },
            )
            .await?;
        recorder
            .record_workflow_reopened(
                recorded_at(3),
                run_id,
                vec![aion_core::ActivityId::from_sequence_position(2)],
            )
            .await?;

        let history = store.read_history(&workflow_id).await?;
        assert_eq!(history.len(), 3);
        assert_eq!(history[2].seq(), 3);
        assert_eq!(recorder.current_head(), 3);
        assert!(matches!(history[2], Event::WorkflowReopened { .. }));
        assert_eq!(
            aion_core::status_from_events(&history),
            aion_core::WorkflowStatus::Running,
            "a recorded reopen returns the failed workflow to Running"
        );
        Ok(())
    }

    #[tokio::test]
    async fn records_workflow_continued_as_new_terminal_event()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(2);
        let parent_run_id = aion_core::RunId::new(uuid::Uuid::from_u128(20));
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone());
        let continued_at = recorded_at(2);
        let continued_input = payload("continued-input")?;
        let workflow_type = Some(String::from("checkout-v2"));

        recorder
            .record_workflow_started(
                recorded_at(1),
                super::WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("input")?,
                    run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
            )
            .await?;
        recorder
            .record_workflow_continued_as_new(
                continued_at,
                continued_input.clone(),
                workflow_type.clone(),
                parent_run_id.clone(),
            )
            .await?;

        let history = store.read_history(&workflow_id).await?;
        assert_eq!(history.len(), 2);
        assert_eq!(history[0].seq(), 1);
        assert_eq!(history[1].seq(), 2);
        match &history[1] {
            Event::WorkflowContinuedAsNew {
                envelope,
                input,
                workflow_type: recorded_workflow_type,
                parent_run_id: recorded_parent_run_id,
            } => {
                assert_eq!(envelope.recorded_at, continued_at);
                assert_eq!(input, &continued_input);
                assert_eq!(recorded_workflow_type, &workflow_type);
                assert_eq!(recorded_parent_run_id, &parent_run_id);
            }
            other => return Err(format!("expected WorkflowContinuedAsNew, got {other:?}").into()),
        }
        assert_eq!(recorder.current_head(), 2);
        Ok(())
    }

    #[tokio::test]
    async fn records_workflow_timed_out_terminal_and_projects_timed_out()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(41);
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone());

        recorder
            .record_workflow_started(
                recorded_at(1),
                super::WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("input")?,
                    run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
            )
            .await?;
        recorder
            .record_workflow_timed_out(recorded_at(2), String::from("workflow"))
            .await?;

        let history = store.read_history(&workflow_id).await?;
        match history.as_slice() {
            [
                Event::WorkflowStarted { .. },
                Event::WorkflowTimedOut { envelope, timeout },
            ] => {
                assert_eq!(envelope.seq, 2);
                assert_eq!(timeout, "workflow");
            }
            other => return Err(format!("expected started then timed out, found {other:?}").into()),
        }
        assert_eq!(
            aion_core::status_from_events(&history),
            aion_core::WorkflowStatus::TimedOut,
            "a recorded WorkflowTimedOut projects to TimedOut"
        );
        assert_eq!(recorder.current_head(), 2);
        Ok(())
    }

    #[tokio::test]
    async fn records_validated_search_attributes_updated_event()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(7);
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone());
        let mut schema = SearchAttributeSchema::new();
        schema.register("customer_id", SearchAttributeType::String)?;
        schema.register("attempt", SearchAttributeType::Int)?;
        let attributes = HashMap::from([
            (
                String::from("customer_id"),
                SearchAttributeValue::String(String::from("customer-123")),
            ),
            (String::from("attempt"), SearchAttributeValue::Int(2)),
        ]);

        recorder
            .record_workflow_started(
                recorded_at(1),
                super::WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("input")?,
                    run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
            )
            .await?;
        recorder
            .record_search_attributes_updated(recorded_at(2), attributes.clone(), &schema)
            .await?;

        let history = store.read_history(&workflow_id).await?;
        match history.as_slice() {
            [
                Event::WorkflowStarted { .. },
                Event::SearchAttributesUpdated {
                    envelope,
                    workflow_id: recorded_workflow_id,
                    attributes: stored_attributes,
                },
            ] => {
                assert_eq!(envelope.seq, 2);
                assert_eq!(recorded_workflow_id, &workflow_id);
                assert_eq!(stored_attributes, &attributes);
            }
            other => {
                return Err(
                    format!("expected started then search attributes, found {other:?}").into(),
                );
            }
        }
        assert_eq!(recorder.current_head(), 2);
        Ok(())
    }

    #[tokio::test]
    async fn started_with_attributes_appends_both_events_in_one_atomic_batch()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(11);
        let run_id = aion_core::RunId::new(uuid::Uuid::from_u128(1));
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone());
        let mut schema = SearchAttributeSchema::new();
        schema.register("aion.namespace", SearchAttributeType::String)?;
        let attributes = HashMap::from([(
            String::from("aion.namespace"),
            SearchAttributeValue::String(String::from("tenant-a")),
        )]);

        recorder
            .record_workflow_started_with_attributes(
                recorded_at(1),
                super::WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("input")?,
                    run_id: run_id.clone(),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
                attributes.clone(),
                &schema,
            )
            .await?;

        let history = store.read_history(&workflow_id).await?;
        match history.as_slice() {
            [
                Event::WorkflowStarted {
                    envelope: started_envelope,
                    run_id: started_run_id,
                    ..
                },
                Event::SearchAttributesUpdated {
                    envelope: attributes_envelope,
                    workflow_id: recorded_workflow_id,
                    attributes: stored_attributes,
                },
            ] => {
                assert_eq!(started_envelope.seq, 1);
                assert_eq!(started_run_id, &run_id);
                assert_eq!(attributes_envelope.seq, 2);
                assert_eq!(recorded_workflow_id, &workflow_id);
                assert_eq!(stored_attributes, &attributes);
            }
            other => {
                return Err(
                    format!("expected started then search attributes, found {other:?}").into(),
                );
            }
        }
        assert_eq!(recorder.current_head(), 2);

        // The recorder must keep appending correctly after the two-event batch.
        recorder
            .record_workflow_completed(recorded_at(3), payload("result")?)
            .await?;
        let history = store.read_history(&workflow_id).await?;
        assert_eq!(history.len(), 3);
        assert_eq!(history[2].seq(), 3);
        Ok(())
    }

    #[tokio::test]
    async fn started_with_invalid_attributes_appends_nothing()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(12);
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone());
        let schema = SearchAttributeSchema::new();
        let attributes = HashMap::from([(
            String::from("aion.namespace"),
            SearchAttributeValue::String(String::from("tenant-a")),
        )]);

        let result = recorder
            .record_workflow_started_with_attributes(
                recorded_at(1),
                super::WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("input")?,
                    run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
                attributes,
                &schema,
            )
            .await;

        assert!(matches!(
            result,
            Err(DurabilityError::SearchAttribute(
                SearchAttributeError::UnregisteredAttribute { name }
            )) if name == "aion.namespace"
        ));
        assert!(store.read_history(&workflow_id).await?.is_empty());
        assert_eq!(recorder.current_head(), 0);
        Ok(())
    }

    #[tokio::test]
    async fn started_with_empty_attributes_appends_only_the_start_event()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(13);
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone());
        let schema = SearchAttributeSchema::new();

        recorder
            .record_workflow_started_with_attributes(
                recorded_at(1),
                super::WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("input")?,
                    run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
                HashMap::new(),
                &schema,
            )
            .await?;

        let history = store.read_history(&workflow_id).await?;
        assert!(matches!(
            history.as_slice(),
            [Event::WorkflowStarted { .. }]
        ));
        assert_eq!(recorder.current_head(), 1);
        Ok(())
    }

    #[tokio::test]
    async fn invalid_search_attributes_return_error_without_appending()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(8);
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone());
        let mut schema = SearchAttributeSchema::new();
        schema.register("attempt", SearchAttributeType::Int)?;

        recorder
            .record_workflow_started(
                recorded_at(1),
                super::WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("input")?,
                    run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
            )
            .await?;
        let attributes = HashMap::from([(
            String::from("attempt"),
            SearchAttributeValue::String(String::from("two")),
        )]);
        let error = recorder
            .record_search_attributes_updated(recorded_at(2), attributes, &schema)
            .await;

        match error {
            Err(DurabilityError::SearchAttribute(SearchAttributeError::TypeMismatch {
                name,
                expected,
                actual,
            })) => {
                assert_eq!(name, "attempt");
                assert_eq!(expected, SearchAttributeType::Int);
                assert_eq!(actual, SearchAttributeType::String);
            }
            Err(other) => {
                return Err(format!("expected search attribute error, got {other:?}").into());
            }
            Ok(()) => return Err("expected search attribute validation error".into()),
        }
        assert_eq!(recorder.current_head(), 1);
        assert_eq!(store.read_history(&workflow_id).await?.len(), 1);
        Ok(())
    }

    /// #214: after a workloop boundary the predecessor's row is closed and the
    /// successor's row is the one running row.
    #[tokio::test]
    async fn a_workloop_boundary_closes_the_predecessor_row_and_opens_the_successors()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(214);
        let first = aion_core::RunId::new(uuid::Uuid::from_u128(2141));
        let second = aion_core::RunId::new(uuid::Uuid::from_u128(2142));
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone())
            .with_visibility(first.clone(), store.clone());
        let start =
            |run_id: aion_core::RunId| -> Result<super::WorkflowStartRecord, Box<dyn std::error::Error>> {
                Ok(super::WorkflowStartRecord {
                    workflow_type: String::from("disk_reaper"),
                    input: aion_core::Payload::from_json(&serde_json::json!({}))?,
                    run_id,
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                })
            };
        let opened = chrono::Utc::now();
        recorder
            .record_workflow_started(opened, start(first.clone())?)
            .await?;
        let closed = opened + chrono::Duration::minutes(10);
        recorder
            .record_workloop_iteration_boundary(
                closed,
                Vec::new(),
                Vec::new(),
                aion_core::Payload::from_json(&serde_json::json!({}))?,
                first.clone(),
                start(second.clone())?,
            )
            .await?;

        // ONE ROW PER WORKFLOW: the boundary's successor upsert REPLACED
        // the predecessor's row — the store holds exactly the current
        // generation, Running, and no ContinuedAsNew row survives anywhere
        // a list can see. The chain stays in history, under describe.
        let row = store
            .get_visibility(&workflow_id)
            .await?
            .ok_or("the workflow keeps its one row")?;
        assert_eq!(row.run_id, second);
        assert_eq!(row.status, aion_core::WorkflowStatus::Running);
        assert_eq!(row.ended_at, None);
        assert_eq!(row.started_at, closed);
        Ok(())
    }

    /// A straggler through a SUPERSEDED generation's recorder (a lease event
    /// for the predecessor landing after the boundary replaced its row) may
    /// heal the workflow's row forward, never regress it: the row keeps the
    /// successor's run and start.
    #[tokio::test]
    async fn a_straggling_touch_from_a_superseded_generation_never_regresses_the_row()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(13);
        let first = aion_core::RunId::new(uuid::Uuid::from_u128(1301));
        let second = aion_core::RunId::new(uuid::Uuid::from_u128(1302));
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone())
            .with_visibility(first.clone(), store.clone());
        let opened = recorded_at(1);
        recorder
            .record_workflow_started(
                opened,
                super::WorkflowStartRecord {
                    workflow_type: String::from("disk_reaper"),
                    input: payload("input")?,
                    run_id: first.clone(),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
            )
            .await?;
        let start = |run_id: aion_core::RunId| -> Result<_, Box<dyn std::error::Error>> {
            Ok(super::WorkflowStartRecord {
                workflow_type: String::from("disk_reaper"),
                input: payload("input")?,
                run_id,
                parent_run_id: Some(first.clone()),
                parent_workflow_id: None,
                package_version: aion_core::PackageVersion::new("a".repeat(64)),
            })
        };
        let closed = opened + chrono::Duration::minutes(10);
        recorder
            .record_workloop_iteration_boundary(
                closed,
                Vec::new(),
                Vec::new(),
                aion_core::Payload::from_json(&serde_json::json!({}))?,
                first.clone(),
                start(second.clone())?,
            )
            .await?;
        let head = recorder.current_head();

        // The straggler: a recorder still pointed at the FIRST generation
        // records a lease event after the boundary.
        let mut straggler = Recorder::resume_at(workflow_id.clone(), store.clone(), head)
            .with_visibility(first.clone(), store.clone());
        let worker = aion_core::WorkerAttribution {
            identity: String::from("w-late"),
            task_queue: String::from("reaper"),
            node: None,
            deployment: None,
            instance_id: None,
            transport: aion_core::WorkerTransport::Grpc,
        };
        straggler
            .record_activity_leased(
                closed + chrono::Duration::seconds(1),
                aion_core::ActivityId::from_sequence_position(head + 1),
                1,
                worker,
            )
            .await?;

        let row = store
            .get_visibility(&workflow_id)
            .await?
            .ok_or("the workflow keeps its one row")?;
        assert_eq!(
            row.run_id, second,
            "the row never regresses to the superseded generation"
        );
        assert_eq!(row.started_at, closed);
        assert_eq!(row.status, aion_core::WorkflowStatus::Running);
        Ok(())
    }

    #[tokio::test]
    async fn recorder_visibility_updates_after_search_attributes()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(9);
        let run_id = aion_core::RunId::new(uuid::Uuid::from_u128(90));
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone())
            .with_visibility(run_id.clone(), store.clone());
        let mut schema = SearchAttributeSchema::new();
        schema.register("customer_id", SearchAttributeType::String)?;
        let attributes = HashMap::from([(
            String::from("customer_id"),
            SearchAttributeValue::String(String::from("customer-123")),
        )]);

        recorder
            .record_workflow_started(
                recorded_at(1),
                super::WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("input")?,
                    // The run the projection is for IS the run that started: a row is
                    // projected from its own generation's window.
                    run_id: run_id.clone(),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
            )
            .await?;
        recorder
            .record_search_attributes_updated(recorded_at(2), attributes.clone(), &schema)
            .await?;

        let row = store
            .get_visibility(&workflow_id)
            .await?
            .ok_or("the attribute update projects the run's row")?;
        assert_eq!(row.workflow_id, workflow_id);
        assert_eq!(row.run_id, run_id);
        assert_eq!(row.search_attributes, attributes);
        assert_eq!(row.updated_at, recorded_at(2));
        Ok(())
    }

    /// WA-010 R4: the per-append touch moves `current_worker` on the stored
    /// row through the same transition the whole-history fold uses, so the
    /// row a list reads and the summary a describe builds from history agree
    /// after every lease and every attempt terminal — without the touch
    /// re-reading history.
    #[tokio::test]
    async fn recorder_touch_moves_current_worker_on_the_row_like_the_fold()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(11);
        let run_id = RunId::new(uuid::Uuid::from_u128(110));
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone())
            .with_visibility(run_id.clone(), store.clone());
        recorder
            .record_workflow_started(
                recorded_at(1),
                super::WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("input")?,
                    run_id: run_id.clone(),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
            )
            .await?;
        let worker = |identity: &str| aion_core::WorkerAttribution {
            identity: identity.to_owned(),
            task_queue: String::from("billing"),
            node: None,
            deployment: None,
            instance_id: None,
            transport: aion_core::WorkerTransport::Grpc,
        };
        let activity = aion_core::ActivityId::from_sequence_position(2);

        recorder
            .record_activity_leased(recorded_at(2), activity.clone(), 1, worker("w-a"))
            .await?;
        let row = store
            .get_visibility(&workflow_id)
            .await?
            .ok_or("the start upsert projects the row")?;
        assert_eq!(
            row.summary().current_worker.map(|w| w.identity).as_deref(),
            Some("w-a"),
            "the lease touch names the worker on the row"
        );
        assert_eq!(row.updated_at, recorded_at(2));

        // A redelivery of the same attempt: the later lease wins.
        recorder
            .record_activity_leased(recorded_at(3), activity.clone(), 1, worker("w-b"))
            .await?;
        recorder
            .record_activity_completed(recorded_at(4), activity, payload("done")?, 1)
            .await?;
        let row = store
            .get_visibility(&workflow_id)
            .await?
            .ok_or("the row survives every touch")?;
        assert_eq!(
            row.summary().current_worker,
            None,
            "the attempt's completion clears its lease on the row"
        );
        assert_eq!(row.updated_at, recorded_at(4));

        // The row equals what a full projection of the same history builds.
        let history = store.read_history(&workflow_id).await?;
        let projected = crate::lifecycle::visibility::project_visibility(&history, &run_id)
            .ok_or("a started history projects")?;
        assert_eq!(
            row, projected,
            "touched row == projected row, leases included"
        );
        Ok(())
    }

    /// A visibility store that refuses writes while armed and delegates
    /// otherwise — the crash window between a durable append and its row.
    struct ArmedVisibilityStore {
        inner: Arc<InMemoryStore>,
        refuse: std::sync::atomic::AtomicBool,
    }

    #[async_trait]
    impl VisibilityStore for ArmedVisibilityStore {
        async fn record_visibility(&self, record: VisibilityRecord) -> Result<(), StoreError> {
            if self.refuse.load(std::sync::atomic::Ordering::SeqCst) {
                return Err(StoreError::Backend(String::from("visibility unavailable")));
            }
            self.inner.record_visibility(record).await
        }

        async fn get_visibility(
            &self,
            workflow_id: &aion_core::WorkflowId,
        ) -> Result<Option<VisibilityRecord>, StoreError> {
            self.inner.get_visibility(workflow_id).await
        }

        async fn remove_visibility(
            &self,
            workflow_id: &aion_core::WorkflowId,
            run_id: &RunId,
        ) -> Result<bool, StoreError> {
            self.inner.remove_visibility(workflow_id, run_id).await
        }

        async fn list_workflows(
            &self,
            request: &WorkflowListRequest,
        ) -> Result<VisibilityPage, StoreError> {
            self.inner.list_workflows(request).await
        }
    }

    /// A recorder over an [`ArmedVisibilityStore`] with a started workflow
    /// whose row was projected the way the lifecycle handlers project it
    /// (`record_workflow_started` and `record_workflow_failed` append without
    /// projecting; the lifecycle handlers project after them).
    struct ArmedFixture {
        workflow_id: aion_core::WorkflowId,
        run_id: RunId,
        store: Arc<InMemoryStore>,
        visibility: Arc<ArmedVisibilityStore>,
        recorder: Recorder,
    }

    impl ArmedFixture {
        async fn started() -> Result<Self, Box<dyn std::error::Error>> {
            let workflow_id = workflow_id(11);
            let run_id = RunId::new(uuid::Uuid::from_u128(110));
            let store = Arc::new(InMemoryStore::default());
            let visibility = Arc::new(ArmedVisibilityStore {
                inner: Arc::clone(&store),
                refuse: std::sync::atomic::AtomicBool::new(false),
            });
            let mut recorder = Recorder::new(workflow_id.clone(), store.clone()).with_visibility(
                run_id.clone(),
                Arc::clone(&visibility) as Arc<dyn VisibilityStore>,
            );
            recorder
                .record_workflow_started(
                    recorded_at(1),
                    super::WorkflowStartRecord {
                        workflow_type: String::from("checkout"),
                        input: payload("input")?,
                        run_id: run_id.clone(),
                        parent_run_id: None,
                        parent_workflow_id: None,
                        package_version: aion_core::PackageVersion::new("a".repeat(64)),
                    },
                )
                .await?;
            let fixture = Self {
                workflow_id,
                run_id,
                store,
                visibility,
                recorder,
            };
            fixture.project().await?;
            assert_eq!(
                fixture.row().await?.head_seq,
                1,
                "the start projects a row at 1"
            );
            Ok(fixture)
        }

        /// Projects the whole row from history, as a lifecycle handler does.
        async fn project(&self) -> Result<(), Box<dyn std::error::Error>> {
            let history = self.store.read_history(&self.workflow_id).await?;
            let row = crate::lifecycle::visibility::project_visibility(&history, &self.run_id)
                .ok_or("the history projects a row")?;
            self.visibility.record_visibility(row).await?;
            Ok(())
        }

        async fn row(&self) -> Result<VisibilityRecord, Box<dyn std::error::Error>> {
            Ok(self
                .store
                .get_visibility(&self.workflow_id)
                .await?
                .ok_or("the row survives")?)
        }

        /// A non-lifecycle append at `seq`: the per-append touch path.
        async fn touch(&mut self, seq: u64) -> Result<(), Box<dyn std::error::Error>> {
            self.recorder
                .record_activity_scheduled(
                    recorded_at(i64::try_from(seq)?),
                    aion_core::ActivityId::from_sequence_position(seq),
                    format!("step_{seq}"),
                    payload("input")?,
                    String::from("default"),
                    None,
                )
                .await?;
            Ok(())
        }

        fn refuse(&self, refuse: bool) {
            self.visibility
                .refuse
                .store(refuse, std::sync::atomic::Ordering::SeqCst);
        }
    }

    /// A touch right behind the row's head stamps it: the row provably
    /// folded everything before that event.
    #[tokio::test]
    async fn a_contiguous_touch_carries_the_stamp_forward() -> Result<(), Box<dyn std::error::Error>>
    {
        let mut fixture = ArmedFixture::started().await?;
        fixture.touch(2).await?;
        let row = fixture.row().await?;
        assert_eq!(
            row.head_seq, 2,
            "a contiguous touch carries the stamp forward"
        );
        assert_eq!(
            aion_store::visibility::head::verdict(Some(&row), 2),
            aion_store::visibility::head::RowVerdict::InFlight
        );
        Ok(())
    }

    /// The stamp is inductive: after a projection the store refused, later
    /// touches must NOT carry the row to the stream head — a row stamped at
    /// the head with a status that missed a lifecycle event would be trusted
    /// by the next boot. Left behind, it is re-derived instead.
    #[tokio::test]
    async fn a_touch_never_stamps_a_row_past_a_projection_that_failed()
    -> Result<(), Box<dyn std::error::Error>> {
        let mut fixture = ArmedFixture::started().await?;
        fixture.touch(2).await?;
        assert_eq!(fixture.row().await?.head_seq, 2);

        // The lifecycle event lands durably; its projection is refused.
        fixture
            .recorder
            .record_workflow_failed(
                recorded_at(3),
                aion_core::WorkflowError {
                    message: String::from("boom"),
                    details: None,
                },
            )
            .await?;
        fixture.refuse(true);
        assert!(
            fixture.project().await.is_err(),
            "the failure's projection was refused"
        );
        fixture.refuse(false);

        // A touch follows the unfolded failure.
        fixture.touch(4).await?;
        let history = fixture.store.read_history(&fixture.workflow_id).await?;
        assert_eq!(
            history.len(),
            4,
            "the history holds the failure the row never folded"
        );
        let row = fixture.row().await?;
        assert_eq!(
            row.status,
            aion_core::WorkflowStatus::Running,
            "the row never saw the failure"
        );
        assert_eq!(
            row.head_seq, 2,
            "the touch refuses to stamp across the unfolded event: the row stays behind the head"
        );
        assert_eq!(
            aion_store::visibility::head::verdict(Some(&row), 4),
            aion_store::visibility::head::RowVerdict::Unsettled,
            "a boot re-derives this row instead of trusting it"
        );
        Ok(())
    }

    #[tokio::test]
    async fn visibility_upsert_failure_after_append_logs_warning_and_succeeds()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(10);
        let run_id = RunId::new(uuid::Uuid::from_u128(100));
        let store = Arc::new(InMemoryStore::default());
        let visibility_store = Arc::new(FailingVisibilityStore);
        let (logs, subscriber) = crate::log_capture::LogCapture::new()?;
        let guard = tracing::subscriber::set_default(subscriber);
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone())
            .with_visibility(run_id.clone(), visibility_store);
        let mut schema = SearchAttributeSchema::new();
        schema.register("customer_id", SearchAttributeType::String)?;
        let attributes = HashMap::from([(
            String::from("customer_id"),
            SearchAttributeValue::String(String::from("customer-123")),
        )]);

        recorder
            .record_workflow_started(
                recorded_at(1),
                super::WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("input")?,
                    run_id: run_id.clone(),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
            )
            .await?;
        recorder
            .record_search_attributes_updated(recorded_at(2), attributes, &schema)
            .await?;
        drop(guard);

        let history = store.read_history(&workflow_id).await?;
        assert_eq!(history.len(), 2);
        assert_eq!(recorder.current_head(), 2);

        // ONE warning, and everything the operator needs is in THAT one. An
        // earlier revision of this test held its own `Vec<u8>` capture and
        // asserted over the joined rendering of every line, which could not
        // tell one complete warning from five partial ones spread across five
        // events — and, holding its own capture, never installed the
        // interest floor that `crate::log_capture` exists to guarantee.
        let warnings = logs.at_level("WARN")?;
        assert_eq!(
            warnings.len(),
            1,
            "the failed visibility upsert must produce exactly one WARN — more than one means the \
             warning was split across events and no single line carries the whole fact: \
             {warnings:?}"
        );
        let warning = warnings
            .first()
            .ok_or("the visibility-upsert warning was not captured at all")?;
        for needle in [
            "visibility upsert failed after durable append",
            "crash-consistency window",
            "visibility unavailable",
            workflow_id.to_string().as_str(),
            run_id.to_string().as_str(),
        ] {
            assert!(
                warning.mentions(needle),
                "the visibility-upsert warning did not mention `{needle}`: {warning}"
            );
        }
        Ok(())
    }

    #[tokio::test]
    async fn records_activity_events_in_sequence_order() -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(6);
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone());
        let activity_id = aion_core::ActivityId::from_sequence_position(1);

        recorder
            .record_activity_scheduled(
                recorded_at(1),
                activity_id.clone(),
                String::from("charge-card"),
                payload("input")?,
                String::from("default"),
                None,
            )
            .await?;
        recorder
            .record_activity_completed(recorded_at(2), activity_id.clone(), payload("result")?, 1)
            .await?;

        let history = store.read_history(&workflow_id).await?;
        assert_eq!(history.len(), 2);
        assert_eq!(history[0].seq(), 1);
        assert_eq!(history[1].seq(), 2);
        match &history[0] {
            Event::ActivityScheduled {
                activity_id: recorded_activity_id,
                ..
            } => assert_eq!(recorded_activity_id, &activity_id),
            other => return Err(format!("expected ActivityScheduled, got {other:?}").into()),
        }
        match &history[1] {
            Event::ActivityCompleted {
                activity_id: recorded_activity_id,
                ..
            } => assert_eq!(recorded_activity_id, &activity_id),
            other => return Err(format!("expected ActivityCompleted, got {other:?}").into()),
        }
        Ok(())
    }

    #[tokio::test]
    async fn resume_at_continues_from_existing_history_head()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(3);
        let store = Arc::new(InMemoryStore::default());
        let seeded = [
            workflow_started(1, &workflow_id)?,
            workflow_started(2, &workflow_id)?,
        ];
        store
            .append(WriteToken::recorder(), &workflow_id, &seeded, 0)
            .await?;

        let mut recorder = Recorder::resume_at(workflow_id.clone(), store.clone(), 2);
        recorder
            .record_signal_received(recorded_at(3), String::from("approve"), payload("signal")?)
            .await?;

        let history = store.read_history(&workflow_id).await?;
        assert_eq!(history.len(), 3);
        assert_eq!(history[2].seq(), 3);
        assert_eq!(recorder.current_head(), 3);
        Ok(())
    }

    #[tokio::test]
    async fn sequence_conflict_surfaces_without_advancing_or_retrying()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(4);
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone());
        let rogue_event = workflow_started(1, &workflow_id)?;
        store
            .append(WriteToken::recorder(), &workflow_id, &[rogue_event], 0)
            .await?;

        let error = recorder
            .record_timer_fired(recorded_at(2), TimerId::anonymous(2))
            .await;

        match error {
            Err(DurabilityError::Store(StoreError::SequenceConflict { expected, found })) => {
                assert_eq!(expected, 0);
                assert_eq!(found, 1);
            }
            Err(other) => return Err(format!("expected sequence conflict, got {other:?}").into()),
            Ok(()) => return Err("expected sequence conflict".into()),
        }
        assert_eq!(recorder.current_head(), 0);
        let history = store.read_history(&workflow_id).await?;
        assert_eq!(history.len(), 1);
        assert_eq!(history[0].seq(), 1);
        Ok(())
    }

    /// aion#145: the ack-loss repair is FORWARD-ONLY and restores the append
    /// path. It exists for exactly one shape — an append this recorder issued
    /// landed while its acknowledgement was lost, leaving the tracked head
    /// below the durable head — and it must never move the head backward
    /// (which could mask a genuine double-writer conflict by re-arming a stale
    /// expectation). The sibling pin
    /// `sequence_conflict_surfaces_without_advancing_or_retrying` guards the
    /// other side: a conflict itself never reconciles anything.
    #[tokio::test]
    async fn reconcile_head_forward_is_forward_only_and_restores_the_append_path()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(42);
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::new(workflow_id.clone(), store.clone());
        // Reproduce the ack-lost store state: the event the recorder minted
        // landed durably, but `mark_append_success` was never reached.
        let landed = workflow_started(1, &workflow_id)?;
        store
            .append(WriteToken::recorder(), &workflow_id, &[landed], 0)
            .await?;
        assert_eq!(
            recorder.current_head(),
            0,
            "the ack loss left the tracker stale-low"
        );

        // At-head (here: still-stale-equal) reconciliation is a no-op.
        recorder.reconcile_head_forward(0);
        assert_eq!(recorder.current_head(), 0);

        // Forward reconciliation adopts the durable head.
        recorder.reconcile_head_forward(1);
        assert_eq!(recorder.current_head(), 1);

        // Never backward.
        recorder.reconcile_head_forward(0);
        assert_eq!(recorder.current_head(), 1);

        // The append path is healthy again: the next event lands at seq 2.
        recorder
            .record_signal_received(recorded_at(2), String::from("next"), payload("signal")?)
            .await?;
        let history = store.read_history(&workflow_id).await?;
        assert_eq!(history.len(), 2);
        assert_eq!(history[1].seq(), 2);
        assert_eq!(recorder.current_head(), 2);
        Ok(())
    }

    #[tokio::test]
    async fn sequence_overflow_returns_error_without_appending()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(5);
        let store = Arc::new(InMemoryStore::default());
        let mut recorder = Recorder::resume_at(workflow_id.clone(), store.clone(), u64::MAX);

        let error = recorder
            .record_workflow_completed(recorded_at(1), payload("result")?)
            .await;

        match error {
            Err(DurabilityError::HistoryShape { reason }) => {
                assert!(reason.contains("sequence head overflow"));
            }
            Err(other) => return Err(format!("expected sequence overflow, got {other:?}").into()),
            Ok(()) => return Err("expected sequence overflow".into()),
        }
        assert_eq!(recorder.current_head(), u64::MAX);
        assert!(store.read_history(&workflow_id).await?.is_empty());
        Ok(())
    }

    /// #253 settle-at-terminal: every workflow terminal the Recorder records
    /// (completed / failed / cancelled) settles the workflow's outbox rows
    /// through the store seam, AND a settle failure never fails the recorded
    /// terminal. The spy's `settle_workflow_outbox_rows_cancelled` returns an
    /// `Err` sentinel, so each terminal here succeeding while the call is
    /// recorded proves both halves at once.
    #[tokio::test]
    async fn workflow_terminals_settle_outbox_rows_and_settle_errors_are_nonfatal()
    -> Result<(), Box<dyn std::error::Error>> {
        use aion_store::testing::ShardSeamSpy;

        type TerminalRecorder = fn(
            &mut Recorder,
        ) -> std::pin::Pin<
            Box<dyn Future<Output = Result<(), DurabilityError>> + Send + '_>,
        >;
        let terminals: [(&str, TerminalRecorder); 3] = [
            ("completed", |recorder| {
                Box::pin(async move {
                    recorder
                        .record_workflow_completed(
                            recorded_at(2),
                            Payload::from_json(&json!({"label": "result"})).map_err(|error| {
                                DurabilityError::HistoryShape {
                                    reason: error.to_string(),
                                }
                            })?,
                        )
                        .await
                })
            }),
            ("failed", |recorder| {
                Box::pin(async move {
                    recorder
                        .record_workflow_failed(
                            recorded_at(2),
                            aion_core::WorkflowError {
                                message: String::from("boom"),
                                details: None,
                            },
                        )
                        .await
                })
            }),
            ("cancelled", |recorder| {
                Box::pin(async move {
                    recorder
                        .record_workflow_cancelled(recorded_at(2), String::from("operator"))
                        .await
                })
            }),
        ];

        for (label, record_terminal) in terminals {
            let workflow_id = workflow_id(31);
            let spy = Arc::new(ShardSeamSpy::new());
            let store: Arc<dyn aion_store::EventStore> = Arc::clone(&spy) as _;
            let mut recorder = Recorder::new(workflow_id.clone(), Arc::clone(&store));
            recorder
                .record_workflow_started(
                    recorded_at(1),
                    super::WorkflowStartRecord {
                        workflow_type: String::from("checkout"),
                        input: payload("input")?,
                        run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
                        parent_run_id: None,
                        parent_workflow_id: None,
                        package_version: aion_core::PackageVersion::new("a".repeat(64)),
                    },
                )
                .await?;

            record_terminal(&mut recorder).await.map_err(|error| {
                format!("terminal `{label}` must record despite the settle Err sentinel: {error}")
            })?;

            let history = store.read_history(&workflow_id).await?;
            assert_eq!(
                history.len(),
                2,
                "terminal `{label}` must be durably recorded"
            );
            let calls = spy.calls();
            assert!(
                calls.contains(&format!(
                    "settle_workflow_outbox_rows_cancelled:{workflow_id}"
                )),
                "terminal `{label}` must settle the workflow's outbox rows; saw {calls:?}"
            );
        }
        Ok(())
    }

    /// #253: continue-as-new is NOT a settle point — the workflow continues,
    /// and the old run's stragglers are neutralized by run-scoped completion
    /// threading (OBX-011), so its rows must stay live.
    #[tokio::test]
    async fn continue_as_new_does_not_settle_outbox_rows() -> Result<(), Box<dyn std::error::Error>>
    {
        use aion_store::testing::ShardSeamSpy;

        let workflow_id = workflow_id(32);
        let spy = Arc::new(ShardSeamSpy::new());
        let store: Arc<dyn aion_store::EventStore> = Arc::clone(&spy) as _;
        let mut recorder = Recorder::new(workflow_id.clone(), store);
        recorder
            .record_workflow_started(
                recorded_at(1),
                super::WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("input")?,
                    run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
            )
            .await?;
        recorder
            .record_workflow_continued_as_new(
                recorded_at(2),
                payload("continued")?,
                None,
                aion_core::RunId::new(uuid::Uuid::from_u128(2)),
            )
            .await?;

        let calls = spy.calls();
        assert!(
            !calls
                .iter()
                .any(|call| call.starts_with("settle_workflow_outbox_rows_cancelled:")),
            "continue-as-new must never settle the workflow's outbox rows; saw {calls:?}"
        );
        Ok(())
    }
    /// AD-014 (aion#192), the shape that bit: a COMMAND append — `TimerStarted`
    /// — whose write lands while its acknowledgement is lost. The recorder
    /// resolves it against the store instead of surfacing the error: the call
    /// returns Ok with the armed sequence, the tracked head equals the durable
    /// head, the history holds the event exactly once, the loss is said at
    /// WARN with the landed range, and the NEXT append through the same
    /// recorder lands at the correct sequence.
    #[tokio::test]
    async fn an_ack_lost_command_append_is_resolved_as_recorded_and_the_run_keeps_appending()
    -> Result<(), Box<dyn std::error::Error>> {
        let (capture, subscriber) = crate::log_capture::LogCapture::new()?;
        let _guard = tracing::subscriber::set_default(subscriber);
        let workflow_id = workflow_id(40);
        let store = Arc::new(crate::store_faults::FlakyStore::new());
        store
            .append(
                WriteToken::recorder(),
                &workflow_id,
                &[workflow_started(1, &workflow_id)?],
                0,
            )
            .await?;
        let mut recorder = Recorder::resume_at(
            workflow_id.clone(),
            Arc::clone(&store) as Arc<dyn aion_store::EventStore>,
            1,
        );
        store.lose_ack_of_next_appends(1);

        let armed = recorder
            .record_timer_started(recorded_at(2), TimerId::anonymous(1), recorded_at(60))
            .await?;

        assert_eq!(
            armed, 2,
            "the command append reports the sequence it landed at"
        );
        assert_eq!(
            recorder.current_head(),
            2,
            "the tracked head equals the durable head"
        );
        let history = store.recorded_history(&workflow_id).await?;
        assert_eq!(
            history.len(),
            2,
            "the event is in the history exactly once: {history:#?}"
        );
        assert!(matches!(history[1], Event::TimerStarted { .. }));

        let warned = capture.at_level("WARN")?;
        let resolved = warned
            .iter()
            .find(|event| event.mentions("acknowledgement was lost"))
            .ok_or("the resolved append must be said at WARN")?;
        assert_eq!(resolved.field("landed_seq_from"), Some("2"));
        assert_eq!(resolved.field("landed_seq_to"), Some("2"));
        assert!(
            resolved
                .field("store_error")
                .is_some_and(|text| text.contains("injected by FlakyStore")),
            "the WARN carries the store's original error: {resolved:?}"
        );

        recorder
            .record_timer_fired(recorded_at(3), TimerId::anonymous(1))
            .await?;
        assert_eq!(recorder.current_head(), 3);
        assert_eq!(store.recorded_history(&workflow_id).await?.len(), 3);
        Ok(())
    }

    /// AD-014: a terminal append (`WorkflowCancelled`) under a lost
    /// acknowledgement is resolved the same way — Ok, ONE terminal in the
    /// history, head advanced — because the run's status is a projection of
    /// exactly that event and a second one must never be minted.
    #[tokio::test]
    async fn an_ack_lost_terminal_append_is_resolved_with_one_terminal_in_the_history()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(41);
        let store = Arc::new(crate::store_faults::FlakyStore::new());
        store
            .append(
                WriteToken::recorder(),
                &workflow_id,
                &[workflow_started(1, &workflow_id)?],
                0,
            )
            .await?;
        let mut recorder = Recorder::resume_at(
            workflow_id.clone(),
            Arc::clone(&store) as Arc<dyn aion_store::EventStore>,
            1,
        );
        store.lose_ack_of_next_appends(1);

        recorder
            .record_workflow_cancelled(recorded_at(2), String::from("operator"))
            .await?;

        assert_eq!(recorder.current_head(), 2);
        let history = store.recorded_history(&workflow_id).await?;
        let terminals = history
            .iter()
            .filter(|event| matches!(event, Event::WorkflowCancelled { .. }))
            .count();
        assert_eq!(terminals, 1, "exactly one terminal: {history:#?}");
        Ok(())
    }

    /// AD-014: an append the store refused BEFORE writing is not a resolution
    /// case — nothing is at the expected sequence, so the ORIGINAL error is
    /// returned unchanged, the head is untouched, and the caller's own retry
    /// then lands at the same sequence.
    #[tokio::test]
    async fn a_refused_append_that_did_not_land_returns_the_refusal_and_leaves_the_head()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(42);
        let store = Arc::new(crate::store_faults::FlakyStore::new());
        store
            .append(
                WriteToken::recorder(),
                &workflow_id,
                &[workflow_started(1, &workflow_id)?],
                0,
            )
            .await?;
        let mut recorder = Recorder::resume_at(
            workflow_id.clone(),
            Arc::clone(&store) as Arc<dyn aion_store::EventStore>,
            1,
        );
        store.fail_next_appends(1);

        let refused = recorder
            .record_timer_started(recorded_at(2), TimerId::anonymous(1), recorded_at(60))
            .await;
        match refused {
            Err(DurabilityError::Store(StoreError::Backend(text))) => {
                assert!(text.contains("transient append failure injected by FlakyStore"));
            }
            other => return Err(format!("expected the store's refusal, got {other:?}").into()),
        }
        assert_eq!(
            recorder.current_head(),
            1,
            "nothing landed, nothing advances"
        );
        assert_eq!(store.recorded_history(&workflow_id).await?.len(), 1);

        let armed = recorder
            .record_timer_started(recorded_at(2), TimerId::anonymous(1), recorded_at(60))
            .await?;
        assert_eq!(armed, 2, "the caller's retry lands at the same sequence");
        Ok(())
    }

    /// AD-014: a fenced write (`NotOwner`) is a definite did-not-land and a
    /// typed routing signal, so it is surfaced AS IS — no resolving read.
    /// The rogue planted at the expected sequence is the discriminator: a
    /// resolver that ran would have found it and minted a `SequenceConflict`
    /// (a double-writer alarm out of a routing condition), and a resolver
    /// whose read failed would have folded the variant into `Backend`.
    #[tokio::test]
    async fn a_fenced_append_is_surfaced_typed_without_a_resolving_read()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(45);
        let store = Arc::new(crate::store_faults::FlakyStore::new());
        store
            .append(
                WriteToken::recorder(),
                &workflow_id,
                &[workflow_started(1, &workflow_id)?],
                0,
            )
            .await?;
        let mut recorder = Recorder::resume_at(
            workflow_id.clone(),
            Arc::clone(&store) as Arc<dyn aion_store::EventStore>,
            1,
        );
        let rogue = Event::TimerFired {
            envelope: aion_core::EventEnvelope {
                seq: 2,
                recorded_at: recorded_at(2),
                workflow_id: workflow_id.clone(),
            },
            timer_id: TimerId::anonymous(9),
        };
        store
            .append(WriteToken::recorder(), &workflow_id, &[rogue], 1)
            .await?;
        store.fence_next_appends(1);
        store.fail_next_range_reads(1);

        let outcome = recorder
            .record_timer_started(recorded_at(2), TimerId::anonymous(1), recorded_at(60))
            .await;
        match outcome {
            Err(DurabilityError::Store(StoreError::NotOwner { shard })) => {
                assert_eq!(shard, 7, "the fence comes back typed, shard and all");
            }
            other => {
                return Err(format!("expected the typed NotOwner refusal, got {other:?}").into());
            }
        }
        assert_eq!(recorder.current_head(), 1, "a fence never moves the head");
        Ok(())
    }

    /// AD-014 keeps aion#145's other side: when the resolving read finds
    /// SOMETHING ELSE at the expected sequence — a rogue writer's event, not
    /// this batch — the result is a genuine `SequenceConflict` and the head
    /// is NOT moved. Resyncing here would silence the double-writer alarm.
    #[tokio::test]
    async fn a_refused_append_over_a_rogue_event_is_a_conflict_not_a_resync()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(43);
        let store = Arc::new(crate::store_faults::FlakyStore::new());
        store
            .append(
                WriteToken::recorder(),
                &workflow_id,
                &[workflow_started(1, &workflow_id)?],
                0,
            )
            .await?;
        let mut recorder = Recorder::resume_at(
            workflow_id.clone(),
            Arc::clone(&store) as Arc<dyn aion_store::EventStore>,
            1,
        );
        // The rogue lands at seq 2 behind this recorder's back; the recorder's
        // own append is then refused before writing, so the resolving read
        // sees the rogue where its batch would have been.
        let rogue = Event::TimerFired {
            envelope: aion_core::EventEnvelope {
                seq: 2,
                recorded_at: recorded_at(2),
                workflow_id: workflow_id.clone(),
            },
            timer_id: TimerId::anonymous(9),
        };
        store
            .append(WriteToken::recorder(), &workflow_id, &[rogue], 1)
            .await?;
        store.fail_next_appends(1);

        let outcome = recorder
            .record_timer_started(recorded_at(2), TimerId::anonymous(1), recorded_at(60))
            .await;
        match outcome {
            Err(DurabilityError::Store(StoreError::SequenceConflict { expected, found })) => {
                assert_eq!((expected, found), (1, 2));
            }
            other => return Err(format!("expected a sequence conflict, got {other:?}").into()),
        }
        assert_eq!(
            recorder.current_head(),
            1,
            "a conflict never reconciles the head"
        );
        Ok(())
    }

    /// AD-014: when the resolving read itself fails, nothing was learned — the
    /// ORIGINAL append error comes back with the read failure attached, and
    /// the head is untouched.
    #[tokio::test]
    async fn a_failed_resolving_read_returns_the_original_error_with_the_read_attached()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = workflow_id(44);
        let store = Arc::new(crate::store_faults::FlakyStore::new());
        store
            .append(
                WriteToken::recorder(),
                &workflow_id,
                &[workflow_started(1, &workflow_id)?],
                0,
            )
            .await?;
        let mut recorder = Recorder::resume_at(
            workflow_id.clone(),
            Arc::clone(&store) as Arc<dyn aion_store::EventStore>,
            1,
        );
        store.lose_ack_of_next_appends(1);
        store.fail_next_range_reads(1);

        let outcome = recorder
            .record_timer_started(recorded_at(2), TimerId::anonymous(1), recorded_at(60))
            .await;
        match outcome {
            Err(DurabilityError::Store(StoreError::Backend(text))) => {
                assert!(
                    text.contains("acknowledgement was lost, injected by FlakyStore"),
                    "{text}"
                );
                assert!(
                    text.contains("transient range read failure injected by FlakyStore"),
                    "{text}"
                );
            }
            other => return Err(format!("expected the original error, got {other:?}").into()),
        }
        assert_eq!(
            recorder.current_head(),
            1,
            "nothing was learned, nothing advances"
        );
        Ok(())
    }
}