liminal-server 0.23.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::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 crate::server::signals::ServerSignals;

use liminal::durability::bridge::block_on;
use liminal::durability::{DurabilityError, DurableStore};

use liminal_protocol::lifecycle::{CapacityCounter, ObserverRecoveryAggregate};
use liminal_protocol::wire::ConversationId;

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

use super::barrier::{OperationFacts, ReceiptCapacityLimits};
use super::capacity::ServerCapacity;
use super::capacity_snapshot::CapacitySeed;
use super::checkpoint_batches::{
    CheckpointPublication, CheckpointPublisher, PublicationBound, PublicationLayoutError,
    PublicationRefusal, publication_bound,
};
#[cfg(test)]
use super::dispatch_work::{ObligationDispatchWorkCounters, ObligationDispatchWorkSnapshot};
use super::facts;
use super::lazy_restore::PendingRestores;
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>,
}

/// One conversation's owner, absent until it restores or enrolls.
pub(super) type ConversationCell = Arc<Mutex<Option<ConversationAuthority>>>;

/// 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>>>>>,
    /// DUR-020: registered conversations not yet restored in this process.
    /// Construction fills it from the registry and restores nothing; each
    /// conversation's first touch runs its restore unit (`lazy_restore`).
    pub(super) pending: PendingRestores,
    /// This handler as its warm thread holds it: weakly, so a server that
    /// drops its handler ends the warm at the next conversation boundary.
    pub(super) self_ref: OnceLock<std::sync::Weak<Self>>,
    /// Set when the server stops; the warm checks it between conversations.
    pub(super) warm_stop: AtomicBool,
    pub(super) warm_worker: Arc<super::warm::WarmWorker>,
    /// 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), folded from each conversation as its restore
    /// unit runs and kept exact by commit reservations and replay folds. It
    /// is exact against durable truth once `pending` is empty, which is why
    /// enrollment, the one request a server-scope count refuses, finishes
    /// the warm first (DUR-020 R3).
    pub(super) capacity: ServerCapacity,
    /// The clean-stop capacity snapshot this process seeded from, if any,
    /// and the witness its next write names (DUR-020 R3).
    pub(super) capacity_seed: CapacitySeed,
    /// Durable registry of created conversations: one row appended before
    /// each conversation's genesis append, read at startup to record every
    /// durable conversation as pending (DUR-020).
    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,
    /// DUR-024: checkpoint publications run off the owner's write path, and
    /// the ledger of where each conversation's publication stands.
    pub(super) publisher: CheckpointPublisher,
    /// Takes of a conversation's owner cell through [`Self::cell`], for the
    /// pin that a publication takes none while it submits (DUR-024 R3).
    #[cfg(test)]
    pub(super) owner_lock_takes: 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. It reads
    /// the conversation registry and records every registered conversation
    /// as pending; it restores none of them (DUR-020 R1), so the server
    /// listens in the same time whatever the number of conversations. Each
    /// conversation restores on its first touch, and the warm restores the
    /// rest while the server serves. A restart still forgets no reserved
    /// identity slot: enrollment, the one request a server-scope count
    /// refuses, finishes the warm before it decides.
    ///
    /// # Errors
    ///
    /// Returns [`ParticipantSemanticError`] when the registry cannot be read:
    /// without it no conversation can be named, so 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()),
            pending: PendingRestores::default(),
            self_ref: OnceLock::new(),
            warm_stop: AtomicBool::new(false),
            warm_worker: Arc::new(super::warm::WarmWorker::default()),
            service_fatal: Mutex::new(None),
            observer: Mutex::new(None),
            observer_head_uncertain: AtomicBool::new(false),
            capacity: ServerCapacity::default(),
            capacity_seed: CapacitySeed::default(),
            registry,
            unloadable: UnloadableConversationRecord::default(),
            owed_receipts: Mutex::new(HashMap::new()),
            #[cfg(test)]
            obligation_dispatch_work: ObligationDispatchWorkCounters::default(),
            now_override: AtomicU64::new(0),
            publisher: CheckpointPublisher::default(),
            #[cfg(test)]
            owner_lock_takes: AtomicU64::new(0),
        };
        if handler.config.pruning.is_none() {
            // DUR-024 R4: no section, the pinned default, said once by name
            // at start. Pruning is never off: a conversation whose rows are
            // never cut grows without limit, and its next checkpoint with it.
            tracing::info!(
                keep_acknowledged_behind_checkpoint =
                    super::prune::DEFAULT_KEEP_ACKNOWLEDGED_BEHIND_CHECKPOINT,
                "pruning on by default: {} rows behind the checkpoint ([participant.pruning] is absent)",
                super::prune::DEFAULT_KEEP_ACKNOWLEDGED_BEHIND_CHECKPOINT
            );
        }
        handler.note_registered_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)
    }

    /// Builds the handler and finishes the warm before returning: the
    /// pre-DUR-020 boot, which every test that pins what a boot restores,
    /// refuses or drains still means.
    #[cfg(test)]
    pub(super) fn new_restored(
        store: Arc<dyn DurableStore>,
        config: ParticipantConfig,
    ) -> Result<Self, ParticipantSemanticError> {
        let handler = Self::new(store, config)?;
        handler.restore_remaining("a test boot restores every conversation before returning")?;
        Ok(handler)
    }

    /// Startup: reads the registry and records every registered conversation
    /// as pending. Nothing about any conversation is read here (DUR-020 R1).
    fn note_registered_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 pending = self.pending.register(conversation_ids)?;
        #[cfg(test)]
        crate::boot_history_harness::during_stage("registry", || {
            self.seed_capacity_from_snapshot();
        });
        #[cfg(not(test))]
        self.seed_capacity_from_snapshot();
        tracing::info!(
            step = "participant restore",
            conversations = pending,
            elapsed_ms = elapsed_ms(started),
            "participant restore deferred: every registered conversation restores on its first touch or in the warm"
        );
        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.
    /// Restores one conversation and answers its log rows and whether it
    /// still owes the checkpoint its rebuild could not publish (DUR-020 R6):
    /// the restore unit publishes it once the incarnation's repair has run.
    pub(super) fn restore_one_conversation(
        &self,
        conversation_id: ConversationId,
    ) -> Result<(u64, bool), ParticipantSemanticError> {
        let mut rows = 0_u64;
        let mut still_owed = false;
        {
            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, publication) = block_on(
                    super::restore_checkpoint_rebuild::restore_at_boot_reporting(
                        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, false));
                }
                // 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()?;
                // DUR-020 R6: a rebuild whose owner could not be captured
                // published nothing, and left every later load a full rebuild.
                // Once the drain has finished the fate that refused the
                // capture, this unit publishes the checkpoint it owes.
                let unpublished =
                    publication == super::restore_checkpoint_rebuild::BootCheckpoint::Unpublished;
                // Owed only when the whole checkpoint can be framed: its
                // positions as well as its lifecycle. The owed crossing is
                // forced, and a forced crossing refused at its positions
                // fails the restore unit instead of serving on (11 October
                // 2026: a drained Detached terminal moves observer progress
                // past the outbox's highest delivery).
                let owe_checkpoint = unpublished
                    && replayed.checkpoint_lifecycle_capturable().is_ok()
                    && replayed.checkpoint_positions_framable().is_ok();
                still_owed = unpublished && !owe_checkpoint;
                if drained || owe_checkpoint {
                    self.commit_boot_drain(
                        conversation_id,
                        &mut replayed,
                        &log,
                        before_drain,
                        owe_checkpoint,
                    )?;
                }
                rows = replayed.next_log_sequence;
                *owner = Some(replayed);
            }
            drop(owner);
        }
        Ok((rows, still_owed))
    }

    /// DUR-020 R6, after the restore unit's repair: a rebuild that could not
    /// be captured before the repair publishes its checkpoint now if the
    /// repair finished what refused the capture. Still uncapturable, it
    /// publishes nothing here and the next crossing publishes, said by name.
    /// Answers whether the checkpoint was published (or handed to the
    /// publisher); `false` when the owner is absent or still uncapturable.
    pub(super) fn publish_owed_checkpoint(
        &self,
        conversation_id: ConversationId,
    ) -> Result<bool, ParticipantSemanticError> {
        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"
                ),
            })?;
        let Some(authority) = owner.as_mut() else {
            return Ok(false);
        };
        if let Err(refusal) = authority
            .checkpoint_lifecycle_capturable()
            .and_then(|()| authority.checkpoint_positions_framable())
        {
            tracing::info!(
                step = "participant restore",
                conversation_id,
                refusal = %refusal,
                "rebuilt conversation still uncapturable; it publishes after its repair or at the next crossing"
            );
            return Ok(false);
        }
        let log = OperationLog::new(Arc::clone(&self.store), conversation_id);
        let before = authority.next_log_sequence;
        self.commit_boot_drain(conversation_id, authority, &log, before, true)?;
        drop(owner);
        Ok(true)
    }

    /// 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.
    pub(super) fn commit_boot_drain(
        &self,
        conversation_id: ConversationId,
        replayed: &mut ConversationAuthority,
        log: &OperationLog,
        before_drain: u64,
        owe_checkpoint: bool,
    ) -> 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,
            Some(super::prune::effective_keep(self.config.pruning)),
            (!flat_legacy).then_some(&batch),
            0,
            &self.publisher,
        );
        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 = owe_checkpoint
            || 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, owe_checkpoint)
                .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.
    pub(super) 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()
    }

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

    /// Every cell this process holds, with its conversation id, taken under
    /// one short registry lock and locked by the caller one at a time.
    pub(super) fn conversation_cells(
        &self,
    ) -> Result<Vec<(ConversationId, ConversationCell)>, ParticipantSemanticError> {
        let conversations =
            self.conversations
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: "participant conversation registry lock is poisoned".to_owned(),
                })?;
        Ok(conversations
            .iter()
            .map(|(conversation_id, cell)| (*conversation_id, Arc::clone(cell)))
            .collect())
    }

    pub(super) fn existing_cell(
        &self,
        conversation_id: ConversationId,
    ) -> Result<Option<Arc<Mutex<Option<ConversationAuthority>>>>, ParticipantSemanticError> {
        // DUR-020 R2: a pending conversation is restored before any reader
        // sees its cell, so "no cell" keeps meaning "no such conversation".
        self.ensure_restored(conversation_id)?;
        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> {
        // DUR-020 R2: every path that reaches a conversation's cell reaches
        // it through here or `existing_cell`, so the first touch, whatever
        // touches, runs the conversation's restore unit first.
        self.ensure_restored(conversation_id)?;
        #[cfg(test)]
        self.owner_lock_takes.fetch_add(1, Ordering::SeqCst);
        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.
            // The load reads and writes through its own log handle: a
            // rebuild's checkpoint publication is the load's settled write,
            // not this request's, and must not mark the request's write as
            // issued. Sharing the handle turned a request refused before it
            // wrote anything into "write outcome unknown" whenever its load
            // had published (11 October 2026).
            let load_log = OperationLog::new(Arc::clone(&self.store), conversation_id);
            let replayed = self
                .replay_and_repair(conversation_id, &load_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,
            Some(super::prune::effective_keep(self.config.pruning)),
            (!flat_legacy).then_some(&batch),
            timestamp,
            &self.publisher,
        );
        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,
            Some(super::prune::effective_keep(self.config.pruning)),
            (!flat_legacy).then_some(&batch),
            timestamp,
            &self.publisher,
        );
        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,
    /// Acknowledged rows kept behind a checkpoint this appender stages
    /// (DUR-018, DUR-024 R4: [`super::prune::effective_keep`]'s answer, the
    /// configured window or the pinned default); `None` for a drain that
    /// stages no cut of its own.
    pub(super) prune_keep: Option<u64>,
    /// 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,
    /// Where a checkpoint too large for one batch is handed off (DUR-024).
    pub(super) publisher: &'a CheckpointPublisher,
}

/// 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,
        prune_keep: Option<u64>,
        batch: Option<&'a RefCell<Option<OperationBatch>>>,
        timestamp: u64,
        publisher: &'a CheckpointPublisher,
    ) -> Self {
        Self {
            log,
            registry,
            conversation_id: log.conversation_id(),
            outbox_log,
            prune_keep,
            outstanding_extension_rows: Cell::new(0),
            batch,
            timestamp,
            publisher,
        }
    }

    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, false)?;
        }
        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.
    /// `force` stages the checkpoint with no crossing: the boot drain's owed
    /// publication of a rebuild that could not be captured (DUR-020 R6).
    fn commit_with_checkpoint(
        &self,
        authority: &mut ConversationAuthority,
        starting_log_sequence: u64,
        config: &ParticipantConfig,
        force: bool,
    ) -> 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,
            force,
        )?;
        // A checkpoint handed to the publisher stages nothing here; an
        // operation with no rows of its own (the boot drain's owed
        // publication) then has nothing to commit, and an empty commit is
        // an unsettled write that contains the room (11 October, 20:48Z).
        if self.has_staged()? {
            self.commit_staged()?;
        }
        let crossed = force
            || OperationBatch::checkpoint_due(
                starting_log_sequence,
                authority.next_log_sequence,
                authority.closed,
            )
            .map_err(OperationLogError::from)?;
        authority
            .install_checkpoint_metadata(metadata)
            .map_err(checkpoint_state_error)?;
        if crossed {
            self.publisher
                .crossing_committed(&crate::server::connection::CheckpointCrossing {
                    conversation_id: self.conversation_id,
                    covered_through: authority.next_log_sequence,
                    delivery_head: authority
                        .outbox
                        .as_ref()
                        .map_or(0, |outbox| outbox.checkpoint_boundaries().1),
                    recipients: authority.crossing_census(),
                });
        }
        Ok(())
    }

    fn prepare_crossing_checkpoint(
        &self,
        authority: &mut ConversationAuthority,
        before: u64,
        mut metadata: super::restore_from_checkpoint::CheckpointMetadata,
        audit: &super::fenced_attach_terminal::ComposedTerminalValidation,
        force: bool,
    ) -> 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 = force
            || OperationBatch::checkpoint_due(
                before,
                authority.next_log_sequence,
                authority.closed,
            )
            .map_err(OperationLogError::from)?;
        // DUR-024 R3: a publication still in flight for this conversation
        // is left to finish; this crossing stages no checkpoint and the next
        // one retries over whatever the publisher committed.
        let due = due && !self.publication_in_flight();
        let mut staged = false;
        // A checkpoint the owner cannot frame is refused here, before
        // anything is staged, and the operation commits without it: the
        // owner serves on and the next crossing retries. A framing failure
        // never fails the operation (11 October, 21:16Z: the door room was
        // contained by its first post's crossing).
        let body_bytes = if due {
            self.capturable_body_bytes(authority)
        } else {
            None
        };
        if let Some(body_bytes) = body_bytes {
            let store = self.log.store();
            let bound = publication_bound(store.as_ref());
            let mut history = block_on(
                super::restore_checkpoint_history_capture::capture_identity_history(
                    store.as_ref(),
                    authority,
                    root,
                ),
            )?
            .map_err(|error| checkpoint_state_error(error.checkpoint()))?;
            // The operation's own commit carries the checkpoint only when
            // the history, the body and the operation's rows together fit
            // the bound. Measured on the history alone, the door room's
            // held Open 44 (11 October) put a 30 MB history and a 101 MB
            // body into one 131 MB write; the store refused it, the
            // outcome was unknown, and the room was contained.
            let inline_bytes = history
                .byte_total()
                .saturating_add(body_bytes)
                .saturating_add(self.staged_value_bytes()?);
            if inline_bytes <= bound.bound {
                // Under the bound: one commit with the operation's own rows,
                // byte for byte what every crossing committed before DUR-024.
                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,
                ))?;
                // The owner adopts the new root below, and with it the
                // digests of what that root holds, so the next crossing puts
                // only what changes before it.
                authority.history_digests = history.take_digests();
                history.stage(pending).map_err(OperationLogError::from)?;
                staged = true;
            } else {
                // Over the bound: the history, the body and the locator go
                // to the publisher in batches, off this write's path. The
                // owner keeps the root whose nodes are all present, so the
                // next crossing's capture stages only what is still absent.
                self.publisher.ledger().log_bound_once(bound);
                self.hand_off_publication(authority, &metadata, history, bound, audit)?;
            }
        }
        authority
            .retain_checkpoint_metadata(root, audit)
            .map_err(checkpoint_state_error)?;
        if staged {
            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)
    }

    /// The bytes of the owner's checkpoint body, or `None`, said by name,
    /// when the owner cannot be framed now.
    ///
    /// The positions the staged checkpoint will carry are validated here
    /// too, before anything is staged: a position refused after staging
    /// failed the operation and contained the room, where every other
    /// framing refusal lets the owner serve on (11 October 2026: a drained
    /// `Detached` terminal advances observer progress past the outbox's
    /// highest delivery, and once the drained suffix could be captured the
    /// position check was the first refusal it reached).
    fn capturable_body_bytes(&self, authority: &ConversationAuthority) -> Option<usize> {
        let framed = authority
            .checkpoint_lifecycle_capturable()
            .and_then(|()| authority.checkpoint_positions_framable())
            .and_then(|()| super::restore_checkpoint::CheckpointLiveState::encode_live(authority));
        match framed {
            Ok(body) => Some(body.len()),
            Err(refusal) => {
                tracing::warn!(
                    conversation_id = self.conversation_id,
                    refusal = %refusal,
                    "checkpoint not captured at this crossing; the owner serves on and the next crossing retries"
                );
                None
            }
        }
    }

    /// Whether this conversation's publication is still in flight, said
    /// once per crossing that finds it so.
    fn publication_in_flight(&self) -> bool {
        let in_flight = self.publisher.ledger().in_flight(self.conversation_id);
        if in_flight {
            tracing::info!(
                conversation_id = self.conversation_id,
                "checkpoint publication still in flight; this crossing stages no checkpoint and the next one retries"
            );
        }
        in_flight
    }

    /// DUR-024 R1 and R3: a history larger than the batch bound is laid out
    /// as batches and handed to the publisher, with the body and the
    /// locator (captured here, under the owner's lock, at this crossing's
    /// positions) and the prune receipts in the final batch. The owner's
    /// lock is released before anything is submitted, and the publisher
    /// never takes it. A publication refused before submission (one item
    /// over the store's bound, R2) is recorded owed and nothing is sent.
    fn hand_off_publication(
        &self,
        authority: &mut ConversationAuthority,
        metadata: &super::restore_from_checkpoint::CheckpointMetadata,
        history: super::restore_checkpoint_history_capture::PreparedCheckpointHistory,
        bound: PublicationBound,
        audit: &super::fenced_attach_terminal::ComposedTerminalValidation,
    ) -> Result<(), StateError> {
        let store = self.log.store();
        let route = checkpoint_conversation_route(self.conversation_id);
        let (new_root, nodes) = history.into_nodes();
        // The body names the root its publication installs, and the owner's
        // live validation frames a body only against the root it retains;
        // so the new root is retained for the framing, and the caller
        // retains the old root again afterwards, the one whose nodes are
        // all present until this publication lands.
        authority
            .retain_checkpoint_metadata(new_root, audit)
            .map_err(checkpoint_state_error)?;
        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)?;
        let mut final_batch = OperationBatch::new(&route);
        block_on(final_batch.stage_borrowed_checkpoint(
            store.as_ref(),
            positions,
            &new_root.encode_canonical(),
            authority,
            metadata.prefix_digest(),
        ))?
        .map_err(OperationLogError::from)?;
        if let Some(plan) = self.plan_prune(authority, extension)? {
            final_batch.stage_prune(plan)?;
        }
        if !self.publisher.ledger().begin(self.conversation_id) {
            return Ok(());
        }
        let publication = match CheckpointPublication::new(
            self.conversation_id,
            &route,
            nodes,
            final_batch,
            bound,
        ) {
            Ok(publication) => publication,
            Err(PublicationLayoutError::Refused(refusal)) => {
                let super::checkpoint_batches::CheckpointPublishError::ItemExceedsStoreBound {
                    bytes,
                    ..
                } = &refusal;
                self.publisher.refuse_before_submission(
                    self.conversation_id,
                    PublicationRefusal {
                        batch_index: 0,
                        bytes: *bytes,
                        bound: bound.bound,
                        refusal: refusal.to_string(),
                    },
                );
                return Ok(());
            }
            Err(PublicationLayoutError::Store(error)) => {
                self.publisher.refuse_before_submission(
                    self.conversation_id,
                    PublicationRefusal {
                        batch_index: 0,
                        bytes: 0,
                        bound: bound.bound,
                        refusal: error.to_string(),
                    },
                );
                return Err(OperationLogError::from(error).into());
            }
        };
        // A thread that could not start is recorded owed by `spawn` itself.
        drop(self.publisher.spawn(store, publication));
        Ok(())
    }

    /// Decides and stages the cut behind the checkpoint just staged
    /// (DUR-018): acknowledged rows behind the checkpoint beyond the kept
    /// window (configured, or DUR-024 R4's pinned default), with a receipt
    /// in the same batch. An appender with no window of its own, or nothing
    /// to cut, stages nothing.
    fn stage_prune(
        &self,
        authority: &ConversationAuthority,
        next_extension_sequence: u64,
    ) -> Result<(), StateError> {
        let Some(plan) = self.plan_prune(authority, next_extension_sequence)? 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(())
    }

    /// The cut this checkpoint decides, less any receipt already written:
    /// `None` when nothing is to be cut.
    fn plan_prune(
        &self,
        authority: &ConversationAuthority,
        next_extension_sequence: u64,
    ) -> Result<Option<super::prune::PrunePlan>, StateError> {
        let keep = self.prune_keep;
        let Some(outbox) = authority.outbox.as_ref() else {
            return Ok(None);
        };
        // Both streams are covered through the checkpoint this cut sits
        // behind: it carries every record a recipient has not acknowledged,
        // so a recipient that never acknowledges holds its own records,
        // never the whole conversation's cut (the 11 October finding).
        let held = outbox.held_records();
        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::behind_checkpoint(authority.next_log_sequence),
            super::prune::StreamBounds::behind_checkpoint(next_extension_sequence),
            &recipients,
            held,
            self.timestamp,
        ) else {
            return Ok(None);
        };
        // A cut repeated to the same point (a crash between receipt and
        // cut) writes no second receipt.
        Ok(block_on(super::prune::unwritten(
            self.log.store().as_ref(),
            self.conversation_id,
            plan,
        ))
        .map_err(OperationLogError::from)?
        .map_err(OperationLogError::from)?)
    }

    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>(())
        })?
    }

    /// The value bytes this operation has staged so far; a flat appender
    /// stages nothing.
    fn staged_value_bytes(&self) -> Result<usize, OperationLogError> {
        let Some(batch) = self.batch else {
            return Ok(0);
        };
        let pending = batch
            .try_borrow()
            .map_err(|_| OperationLogError::RoutedStream(RoutedStreamRefusal::BatchRefused))?;
        Ok(pending.as_ref().map_or(0, OperationBatch::value_bytes))
    }

    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();
        let checkpointed = pending.take_checkpointed();
        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.
            block_on(super::prune::report_cut(
                store.as_ref(),
                self.conversation_id,
                &plan,
            ))?;
        }
        if checkpointed {
            block_on(super::restore_checkpoint_history::retire_v1_history(
                store.as_ref(),
                self.conversation_id,
            ))?;
        }
        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;