cookie_agent_engine/runtime/

delegation.rs

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

use cookie_agent_protocol::{
    DelegateRequestPayload, DelegatedContextRole, DelegatedContextTurn, InvocationId,
    PersistedAssistantPart, PersistedToolResult as ToolResult, ProducerId, ProducerOwner, RunId,
    RunStartParams, SafeToolError, SessionId, SessionOrigin, SessionStatus, ToolCallId,
    ToolCallTermination, ToolTerminationOutcome,
};
use serde_json::Value;
use tokio::{sync::oneshot, task::JoinHandle};
use tokio_util::sync::CancellationToken;

use super::{
    Engine, EngineError, Event, SessionCommand,
    admission::{AdmissionGuard, InflightDelegation},
    handles::SubagentScope,
    helpers::{invocation_id, safe_code, safe_display, safe_error, session_depth},
    producers::ProducerAuthority,
};
use crate::{
    EngineHistoryView,
    delegation_api::{DelegateAwait, DelegateHandle, DelegateInvocation},
    delegation_events::{self, DelegationEventError},
    policy::{self, FrozenRunPolicy, freeze_delegated_agent_policy, resolve_agent},
    session,
};

/// How long [`Engine::resume`] waits for the delegation recovery its adoption
/// scheduled before giving up on the ordering guarantee.
///
/// Recovery is a handful of local appends and mailbox round-trips per recovered
/// child, so this is generous. Exceeding it is not an error: the wait degrades
/// to the eventually-consistent behaviour, logs once, and lets the resume
/// succeed.
pub(super) const DELEGATION_RECOVERY_SETTLE_BOUND: std::time::Duration =
    std::time::Duration::from_secs(5);

/// Delegation runtime state owned by [`super::Inner`].
pub(crate) struct DelegationRuntimeState {
    pub(super) inflight: Mutex<HashMap<InvocationId, HashMap<u64, InflightDelegation>>>,
    pub(super) by_session: Mutex<HashMap<SessionId, DelegationRecord>>,
    pub(crate) queue: Mutex<VecDeque<SessionId>>,
    pub(crate) admission: tokio::sync::Mutex<()>,
    pub(crate) reconciliation_running: AtomicBool,
    pub(crate) reconciliation_requested: AtomicBool,
    pub(crate) recovery_stale_producers: Mutex<Vec<(SessionId, InvocationId, ProducerId)>>,
    pub(crate) next_admission_generation: AtomicU64,
    pub(crate) admission_tasks: Mutex<Vec<JoinHandle<()>>>,
    pub(crate) admission_blocking_tasks: Mutex<Vec<JoinHandle<()>>>,
    pub(crate) admission_tasks_closing: AtomicBool,
    pub(crate) recovery_waiters: Mutex<HashSet<(SessionId, RunId, ToolCallId)>>,
    /// Recovered background monitors claimed but not yet resolved, per root.
    ///
    /// A claim is made under [`Engine::claim_recovered_monitor`] and released
    /// when the monitor that took it has finished resolving its delegation
    /// record. [`Engine::await_delegation_recovery`] reads this to decide when
    /// an adoption's recovery has settled.
    pub(crate) recovery_pending: Mutex<HashMap<SessionId, usize>>,
    /// Woken whenever a recovery claim is released or a reconciliation pass
    /// ends, so a settling waiter need not poll on its own.
    pub(crate) recovery_settled: tokio::sync::Notify,
}

impl Default for DelegationRuntimeState {
    fn default() -> Self {
        Self {
            inflight: Mutex::new(HashMap::new()),
            by_session: Mutex::new(HashMap::new()),
            queue: Mutex::new(VecDeque::new()),
            admission: tokio::sync::Mutex::new(()),
            reconciliation_running: AtomicBool::new(false),
            reconciliation_requested: AtomicBool::new(false),
            recovery_stale_producers: Mutex::new(Vec::new()),
            // Generation zero is the "never admitted" sentinel.
            next_admission_generation: AtomicU64::new(1),
            admission_tasks: Mutex::new(Vec::new()),
            admission_blocking_tasks: Mutex::new(Vec::new()),
            admission_tasks_closing: AtomicBool::new(false),
            recovery_waiters: Mutex::new(HashSet::new()),
            recovery_pending: Mutex::new(HashMap::new()),
            recovery_settled: tokio::sync::Notify::new(),
        }
    }
}

impl Engine {
    pub(crate) fn validate_resume_target(
        &self,
        parent_session_id: SessionId,
        resume_session_id: SessionId,
        agent_type: &cookie_agent_protocol::AgentId,
        preset: Option<&str>,
    ) -> Result<session::SessionProjection, EngineError> {
        if resume_session_id == parent_session_id {
            return Err(EngineError::ToolFailed(
                "resume_session_id cannot reference the delegating session itself".into(),
            ));
        }
        let mut cursor = parent_session_id;
        let mut visited = std::collections::HashSet::new();
        while visited.insert(cursor) {
            let current = self.inner.store.get(cursor)?;
            let SessionOrigin::Delegated {
                parent_session_id, ..
            } = current.meta.origin
            else {
                break;
            };
            if parent_session_id == resume_session_id {
                return Err(EngineError::ToolFailed(
                    "resume_session_id cannot reference an ancestor session".into(),
                ));
            }
            cursor = parent_session_id;
        }
        let resumed = self.inner.store.get(resume_session_id).map_err(|_| {
            EngineError::ToolFailed(format!("resume session {resume_session_id} was not found"))
        })?;
        let direct_child = matches!(
            resumed.meta.origin,
            SessionOrigin::Delegated {
                parent_session_id: owner,
                ..
            } if owner == parent_session_id
        ) && self.inner.delegation_events.entries().iter().any(|entry| {
            entry.reservation.parent_session_id == parent_session_id
                && entry.reservation.child_session_id == resume_session_id
        });
        if !direct_child {
            return Err(EngineError::ToolFailed(
                "resume session is not a prior direct child of the delegating parent".into(),
            ));
        }
        if &resumed.meta.creation_selection.agent != agent_type {
            return Err(EngineError::ToolFailed(
                "resume session agent does not match agent_type".into(),
            ));
        }
        if resumed.meta.creation_selection.preset.as_deref() != preset {
            return Err(EngineError::ToolFailed(
                "resume target belongs to a different agent preset".into(),
            ));
        }
        if let Some(record) = self
            .inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .get(&resume_session_id)
            .copied()
            && matches!(
                record.state,
                DelegationState::Queued | DelegationState::Starting
            )
        {
            return Err(EngineError::ToolFailed(format!(
                "resume session already has in-flight delegation {} in {} state",
                record.invocation_id,
                if record.state == DelegationState::Queued {
                    "queued"
                } else {
                    "starting"
                }
            )));
        }
        Ok(resumed)
    }

    async fn delegated_context_seed(
        &self,
        parent_session_id: SessionId,
    ) -> Result<Vec<DelegatedContextTurn>, EngineError> {
        let history = self
            .get_history(parent_session_id, EngineHistoryView::Assembled)
            .await?;
        Ok(context_seed_from_history(history))
    }

    pub(super) fn is_delegate_call_result(
        &self,
        session: SessionId,
        run: RunId,
        tool_call_id: ToolCallId,
        result: &ToolResult,
    ) -> bool {
        let Ok(parent) = self.inner.store.get(session) else {
            return false;
        };
        if parent
            .runs
            .get(&run)
            .and_then(|run| run.pending_calls.get(&tool_call_id))
            .is_none_or(|name| name != "delegate_subagent")
        {
            return false;
        }
        self.inner
            .delegation_events
            .get(invocation_id(session, run, tool_call_id))
            .is_some_and(|entry| {
                entry.reservation.parent_session_id == session
                    && entry.reservation.parent_run_id == run
                    && entry.reservation.parent_tool_call_id == tool_call_id
                    && crate::delegation_api::delegate_result_matches_child(
                        result,
                        entry.reservation.child_session_id,
                    )
            })
    }

    /// Serializes the durable parent backlink per invocation. Every admission
    /// path re-checks under this barrier; only the first appends it.
    pub(super) async fn ensure_parent_link(
        &self,
        parent_session_id: SessionId,
        parent_run_id: RunId,
        parent_tool_call_id: ToolCallId,
        child_session_id: SessionId,
    ) -> Result<(), EngineError> {
        self.request(parent_session_id, |reply| {
            SessionCommand::EnsureToolCallLinked {
                run: parent_run_id,
                tool_call_id: parent_tool_call_id,
                child_session_id,
                reply,
            }
        })
        .await
    }

    pub(super) async fn terminal_parent_delegate(
        &self,
        parent_session_id: SessionId,
        parent_run_id: RunId,
        parent_tool_call_id: ToolCallId,
    ) -> Result<bool, EngineError> {
        let parent = self.inner.store.get(parent_session_id)?;
        let Some(run) = parent.runs.get(&parent_run_id) else {
            return Ok(true);
        };
        if !matches!(
            run.status,
            SessionStatus::Cancelled | SessionStatus::Failed | SessionStatus::Completed
        ) {
            return Ok(false);
        }
        if run
            .pending_calls
            .get(&parent_tool_call_id)
            .is_some_and(|tool| tool == "delegate_subagent")
        {
            let result = cancelled_delegate_result_with_reason(
                None,
                None,
                "parent run was already terminal",
            );
            self.append(
                parent_session_id,
                Some(parent_run_id),
                super::event_origin("engine:delegation"),
                Event::ToolCallTerminated {
                    termination: ToolCallTermination {
                        tool_call_id: parent_tool_call_id,
                        owner: self.tool_call_owner(
                            parent_session_id,
                            parent_run_id,
                            parent_tool_call_id,
                        )?,
                        outcome: ToolTerminationOutcome::Cancelled,
                        result: Some(result),
                        error: Some(SafeToolError {
                            code: safe_code("parent_terminal"),
                            message: safe_error("parent run was already terminal"),
                        }),
                    },
                },
            )
            .await?;
        }
        Ok(true)
    }

    /// Admits a delegate invocation, creates/attaches its child, and starts the
    /// invocation-derived child run exactly once.
    pub async fn delegate_invoke(
        &self,
        invocation: DelegateInvocation,
    ) -> Result<DelegateHandle, EngineError> {
        let invocation_id = invocation_id(
            invocation.parent_session_id,
            invocation.parent_run_id,
            invocation.parent_tool_call_id,
        );
        let mut admission =
            AdmissionGuard::begin(self.inner.clone(), invocation_id, invocation.parent_run_id);
        admission.set_parent(invocation.parent_session_id, invocation.parent_tool_call_id);
        // The admission task, rather than this observer, owns the durable start
        // confirmation and therefore survives a dropped caller future.
        let (reply, receiver) = oneshot::channel();
        let engine = self.clone();
        let generation = admission.generation;
        let Some(runtime) = tokio::runtime::Handle::try_current().ok() else {
            return Err(EngineError::ActorStopped);
        };
        if !self.spawn_admission_task(&runtime, async move {
            let result = engine
                .delegate_invoke_admitted(invocation, invocation_id, generation)
                .await;
            let _ = reply.send(result);
        }) {
            return Err(EngineError::ActorStopped);
        }
        let result = receiver.await.map_err(|_| EngineError::ActorStopped)?;
        if result.is_ok() {
            admission.handoff();
        } else {
            admission.complete();
        }
        result
    }

    pub(super) async fn delegate_invoke_admitted(
        &self,
        invocation: DelegateInvocation,
        invocation_id: InvocationId,
        generation: u64,
    ) -> Result<DelegateHandle, EngineError> {
        let admission_guard = self.inner.delegation.admission.lock().await;
        if invocation
            .prompt
            .starts_with(super::skills::RESERVED_STAGED_SKILL_PREFIX)
        {
            return Err(EngineError::ToolFailed(
                "delegate prompt uses a reserved staged-skill prefix".into(),
            ));
        }
        let staged_skill = self
            .inner
            .skills_runtime
            .pending_forks
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .get(&invocation.parent_tool_call_id)
            .map(super::skills::PreparedSkillInvocation::staged_payload);
        if invocation.resume_session_id.is_some() && invocation.inherit_context {
            return Err(EngineError::ToolFailed(
                "resume_session_id and inherit_context cannot both be set".into(),
            ));
        }
        let resume_session_id = invocation
            .resume_session_id
            .as_deref()
            .map(|reference| {
                self.resolve_subagent_target(
                    invocation.parent_session_id,
                    reference,
                    SubagentScope::DirectChildren,
                )
            })
            .transpose()?;
        if let Some(existing) = self.delegation_event_get(invocation_id).await?
            && existing.child_run_id.is_some()
        {
            validate_redelivery_mode(existing.request.background, invocation.background)?;
            let request_matches = existing.reservation.parent_session_id
                == invocation.parent_session_id
                && existing.reservation.parent_run_id == invocation.parent_run_id
                && existing.reservation.parent_tool_call_id == invocation.parent_tool_call_id
                && existing.child_agent.agent == invocation.agent_type
                && existing.request.description == invocation.description
                && existing.request.prompt == invocation.prompt
                && existing.request.resume_session_id == resume_session_id
                && existing.request.inherit_context == invocation.inherit_context
                && existing.request.staged_skill == staged_skill;
            if !request_matches {
                return Err(EngineError::ToolFailed(
                    "delegate redelivery does not match its durable reservation".into(),
                ));
            }
            let registry_owns_invocation = self
                .inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?
                .get(&existing.reservation.child_session_id)
                .is_some_and(|record| record.invocation_id == invocation_id);
            if existing.run_attached || registry_owns_invocation {
                if existing.request.background && registry_owns_invocation {
                    let producer_missing = self
                        .inner
                        .delegation
                        .by_session
                        .lock()
                        .map_err(|_| EngineError::ActorStopped)?
                        .get(&existing.reservation.child_session_id)
                        .is_some_and(|record| {
                            !record.notification_sent && record.producer_id.is_none()
                        });
                    if producer_missing {
                        let mut guard = self
                            .register_background_delegation_producer(
                                existing.reservation.parent_session_id,
                                invocation_id,
                            )
                            .await?;
                        let mut records = self
                            .inner
                            .delegation
                            .by_session
                            .lock()
                            .map_err(|_| EngineError::ActorStopped)?;
                        if let Some(record) = records
                            .get_mut(&existing.reservation.child_session_id)
                            .filter(|record| {
                                record.invocation_id == invocation_id
                                    && record.producer_id.is_none()
                            })
                        {
                            record.producer_id = Some(guard.producer_id);
                            guard.handoff();
                        }
                    }
                }
                return Ok(DelegateHandle {
                    invocation_id,
                    child_session_id: existing.reservation.child_session_id,
                    child_run_id: existing.child_run_id,
                });
            }
        }
        let parent = self.inner.store.get(invocation.parent_session_id)?;
        if parent
            .runs
            .get(&invocation.parent_run_id)
            .is_some_and(|run| run.status == SessionStatus::Interrupted)
            && self.delegation_event_get(invocation_id).await?.is_none()
        {
            return Err(EngineError::ToolFailed(
                "delegate parent run is interrupted; use recovery".into(),
            ));
        }
        let active_parent = self
            .inner
            .sessions
            .active
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .get(&invocation.parent_run_id)
            .cloned()
            .ok_or_else(|| EngineError::ToolFailed("delegate parent run is not active".into()))?;
        let parent_policy = Arc::clone(&active_parent.policy);
        let parent_delegation = parent_policy
            .agent
            .delegation
            .as_ref()
            .ok_or_else(|| EngineError::ToolFailed("delegation is disabled".into()))?;
        if !parent_delegation.targets.contains(&invocation.agent_type)
            || session_depth(&parent.meta.origin) >= parent_delegation.effective_depth_ceiling
        {
            return Err(EngineError::ToolFailed(
                "delegate target or depth is not allowed".into(),
            ));
        }
        let resume_child = resume_session_id
            .map(|resume_session_id| {
                self.validate_resume_target(
                    invocation.parent_session_id,
                    resume_session_id,
                    &invocation.agent_type,
                    parent_policy.preset.as_deref(),
                )
            })
            .transpose()?;
        let child_agent = resolve_agent(&parent_policy.registry, &invocation.agent_type)?;
        if !child_agent.document.frontmatter.enabled
            || !matches!(
                child_agent.document.frontmatter.mode,
                cookie_agent_config::AgentMode::Subagent | cookie_agent_config::AgentMode::All
            )
        {
            return Err(EngineError::DisabledAgent(invocation.agent_type));
        }
        let fallback_index = active_parent.fallback_index.load(Ordering::Acquire) as usize;
        let inherited = parent_policy.active_suffix(fallback_index);
        let child_selection = child_agent
            .resolved_fallback
            .iter()
            .filter_map(|fallback| match fallback {
                crate::runtime_snapshot::ResolvedAgentFallback::Selection { selection, .. } => {
                    Some(selection)
                }
                crate::runtime_snapshot::ResolvedAgentFallback::ParentModel { .. } => None,
            })
            .find(|selection| {
                parent_policy
                    .runtime
                    .models
                    .model(&selection.model)
                    .is_some_and(|model| {
                        model.model.status
                            == cookie_agent_models::compiler::CompiledModelStatus::Available
                    })
            })
            .cloned()
            .or_else(|| {
                if child_agent.resolved_fallback.is_empty() {
                    inherited.first().map(|binding| binding.selection.clone())
                } else {
                    None
                }
            })
            .ok_or_else(|| EngineError::ToolFailed("delegate child has no model suffix".into()))?;
        let child_policy = freeze_delegated_child_policy(
            child_agent,
            &parent_policy,
            &child_selection,
            inherited,
            parent_delegation.effective_depth_ceiling,
            policy::ResultLimits {
                tool_output_max_lines: self.inner.config.runtime.tool_output.max_lines,
                tool_output_max_bytes: self.inner.config.runtime.tool_output.max_bytes,
            },
        )?;
        let seeded_context = if invocation.inherit_context {
            self.delegated_context_seed(invocation.parent_session_id)
                .await?
        } else {
            Vec::new()
        };
        let proposed_title = super::titles::delegated_title(
            &invocation.description,
            self.inner.config.runtime.session_title.max_chars,
        )?;
        let title = resume_child
            .as_ref()
            .and_then(|child| child.meta.title.clone())
            .unwrap_or(proposed_title);
        let request = DelegateRequestPayload {
            description: invocation.description,
            prompt: invocation.prompt,
            title,
            resume_session_id,
            inherit_context: invocation.inherit_context,
            seeded_context,
            background: invocation.background,
            staged_skill,
        };
        let cache_strategies = policy::wire_cache_strategies(&child_policy.cache_strategies)?;
        let fingerprint = delegation_events::delegation_request_fingerprint(
            &child_policy.agent,
            &child_policy.selected_suffix,
            &cache_strategies,
            &request,
        )
        .map_err(|()| EngineError::RuntimeCompileFailed)?;

        let root_session_id = match parent.meta.origin {
            SessionOrigin::Delegated {
                root_session_id, ..
            } => root_session_id,
            _ => parent.meta.session_id,
        };
        let resumed_running = if resume_child
            .as_ref()
            .is_some_and(|child| child.status == SessionStatus::Running)
        {
            let record = self
                .inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?
                .get(
                    &resume_child
                        .as_ref()
                        .expect("checked resume child")
                        .meta
                        .session_id,
                )
                .copied()
                .ok_or_else(|| {
                    EngineError::ToolFailed(
                        "running resume session has no active delegation record".into(),
                    )
                })?;
            let run_id = record.child_run_id.ok_or_else(|| {
                EngineError::ToolFailed("running resume session has no active run".into())
            })?;
            Some((run_id, record))
        } else {
            None
        };
        let fresh_counts_slot =
            invocation.background && !matches!(parent.meta.origin, SessionOrigin::Delegated { .. });
        let counts_slot = resumed_running
            .map(|(_, record)| record.counts_slot)
            .unwrap_or(fresh_counts_slot);
        let queued = resumed_running.is_none()
            && fresh_counts_slot
            && self.background_slot_unavailable(root_session_id)?;
        if queued && self.background_queue_full(root_session_id)? {
            return Err(EngineError::ToolFailed(format!(
                "delegate admission denied: background queue is full for concurrency limit {}",
                self.inner
                    .config
                    .runtime
                    .delegation
                    .max_concurrency
                    .unwrap_or(4)
            )));
        }
        let mut producer_guard = if invocation.background {
            Some(
                self.register_background_delegation_producer(
                    invocation.parent_session_id,
                    invocation_id,
                )
                .await?,
            )
        } else {
            None
        };
        let child = match self
            .create_child(
                invocation.parent_session_id,
                invocation.parent_run_id,
                invocation.parent_tool_call_id,
                &invocation.agent_type,
                child_policy,
                cache_strategies,
                fingerprint,
                request,
                Some((invocation_id, generation)),
            )
            .await
        {
            Ok(child) => child,
            Err(error) => {
                if is_delegation_event_append_failure(&error) {
                    let result = delegate_failure_result(
                        None,
                        None,
                        "delegate reservation event append failed",
                    );
                    self.resolve_delegate_failure_if_pending(
                        invocation.parent_session_id,
                        invocation.parent_run_id,
                        invocation.parent_tool_call_id,
                        result,
                    )
                    .await?;
                }
                return Err(error);
            }
        };
        self.transfer_pending_skill_fork(invocation.parent_tool_call_id, child.session_id);
        self.publish_admission_child(invocation_id, generation, child.session_id)?;
        let entry = self
            .delegation_event_get(invocation_id)
            .await?
            .ok_or_else(|| EngineError::ToolFailed("delegate reservation disappeared".into()))?;
        if let Some(guard) = producer_guard.as_mut() {
            let mut records = self
                .inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?;
            if let Some(record) = records.get_mut(&child.session_id).filter(|record| {
                record.invocation_id == invocation_id && record.producer_id.is_none()
            }) {
                record.producer_id = Some(guard.producer_id);
                guard.handoff();
            }
        }
        #[cfg(test)]
        let skill_fork_hook = if entry.request.staged_skill.is_some() {
            self.inner
                .test_hooks
                .skill_fork_reservation_hook
                .lock()
                .expect("skill fork reservation hook lock poisoned")
                .take()
        } else {
            None
        };
        #[cfg(test)]
        if let Some(hook) = skill_fork_hook {
            if let Some(reached) = hook
                .reached
                .lock()
                .expect("skill fork reservation reached lock poisoned")
                .take()
            {
                let _ = reached.send(());
            }
            hook.release.notified().await;
        }
        if resume_child.is_some() {
            self.inner
                .delegation
                .queue
                .lock()
                .map_err(|_| EngineError::ActorStopped)?
                .retain(|session_id| *session_id != child.session_id);
        }
        if let Some((child_run_id, previous_record)) = resumed_running {
            if entry.child_run_id == Some(child_run_id)
                && (entry.run_attached || previous_record.invocation_id == invocation_id)
            {
                return Ok(DelegateHandle {
                    invocation_id,
                    child_session_id: child.session_id,
                    child_run_id: Some(child_run_id),
                });
            }
            let delegation_events = self.inner.delegation_events.clone();
            self.spawn_admission_blocking(move || {
                delegation_events.mark_run_attached(invocation_id, child_run_id)
            })
            .await?;
            #[cfg(test)]
            self.wait_for_resume_test_hook(&self.inner.test_hooks.resume_attachment_hook)
                .await;
            if let Err(error) = self.publish_resume_admission_target(
                invocation_id,
                generation,
                child.session_id,
                child_run_id,
                previous_record,
            ) {
                let delegation_events = self.inner.delegation_events.clone();
                self.spawn_admission_blocking(move || {
                    delegation_events.mark_finished(invocation_id, SessionStatus::Cancelled)
                })
                .await?;
                return Err(error);
            }
            let handle = DelegateHandle {
                invocation_id,
                child_session_id: child.session_id,
                child_run_id: Some(child_run_id),
            };
            let monitor_release = if invocation.background || counts_slot {
                match self.spawn_background_monitor_gated(handle) {
                    Ok(release) => Some(release),
                    Err(error) => {
                        let delegation_events = self.inner.delegation_events.clone();
                        self.spawn_admission_blocking(move || {
                            delegation_events.mark_finished(invocation_id, SessionStatus::Cancelled)
                        })
                        .await?;
                        return Err(error);
                    }
                }
            } else {
                None
            };
            #[cfg(test)]
            if let Some(hook) = {
                self.inner
                    .test_hooks
                    .resume_admission_hook
                    .lock()
                    .expect("resume admission hook lock poisoned")
                    .clone()
            } {
                if let Some(reached) = hook
                    .reached
                    .lock()
                    .expect("resume admission reached lock poisoned")
                    .take()
                {
                    let _ = reached.send(());
                }
                hook.release.notified().await;
            }
            let admission = match self
                .request(child.session_id, |reply| {
                    SessionCommand::AdmitDelegatedResume {
                        run: child_run_id,
                        input: entry.request.prompt.clone(),
                        reply,
                    }
                })
                .await
            {
                Ok(admission) => admission,
                Err(error) => {
                    let delegation_events = self.inner.delegation_events.clone();
                    self.spawn_admission_blocking(move || {
                        delegation_events.mark_finished(invocation_id, SessionStatus::Cancelled)
                    })
                    .await?;
                    return Err(error);
                }
            };
            if !admission.accepted {
                let delegation_events = self.inner.delegation_events.clone();
                self.spawn_admission_blocking(move || {
                    delegation_events.mark_finished(invocation_id, SessionStatus::Cancelled)
                })
                .await?;
                return Err(EngineError::ToolFailed(
                    "resume session stopped before its prompt was admitted".into(),
                ));
            }
            let Some(admission_seq) = admission.admission_seq else {
                let delegation_events = self.inner.delegation_events.clone();
                self.spawn_admission_blocking(move || {
                    delegation_events.mark_finished(invocation_id, SessionStatus::Cancelled)
                })
                .await?;
                let _ = self.cancel_run_durably(
                    child_run_id,
                    Some("resume admission sequence was not published".into()),
                );
                return Err(EngineError::ToolFailed(
                    "resume admission sequence is missing".into(),
                ));
            };
            if let Err(error) =
                self.publish_resume_admission_seq(invocation_id, generation, admission_seq)
            {
                drop(monitor_release);
                self.rollback_running_resume(
                    invocation_id,
                    child.session_id,
                    child_run_id,
                    admission_seq,
                    previous_record,
                    SessionStatus::Cancelled,
                )
                .await?;
                return Err(error);
            }
            #[cfg(test)]
            self.wait_for_resume_test_hook(&self.inner.test_hooks.resume_rollback_hook)
                .await;
            if !self.admission_generation_live(invocation_id, generation) {
                drop(monitor_release);
                self.rollback_running_resume(
                    invocation_id,
                    child.session_id,
                    child_run_id,
                    admission_seq,
                    previous_record,
                    SessionStatus::Cancelled,
                )
                .await?;
                return Err(EngineError::ToolFailed(
                    "delegate admission was abandoned".into(),
                ));
            }
            let registry_failed = {
                match self.inner.delegation.by_session.lock() {
                    Ok(mut records) => {
                        let record = DelegationRecord {
                            parent_session_id: invocation.parent_session_id,
                            parent_run_id: invocation.parent_run_id,
                            parent_tool_call_id: invocation.parent_tool_call_id,
                            invocation_id,
                            root_session_id,
                            child_run_id: Some(child_run_id),
                            state: DelegationState::Running,
                            background: invocation.background,
                            counts_slot,
                            notification_sent: false,
                            producer_id: producer_guard.as_ref().map(|guard| guard.producer_id),
                            monitor_started: monitor_release.is_some(),
                        };
                        debug_assert!(record.background || record.producer_id.is_none());
                        records.insert(child.session_id, record);
                        false
                    }
                    Err(_) => true,
                }
            };
            if registry_failed {
                drop(monitor_release);
                self.rollback_running_resume(
                    invocation_id,
                    child.session_id,
                    child_run_id,
                    admission_seq,
                    previous_record,
                    SessionStatus::Cancelled,
                )
                .await?;
                return Err(EngineError::ActorStopped);
            }
            if let Some(release) = monitor_release
                && release.send(()).is_err()
            {
                self.rollback_running_resume(
                    invocation_id,
                    child.session_id,
                    child_run_id,
                    admission_seq,
                    previous_record,
                    SessionStatus::Cancelled,
                )
                .await?;
                return Err(EngineError::ActorStopped);
            }
            if !self.admission_generation_live(invocation_id, generation) {
                self.rollback_running_resume(
                    invocation_id,
                    child.session_id,
                    child_run_id,
                    admission_seq,
                    previous_record,
                    SessionStatus::Cancelled,
                )
                .await?;
                return Err(EngineError::ToolFailed(
                    "delegate admission was abandoned".into(),
                ));
            }
            if let Some(guard) = producer_guard.as_mut() {
                guard.handoff();
            }
            return Ok(handle);
        }
        let record = DelegationRecord {
            parent_session_id: invocation.parent_session_id,
            parent_run_id: invocation.parent_run_id,
            parent_tool_call_id: invocation.parent_tool_call_id,
            invocation_id,
            root_session_id,
            child_run_id: None,
            state: if queued {
                DelegationState::Queued
            } else {
                DelegationState::Starting
            },
            background: invocation.background,
            counts_slot,
            notification_sent: false,
            producer_id: producer_guard.as_ref().map(|guard| guard.producer_id),
            monitor_started: false,
        };
        debug_assert!(record.background || record.producer_id.is_none());
        self.inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .insert(child.session_id, record);
        if let Some(guard) = producer_guard.as_mut() {
            guard.handoff();
        }
        if queued {
            let position = {
                let mut queue = self
                    .inner
                    .delegation
                    .queue
                    .lock()
                    .map_err(|_| EngineError::ActorStopped)?;
                queue.push_back(child.session_id);
                u32::try_from(queue.len()).ok()
            };
            if let Err(error) = self
                .append(
                    invocation.parent_session_id,
                    Some(invocation.parent_run_id),
                    super::event_origin("engine:delegation"),
                    Event::DelegateQueued {
                        session_id: child.session_id,
                        position,
                    },
                )
                .await
            {
                self.inner
                    .delegation
                    .queue
                    .lock()
                    .map_err(|_| EngineError::ActorStopped)?
                    .retain(|session_id| *session_id != child.session_id);
                let delegation_events = self.inner.delegation_events.clone();
                let _ = self
                    .spawn_admission_blocking(move || {
                        delegation_events.mark_finished(invocation_id, SessionStatus::Failed)
                    })
                    .await;
                let terminalized = self
                    .terminalize_child_without_run(
                        child.session_id,
                        SessionStatus::Failed,
                        "delegate queue event append failed",
                    )
                    .await;
                if let Some(record) = self
                    .inner
                    .delegation
                    .by_session
                    .lock()
                    .map_err(|_| EngineError::ActorStopped)?
                    .get_mut(&child.session_id)
                {
                    record.state = DelegationState::Finished(SessionStatus::Failed);
                }
                drop(admission_guard);
                if terminalized.is_ok() {
                    self.finish_background_or_retry(child.session_id, invocation_id)
                        .await;
                } else {
                    let _ = self.start_queued_delegation(root_session_id).await;
                }
                return Err(error);
            }
            return Ok(DelegateHandle {
                invocation_id,
                child_session_id: child.session_id,
                child_run_id: None,
            });
        }
        let child_run_id = match self
            .ensure_delegate_run(&entry, Some((invocation_id, generation)))
            .await
        {
            Ok(run_id) => run_id,
            Err(error) => {
                if is_delegation_event_append_failure(&error) {
                    let result = delegate_failure_result(
                        Some(entry.reservation.child_session_id),
                        child.short_id.as_deref(),
                        "delegate run event confirmation failed",
                    );
                    let _ = self
                        .resolve_delegate_failure_if_pending(
                            entry.reservation.parent_session_id,
                            entry.reservation.parent_run_id,
                            entry.reservation.parent_tool_call_id,
                            result,
                        )
                        .await;
                }
                let delegation_events = self.inner.delegation_events.clone();
                let _ = self
                    .spawn_admission_blocking(move || {
                        delegation_events.mark_finished(invocation_id, SessionStatus::Failed)
                    })
                    .await;
                let terminalized = self
                    .terminalize_child_without_run(
                        child.session_id,
                        SessionStatus::Failed,
                        "delegate child startup failed",
                    )
                    .await;
                if let Some(record) = self
                    .inner
                    .delegation
                    .by_session
                    .lock()
                    .map_err(|_| EngineError::ActorStopped)?
                    .get_mut(&child.session_id)
                {
                    record.state = DelegationState::Finished(SessionStatus::Failed);
                }
                drop(admission_guard);
                if invocation.background {
                    if terminalized.is_ok() {
                        self.finish_background_or_retry(child.session_id, invocation_id)
                            .await;
                    } else {
                        let _ = self.start_queued_delegation(root_session_id).await;
                    }
                }
                return Err(error);
            }
        };
        if let Some(record) = self
            .inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .get_mut(&child.session_id)
        {
            record.child_run_id = Some(child_run_id);
            record.state = DelegationState::Running;
            record.monitor_started = invocation.background;
        }
        let handle = DelegateHandle {
            invocation_id,
            child_session_id: child.session_id,
            child_run_id: Some(child_run_id),
        };
        if invocation.background
            && let Err(error) = self.spawn_background_monitor(handle, None)
        {
            self.release_recovered_monitor(child.session_id, invocation_id);
            let _ = self.cancel_run_durably(
                child_run_id,
                Some("background delegate monitor failed to start".into()),
            );
            drop(admission_guard);
            self.finish_background_or_retry(child.session_id, invocation_id)
                .await;
            return Err(error);
        }
        Ok(handle)
    }

    /// Waits for a child terminal state and returns the bounded model-visible
    /// delegate result. Cancellation retains the child session identity.
    pub fn await_delegate(&self, handle: DelegateHandle) -> DelegateAwait {
        let engine = self.clone();
        DelegateAwait {
            future: Box::pin(async move { engine.await_delegate_inner(handle).await }),
            engine: self.clone(),
            runtime: self.inner.runtime.clone(),
            handle,
            completed: false,
        }
    }

    pub(super) async fn await_delegate_inner(
        &self,
        handle: DelegateHandle,
    ) -> Result<ToolResult, EngineError> {
        loop {
            let child = match self.inner.store.get(handle.child_session_id) {
                Ok(child) => child,
                Err(_) => {
                    return Ok(delegate_failure_result(
                        Some(handle.child_session_id),
                        None,
                        "child session is missing",
                    ));
                }
            };
            let status = handle
                .child_run_id
                .and_then(|run_id| child.runs.get(&run_id).map(|run| run.status))
                .unwrap_or(child.status);
            match status {
                SessionStatus::Running | SessionStatus::Idle => {
                    let active = {
                        self.inner
                            .sessions
                            .active
                            .lock()
                            .unwrap_or_else(|poisoned| poisoned.into_inner())
                            .get(&handle.child_run_id.expect("running delegate has a run"))
                            .cloned()
                    };
                    if let Some(active) = active {
                        // Event-driven terminal-state wakeups are post-MVP; this
                        // bounded cancellation-aware wait keeps the MVP responsive.
                        tokio::select! {
                            () = active.cancellation.cancelled() => {},
                            () = tokio::time::sleep(std::time::Duration::from_millis(25)) => {},
                        }
                    } else {
                        tokio::time::sleep(std::time::Duration::from_millis(25)).await;
                    }
                }
                SessionStatus::Completed
                | SessionStatus::Cancelled
                | SessionStatus::Failed
                | SessionStatus::Interrupted => {
                    let delegation_events = self.inner.delegation_events.clone();
                    self.spawn_admission_blocking(move || {
                        delegation_events.mark_finished(handle.invocation_id, status)
                    })
                    .await?;
                    let result = terminal_delegate_result(&child, handle.child_run_id, status);
                    let mut records = self
                        .inner
                        .delegation
                        .by_session
                        .lock()
                        .map_err(|_| EngineError::ActorStopped)?;
                    if let Some(record) = records
                        .get_mut(&handle.child_session_id)
                        .filter(|record| record.invocation_id == handle.invocation_id)
                    {
                        record.state = DelegationState::Finished(status);
                    }
                    drop(records);
                    self.clear_delegate_admissions(handle.invocation_id);
                    return Ok(result);
                }
            }
        }
    }

    /// Cancels the child run and returns the structured delegate cancellation
    /// result used by the parent tool invocation.
    pub async fn cancel_delegate(&self, handle: DelegateHandle) -> Result<ToolResult, EngineError> {
        let child = match self.inner.store.get(handle.child_session_id) {
            Ok(child) => child,
            Err(_) => {
                return Ok(delegate_failure_result(
                    Some(handle.child_session_id),
                    None,
                    "child session is missing",
                ));
            }
        };
        if !matches!(child.status, SessionStatus::Running | SessionStatus::Idle) {
            return self.await_delegate(handle).await;
        }
        let Some(child_run_id) = handle.child_run_id else {
            return Ok(cancelled_delegate_result(
                handle.child_session_id,
                child.meta.short_id.as_deref(),
                None,
            ));
        };
        let _ = self.cancel_run(child_run_id).await;
        self.await_delegate(handle).await
    }

    pub(super) async fn delegation_event_get(
        &self,
        invocation_id: InvocationId,
    ) -> Result<Option<delegation_events::DelegationEntry>, EngineError> {
        let delegation_events = self.inner.delegation_events.clone();
        self.spawn_admission_blocking(move || {
            Ok::<_, EngineError>(delegation_events.get(invocation_id))
        })
        .await
    }

    pub(super) fn clear_delegate_admissions(&self, invocation_id: InvocationId) {
        self.inner
            .delegation
            .inflight
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .remove(&invocation_id);
    }

    pub(super) async fn resolve_delegate_failure_if_pending(
        &self,
        session_id: SessionId,
        run_id: RunId,
        tool_call_id: ToolCallId,
        result: ToolResult,
    ) -> Result<bool, EngineError> {
        self.request(session_id, |reply| {
            SessionCommand::ResolveDelegateFailureIfPending {
                run: run_id,
                tool_call_id,
                result,
                reply,
            }
        })
        .await
    }

    pub(super) fn resolve_delegate_failure_if_pending_direct(
        &self,
        session_id: SessionId,
        run_id: RunId,
        tool_call_id: ToolCallId,
        result: ToolResult,
    ) -> Result<bool, EngineError> {
        let pending = self
            .inner
            .store
            .get(session_id)?
            .runs
            .get(&run_id)
            .is_some_and(|run| {
                run.pending_calls
                    .get(&tool_call_id)
                    .is_some_and(|tool| tool == "delegate_subagent")
            });
        if !pending {
            return Ok(false);
        }
        self.append_direct(
            session_id,
            Some(run_id),
            super::event_origin("engine:tool-result"),
            Event::ToolCallTerminated {
                termination: ToolCallTermination {
                    tool_call_id,
                    owner: self.tool_call_owner(session_id, run_id, tool_call_id)?,
                    outcome: ToolTerminationOutcome::Failed,
                    result: Some(result),
                    error: Some(SafeToolError {
                        code: safe_code("delegate_failed"),
                        message: safe_error("delegated run failed"),
                    }),
                },
            },
        )?;
        Ok(true)
    }

    pub(super) fn resolve_abandoned_delegate_failure_if_pending_direct(
        &self,
        invocation_id: InvocationId,
        generation: u64,
        session_id: SessionId,
        run_id: RunId,
        tool_call_id: ToolCallId,
        result: ToolResult,
    ) -> Result<bool, EngineError> {
        let mut admissions = self
            .inner
            .delegation
            .inflight
            .lock()
            .map_err(|_| EngineError::ActorStopped)?;
        let still_abandoned = admissions.get(&invocation_id).is_some_and(|entries| {
            entries
                .get(&generation)
                .is_some_and(|entry| entry.cancelled)
                && entries.values().all(|entry| entry.cancelled)
        });
        if !still_abandoned {
            return Ok(false);
        }
        let resolved = self.resolve_delegate_failure_if_pending_direct(
            session_id,
            run_id,
            tool_call_id,
            result,
        )?;
        admissions.remove(&invocation_id);
        Ok(resolved)
    }

    pub(super) async fn ensure_delegate_run(
        &self,
        entry: &delegation_events::DelegationEntry,
        admission: Option<(InvocationId, u64)>,
    ) -> Result<RunId, EngineError> {
        if entry.terminal_status.is_some() {
            return Err(EngineError::ToolFailed(
                "delegate invocation was cancelled before startup".into(),
            ));
        }
        #[cfg(test)]
        if self
            .inner
            .test_hooks
            .delegate_start_failures
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| {
                (remaining > 0).then(|| remaining - 1)
            })
            .is_ok()
        {
            self.inner
                .test_hooks
                .delegate_start_failure_observed
                .notify_one();
            return Err(EngineError::ToolFailed(
                "injected delegate startup failure".into(),
            ));
        }
        if admission.is_some_and(|(invocation_id, generation)| {
            !self.admission_generation_live(invocation_id, generation)
        }) {
            return Err(EngineError::ToolFailed(
                "delegate cancelled before child start".into(),
            ));
        }
        let child = self.inner.store.get(entry.reservation.child_session_id)?;
        if let Some(staged_skill) = &entry.request.staged_skill {
            self.stage_child_skill_from_event(entry.reservation.child_session_id, staged_skill);
        }
        if let Some((invocation_id, generation)) = admission {
            self.mark_admission_starting(invocation_id, generation)?;
        }
        if let Some(run_id) = entry.child_run_id
            && child.runs.contains_key(&run_id)
        {
            if let Some((invocation_id, generation)) = admission {
                self.publish_admission_run(
                    invocation_id,
                    generation,
                    entry.reservation.child_session_id,
                    run_id,
                )?;
                if !self.admission_generation_live(invocation_id, generation) {
                    self.sweep_abandoned_admission(invocation_id, generation)
                        .await?;
                    return Err(EngineError::ToolFailed(
                        "delegate cancelled before child start".into(),
                    ));
                }
            }
            return Ok(run_id);
        }
        let client_run_id = delegate_client_run_id(entry.reservation.invocation_id);
        let existing_run = child
            .runs
            .values()
            .find(|run| run.client_run_id == client_run_id)
            .map(|run| run.id);
        let run_id = match existing_run {
            Some(run_id) => run_id,
            None => match self
                .request(entry.reservation.child_session_id, |reply| {
                    SessionCommand::Start {
                        params: RunStartParams {
                            reset_fallback: false,
                            session_id: entry.reservation.child_session_id,
                            client_run_id: client_run_id.clone(),
                            selection: child.meta.creation_selection.clone(),
                            input: render_delegate_input(&entry.request),
                        },
                        origin: super::event_origin("engine:delegation"),
                        admission,
                        reply,
                    }
                })
                .await
            {
                Ok(started) => started.run_id,
                Err(error) => {
                    if let Some((invocation_id, generation)) = admission {
                        self.clear_admission_starting(invocation_id, generation);
                        if !self.admission_generation_live(invocation_id, generation) {
                            self.sweep_abandoned_admission(invocation_id, generation)
                                .await?;
                        }
                    }
                    return Err(error);
                }
            },
        };
        let cancelled = if let Some((invocation_id, generation)) = admission {
            // The Start actor published a newly-created run before delivering
            // its reply. Existing local runs reach the same state here.
            self.publish_admission_run(
                invocation_id,
                generation,
                entry.reservation.child_session_id,
                run_id,
            )?;
            !self.admission_generation_live(invocation_id, generation)
        } else {
            false
        };
        // The actor-owned confirmation must run even after its observer drops:
        // the start reply may already have created the child run.
        #[cfg(test)]
        let confirmation_hook = self
            .inner
            .test_hooks
            .admission_confirmation_hook
            .lock()
            .expect("admission confirmation hook lock poisoned")
            .clone();
        #[cfg(test)]
        if let Some(hook) = confirmation_hook {
            let _ = hook.reached.send(());
            hook.release.wait().await;
        }
        let delegation_events = self.inner.delegation_events.clone();
        let invocation_id = entry.reservation.invocation_id;
        let confirmation = self
            .spawn_admission_blocking(move || {
                delegation_events.mark_run_started(invocation_id, run_id)
            })
            .await;
        if let Err(error) = confirmation {
            // The child already has an active run, so terminally cancel it before
            // the caller resolves the parent through its actor.
            let _ = self.cancel_run_durably(
                run_id,
                Some("delegate run event confirmation failed".into()),
            );
            return Err(error);
        }
        if cancelled {
            if let Some((invocation_id, generation)) = admission {
                self.sweep_abandoned_admission(invocation_id, generation)
                    .await?;
            }
            return Err(EngineError::ToolFailed(
                "delegate cancelled during child start".into(),
            ));
        }
        Ok(run_id)
    }

    fn background_slot_unavailable(&self, root_session_id: SessionId) -> Result<bool, EngineError> {
        let Some(limit) = self.inner.config.runtime.delegation.max_concurrency else {
            return Ok(false);
        };
        let running = self
            .inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .values()
            .filter(|record| {
                record.root_session_id == root_session_id
                    && record.counts_slot
                    && record.state == DelegationState::Running
            })
            .count() as u32;
        Ok(running >= limit)
    }

    pub(super) fn interrupt_delegation_accounting(
        &self,
        parent_session_id: SessionId,
        interrupted_runs: &[RunId],
    ) -> Result<(), EngineError> {
        let mut records = self
            .inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?;
        for record in records.values_mut() {
            if record.parent_session_id == parent_session_id
                && interrupted_runs.contains(&record.parent_run_id)
                && record.state == DelegationState::Running
            {
                record.state = DelegationState::Finished(SessionStatus::Interrupted);
            }
        }
        Ok(())
    }

    #[cfg(test)]
    pub(crate) fn running_background_delegations_for_test(
        &self,
        root_session_id: SessionId,
    ) -> usize {
        self.inner
            .delegation
            .by_session
            .lock()
            .expect("delegation registry")
            .values()
            .filter(|record| {
                record.root_session_id == root_session_id
                    && record.counts_slot
                    && record.state == DelegationState::Running
            })
            .count()
    }

    fn background_queue_full(&self, root_session_id: SessionId) -> Result<bool, EngineError> {
        let Some(limit) = self.inner.config.runtime.delegation.max_concurrency else {
            return Ok(false);
        };
        let queued = self
            .inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .values()
            .filter(|record| {
                record.root_session_id == root_session_id && record.state == DelegationState::Queued
            })
            .count() as u32;
        Ok(background_queue_limit_reached(limit, queued))
    }

    /// Spawns the completion monitor for one background delegate.
    ///
    /// `recovery` carries the settling ticket when this monitor was claimed by
    /// startup or adoption recovery; the task holds it until the record is
    /// resolved so `resume` can wait for exactly that work.
    fn spawn_background_monitor(
        &self,
        handle: DelegateHandle,
        recovery: Option<RecoveryMonitorTicket>,
    ) -> Result<(), EngineError> {
        let runtime = self
            .inner
            .runtime
            .clone()
            .or_else(|| tokio::runtime::Handle::try_current().ok())
            .ok_or(EngineError::ActorStopped)?;
        let engine = self.clone();
        let monitor = engine.clone();
        if !self.spawn_admission_task(&runtime, async move {
            let _recovery = recovery;
            let _ = monitor.await_delegate_inner(handle).await;
            loop {
                match monitor
                    .finish_background_delegate(handle.child_session_id, handle.invocation_id)
                    .await
                {
                    Ok(()) => break,
                    Err(error) => {
                        if monitor
                            .inner
                            .delegation
                            .admission_tasks_closing
                            .load(Ordering::Acquire)
                        {
                            break;
                        }
                        eprintln!("background delegate completion retrying: {error}");
                        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
                    }
                }
            }
        }) {
            return Err(EngineError::ActorStopped);
        }
        Ok(())
    }

    async fn register_background_delegation_producer(
        &self,
        parent_session_id: SessionId,
        invocation_id: InvocationId,
    ) -> Result<BackgroundProducerGuard, EngineError> {
        let producer_id = self
            .register_producer(
                parent_session_id,
                delegation_producer_authority(invocation_id),
            )
            .await?;
        Ok(BackgroundProducerGuard {
            engine: self.clone(),
            runtime: self.inner.runtime.clone(),
            parent_session_id,
            invocation_id,
            producer_id,
            active: true,
        })
    }

    fn spawn_background_monitor_gated(
        &self,
        handle: DelegateHandle,
    ) -> Result<tokio::sync::oneshot::Sender<()>, EngineError> {
        #[cfg(test)]
        if self
            .inner
            .test_hooks
            .resume_monitor_failures
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| {
                (remaining > 0).then(|| remaining - 1)
            })
            .is_ok()
        {
            return Err(EngineError::ActorStopped);
        }
        let runtime = self
            .inner
            .runtime
            .clone()
            .or_else(|| tokio::runtime::Handle::try_current().ok())
            .ok_or(EngineError::ActorStopped)?;
        let (release, admitted) = tokio::sync::oneshot::channel();
        let engine = self.clone();
        let monitor = engine.clone();
        if !self.spawn_admission_task(&runtime, async move {
            if admitted.await.is_err() {
                return;
            }
            let _ = monitor.await_delegate_inner(handle).await;
            loop {
                match monitor
                    .finish_background_delegate(handle.child_session_id, handle.invocation_id)
                    .await
                {
                    Ok(()) => break,
                    Err(error) => {
                        if monitor
                            .inner
                            .delegation
                            .admission_tasks_closing
                            .load(Ordering::Acquire)
                        {
                            break;
                        }
                        eprintln!("background delegate completion retrying: {error}");
                        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
                    }
                }
            }
        }) {
            return Err(EngineError::ActorStopped);
        }
        Ok(release)
    }

    pub(super) async fn rollback_running_resume(
        &self,
        invocation_id: InvocationId,
        child_session_id: SessionId,
        child_run_id: RunId,
        admission_seq: u64,
        previous_record: DelegationRecord,
        status: SessionStatus,
    ) -> Result<(), EngineError> {
        let recalled = self
            .request(child_session_id, |reply| {
                SessionCommand::RecallDelegatedResume {
                    run: child_run_id,
                    admission_seq,
                    reply,
                }
            })
            .await
            .unwrap_or(false);
        let delegation_events = self.inner.delegation_events.clone();
        self.spawn_admission_blocking(move || {
            delegation_events.mark_finished(invocation_id, status)
        })
        .await?;
        if let Ok(mut records) = self.inner.delegation.by_session.lock()
            && records
                .get(&child_session_id)
                .is_some_and(|record| record.invocation_id == invocation_id)
        {
            records.insert(child_session_id, previous_record);
        }
        if !recalled {
            let child = self.inner.store.get(child_session_id)?;
            if child
                .runs
                .get(&child_run_id)
                .is_some_and(|run| run.status == SessionStatus::Running)
            {
                let _ = self.cancel_run_durably(
                    child_run_id,
                    Some("abandoned resume prompt could not be recalled".into()),
                );
            }
        }
        Ok(())
    }

    async fn finish_background_or_retry(
        &self,
        child_session_id: SessionId,
        invocation_id: InvocationId,
    ) {
        if self
            .finish_background_delegate(child_session_id, invocation_id)
            .await
            .is_err()
        {
            let _ = self.spawn_background_finish_retry(child_session_id, invocation_id);
        }
    }

    fn spawn_background_finish_retry(
        &self,
        child_session_id: SessionId,
        invocation_id: InvocationId,
    ) -> Result<(), EngineError> {
        let runtime = self
            .inner
            .runtime
            .clone()
            .or_else(|| tokio::runtime::Handle::try_current().ok())
            .ok_or(EngineError::ActorStopped)?;
        let engine = self.clone();
        let retry = engine.clone();
        if !self.spawn_admission_task(&runtime, async move {
            loop {
                match retry
                    .finish_background_delegate(child_session_id, invocation_id)
                    .await
                {
                    Ok(()) => break,
                    Err(error) => {
                        if retry
                            .inner
                            .delegation
                            .admission_tasks_closing
                            .load(Ordering::Acquire)
                        {
                            break;
                        }
                        eprintln!("background delegate completion retrying: {error}");
                        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
                    }
                }
            }
        }) {
            return Err(EngineError::ActorStopped);
        }
        Ok(())
    }

    async fn finish_background_delegate(
        &self,
        child_session_id: SessionId,
        invocation_id: InvocationId,
    ) -> Result<(), EngineError> {
        let child = self.inner.store.get(child_session_id)?;
        // A revert can remove the reservation while its child monitor finishes.
        let Some(entry) = self.delegation_event_get(invocation_id).await? else {
            return Ok(());
        };
        if entry.reservation.child_session_id != child_session_id {
            return Err(EngineError::ToolFailed(
                "delegate reservation child does not match completion monitor".into(),
            ));
        }
        let parent_session_id = entry.reservation.parent_session_id;
        let parent = self.inner.store.get(parent_session_id)?;
        let root_session_id = match parent.meta.origin {
            SessionOrigin::Delegated {
                root_session_id, ..
            } => root_session_id,
            _ => parent_session_id,
        };
        let exact_status = if let Some(status) = entry.terminal_status {
            status
        } else if let Some(child_run_id) = entry.child_run_id {
            child
                .runs
                .get(&child_run_id)
                .map(|run| run.status)
                .ok_or_else(|| EngineError::ToolFailed("delegate child run is missing".into()))?
        } else {
            child.status
        };
        if !matches!(
            exact_status,
            SessionStatus::Completed
                | SessionStatus::Failed
                | SessionStatus::Interrupted
                | SessionStatus::Cancelled
        ) {
            return Err(EngineError::ToolFailed(
                "delegate child run is not terminal".into(),
            ));
        }
        let (owns_registry, mut producer_id) = {
            let mut records = self
                .inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?;
            if let Some(record) = records
                .get_mut(&child_session_id)
                .filter(|record| record.invocation_id == invocation_id)
            {
                record.state = DelegationState::Finished(exact_status);
                (true, record.producer_id)
            } else {
                (false, None)
            }
        };
        let parent_events = parent.log.event_snapshot();
        let mut already_logged = false;
        let mut already_accepted = false;
        for event in parent_events.iter() {
            match &event.payload {
                Event::DelegateFinishedV2 {
                    invocation_id: logged_invocation,
                    session_id,
                    ..
                } if *logged_invocation == invocation_id && *session_id == child_session_id => {
                    already_logged = true;
                }
                Event::ProducerMessageAccepted {
                    producer_owner:
                        ProducerOwner::Delegation {
                            invocation_id: accepted_invocation,
                        },
                    ..
                } if *accepted_invocation == invocation_id => {
                    already_accepted = true;
                }
                _ => {}
            }
        }
        let background = entry.request.background;
        let teaser = if entry.terminal_status.is_some() && entry.child_run_id.is_none() {
            DelegateTeaser {
                session_id: child_session_id,
                short_id: child
                    .meta
                    .short_id
                    .clone()
                    .unwrap_or_else(|| child_session_id.to_string()),
                status: SessionStatus::Cancelled,
                preview: String::new(),
                total_lines: 0,
            }
        } else {
            delegate_teaser(&child, child_session_id, exact_status, entry.child_run_id)
        };
        // Historical V2-only completions have already been model-visible. New
        // completions durably accept the producer message before the V2 teaser.
        let mut transient_guard = None;
        if background && !already_accepted && !already_logged && producer_id.is_none() {
            let mut guard = self
                .register_background_delegation_producer(parent_session_id, invocation_id)
                .await?;
            let registered_id = guard.producer_id;
            producer_id = Some(registered_id);
            if owns_registry {
                let mut records = self
                    .inner
                    .delegation
                    .by_session
                    .lock()
                    .map_err(|_| EngineError::ActorStopped)?;
                if let Some(record) = records
                    .get_mut(&child_session_id)
                    .filter(|record| record.invocation_id == invocation_id)
                {
                    if record.producer_id.is_none() {
                        record.producer_id = Some(registered_id);
                        guard.handoff();
                    }
                    producer_id = record.producer_id;
                }
            }
            transient_guard = Some(guard);
        }
        if background && !already_logged {
            let send = self
                .commit_background_delegation_completion(
                    entry.reservation.clone(),
                    producer_id,
                    teaser.clone(),
                )
                .await;
            let committed = if let Err(EngineError::Producer(_)) = send {
                let failed_id = producer_id;
                let mut guard = self
                    .register_background_delegation_producer(parent_session_id, invocation_id)
                    .await?;
                let registered_id = guard.producer_id;
                producer_id = Some(registered_id);
                if owns_registry {
                    let mut records = self
                        .inner
                        .delegation
                        .by_session
                        .lock()
                        .map_err(|_| EngineError::ActorStopped)?;
                    if let Some(record) = records
                        .get_mut(&child_session_id)
                        .filter(|record| record.invocation_id == invocation_id)
                    {
                        if record.producer_id.is_none() || record.producer_id == failed_id {
                            record.producer_id = Some(registered_id);
                            guard.handoff();
                        }
                        producer_id = record.producer_id;
                    }
                }
                transient_guard = Some(guard);
                self.commit_background_delegation_completion(
                    entry.reservation.clone(),
                    producer_id,
                    teaser,
                )
                .await?
            } else {
                send?
            };
            if !committed {
                return Ok(());
            }
            already_accepted = true;
            already_logged = true;
        }
        if owns_registry
            && (already_accepted || already_logged)
            && let Some(record) = self
                .inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?
                .get_mut(&child_session_id)
        {
            record.notification_sent = true;
        }
        if let Some(id) = producer_id {
            match self
                .unregister_producer(
                    parent_session_id,
                    delegation_producer_authority(invocation_id),
                    id,
                )
                .await
            {
                Ok(()) | Err(EngineError::Producer(_)) => {
                    if let Some(guard) = transient_guard.as_mut() {
                        guard.handoff();
                    }
                    if owns_registry
                        && let Some(record) = self
                            .inner
                            .delegation
                            .by_session
                            .lock()
                            .map_err(|_| EngineError::ActorStopped)?
                            .get_mut(&child_session_id)
                    {
                        record.producer_id = None;
                    }
                }
                Err(error) => return Err(error),
            }
        }
        let drain_result = if owns_registry {
            self.start_queued_delegation(root_session_id).await
        } else {
            Ok(())
        };
        let completion = drain_result;
        if completion.is_ok()
            && let Err(error) = self.evict_idle_subagents().await
        {
            eprintln!("subagent session completion eviction failed: {error}");
        }
        completion
    }

    async fn commit_background_delegation_completion(
        &self,
        reservation: cookie_agent_protocol::DelegationReservation,
        producer_id: Option<ProducerId>,
        teaser: DelegateTeaser,
    ) -> Result<bool, EngineError> {
        self.request(reservation.parent_session_id, |reply| {
            SessionCommand::Producer(
                super::producers::ProducerCommand::CommitDelegationCompletion {
                    reservation,
                    producer_id,
                    teaser,
                    reply,
                },
            )
        })
        .await
    }

    async fn terminalize_child_without_run(
        &self,
        child_session_id: SessionId,
        status: SessionStatus,
        reason: &str,
    ) -> Result<(), EngineError> {
        #[cfg(test)]
        if self
            .inner
            .test_hooks
            .delegate_terminal_append_failures
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| {
                (remaining > 0).then(|| remaining - 1)
            })
            .is_ok()
        {
            return Err(EngineError::ToolFailed(
                "injected delegate terminal append failure".into(),
            ));
        }
        let child = self.inner.store.get(child_session_id)?;
        if child.status == SessionStatus::Idle {
            self.append(
                child_session_id,
                None,
                super::event_origin("engine:delegation"),
                Event::DelegateChildTerminated {
                    status,
                    reason: Some(safe_error(reason)),
                },
            )
            .await?;
        }
        Ok(())
    }

    async fn void_runless_pending_inputs(
        &self,
        child_session_id: SessionId,
    ) -> Result<(), EngineError> {
        let events = self.inner.store.get(child_session_id)?.log.event_snapshot();
        let boundary = events
            .iter()
            .rev()
            .find(|event| {
                matches!(
                    event.payload,
                    Event::RunCompleted { .. }
                        | Event::RunFailed { .. }
                        | Event::RunCancelled { .. }
                        | Event::RunInterrupted { .. }
                )
            })
            .map_or(0, |event| event.seq);
        let mut pending = std::collections::VecDeque::new();
        for event in events
            .iter()
            .filter(|event| event.seq > boundary && event.run_id.is_none())
        {
            match &event.payload {
                Event::UserInputAdmitted { input } => pending.push_back(input.clone()),
                Event::UserInputRecalled { .. } => {
                    pending.pop_back();
                }
                _ => {}
            }
        }
        for input in pending.into_iter().rev() {
            self.append(
                child_session_id,
                None,
                super::event_origin("engine:delegation"),
                Event::UserInputRecalled { input },
            )
            .await?;
        }
        Ok(())
    }

    async fn start_queued_delegation(&self, root_session_id: SessionId) -> Result<(), EngineError> {
        let admission_guard = self.inner.delegation.admission.lock().await;
        if self.background_slot_unavailable(root_session_id)? {
            return Ok(());
        }
        let child_session_id = {
            let records = self
                .inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?;
            let queue = self
                .inner
                .delegation
                .queue
                .lock()
                .map_err(|_| EngineError::ActorStopped)?;
            let position = queue.iter().position(|session_id| {
                records.get(session_id).is_some_and(|record| {
                    record.root_session_id == root_session_id
                        && record.state == DelegationState::Queued
                })
            });
            position.and_then(|position| queue.get(position).copied())
        };
        let Some(child_session_id) = child_session_id else {
            return Ok(());
        };
        let invocation_id = self
            .inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .get(&child_session_id)
            .map(|record| record.invocation_id)
            .ok_or_else(|| {
                EngineError::ToolFailed("queued subagent registry entry is missing".into())
            })?;
        let entry = self
            .delegation_event_get(invocation_id)
            .await?
            .ok_or_else(|| {
                EngineError::ToolFailed("queued subagent reservation is missing".into())
            })?;
        let child_run_id = self.ensure_delegate_run(&entry, None).await?;
        let removed = {
            let mut records = self
                .inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?;
            let record = records.get_mut(&child_session_id).ok_or_else(|| {
                EngineError::ToolFailed("queued subagent registry entry is missing".into())
            })?;
            record.child_run_id = Some(child_run_id);
            record.state = DelegationState::Running;
            record.monitor_started = true;
            let mut queue = self
                .inner
                .delegation
                .queue
                .lock()
                .map_err(|_| EngineError::ActorStopped)?;
            queue
                .iter()
                .position(|session_id| *session_id == child_session_id)
                .is_some_and(|position| queue.remove(position).is_some())
        };
        if !removed {
            let error =
                EngineError::ToolFailed("queued subagent disappeared during startup".into());
            let _ = self.cancel_run_durably(
                child_run_id,
                Some("queued subagent disappeared during startup".into()),
            );
            drop(admission_guard);
            let _ = self.spawn_background_finish_retry(child_session_id, invocation_id);
            return Err(error);
        }
        let handle = DelegateHandle {
            invocation_id,
            child_session_id,
            child_run_id: Some(child_run_id),
        };
        // A promotion starts fresh work rather than recovering an abandoned
        // record, so it carries no settling ticket: `resume` has no reason to
        // wait for a delegate it just started.
        if let Err(error) = self.spawn_background_monitor(handle, None) {
            self.release_recovered_monitor(child_session_id, invocation_id);
            let _ = self.cancel_run_durably(
                child_run_id,
                Some("background delegate monitor failed to start".into()),
            );
            drop(admission_guard);
            let _ = self.spawn_background_finish_retry(child_session_id, invocation_id);
            return Err(error);
        }
        Ok(())
    }

    pub fn subagent_agent_type(
        &self,
        caller_session_id: SessionId,
        child_session_id: SessionId,
    ) -> Result<cookie_agent_protocol::AgentId, EngineError> {
        let child = self.inner.store.get(child_session_id)?;
        match child.meta.origin {
            SessionOrigin::Delegated {
                parent_session_id, ..
            } if parent_session_id == caller_session_id => {
                Ok(child.meta.creation_selection.agent.clone())
            }
            _ => Err(EngineError::ToolFailed(
                "subagent session is not owned by the caller".into(),
            )),
        }
    }

    async fn ensure_subagent_owned(
        &self,
        caller_session_id: SessionId,
        child_session_id: SessionId,
    ) -> Result<DelegateHandle, EngineError> {
        self.subagent_agent_type(caller_session_id, child_session_id)?;
        let child = self.inner.store.get(child_session_id)?;
        let SessionOrigin::Delegated {
            parent_run_id,
            parent_tool_call_id,
            invocation_id,
            ..
        } = child.meta.origin
        else {
            unreachable!("ownership check requires a delegated origin");
        };
        let registry_record = self
            .inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .get(&child_session_id)
            .copied();
        if registry_record.is_some_and(|record| record.parent_session_id != caller_session_id) {
            return Err(EngineError::ToolFailed(
                "subagent registry ownership does not match its parent link".into(),
            ));
        }
        let (parent_run_id, parent_tool_call_id, invocation_id) = registry_record.map_or(
            (parent_run_id, parent_tool_call_id, invocation_id),
            |record| {
                (
                    record.parent_run_id,
                    record.parent_tool_call_id,
                    record.invocation_id,
                )
            },
        );
        self.ensure_parent_link(
            caller_session_id,
            parent_run_id,
            parent_tool_call_id,
            child_session_id,
        )
        .await?;
        let child_run_id = registry_record
            .and_then(|record| record.child_run_id)
            .or_else(|| {
                self.inner
                    .delegation_events
                    .get(invocation_id)
                    .and_then(|entry| entry.child_run_id)
            });
        Ok(DelegateHandle {
            invocation_id,
            child_session_id,
            child_run_id,
        })
    }

    /// Reports **session liveness**, not delegation lifecycle. A running
    /// (including queued/starting, or finished-then-woken) subagent reports
    /// "running"; once it ends, the result is the last assistant message of its
    /// most recent turn, paginated by `offset`/`limit`.
    pub async fn get_subagent_result(
        &self,
        caller_session_id: SessionId,
        reference: impl std::fmt::Display,
        wait: bool,
        offset: u32,
        limit: u32,
        cancellation: CancellationToken,
    ) -> Result<ToolResult, EngineError> {
        let child_session_id = self.resolve_subagent_target(
            caller_session_id,
            reference,
            SubagentScope::DirectChildren,
        )?;
        self.ensure_subagent_owned(caller_session_id, child_session_id)
            .await?;
        loop {
            let child = self.inner.store.get(child_session_id)?;
            // Liveness is the session's own state *plus* an owed producer wake.
            // A `send_message` to a finished child is durably accepted before it
            // replies, but its `RunStarted` is appended asynchronously: pending
            // mail (and an in-flight/scheduled wake) covers that gap so a
            // same-turn read cannot return the previous turn's text.
            let running = matches!(child.status, SessionStatus::Running | SessionStatus::Idle)
                || super::producers::has_pending_producer_message(&child.log.event_snapshot())
                || self.producer_wake_in_flight(child_session_id);
            if !running {
                let text = last_assistant_message(&child);
                return Ok(paginated_subagent_result(
                    session_status_name(child.status),
                    &text,
                    offset,
                    limit,
                ));
            }
            if !wait {
                return Ok(paginated_subagent_result("running", "", offset, limit));
            }
            tokio::select! {
                () = cancellation.cancelled() => {
                    return Err(EngineError::ToolFailed("subagent result wait cancelled".into()));
                }
                () = tokio::time::sleep(std::time::Duration::from_millis(25)) => {}
            }
        }
    }

    pub async fn steer_subagent(
        &self,
        caller_session_id: SessionId,
        child_session_id: SessionId,
        message: String,
    ) -> Result<ToolResult, EngineError> {
        if message.trim().is_empty() {
            return Err(EngineError::ToolFailed(
                "subagent steer message must not be empty".into(),
            ));
        }
        let handle = self
            .ensure_subagent_owned(caller_session_id, child_session_id)
            .await?;
        let admission_guard = self.inner.delegation.admission.lock().await;
        let record = self
            .inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .get(&child_session_id)
            .copied()
            .ok_or_else(|| EngineError::ToolFailed("subagent registry entry is missing".into()))?;
        let child = self.inner.store.get(child_session_id)?;
        match record.state {
            DelegationState::Queued => {
                self.append(
                    child_session_id,
                    None,
                    super::event_origin("engine:delegation"),
                    Event::UserInputAdmitted { input: message },
                )
                .await?;
                drop(admission_guard);
                Ok(steered_delegate_result(
                    child_session_id,
                    child.meta.short_id.as_deref(),
                    "queued",
                ))
            }
            DelegationState::Starting | DelegationState::Running => {
                if matches!(
                    child.status,
                    SessionStatus::Completed
                        | SessionStatus::Failed
                        | SessionStatus::Interrupted
                        | SessionStatus::Cancelled
                ) {
                    return Err(EngineError::ToolFailed(format!(
                        "subagent is terminal ({}) and cannot be steered",
                        session_status_name(child.status)
                    )));
                }
                let child_run_id =
                    record.child_run_id.or(handle.child_run_id).ok_or_else(|| {
                        EngineError::ToolFailed("subagent has not started a run".into())
                    })?;
                drop(admission_guard);
                let result = self
                    .request(child_session_id, |reply| SessionCommand::Steer {
                        run: child_run_id,
                        origin: super::event_origin("engine:delegation"),
                        input: message,
                        original_input: None,
                        reply,
                    })
                    .await?;
                if !result.accepted {
                    return Err(EngineError::ToolFailed(
                        "subagent is no longer running".into(),
                    ));
                }
                Ok(steered_delegate_result(
                    child_session_id,
                    child.meta.short_id.as_deref(),
                    "running",
                ))
            }
            DelegationState::Finished(status) => Err(EngineError::ToolFailed(format!(
                "subagent is terminal ({}) and cannot be steered",
                session_status_name(status)
            ))),
        }
    }

    pub async fn cancel_subagent(
        &self,
        caller_session_id: SessionId,
        reference: impl std::fmt::Display,
        reason: Option<String>,
    ) -> Result<ToolResult, EngineError> {
        let child_session_id = self.resolve_subagent_target(
            caller_session_id,
            reference,
            SubagentScope::DirectChildren,
        )?;
        let handle = self
            .ensure_subagent_owned(caller_session_id, child_session_id)
            .await?;
        let admission_guard = self.inner.delegation.admission.lock().await;
        let record = self
            .inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .get(&child_session_id)
            .copied()
            .ok_or_else(|| EngineError::ToolFailed("subagent registry entry is missing".into()))?;
        if record.state == DelegationState::Queued {
            self.inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?
                .get(&child_session_id)
                .filter(|record| record.state == DelegationState::Queued)
                .map(|_| ())
                .ok_or_else(|| {
                    EngineError::ToolFailed("subagent is not queued for cancellation".into())
                })?;
            let cancellation_reason = reason
                .as_deref()
                .unwrap_or("queued subagent cancelled before startup");
            let delegation_events = self.inner.delegation_events.clone();
            self.spawn_admission_blocking(move || {
                delegation_events.mark_finished(record.invocation_id, SessionStatus::Cancelled)
            })
            .await?;
            self.void_runless_pending_inputs(child_session_id).await?;
            self.terminalize_child_without_run(
                child_session_id,
                SessionStatus::Cancelled,
                cancellation_reason,
            )
            .await?;
            self.inner
                .delegation
                .queue
                .lock()
                .map_err(|_| EngineError::ActorStopped)?
                .retain(|session_id| *session_id != child_session_id);
            if let Some(record) = self
                .inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?
                .get_mut(&child_session_id)
            {
                record.state = DelegationState::Finished(SessionStatus::Cancelled);
            }
            drop(admission_guard);
            self.finish_background_or_retry(child_session_id, record.invocation_id)
                .await;
            let short_id = self.session_short_id(child_session_id);
            return Ok(cancelled_delegate_result(
                child_session_id,
                short_id.as_deref(),
                None,
            ));
        }
        let child_run_id = record
            .child_run_id
            .or(handle.child_run_id)
            .or_else(|| {
                self.inner
                    .delegation_events
                    .get(handle.invocation_id)
                    .and_then(|entry| entry.child_run_id)
            })
            .ok_or_else(|| EngineError::ToolFailed("subagent has not started a run".into()))?;
        drop(admission_guard);
        let child = self.inner.store.get(child_session_id)?;
        if matches!(child.status, SessionStatus::Running | SessionStatus::Idle) {
            self.cancel_run_durably(child_run_id, reason)?;
        }
        loop {
            let child = self.inner.store.get(child_session_id)?;
            if !matches!(child.status, SessionStatus::Running | SessionStatus::Idle) {
                return Ok(cancelled_delegate_result(
                    child_session_id,
                    child.meta.short_id.as_deref(),
                    Some(delegate_final_text(&child, Some(child_run_id)).to_owned()),
                ));
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
    }

    #[cfg(test)]
    pub(crate) fn delegation_queue_contains(
        &self,
        child_session_id: SessionId,
    ) -> Result<bool, EngineError> {
        Ok(self
            .inner
            .delegation
            .queue
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .contains(&child_session_id))
    }

    #[cfg(test)]
    pub(crate) async fn wait_for_delegation_reconciliation_for_test(&self) {
        while self
            .inner
            .delegation
            .reconciliation_running
            .load(Ordering::Acquire)
            || self
                .inner
                .delegation
                .reconciliation_requested
                .load(Ordering::Acquire)
        {
            tokio::task::yield_now().await;
        }
    }

    #[cfg(test)]
    pub(crate) fn delegation_registry_snapshot(
        &self,
        child_session_id: SessionId,
    ) -> Result<(InvocationId, Option<SessionStatus>, bool), EngineError> {
        self.inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .get(&child_session_id)
            .map(|record| {
                (
                    record.invocation_id,
                    match record.state {
                        DelegationState::Finished(status) => Some(status),
                        _ => None,
                    },
                    record.counts_slot,
                )
            })
            .ok_or_else(|| EngineError::ToolFailed("subagent registry entry is missing".into()))
    }

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

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

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

    #[cfg(test)]
    pub(crate) fn install_resume_attachment_hook(
        &self,
    ) -> (tokio::sync::oneshot::Receiver<()>, Arc<tokio::sync::Notify>) {
        self.install_resume_test_hook(&self.inner.test_hooks.resume_attachment_hook)
    }

    #[cfg(test)]
    pub(crate) fn install_resume_rollback_hook(
        &self,
    ) -> (tokio::sync::oneshot::Receiver<()>, Arc<tokio::sync::Notify>) {
        self.install_resume_test_hook(&self.inner.test_hooks.resume_rollback_hook)
    }

    #[cfg(test)]
    fn install_resume_test_hook(
        &self,
        slot: &std::sync::Mutex<Option<Arc<super::ResumeAdmissionHook>>>,
    ) -> (tokio::sync::oneshot::Receiver<()>, Arc<tokio::sync::Notify>) {
        let (reached, receiver) = tokio::sync::oneshot::channel();
        let release = Arc::new(tokio::sync::Notify::new());
        *slot.lock().expect("resume test hook lock poisoned") =
            Some(Arc::new(super::ResumeAdmissionHook {
                reached: std::sync::Mutex::new(Some(reached)),
                release: Arc::clone(&release),
            }));
        (receiver, release)
    }

    #[cfg(test)]
    async fn wait_for_resume_test_hook(
        &self,
        slot: &std::sync::Mutex<Option<Arc<super::ResumeAdmissionHook>>>,
    ) {
        let hook = slot.lock().expect("resume test hook lock poisoned").clone();
        if let Some(hook) = hook {
            if let Some(reached) = hook
                .reached
                .lock()
                .expect("resume test hook reached lock poisoned")
                .take()
            {
                let _ = reached.send(());
            }
            hook.release.notified().await;
        }
    }

    #[cfg(test)]
    pub(crate) fn set_delegation_slot_ownership(
        &self,
        child_session_id: SessionId,
        background: bool,
        counts_slot: bool,
    ) -> Result<(), EngineError> {
        let mut records = self
            .inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?;
        let record = records
            .get_mut(&child_session_id)
            .ok_or_else(|| EngineError::ToolFailed("subagent registry entry is missing".into()))?;
        record.background = background;
        record.counts_slot = counts_slot;
        Ok(())
    }

    /// Simulates a corrupted foreground record by registering a real delegation
    /// producer and writing it onto the child's record. The registry rebuild is
    /// expected to strip it again and retire the registration.
    #[cfg(test)]
    pub(crate) async fn plant_foreground_delegation_producer_for_test(
        &self,
        child_session_id: SessionId,
    ) -> Result<ProducerId, EngineError> {
        let (parent_session_id, invocation_id) = {
            let records = self
                .inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?;
            let record = records.get(&child_session_id).ok_or_else(|| {
                EngineError::ToolFailed("subagent registry entry is missing".into())
            })?;
            (record.parent_session_id, record.invocation_id)
        };
        let mut guard = self
            .register_background_delegation_producer(parent_session_id, invocation_id)
            .await?;
        let producer_id = guard.producer_id;
        {
            let mut records = self
                .inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?;
            let record = records
                .get_mut(&child_session_id)
                .filter(|record| record.invocation_id == invocation_id)
                .ok_or_else(|| {
                    EngineError::ToolFailed("subagent registry entry is missing".into())
                })?;
            record.background = false;
            record.producer_id = Some(producer_id);
        }
        guard.handoff();
        Ok(producer_id)
    }

    #[cfg(test)]
    pub(crate) fn rebuild_delegation_registry_for_test(&self) -> Result<(), EngineError> {
        self.rebuild_delegation_registry(&self.inner.delegation_events.entries(), true)
    }

    #[cfg(test)]
    pub(crate) fn delegation_producer_ownership_for_test(
        &self,
        child_session_id: SessionId,
    ) -> Result<(bool, Option<ProducerId>), EngineError> {
        self.inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .get(&child_session_id)
            .map(|record| (record.background, record.producer_id))
            .ok_or_else(|| EngineError::ToolFailed("subagent registry entry is missing".into()))
    }

    pub(super) fn rebuild_delegation_registry(
        &self,
        entries: &[delegation_events::DelegationEntry],
        recover: bool,
    ) -> Result<(), EngineError> {
        let mut latest = std::collections::HashMap::new();
        for entry in entries {
            latest.insert(entry.reservation.child_session_id, entry);
        }
        let mut stale_producers = {
            let mut records = self
                .inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?;
            let stale = records
                .iter()
                .filter(|(child_session_id, record)| {
                    latest
                        .get(child_session_id)
                        .is_none_or(|entry| entry.reservation.invocation_id != record.invocation_id)
                })
                .filter_map(|(_, record)| {
                    record.producer_id.map(|producer_id| {
                        (record.parent_session_id, record.invocation_id, producer_id)
                    })
                })
                .collect::<Vec<_>>();
            records.retain(|child_session_id, record| {
                latest
                    .get(child_session_id)
                    .is_some_and(|entry| entry.reservation.invocation_id == record.invocation_id)
            });
            stale
        };
        self.inner
            .delegation
            .queue
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .retain(|child_session_id| latest.contains_key(child_session_id));
        for entry in latest.into_values() {
            let child_id = entry.reservation.child_session_id;
            // No child event log is read to rebuild the registry (§4.1.3):
            // existence comes from placement, run status from the tree's
            // terminal-run cache. An unknown child stays unregistered, as before.
            if !self.inner.store.session_exists(child_id) {
                continue;
            }
            // Facts about the parent as a delegation *parent*, taken from the
            // parent's own log the one time it was read: by this load pass for a
            // child, by the startup fold for a root. Reopening a child log here
            // would fold a second time what the bulk pass already produced
            // (§4.1.3).
            //
            // `None` means the parent is a child whose tree is still unloaded.
            // Registering it on guesses would be the wrong direction to be wrong
            // in, so the entry is left out and this rebuild runs again when that
            // tree's load delivers its `parent_facts`: the registry is a fold
            // over all known entries that preserves each matching record's
            // producer and monitor state, so a later pass resolves this entry
            // without disturbing any other one.
            let Some(facts) = self
                .inner
                .store
                .parent_run_facts(entry.reservation.parent_session_id)?
            else {
                continue;
            };
            let root_session_id = facts.root_session_id;
            let queued_event = facts.queued.iter().any(|(run_id, session)| {
                *run_id == entry.reservation.parent_run_id && *session == child_id
            });
            let notification_sent = facts.notified.iter().any(|(invocation_id, session)| {
                *invocation_id == entry.reservation.invocation_id && *session == child_id
            }) || facts
                .producer_accepted
                .contains(&entry.reservation.invocation_id);
            let background = entry.request.background;
            // A cache miss means "not terminal", the conservative reading that
            // startup recovery then corrects — never the other way round.
            let exact_run_status = entry
                .child_run_id
                .and_then(|run_id| self.inner.store.terminal_run_status(child_id, run_id));
            let parent_interrupted = facts
                .interrupted_runs
                .contains(&entry.reservation.parent_run_id);
            let state = if let Some(status) = entry.terminal_status {
                DelegationState::Finished(status)
            } else if entry.child_run_id.is_none() && queued_event {
                DelegationState::Queued
            } else if parent_interrupted {
                DelegationState::Finished(SessionStatus::Interrupted)
            } else if exact_run_status.is_some_and(|status| {
                matches!(
                    status,
                    SessionStatus::Completed
                        | SessionStatus::Failed
                        | SessionStatus::Interrupted
                        | SessionStatus::Cancelled
                )
            }) {
                DelegationState::Finished(exact_run_status.expect("checked terminal status"))
            } else {
                DelegationState::Running
            };
            let counts_slot = background && !facts.delegated;
            let (mut producer_id, mut monitor_started) = self
                .inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?
                .get(&child_id)
                .filter(|record| record.invocation_id == entry.reservation.invocation_id)
                .map_or((None, false), |record| {
                    (record.producer_id, record.monitor_started)
                });
            // A foreground delegation must never own a background producer
            // registration. If a preserved record carries one anyway, drop it and
            // route the registration through the stale sweep so the normal
            // unregister path retires it instead of leaving it orphaned.
            let mut stale_foreground_producer = None;
            if !background && let Some(stale) = producer_id {
                stale_foreground_producer = Some(stale);
                producer_id = None;
                monitor_started = false;
            }
            let parent_session_id = entry.reservation.parent_session_id;
            let invocation_id = entry.reservation.invocation_id;
            self.inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?
                .insert(
                    child_id,
                    DelegationRecord {
                        parent_session_id,
                        parent_run_id: entry.reservation.parent_run_id,
                        parent_tool_call_id: entry.reservation.parent_tool_call_id,
                        invocation_id,
                        root_session_id,
                        child_run_id: entry.child_run_id,
                        state,
                        background,
                        counts_slot,
                        notification_sent,
                        producer_id,
                        monitor_started,
                    },
                );
            if let Some(stale) = stale_foreground_producer {
                stale_producers.push((parent_session_id, invocation_id, stale));
            }
            if state == DelegationState::Queued {
                let mut queue = self
                    .inner
                    .delegation
                    .queue
                    .lock()
                    .map_err(|_| EngineError::ActorStopped)?;
                if !queue.contains(&child_id) {
                    queue.push_back(child_id);
                }
            }
        }
        if recover
            && let Some(runtime) = self
                .inner
                .runtime
                .clone()
                .or_else(|| tokio::runtime::Handle::try_current().ok())
        {
            self.schedule_delegation_reconciliation(&runtime, stale_producers)?;
        }
        Ok(())
    }

    fn schedule_delegation_reconciliation(
        &self,
        runtime: &tokio::runtime::Handle,
        stale_producers: Vec<(SessionId, InvocationId, ProducerId)>,
    ) -> Result<(), EngineError> {
        self.inner
            .delegation
            .recovery_stale_producers
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .extend(stale_producers);
        self.inner
            .delegation
            .reconciliation_requested
            .store(true, Ordering::Release);
        if self
            .inner
            .delegation
            .reconciliation_running
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return Ok(());
        }
        let engine = self.clone();
        let reconcile = engine.clone();
        if !self.spawn_admission_task(runtime, async move {
            reconcile.run_delegation_reconciliation().await;
        }) {
            self.inner
                .delegation
                .reconciliation_running
                .store(false, Ordering::Release);
            self.inner.delegation.recovery_settled.notify_waiters();
            return Err(EngineError::ActorStopped);
        }
        Ok(())
    }

    async fn run_delegation_reconciliation(&self) {
        loop {
            self.inner
                .delegation
                .reconciliation_requested
                .store(false, Ordering::Release);
            let stale_producers = std::mem::take(
                &mut *self
                    .inner
                    .delegation
                    .recovery_stale_producers
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner()),
            );
            if let Err(error) = self.reconcile_delegation_producers(stale_producers).await {
                eprintln!("delegation producer recovery deferred: {error}");
            }
            if self
                .inner
                .delegation
                .reconciliation_requested
                .load(Ordering::Acquire)
            {
                continue;
            }
            self.inner
                .delegation
                .reconciliation_running
                .store(false, Ordering::Release);
            if !self
                .inner
                .delegation
                .reconciliation_requested
                .load(Ordering::Acquire)
                || self
                    .inner
                    .delegation
                    .reconciliation_running
                    .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
                    .is_err()
            {
                // Every claim this pass made is now owned by its monitor, so a
                // settling waiter can make progress.
                self.inner.delegation.recovery_settled.notify_waiters();
                break;
            }
        }
    }

    /// Recreates volatile delegation registrations before recovered work is
    /// monitored or dequeued.
    async fn reconcile_delegation_producers(
        &self,
        stale_producers: Vec<(SessionId, InvocationId, ProducerId)>,
    ) -> Result<(), EngineError> {
        for (parent_session_id, invocation_id, producer_id) in stale_producers {
            if !self.inner.store.is_owned(parent_session_id) {
                continue;
            }
            match self
                .unregister_producer(
                    parent_session_id,
                    delegation_producer_authority(invocation_id),
                    producer_id,
                )
                .await
            {
                Ok(()) | Err(EngineError::Producer(_)) => {}
                Err(error) => return Err(error),
            }
        }
        let records = self
            .inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .iter()
            .map(|(session_id, record)| (*session_id, *record))
            .collect::<Vec<_>>();

        for (child_session_id, record) in &records {
            if !record.background
                || record.notification_sent
                || record.producer_id.is_some()
                || !self.inner.store.is_owned(record.parent_session_id)
            {
                continue;
            }
            let mut guard = self
                .register_background_delegation_producer(
                    record.parent_session_id,
                    record.invocation_id,
                )
                .await?;
            let mut current = self
                .inner
                .delegation
                .by_session
                .lock()
                .map_err(|_| EngineError::ActorStopped)?;
            if let Some(current_record) = current.get_mut(child_session_id).filter(|current| {
                current.invocation_id == record.invocation_id && current.producer_id.is_none()
            }) {
                current_record.producer_id = Some(guard.producer_id);
                guard.handoff();
            }
        }

        let records = self
            .inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?
            .iter()
            .map(|(session_id, record)| (*session_id, *record))
            .collect::<Vec<_>>();
        let mut queued_roots = Vec::new();
        let mut finished = Vec::new();
        for (child_session_id, record) in records {
            if !record.background || !self.inner.store.is_owned(record.parent_session_id) {
                continue;
            }
            match record.state {
                DelegationState::Finished(_) => {
                    finished.push((child_session_id, record.invocation_id));
                }
                DelegationState::Queued => queued_roots.push(record.root_session_id),
                DelegationState::Running if record.child_run_id.is_some() => {
                    let Some(ticket) =
                        self.claim_recovered_monitor(child_session_id, record.invocation_id)?
                    else {
                        continue;
                    };
                    self.terminalize_abandoned_delegate_run(
                        child_session_id,
                        record.child_run_id.expect("checked child run"),
                    )
                    .await;
                    if let Err(error) = self.spawn_background_monitor(
                        DelegateHandle {
                            invocation_id: record.invocation_id,
                            child_session_id,
                            child_run_id: record.child_run_id,
                        },
                        Some(ticket),
                    ) {
                        self.release_recovered_monitor(child_session_id, record.invocation_id);
                        return Err(error);
                    }
                }
                DelegationState::Running | DelegationState::Starting => {
                    let Some(ticket) =
                        self.claim_recovered_monitor(child_session_id, record.invocation_id)?
                    else {
                        continue;
                    };
                    let entry = self
                        .delegation_event_get(record.invocation_id)
                        .await
                        .inspect_err(|_| {
                            self.release_recovered_monitor(child_session_id, record.invocation_id);
                        })?
                        .ok_or_else(|| {
                            self.release_recovered_monitor(child_session_id, record.invocation_id);
                            EngineError::ToolFailed(
                                "recovered subagent reservation is missing".into(),
                            )
                        })?;
                    let child_run_id =
                        self.ensure_delegate_run(&entry, None)
                            .await
                            .inspect_err(|_| {
                                self.release_recovered_monitor(
                                    child_session_id,
                                    record.invocation_id,
                                );
                            })?;
                    if let Some(current) = self
                        .inner
                        .delegation
                        .by_session
                        .lock()
                        .map_err(|_| EngineError::ActorStopped)?
                        .get_mut(&child_session_id)
                        .filter(|current| current.invocation_id == record.invocation_id)
                    {
                        current.child_run_id = Some(child_run_id);
                        current.state = DelegationState::Running;
                    }
                    if let Err(error) = self.spawn_background_monitor(
                        DelegateHandle {
                            invocation_id: record.invocation_id,
                            child_session_id,
                            child_run_id: Some(child_run_id),
                        },
                        Some(ticket),
                    ) {
                        self.release_recovered_monitor(child_session_id, record.invocation_id);
                        return Err(error);
                    }
                }
            }
        }
        for (child_session_id, invocation_id) in finished {
            self.finish_background_or_retry(child_session_id, invocation_id)
                .await;
        }
        queued_roots.sort_unstable();
        queued_roots.dedup();
        for root_session_id in queued_roots {
            self.start_queued_delegation(root_session_id).await?;
        }
        Ok(())
    }

    /// Adopts a recovered background delegate whose run is not live here.
    ///
    /// A record the rebuild read as `Running` from durable facts alone may
    /// belong to a run that died with the previous process: the child log has a
    /// `RunStarted` and no terminal event, and nothing in this process is going
    /// to append one. Adopting the child runs its own reconciliation, which
    /// terminalizes the run as interrupted, so the monitor spawned next
    /// observes a terminal state instead of waiting forever and leaking the
    /// parent's background-delegation slot.
    ///
    /// Best-effort by design: a child owned by another process, or one this
    /// process already adopted, is left to that owner.
    async fn terminalize_abandoned_delegate_run(
        &self,
        child_session_id: SessionId,
        child_run_id: RunId,
    ) {
        let live = self
            .inner
            .sessions
            .active
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .contains_key(&child_run_id);
        if live {
            return;
        }
        let _residency = self.inner.sessions.residency_mutation.lock().await;
        if let Err(error) = self.ensure_session_owned(child_session_id) {
            eprintln!("session {child_session_id} delegate recovery deferred: {error}");
        }
    }

    /// Claims the one recovered monitor for `child_session_id`.
    ///
    /// The returned ticket keeps the record's root marked as still settling
    /// until it is dropped, which is what lets [`Engine::resume`] return only
    /// after the recovery its adoption scheduled has run to a conclusion.
    fn claim_recovered_monitor(
        &self,
        child_session_id: SessionId,
        invocation_id: InvocationId,
    ) -> Result<Option<RecoveryMonitorTicket>, EngineError> {
        let mut records = self
            .inner
            .delegation
            .by_session
            .lock()
            .map_err(|_| EngineError::ActorStopped)?;
        let Some(record) = records
            .get_mut(&child_session_id)
            .filter(|record| record.invocation_id == invocation_id)
        else {
            return Ok(None);
        };
        if record.monitor_started
            || !matches!(
                record.state,
                DelegationState::Running | DelegationState::Starting
            )
        {
            return Ok(None);
        }
        record.monitor_started = true;
        let root_session_id = record.root_session_id;
        drop(records);
        *self
            .inner
            .delegation
            .recovery_pending
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .entry(root_session_id)
            .or_insert(0) += 1;
        Ok(Some(RecoveryMonitorTicket {
            engine: self.clone(),
            root_session_id,
        }))
    }

    fn release_recovery_claim(&self, root_session_id: SessionId) {
        {
            let mut pending = self
                .inner
                .delegation
                .recovery_pending
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            if let Some(count) = pending.get_mut(&root_session_id) {
                *count = count.saturating_sub(1);
                if *count == 0 {
                    pending.remove(&root_session_id);
                }
            }
        }
        self.inner.delegation.recovery_settled.notify_waiters();
    }

    fn delegation_recovery_settled(&self, root_session_id: SessionId) -> bool {
        if self
            .inner
            .delegation
            .reconciliation_running
            .load(Ordering::Acquire)
            || self
                .inner
                .delegation
                .reconciliation_requested
                .load(Ordering::Acquire)
        {
            return false;
        }
        !self
            .inner
            .delegation
            .recovery_pending
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .contains_key(&root_session_id)
    }

    /// Waits until the delegation recovery scheduled for `root_session_id` has
    /// settled, or `bound` expires.
    ///
    /// Returns `false` when the bound expired, which leaves the caller with the
    /// pre-existing eventually-consistent behaviour rather than a failure.
    ///
    /// This must never be awaited from inside a session actor: resolving a
    /// recovered delegation goes through its parent session's mailbox.
    pub(crate) async fn await_delegation_recovery(
        &self,
        root_session_id: SessionId,
        bound: std::time::Duration,
    ) -> bool {
        let settle = async {
            while !self.delegation_recovery_settled(root_session_id) {
                // A bounded wait on the notification, so a wakeup that lands
                // between the check and the registration costs a short delay
                // rather than the whole deadline.
                let _ = tokio::time::timeout(
                    std::time::Duration::from_millis(10),
                    self.inner.delegation.recovery_settled.notified(),
                )
                .await;
            }
        };
        tokio::time::timeout(bound, settle).await.is_ok()
    }

    fn release_recovered_monitor(&self, child_session_id: SessionId, invocation_id: InvocationId) {
        if let Ok(mut records) = self.inner.delegation.by_session.lock()
            && let Some(record) = records
                .get_mut(&child_session_id)
                .filter(|record| record.invocation_id == invocation_id)
        {
            record.monitor_started = false;
        }
    }
}

/// Keeps a root marked as "recovery still settling" for the lifetime of one
/// recovered background monitor.
///
/// Dropping it is the single release point: the monitor task owns the ticket
/// for as long as it is resolving its record, and every early-return path in
/// the reconciliation drops it on the spot.
pub(super) struct RecoveryMonitorTicket {
    engine: Engine,
    root_session_id: SessionId,
}

impl Drop for RecoveryMonitorTicket {
    fn drop(&mut self) {
        self.engine.release_recovery_claim(self.root_session_id);
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) enum DelegationState {
    Queued,
    Starting,
    Running,
    Finished(SessionStatus),
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) struct DelegationRecord {
    pub(super) parent_session_id: SessionId,
    pub(super) parent_run_id: RunId,
    pub(super) parent_tool_call_id: ToolCallId,
    pub(super) invocation_id: InvocationId,
    pub(super) root_session_id: SessionId,
    pub(super) child_run_id: Option<RunId>,
    pub(super) state: DelegationState,
    pub(super) background: bool,
    pub(super) counts_slot: bool,
    pub(super) notification_sent: bool,
    pub(super) producer_id: Option<ProducerId>,
    pub(super) monitor_started: bool,
}

struct BackgroundProducerGuard {
    engine: Engine,
    runtime: Option<tokio::runtime::Handle>,
    parent_session_id: SessionId,
    invocation_id: InvocationId,
    producer_id: ProducerId,
    active: bool,
}

impl BackgroundProducerGuard {
    fn handoff(&mut self) {
        self.active = false;
    }
}

impl Drop for BackgroundProducerGuard {
    fn drop(&mut self) {
        if !self.active {
            return;
        }
        let Some(runtime) = self
            .runtime
            .clone()
            .or_else(|| tokio::runtime::Handle::try_current().ok())
        else {
            return;
        };
        let engine = self.engine.clone();
        let unregister_engine = engine.clone();
        let session_id = self.parent_session_id;
        let authority = delegation_producer_authority(self.invocation_id);
        let producer_id = self.producer_id;
        let _ = engine.spawn_admission_task(&runtime, async move {
            let _ = unregister_engine
                .unregister_producer(session_id, authority, producer_id)
                .await;
        });
    }
}

fn delegation_producer_authority(invocation_id: InvocationId) -> ProducerAuthority {
    ProducerAuthority {
        owner: ProducerOwner::Delegation { invocation_id },
        connection_epoch: None,
    }
}

/// The completion body shared by the foreground `delegate_subagent` result and
/// the pushed `<subagent_notification>`. Both the runtime renderer
/// ([`render_background_completion`]) and the history-replay renderer
/// (`model_history`) call this so the stored-body dedup comparison always sees
/// byte-identical text.
pub(crate) fn render_delegate_teaser_body(
    preview: &str,
    status: SessionStatus,
    total_lines: u64,
    handle: &str,
) -> String {
    // Only a cut preview points at `get_subagent_result`; a complete one
    // would send the model to re-read what it already has.
    let more = if preview_is_truncated(preview, total_lines) {
        format!("; use get_subagent_result with session_id \"{handle}\" for the full output")
    } else {
        "; full output shown".to_owned()
    };
    // Line breaks and tabs survive so lists, tables, and code in the report
    // keep their shape; every other control character is folded away.
    let preview = crate::tool_api::sanitize_tool_display(preview, PREVIEW_MAX_BYTES);
    let preview = if preview.trim().is_empty() {
        "(no output)"
    } else {
        preview.as_str()
    };
    format!(
        "{preview}\n\n[subagent session {handle}; {}; {total_lines} lines{more}]",
        session_status_name(status),
    )
}

/// Whether [`preview_text`] cut the output, judged from the preview and line
/// count alone so history replay, which keeps only those, renders the same
/// body. A byte cut lands within a character of the cap, so a complete
/// preview of 2045..=2048 bytes also counts as cut: that errs toward the hint.
fn preview_is_truncated(preview: &str, total_lines: u64) -> bool {
    let preview_lines = if preview.is_empty() {
        0
    } else {
        // Counted by separator, not `lines()`, which drops a trailing empty
        // line the full count kept.
        preview.split('\n').count() as u64
    };
    // A UTF-8 character is at most four bytes, so a cut ends past this.
    preview_lines < total_lines || preview.len() > PREVIEW_MAX_BYTES - 4
}

/// The pushed `<subagent_notification>` envelope, shared by the runtime and
/// history-replay renderers.
pub(crate) fn render_subagent_notification(
    preview: &str,
    status: SessionStatus,
    total_lines: u64,
    handle: &str,
) -> String {
    format!(
        "<subagent_notification>{}</subagent_notification>",
        render_delegate_teaser_body(preview, status, total_lines, handle)
    )
}

pub(super) fn render_background_completion(teaser: &DelegateTeaser) -> String {
    render_subagent_notification(
        &teaser.preview,
        teaser.status,
        teaser.total_lines,
        &teaser.short_id,
    )
}

pub(crate) fn freeze_delegated_child_policy(
    child_agent: &crate::runtime_snapshot::ResolvedAgent,
    parent_policy: &FrozenRunPolicy,
    child_selection: &cookie_agent_protocol::ModelSelection,
    inherited_suffix: &[cookie_agent_protocol::FrozenModelBinding],
    inherited_depth_ceiling: u32,
    result_limits: policy::ResultLimits,
) -> Result<FrozenRunPolicy, EngineError> {
    freeze_delegated_agent_policy(
        child_agent,
        Arc::clone(&parent_policy.registry),
        Arc::clone(&parent_policy.runtime),
        child_selection,
        inherited_suffix,
        inherited_depth_ceiling,
        policy::FreezeOptions {
            result_limits,
            model_retry: parent_policy.model_retry,
            inherited_cache_strategies: Some(
                inherited_suffix
                    .iter()
                    .map(|binding| parent_policy.raw_cache_strategy(binding))
                    .collect(),
            ),
        },
    )
}

pub(super) fn delegate_client_run_id(
    invocation_id: InvocationId,
) -> cookie_agent_protocol::ClientRunId {
    cookie_agent_protocol::ClientRunId::new(format!("delegate:{invocation_id}"))
        .expect("bounded delegate client run id")
}

pub(super) fn render_delegate_input(request: &DelegateRequestPayload) -> String {
    request.prompt.clone()
}

const DELEGATED_CONTEXT_MAX_BYTES: usize = 64 * 1024;

fn context_seed_from_history(history: Vec<oven_sdk::HistoryTurn>) -> Vec<DelegatedContextTurn> {
    let mut turns = history
        .into_iter()
        .filter_map(|turn| match turn {
            oven_sdk::HistoryTurn::User(message) => {
                let text = message
                    .content
                    .into_iter()
                    .filter_map(|part| match part {
                        oven_sdk::InputPart::Text(text) => Some(text.text),
                        _ => None,
                    })
                    .collect::<String>();
                (!text.is_empty()).then_some(DelegatedContextTurn {
                    role: DelegatedContextRole::User,
                    text,
                })
            }
            oven_sdk::HistoryTurn::Assistant(turn) => {
                let text = turn
                    .message
                    .content
                    .into_iter()
                    .filter_map(|part| match part {
                        oven_sdk::AssistantPart::Text(text) => Some(text.text),
                        _ => None,
                    })
                    .collect::<String>();
                (!text.is_empty()).then_some(DelegatedContextTurn {
                    role: DelegatedContextRole::Assistant,
                    text,
                })
            }
            oven_sdk::HistoryTurn::System(_) | oven_sdk::HistoryTurn::Tool(_) => None,
        })
        .collect::<Vec<_>>();
    let mut excess = turns
        .iter()
        .map(|turn| turn.text.len())
        .sum::<usize>()
        .saturating_sub(DELEGATED_CONTEXT_MAX_BYTES);
    let mut retained_from = 0;
    while retained_from < turns.len() && turns[retained_from].text.len() <= excess {
        excess -= turns[retained_from].text.len();
        retained_from += 1;
    }
    turns.drain(..retained_from);
    if excess > 0
        && let Some(turn) = turns.first_mut()
    {
        let mut boundary = excess;
        while !turn.text.is_char_boundary(boundary) {
            boundary += 1;
        }
        turn.text.drain(..boundary);
    }
    turns
}

#[derive(Clone)]
pub(super) struct DelegateTeaser {
    pub(super) session_id: SessionId,
    /// Model-facing handle; falls back to the UUID for pre-handle sessions.
    pub(super) short_id: String,
    pub(super) status: SessionStatus,
    pub(super) preview: String,
    pub(super) total_lines: u64,
}

fn delegate_teaser(
    child: &session::SessionProjection,
    child_session_id: SessionId,
    status: SessionStatus,
    child_run_id: Option<RunId>,
) -> DelegateTeaser {
    let text = delegate_final_text(child, child_run_id);
    DelegateTeaser {
        session_id: child_session_id,
        short_id: child
            .meta
            .short_id
            .clone()
            .unwrap_or_else(|| child_session_id.to_string()),
        status,
        preview: preview_text(text),
        total_lines: text.lines().count() as u64,
    }
}

fn delegate_final_text(child: &session::SessionProjection, child_run_id: Option<RunId>) -> &str {
    child_run_id
        .and_then(|run_id| child.runs.get(&run_id))
        .and_then(|run| run.final_text.as_deref())
        .unwrap_or("")
}

/// The last assistant message the session emitted: the most recent completed
/// run's final text, falling back to the most recent committed model turn's
/// text when a run ended without a completed turn (cancel/failure/interrupt).
fn last_assistant_message(child: &session::SessionProjection) -> String {
    let events = child.log.event_snapshot();
    if let Some(text) = events.iter().rev().find_map(|event| match &event.payload {
        Event::RunCompleted {
            final_text: Some(text),
        } => Some(text.clone()),
        _ => None,
    }) {
        return text;
    }
    events
        .iter()
        .rev()
        .find_map(|event| match &event.payload {
            Event::ModelTurnCommitted { turn, .. } => {
                let text = turn
                    .content
                    .iter()
                    .filter_map(|part| match part {
                        PersistedAssistantPart::Text { text, .. } => Some(text.as_str()),
                        _ => None,
                    })
                    .collect::<String>();
                (!text.is_empty()).then_some(text)
            }
            _ => None,
        })
        .unwrap_or_default()
}

const PREVIEW_MAX_LINES: usize = 20;
const PREVIEW_MAX_BYTES: usize = 2048;

fn preview_text(text: &str) -> String {
    let first_lines = text
        .lines()
        .take(PREVIEW_MAX_LINES)
        .collect::<Vec<_>>()
        .join("\n");
    if first_lines.len() <= PREVIEW_MAX_BYTES {
        return first_lines;
    }
    let mut end = PREVIEW_MAX_BYTES;
    while !first_lines.is_char_boundary(end) {
        end -= 1;
    }
    first_lines[..end].to_owned()
}

pub(super) const fn session_status_name(status: SessionStatus) -> &'static str {
    match status {
        SessionStatus::Idle => "running",
        SessionStatus::Running => "running",
        SessionStatus::Completed => "completed",
        SessionStatus::Failed => "failed",
        SessionStatus::Cancelled => "cancelled",
        SessionStatus::Interrupted => "interrupted",
    }
}

const fn background_queue_limit_reached(limit: u32, queued: u32) -> bool {
    queued >= limit.saturating_mul(4)
}

fn paginated_subagent_result(status: &str, text: &str, offset: u32, limit: u32) -> ToolResult {
    let lines = text.lines().collect::<Vec<_>>();
    let mut output = format!("<status>{status}</status>\n<content>\n");
    for (index, line) in lines
        .iter()
        .enumerate()
        .skip(offset as usize)
        .take(limit as usize)
    {
        output.push_str(&format!("{}: {line}\n", index + 1));
    }
    output.push_str("</content>");
    ToolResult {
        display: Some(crate::tool_api::bounded_tool_display(&output)),
        retained_output: None,
        title: safe_display("Read subagent result"),
        output,
        metadata: serde_json::json!({
            "offset": offset,
            "limit": limit,
            "total_lines": lines.len(),
        }),
        truncation: None,
        attachments: Vec::new(),
        additional_messages: Vec::new(),
    }
}

/// The model-facing reference for a session: its handle when one exists, else
/// the full UUID.
fn reference_label(session_id: Option<SessionId>, short_id: Option<&str>) -> Option<String> {
    short_id
        .map(str::to_owned)
        .or_else(|| session_id.map(|id| id.to_string()))
}

fn steered_delegate_result(
    child_session_id: SessionId,
    short_id: Option<&str>,
    status: &str,
) -> ToolResult {
    let label = reference_label(Some(child_session_id), short_id)
        .expect("steer result has a session reference");
    structured_delegate_result(
        "Subagent steered",
        format!(
            "Subagent steered. [subagent session {label}; {}]",
            safe_display(status)
        ),
        serde_json::json!({
            "session_id": child_session_id,
            "handle": short_id,
            "status": status,
        }),
    )
}

pub(crate) fn completed_delegate_result(
    child: &session::SessionProjection,
    child_run_id: Option<RunId>,
) -> ToolResult {
    terminal_delegate_result(child, child_run_id, SessionStatus::Completed)
}

fn terminal_delegate_result(
    child: &session::SessionProjection,
    child_run_id: Option<RunId>,
    status: SessionStatus,
) -> ToolResult {
    let teaser = delegate_teaser(child, child.meta.session_id, status, child_run_id);
    let output = render_delegate_teaser_body(
        &teaser.preview,
        teaser.status,
        teaser.total_lines,
        &teaser.short_id,
    );
    let metadata = serde_json::json!({
        "session_id": teaser.session_id,
        "handle": teaser.short_id,
        "status": session_status_name(teaser.status),
        "total_lines": teaser.total_lines,
    });
    structured_delegate_result("Subagent finished", output, metadata)
}

fn structured_delegate_result(title: &str, output: String, metadata: Value) -> ToolResult {
    ToolResult {
        display: Some(crate::tool_api::bounded_tool_display(&output)),
        retained_output: None,
        title: safe_display(title),
        output,
        metadata,
        truncation: None,
        attachments: Vec::new(),
        additional_messages: Vec::new(),
    }
}

pub(super) fn cancelled_delegate_result(
    child_session_id: SessionId,
    short_id: Option<&str>,
    partial_report: Option<String>,
) -> ToolResult {
    let label = reference_label(Some(child_session_id), short_id)
        .expect("cancel result has a session reference");
    let mut output = format!("Delegate cancelled. [subagent session {label}]");
    if let Some(report) = &partial_report {
        output.push_str(&format!(" Partial report: {}", safe_display(report)));
    }
    structured_delegate_result(
        "Delegate cancelled",
        output,
        serde_json::json!({
            "status": "cancelled",
            "session_id": child_session_id,
            "handle": short_id,
            "partial_report": partial_report,
        }),
    )
}

pub(super) fn cancelled_delegate_result_with_reason(
    child_session_id: Option<SessionId>,
    short_id: Option<&str>,
    reason: &str,
) -> ToolResult {
    let mut output = format!("Delegate cancelled: {}.", safe_display(reason));
    if let Some(label) = reference_label(child_session_id, short_id) {
        output.push_str(&format!(" [subagent session {label}]"));
    }
    structured_delegate_result(
        "Delegate cancelled",
        output,
        serde_json::json!({
            "status": "cancelled",
            "session_id": child_session_id,
            "handle": short_id,
            "reason": reason,
        }),
    )
}

pub(super) fn delegate_failure_result(
    child_session_id: Option<SessionId>,
    short_id: Option<&str>,
    reason: &str,
) -> ToolResult {
    let mut output = format!("Delegate failed: {}.", safe_display(reason));
    if let Some(label) = reference_label(child_session_id, short_id) {
        output.push_str(&format!(" [subagent session {label}]"));
    }
    structured_delegate_result(
        "Delegate failed",
        output,
        serde_json::json!({
            "status": "failed",
            "session_id": child_session_id,
            "handle": short_id,
            "reason": reason,
        }),
    )
}

fn validate_redelivery_mode(
    durable_background: bool,
    redelivery_background: bool,
) -> Result<(), EngineError> {
    if durable_background != redelivery_background {
        let durable = if durable_background {
            "background"
        } else {
            "foreground"
        };
        let redelivery = if redelivery_background {
            "background"
        } else {
            "foreground"
        };
        return Err(EngineError::ToolFailed(format!(
            "delegate redelivery execution mode conflict: durable invocation is {durable}, redelivery requested {redelivery}"
        )));
    }
    Ok(())
}

pub(super) fn is_delegation_event_append_failure(error: &EngineError) -> bool {
    matches!(
        error,
        EngineError::DelegationEvents(DelegationEventError::Session(
            crate::session::SessionError::Event(_)
        )) | EngineError::ActorStopped
    )
}

#[cfg(test)]
mod concurrency_tests {
    use super::{
        DELEGATED_CONTEXT_MAX_BYTES, DelegateTeaser, background_queue_limit_reached,
        cancelled_delegate_result, cancelled_delegate_result_with_reason,
        context_seed_from_history, delegate_failure_result, paginated_subagent_result,
        preview_is_truncated, preview_text, render_background_completion,
        render_delegate_teaser_body, steered_delegate_result, validate_redelivery_mode,
    };

    #[test]
    fn delegate_outputs_are_concise_and_preserve_metadata() {
        let session_id = cookie_agent_protocol::SessionId::new_v7();
        let cases = [
            (
                steered_delegate_result(session_id, None, "running"),
                format!("Subagent steered. [subagent session {session_id}; running]"),
                serde_json::json!({"session_id": session_id, "handle": null, "status": "running"}),
            ),
            (
                cancelled_delegate_result(session_id, None, Some("Work so far".into())),
                format!(
                    "Delegate cancelled. [subagent session {session_id}] Partial report: Work so far"
                ),
                serde_json::json!({"session_id": session_id, "handle": null, "status": "cancelled", "partial_report": "Work so far"}),
            ),
            (
                cancelled_delegate_result(session_id, None, None),
                format!("Delegate cancelled. [subagent session {session_id}]"),
                serde_json::json!({"session_id": session_id, "handle": null, "status": "cancelled", "partial_report": null}),
            ),
            (
                cancelled_delegate_result_with_reason(Some(session_id), None, "Parent stopped"),
                format!("Delegate cancelled: Parent stopped. [subagent session {session_id}]"),
                serde_json::json!({"session_id": session_id, "handle": null, "status": "cancelled", "reason": "Parent stopped"}),
            ),
            (
                cancelled_delegate_result_with_reason(None, None, "Parent stopped"),
                "Delegate cancelled: Parent stopped.".into(),
                serde_json::json!({"session_id": null, "handle": null, "status": "cancelled", "reason": "Parent stopped"}),
            ),
            (
                delegate_failure_result(Some(session_id), None, "Child unavailable"),
                format!("Delegate failed: Child unavailable. [subagent session {session_id}]"),
                serde_json::json!({"session_id": session_id, "handle": null, "status": "failed", "reason": "Child unavailable"}),
            ),
            (
                delegate_failure_result(None, None, "Child unavailable"),
                "Delegate failed: Child unavailable.".into(),
                serde_json::json!({"session_id": null, "handle": null, "status": "failed", "reason": "Child unavailable"}),
            ),
        ];
        for (result, output, metadata) in cases {
            assert_eq!(result.output, output);
            assert_eq!(result.metadata, metadata);
            assert!(serde_json::from_str::<serde_json::Value>(&result.output).is_err());
            assert!(result.truncation.is_none());
            assert!(result.attachments.is_empty());
            assert!(result.additional_messages.is_empty());
        }
    }

    #[test]
    fn delegate_outputs_surface_the_handle_in_place_of_the_uuid() {
        let session_id = cookie_agent_protocol::SessionId::new_v7();
        let handle = "explore_1a2b3c4d";
        let steered = steered_delegate_result(session_id, Some(handle), "running");
        assert_eq!(
            steered.output,
            "Subagent steered. [subagent session explore_1a2b3c4d; running]"
        );
        let cancelled = cancelled_delegate_result(session_id, Some(handle), None);
        assert_eq!(
            cancelled.output,
            "Delegate cancelled. [subagent session explore_1a2b3c4d]"
        );
        assert_eq!(
            cancelled.metadata["session_id"],
            serde_json::json!(session_id)
        );
        assert_eq!(cancelled.metadata["handle"], serde_json::json!(handle));
        let failed = delegate_failure_result(Some(session_id), Some(handle), "Child unavailable");
        assert_eq!(
            failed.output,
            "Delegate failed: Child unavailable. [subagent session explore_1a2b3c4d]"
        );
        assert!(!failed.output.contains(&session_id.to_string()));
    }

    #[test]
    fn background_notification_snapshot_matches_the_foreground_teaser_body() {
        let teaser = DelegateTeaser {
            session_id: cookie_agent_protocol::SessionId::new_v7(),
            short_id: "explore_1a2b3c4d".into(),
            status: cookie_agent_protocol::SessionStatus::Completed,
            preview: "first line\nsecond line".into(),
            total_lines: 79,
        };
        // Line breaks survive, exactly as in the foreground result.
        insta::assert_snapshot!(
            render_background_completion(&teaser),
            @r###"
<subagent_notification>first line
second line

[subagent session explore_1a2b3c4d; completed; 79 lines; use get_subagent_result with session_id "explore_1a2b3c4d" for the full output]</subagent_notification>
"###
        );
    }

    #[test]
    fn foreground_tool_result_text_snapshot_matches_the_shared_teaser_body() {
        // `terminal_delegate_result` renders exactly this body, so this is the
        // foreground `delegate_subagent` tool-result text.
        insta::assert_snapshot!(
            render_delegate_teaser_body(
                "first line\nsecond line",
                cookie_agent_protocol::SessionStatus::Completed,
                79,
                "explore_1a2b3c4d",
            ),
            @r###"
first line
second line

[subagent session explore_1a2b3c4d; completed; 79 lines; use get_subagent_result with session_id "explore_1a2b3c4d" for the full output]
"###
        );
    }

    #[test]
    fn delegate_display_fields_are_bounded_and_control_free_without_changing_metadata() {
        let session_id = cookie_agent_protocol::SessionId::new_v7();
        let text = format!("\x1b\n\t{}", "é".repeat(2048));
        for (result, key) in [
            (steered_delegate_result(session_id, None, &text), "status"),
            (
                cancelled_delegate_result(session_id, None, Some(text.clone())),
                "partial_report",
            ),
            (
                cancelled_delegate_result_with_reason(Some(session_id), None, &text),
                "reason",
            ),
            (
                delegate_failure_result(Some(session_id), None, &text),
                "reason",
            ),
        ] {
            assert_eq!(result.metadata[key], text);
            assert!(result.output.len() < 1200);
            assert!(!result.output.chars().any(char::is_control));
            assert_eq!(result.output.matches(&session_id.to_string()).count(), 1);
        }
    }

    #[test]
    fn background_queue_rejects_only_at_four_times_concurrency() {
        assert!(!background_queue_limit_reached(4, 15));
        assert!(background_queue_limit_reached(4, 16));
        assert!(background_queue_limit_reached(4, 17));
    }

    #[test]
    fn redelivery_execution_mode_must_match_in_both_directions() {
        assert!(validate_redelivery_mode(false, false).is_ok());
        assert!(validate_redelivery_mode(true, true).is_ok());
        let foreground_to_background = validate_redelivery_mode(false, true)
            .expect_err("foreground to background must fail")
            .to_string();
        assert!(foreground_to_background.contains("durable invocation is foreground"));
        assert!(foreground_to_background.contains("requested background"));
        let background_to_foreground = validate_redelivery_mode(true, false)
            .expect_err("background to foreground must fail")
            .to_string();
        assert!(background_to_foreground.contains("durable invocation is background"));
        assert!(background_to_foreground.contains("requested foreground"));
    }

    #[test]
    fn complete_preview_drops_the_full_output_hint() {
        insta::assert_snapshot!(
            render_delegate_teaser_body(
                "first line\nsecond line",
                cookie_agent_protocol::SessionStatus::Completed,
                2,
                "explore_1a2b3c4d",
            ),
            @r###"
first line
second line

[subagent session explore_1a2b3c4d; completed; 2 lines; full output shown]
"###
        );
    }

    #[test]
    fn preview_keeps_its_shape_but_not_other_control_characters() {
        let body = render_delegate_teaser_body(
            "- item\n\tindented\x1b[31m red\x07",
            cookie_agent_protocol::SessionStatus::Completed,
            2,
            "explore_1a2b3c4d",
        );
        assert!(body.starts_with("- item\n\tindented [31m red \n\n[subagent session"));
        let empty = render_delegate_teaser_body(
            "",
            cookie_agent_protocol::SessionStatus::Completed,
            0,
            "explore_1a2b3c4d",
        );
        assert!(empty.starts_with("(no output)\n\n[subagent session"));
    }

    #[test]
    fn truncation_is_judged_from_the_preview_the_output_produced() {
        let cases = [
            ("", false),
            ("one line", false),
            ("trailing blank\n\n", false),
            ("two\nlines", false),
        ]
        .into_iter()
        .map(|(text, cut)| (text.to_owned(), cut))
        .chain([
            // Line cap: 21 lines keep 20.
            (vec!["line"; 21].join("\n"), true),
            (vec!["line"; 20].join("\n"), false),
            // Byte cap inside the last kept line.
            ("é".repeat(1100), true),
            // A multibyte cut backs off below the cap and still counts.
            (format!("a{}", "€".repeat(700)), true),
        ]);
        for (text, cut) in cases {
            let preview = preview_text(&text);
            let total_lines = text.lines().count() as u64;
            assert_eq!(
                preview_is_truncated(&preview, total_lines),
                cut,
                "{} bytes, {total_lines} lines",
                text.len()
            );
        }
    }

    #[test]
    fn preview_is_line_and_utf8_bounded() {
        let text = (0..30)
            .map(|index| format!("{index}: {}", "é".repeat(200)))
            .collect::<Vec<_>>()
            .join("\n");
        let preview = preview_text(&text);
        assert!(preview.lines().count() <= 20);
        assert!(preview.len() <= 2048);
        assert!(std::str::from_utf8(preview.as_bytes()).is_ok());
    }

    #[test]
    fn result_pages_match_read_line_numbering_and_empty_running_shape() {
        let page = paginated_subagent_result("completed", "one\ntwo\nthree", 1, 1);
        assert_eq!(
            page.output,
            "<status>completed</status>\n<content>\n2: two\n</content>"
        );
        assert_eq!(page.metadata["total_lines"], 3);

        let running = paginated_subagent_result("running", "", 0, 20);
        assert_eq!(
            running.output,
            "<status>running</status>\n<content>\n</content>"
        );
        assert_eq!(running.metadata["total_lines"], 0);
    }

    #[test]
    fn inherited_context_is_text_only_and_truncates_oldest_bytes_first() {
        let history = vec![
            oven_sdk::HistoryTurn::system(oven_sdk::SystemMessage::new(vec![
                oven_sdk::SystemPart::Text(oven_sdk::TextPart::new("private system prompt")),
            ])),
            oven_sdk::HistoryTurn::user(oven_sdk::UserMessage::new(vec![
                oven_sdk::InputPart::Text(oven_sdk::TextPart::new("old-user".repeat(125))),
            ])),
            oven_sdk::HistoryTurn::tool(oven_sdk::ToolMessage::new(vec![
                oven_sdk::ToolResultPart::new(
                    "call",
                    oven_sdk::ToolContent::Text("secret tool output".into()),
                ),
            ])),
            oven_sdk::HistoryTurn::assistant(oven_sdk::CompletedTurn::new(
                oven_sdk::AssistantMessage::new(vec![
                    oven_sdk::AssistantPart::Text(oven_sdk::TextPart::new("n".repeat(70_000))),
                    oven_sdk::AssistantPart::ToolResult(oven_sdk::ToolResultPart::new(
                        "hosted",
                        oven_sdk::ToolContent::Text("hosted output".into()),
                    )),
                ]),
                oven_sdk::Finish::new(oven_sdk::Usage::default(), oven_sdk::FinishReason::Stop),
            )),
        ];
        let seed = context_seed_from_history(history);
        assert_eq!(seed.len(), 1);
        assert_eq!(
            seed[0].role,
            cookie_agent_protocol::DelegatedContextRole::Assistant
        );
        assert_eq!(seed[0].text.len(), DELEGATED_CONTEXT_MAX_BYTES);
        assert!(!seed[0].text.contains("old-user"));
        assert!(!seed[0].text.contains("tool output"));
    }
}