cookie_agent_engine/runtime/

mailbox.rs

use std::{
    collections::{HashMap, HashSet, VecDeque},
    sync::{Arc, Mutex, atomic::Ordering},
};

use base64::{Engine as _, engine::general_purpose::STANDARD};
use cookie_agent_protocol::{
    ApprovalStatus, EventOrigin, EventSubscriptionMessage, EventsSubscribeResult,
    ExtensionBusEventParams, ExtensionEmitStatus, PersistedToolResult as ToolResult,
    PluginDiagnosticKind, RunCancelResult, RunId, RunRecallSteerResult, RunSteerResult,
    RunToolStdinResult, SafeToolError, SessionId, SessionRenameChange, SessionRenameResult,
    SessionRevertResult, SessionStatus, SessionTitleChange, StoredEvent, ToolCallId,
    ToolCallTermination, ToolTerminationOutcome,
};
use tokio::sync::{mpsc, oneshot};

#[cfg(test)]
use super::PagingRaceHook;
use super::ToolCallFailureCode;
use super::{
    ApprovalTerminal, Engine, EngineError, Event, MAX_COMPACTION_DEFERRED_COMMANDS, PendingInput,
    PendingPromotionState, PredictiveCompactionInput, SESSION_MAILBOX_CAPACITY, SessionCommand,
    ToolFailure,
    approval_projection::{approval_records, approval_run_id},
    helpers::safe_error,
    model_loop,
};
use crate::{actor::SessionActor, events, session::SessionError, tool_api::StdinWrite};

use super::{ActiveRun, producers};

/// Per-session mailbox and run state owned by [`super::Inner`].
#[derive(Default)]
pub(crate) struct SessionRuntimeState {
    pub(super) actors: Mutex<HashMap<SessionId, SessionActor<SessionCommand>>>,
    pub(super) active: Mutex<HashMap<RunId, Arc<ActiveRun>>>,
    pub(super) producers: Mutex<HashMap<SessionId, producers::SessionProducers>>,
    /// The task driving each in-flight run loop.
    ///
    /// Shutdown joins these after cancelling their runs, so a run task gets to
    /// write its `RunCancelled` while the session actors and the store are
    /// still up. A task drops its own entry when it finishes, and a new
    /// registration sweeps whatever finished without being noticed.
    pub(super) run_tasks: Mutex<HashMap<RunId, tokio::task::JoinHandle<()>>>,
    pub(crate) residency_mutation: tokio::sync::Mutex<()>,
}

/// Removes a run task's join handle when its task ends, however it ends.
pub(super) struct RunTaskSlot {
    pub(super) engine: Engine,
    pub(super) run_id: RunId,
}

impl Drop for RunTaskSlot {
    fn drop(&mut self) {
        if let Ok(mut tasks) = self.engine.inner.sessions.run_tasks.lock() {
            tasks.remove(&self.run_id);
        }
    }
}

impl Engine {
    /// Records the task driving `run_id` so shutdown can wait for it.
    ///
    /// A task that finished before its handle arrived here has already dropped
    /// its slot, so the sweep is what retires the handle it left behind.
    pub(super) fn register_run_task(&self, run_id: RunId, task: tokio::task::JoinHandle<()>) {
        let mut tasks = self
            .inner
            .sessions
            .run_tasks
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        tasks.retain(|_, task| !task.is_finished());
        tasks.insert(run_id, task);
    }

    /// Waits for the in-flight run tasks, then aborts whatever is left.
    ///
    /// Called from shutdown while the session actors and the store are still
    /// up, so a cancelled run's terminal event lands in its log instead of
    /// being repaired as a restart interruption on the next startup.
    pub(super) async fn join_run_tasks(&self, bound: std::time::Duration) {
        let tasks = self
            .inner
            .sessions
            .run_tasks
            .lock()
            .map(|mut tasks| tasks.drain().map(|(_, task)| task).collect::<Vec<_>>())
            .unwrap_or_default();
        if tasks.is_empty() {
            return;
        }
        let deadline = tokio::time::Instant::now() + bound;
        for mut task in tasks {
            if tokio::time::timeout_at(deadline, &mut task).await.is_err() {
                task.abort();
                let _ = task.await;
            }
        }
    }

    pub async fn subscribe(
        &self,
        session: SessionId,
        cursor: Option<u64>,
    ) -> Result<
        (
            EventsSubscribeResult,
            mpsc::Receiver<EventSubscriptionMessage>,
        ),
        EngineError,
    > {
        self.inner.store.get(session)?;
        if !self.inner.store.is_owned(session) {
            return Err(EngineError::SessionOwnedByAnotherProcess(session));
        }
        self.request(session, |reply| SessionCommand::Subscribe { cursor, reply })
            .await
    }

    pub fn snapshot_events(
        &self,
        session: SessionId,
        cursor: Option<u64>,
    ) -> Result<EventsSubscribeResult, EngineError> {
        let events = self
            .inner
            .store
            .get(session)?
            .log
            .all_events()
            .into_iter()
            .filter(|event| cursor.is_none_or(|cursor| event.seq > cursor))
            .collect();
        Ok(EventsSubscribeResult { events })
    }

    /// Subscribes to a currently running call's retained output and live tail.
    /// Output is ephemeral and intentionally separate from event cursors.
    pub fn subscribe_tool_output(
        &self,
        call: ToolCallId,
        stream: cookie_agent_protocol::OutputStream,
    ) -> Option<(
        cookie_agent_protocol::OutputSnapshot,
        mpsc::Receiver<events::OutputMessage>,
    )> {
        self.inner
            .output
            .hubs
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .get(&call)
            .cloned()
            .map(|hub| hub.subscribe(stream, 256))
    }

    pub fn tool_output_streams(
        &self,
        call_id: ToolCallId,
    ) -> Option<Vec<cookie_agent_protocol::OutputStream>> {
        self.inner
            .output
            .hubs
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .get(&call_id)
            .map(|hub| hub.streams())
    }

    pub(super) fn retain_finalized_output_hub(&self, call: ToolCallId) {
        const FINALIZED_HUB_RETENTION: usize = 128;
        let mut finalized = self
            .inner
            .output
            .finalized_hubs
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        finalized.push_back(call);
        if finalized.len() > FINALIZED_HUB_RETENTION
            && let Some(expired) = finalized.pop_front()
        {
            self.inner
                .output
                .hubs
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .remove(&expired);
        }
    }

    pub async fn append(
        &self,
        session: SessionId,
        run: Option<RunId>,
        origin: EventOrigin,
        event: Event,
    ) -> Result<(), EngineError> {
        self.request(session, |reply| SessionCommand::Append {
            run,
            origin,
            event: Box::new(event),
            reply,
        })
        .await
    }

    pub(super) async fn enqueue_append(
        &self,
        session: SessionId,
        run: Option<RunId>,
        origin: EventOrigin,
        event: Event,
    ) -> Result<oneshot::Receiver<Result<(), EngineError>>, EngineError> {
        let (reply, receiver) = oneshot::channel();
        let _residency = self.inner.sessions.residency_mutation.lock().await;
        self.ensure_session_owned(session)?;
        self.spawn_actor(session)?;
        let actor = self
            .inner
            .sessions
            .actors
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .get(&session)
            .cloned()
            .ok_or(EngineError::MissingActor(session))?;
        actor
            .send(SessionCommand::Append {
                run,
                origin,
                event: Box::new(event),
                reply,
            })
            .await
            .map_err(|_| EngineError::ActorStopped)?;
        Ok(receiver)
    }

    /// Synchronous setup/CLI wrapper. Do not call from a Tokio runtime.
    pub fn append_blocking(
        &self,
        session: SessionId,
        run: Option<RunId>,
        origin: EventOrigin,
        event: Event,
    ) -> Result<(), EngineError> {
        self.request_blocking(session, |reply| SessionCommand::Append {
            run,
            origin,
            event: Box::new(event),
            reply,
        })
    }

    /// Commits a completed tool invocation through its session actor.
    pub async fn submit_tool_result(
        &self,
        session: SessionId,
        run: RunId,
        tool_call_id: ToolCallId,
        result: Result<ToolResult, String>,
    ) -> Result<(), EngineError> {
        let result = result.map_err(|message| ToolFailure {
            partial_output: None,
            code: ToolCallFailureCode::ExecutionFailed,
            message,
        });
        self.submit_tool_result_status(session, run, tool_call_id, result, false)
            .await
            .and_then(|committed| {
                committed
                    .then_some(())
                    .ok_or_else(|| EngineError::ToolFailed("tool call is no longer pending".into()))
            })
    }

    pub(super) async fn submit_tool_result_status(
        &self,
        session: SessionId,
        run: RunId,
        tool_call_id: ToolCallId,
        result: Result<ToolResult, ToolFailure>,
        cancelled: bool,
    ) -> Result<bool, EngineError> {
        self.request(session, |reply| SessionCommand::ToolResult {
            run,
            tool_call_id,
            result,
            cancelled,
            reply,
        })
        .await
    }

    pub(crate) fn append_direct(
        &self,
        session: SessionId,
        run: Option<RunId>,
        origin: EventOrigin,
        event: Event,
    ) -> Result<(), EngineError> {
        self.append_direct_with_mode(session, run, origin, event, false)
            .map(|_| ())
    }

    pub(super) fn append_direct_record(
        &self,
        session: SessionId,
        run: Option<RunId>,
        origin: EventOrigin,
        event: Event,
    ) -> Result<StoredEvent, EngineError> {
        self.append_direct_with_mode(session, run, origin, event, false)
    }

    pub(crate) fn append_recovery_direct(
        &self,
        session: SessionId,
        run: Option<RunId>,
        origin: EventOrigin,
        event: Event,
    ) -> Result<(), EngineError> {
        self.append_direct_with_mode(session, run, origin, event, true)
            .map(|_| ())
    }

    fn append_direct_with_mode(
        &self,
        session: SessionId,
        run: Option<RunId>,
        origin: EventOrigin,
        mut event: Event,
        recovery: bool,
    ) -> Result<StoredEvent, EngineError> {
        if let Event::ToolCallTerminated { termination } = &mut event {
            // Finalization and after-result hooks can await after the executor succeeds.
            // Freeze status and retained-output state together at this append boundary.
            let cancelled = run.is_some_and(|run| {
                self.inner
                    .sessions
                    .active
                    .lock()
                    .unwrap_or_else(|p| p.into_inner())
                    .get(&run)
                    .is_some_and(|active| {
                        active.session == session && active.cancellation.is_cancelled()
                    })
            });
            if cancelled
                && matches!(
                    termination.outcome,
                    ToolTerminationOutcome::Completed | ToolTerminationOutcome::Failed
                )
            {
                let finalized = termination.outcome == ToolTerminationOutcome::Completed;
                termination.outcome = ToolTerminationOutcome::Cancelled;
                termination.error.get_or_insert_with(|| SafeToolError {
                    code: if finalized {
                        super::safe_code(super::CANCELLED_AFTER_COMPLETION)
                    } else {
                        ToolCallFailureCode::ExecutionFailed.safe_code()
                    },
                    message: safe_error("tool call cancelled after it started"),
                });
            }
            if termination.outcome != ToolTerminationOutcome::Completed
                && let Some(result) = &mut termination.result
            {
                let was_incomplete = result.retained_output.as_mut().map(|retained| {
                    let was_incomplete = retained.incomplete;
                    retained.incomplete = true;
                    was_incomplete
                });
                if termination.outcome == ToolTerminationOutcome::Cancelled
                    || was_incomplete.is_some_and(|incomplete| {
                        !incomplete
                            || result
                                .display
                                .as_ref()
                                .is_none_or(|display| display.trim().is_empty())
                    })
                {
                    result.display = Some(
                        match termination.outcome {
                            ToolTerminationOutcome::Cancelled => {
                                if result.retained_output.is_some() {
                                    "Tool cancelled; retained output is incomplete."
                                } else {
                                    "Tool cancelled; output is incomplete."
                                }
                            }
                            ToolTerminationOutcome::Interrupted => {
                                "Tool interrupted; retained output is incomplete."
                            }
                            _ => "Tool failed; retained output is incomplete.",
                        }
                        .into(),
                    );
                }
            }
        }
        let was_persisted = self.inner.store.is_persisted(session)?;
        let envelope = if recovery {
            self.inner
                .store
                .append_recovery(session, run, origin, event)?
        } else {
            self.inner.store.append(session, run, origin, event)?
        };
        if !was_persisted && self.inner.store.is_persisted(session)? {
            for durable in self.inner.store.get(session)?.log.all_events() {
                let drops = self
                    .inner
                    .plugins
                    .stream_session_event(&durable, durable.origin.as_ref());
                self.record_plugin_drops(session, drops);
            }
        } else {
            let drops = self
                .inner
                .plugins
                .stream_session_event(&envelope, envelope.origin.as_ref());
            self.record_plugin_drops(session, drops);
        }
        Ok(envelope)
    }

    fn record_plugin_drops(
        &self,
        session: SessionId,
        drops: Vec<crate::plugin::PluginDeliveryDrop>,
    ) {
        for drop in drops {
            let class = match drop.class {
                crate::plugin::PluginDeliveryClass::Chunk => "chunk",
                crate::plugin::PluginDeliveryClass::Ordinary => "ordinary",
                crate::plugin::PluginDeliveryClass::Terminal => "terminal",
            };
            self.queue_plugin_diagnostic(
                session,
                drop.plugin,
                PluginDiagnosticKind::EventDrop,
                format!("plugin {class} event buffer overflow; event delivery was dropped"),
                1,
            );
        }
    }

    pub(crate) fn record_plugin_diagnostic(
        &self,
        session: SessionId,
        plugin: String,
        kind: PluginDiagnosticKind,
        message: String,
    ) {
        self.queue_plugin_diagnostic(session, plugin, kind, message, 1);
    }

    fn queue_plugin_diagnostic(
        &self,
        session_id: SessionId,
        plugin: String,
        kind: PluginDiagnosticKind,
        message: String,
        count: u64,
    ) {
        let message = cookie_agent_protocol::diagnostics::detail(&message).to_string();
        self.inner
            .plugin_diagnostics
            .accumulator
            .record((session_id, plugin, kind, message), count);
    }

    pub(crate) async fn publish_plugin_emit(
        &self,
        request: crate::plugin::PluginEmitRequest,
    ) -> crate::plugin::PluginEmitOutcome {
        let session_id = request.session_id;
        if let crate::plugin::PluginEmitContext::Rejected {
            diagnostic_session_id,
            reason,
        } = &request.context
        {
            let message = format!("plugin emit context rejected: {reason}");
            if let Some(diagnostic_session_id) = diagnostic_session_id {
                self.queue_plugin_diagnostic(
                    *diagnostic_session_id,
                    request.plugin.clone(),
                    PluginDiagnosticKind::ContextMismatch,
                    message,
                    1,
                );
            } else {
                self.inner
                    .plugins
                    .note_offender_diagnostic(&request.plugin, message);
            }
            return crate::plugin::PluginEmitOutcome {
                bus: ExtensionEmitStatus::Dropped,
                durable: ExtensionEmitStatus::Rejected,
                reason: Some(reason.clone()),
            };
        }
        let payload_bytes =
            serde_json::to_vec(&request.payload).map_or(usize::MAX, |bytes| bytes.len());
        let event_bytes = serde_json::to_vec(&Event::PluginEventAdded {
            plugin: request.plugin.clone(),
            name: request.name.clone(),
            payload: request.payload.clone(),
        })
        .map_or(usize::MAX, |bytes| bytes.len());
        let invalid_name = request.name.is_empty()
            || request.name.chars().count() > crate::plugin::MAX_PLUGIN_EVENT_NAME_CHARS
            || request.name.chars().any(char::is_control);
        if invalid_name
            || payload_bytes > crate::plugin::MAX_PLUGIN_EVENT_PAYLOAD_BYTES
            || event_bytes > crate::plugin::MAX_PLUGIN_EVENT_TOTAL_BYTES
        {
            self.queue_plugin_diagnostic(
                session_id,
                request.plugin.clone(),
                PluginDiagnosticKind::OversizedEvent,
                "plugin event name or serialized event exceeds its protocol limit".into(),
                1,
            );
            return crate::plugin::PluginEmitOutcome {
                bus: ExtensionEmitStatus::Dropped,
                durable: ExtensionEmitStatus::Rejected,
                reason: Some(format!(
                    "event name must be 1-{} control-free characters, payload at most {} bytes, and total event at most {} bytes",
                    crate::plugin::MAX_PLUGIN_EVENT_NAME_CHARS,
                    crate::plugin::MAX_PLUGIN_EVENT_PAYLOAD_BYTES,
                    crate::plugin::MAX_PLUGIN_EVENT_TOTAL_BYTES,
                )),
            };
        }
        if let Err(reason) =
            self.inner
                .plugins
                .check_publish_quota(&request.plugin, session_id, event_bytes)
        {
            self.queue_plugin_diagnostic(
                session_id,
                request.plugin.clone(),
                PluginDiagnosticKind::RateLimited,
                reason.into(),
                1,
            );
            return crate::plugin::PluginEmitOutcome {
                bus: ExtensionEmitStatus::Dropped,
                durable: ExtensionEmitStatus::Rejected,
                reason: Some(reason.into()),
            };
        }

        let mut bus = ExtensionEmitStatus::Rejected;
        let durable = ExtensionEmitStatus::Rejected;
        let mut reason = None;
        if request.publish_bus {
            let event = crate::EngineEvent::PluginEvent {
                session_id,
                plugin: request.plugin.clone(),
                name: request.name.clone(),
                payload: request.payload.clone(),
            };
            let _ = self.inner.engine_events.send(event);
            let drops = self.inner.plugins.stream_bus_event(
                &ExtensionBusEventParams {
                    session_id,
                    context_id: None,
                    plugin: request.plugin.clone(),
                    name: request.name.clone(),
                    payload: request.payload.clone(),
                },
                Some(&request.plugin),
            );
            self.record_plugin_drops(session_id, drops);
            bus = ExtensionEmitStatus::Published;
        }
        if request.publish_session_events {
            reason = Some("model-bound session emission requires explicit producer.register and producer.send".into());
        }
        if !request.publish_bus && !request.publish_session_events {
            reason = Some("plugin did not declare an event publishing capability".into());
        }
        crate::plugin::PluginEmitOutcome {
            bus,
            durable,
            reason,
        }
    }

    pub(super) async fn request<T>(
        &self,
        session: SessionId,
        command: impl FnOnce(oneshot::Sender<Result<T, EngineError>>) -> SessionCommand,
    ) -> Result<T, EngineError> {
        let (reply, receiver) = oneshot::channel();
        {
            let _residency = self.inner.sessions.residency_mutation.lock().await;
            self.ensure_session_owned(session)?;
            self.spawn_actor(session)?;
            let actor = self
                .inner
                .sessions
                .actors
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .get(&session)
                .cloned()
                .ok_or(EngineError::MissingActor(session))?;
            actor
                .send(command(reply))
                .await
                .map_err(|_| EngineError::ActorStopped)?;
        }
        receiver.await.map_err(|_| EngineError::ActorStopped)?
    }

    pub(super) async fn prompt_events(
        &self,
        session: SessionId,
        run: RunId,
    ) -> Result<super::producer_claims::ClaimedPrompt, EngineError> {
        #[cfg(test)]
        if let Some(hook) = {
            self.inner
                .test_hooks
                .prompt_before_claim_hook
                .lock()
                .expect("prompt before-claim hook lock poisoned")
                .take()
        } {
            if let Some(reached) = hook
                .reached
                .lock()
                .expect("prompt before-claim reached lock poisoned")
                .take()
            {
                let _ = reached.send(());
            }
            hook.release.notified().await;
        }
        let events = self
            .request(session, |reply| SessionCommand::PromotePendingInputs {
                run,
                reply,
            })
            .await?;
        #[cfg(test)]
        if let Some(hook) = {
            self.inner
                .test_hooks
                .prompt_snapshot_hook
                .lock()
                .expect("prompt snapshot hook lock poisoned")
                .take()
        } {
            if let Some(reached) = hook
                .reached
                .lock()
                .expect("prompt snapshot reached lock poisoned")
                .take()
            {
                let _ = reached.send(());
            }
            hook.release.notified().await;
        }
        Ok(events)
    }

    #[cfg(test)]
    pub(crate) fn install_prompt_before_claim_hook_for_test(
        &self,
    ) -> (oneshot::Receiver<()>, std::sync::Arc<tokio::sync::Notify>) {
        let (reached, receiver) = oneshot::channel();
        let release = std::sync::Arc::new(tokio::sync::Notify::new());
        *self
            .inner
            .test_hooks
            .prompt_before_claim_hook
            .lock()
            .expect("before-claim hook lock") =
            Some(std::sync::Arc::new(super::PromptSnapshotHook {
                reached: std::sync::Mutex::new(Some(reached)),
                release: release.clone(),
            }));
        (receiver, release)
    }

    #[cfg(test)]
    pub(crate) fn install_prompt_snapshot_hook_for_test(
        &self,
    ) -> (oneshot::Receiver<()>, std::sync::Arc<tokio::sync::Notify>) {
        let (reached, receiver) = oneshot::channel();
        let release = std::sync::Arc::new(tokio::sync::Notify::new());
        *self
            .inner
            .test_hooks
            .prompt_snapshot_hook
            .lock()
            .expect("snapshot hook lock") = Some(std::sync::Arc::new(super::PromptSnapshotHook {
            reached: std::sync::Mutex::new(Some(reached)),
            release: release.clone(),
        }));
        (receiver, release)
    }

    #[cfg(test)]
    pub(crate) async fn reconcile_producers_for_test(
        &self,
        session: SessionId,
    ) -> Result<(), EngineError> {
        self.reconcile_producers(session).await
    }

    #[cfg(test)]
    pub(crate) fn compaction_reserved_for_test(&self, session: SessionId) -> bool {
        self.inner
            .compaction
            .in_progress
            .lock()
            .expect("compaction reservation lock poisoned")
            .contains(&session)
    }

    #[cfg(test)]
    pub(crate) fn run_active_for_test(&self, run: RunId) -> bool {
        self.inner
            .sessions
            .active
            .lock()
            .expect("active runs lock poisoned")
            .contains_key(&run)
    }

    #[cfg(test)]
    pub(crate) fn actor_resident_for_test(&self, session: SessionId) -> bool {
        self.inner
            .sessions
            .actors
            .lock()
            .expect("actor registry lock poisoned")
            .contains_key(&session)
    }

    #[cfg(test)]
    pub(crate) fn install_compaction_execution_hook_for_test(
        &self,
    ) -> (oneshot::Receiver<()>, std::sync::Arc<tokio::sync::Notify>) {
        let (reached, receiver) = oneshot::channel();
        let release = std::sync::Arc::new(tokio::sync::Notify::new());
        *self
            .inner
            .test_hooks
            .compaction_execution_hook
            .lock()
            .expect("compaction execution hook lock poisoned") =
            Some(std::sync::Arc::new(PagingRaceHook {
                reached: std::sync::Mutex::new(Some(reached)),
                release: std::sync::Arc::clone(&release),
            }));
        (receiver, release)
    }

    #[cfg(test)]
    pub(crate) async fn enqueue_compact_without_residency_for_test(
        &self,
        session: SessionId,
    ) -> Result<
        oneshot::Receiver<Result<cookie_agent_protocol::SessionCompactResult, EngineError>>,
        EngineError,
    > {
        let actor = self
            .inner
            .sessions
            .actors
            .lock()
            .expect("actor registry lock poisoned")
            .get(&session)
            .cloned()
            .ok_or(EngineError::MissingActor(session))?;
        let (reply, receiver) = oneshot::channel();
        actor
            .send(SessionCommand::Compact {
                focus: None,
                origin: super::event_origin("engine:auto-compact"),
                reply,
            })
            .await
            .map_err(|_| EngineError::ActorStopped)?;
        Ok(receiver)
    }

    pub(super) fn request_blocking<T>(
        &self,
        session: SessionId,
        command: impl FnOnce(oneshot::Sender<Result<T, EngineError>>) -> SessionCommand,
    ) -> Result<T, EngineError> {
        let (reply, receiver) = oneshot::channel();
        {
            let _residency = self.inner.sessions.residency_mutation.blocking_lock();
            self.ensure_session_owned(session)?;
            self.spawn_actor(session)?;
            let actor = self
                .inner
                .sessions
                .actors
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .get(&session)
                .cloned()
                .ok_or(EngineError::MissingActor(session))?;
            actor
                .blocking_send(command(reply))
                .map_err(|_| EngineError::ActorStopped)?;
        }
        receiver
            .blocking_recv()
            .map_err(|_| EngineError::ActorStopped)?
    }

    pub(crate) fn spawn_actor(&self, session: SessionId) -> Result<(), EngineError> {
        let mut actors = self
            .inner
            .sessions
            .actors
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        self.ensure_not_shutting_down()?;
        if !self.inner.store.is_owned(session) {
            return Err(EngineError::SessionOwnedByAnotherProcess(session));
        }
        if actors.contains_key(&session) {
            return Ok(());
        }
        self.inner
            .compaction
            .context_token_estimators
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .entry(session)
            .or_default();
        let engine = self.clone();
        let actor = SessionActor::spawn(SESSION_MAILBOX_CAPACITY, move |command| {
            let engine = engine.clone();
            async move { engine.handle_actor_command(session, command).await }
        });
        actors.insert(session, actor);
        Ok(())
    }

    pub(crate) fn ensure_session_owned(&self, session: SessionId) -> Result<(), EngineError> {
        self.ensure_not_shutting_down()?;
        let adoption_lock = self.inner.store.adoption_lock(session);
        let _adoption = adoption_lock
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        self.ensure_not_shutting_down()?;
        let write_open = self
            .inner
            .store
            .begin_write(session)
            .map_err(|error| match error {
                SessionError::SessionLocked(_) => {
                    EngineError::SessionOwnedByAnotherProcess(session)
                }
                SessionError::StoreClosed => EngineError::ActorStopped,
                error => error.into(),
            })?;
        if let Err(error) = self.ensure_not_shutting_down() {
            if write_open == crate::session::WriteOpen::Adopting {
                self.inner.store.rollback_adoption(session);
            }
            return Err(error);
        }
        if write_open == crate::session::WriteOpen::Adopting {
            if let Err(error) = self.reconcile_session(session) {
                self.inner.store.rollback_adoption(session);
                return Err(error);
            }
            if let Err(error) = self.ensure_not_shutting_down() {
                self.inner.store.rollback_adoption(session);
                return Err(error);
            }
            if let Err(error) = self.inner.store.commit_adoption(session) {
                self.inner.store.rollback_adoption(session);
                return Err(if matches!(error, SessionError::StoreClosed) {
                    EngineError::ActorStopped
                } else {
                    error.into()
                });
            }
            self.ensure_not_shutting_down()?;
            if let Err(error) = self.recover_missing_delegation_children(session) {
                eprintln!("session {session} missing-child recovery deferred: {error}");
            }
            if let Err(error) =
                self.rebuild_delegation_registry(&self.inner.delegation_events.entries(), true)
            {
                eprintln!("session {session} delegation recovery deferred: {error}");
            }
        }
        Ok(())
    }

    pub(super) fn ensure_not_shutting_down(&self) -> Result<(), EngineError> {
        if self
            .inner
            .delegation
            .admission_tasks_closing
            .load(std::sync::atomic::Ordering::Acquire)
        {
            Err(EngineError::ActorStopped)
        } else {
            Ok(())
        }
    }

    pub(super) fn reserve_compaction(&self, session: SessionId) -> bool {
        self.inner
            .compaction
            .in_progress
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .insert(session)
    }

    pub(super) async fn finish_compaction(&self, session: SessionId) -> Result<(), EngineError> {
        self.request(session, |reply| SessionCommand::CompactionFinished {
            reply,
        })
        .await
    }

    async fn release_compaction_direct(&self, session: SessionId) {
        if let Some(state) = self
            .inner
            .sessions
            .producers
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .get_mut(&session)
        {
            state.starting = false;
        }
        self.inner
            .compaction
            .in_progress
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .remove(&session);
        let deferred = self
            .inner
            .compaction
            .deferred
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .remove(&session)
            .unwrap_or_default();
        for command in deferred {
            Box::pin(self.handle_actor_command(session, command)).await;
        }
    }

    pub(super) async fn handle_actor_command(&self, session: SessionId, command: SessionCommand) {
        if matches!(&command, SessionCommand::Start { .. }) {
            {
                let mut registry = self
                    .inner
                    .sessions
                    .producers
                    .lock()
                    .unwrap_or_else(|e| e.into_inner());
                let state = registry.entry(session).or_default();
                if state.starting {
                    state.preempted = true;
                }
            }
        }
        if let Some(kind) = command.compaction_deferred_kind()
            && self
                .inner
                .compaction
                .in_progress
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .contains(&session)
        {
            let rejected = {
                let mut deferred = self
                    .inner
                    .compaction
                    .deferred
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                let queue = deferred.entry(session).or_default();
                let duplicate = queue
                    .iter()
                    .any(|pending| pending.compaction_deferred_kind() == Some(kind));
                let rejected = if duplicate {
                    Some(command)
                } else {
                    queue.push_back(command);
                    None
                };
                debug_assert!(queue.len() <= MAX_COMPACTION_DEFERRED_COMMANDS);
                rejected
            };
            if let Some(rejected) = rejected {
                rejected.reject_duplicate_during_compaction(session);
            }
            return;
        }
        let reconcile = match &command {
            SessionCommand::EvictionBarrier { .. } => false,
            SessionCommand::Append { event, .. } => matches!(
                &**event,
                Event::ModelTurnCommitted { .. }
                    | Event::RunCompleted { .. }
                    | Event::RunFailed { .. }
                    | Event::RunCancelled { .. }
                    | Event::RunInterrupted { .. }
                    | Event::GoalActivated { .. }
                    | Event::GoalChecklistRevised { .. }
                    | Event::GoalLifecycleChanged { .. }
                    | Event::ProducerMessageAccepted { .. }
                    | Event::ProducerMessageConsumed { .. }
                    | Event::ProducerMessageDiscarded { .. }
                    | Event::UserInputAdmitted { .. }
                    | Event::UserInputSubmitted { .. }
                    | Event::UserInputRecalled { .. }
                    | Event::UserInputRecalledV2 { .. }
                    | Event::SessionReverted { .. }
            ),
            SessionCommand::Producer(
                super::producers::ProducerCommand::WakeFinished { .. }
                | super::producers::ProducerCommand::Reconcile { .. }
                | super::producers::ProducerCommand::CommitStart { .. },
            ) => false,
            _ => true,
        };
        match command {
            SessionCommand::Producer(command) => {
                self.handle_producer_command(session, command).await
            }
            SessionCommand::EvictionBarrier { reply } => {
                let _ = reply.send(Ok(()));
            }
            SessionCommand::Append {
                run,
                origin,
                event,
                reply,
            } => {
                #[cfg(test)]
                if matches!(
                    &*event,
                    Event::MessageInjected { .. }
                        | Event::UserInputTransformed { .. }
                        | Event::UserInputSubmitted { .. }
                        | Event::RunFailed { .. }
                ) && self
                    .inner
                    .test_hooks
                    .run_setup_append_failures
                    .fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| {
                        remaining.checked_sub(1)
                    })
                    .is_ok()
                {
                    let event_name = match &*event {
                        Event::MessageInjected { .. } => "message injected",
                        Event::UserInputTransformed { .. } => "user input transformed",
                        Event::UserInputSubmitted { .. } => "user input submitted",
                        Event::RunFailed { .. } => "run failed",
                        _ => unreachable!("run setup failure filter checked"),
                    };
                    let _ = reply.send(Err(EngineError::Mcp(format!(
                        "injected {event_name} append failure"
                    ))));
                    return;
                }
                let reverted = matches!(&*event, Event::SessionReverted { .. });
                let result = self
                    .append_direct(session, run, origin, *event)
                    .and_then(|result| {
                        if reverted {
                            self.inner.delegation_events.reconcile_parent(session)?;
                            self.reconcile_reverted_producers_direct(session)?;
                        }
                        Ok(result)
                    });
                let _ = reply.send(result);
            }
            SessionCommand::EnsureToolCallLinked {
                run,
                tool_call_id,
                child_session_id,
                reply,
            } => {
                let result = (|| {
                    let linked = self
                        .inner
                        .store
                        .get(session)?
                        .log
                        .event_snapshot()
                        .iter()
                        .any(|event| {
                            matches!(event.payload, Event::ToolCallLinked { tool_call_id: linked_call, child_session_id: linked_child }
                                if linked_call == tool_call_id && linked_child == child_session_id)
                        });
                    if !linked {
                        self.append_direct(
                            session,
                            Some(run),
                            super::event_origin("engine:delegation"),
                            Event::ToolCallLinked {
                                tool_call_id,
                                child_session_id,
                            },
                        )?;
                    }
                    Ok(())
                })();
                let _ = reply.send(result);
            }
            SessionCommand::Start {
                params,
                origin,
                admission,
                reply,
            } => {
                if !self.reserve_compaction(session) {
                    let _ = reply.send(Err(EngineError::SessionRunning(session)));
                } else {
                    let engine = self.clone();
                    tokio::spawn(async move {
                        let child_session_id = params.session_id;
                        let mut result = engine
                            .start_run_direct(params, origin, admission, false)
                            .await;
                        if let (Some((invocation_id, generation)), Ok(started)) =
                            (admission, result.as_ref())
                            && let Err(error) = engine.publish_admission_run(
                                invocation_id,
                                generation,
                                child_session_id,
                                started.run_id,
                            )
                        {
                            let _ = engine.cancel_run_durably(
                                started.run_id,
                                Some("delegate admission could not be published".into()),
                            );
                            result = Err(error);
                        }
                        if let Err(error) = engine.finish_compaction(session).await
                            && result.is_ok()
                        {
                            result = Err(error);
                        }
                        let started = result.as_ref().ok().map(|result| result.run_id);
                        if reply.send(result).is_err()
                            && admission.is_some()
                            && let Some(run_id) = started
                        {
                            let _ = engine.cancel_run_durably(
                                run_id,
                                Some("delegate admission reply abandoned".into()),
                            );
                        }
                    });
                }
            }
            SessionCommand::Steer {
                run,
                origin,
                input,
                original_input,
                reply,
            } => {
                let active = self
                    .inner
                    .sessions
                    .active
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner())
                    .get(&run)
                    .cloned()
                    .filter(|active| active.session == session)
                    .ok_or(EngineError::MissingRun(run));
                let result = match active {
                    Err(error) => Err(error),
                    Ok(_) => {
                        let projection = self.inner.store.get(session);
                        match projection {
                            Err(error) => Err(error.into()),
                            Ok(projection)
                                if !projection
                                    .runs
                                    .get(&run)
                                    .is_some_and(|run| run.status == SessionStatus::Running) =>
                            {
                                Ok(RunSteerResult {
                                    accepted: false,
                                    handled_reason: None,
                                })
                            }
                            Ok(_) => (|| {
                                if let Some(original_input) = original_input {
                                    self.append_direct(
                                        session,
                                        Some(run),
                                        origin.clone(),
                                        Event::UserInputTransformed {
                                            original_input,
                                            input: input.clone(),
                                        },
                                    )?;
                                }
                                self.append_direct(
                                    session,
                                    Some(run),
                                    origin,
                                    Event::UserInputAdmitted { input },
                                )?;
                                self.clear_skill_turn_state(session);
                                Ok(RunSteerResult {
                                    accepted: true,
                                    handled_reason: None,
                                })
                            })(),
                        }
                    }
                };
                let _ = reply.send(result);
            }
            SessionCommand::AdmitDelegatedResume { run, input, reply } => {
                let active = self
                    .inner
                    .sessions
                    .active
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner())
                    .get(&run)
                    .cloned()
                    .filter(|active| active.session == session)
                    .ok_or(EngineError::MissingRun(run));
                let result = match active {
                    Err(error) => Err(error),
                    Ok(_) => {
                        let projection = self.inner.store.get(session);
                        match projection {
                            Err(error) => Err(error.into()),
                            Ok(projection)
                                if !projection
                                    .runs
                                    .get(&run)
                                    .is_some_and(|run| run.status == SessionStatus::Running) =>
                            {
                                Ok(super::DelegatedResumeAdmission {
                                    accepted: false,
                                    admission_seq: None,
                                })
                            }
                            Ok(_) => self
                                .append_direct(
                                    session,
                                    Some(run),
                                    super::event_origin("engine:delegation"),
                                    Event::UserInputAdmitted { input },
                                )
                                .and_then(|()| {
                                    self.clear_skill_turn_state(session);
                                    let admission_seq =
                                        self.inner.store.get(session)?.meta.last_event_seq;
                                    Ok(super::DelegatedResumeAdmission {
                                        accepted: true,
                                        admission_seq: Some(admission_seq),
                                    })
                                }),
                        }
                    }
                };
                let _ = reply.send(result);
            }
            SessionCommand::RecallDelegatedResume {
                run,
                admission_seq,
                reply,
            } => {
                let result = (|| {
                    let projection = self.inner.store.get(session)?;
                    if !projection.runs.contains_key(&run) {
                        return Err(EngineError::MissingRun(run));
                    }
                    let recalled = pending_inputs(&projection.log.event_snapshot(), run)
                        .into_iter()
                        .find(|pending| pending.admission_seq == admission_seq);
                    if let Some(pending) = recalled {
                        self.append_direct(
                            session,
                            Some(run),
                            super::event_origin("engine:delegation"),
                            Event::UserInputRecalledV2 {
                                user_input_seq: admission_seq,
                                input: pending.input,
                            },
                        )?;
                        Ok(true)
                    } else {
                        Ok(false)
                    }
                })();
                let _ = reply.send(result);
            }
            SessionCommand::RecallSteer { run, reply } => {
                let result = (|| {
                    self.inner
                        .sessions
                        .active
                        .lock()
                        .unwrap_or_else(|poisoned| poisoned.into_inner())
                        .get(&run)
                        .cloned()
                        .filter(|active| active.session == session)
                        .ok_or(EngineError::MissingRun(run))?;
                    let projection = self.inner.store.get(session)?;
                    if !projection
                        .runs
                        .get(&run)
                        .is_some_and(|run| run.status == SessionStatus::Running)
                    {
                        return Ok(RunRecallSteerResult { recalled: None });
                    }
                    let recalled = pending_inputs(&projection.log.event_snapshot(), run)
                        .pop()
                        .map(|pending| pending.input);
                    if let Some(input) = &recalled {
                        self.append_direct(
                            session,
                            Some(run),
                            super::event_origin("user"),
                            Event::UserInputRecalled {
                                input: input.clone(),
                            },
                        )?;
                    }
                    Ok(RunRecallSteerResult { recalled })
                })();
                let _ = reply.send(result);
            }
            SessionCommand::CommitPendingPromotion {
                run,
                through_admission_seq,
                final_text,
                complete_if_empty,
                already_promoted,
                reply,
            } => {
                let result = (|| {
                    let projection = self.inner.store.get(session)?;
                    if !projection
                        .runs
                        .get(&run)
                        .is_some_and(|run| run.status == SessionStatus::Running)
                    {
                        return Ok(PendingPromotionState {
                            promoted: already_promoted,
                            pending: Vec::new(),
                            continue_run: false,
                        });
                    }
                    let eligible = pending_inputs(&projection.log.event_snapshot(), run)
                        .into_iter()
                        .take_while(|pending| pending.admission_seq <= through_admission_seq)
                        .collect::<Vec<_>>();
                    for pending in &eligible {
                        self.append_direct(
                            session,
                            Some(run),
                            pending.origin.clone(),
                            Event::UserInputSubmitted {
                                input: pending.input.clone(),
                            },
                        )?;
                    }
                    let producer_promoted =
                        self.promote_producer_inputs_direct(session, run, false)?;
                    let promoted = already_promoted || !eligible.is_empty() || producer_promoted;
                    let pending =
                        pending_inputs(&self.inner.store.get(session)?.log.event_snapshot(), run);
                    if pending.is_empty() && !promoted && complete_if_empty {
                        self.append_direct(
                            session,
                            Some(run),
                            super::event_origin("engine:model-loop"),
                            Event::RunCompleted { final_text },
                        )?;
                    }
                    Ok(PendingPromotionState {
                        promoted,
                        continue_run: promoted,
                        pending,
                    })
                })();
                if result.as_ref().is_ok_and(|state| state.pending.is_empty()) {
                    self.release_compaction_direct(session).await;
                }
                let _ = reply.send(result);
            }
            SessionCommand::Compact {
                focus,
                origin,
                reply,
            } => {
                if !self.reserve_compaction(session) {
                    let _ = reply.send(Err(EngineError::SessionRunning(session)));
                } else {
                    let engine = self.clone();
                    tokio::spawn(async move {
                        #[cfg(test)]
                        let hook = engine
                            .inner
                            .test_hooks
                            .compaction_execution_hook
                            .lock()
                            .expect("compaction execution hook lock poisoned")
                            .take();
                        #[cfg(test)]
                        if let Some(hook) = hook {
                            if let Some(reached) = hook
                                .reached
                                .lock()
                                .expect("compaction execution reached lock poisoned")
                                .take()
                            {
                                let _ = reached.send(());
                            }
                            hook.release.notified().await;
                        }
                        let mut result = engine
                            .compact_session_direct(session, focus.as_deref(), origin)
                            .await;
                        if let Err(error) = engine.finish_compaction(session).await
                            && result.is_ok()
                        {
                            result = Err(error);
                        }
                        let _ = reply.send(result);
                    });
                }
            }
            SessionCommand::Revert {
                through_seq,
                origin,
                instructions_override,
                reply,
            } => {
                if !self.reserve_compaction(session) {
                    let _ = reply.send(Err(EngineError::SessionRunning(session)));
                } else {
                    let result = (|| {
                        let projection = self.inner.store.get(session)?;
                        if projection.status == SessionStatus::Running {
                            return Err(EngineError::SessionRunning(session));
                        }
                        let tip = projection
                            .log
                            .all_events()
                            .last()
                            .map_or(0, |event| event.seq);
                        if through_seq == 0 || through_seq > tip {
                            return Err(SessionError::InvalidSequence {
                                session_id: session,
                                through_seq,
                            }
                            .into());
                        }
                        self.append_direct(
                            session,
                            None,
                            origin,
                            Event::SessionReverted { through_seq },
                        )?;
                        self.inner
                            .compaction
                            .context_token_estimators
                            .lock()
                            .unwrap_or_else(|poisoned| poisoned.into_inner())
                            .remove(&session);
                        self.rebuild_visible_tree_grants();
                        self.inner.delegation_events.reconcile_parent(session)?;
                        self.reconcile_reverted_producers_direct(session)?;
                        Ok(SessionRevertResult {
                            session: self.inner.store.get(session)?.metadata(),
                            instructions_override,
                        })
                    })();
                    self.release_compaction_direct(session).await;
                    let _ = reply.send(result);
                }
            }
            SessionCommand::CompactionFinished { reply } => {
                self.release_compaction_direct(session).await;
                let _ = reply.send(Ok(()));
            }
            SessionCommand::Cancel { run, reply } => {
                let result = (|| {
                    let active = self
                        .inner
                        .sessions
                        .active
                        .lock()
                        .unwrap_or_else(|poisoned| poisoned.into_inner())
                        .get(&run)
                        .cloned()
                        .filter(|active| active.session == session)
                        .ok_or(EngineError::MissingRun(run))?;
                    active.cancellation.cancel();
                    active
                        .stdin
                        .lock()
                        .unwrap_or_else(|poisoned| poisoned.into_inner())
                        .clear();
                    let events = self.inner.store.get(session)?.log.event_snapshot();
                    let pending = approval_records(session, &events)
                        .into_values()
                        .filter(|record| {
                            matches!(
                                record.status,
                                ApprovalStatus::Pending | ApprovalStatus::Escalated
                            ) && approval_run_id(&events, record.request.approval_id()) == Some(run)
                        })
                        .map(|record| record.request.approval_id())
                        .collect::<Vec<_>>();
                    for approval_id in pending {
                        self.approval_terminal_direct(
                            session,
                            run,
                            approval_id,
                            ApprovalTerminal::Cancelled,
                        )?;
                    }
                    Ok(RunCancelResult { cancelled: true })
                })();
                let _ = reply.send(result);
            }
            SessionCommand::Stdin { params, reply } => {
                let result = (|| {
                    let active = self
                        .inner
                        .sessions
                        .active
                        .lock()
                        .unwrap_or_else(|poisoned| poisoned.into_inner())
                        .get(&params.run_id)
                        .cloned()
                        .filter(|active| active.session == session)
                        .ok_or(EngineError::MissingRun(params.run_id))?;
                    let data = params
                        .data
                        .map(|encoded| STANDARD.decode(encoded))
                        .transpose()?
                        .unwrap_or_default();
                    let sender = active
                        .stdin
                        .lock()
                        .unwrap_or_else(|poisoned| poisoned.into_inner())
                        .get(&params.call_id)
                        .cloned()
                        .ok_or(EngineError::StdinUnavailable)?;
                    sender
                        .try_send(StdinWrite {
                            data: data.clone(),
                            eof: params.eof,
                        })
                        .map_err(|_| EngineError::StdinUnavailable)?;
                    if params.eof {
                        active
                            .stdin
                            .lock()
                            .unwrap_or_else(|poisoned| poisoned.into_inner())
                            .remove(&params.call_id);
                    }
                    self.append_direct(
                        session,
                        Some(params.run_id),
                        super::event_origin("engine:tool-execution"),
                        Event::ToolStdinSubmitted {
                            tool_call_id: params.call_id,
                            byte_count: data.len() as u64,
                        },
                    )?;
                    Ok(RunToolStdinResult { accepted: true })
                })();
                let _ = reply.send(result);
            }
            SessionCommand::Subscribe { cursor, reply } => {
                let result = self.inner.store.subscribe_events(session, cursor);
                let _ = reply.send(result.map_err(EngineError::from));
            }
            SessionCommand::Resume { reply } => {
                let result = self
                    .resolve_interrupted_direct(session)
                    .await
                    .and_then(|()| Ok(self.inner.store.get(session)?.metadata()));
                let _ = reply.send(result);
            }
            SessionCommand::Rename {
                params,
                origin: _request_origin,
                reply,
            } => {
                let result = (|| {
                    let projection = self.inner.store.get(session)?;
                    if let Some(record) = projection.rename_records.get(&params.client_rename_id) {
                        if record.conflicts_with(&params) {
                            return Err(EngineError::RenameConflict);
                        }
                        return Ok(SessionRenameResult {
                            client_rename_id: params.client_rename_id,
                            session: projection.metadata(),
                        });
                    }
                    let commit = match params.change {
                        SessionRenameChange::Set { title } => SessionTitleChange::UserSet {
                            title,
                            client_rename_id: params.client_rename_id.clone(),
                        },
                        SessionRenameChange::Clear => SessionTitleChange::UserClear {
                            client_rename_id: params.client_rename_id.clone(),
                        },
                        SessionRenameChange::Reset => SessionTitleChange::UserReset {
                            client_rename_id: params.client_rename_id.clone(),
                        },
                    };
                    let input_through_seq = projection
                        .log
                        .all_events()
                        .last()
                        .map_or(0, |event| event.seq);
                    self.append_direct(
                        session,
                        None,
                        super::event_origin("user"),
                        Event::SessionTitleCommitted {
                            input_through_seq,
                            change: commit,
                        },
                    )?;
                    Ok(SessionRenameResult {
                        client_rename_id: params.client_rename_id,
                        session: self.inner.store.get(session)?.metadata(),
                    })
                })();
                let _ = reply.send(result);
            }
            SessionCommand::ApprovalRespond {
                params,
                origin: _request_origin,
                reply,
            } => {
                let _ = reply.send(self.approval_respond_direct(params));
            }
            SessionCommand::ApprovalCapabilityInvalid {
                params,
                invalidation,
                reply,
            } => {
                let _ = reply.send(self.approval_capability_invalid_direct(params, invalidation));
            }
            SessionCommand::ApprovalEvaluationComplete {
                run,
                request,
                executor,
                decision,
                permission_mode,
                cancelled,
                reply,
            } => {
                let _ = reply.send(self.approval_evaluation_complete_direct(
                    session,
                    run,
                    request,
                    executor,
                    (permission_mode, decision),
                    cancelled,
                ));
            }
            SessionCommand::ApprovalTerminal {
                run,
                approval_id,
                terminal,
                reply,
            } => {
                let _ =
                    reply.send(self.approval_terminal_direct(session, run, approval_id, terminal));
            }
            SessionCommand::ToolResult {
                run,
                tool_call_id,
                result,
                cancelled,
                reply,
            } => {
                let pending = self
                    .inner
                    .store
                    .get(session)
                    .ok()
                    .and_then(|projection| projection.runs.get(&run).cloned())
                    .is_some_and(|run| run.pending_calls.contains_key(&tool_call_id));
                let response = if !pending {
                    Ok(false)
                } else {
                    (|| {
                        let owner = self.tool_call_owner(session, run, tool_call_id)?;
                        let event = match result {
                            Ok(result) => Event::ToolCallTerminated {
                                termination: ToolCallTermination {
                                    tool_call_id,
                                    owner,
                                    outcome: if cancelled {
                                        ToolTerminationOutcome::Cancelled
                                    } else {
                                        ToolTerminationOutcome::Completed
                                    },
                                    result: Some(result),
                                    error: cancelled.then(|| SafeToolError {
                                        code: super::safe_code(super::CANCELLED_AFTER_COMPLETION),
                                        message: safe_error("tool call cancelled after it started"),
                                    }),
                                },
                            },
                            Err(failure) => Event::ToolCallTerminated {
                                termination: ToolCallTermination {
                                    tool_call_id,
                                    owner,
                                    outcome: if cancelled {
                                        ToolTerminationOutcome::Cancelled
                                    } else {
                                        ToolTerminationOutcome::Failed
                                    },
                                    result: failure.partial_output.map(|result| *result),
                                    error: Some(SafeToolError {
                                        code: failure.code.safe_code(),
                                        message: safe_error(&failure.message),
                                    }),
                                },
                            },
                        };
                        self.append_direct(
                            session,
                            Some(run),
                            super::event_origin("engine:tool-result"),
                            event,
                        )?;
                        self.inner.store.get(session)?.log.flush()?;
                        Ok(true)
                    })()
                };
                if let Some(capture) = self
                    .inner
                    .output
                    .captures
                    .lock()
                    .unwrap_or_else(|p| p.into_inner())
                    .remove(&tool_call_id)
                {
                    capture.release_publication();
                }
                let _ = reply.send(response);
            }
            SessionCommand::ResolveDelegateFailureIfPending {
                run,
                tool_call_id,
                result,
                reply,
            } => {
                let result = self.resolve_delegate_failure_if_pending_direct(
                    session,
                    run,
                    tool_call_id,
                    result,
                );
                let _ = reply.send(result);
            }
            SessionCommand::ResolveAbandonedDelegateFailureIfPending {
                invocation_id,
                generation,
                run,
                tool_call_id,
                result,
                reply,
            } => {
                let result = self.resolve_abandoned_delegate_failure_if_pending_direct(
                    invocation_id,
                    generation,
                    session,
                    run,
                    tool_call_id,
                    result,
                );
                let _ = reply.send(result);
            }
            SessionCommand::PromotePendingOrComplete {
                run,
                final_text,
                complete_if_empty,
                reply,
            } => {
                let producer_promoted =
                    match self.promote_producer_inputs_direct(session, run, false) {
                        Ok(promoted) => promoted,
                        Err(error) => {
                            let _ = reply.send(Err(error));
                            return;
                        }
                    };
                let active = self
                    .inner
                    .sessions
                    .active
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner())
                    .get(&run)
                    .cloned()
                    .filter(|active| active.session == session)
                    .ok_or(EngineError::MissingRun(run));
                let active = match active {
                    Ok(active) => active,
                    Err(error) => {
                        let _ = reply.send(Err(error));
                        return;
                    }
                };
                let pending = match self.inner.store.get(session) {
                    Ok(projection) => pending_inputs(&projection.log.event_snapshot(), run),
                    Err(error) => {
                        let _ = reply.send(Err(error.into()));
                        return;
                    }
                };
                if pending.is_empty() {
                    let result = if complete_if_empty && !producer_promoted {
                        self.append_direct(
                            session,
                            Some(run),
                            super::event_origin("engine:model-loop"),
                            Event::RunCompleted { final_text },
                        )
                    } else {
                        Ok(())
                    }
                    .map(|()| producer_promoted);
                    let _ = reply.send(result);
                } else if !self.reserve_compaction(session) {
                    let _ = reply.send(Err(EngineError::SessionRunning(session)));
                } else {
                    let engine = self.clone();
                    tokio::spawn(async move {
                        let mut pending = pending;
                        let mut promoted = false;
                        let result = loop {
                            let fallback_index =
                                active.fallback_index.load(Ordering::Acquire) as usize;
                            let through_admission_seq = pending
                                .last()
                                .expect("pending batch is nonempty")
                                .admission_seq;
                            let mut failed = None;
                            let mut serialized_message_bytes = 0_usize;
                            for pending_input in &pending {
                                let input_bytes = match model_loop::serialized_input_bytes(
                                    &pending_input.input,
                                ) {
                                    Ok(input_bytes) => input_bytes,
                                    Err(error) => {
                                        failed = Some(error);
                                        break;
                                    }
                                };
                                serialized_message_bytes =
                                    serialized_message_bytes.saturating_add(input_bytes);
                                match engine
                                    .maybe_predictive_compact_before_input(
                                        PredictiveCompactionInput {
                                            session,
                                            run,
                                            serialized_message_bytes,
                                            policy: &active.policy,
                                            fallback_index,
                                            cancellation: &active.cancellation,
                                            actor_direct: false,
                                        },
                                    )
                                    .await
                                {
                                    Ok(true) => break,
                                    Ok(false) => {}
                                    Err(error) => {
                                        failed = Some(error);
                                        break;
                                    }
                                }
                            }
                            if let Some(error) = failed {
                                break Err(error);
                            }
                            match engine
                                .request(session, |reply| SessionCommand::CommitPendingPromotion {
                                    run,
                                    through_admission_seq,
                                    final_text: final_text.clone(),
                                    complete_if_empty,
                                    already_promoted: promoted,
                                    reply,
                                })
                                .await
                            {
                                Ok(state) if state.pending.is_empty() => {
                                    break Ok(state.continue_run);
                                }
                                Ok(state) => {
                                    promoted = state.promoted;
                                    pending = state.pending;
                                }
                                Err(error) => break Err(error),
                            }
                        };
                        let mut result = result;
                        if result.is_err()
                            && let Err(error) = engine.finish_compaction(session).await
                        {
                            result = Err(error);
                        }
                        let _ = reply.send(result);
                    });
                }
            }
            SessionCommand::PromotePendingInputs { run, reply } => {
                let result = self
                    .inner
                    .sessions
                    .active
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner())
                    .get(&run)
                    .cloned()
                    .filter(|active| active.session == session)
                    .ok_or(EngineError::MissingRun(run))
                    .and_then(|_| {
                        self.promote_producer_inputs_direct(session, run, false)?;
                        let events = self.inner.store.get(session)?.log.event_snapshot();
                        let applied: HashSet<u64> = events
                            .iter()
                            .filter_map(|event| match &event.payload {
                                Event::UserInputApplied { user_input_seq }
                                    if event.run_id == Some(run) =>
                                {
                                    Some(*user_input_seq)
                                }
                                _ => None,
                            })
                            .collect();
                        for user_input_seq in
                            events.iter().filter_map(|event| match &event.payload {
                                Event::UserInputSubmitted { .. }
                                    if event.run_id == Some(run)
                                        && !applied.contains(&event.seq) =>
                                {
                                    Some(event.seq)
                                }
                                _ => None,
                            })
                        {
                            self.append_direct(
                                session,
                                Some(run),
                                super::event_origin("engine:model-loop"),
                                Event::UserInputApplied { user_input_seq },
                            )?;
                        }
                        self.claim_producer_snapshot_direct(session, run)
                    });
                let _ = reply.send(result);
            }
        }
        if reconcile && let Err(error) = self.reconcile_producers_direct(session, None) {
            eprintln!("session {session} producer reconciliation failed: {error}");
        }
    }
}

fn pending_inputs(events: &[StoredEvent], run: RunId) -> Vec<PendingInput> {
    let mut pending = VecDeque::new();
    let mut initial_input_submitted = false;
    let run_start_seq = events.iter().find_map(|event| {
        (event.run_id == Some(run) && matches!(event.payload, Event::RunStarted { .. }))
            .then_some(event.seq)
    });
    let previous_terminal_seq = run_start_seq
        .and_then(|run_start_seq| {
            events
                .iter()
                .rev()
                .find(|event| {
                    event.seq < run_start_seq
                        && matches!(
                            event.payload,
                            Event::RunCompleted { .. }
                                | Event::RunFailed { .. }
                                | Event::RunCancelled { .. }
                                | Event::RunInterrupted { .. }
                        )
                })
                .map(|event| event.seq)
        })
        .unwrap_or(0);
    for event in events.iter().filter(|event| {
        event.run_id == Some(run)
            || (run_start_seq.is_some_and(|run_start_seq| event.seq < run_start_seq)
                && event.seq > previous_terminal_seq
                && event.run_id.is_none()
                && matches!(
                    event.payload,
                    Event::UserInputAdmitted { .. } | Event::UserInputRecalled { .. }
                ))
    }) {
        match &event.payload {
            Event::ProducerMessageAdmitted { .. } => initial_input_submitted = true,
            Event::UserInputAdmitted { input } => pending.push_back(PendingInput {
                admission_seq: event.seq,
                origin: event
                    .origin
                    .clone()
                    .unwrap_or_else(|| super::event_origin("user")),
                input: input.clone(),
            }),
            Event::UserInputRecalled { .. } => {
                pending.pop_back();
            }
            Event::UserInputRecalledV2 { user_input_seq, .. } => {
                if let Some(position) = pending
                    .iter()
                    .position(|pending| pending.admission_seq == *user_input_seq)
                {
                    pending.remove(position);
                }
            }
            Event::UserInputSubmitted { .. } => {
                if initial_input_submitted {
                    pending.pop_front();
                } else {
                    initial_input_submitted = true;
                }
            }
            Event::RunCompleted { .. }
            | Event::RunFailed { .. }
            | Event::RunCancelled { .. }
            | Event::RunInterrupted { .. } => pending.clear(),
            _ => {}
        }
    }
    pending.into_iter().collect()
}

#[cfg(test)]
mod tests {
    use cookie_agent_protocol::{RunId, SessionId, StoredEvent};

    use super::{Event, PendingInput, pending_inputs};

    fn event(run: RunId, seq: u64, payload: Event) -> StoredEvent {
        StoredEvent {
            engine_version: None,
            origin: None,
            session_id: SessionId::new_v7(),
            run_id: Some(run),
            seq,
            timestamp: jiff::Timestamp::now(),
            payload,
        }
    }

    #[test]
    fn pending_lane_replays_fifo_promotion_lifo_recall_and_terminal_voiding() {
        let run = RunId::new_v7();
        let events = vec![
            event(
                run,
                1,
                Event::UserInputSubmitted {
                    input: "initial".into(),
                },
            ),
            event(
                run,
                2,
                Event::UserInputAdmitted {
                    input: "one".into(),
                },
            ),
            event(
                run,
                3,
                Event::UserInputAdmitted {
                    input: "two".into(),
                },
            ),
            event(
                run,
                4,
                Event::UserInputRecalled {
                    input: "two".into(),
                },
            ),
            event(
                run,
                5,
                Event::UserInputAdmitted {
                    input: "three".into(),
                },
            ),
            event(
                run,
                6,
                Event::UserInputSubmitted {
                    input: "one".into(),
                },
            ),
        ];
        assert_eq!(
            pending_inputs(&events, run),
            vec![PendingInput {
                admission_seq: 5,
                origin: cookie_agent_protocol::EventOrigin::new("user").unwrap(),
                input: "three".into(),
            }]
        );
        let replayed = serde_json::from_slice::<Vec<StoredEvent>>(
            &serde_json::to_vec(&events).expect("serialize replay events"),
        )
        .expect("deserialize replay events");
        assert_eq!(pending_inputs(&replayed, run), pending_inputs(&events, run));

        let mut terminal = replayed;
        terminal.push(event(run, 7, Event::RunCancelled { reason: None }));
        assert!(pending_inputs(&terminal, run).is_empty());
    }

    #[test]
    fn targeted_recall_removes_only_its_admission_sequence() {
        let run = RunId::new_v7();
        let events = vec![
            event(
                run,
                1,
                Event::UserInputSubmitted {
                    input: "initial".into(),
                },
            ),
            event(
                run,
                2,
                Event::UserInputAdmitted {
                    input: "resume".into(),
                },
            ),
            event(
                run,
                3,
                Event::UserInputAdmitted {
                    input: "direct steer".into(),
                },
            ),
            event(
                run,
                4,
                Event::UserInputRecalledV2 {
                    user_input_seq: 2,
                    input: "resume".into(),
                },
            ),
        ];
        assert_eq!(
            pending_inputs(&events, run),
            vec![PendingInput {
                admission_seq: 3,
                origin: cookie_agent_protocol::EventOrigin::new("user").unwrap(),
                input: "direct steer".into(),
            }]
        );
    }
}