liminal-server 0.20.1

//! The single production participant semantic handler.
//!
//! One handler per server. Each conversation has exactly one live in-memory
//! authority owner, rebuilt from its durable transition-input log on first
//! touch (and after any failed operation, which discards the owner so the
//! next touch cold-replays durable reality). A short registry lock selects
//! the conversation cell; a per-conversation lock covers replay and each
//! operation. Everything is event-driven: cells are created on request
//! arrival, discarded on error, and evicted entirely when the touched
//! conversation has no durable log — no timer, sweep, or polling loop
//! exists, and refused probes of unknown conversation ids leave neither
//! durable nor in-memory residue.

use crate::server::signals::ServerSignals;
use std::sync::OnceLock;

use std::cell::{Cell, RefCell};
use std::collections::{BTreeMap, HashMap};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Instant;

use liminal::durability::bridge::block_on;
use liminal::durability::{DurabilityError, DurableStore};
use liminal_protocol::lifecycle::{CapacityCounter, ObserverRecoveryAggregate};
use liminal_protocol::wire::ConversationId;

use crate::config::types::ParticipantConfig;
use crate::health::reissue::{
    OperatorCredentialReissueError, OperatorCredentialReissueOutcome,
    OperatorCredentialReissueRequest, OperatorCredentialReissuer,
};
use crate::health::unloadable::{UnloadableConversation, UnloadableConversationRecord};
use crate::server::participant::{
    ObserverPublicationTarget, ParticipantConnectionContext, ParticipantConnectionConversations,
    ParticipantSemanticError, ParticipantServiceFatal, dispatch_impact::DispatchImpactAccumulator,
};

use super::barrier::{OperationFacts, ReceiptCapacityLimits};
use super::capacity::ServerCapacity;
#[cfg(test)]
use super::dispatch_work::{ObligationDispatchWorkCounters, ObligationDispatchWorkSnapshot};
use super::facts;
use super::log::{OperationLog, OperationLogError, StoredOperation};
use super::operation_batch::OperationBatch;
use super::outbox::{ConversationOutboxError, ConversationOutboxLimits};
use super::outbox_log::{OutboxLog, OutboxLogError};
use super::outbox_projection::owes_extension_row;
use super::registry::ConversationRegistry;
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::routed_stream::RoutedStreamRefusal;
use super::state::{ConversationAuthority, DurableAppend, LifecycleReceipt, StateError};
use super::write_issuance::WriteIssuance;

#[derive(Debug)]
pub(super) struct ObserverArmTarget {
    pub(super) refused_epoch: u64,
    pub(super) connection_incarnation: liminal_protocol::wire::ConnectionIncarnation,
    pub(super) target: ObserverPublicationTarget,
}

/// One exclusively serialized observer owner: durable protocol aggregate/head
/// plus volatile weak live targets for its installed arms. Restoring durable
/// arms intentionally starts with an empty target map because no socket
/// survives a process restart.
#[derive(Debug)]
pub(super) struct ObserverOwner {
    pub(super) aggregate: ObserverRecoveryAggregate,
    pub(super) head: u64,
    pub(super) arm_targets: BTreeMap<ConversationId, ObserverArmTarget>,
}

/// Lifecycle receipts held for a commit whose outcome was not confirmed, with
/// the log head its operation started from.
#[derive(Debug)]
struct OwedReceipts {
    starting_log_sequence: u64,
    receipts: Vec<LifecycleReceipt>,
}

/// Production semantic handler backed by the shared durable store.
///
/// Constructed exactly once per server by the connection-services layer when
/// the deployment's `[participant]` configuration is present.
#[derive(Debug)]
pub struct ProductionParticipantHandler {
    signals: OnceLock<ServerSignals>,
    pub(super) store: Arc<dyn DurableStore>,
    pub(super) config: ParticipantConfig,
    outbox_limits: ConversationOutboxLimits,
    conversations: Mutex<HashMap<ConversationId, Arc<Mutex<Option<ConversationAuthority>>>>>,
    /// First post-Open fatal; once set, every semantic/publication entry refuses.
    service_fatal: Mutex<Option<ParticipantServiceFatal>>,
    /// Server-wide observer-recovery aggregate paired with its durable row
    /// head (`None` until first restored).
    pub(super) observer: Mutex<Option<ObserverOwner>>,
    pub(super) observer_head_uncertain: AtomicBool,
    /// Server-scope stage-8 occupancy ledger (identity slots, live receipts,
    /// provenance fingerprints), restored from every durable conversation at
    /// construction and kept exact by commit reservations and replay folds.
    pub(super) capacity: ServerCapacity,
    /// Durable registry of created conversations: one row appended before
    /// each conversation's genesis append, read at startup to enumerate
    /// every durable conversation for the capacity restore.
    pub(super) registry: ConversationRegistry,
    /// CONTAINMENT: conversations whose durable state could not be loaded,
    /// each against the load failure's own class and text.
    ///
    /// A conversation lands here instead of taking the node down with it. The
    /// record is the node's answer to "which one is broken?", which is the half
    /// of the property that a boot-succeeds assertion cannot see: containment
    /// without attribution is a node that serves nothing on one conversation
    /// forever and never says so.
    ///
    /// The record is SHARED, not private: a clone of it is published onto the
    /// health endpoint's `GET /unloadable-conversations` at server startup
    /// ([`Self::unloadable_record`]), so the answer this handler writes is the
    /// same answer an operator reads. Nothing pushes — the endpoint snapshots
    /// this record only when it is scraped.
    unloadable: UnloadableConversationRecord,
    /// Lifecycle receipts owed by a commit whose outcome this process could
    /// not confirm: its batch was issued, and the operation failed without
    /// proof that its rows landed. A receipt is owed by the commit, not by
    /// the answer that first reported it, so it is held here, with the log
    /// head the operation started from, until the next load of the
    /// conversation shows whether the rows landed. Landed, the receipts are
    /// emitted then, exactly once. Not landed, they are dropped, because a
    /// re-presented request commits live and records its own. Held in memory
    /// only: a process that stops before that load owes them to nobody, as
    /// a crash between any commit and its emission does.
    owed_receipts: Mutex<HashMap<ConversationId, OwedReceipts>>,
    /// Exact W2 work observation points, isolated per handler and test-only.
    #[cfg(test)]
    pub(super) obligation_dispatch_work: ObligationDispatchWorkCounters,
    /// The participant clock's pin, for deterministic time-window tests.
    ///
    /// `0` means the wall clock, and production never writes anything else:
    /// only the test-only [`Self::pin_clock_ms`] stores a nonzero reading, so a
    /// test can step through the receipt and provenance window states instead
    /// of racing the wall clock. Production pays one atomic load per clock
    /// read for it, and in return [`Self::now_ms`] is one method in every build.
    now_override: AtomicU64,
}

impl ProductionParticipantHandler {
    pub(in crate::server) fn install_signals(
        &self,
        signals: ServerSignals,
    ) -> Result<(), ParticipantSemanticError> {
        self.signals
            .set(signals)
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant lifecycle signals were already installed".to_owned(),
            })
    }

    /// Creates the handler over the server's shared durable store, replaying
    /// every durable conversation so the server-scope capacity ledger is
    /// exact against durable truth from the first request (a restart must
    /// not forget reserved identity slots or in-window receipts).
    ///
    /// # Errors
    ///
    /// Returns [`ParticipantSemanticError`] when the durable store cannot be
    /// scanned or a conversation log fails replay — the server refuses to
    /// start over state it cannot account for.
    pub fn new(
        store: Arc<dyn DurableStore>,
        config: ParticipantConfig,
    ) -> Result<Self, ParticipantSemanticError> {
        let outbox_limits = ConversationOutboxLimits::try_new(
            config.max_retained_record_rows,
            config.identity_slots,
        )
        .map_err(|error| ParticipantSemanticError::Internal {
            message: format!("participant outbox limit configuration failed: {error}"),
        })?;
        let registry = ConversationRegistry::new(Arc::clone(&store));
        let handler = Self {
            signals: OnceLock::new(),
            store,
            config,
            outbox_limits,
            conversations: Mutex::new(HashMap::new()),
            service_fatal: Mutex::new(None),
            observer: Mutex::new(None),
            observer_head_uncertain: AtomicBool::new(false),
            capacity: ServerCapacity::default(),
            registry,
            unloadable: UnloadableConversationRecord::default(),
            owed_receipts: Mutex::new(HashMap::new()),
            #[cfg(test)]
            obligation_dispatch_work: ObligationDispatchWorkCounters::default(),
            now_override: AtomicU64::new(0),
        };
        if handler.config.pruning.is_none() {
            // DUR-018 R1: no section, no cut, said once by name at start.
            tracing::info!(
                "pruning is off: [participant.pruning] keep_acknowledged_behind_checkpoint is not configured, so no conversation row is ever cut"
            );
        }
        handler.restore_all_conversations()?;
        Ok(handler)
    }

    /// Reads the participant clock as Unix milliseconds: the wall clock, or
    /// the reading a test pinned with [`Self::pin_clock_ms`].
    fn now_ms(&self) -> Result<u64, facts::FactsError> {
        match self.now_override.load(Ordering::SeqCst) {
            0 => facts::now_unix_millis(),
            fixed => Ok(fixed),
        }
    }

    /// Pins the participant clock to a fixed Unix-millisecond reading, or
    /// releases it back to the wall clock when passed `0`.
    ///
    /// Harness-owned and test-only: it is the only writer of
    /// [`Self::now_override`]. Time-window tests set an explicit
    /// base and step it across the deterministic window states they assert.
    #[cfg(test)]
    pub(super) fn pin_clock_ms(&self, now_ms: u64) {
        self.now_override.store(now_ms, Ordering::SeqCst);
    }

    #[cfg(test)]
    pub(crate) fn obligation_dispatch_work_snapshot(&self) -> ObligationDispatchWorkSnapshot {
        self.obligation_dispatch_work.snapshot()
    }

    pub(super) fn current_service_fatal(
        &self,
    ) -> Result<Option<ParticipantServiceFatal>, ParticipantSemanticError> {
        let fatal = self
            .service_fatal
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant service fatal latch is poisoned".to_owned(),
            })?
            .clone();
        Ok(fatal)
    }

    pub(super) fn ensure_service_live(&self) -> Result<(), ParticipantSemanticError> {
        if let Some(fatal) = self.current_service_fatal()? {
            return Err(ParticipantSemanticError::ServiceFatal(fatal));
        }
        Ok(())
    }

    pub(super) fn latch_connection_fate_fatal(
        &self,
        open_sequence: u64,
        conversation_id: ConversationId,
    ) -> Result<ParticipantServiceFatal, ParticipantSemanticError> {
        let mut fatal =
            self.service_fatal
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: "participant service fatal latch is poisoned".to_owned(),
                })?;
        let selected = fatal
            .get_or_insert_with(|| ParticipantServiceFatal::ConnectionFateIntentIncomplete {
                open_sequence,
                conversation_id,
            })
            .clone();
        drop(fatal);
        Ok(selected)
    }

    /// Startup restore: enumerates every registered conversation and replays
    /// it, folding each conversation's server-scope contribution into the
    /// capacity ledger.
    ///
    /// A registry row whose conversation never got its genesis append (the
    /// crash window between the two ordered appends) replays empty and is
    /// evicted exactly like a refused probe.
    fn restore_all_conversations(&self) -> Result<(), ParticipantSemanticError> {
        // THE REGISTRY READ STAYS FATAL, and that is the property stated
        // precisely rather than an exception carved out of it. Containment
        // here is PER-CONVERSATION containment: without the registry read
        // there is no per-conversation subject to attribute a failure to, so a
        // failure with no conversation to name is the one thing that still
        // should stop the node.
        let started = Instant::now();
        #[cfg(test)]
        let conversation_ids = crate::boot_history_harness::during_stage("registry", || {
            self.registry.restore().map_err(|error| log_error(&error))
        })?;
        #[cfg(not(test))]
        let conversation_ids = self.registry.restore().map_err(|error| log_error(&error))?;
        let total = conversation_ids.len();
        tracing::info!(
            step = "participant restore",
            conversations = total,
            "participant restore started"
        );
        let mut unloadable = 0_usize;
        for (index, conversation_id) in conversation_ids.into_iter().enumerate() {
            let conversation_started = Instant::now();
            tracing::info!(
                step = "participant restore",
                conversation_id,
                done = index,
                total,
                elapsed_ms = elapsed_ms(started),
                "participant conversation restore starting"
            );
            // ONE FALLIBLE UNIT, not a guard per `?`. A guard per `?` is the
            // same defect with a longer fuse: the next `?` added to the
            // restore body is fatal again and nothing complains.
            #[cfg(test)]
            let outcome = crate::boot_history_harness::during_stage("authority", || {
                self.restore_one_conversation(conversation_id)
            });
            #[cfg(not(test))]
            let outcome = self.restore_one_conversation(conversation_id);
            let done = index.saturating_add(1);
            match outcome {
                Ok(rows) => tracing::info!(
                    step = "participant restore",
                    conversation_id,
                    done,
                    total,
                    log_rows = rows,
                    conversation_elapsed_ms = elapsed_ms(conversation_started),
                    elapsed_ms = elapsed_ms(started),
                    "participant conversation restored"
                ),
                Err(error) => {
                    unloadable = unloadable.saturating_add(1);
                    let refusal = self.record_unloadable(conversation_id, &error);
                    tracing::warn!(
                        step = "participant restore",
                        conversation_id,
                        done,
                        total,
                        elapsed_ms = elapsed_ms(started),
                        refusal = %refusal,
                        "participant conversation refused at restore"
                    );
                }
            }
        }
        tracing::info!(
            step = "participant restore",
            conversations = total,
            unloadable,
            elapsed_ms = elapsed_ms(started),
            "participant restore finished"
        );
        Ok(())
    }

    /// Restores exactly one conversation, and fails as exactly one
    /// conversation.
    ///
    /// Every propagation point in here belongs to a single conversation, so
    /// its `Err` is that conversation's answer and nobody else's. The caller
    /// is what makes that true — this function is deliberately allowed to be
    /// fallible, and deliberately not allowed to be fatal.
    ///
    /// Answers the number of operation-log rows the restored owner holds.
    fn restore_one_conversation(
        &self,
        conversation_id: ConversationId,
    ) -> Result<u64, ParticipantSemanticError> {
        let mut rows = 0_u64;
        {
            let cell = self.cell(conversation_id)?;
            let mut owner = cell
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: format!(
                        "participant conversation {conversation_id} owner lock is poisoned"
                    ),
                })?;
            if owner.is_none() {
                let log = OperationLog::new(Arc::clone(&self.store), conversation_id);
                let outbox_log = OutboxLog::new(Arc::clone(&self.store), conversation_id);
                let mut replayed = block_on(super::restore_checkpoint_rebuild::restore_at_boot(
                    self.store.as_ref(),
                    &log,
                    &outbox_log,
                    &self.config,
                    self.outbox_limits,
                ))
                .map_err(|error| bridge_error(&error))?
                .map_err(|error| ParticipantSemanticError::Internal {
                    message: format!("participant checkpoint restore failed: {error}"),
                })?;
                self.repair_loaded_fates(conversation_id, &mut replayed, &log, &outbox_log)?;
                let witnesses = replayed.observer_progress_witnesses();
                if !replayed.tokens.is_empty() || replayed.is_closed() {
                    self.reconcile_observer_progress(
                        conversation_id,
                        witnesses,
                        replayed.observer_progress_pruned_through(),
                        replayed.observer_progress,
                    )?;
                    replayed
                        .prune_durable_observer_witnesses()
                        .map_err(|error| state_error(&error))?;
                } else if !witnesses.is_empty() {
                    return Err(ParticipantSemanticError::Internal {
                        message: format!(
                            "unenrolled conversation {conversation_id} projected observer progress"
                        ),
                    });
                }
                let now = self
                    .now_ms()
                    .map_err(|error| ParticipantSemanticError::Internal {
                        message: format!("participant clock read failed: {error}"),
                    })?;
                replayed.prune_expired_provenance(u128::from(now));
                let contribution = replayed
                    .capacity_contribution(u128::from(now))
                    .map_err(|error| state_error(&error))?;
                self.capacity
                    .fold_conversation(conversation_id, contribution)
                    .map_err(|error| state_error(&error))?;
                let durably_empty = replayed.next_log_sequence == 0;
                if durably_empty {
                    drop(owner);
                    self.evict_uncommitted(conversation_id, &cell)?;
                    return Ok(0);
                }
                // R-BOOT-DRAIN (F8B §6.2): empty this conversation's
                // immutable-candidate lane before any retained connection-fate
                // `Open` replays, and before the owner is installed. The drain
                // lives HERE and not in `replay_and_repair`, which is also the
                // load a request runs for a conversation with no live owner.
                let before_drain = replayed.next_log_sequence;
                let verdict =
                    self.drain_restored_candidate_lane(conversation_id, &mut replayed, &log);
                let drained = verdict.drained_any();
                verdict.observe()?;
                if drained {
                    self.commit_boot_drain(conversation_id, &mut replayed, &log, before_drain)?;
                }
                rows = replayed.next_log_sequence;
                *owner = Some(replayed);
            }
            drop(owner);
        }
        Ok(rows)
    }

    /// The commit tail a boot drain owes once its heads are committed: the
    /// observer reconciliation and capacity re-fold every live commit runs,
    /// then the checkpoint the crossing owes.
    fn commit_boot_drain(
        &self,
        conversation_id: ConversationId,
        replayed: &mut ConversationAuthority,
        log: &OperationLog,
        before_drain: u64,
    ) -> Result<(), ParticipantSemanticError> {
        // The drain appended durable rows and completed each one's
        // extension row in place, exactly as a live commit does, so
        // the restored owner is its log's current state. What the
        // drain still owes is the commit tail every live commit
        // runs: the observer reconciliation for the projections it
        // recorded and the capacity re-fold for the binding slot it
        // released.
        let outbox_log = OutboxLog::new(Arc::clone(&self.store), conversation_id);
        let batch = RefCell::new(Some(OperationBatch::new(&checkpoint_conversation_route(
            conversation_id,
        ))));
        #[cfg(test)]
        let flat_legacy = super::log::legacy_rows::armed();
        #[cfg(not(test))]
        let flat_legacy = false;
        let appender = LogAppender::new(
            log,
            &self.registry,
            &outbox_log,
            self.config.pruning,
            (!flat_legacy).then_some(&batch),
            0,
        );
        self.complete_live_commit(conversation_id, replayed, &appender)?;
        // The commit tail every live commit runs. The drain's own
        // rows were committed head by head, so nothing staged is
        // not "nothing owed": when the drain crossed a period or
        // sealed the conversation, the checkpoint that crossing
        // owes (`checkpoint_due`, the closed marker included) is
        // captured, committed and installed exactly as a live
        // seal's is through `finish_operation`.
        let due = OperationBatch::checkpoint_due(
            before_drain,
            replayed.next_log_sequence,
            replayed.is_closed(),
        )
        .map_err(|error| log_error(&OperationLogError::from(error)))?;
        if due {
            appender
                .commit_with_checkpoint(replayed, before_drain, &self.config)
                .map_err(|error| state_error(&error))?;
        } else if appender.has_staged().map_err(|error| log_error(&error))? {
            appender
                .commit_staged()
                .map_err(|error| log_error(&error))?;
        }
        Ok(())
    }

    /// Records one conversation as unloadable and renders the NAMED refusal
    /// every later request on it is answered with.
    ///
    /// This is the attribution half of containment, and it is deliberately the
    /// same call on both sides: the boot loop records and continues, the
    /// request path records and refuses, and both produce a refusal carrying
    /// the conversation id. An operator therefore never has to correlate a
    /// bare decode failure against a log to learn which conversation stopped.
    fn record_unloadable(
        &self,
        conversation_id: ConversationId,
        error: &ParticipantSemanticError,
    ) -> ParticipantSemanticError {
        let reason = error.to_string();
        let class = error.class();
        // The operator's first tell, and it carries the class as its own field:
        // the refusal text of a failed load ("expected value at line 1 column
        // 1") names no class at all, so a reader with only the message has to
        // discriminate on a substring or not at all.
        tracing::error!(
            conversation_id,
            class,
            reason = %reason,
            "CONTAINMENT: participant conversation is unloadable and is refused on its own; every \
             other conversation keeps being served"
        );
        if !self.unloadable.record(UnloadableConversation {
            conversation_id,
            class,
            reason: reason.clone(),
        }) {
            // Never a silent drop: a refusal that is reported but not retained
            // is a different, weaker state than one that was retained, and an
            // operator reading `GET /unloadable-conversations` must not be told
            // a shorter story than the log tells.
            tracing::error!(
                conversation_id,
                "the unloadable-conversation record is poisoned; this refusal is reported but not \
                 retained"
            );
        }
        ParticipantSemanticError::ConversationUnloadable {
            conversation_id,
            reason,
        }
    }

    /// Retires an unloadable record once the conversation has actually loaded,
    /// so the map never keeps reporting a conversation that recovered.
    fn clear_unloadable(&self, conversation_id: ConversationId) {
        if !self.unloadable.retire(conversation_id) {
            tracing::error!(
                conversation_id,
                "the unloadable-conversation record is poisoned; a conversation that recovered may \
                 still be reported as unloadable"
            );
        }
    }

    /// The shared refused-load record, for publication onto the operator read
    /// surface.
    ///
    /// The clone shares the record rather than copying it, so a refusal
    /// recorded after publication is still the answer the surface gives. This
    /// is the only way out of the handler for the containment record, and the
    /// server's startup path (`server/runtime.rs`) is its one production
    /// caller.
    pub(crate) fn unloadable_record(&self) -> UnloadableConversationRecord {
        self.unloadable.clone()
    }

    /// Every conversation this node has refused as unloadable, against the
    /// load failure's own text.
    ///
    /// In-process view for the tests that assert on containment attribution.
    /// The operator's view of the same record is
    /// `GET /unloadable-conversations`, served from the clone published by
    /// [`Self::unloadable_record`]; this accessor exists so a test can read the
    /// record without standing up an HTTP endpoint.
    #[cfg(test)]
    pub(super) fn unloadable_conversations(&self) -> BTreeMap<ConversationId, String> {
        self.unloadable
            .snapshot()
            .into_iter()
            .map(|entry| (entry.conversation_id, entry.reason))
            .collect()
    }

    pub(super) fn registered_conversation_ids(
        &self,
    ) -> Result<Vec<ConversationId>, ParticipantSemanticError> {
        self.registry.restore().map_err(|error| log_error(&error))
    }

    pub(super) fn existing_cell(
        &self,
        conversation_id: ConversationId,
    ) -> Result<Option<Arc<Mutex<Option<ConversationAuthority>>>>, ParticipantSemanticError> {
        let conversations =
            self.conversations
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: "participant conversation registry lock is poisoned".to_owned(),
                })?;
        Ok(conversations.get(&conversation_id).map(Arc::clone))
    }

    pub(super) fn cell(
        &self,
        conversation_id: ConversationId,
    ) -> Result<Arc<Mutex<Option<ConversationAuthority>>>, ParticipantSemanticError> {
        let mut conversations =
            self.conversations
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: "participant conversation registry lock is poisoned".to_owned(),
                })?;
        let cell = Arc::clone(
            conversations
                .entry(conversation_id)
                .or_insert_with(|| Arc::new(Mutex::new(None))),
        );
        drop(conversations);
        Ok(cell)
    }

    pub(super) fn with_conversation_impact<T>(
        &self,
        conversation_id: ConversationId,
        impact: &mut DispatchImpactAccumulator,
        operation: impl FnOnce(
            &mut ConversationAuthority,
            &dyn DurableAppend,
            &mut DispatchImpactAccumulator,
        ) -> Result<T, StateError>,
    ) -> Result<T, ParticipantSemanticError> {
        self.with_conversation_reconciliation(
            conversation_id,
            true,
            Some(impact),
            |authority, appender, impact| {
                impact.map_or_else(
                    || Err(StateError::invariant("impact owner is unavailable")),
                    |impact| operation(authority, appender, impact),
                )
            },
        )
    }

    /// Runs a fate-source append and reconciles its exact Unit 2 projection before
    /// retaining the transitioned live owner. Died/Detached joined the exhaustive
    /// v3 replay projection pass with W1b, so this boundary must use the same
    /// post-append repair as semantic sources to keep live and cold owners equal.
    pub(super) fn with_conversation_fate_source<T>(
        &self,
        conversation_id: ConversationId,
        impact: Option<&mut DispatchImpactAccumulator>,
        operation: impl FnOnce(
            &mut ConversationAuthority,
            &dyn DurableAppend,
            Option<&mut DispatchImpactAccumulator>,
        ) -> Result<T, StateError>,
    ) -> Result<T, ParticipantSemanticError> {
        self.with_conversation_reconciliation(conversation_id, true, impact, operation)
    }

    fn operation_timestamp(&self) -> Result<u64, ParticipantSemanticError> {
        self.now_ms()
            .map_err(|error| ParticipantSemanticError::Internal {
                message: format!("participant clock read failed: {error}"),
            })?
            .checked_mul(1_000_000)
            .ok_or_else(|| ParticipantSemanticError::Internal {
                message: "participant timestamp exceeds nanoseconds".to_owned(),
            })
    }

    fn with_conversation_reconciliation<T>(
        &self,
        conversation_id: ConversationId,
        reconcile_appended_source: bool,
        mut impact: Option<&mut DispatchImpactAccumulator>,
        operation: impl FnOnce(
            &mut ConversationAuthority,
            &dyn DurableAppend,
            Option<&mut DispatchImpactAccumulator>,
        ) -> Result<T, StateError>,
    ) -> Result<T, ParticipantSemanticError> {
        let cell = self.cell(conversation_id)?;
        let mut owner = lock_owner(&cell, conversation_id)?;
        let log = OperationLog::new(Arc::clone(&self.store), conversation_id);
        let outbox_log = OutboxLog::new(Arc::clone(&self.store), conversation_id);
        let mut owed = Vec::new();
        if owner.is_none() {
            // The request-side half of containment. A conversation the node
            // cannot load refuses HERE, by name, instead of surfacing a bare
            // decode failure that names no subject at all.
            let replayed = self
                .replay_and_repair(conversation_id, &log)
                .map_err(|error| self.record_unloadable(conversation_id, &error))?;
            self.clear_unloadable(conversation_id);
            owed.extend(self.settle_owed_receipts(conversation_id, replayed.next_log_sequence)?);
            *owner = Some(replayed);
        }
        let Some(authority) = owner.as_mut() else {
            return Err(ParticipantSemanticError::Internal {
                message: format!("participant conversation {conversation_id} owner is absent"),
            });
        };
        let timestamp = self.operation_timestamp()?;
        let batch = RefCell::new(Some(OperationBatch::new(&checkpoint_conversation_route(
            conversation_id,
        ))));
        // A test that arms legacy rows writes them where a legacy store in the
        // field holds them: flat, one append per row, never the routed batch.
        #[cfg(test)]
        let flat_legacy = super::log::legacy_rows::armed();
        #[cfg(not(test))]
        let flat_legacy = false;
        let appender = LogAppender::new(
            &log,
            &self.registry,
            &outbox_log,
            self.config.pruning,
            (!flat_legacy).then_some(&batch),
            timestamp,
        );
        let starting_log_sequence = authority.next_log_sequence;
        let operation_result =
            operation(authority, &appender, impact.as_deref_mut()).and_then(|value| {
                appender.finish_operation(
                    authority,
                    starting_log_sequence,
                    reconcile_appended_source,
                    &self.config,
                )?;
                Ok(value)
            });
        let mut receipts = owed;
        let (result, durably_empty) = match operation_result {
            Ok(value)
                if reconcile_appended_source
                    && authority.next_log_sequence > starting_log_sequence =>
            {
                // The v2 source barrier crossed. Complete its exact Unit 2
                // projection under this same conversation lock before the
                // caller can publish the correlated terminal response.
                match self.complete_appended_source(conversation_id, authority, &appender) {
                    Ok(()) => {
                        install_staged(impact.as_deref_mut());
                        let durably_empty = authority.next_log_sequence == 0;
                        (Ok(value), durably_empty)
                    }
                    Err(error) => {
                        // The committed rows are durable; this owner may not
                        // be. The next request loads the conversation, which
                        // completes any row this one left owed.
                        install_staged(impact.as_deref_mut());
                        receipts.extend(std::mem::take(&mut authority.lifecycle_receipts));
                        *owner = None;
                        (
                            Err(ParticipantSemanticError::WriteOutcomeUnknown {
                                message: format!("committed participant follow-up failed: {error}"),
                            }),
                            false,
                        )
                    }
                }
            }
            Ok(value) => {
                install_staged(impact.as_deref_mut());
                let durably_empty = authority.next_log_sequence == 0;
                (Ok(value), durably_empty)
            }
            Err(error) => {
                // Pending rows and receipts have no publication authority.
                // The next request restores the store's actual committed state.
                if let Some(impact) = impact {
                    impact.discard_staged();
                }
                let durably_empty =
                    self.fail_operation(conversation_id, starting_log_sequence, authority, &log)?;
                *owner = None;
                (
                    Err(state_error_with_issuance(&error, log.issuance())),
                    durably_empty,
                )
            }
        };
        if let Some(authority) = owner.as_mut() {
            receipts.extend(std::mem::take(&mut authority.lifecycle_receipts));
        }
        self.emit_lifecycle_receipts(conversation_id, receipts, result.as_ref().err())?;
        drop(owner);
        if durably_empty {
            self.evict_uncommitted(conversation_id, &cell)?;
        }
        result
    }

    /// Settles a failed operation's receipts and answers whether the
    /// conversation is durably empty. A batch that was never issued left
    /// nothing behind, so the conversation is empty exactly when its log is.
    /// A batch that was issued and whose landing nothing proved leaves the
    /// conversation possibly written, and its receipts owed to the next
    /// load, which decides.
    fn fail_operation(
        &self,
        conversation_id: ConversationId,
        starting_log_sequence: u64,
        authority: &mut ConversationAuthority,
        log: &OperationLog,
    ) -> Result<bool, ParticipantSemanticError> {
        if !log.issuance().was_submitted() {
            return Ok(authority.next_log_sequence == 0);
        }
        self.hold_owed_receipts(
            conversation_id,
            starting_log_sequence,
            std::mem::take(&mut authority.lifecycle_receipts),
        )?;
        Ok(false)
    }

    /// Holds `receipts` for `conversation_id` until its next load shows
    /// whether the operation that started at `starting_log_sequence` landed.
    fn hold_owed_receipts(
        &self,
        conversation_id: ConversationId,
        starting_log_sequence: u64,
        receipts: Vec<LifecycleReceipt>,
    ) -> Result<(), ParticipantSemanticError> {
        if receipts.is_empty() {
            return Ok(());
        }
        let mut owed = self.lock_owed_receipts()?;
        owed.insert(
            conversation_id,
            OwedReceipts {
                starting_log_sequence,
                receipts,
            },
        );
        drop(owed);
        Ok(())
    }

    /// The receipts owed to `conversation_id`, now that a load found its log
    /// head at `loaded_log_head`: all of them when the operation that owed
    /// them landed rows past where it started, none when it did not. Either
    /// way they are no longer held.
    fn settle_owed_receipts(
        &self,
        conversation_id: ConversationId,
        loaded_log_head: u64,
    ) -> Result<Vec<LifecycleReceipt>, ParticipantSemanticError> {
        let mut owed = self.lock_owed_receipts()?;
        let held = owed.remove(&conversation_id);
        drop(owed);
        Ok(held
            .filter(|held| loaded_log_head > held.starting_log_sequence)
            .map(|held| held.receipts)
            .unwrap_or_default())
    }

    /// Emits the receipts owed to `conversation_id`, if its load at
    /// `loaded_log_head` shows that the operation that owed them landed: the
    /// one call for a load path that installs an owner outside an operation.
    pub(super) fn emit_owed_receipts(
        &self,
        conversation_id: ConversationId,
        loaded_log_head: u64,
    ) -> Result<(), ParticipantSemanticError> {
        let owed = self.settle_owed_receipts(conversation_id, loaded_log_head)?;
        if owed.is_empty() {
            return Ok(());
        }
        self.emit_lifecycle_receipts(conversation_id, owed, None)
    }

    fn lock_owed_receipts(
        &self,
    ) -> Result<MutexGuard<'_, HashMap<ConversationId, OwedReceipts>>, ParticipantSemanticError>
    {
        self.owed_receipts
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "owed lifecycle receipt lock poisoned".to_owned(),
            })
    }

    fn emit_lifecycle_receipts(
        &self,
        conversation_id: ConversationId,
        receipts: Vec<LifecycleReceipt>,
        operation_error: Option<&ParticipantSemanticError>,
    ) -> Result<(), ParticipantSemanticError> {
        if let Some(signals) = self.signals.get() {
            if let Err(error) = signals.lifecycle_batch(
                conversation_id,
                receipts
                    .into_iter()
                    .map(|receipt| (receipt.generation, receipt.transition)),
            ) {
                if let Some(operation_error) = operation_error {
                    tracing::error!(%operation_error, "operation and lifecycle batch both failed");
                } else {
                    tracing::error!(conversation_id, %error, "committed lifecycle batch failed");
                    return Ok(());
                }
                return Err(ParticipantSemanticError::WriteOutcomeUnknown {
                    message: format!("committed lifecycle batch failed: {error}"),
                });
            }
        }
        Ok(())
    }

    /// Completes the sources this operation appended.
    ///
    /// Every source writes its own Unit 2 extension row in place right after
    /// its base row, and a failed write already refused the operation, so a
    /// commit that reaches here owes nothing and needs only the commit tail.
    /// A nonzero owed count is an invariant break: the commit refuses by name,
    /// never by rebuilding the conversation from its log, and the next
    /// request's load writes the owed row.
    fn complete_appended_source(
        &self,
        conversation_id: ConversationId,
        authority: &mut ConversationAuthority,
        appender: &LogAppender<'_>,
    ) -> Result<(), ParticipantSemanticError> {
        let owed = appender.outstanding_extension_rows();
        if owed != 0 {
            return Err(state_error(&StateError::ExtensionRowsOwed {
                conversation_id,
                owed,
            }));
        }
        self.complete_live_commit(conversation_id, authority, appender)
    }

    /// Completes one in-process commit without re-deriving the prefix.
    ///
    /// Board #60 §3c. This is the tail of [`Self::replay_and_repair`] applied
    /// to the live owner — the same calls, in the same order, minus the three
    /// passes whose only job was to rebuild a prefix this owner never lost:
    ///
    /// - `repair_pending_specific_fates` is KEPT. It is not a load-time repair
    ///   in disguise: a live Died/Detached commit can leave a pending specific
    ///   fate whose finalizer must be appended before the response publishes,
    ///   and today's post-append replay is what performs it.
    /// - `reconcile_observer_progress` is KEPT, over the owner's OWN witness
    ///   vector, which is now retained rather than drained. The planner reads
    ///   the whole source history (it validates the durable observer prefix
    ///   against it), and the retained vector is that history — the same
    ///   vector, source for source, that a from-zero replay would rebuild.
    /// - `prune_expired_provenance` and the capacity fold are KEPT: both are
    ///   per-touch clock work over the current owner, not prefix derivation.
    ///
    /// Deliberately NOT carried over: `validate_operation_schema`, the base-log
    /// replay, the extension merge, `validate_replayed_seal` and
    /// `reconcile_load_end_marker_anchors`. The first three re-derive a prefix
    /// the live owner already holds; the last two are load-end repairs of crash
    /// residue, and a source committed in-process under this lock has no crash
    /// residue to repair.
    fn complete_live_commit(
        &self,
        conversation_id: ConversationId,
        authority: &mut ConversationAuthority,
        appender: &LogAppender<'_>,
    ) -> Result<(), ParticipantSemanticError> {
        authority
            .repair_pending_specific_fates(appender)
            .map_err(|error| state_error(&error))?;
        self.reconcile_live_observer(conversation_id, authority)?;
        let now = self
            .now_ms()
            .map_err(|error| ParticipantSemanticError::Internal {
                message: format!("participant clock read failed: {error}"),
            })?;
        let now = u128::from(now);
        authority.prune_expired_provenance(now);
        let contribution = authority
            .capacity_contribution(now)
            .map_err(|error| state_error(&error))?;
        self.capacity
            .fold_conversation(conversation_id, contribution)
            .map_err(|error| state_error(&error))?;
        Ok(())
    }

    /// Reconciles a live owner's observer progress with the durable observer
    /// store and prunes the witnesses that are now durable.
    pub(super) fn reconcile_live_observer(
        &self,
        conversation_id: ConversationId,
        authority: &mut ConversationAuthority,
    ) -> Result<(), ParticipantSemanticError> {
        let observer_progress = authority.observer_progress;
        let pruned_through = authority.observer_progress_pruned_through();
        let witnesses = authority.observer_progress_witnesses();
        if !authority.tokens.is_empty() || authority.is_closed() {
            self.reconcile_observer_progress(
                conversation_id,
                witnesses,
                pruned_through,
                observer_progress,
            )?;
            authority
                .prune_durable_observer_witnesses()
                .map_err(|error| state_error(&error))?;
        } else if !witnesses.is_empty() {
            return Err(ParticipantSemanticError::Internal {
                message: format!(
                    "unenrolled conversation {conversation_id} projected observer progress"
                ),
            });
        }
        Ok(())
    }

    /// Completes the pending binding fates a load found, through one routed
    /// batch on the conversation's route committed once, the way every
    /// operation commits. The flat legacy-rows arm wrote them flat beside
    /// the routed rows, where the store refused them by sequence and the
    /// load failed; a test that arms legacy rows still writes them flat.
    fn repair_loaded_fates(
        &self,
        conversation_id: ConversationId,
        replayed: &mut ConversationAuthority,
        log: &OperationLog,
        outbox_log: &OutboxLog,
    ) -> Result<(), ParticipantSemanticError> {
        let timestamp = self.operation_timestamp()?;
        let batch = RefCell::new(Some(OperationBatch::new(&checkpoint_conversation_route(
            conversation_id,
        ))));
        #[cfg(test)]
        let flat_legacy = super::log::legacy_rows::armed();
        #[cfg(not(test))]
        let flat_legacy = false;
        let appender = LogAppender::new(
            log,
            &self.registry,
            outbox_log,
            self.config.pruning,
            (!flat_legacy).then_some(&batch),
            timestamp,
        );
        replayed
            .repair_pending_specific_fates(&appender)
            .map_err(|error| state_error(&error))?;
        if appender.has_staged().map_err(|error| log_error(&error))? {
            appender
                .commit_staged()
                .map_err(|error| log_error(&error))?;
        }
        Ok(())
    }

    /// Cold-replays one conversation's durable log and repairs its observer
    /// registration.
    ///
    /// An enrolled conversation whose durable observer `Track` row was lost
    /// to a crash between the enrollment append and the tracking append is
    /// re-registered idempotently here, so observer recovery is derivable
    /// from the conversation log itself on any first touch.
    pub(super) fn replay_and_repair(
        &self,
        conversation_id: ConversationId,
        log: &OperationLog,
    ) -> Result<ConversationAuthority, ParticipantSemanticError> {
        let outbox_log = OutboxLog::new(Arc::clone(&self.store), conversation_id);
        // The load the boot runs: the validated checkpoint and its routed
        // tail, or a full rebuild through the routed route. The legacy replay
        // walked the flat stream alone and saw none of the rows a live commit
        // stages routed, so an owner discarded after a failed operation came
        // back from its durable log empty, and a retry on a bound member was
        // answered ParticipantUnknown.
        let mut replayed = block_on(super::restore_checkpoint_rebuild::restore_at_boot(
            self.store.as_ref(),
            log,
            &outbox_log,
            &self.config,
            self.outbox_limits,
        ))
        .map_err(|error| bridge_error(&error))?
        .map_err(|error| ParticipantSemanticError::Internal {
            message: format!("participant checkpoint restore failed: {error}"),
        })?;
        self.repair_loaded_fates(conversation_id, &mut replayed, log, &outbox_log)?;
        let pruned_through = replayed.observer_progress_pruned_through();
        let observer_witnesses = replayed.observer_progress_witnesses();
        // F8B R-SEAL (§6.6). This is the THIRD site that reads `tokens` empty
        // as "never enrolled" — the two the ruling names are the replay
        // invariant twins. A Closed conversation is the state that proxy never
        // anticipated: no tokens, but a log full of records, terminals and
        // drain rows whose observer-progress witnesses are durable truth. It
        // must reconcile exactly as an enrolled conversation does, or the
        // handler's observer state is left behind its own durable log — the
        // failure §6.2's post-drain reconciliation exists to prevent. Bare
        // tokens-empty-with-witnesses REMAINS a refusal for the never-enrolled
        // shape.
        if !replayed.tokens.is_empty() || replayed.is_closed() {
            self.reconcile_observer_progress(
                conversation_id,
                observer_witnesses,
                pruned_through,
                replayed.observer_progress,
            )?;
            replayed
                .prune_durable_observer_witnesses()
                .map_err(|error| state_error(&error))?;
        } else if !observer_witnesses.is_empty() {
            return Err(ParticipantSemanticError::Internal {
                message: format!(
                    "unenrolled conversation {conversation_id} projected observer progress"
                ),
            });
        }
        // Request-time expiry over the replayed state (replay rebuilds every
        // retired rotation's fingerprint; the ones past their deadlines are
        // dropped under this touch's clock read), then fold the
        // conversation's complete server-scope contribution into the ledger
        // — a replace, so a discarded-and-replayed owner never double
        // counts and the ledger self-heals from durable truth.
        let now = self
            .now_ms()
            .map_err(|error| ParticipantSemanticError::Internal {
                message: format!("participant clock read failed: {error}"),
            })?;
        let now = u128::from(now);
        replayed.prune_expired_provenance(now);
        let contribution = replayed
            .capacity_contribution(now)
            .map_err(|error| state_error(&error))?;
        self.capacity
            .fold_conversation(conversation_id, contribution)
            .map_err(|error| state_error(&error))?;
        Ok(replayed)
    }

    /// Runs the frozen pre-W3 aggregate reference without installing any
    /// owner, observer, capacity, or publication state.
    #[cfg(test)]
    pub(super) fn replay_aggregate_reference(
        &self,
        conversation_id: ConversationId,
        log: &OperationLog,
    ) -> Result<ConversationAuthority, ParticipantSemanticError> {
        let outbox_log = OutboxLog::new(Arc::clone(&self.store), conversation_id);
        let extension_rows = block_on(outbox_log.read_all())
            .map_err(|error| bridge_error(&error))?
            .map_err(|error| outbox_log_error(&error))?;
        block_on(ConversationAuthority::replay_aggregate_reference(
            conversation_id,
            log,
            &outbox_log,
            extension_rows,
            &self.config,
            self.outbox_limits,
        ))
        .map_err(|error| bridge_error(&error))?
        .map_err(|error| state_error(&error))
    }

    /// Removes a conversation's registry cell after a durably empty touch.
    ///
    /// Only the exact cell this operation used is removed (a racing request
    /// may have installed a fresh cell already); a concurrent holder of the
    /// evicted cell stays correct because every durable append is optimistic
    /// on its exact sequence and cold replay is the source of truth.
    pub(super) fn evict_uncommitted(
        &self,
        conversation_id: ConversationId,
        cell: &Arc<Mutex<Option<ConversationAuthority>>>,
    ) -> Result<(), ParticipantSemanticError> {
        let mut conversations =
            self.conversations
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: "participant conversation registry lock is poisoned".to_owned(),
                })?;
        if let Some(existing) = conversations.get(&conversation_id) {
            if Arc::ptr_eq(existing, cell) {
                conversations.remove(&conversation_id);
            }
        }
        drop(conversations);
        Ok(())
    }

    /// Number of live conversation registry cells (test observability).
    #[cfg(test)]
    pub(super) fn registry_len(&self) -> usize {
        self.conversations
            .lock()
            .map_or(usize::MAX, |conversations| conversations.len())
    }

    /// Drops only volatile participant/observer owners for cold-first-touch tests.
    #[cfg(test)]
    pub(super) fn discard_owners_for_test(&self) -> Result<(), ParticipantSemanticError> {
        self.conversations
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant conversation registry lock is poisoned".to_owned(),
            })?
            .clear();
        self.observer
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "observer recovery aggregate lock is poisoned".to_owned(),
            })?
            .take();
        Ok(())
    }

    /// Runs one `OperatorCredentialReissue` at the serialized
    /// participant-state point (R18 amendment A7, §0.18).
    ///
    /// It reaches the conversation through the SAME owner lock, cold-replay,
    /// and post-commit reconciliation seam as enrollment and credential
    /// attach — never a second path — so every guard reads the same exact
    /// state those operations read, and an unknown conversation is evicted
    /// with no residue exactly as a refused probe already is.
    ///
    /// # Errors
    ///
    /// Returns [`ParticipantSemanticError`] when the operation could not be
    /// DECIDED (a latched service fatal, an unloadable conversation, a durable
    /// failure). Every decided refusal is an `Ok`.
    pub(crate) fn operator_credential_reissue(
        &self,
        request: OperatorCredentialReissueRequest,
    ) -> Result<OperatorCredentialReissueOutcome, ParticipantSemanticError> {
        self.ensure_service_live()?;
        let now_ms = self
            .now_ms()
            .map_err(|error| ParticipantSemanticError::Internal {
                message: format!("participant clock read failed: {error}"),
            })?;
        self.with_conversation_fate_source(
            request.conversation_id,
            None,
            |authority, appender, _| {
                authority.apply_operator_credential_reissue(request, now_ms, appender)
            },
        )
    }

    pub(super) fn operation_facts(
        &self,
        context: ParticipantConnectionContext,
        conversation_id: ConversationId,
        conversations: &ParticipantConnectionConversations,
    ) -> Result<OperationFacts, ParticipantSemanticError> {
        let now_ms = self
            .now_ms()
            .map_err(|error| ParticipantSemanticError::Internal {
                message: format!("participant clock read failed: {error}"),
            })?;
        // The connection map only grows through capacity commits, so its
        // occupancy always fits the validated nonzero signed limit; a counter
        // rejection here is genuine internal drift and fails closed.
        let connection_capacity = CapacityCounter::try_new(
            self.config.max_semantic_conversations_per_connection,
            conversations.occupied(),
        )
        .map_err(|error| ParticipantSemanticError::Internal {
            message: format!(
                "connection-conversation occupancy disagrees with its signed limit: {error:?}"
            ),
        })?;
        Ok(OperationFacts {
            receiving_incarnation: context.connection_incarnation(),
            now_ms,
            identity_slots: self.config.identity_slots,
            attach_receipt_ttl_ms: self.config.attach_receipt_ttl_ms,
            receipt_provenance_ttl_ms: self.config.receipt_provenance_ttl_ms,
            receipt_limits: ReceiptCapacityLimits::from_config(&self.config),
            connection_tracking: conversations.tracking(conversation_id),
            connection_capacity,
        })
    }
}

/// The operator surface reaches the participant through the handler itself
/// (R18 amendment A7, §0.18): possession of the operator surface is the
/// authority, and this impl is the only bridge between it and the serialized
/// participant-state point.
impl OperatorCredentialReissuer for ProductionParticipantHandler {
    fn reissue(
        &self,
        request: OperatorCredentialReissueRequest,
    ) -> Result<OperatorCredentialReissueOutcome, OperatorCredentialReissueError> {
        self.operator_credential_reissue(request)
            .map_err(|error| OperatorCredentialReissueError {
                message: error.to_string(),
            })
    }
}

/// Bridges the synchronous state seam onto the async durable log, and keeps
/// the conversation registry complete by construction: the one
/// conversation-creating append (genesis at sequence zero) is preceded by a
/// durable registry row, so startup can enumerate every conversation stream
/// that exists.
pub(super) struct LogAppender<'a> {
    pub(super) log: &'a OperationLog,
    pub(super) registry: &'a ConversationRegistry,
    pub(super) conversation_id: ConversationId,
    /// Board #60 §3c. The Unit 2 extension log a committed source completes
    /// itself into, under the same conversation lock as its base append.
    pub(super) outbox_log: &'a OutboxLog,
    /// `[participant.pruning]`, `None` when pruning is off (DUR-018).
    pub(super) pruning: Option<crate::config::types::PruningConfig>,
    /// Owed-minus-written Unit 2 extension rows for the operation this
    /// appender is serving. Incremented by every appended row that owes one
    /// ([`owes_extension_row`]), decremented by every row written in place.
    /// Zero at the end of the operation is the OUTCOME GATE that lets the
    /// commit path skip its from-zero replay.
    pub(super) outstanding_extension_rows: Cell<u64>,
    pub(super) batch: Option<&'a RefCell<Option<OperationBatch>>>,
    pub(super) timestamp: u64,
}

/// The conversation's owner cell, locked; a poisoned lock refuses by name.
fn lock_owner(
    cell: &Mutex<Option<ConversationAuthority>>,
    conversation_id: ConversationId,
) -> Result<MutexGuard<'_, Option<ConversationAuthority>>, ParticipantSemanticError> {
    cell.lock().map_err(|_| ParticipantSemanticError::Internal {
        message: format!("participant conversation {conversation_id} owner lock is poisoned"),
    })
}

/// Installs the effects an operation staged on its impact accumulator, when
/// the caller gave one.
fn install_staged(impact: Option<&mut DispatchImpactAccumulator>) {
    if let Some(impact) = impact {
        impact.install_staged();
    }
}

fn checkpoint_state_error(error: super::restore_checkpoint::CheckpointError) -> StateError {
    OperationLogError::RoutedStream(RoutedStreamRefusal::Checkpoint(Box::new(error))).into()
}

impl<'a> LogAppender<'a> {
    /// An appender over `log`: routed through `batch` when one is given, flat
    /// (one append per row) when none is.
    pub(super) const fn new(
        log: &'a OperationLog,
        registry: &'a ConversationRegistry,
        outbox_log: &'a OutboxLog,
        pruning: Option<crate::config::types::PruningConfig>,
        batch: Option<&'a RefCell<Option<OperationBatch>>>,
        timestamp: u64,
    ) -> Self {
        Self {
            log,
            registry,
            conversation_id: log.conversation_id(),
            outbox_log,
            pruning,
            outstanding_extension_rows: Cell::new(0),
            batch,
            timestamp,
        }
    }

    fn finish_operation(
        &self,
        authority: &mut ConversationAuthority,
        starting_log_sequence: u64,
        reconcile_appended_source: bool,
        config: &ParticipantConfig,
    ) -> Result<(), StateError> {
        if reconcile_appended_source && authority.next_log_sequence > starting_log_sequence {
            authority.repair_pending_specific_fates(self)?;
            let owed = self.outstanding_extension_rows();
            if owed != 0 {
                return Err(StateError::ExtensionRowsOwed {
                    conversation_id: self.conversation_id,
                    owed,
                });
            }
        }
        if self.has_staged()? {
            self.commit_with_checkpoint(authority, starting_log_sequence, config)?;
        }
        Ok(())
    }

    /// Commits what is staged together with the checkpoint the crossing from
    /// `starting_log_sequence` owes (`checkpoint_due`: a period boundary or the
    /// closed marker), and installs the checkpoint metadata the commit proved.
    fn commit_with_checkpoint(
        &self,
        authority: &mut ConversationAuthority,
        starting_log_sequence: u64,
        config: &ParticipantConfig,
    ) -> Result<(), StateError> {
        let (metadata, audit) =
            self.staged_checkpoint_metadata(authority, starting_log_sequence, config)?;
        let metadata =
            self.prepare_crossing_checkpoint(authority, starting_log_sequence, metadata, &audit)?;
        self.commit_staged()?;
        authority
            .install_checkpoint_metadata(metadata)
            .map_err(checkpoint_state_error)
    }

    fn prepare_crossing_checkpoint(
        &self,
        authority: &mut ConversationAuthority,
        before: u64,
        mut metadata: super::restore_from_checkpoint::CheckpointMetadata,
        audit: &super::fenced_attach_terminal::ComposedTerminalValidation,
    ) -> Result<super::restore_from_checkpoint::CheckpointMetadata, StateError> {
        use super::restore_checkpoint_history::CheckpointHistoryRoot;
        let mut root =
            CheckpointHistoryRoot::decode_canonical(metadata.history_root(), self.conversation_id)
                .map_err(|source| {
                    checkpoint_state_error(super::restore_checkpoint::CheckpointError::LiveState {
                        source: Box::new(source),
                    })
                })?;
        let due =
            OperationBatch::checkpoint_due(before, authority.next_log_sequence, authority.closed)
                .map_err(OperationLogError::from)?;
        if due {
            let store = self.log.store();
            let history = block_on(
                super::restore_checkpoint_history_capture::capture_identity_history(
                    store.as_ref(),
                    authority,
                    root,
                ),
            )?
            .map_err(|error| checkpoint_state_error(error.checkpoint()))?;
            root = history.root;
            metadata = metadata
                .with_history_root(root.encode_canonical())
                .map_err(checkpoint_state_error)?;
            let pending = self.batch.ok_or(OperationLogError::RoutedStream(
                RoutedStreamRefusal::BatchRefused,
            ))?;
            let mut held = pending
                .try_borrow_mut()
                .map_err(|_| OperationLogError::RoutedStream(RoutedStreamRefusal::BatchRefused))?;
            let pending = held.as_mut().ok_or(OperationLogError::RoutedStream(
                RoutedStreamRefusal::BatchRefused,
            ))?;
            history.stage(pending).map_err(OperationLogError::from)?;
        }
        authority
            .retain_checkpoint_metadata(root, audit)
            .map_err(checkpoint_state_error)?;
        if due {
            let extension = authority
                .outbox
                .as_ref()
                .ok_or_else(|| StateError::invariant("checkpoint outbox is absent"))?
                .next_extension_sequence();
            let positions = authority
                .checkpoint_positions(metadata.schema(), authority.next_log_sequence, extension)
                .map_err(checkpoint_state_error)?;
            self.stage_borrowed_checkpoint(
                positions,
                metadata.history_root(),
                authority,
                metadata.prefix_digest(),
            )?;
            self.stage_prune(authority, extension)?;
        }
        Ok(metadata)
    }

    /// Decides and stages the cut behind the checkpoint just staged
    /// (DUR-018): acknowledged rows behind the checkpoint beyond the
    /// configured kept window, with a receipt in the same batch. Off, or
    /// nothing to cut, stages nothing.
    fn stage_prune(
        &self,
        authority: &ConversationAuthority,
        next_extension_sequence: u64,
    ) -> Result<(), StateError> {
        let keep = self
            .pruning
            .map(|pruning| pruning.keep_acknowledged_behind_checkpoint);
        let Some(outbox) = authority.outbox.as_ref() else {
            return Ok(());
        };
        let acknowledged_log_through = outbox.acknowledged_frontier(authority.next_log_sequence);
        let acknowledged_outbox_through = outbox.acknowledged_frontier(next_extension_sequence);
        let recipients: Vec<super::prune::RecipientAcknowledgement> = outbox
            .recipient_frontiers()
            .into_iter()
            .map(
                |(participant_id, acknowledged_through)| super::prune::RecipientAcknowledgement {
                    participant_id,
                    acknowledged_through,
                },
            )
            .collect();
        let Some(plan) = super::prune::plan(
            keep,
            self.conversation_id,
            super::prune::StreamBounds {
                checkpoint_position: authority.next_log_sequence,
                acknowledged_through: acknowledged_log_through,
            },
            super::prune::StreamBounds {
                checkpoint_position: next_extension_sequence,
                acknowledged_through: acknowledged_outbox_through,
            },
            &recipients,
            self.timestamp,
        ) else {
            return Ok(());
        };
        // A cut repeated to the same point (a crash between receipt and
        // cut) writes no second receipt.
        let Some(plan) = block_on(super::prune::unwritten(
            self.log.store().as_ref(),
            self.conversation_id,
            plan,
        ))
        .map_err(OperationLogError::from)?
        .map_err(OperationLogError::from)?
        else {
            return Ok(());
        };
        let pending = self.batch.ok_or(OperationLogError::RoutedStream(
            RoutedStreamRefusal::BatchRefused,
        ))?;
        let mut held = pending
            .try_borrow_mut()
            .map_err(|_| OperationLogError::RoutedStream(RoutedStreamRefusal::BatchRefused))?;
        let pending = held.as_mut().ok_or(OperationLogError::RoutedStream(
            RoutedStreamRefusal::BatchRefused,
        ))?;
        pending.stage_prune(plan)?;
        Ok(())
    }

    fn staged_checkpoint_metadata(
        &self,
        authority: &ConversationAuthority,
        starting_log_sequence: u64,
        config: &ParticipantConfig,
    ) -> Result<
        (
            super::restore_from_checkpoint::CheckpointMetadata,
            super::fenced_attach_terminal::ComposedTerminalValidation,
        ),
        StateError,
    > {
        use super::fenced_attach_terminal::{AuditSourceError, ComposedTerminalValidation};
        use super::log::OperationSchemaPhase;
        use super::restore_checkpoint::CheckpointSchema;
        use super::restore_checkpoint_history::CheckpointHistoryRoot;
        use super::restore_checkpoint_locator::checkpoint_prefix_seed;
        use super::restore_from_checkpoint::CheckpointMetadata;
        let (root, prefix, mut phase, mut audit) = if starting_log_sequence == 0 {
            (
                CheckpointHistoryRoot::empty(self.conversation_id).encode_canonical(),
                checkpoint_prefix_seed(),
                OperationSchemaPhase::V2Prefix,
                ComposedTerminalValidation::for_identity_slots(config.identity_slots)?,
            )
        } else {
            let metadata = authority.checkpoint_metadata()?;
            let phase = match metadata.schema() {
                CheckpointSchema::V2Prefix => OperationSchemaPhase::V2Prefix,
                CheckpointSchema::V3Suffix => OperationSchemaPhase::V3Suffix,
                CheckpointSchema::V4Suffix => OperationSchemaPhase::V4Suffix,
            };
            let bytes = authority
                .checkpoint_audit_source
                .as_deref()
                .ok_or_else(|| StateError::invariant("saved terminal audit is absent"))?;
            let audit = ComposedTerminalValidation::decode_checkpoint(
                bytes,
                config.identity_slots,
                starting_log_sequence - 1,
            )
            .map_err(checkpoint_state_error)?;
            (
                *metadata.history_root(),
                metadata.prefix_digest(),
                phase,
                audit,
            )
        };
        let pending = self.batch.ok_or(OperationLogError::RoutedStream(
            RoutedStreamRefusal::BatchRefused,
        ))?;
        let mut held = pending
            .try_borrow_mut()
            .map_err(|_| OperationLogError::RoutedStream(RoutedStreamRefusal::BatchRefused))?;
        let pending = held.as_mut().ok_or(OperationLogError::RoutedStream(
            RoutedStreamRefusal::BatchRefused,
        ))?;
        let mut sequence = starting_log_sequence;
        block_on(async {
            while sequence < authority.next_log_sequence {
                let page = self.log.read_staged_page(pending, sequence, phase).await?;
                if page.rows.is_empty() {
                    return Err(StateError::invariant("staged operation has a missing row"));
                }
                phase = page.next_phase;
                for row in page.rows {
                    if row.sequence != sequence || row.sequence >= authority.next_log_sequence {
                        return Err(StateError::invariant("staged operation sequence differs"));
                    }
                    let source = |position| self.log.read_staged_at(pending, position);
                    match audit
                        .validate_from(&source, row.sequence, &row.operation)
                        .await
                    {
                        Ok(()) => {}
                        Err(AuditSourceError::Semantic(error)) => return Err(error),
                        Err(AuditSourceError::Source(error)) => return Err(error.into()),
                    }
                    sequence = sequence.checked_add(1).ok_or_else(|| {
                        StateError::invariant("staged operation sequence overflow")
                    })?;
                }
            }
            Ok::<(), StateError>(())
        })??;
        let prefix = pending
            .checkpoint_prefix(prefix)
            .map_err(OperationLogError::from)?;
        let schema = match phase {
            OperationSchemaPhase::V2Prefix => CheckpointSchema::V2Prefix,
            OperationSchemaPhase::V3Suffix => CheckpointSchema::V3Suffix,
            OperationSchemaPhase::V4Suffix => CheckpointSchema::V4Suffix,
        };
        let extension = authority
            .outbox
            .as_ref()
            .ok_or_else(|| StateError::invariant("checkpoint outbox is absent"))?
            .next_extension_sequence();
        let positions = authority
            .checkpoint_positions(schema, authority.next_log_sequence, extension)
            .map_err(checkpoint_state_error)?;
        let metadata = CheckpointMetadata::validated(positions, root, prefix)
            .map_err(checkpoint_state_error)?;
        Ok((metadata, audit))
    }

    fn stage_borrowed_checkpoint<
        L: super::restore_checkpoint::CheckpointLiveState,
        const N: usize,
    >(
        &self,
        positions: super::restore_checkpoint::CheckpointPositions,
        history_root: &[u8; N],
        live: &L,
        prefix_digest: [u8; 32],
    ) -> Result<(), StateError> {
        let batch = self.batch.ok_or(OperationLogError::RoutedStream(
            RoutedStreamRefusal::BatchRefused,
        ))?;
        let mut pending = batch
            .try_borrow_mut()
            .map_err(|_| OperationLogError::RoutedStream(RoutedStreamRefusal::BatchRefused))?;
        let pending = pending.as_mut().ok_or(OperationLogError::RoutedStream(
            RoutedStreamRefusal::BatchRefused,
        ))?;
        let store = self.log.store();
        block_on(async {
            self.log
                .stage_bootstrap_head(pending, positions.next_log_sequence)
                .await?;
            self.outbox_log
                .stage_bootstrap_head(pending, positions.extension_sequence)
                .await?;
            pending
                .stage_borrowed_checkpoint(
                    store.as_ref(),
                    positions,
                    history_root,
                    live,
                    prefix_digest,
                )
                .await
                .map_err(OperationLogError::from)?;
            Ok::<(), StateError>(())
        })?
    }

    pub(super) fn has_staged(&self) -> Result<bool, OperationLogError> {
        // A flat appender stages nothing: its rows are durable as appended.
        let Some(batch) = self.batch else {
            return Ok(false);
        };
        let pending = batch
            .try_borrow()
            .map_err(|_| OperationLogError::RoutedStream(RoutedStreamRefusal::BatchRefused))?;
        let operation = pending.as_ref().ok_or(OperationLogError::RoutedStream(
            RoutedStreamRefusal::BatchRefused,
        ))?;
        Ok(!operation.is_empty()?)
    }

    pub(super) fn commit_staged(&self) -> Result<(), OperationLogError> {
        let Some(batch) = self.batch else {
            return Err(OperationLogError::RoutedStream(
                RoutedStreamRefusal::BatchRefused,
            ));
        };
        let pending = {
            let mut pending = batch
                .try_borrow_mut()
                .map_err(|_| OperationLogError::RoutedStream(RoutedStreamRefusal::BatchRefused))?;
            pending.take().ok_or(OperationLogError::RoutedStream(
                RoutedStreamRefusal::BatchRefused,
            ))?
        };
        let store = self.log.store();
        let mut pending = pending;
        let prune = pending.take_prune();
        block_on(pending.commit(store.as_ref(), || self.log.issuance().mark_submitted()))??;
        if let Some(plan) = prune {
            // The receipts are durable with the checkpoint; the cut follows,
            // and a store that refuses is recorded by name, never claimed.
            for outcome in block_on(super::prune::cut(store.as_ref(), &plan))? {
                match outcome.refused {
                    None => tracing::info!(
                        stream = ?outcome.stream,
                        source = ?outcome.source,
                        cut_before = outcome.cut_before,
                        removed = outcome.removed,
                        "pruned acknowledged rows behind the checkpoint"
                    ),
                    Some(refusal) => tracing::warn!(
                        stream = ?outcome.stream,
                        source = ?outcome.source,
                        cut_before = outcome.cut_before,
                        refusal = %refusal,
                        "prune receipt written, cut refused by the store"
                    ),
                }
            }
        }
        Ok(())
    }

    /// Rows this operation appended that owe an extension row and have not
    /// been given one in place.
    pub(super) fn outstanding_extension_rows(&self) -> u64 {
        self.outstanding_extension_rows.get()
    }
}

impl DurableAppend for LogAppender<'_> {
    fn append_extension(
        &self,
        row: &crate::server::participant::production::outbox_log::OutboxRow,
        expected_sequence: u64,
    ) -> Result<(), crate::server::participant::production::state::ExtensionRowCause> {
        if let Some(batch) = self.batch {
            let mut pending = batch.try_borrow_mut().map_err(|_| {
                super::state::ExtensionRowCause::Log(OutboxLogError::RoutedStream(
                    RoutedStreamRefusal::BatchRefused,
                ))
            })?;
            let pending = pending.as_mut().ok_or_else(|| {
                super::state::ExtensionRowCause::Log(OutboxLogError::RoutedStream(
                    RoutedStreamRefusal::BatchRefused,
                ))
            })?;
            return block_on(self.outbox_log.stage_into(
                pending,
                row,
                expected_sequence,
                self.timestamp,
            ))
            .map_err(super::state::ExtensionRowCause::Bridge)?
            .map_err(super::state::ExtensionRowCause::Log);
        }
        let Some(log) = self.extension_log() else {
            return Err(
                crate::server::participant::production::state::ExtensionRowCause::Log(
                    crate::server::participant::production::outbox_log::OutboxLogError::Durability(
                        liminal::durability::DurabilityError::BatchUnsupported {
                            store: std::any::type_name::<Self>(),
                        },
                    ),
                ),
            );
        };
        liminal::durability::bridge::block_on(log.append_flat(row, expected_sequence))
            .map_err(crate::server::participant::production::state::ExtensionRowCause::Bridge)?
            .map_err(crate::server::participant::production::state::ExtensionRowCause::Log)
    }
    fn append(
        &self,
        operation: &StoredOperation,
        expected_sequence: u64,
    ) -> Result<(), OperationLogError> {
        if expected_sequence == 0 && matches!(operation, StoredOperation::Genesis { .. }) {
            self.registry
                .register_with_issuance(self.conversation_id, self.log.issuance())?;
        }
        if let Some(batch) = self.batch {
            let mut pending = batch
                .try_borrow_mut()
                .map_err(|_| OperationLogError::RoutedStream(RoutedStreamRefusal::BatchRefused))?;
            let pending = pending.as_mut().ok_or(OperationLogError::RoutedStream(
                RoutedStreamRefusal::BatchRefused,
            ))?;
            block_on(
                self.log
                    .stage_into(pending, operation, expected_sequence, self.timestamp),
            )??;
        } else {
            block_on(self.log.append_flat(operation, expected_sequence))??;
        }
        if owes_extension_row(operation) {
            self.outstanding_extension_rows
                .set(self.outstanding_extension_rows.get().saturating_add(1));
        }
        Ok(())
    }

    fn extension_log(&self) -> Option<&OutboxLog> {
        Some(self.outbox_log)
    }

    fn owed_extension_rows(&self) -> u64 {
        self.outstanding_extension_rows.get()
    }

    fn discharge_owed_extension_row(&self) {
        self.outstanding_extension_rows
            .set(self.outstanding_extension_rows.get().saturating_sub(1));
    }
}

/// Milliseconds since `started`, saturating, for progress lines.
pub(super) fn elapsed_ms(started: Instant) -> u64 {
    u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX)
}

/// Every store failure on an issued operation leaves its durability unconfirmed.
const fn unsettled_write(error: &DurabilityError) -> bool {
    match error {
        DurabilityError::EmptyOperationBatch
        | DurabilityError::BatchUnsupported { .. }
        | DurabilityError::AppendUnresolved { .. }
        | DurabilityError::PendingAppendUnreadable { .. }
        | DurabilityError::OutcomeUnknown { .. }
        | DurabilityError::NotStarted { .. }
        | DurabilityError::StoreError(_)
        | DurabilityError::SequenceConflict { .. }
        | DurabilityError::CursorRegression { .. }
        | DurabilityError::DedupCollision { .. }
        | DurabilityError::ConfigError(_)
        | DurabilityError::EphemeralStoreOpen(_)
        | DurabilityError::EphemeralStoreDetached
        | DurabilityError::TruncateUnsupported { .. }
        | DurabilityError::RoutedScanUnsupported { .. }
        | DurabilityError::RoutedRemoveUnsupported { .. }
        | DurabilityError::EnvelopeError(_) => true,
    }
}

const fn unsettled_outbox_write(error: &OutboxLogError) -> bool {
    matches!(error, OutboxLogError::Durability(error) if unsettled_write(error))
}

pub(super) fn state_error(error: &StateError) -> ParticipantSemanticError {
    let issued = match error {
        StateError::Log(error) => {
            matches!(error, OperationLogError::Durability(error) if unsettled_write(error))
        }
        StateError::OutboxLog(error) => unsettled_outbox_write(error),
        StateError::Outbox(error) => {
            matches!(error, ConversationOutboxError::Log(error) if unsettled_outbox_write(error))
        }
        StateError::ExtensionRowsOwed { .. } | StateError::ExtensionRowUnconfirmed { .. } => true,
        StateError::Bridge(_)
        | StateError::Facts(_)
        | StateError::ObserverProgressConformance(_)
        | StateError::FateOccurrenceConflict(_)
        | StateError::ShellRefused { .. }
        | StateError::ShellUnavailable
        | StateError::FrontierUnavailable
        | StateError::ObligationDebtOwner { .. }
        | StateError::ReplayedEventDrift { .. }
        | StateError::ReplayedDiedTerminalHistory { .. }
        | StateError::BindingTerminalAdmissionRefused { .. }
        | StateError::ConversationSealed { .. }
        | StateError::PresentedRefusal(_)
        | StateError::Invariant { .. }
        | StateError::AllocationExhausted { .. } => false,
    };
    classify_state_error(error, issued)
}

fn state_error_with_issuance(
    error: &StateError,
    issuance: &WriteIssuance,
) -> ParticipantSemanticError {
    classify_state_error(error, issuance.was_submitted())
}

fn classify_state_error(error: &StateError, issued: bool) -> ParticipantSemanticError {
    let uncertain = match error {
        StateError::ReplayedDiedTerminalHistory { .. } => false,
        StateError::BindingTerminalAdmissionRefused { error } => {
            return ParticipantSemanticError::BindingTerminalAdmissionRefused { error: *error };
        }
        StateError::ConversationSealed { conversation_id } => {
            return ParticipantSemanticError::ConversationSealed {
                conversation_id: *conversation_id,
            };
        }
        StateError::ExtensionRowsOwed { .. }
        | StateError::ExtensionRowUnconfirmed { .. }
        | StateError::Log(_)
        | StateError::Outbox(_)
        | StateError::OutboxLog(_)
        | StateError::Bridge(_)
        | StateError::Facts(_)
        | StateError::ObserverProgressConformance(_)
        | StateError::FateOccurrenceConflict(_)
        | StateError::ShellRefused { .. }
        | StateError::ShellUnavailable
        | StateError::FrontierUnavailable
        | StateError::ObligationDebtOwner { .. }
        | StateError::ReplayedEventDrift { .. }
        | StateError::PresentedRefusal(_)
        | StateError::Invariant { .. }
        | StateError::AllocationExhausted { .. } => issued,
    };
    let message = format!("participant production operation failed: {error}");
    if uncertain {
        ParticipantSemanticError::WriteOutcomeUnknown { message }
    } else {
        ParticipantSemanticError::Internal { message }
    }
}

pub(super) fn log_error(error: &OperationLogError) -> ParticipantSemanticError {
    let message = format!("participant production log failed: {error}");
    if matches!(error, OperationLogError::Durability(error) if unsettled_write(error)) {
        return ParticipantSemanticError::WriteOutcomeUnknown { message };
    }
    ParticipantSemanticError::Internal { message }
}

#[cfg(test)]
pub(super) fn outbox_log_error(error: &OutboxLogError) -> ParticipantSemanticError {
    let message = format!("participant Unit 2 extension log failed: {error}");
    if unsettled_outbox_write(error) {
        return ParticipantSemanticError::WriteOutcomeUnknown { message };
    }
    ParticipantSemanticError::Internal { message }
}

pub(super) fn bridge_error(
    error: &liminal::durability::bridge::BridgeError,
) -> ParticipantSemanticError {
    ParticipantSemanticError::Internal {
        message: format!("participant durability bridge failed: {error}"),
    }
}

#[cfg(test)]
mod batch_tests;