bamboo-engine 2026.10.22

//! Phase 2 SDK runner tests (S-T2.1 .. S-T2.5).
//!
//! These exercise the canonical spawn core (`run_child_spawn`) and the ergonomic
//! `ChildRunner` facade directly at the engine level, using a mock LLM provider
//! to avoid network I/O.

use std::collections::HashMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use futures::stream;
use tokio::sync::{broadcast, Notify, RwLock};

use bamboo_agent_core::storage::Storage;
use bamboo_agent_core::tools::{ToolCall, ToolError, ToolResult, ToolSchema};
use bamboo_agent_core::{AgentEvent, Message, Session};
use bamboo_domain::session::runtime_state::{
    AgentRuntimeState, AgentStatusState, ChildWaitPolicy, SuspensionState, WaitingForChildrenState,
};
use bamboo_domain::{
    SessionActivationError, SessionActivationPort, SessionInboxPort, SessionMessageEnvelope,
    SessionMessageId,
};
use bamboo_llm::{LLMChunk, LLMError, LLMProvider, LLMStream};

use crate::runtime::execution::spawn::{ExternalChildRunner, SpawnContext, SpawnJob};
use crate::sdk::runner::{ChildRunner, RunChildInput};
use crate::sdk::spawn::run_child_spawn;
use bamboo_metrics::MetricsCollector;
use bamboo_skills::SkillManager;

// ---------------------------------------------------------------------------
// Test doubles
// ---------------------------------------------------------------------------

/// Provider that returns a single token + Done, so the child loop completes.
struct CompletedProvider;

#[async_trait]
impl LLMProvider for CompletedProvider {
    async fn chat_stream(
        &self,
        _messages: &[Message],
        _tools: &[ToolSchema],
        _max_output_tokens: Option<u32>,
        _model: &str,
    ) -> Result<LLMStream, LLMError> {
        let items: Vec<bamboo_llm::provider::Result<LLMChunk>> =
            vec![Ok(LLMChunk::Token("done".to_string())), Ok(LLMChunk::Done)];
        Ok(Box::pin(stream::iter(items)))
    }
}

/// Holds a resumed activation inside its provider boundary so registration
/// races can inspect the published router owner and raw runner deterministically.
struct BlockingCompletedProvider {
    calls: Arc<AtomicUsize>,
    entered: Arc<Notify>,
    release: Arc<Notify>,
}

#[async_trait]
impl LLMProvider for BlockingCompletedProvider {
    async fn chat_stream(
        &self,
        _messages: &[Message],
        _tools: &[ToolSchema],
        _max_output_tokens: Option<u32>,
        _model: &str,
    ) -> Result<LLMStream, LLMError> {
        self.calls.fetch_add(1, Ordering::SeqCst);
        self.entered.notify_one();
        self.release.notified().await;
        let items: Vec<bamboo_llm::provider::Result<LLMChunk>> =
            vec![Ok(LLMChunk::Token("done".to_string())), Ok(LLMChunk::Done)];
        Ok(Box::pin(stream::iter(items)))
    }
}

/// Records the second production run's provider context. The terminal-race
/// envelope must be absent from the first call and present exactly once in the
/// successor call.
struct SuccessorRecordingProvider {
    calls: Arc<AtomicUsize>,
    successor_call: usize,
    expected_message_id: String,
    successor_seen: Arc<Notify>,
    successor_match_count: Arc<AtomicUsize>,
}

/// Observes the real parent execution launched after a completed child clears
/// a durable synchronous wait. The provider must see the hidden child result.
struct ChildOutcomeRecordingProvider {
    calls: Arc<AtomicUsize>,
    outcome_seen: Arc<Notify>,
    saw_final_response: Arc<std::sync::atomic::AtomicBool>,
}

#[async_trait]
impl LLMProvider for ChildOutcomeRecordingProvider {
    async fn chat_stream(
        &self,
        messages: &[Message],
        _tools: &[ToolSchema],
        _max_output_tokens: Option<u32>,
        _model: &str,
    ) -> Result<LLMStream, LLMError> {
        self.calls.fetch_add(1, Ordering::SeqCst);
        self.saw_final_response.store(
            messages.iter().any(|message| {
                message
                    .content
                    .contains("Child final response:\nfast child reply")
                    && message
                        .metadata
                        .as_ref()
                        .and_then(|metadata| metadata.get("runtime_kind"))
                        .and_then(|kind| kind.as_str())
                        == Some("child_completion_resume")
            }),
            Ordering::SeqCst,
        );
        self.outcome_seen.notify_one();
        Ok(Box::pin(stream::iter(vec![
            Ok(LLMChunk::Token("parent resumed".to_string())),
            Ok(LLMChunk::Done),
        ])))
    }
}

#[async_trait]
impl LLMProvider for SuccessorRecordingProvider {
    async fn chat_stream(
        &self,
        messages: &[Message],
        _tools: &[ToolSchema],
        _max_output_tokens: Option<u32>,
        _model: &str,
    ) -> Result<LLMStream, LLMError> {
        let call = self.calls.fetch_add(1, Ordering::SeqCst) + 1;
        let matches = messages
            .iter()
            .filter(|message| message.id == self.expected_message_id)
            .count();
        if call < self.successor_call {
            assert_eq!(
                matches, 0,
                "queued message must not reach a provider before its successor"
            );
        } else if call == self.successor_call {
            self.successor_match_count.store(matches, Ordering::SeqCst);
            self.successor_seen.notify_one();
        }
        let items: Vec<bamboo_llm::provider::Result<LLMChunk>> =
            vec![Ok(LLMChunk::Token("done".to_string())), Ok(LLMChunk::Done)];
        Ok(Box::pin(stream::iter(items)))
    }
}

/// Provider whose stream drips tokens forever without ever sending `Done`, so the
/// child runner never finishes on its own and the watchdog must terminate it.
///
/// The stream MUST keep yielding (rather than `stream::pending()`): the stream
/// consumer only checks the cancel token between chunks, mirroring real LLM
/// streams that emit periodic chunks.
struct HangingProvider;

#[async_trait]
impl LLMProvider for HangingProvider {
    async fn chat_stream(
        &self,
        _messages: &[Message],
        _tools: &[ToolSchema],
        _max_output_tokens: Option<u32>,
        _model: &str,
    ) -> Result<LLMStream, LLMError> {
        // A deliberately slow but *finite* stream: a handful of tokens spaced out,
        // then `Done`. Each LLM call (including any auxiliary calls that do not
        // thread the cancel token) terminates, but the main agent round takes long
        // enough that the child watchdog fires and cancels the run first — the
        // round-boundary cancel check then yields `AgentError::Cancelled`, which the
        // spawn core maps to a `timeout` terminal status (because `timeout_reason`
        // was set by the watchdog).
        let chunks: Vec<bamboo_llm::provider::Result<LLMChunk>> = vec![
            Ok(LLMChunk::Token("slow-1".to_string())),
            Ok(LLMChunk::Token("slow-2".to_string())),
            Ok(LLMChunk::Done),
        ];
        // Each chunk is paced well beyond the watchdog's total-timeout window, so
        // the watchdog deterministically fires (and cancels) while the run is still
        // streaming — never racing a fast natural completion.
        let s = stream::unfold(chunks.into_iter(), |mut it| async move {
            let next = it.next()?;
            tokio::time::sleep(Duration::from_secs(2)).await;
            Some((next, it))
        });
        Ok(Box::pin(s))
    }
}

/// Tool executor exposing a fixed catalog of tool names.
struct CatalogToolExecutor {
    names: Vec<String>,
}

#[async_trait]
impl bamboo_agent_core::tools::ToolExecutor for CatalogToolExecutor {
    async fn execute(&self, _call: &ToolCall) -> Result<ToolResult, ToolError> {
        Err(ToolError::NotFound("noop".to_string()))
    }

    fn list_tools(&self) -> Vec<ToolSchema> {
        self.names
            .iter()
            .map(|name| ToolSchema {
                schema_type: "function".to_string(),
                function: bamboo_agent_core::tools::FunctionSchema {
                    name: name.clone(),
                    description: String::new(),
                    parameters: serde_json::json!({}),
                },
            })
            .collect()
    }
}

/// Test-only child runner that drives the child loop in-process. The
/// production in-process spawn path was removed (sub-agents always run as
/// actors), but `run_child_spawn`'s orchestration — event ordering, watchdog,
/// completion/status mapping, persistence — is runner-agnostic. This stand-in
/// keeps that orchestration under fast, network-free unit test without spinning
/// up a real actor worker subprocess.
struct InProcessTestRunner {
    agent: Arc<crate::Agent>,
    tools: Arc<dyn bamboo_agent_core::tools::ToolExecutor>,
}

#[async_trait]
impl ExternalChildRunner for InProcessTestRunner {
    async fn should_handle(&self, _session: &Session) -> bool {
        true
    }

    async fn execute_external_child(
        &self,
        session: &mut Session,
        job: &SpawnJob,
        event_tx: tokio::sync::mpsc::Sender<AgentEvent>,
        cancel_token: tokio_util::sync::CancellationToken,
    ) -> crate::runtime::runner::Result<()> {
        let mut builder =
            crate::runtime::ExecuteRequestBuilder::new(String::new(), event_tx, cancel_token)
                .tools(self.tools.clone())
                .model_roster(crate::runtime::ModelRoster {
                    model: Some(job.model.clone()),
                    provider_name: None,
                    provider_type: None,
                    fast: None,
                    background: None,
                    summarization: None,
                });
        if let Some(disabled) = &job.disabled_tools {
            builder = builder.disabled_tools(disabled.iter().cloned().collect());
        }
        self.agent.execute(session, builder.build()).await
    }
}

struct TerminalBarrierRunner {
    inner: InProcessTestRunner,
    completed_runs: AtomicUsize,
    durable_confirmations: Arc<AtomicUsize>,
    first_execute_returned: Arc<Notify>,
    release_first_terminal: Arc<Notify>,
}

#[async_trait]
impl ExternalChildRunner for TerminalBarrierRunner {
    async fn should_handle(&self, session: &Session) -> bool {
        self.inner.should_handle(session).await
    }

    async fn execute_external_child(
        &self,
        session: &mut Session,
        job: &SpawnJob,
        event_tx: tokio::sync::mpsc::Sender<AgentEvent>,
        cancel_token: tokio_util::sync::CancellationToken,
    ) -> crate::runtime::runner::Result<()> {
        let result = self
            .inner
            .execute_external_child(session, job, event_tx, cancel_token)
            .await;
        let completed = self.completed_runs.fetch_add(1, Ordering::SeqCst);
        if completed == 0 {
            // The final provider boundary has completed, while run_child_spawn's
            // real terminal block has not begun.
            self.first_execute_returned.notify_one();
            self.release_first_terminal.notified().await;
        }
        result
    }

    async fn confirm_durable_child_delivery(
        &self,
        session: &Session,
        _activation_run_id: &str,
        save_succeeded: bool,
    ) -> Result<(), String> {
        if !save_succeeded {
            return Ok(());
        }
        let saved = self
            .inner
            .agent
            .storage()
            .load_session(&session.id)
            .await
            .map_err(|error| error.to_string())?
            .ok_or_else(|| "terminal Child snapshot missing before ACK callback".to_string())?;
        if saved.created_at != session.created_at
            || saved.last_run_status() != session.last_run_status()
            || saved.messages.last().map(|message| &message.content)
                != session.messages.last().map(|message| &message.content)
        {
            return Err("terminal Child snapshot was not durable before ACK callback".into());
        }
        self.durable_confirmations.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }
}

struct GatedCoordinatorSpawner {
    inner: Arc<crate::ChildCompletionCoordinator>,
    reservations: Arc<AtomicUsize>,
    reservation_entered: Arc<Notify>,
    allow_reservation: Arc<Notify>,
}

#[async_trait]
impl crate::SessionActivationSpawner for GatedCoordinatorSpawner {
    async fn reserve_activation(
        &self,
        target_session_id: &str,
        inbox_generation: u64,
    ) -> Result<crate::SessionActivationReserveOutcome, SessionActivationError> {
        self.reservations.fetch_add(1, Ordering::SeqCst);
        self.reservation_entered.notify_one();
        self.allow_reservation.notified().await;
        crate::SessionActivationSpawner::reserve_activation(
            self.inner.as_ref(),
            target_session_id,
            inbox_generation,
        )
        .await
    }
}

/// Fails exactly the first transcript checkpoint that contains the selected
/// typed inbox message. Legacy queue cleanup and terminal persistence still
/// use the real locked store, reproducing the boundary failure where the
/// compatibility source has already been CAS-cleared but the typed claim has
/// not yet been admitted.
struct FailFirstInboxCheckpointPersistence {
    inner: Arc<bamboo_storage::LockedSessionStore>,
    message_id: String,
    failures: AtomicUsize,
}

#[async_trait]
impl bamboo_domain::RuntimeSessionPersistence for FailFirstInboxCheckpointPersistence {
    async fn save_runtime_session(&self, session: &mut Session) -> std::io::Result<()> {
        bamboo_domain::RuntimeSessionPersistence::save_runtime_session(self.inner.as_ref(), session)
            .await
    }

    async fn checkpoint_runtime_session(&self, session: &mut Session) -> std::io::Result<()> {
        if session
            .messages
            .iter()
            .any(|message| message.id == self.message_id)
            && self.failures.fetch_add(1, Ordering::SeqCst) == 0
        {
            return Err(std::io::Error::other(
                "injected first SessionInbox transcript checkpoint failure",
            ));
        }
        bamboo_domain::RuntimeSessionPersistence::checkpoint_runtime_session(
            self.inner.as_ref(),
            session,
        )
        .await
    }

    async fn load_runtime_session(&self, session_id: &str) -> std::io::Result<Option<Session>> {
        bamboo_domain::RuntimeSessionPersistence::load_runtime_session(
            self.inner.as_ref(),
            session_id,
        )
        .await
    }

    async fn clear_legacy_pending_messages(
        &self,
        session_id: &str,
        expected: &[serde_json::Value],
    ) -> std::io::Result<bool> {
        bamboo_domain::RuntimeSessionPersistence::clear_legacy_pending_messages(
            self.inner.as_ref(),
            session_id,
            expected,
        )
        .await
    }

    async fn append_token_usage_record(
        &self,
        session_id: &str,
        json_line: &str,
    ) -> std::io::Result<()> {
        bamboo_domain::RuntimeSessionPersistence::append_token_usage_record(
            self.inner.as_ref(),
            session_id,
            json_line,
        )
        .await
    }
}

struct Harness {
    ctx: SpawnContext,
    storage: Arc<dyn Storage>,
    parent_session_id: String,
    child_session_id: String,
    parent_rx: broadcast::Receiver<AgentEvent>,
    parent_tx: broadcast::Sender<AgentEvent>,
}

struct DeliveryHarness {
    ctx: SpawnContext,
    storage: Arc<dyn Storage>,
    inbox: Arc<dyn SessionInboxPort>,
    messenger: Arc<crate::SessionMessenger>,
    coordinator: Arc<crate::ChildCompletionCoordinator>,
    parent_session_id: String,
    child_session_id: String,
    parent_rx: broadcast::Receiver<AgentEvent>,
    agent_runners: Arc<RwLock<HashMap<String, crate::execution::AgentRunner>>>,
    first_execute_returned: Arc<Notify>,
    release_first_terminal: Arc<Notify>,
    reservation_entered: Arc<Notify>,
    allow_reservation: Arc<Notify>,
    reservations: Arc<AtomicUsize>,
    durable_confirmations: Arc<AtomicUsize>,
    _spawn_scheduler: Arc<crate::runtime::execution::spawn::SpawnScheduler>,
}

fn temp_dir(prefix: &str) -> std::path::PathBuf {
    std::env::temp_dir().join(format!("{prefix}-{}", uuid::Uuid::new_v4()))
}

async fn build_harness(
    provider: Arc<dyn LLMProvider>,
    tool_names: Vec<String>,
    child_metadata: &[(&str, &str)],
) -> Harness {
    let home = temp_dir("bamboo-sdk-test");
    tokio::fs::create_dir_all(&home).await.unwrap();

    let session_store = Arc::new(
        bamboo_storage::SessionStoreV2::new(home.clone())
            .await
            .unwrap(),
    );
    let storage_dir = home.join("storage");
    tokio::fs::create_dir_all(&storage_dir).await.unwrap();
    let jsonl = bamboo_storage::JsonlStorage::new(&storage_dir);
    jsonl.init().await.unwrap();
    let storage: Arc<dyn Storage> = Arc::new(jsonl);

    let metrics_storage = Arc::new(bamboo_metrics::SqliteMetricsStorage::new(
        home.join("metrics.db"),
    ));
    let metrics_collector = MetricsCollector::spawn(metrics_storage, 7);

    let sessions_cache: crate::SessionCache = Arc::default();
    let agent_runners = Arc::new(RwLock::new(HashMap::new()));
    let session_event_senders = Arc::new(RwLock::new(HashMap::<
        String,
        broadcast::Sender<AgentEvent>,
    >::new()));

    let parent_session_id = "root-session".to_string();
    let child_session_id = "child-session".to_string();
    let (parent_tx, parent_rx) = broadcast::channel(1000);
    {
        let mut senders = session_event_senders.write().await;
        senders.insert(parent_session_id.clone(), parent_tx.clone());
    }

    let mut parent = Session::new(parent_session_id.clone(), "gpt-5");
    parent.title = "Root".to_string();
    storage.save_session(&parent).await.unwrap();

    let mut child = Session::new_child(
        child_session_id.clone(),
        parent_session_id.clone(),
        "gpt-5",
        "Child session",
    );
    child
        .metadata
        .insert("last_run_status".to_string(), "queued".to_string());
    for (k, v) in child_metadata {
        child.metadata.insert(k.to_string(), v.to_string());
    }
    child.add_message(Message::system("child system"));
    child.add_message(Message::user("do the task"));
    storage.save_session(&child).await.unwrap();

    let agent = Arc::new(
        crate::Agent::builder()
            .storage(storage.clone())
            .persistence(Arc::new(bamboo_storage::LockedSessionStore::new(
                storage.clone(),
            )))
            .attachment_reader(session_store.clone())
            .skill_manager(Arc::new(SkillManager::new()))
            .metrics_collector(metrics_collector)
            .config(Arc::new(RwLock::new(bamboo_llm::Config::default())))
            .provider(provider)
            .default_tools(Arc::new(CatalogToolExecutor {
                names: tool_names.clone(),
            }))
            .build()
            .expect("test agent should build"),
    );

    let tools: Arc<dyn bamboo_agent_core::tools::ToolExecutor> =
        Arc::new(CatalogToolExecutor { names: tool_names });

    let ctx = SpawnContext {
        agent: agent.clone(),
        tools: tools.clone(),
        sessions_cache,
        agent_runners,
        session_event_senders,
        external_child_runner: Arc::new(InProcessTestRunner { agent, tools }),
        provider_router: None,
        app_data_dir: None,
        completion_handler: None,
        child_run_launch_hook: None,
        account_feed_inbox: None,
    };

    Harness {
        ctx,
        storage,
        parent_session_id,
        child_session_id,
        parent_rx,
        parent_tx,
    }
}

async fn build_terminal_delivery_harness(
    provider: Arc<dyn LLMProvider>,
    legacy_pending: Option<serde_json::Value>,
    fail_first_checkpoint_for: Option<String>,
) -> DeliveryHarness {
    let home = temp_dir("bamboo-sdk-terminal-inbox-test");
    tokio::fs::create_dir_all(&home).await.unwrap();
    let store = Arc::new(
        bamboo_storage::SessionStoreV2::new(home.clone())
            .await
            .unwrap(),
    );
    let storage: Arc<dyn Storage> = store.clone();
    let locked = Arc::new(bamboo_storage::LockedSessionStore::new(storage.clone()));
    let persistence: Arc<dyn bamboo_domain::RuntimeSessionPersistence> =
        match fail_first_checkpoint_for {
            Some(message_id) => Arc::new(FailFirstInboxCheckpointPersistence {
                inner: locked.clone(),
                message_id,
                failures: AtomicUsize::new(0),
            }),
            None => locked.clone(),
        };
    let inbox: Arc<dyn SessionInboxPort> = Arc::new(bamboo_storage::FileSessionInbox::new(
        store.clone(),
        bamboo_domain::SessionInboxLimits::default(),
    ));
    let router = crate::SessionActivationRouter::new();
    let activation: Arc<dyn SessionActivationPort> = router.clone();
    let messenger = Arc::new(crate::SessionMessenger::new(
        storage.clone(),
        inbox.clone(),
        activation,
    ));

    let metrics_storage = Arc::new(bamboo_metrics::SqliteMetricsStorage::new(
        home.join("metrics.db"),
    ));
    let metrics_collector = MetricsCollector::spawn(metrics_storage, 7);
    let tools: Arc<dyn bamboo_agent_core::tools::ToolExecutor> =
        Arc::new(CatalogToolExecutor { names: Vec::new() });
    let mut config_value = bamboo_llm::Config::default();
    config_value.provider = "test".to_string();
    let config = Arc::new(RwLock::new(config_value));
    let mut providers = HashMap::new();
    providers.insert("test".to_string(), provider.clone());
    let provider_registry = Arc::new(bamboo_llm::ProviderRegistry::new(
        providers,
        "test".to_string(),
    ));
    let provider_router = Arc::new(bamboo_llm::ProviderModelRouter::new(
        provider_registry.clone(),
    ));
    let agent = Arc::new(
        crate::Agent::builder()
            .storage(storage.clone())
            .persistence(persistence)
            .session_inbox(inbox.clone())
            .activation_router(router.clone())
            .session_messenger(messenger.clone())
            .attachment_reader(store)
            .skill_manager(Arc::new(SkillManager::new()))
            .metrics_collector(metrics_collector)
            .config(config.clone())
            .provider(provider)
            .default_tools(tools.clone())
            .build()
            .expect("delivery test agent"),
    );

    let sessions_cache: crate::SessionCache = Arc::default();
    let agent_runners = Arc::new(RwLock::new(HashMap::new()));
    let session_event_senders = Arc::new(RwLock::new(HashMap::<
        String,
        broadcast::Sender<AgentEvent>,
    >::new()));
    let parent_session_id = "terminal-root".to_string();
    let child_session_id = "terminal-child".to_string();
    let (parent_tx, parent_rx) = broadcast::channel(1000);
    session_event_senders
        .write()
        .await
        .insert(parent_session_id.clone(), parent_tx);

    let parent = Session::new(&parent_session_id, "gpt-5");
    storage.save_session(&parent).await.unwrap();
    let mut child = Session::new_child(
        &child_session_id,
        &parent_session_id,
        "gpt-5",
        "Terminal child",
    );
    child.add_message(Message::system("child system"));
    child.add_message(Message::user("first task"));
    if let Some(message) = legacy_pending {
        child.set_pending_injected_messages(vec![message]);
    }
    storage.save_session(&child).await.unwrap();

    let first_execute_returned = Arc::new(Notify::new());
    let release_first_terminal = Arc::new(Notify::new());
    let durable_confirmations = Arc::new(AtomicUsize::new(0));
    let external_child_runner: Arc<dyn ExternalChildRunner> = Arc::new(TerminalBarrierRunner {
        inner: InProcessTestRunner {
            agent: agent.clone(),
            tools: tools.clone(),
        },
        completed_runs: AtomicUsize::new(0),
        durable_confirmations: durable_confirmations.clone(),
        first_execute_returned: first_execute_returned.clone(),
        release_first_terminal: release_first_terminal.clone(),
    });
    let ctx = SpawnContext {
        agent: agent.clone(),
        tools: tools.clone(),
        sessions_cache: sessions_cache.clone(),
        agent_runners: agent_runners.clone(),
        session_event_senders: session_event_senders.clone(),
        external_child_runner,
        provider_router: None,
        app_data_dir: None,
        completion_handler: None,
        child_run_launch_hook: None,
        account_feed_inbox: None,
    };
    let reservations = Arc::new(AtomicUsize::new(0));
    let reservation_entered = Arc::new(Notify::new());
    let allow_reservation = Arc::new(Notify::new());
    let coordinator = Arc::new(crate::ChildCompletionCoordinator::new(
        storage.clone(),
        locked,
        sessions_cache,
        agent_runners,
        session_event_senders,
        agent,
        config,
        provider_registry,
        provider_router,
        home,
        None,
    ));
    coordinator.set_root_tools(tools).await;
    let spawn_scheduler = Arc::new(crate::runtime::execution::spawn::SpawnScheduler::new(
        ctx.clone(),
    ));
    coordinator.set_spawn_scheduler(&spawn_scheduler).await;
    router
        .set_spawner(Arc::new(GatedCoordinatorSpawner {
            inner: coordinator.clone(),
            reservations: reservations.clone(),
            reservation_entered: reservation_entered.clone(),
            allow_reservation: allow_reservation.clone(),
        }))
        .await;

    let delivery_runners = ctx.agent_runners.clone();
    DeliveryHarness {
        ctx,
        storage,
        inbox,
        messenger,
        coordinator,
        parent_session_id,
        child_session_id,
        parent_rx,
        agent_runners: delivery_runners,
        first_execute_returned,
        release_first_terminal,
        reservation_entered,
        allow_reservation,
        reservations,
        durable_confirmations,
        _spawn_scheduler: spawn_scheduler,
    }
}

/// Collect parent events until a `SubAgentCompleted` is observed (or timeout).
async fn collect_until_completed(rx: &mut broadcast::Receiver<AgentEvent>) -> Vec<AgentEvent> {
    collect_until_completed_with_budget(rx, Duration::from_secs(15)).await
}

/// Like [`collect_until_completed`] but with a caller-supplied budget. The
/// watchdog timeout path uses wall-clock time, which can stretch under heavy
/// parallel test load, so the timeout test grants a wider budget.
async fn collect_until_completed_with_budget(
    rx: &mut broadcast::Receiver<AgentEvent>,
    budget: Duration,
) -> Vec<AgentEvent> {
    let mut events = Vec::new();
    let deadline = tokio::time::Instant::now() + budget;
    loop {
        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
        if remaining.is_zero() {
            panic!("timed out waiting for SubAgentCompleted; saw: {events:?}");
        }
        match tokio::time::timeout(remaining, rx.recv()).await {
            Ok(Ok(event)) => {
                let done = matches!(event, AgentEvent::SubAgentCompleted { .. });
                events.push(event);
                if done {
                    return events;
                }
            }
            Ok(Err(broadcast::error::RecvError::Lagged(_))) => continue,
            Ok(Err(broadcast::error::RecvError::Closed)) => {
                panic!("parent channel closed before completion; saw: {events:?}")
            }
            Err(_) => panic!("timed out waiting for SubAgentCompleted; saw: {events:?}"),
        }
    }
}

#[tokio::test]
async fn queue_commit_then_producer_abort_publishes_start_once_before_completion() {
    let mut harness = build_harness(Arc::new(CompletedProvider), vec![], &[]).await;
    let scheduler = crate::runtime::execution::spawn::SpawnScheduler::new(harness.ctx.clone());
    // `prepare_child_launch` needs this map. Holding it parks the producer
    // after queue admission but before Start publication.
    let held_senders = harness.ctx.session_event_senders.write().await;
    let gate = Arc::new(bamboo_domain::AdmissionGate::default());
    let job = SpawnJob {
        parent_session_id: harness.parent_session_id.clone(),
        child_session_id: harness.child_session_id.clone(),
        model: "gpt-5".to_string(),
        disabled_tools: None,
    };
    let producer = {
        let gate = gate.clone();
        tokio::spawn(async move {
            scheduler
                .enqueue_announced_with_gate(job, Some("Child session".into()), Some(&gate))
                .await
        })
    };
    tokio::time::timeout(Duration::from_secs(5), async {
        while !gate.is_committed() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("queue entry must commit before aborting producer");
    producer.abort();
    assert!(producer.await.unwrap_err().is_cancelled());
    drop(held_senders);

    let events = collect_until_completed(&mut harness.parent_rx).await;
    let starts = events
        .iter()
        .filter(|event| {
            matches!(
                event,
                AgentEvent::SubAgentStarted { child_session_id, .. }
                    if child_session_id == &harness.child_session_id
            )
        })
        .count();
    let completed = events.iter().position(|event| {
        matches!(
            event,
            AgentEvent::SubAgentCompleted { child_session_id, .. }
                if child_session_id == &harness.child_session_id
        )
    });
    assert_eq!(
        starts, 1,
        "the worker must publish one fallback Start: {events:?}"
    );
    assert!(completed.is_some_and(|index| index > 0), "{events:?}");
}

// ---------------------------------------------------------------------------
// S-T2.1 — run_child_spawn integration: ordering + persisted status
// ---------------------------------------------------------------------------

#[tokio::test]
async fn distinct_router_collision_is_a_terminal_setup_error_not_a_phantom_run() {
    let mut harness =
        build_terminal_delivery_harness(Arc::new(CompletedProvider), None, None).await;
    let workspace_before =
        bamboo_agent_core::workspace_state::peek_workspace(&harness.child_session_id);
    let rejected_workspace =
        temp_dir("bamboo-rejected-child-workspace").join("must-not-materialize");
    let mut child_with_rejected_workspace = harness
        .storage
        .load_session(&harness.child_session_id)
        .await
        .unwrap()
        .unwrap();
    child_with_rejected_workspace.workspace =
        Some(rejected_workspace.to_string_lossy().to_string());
    harness
        .storage
        .save_session(&child_with_rejected_workspace)
        .await
        .unwrap();

    let router = harness
        .ctx
        .agent
        .activation_router()
        .expect("delivery harness has an activation router")
        .clone();
    let mut existing_owner = router
        .register_run(&harness.child_session_id, "independent-owner")
        .await
        .unwrap();

    let error = run_child_spawn(
        harness.ctx.clone(),
        SpawnJob {
            parent_session_id: harness.parent_session_id.clone(),
            child_session_id: harness.child_session_id.clone(),
            model: "gpt-5".to_string(),
            disabled_tools: None,
        },
    )
    .await
    .expect_err("a distinct logical-session owner must reject child setup");
    assert!(error.contains("owner collision"), "{error}");
    assert!(
        router
            .owns_run(&harness.child_session_id, "independent-owner")
            .await
    );

    let completion = tokio::time::timeout(Duration::from_secs(2), harness.parent_rx.recv())
        .await
        .expect("the waiting parent must be unblocked immediately")
        .expect("parent event channel remains open");
    match completion {
        AgentEvent::SubAgentCompleted {
            child_session_id,
            status,
            error: Some(completion_error),
            ..
        } => {
            assert_eq!(child_session_id, harness.child_session_id);
            assert_eq!(status, "error");
            assert!(completion_error.contains("owner collision"));
        }
        other => panic!("expected terminal collision completion, got {other:?}"),
    }

    let persisted = harness
        .storage
        .load_session(&harness.child_session_id)
        .await
        .unwrap()
        .unwrap();
    let persisted_status = persisted.last_run_status();
    assert_ne!(
        persisted_status.as_deref(),
        Some("running"),
        "rejected setup must not persist a phantom running marker"
    );
    assert_eq!(
        bamboo_agent_core::workspace_state::peek_workspace(&harness.child_session_id),
        workspace_before,
        "rejected setup must not overwrite process-global workspace ownership"
    );
    assert!(
        !rejected_workspace.exists(),
        "rejected setup must not materialize its proposed workspace"
    );
    assert!(matches!(
        harness
            .agent_runners
            .read()
            .await
            .get(&harness.child_session_id)
            .map(|runner| &runner.status),
        Some(crate::execution::AgentStatus::Error(_))
    ));

    existing_owner.begin_finalization().await;
    assert_eq!(existing_owner.finish(0).await.unwrap(), None);
}

#[tokio::test]
async fn s_t2_1_run_child_spawn_emits_started_event_completed_in_order() {
    let mut harness = build_harness(Arc::new(CompletedProvider), Vec::new(), &[]).await;
    let child_tx = crate::runtime::execution::session_events::get_or_create_event_sender(
        &harness.ctx.session_event_senders,
        &harness.child_session_id,
    )
    .await;
    let mut child_rx = child_tx.subscribe();

    // The adapter emits SubAgentStarted before enqueue; simulate that here so the
    // full ordering can be asserted.
    harness
        .parent_tx
        .send(AgentEvent::SubAgentStarted {
            parent_session_id: harness.parent_session_id.clone(),
            child_session_id: harness.child_session_id.clone(),
            title: Some("Child session".to_string()),
        })
        .unwrap();

    run_child_spawn(
        harness.ctx.clone(),
        SpawnJob {
            parent_session_id: harness.parent_session_id.clone(),
            child_session_id: harness.child_session_id.clone(),
            model: "gpt-5".to_string(),
            disabled_tools: None,
        },
    )
    .await
    .unwrap();

    let events = collect_until_completed(&mut harness.parent_rx).await;

    let started_idx = events
        .iter()
        .position(|e| matches!(e, AgentEvent::SubAgentStarted { .. }))
        .expect("SubAgentStarted present");
    let completed_idx = events
        .iter()
        .position(|e| matches!(e, AgentEvent::SubAgentCompleted { .. }))
        .expect("SubAgentCompleted present");
    assert!(
        started_idx < completed_idx,
        "Started must precede Completed: {events:?}"
    );
    assert!(
        events
            .iter()
            .all(|event| !matches!(event, AgentEvent::SubAgentEvent { .. })),
        "raw child events belong only to the child's independently subscribable channel"
    );
    let mut child_events = Vec::new();
    while let Ok(event) = child_rx.try_recv() {
        child_events.push(event);
    }
    assert!(
        !child_events.is_empty(),
        "the child session must retain its independently subscribable event stream"
    );
    assert!(
        child_events
            .iter()
            .any(|event| matches!(event, AgentEvent::Token { .. })),
        "the child channel must retain full-fidelity token events"
    );

    match events.last().unwrap() {
        AgentEvent::SubAgentCompleted { status, .. } => {
            assert_eq!(status, "completed", "child should finish completed");
        }
        other => panic!("last event must be SubAgentCompleted, got {other:?}"),
    }

    // Persisted child status reflects completion.
    let persisted = harness
        .storage
        .load_session(&harness.child_session_id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(
        persisted
            .metadata
            .get("last_run_status")
            .map(String::as_str),
        Some("completed")
    );
}

#[tokio::test]
async fn activation_watermark_wait_gap_defers_then_reserves_same_generation_once() {
    let harness = build_terminal_delivery_harness(Arc::new(CompletedProvider), None, None).await;
    let now = chrono::Utc::now();
    let mut parent = harness
        .storage
        .load_session(&harness.parent_session_id)
        .await
        .unwrap()
        .unwrap();
    let mut runtime = AgentRuntimeState::new("watermark-wait-gap");
    runtime.status = AgentStatusState::Suspended;
    runtime.suspension = Some(SuspensionState {
        reason: "waiting_for_children".to_string(),
        suspended_at: now,
        resumable: true,
        hook_point: None,
    });
    runtime.waiting_for_children = Some(WaitingForChildrenState::for_children(
        vec!["still-running-child".to_string()],
        ChildWaitPolicy::All,
        now,
    ));
    parent.agent_runtime_state = Some(runtime.clone());
    parent.metadata.insert(
        "agent.runtime.state".to_string(),
        serde_json::to_string(&runtime).unwrap(),
    );
    parent.metadata.insert(
        "runtime.suspend_reason".to_string(),
        "waiting_for_children".to_string(),
    );
    harness.storage.save_session(&parent).await.unwrap();

    let admission = harness
        .messenger
        .admit(SessionMessageEnvelope::user_input(
            &harness.parent_session_id,
            "durable work waiting behind a child barrier",
        ))
        .await
        .unwrap();
    harness
        .messenger
        .prepare_activation(&admission)
        .await
        .unwrap();
    let generation = admission.delivery.generation;

    let deferred = crate::SessionActivationSpawner::reserve_activation(
        harness.coordinator.as_ref(),
        &harness.parent_session_id,
        generation,
    )
    .await
    .unwrap();
    assert!(matches!(
        deferred,
        crate::SessionActivationReserveOutcome::NoWork
    ));
    assert!(!harness
        .agent_runners
        .read()
        .await
        .contains_key(&harness.parent_session_id));
    let still_waiting = harness
        .storage
        .load_session(&harness.parent_session_id)
        .await
        .unwrap()
        .unwrap()
        .agent_runtime_state
        .unwrap();
    assert!(still_waiting.waiting_for_children.is_some());
    assert_eq!(still_waiting.status, AgentStatusState::Suspended);

    let mut cleared = harness
        .storage
        .load_session(&harness.parent_session_id)
        .await
        .unwrap()
        .unwrap();
    let mut cleared_runtime = cleared.agent_runtime_state.clone().unwrap();
    cleared_runtime.waiting_for_children = None;
    cleared_runtime.status = AgentStatusState::Idle;
    cleared_runtime.suspension = None;
    cleared.agent_runtime_state = Some(cleared_runtime.clone());
    cleared.metadata.insert(
        "agent.runtime.state".to_string(),
        serde_json::to_string(&cleared_runtime).unwrap(),
    );
    cleared.metadata.remove("runtime.suspend_reason");
    harness.storage.save_session(&cleared).await.unwrap();

    let reserved = crate::SessionActivationSpawner::reserve_activation(
        harness.coordinator.as_ref(),
        &harness.parent_session_id,
        generation,
    )
    .await
    .unwrap();
    let crate::SessionActivationReserveOutcome::Reserved(launch) = reserved else {
        panic!("cleared wait must reserve the same durable generation");
    };
    let run_id = launch.run_id.clone();
    assert_eq!(
        harness
            .agent_runners
            .read()
            .await
            .get(&harness.parent_session_id)
            .map(|runner| runner.run_id.as_str()),
        Some(run_id.as_str())
    );

    let duplicate = crate::SessionActivationSpawner::reserve_activation(
        harness.coordinator.as_ref(),
        &harness.parent_session_id,
        generation,
    )
    .await
    .unwrap();
    assert!(matches!(
        duplicate,
        crate::SessionActivationReserveOutcome::AlreadyRunning {
            run_id: duplicate_run_id
        } if duplicate_run_id == run_id
    ));

    drop(launch);
    tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            if !harness
                .agent_runners
                .read()
                .await
                .contains_key(&harness.parent_session_id)
            {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("dropping an unpublished activation must roll back its reservation");
}

#[tokio::test]
async fn fast_child_completion_automatically_starts_parent_successor() {
    use crate::execution::{ChildCompletion, ChildCompletionHandler};

    let calls = Arc::new(AtomicUsize::new(0));
    let outcome_seen = Arc::new(Notify::new());
    let saw_final_response = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let provider: Arc<dyn LLMProvider> = Arc::new(ChildOutcomeRecordingProvider {
        calls: calls.clone(),
        outcome_seen: outcome_seen.clone(),
        saw_final_response: saw_final_response.clone(),
    });
    let harness = build_terminal_delivery_harness(provider, None, None).await;
    let now = chrono::Utc::now();
    let mut parent = harness
        .storage
        .load_session(&harness.parent_session_id)
        .await
        .unwrap()
        .unwrap();
    let mut runtime = AgentRuntimeState::new("parent-wait-fast-child");
    runtime.status = AgentStatusState::Suspended;
    runtime.suspension = Some(SuspensionState {
        reason: "waiting_for_children".to_string(),
        suspended_at: now,
        resumable: true,
        hook_point: None,
    });
    let mut wait = WaitingForChildrenState::for_children(
        vec![harness.child_session_id.clone()],
        ChildWaitPolicy::All,
        now,
    );
    wait.registered_by_tool_call_id = Some("fast-child-launch".to_string());
    runtime.waiting_for_children = Some(wait);
    parent.agent_runtime_state = Some(runtime);
    parent.metadata.insert(
        "runtime.suspend_reason".to_string(),
        "waiting_for_children".to_string(),
    );
    parent.set_last_run_status("suspended");
    harness.storage.save_session(&parent).await.unwrap();

    let mut child = harness
        .storage
        .load_session(&harness.child_session_id)
        .await
        .unwrap()
        .unwrap();
    child.add_message(Message::assistant("fast child reply", None));
    child.set_last_run_status("completed");
    crate::execution::ChildCompletionSource::prepare(
        &mut child,
        "fast-child-run",
        &Default::default(),
    );
    harness.storage.save_session(&child).await.unwrap();

    let completion = ChildCompletion {
        parent_session_id: harness.parent_session_id.clone(),
        child_session_id: harness.child_session_id.clone(),
        status: "completed".to_string(),
        error: None,
        completed_at: now,
        source: crate::execution::ChildCompletionSource::from_committed_session(&child),
    };
    let coordinator = harness.coordinator.clone();
    let completion_task = tokio::spawn(async move {
        coordinator.on_child_completed(completion).await;
    });

    tokio::time::timeout(
        Duration::from_secs(5),
        harness.reservation_entered.notified(),
    )
    .await
    .expect("child completion must request a real parent successor");
    let cleared = harness
        .storage
        .load_session(&harness.parent_session_id)
        .await
        .unwrap()
        .unwrap();
    assert!(cleared
        .agent_runtime_state
        .as_ref()
        .and_then(|state| state.waiting_for_children.as_ref())
        .is_none());
    assert!(!cleared.metadata.contains_key("runtime.suspend_reason"));
    assert_eq!(harness.reservations.load(Ordering::SeqCst), 1);

    harness.allow_reservation.notify_one();
    completion_task.await.unwrap();
    tokio::time::timeout(Duration::from_secs(10), outcome_seen.notified())
        .await
        .expect("the parent successor must reach its provider without a watchdog");
    assert!(saw_final_response.load(Ordering::SeqCst));
    tokio::time::timeout(Duration::from_secs(10), async {
        loop {
            let settled = harness
                .agent_runners
                .read()
                .await
                .get(&harness.parent_session_id)
                .is_some_and(|runner| {
                    !matches!(runner.status, crate::execution::AgentStatus::Running)
                });
            if settled {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("parent successor must finish through the real runner lifecycle");
    let parent = harness
        .storage
        .load_session(&harness.parent_session_id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(parent.last_run_status().as_deref(), Some("completed"));
    assert!(parent
        .messages
        .iter()
        .any(|message| message.content == "parent resumed"));
    let backlog = harness
        .inbox
        .inspect(&harness.parent_session_id)
        .await
        .unwrap();
    assert_eq!(backlog.pending + backlog.claimed, 0);
    assert_eq!(calls.load(Ordering::SeqCst), 1);
}

#[tokio::test]
async fn activation_already_running_never_replaces_live_session_cache_arc() {
    let harness = build_terminal_delivery_harness(Arc::new(CompletedProvider), None, None).await;
    let admission = harness
        .messenger
        .admit(SessionMessageEnvelope::user_input(
            &harness.parent_session_id,
            "activation racing a manual owner",
        ))
        .await
        .unwrap();
    harness
        .messenger
        .prepare_activation(&admission)
        .await
        .unwrap();

    let mut live = Session::new(&harness.parent_session_id, "live-cache-model");
    live.metadata
        .insert("cache_owner".to_string(), "manual-run".to_string());
    let live_arc = Arc::new(crate::SessionSnapshot::new(live));
    harness
        .ctx
        .sessions_cache
        .insert(harness.parent_session_id.clone(), live_arc.clone());
    let event_sender = crate::runtime::execution::session_events::get_or_create_event_sender(
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
    )
    .await;
    let existing_run_id = match crate::runtime::execution::runner_lifecycle::reserve_runner_core(
        &harness.agent_runners,
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
        &event_sender,
    )
    .await
    {
        crate::runtime::execution::runner_lifecycle::ReserveOutcome::Reserved(reservation) => {
            reservation.run_id
        }
        crate::runtime::execution::runner_lifecycle::ReserveOutcome::AlreadyRunning(_) => {
            panic!("test must establish the manual owner")
        }
    };

    let outcome = crate::SessionActivationSpawner::reserve_activation(
        harness.coordinator.as_ref(),
        &harness.parent_session_id,
        admission.delivery.generation,
    )
    .await
    .unwrap();
    assert!(matches!(
        outcome,
        crate::SessionActivationReserveOutcome::AlreadyRunning { run_id }
            if run_id == existing_run_id
    ));
    let cached = harness
        .ctx
        .sessions_cache
        .get(&harness.parent_session_id)
        .unwrap()
        .value()
        .clone();
    assert!(
        Arc::ptr_eq(&cached, &live_arc),
        "losing activation must not publish its detached prepared snapshot"
    );
    assert_eq!(
        cached
            .read()
            .metadata
            .get("cache_owner")
            .map(String::as_str),
        Some("manual-run")
    );
}

#[tokio::test]
async fn direct_registration_waits_for_activation_without_creating_a_phantom_owner() {
    let provider_calls = Arc::new(AtomicUsize::new(0));
    let provider_entered = Arc::new(Notify::new());
    let provider_release = Arc::new(Notify::new());
    let harness = build_terminal_delivery_harness(
        Arc::new(BlockingCompletedProvider {
            calls: provider_calls.clone(),
            entered: provider_entered.clone(),
            release: provider_release.clone(),
        }),
        None,
        None,
    )
    .await;
    let admission = harness
        .messenger
        .admit(SessionMessageEnvelope::user_input(
            &harness.parent_session_id,
            "activation wins before direct registration",
        ))
        .await
        .unwrap();
    harness
        .messenger
        .prepare_activation(&admission)
        .await
        .unwrap();

    let messenger = harness.messenger.clone();
    let activation = tokio::spawn(async move { messenger.activate_prepared(&admission).await });
    tokio::time::timeout(
        Duration::from_secs(5),
        harness.reservation_entered.notified(),
    )
    .await
    .expect("router must enter the gated runtime reservation");

    let mut live = Session::new(&harness.parent_session_id, "manual-owner-model");
    live.metadata
        .insert("cache_owner".to_string(), "manual-router-owner".to_string());
    let live_arc = Arc::new(crate::SessionSnapshot::new(live));
    let before_commit = live_arc.read();
    harness
        .ctx
        .sessions_cache
        .insert(harness.parent_session_id.clone(), live_arc.clone());

    let router = harness
        .ctx
        .agent
        .activation_router()
        .expect("delivery harness has an activation router")
        .clone();
    let direct_router = router.clone();
    let direct_session_id = harness.parent_session_id.clone();
    let direct = tokio::spawn(async move {
        direct_router
            .register_run(&direct_session_id, "manual-router-owner")
            .await
    });
    tokio::task::yield_now().await;
    assert!(
        !direct.is_finished(),
        "direct registration must remain pending while activation owns the router token"
    );
    assert!(
        !router
            .owns_run(&harness.parent_session_id, "manual-router-owner")
            .await
    );
    let cached_before_launch = harness
        .ctx
        .sessions_cache
        .get(&harness.parent_session_id)
        .unwrap()
        .value()
        .clone();
    assert_eq!(
        cached_before_launch.read().metadata.get("cache_owner"),
        before_commit.metadata.get("cache_owner"),
        "prepared activation cache must remain unpublished before router ownership commits"
    );

    harness.allow_reservation.notify_one();

    let disposition = tokio::time::timeout(Duration::from_secs(5), activation)
        .await
        .expect("serialized activation must settle")
        .expect("activation task must not panic")
        .expect("durable activation remains accepted");
    assert_eq!(
        disposition.activation,
        bamboo_domain::SessionActivationDisposition::ActivationReserved
    );
    tokio::time::timeout(Duration::from_secs(5), provider_entered.notified())
        .await
        .expect("the published activation must reach one real provider execution");

    let collision = match tokio::time::timeout(Duration::from_secs(5), direct)
        .await
        .expect("direct registration must wake after activation publication")
        .expect("direct registration task must not panic")
    {
        Ok(_) => panic!("direct registration must not replace the activation owner"),
        Err(error) => error,
    };
    let activation_run_id = collision.existing_run_id().to_string();
    assert_ne!(activation_run_id, "manual-router-owner");
    assert!(
        router
            .owns_run(&harness.parent_session_id, &activation_run_id)
            .await
    );
    assert!(
        !router
            .owns_run(&harness.parent_session_id, "manual-router-owner")
            .await
    );
    {
        let runners = harness.agent_runners.read().await;
        let runner = runners
            .get(&harness.parent_session_id)
            .expect("the activation must retain its exact raw runner");
        assert_eq!(runner.run_id, activation_run_id);
        assert!(matches!(
            runner.status,
            crate::execution::AgentStatus::Running
        ));
    }

    let cached = harness
        .ctx
        .sessions_cache
        .get(&harness.parent_session_id)
        .unwrap()
        .value()
        .clone();
    assert_eq!(
        before_commit.metadata["cache_owner"], "manual-router-owner",
        "a reader retained across owner commit must keep its original snapshot"
    );
    assert!(
        !cached.read().metadata.contains_key("cache_owner"),
        "the stale manual cache snapshot must not masquerade as the winning execution"
    );

    provider_release.notify_one();
    tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            let runner_settled = harness
                .agent_runners
                .read()
                .await
                .get(&harness.parent_session_id)
                .is_some_and(|runner| {
                    runner.run_id == activation_run_id
                        && !matches!(runner.status, crate::execution::AgentStatus::Running)
                });
            let owner_released = !router
                .owns_run(&harness.parent_session_id, &activation_run_id)
                .await;
            if runner_settled && owner_released {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("the one real activation must settle and release its exact router owner");
    assert!(
        !router
            .owns_run(&harness.parent_session_id, "manual-router-owner")
            .await,
        "the rejected direct run must never leave a phantom router owner"
    );
    assert_eq!(harness.reservations.load(Ordering::SeqCst), 1);
    assert_eq!(provider_calls.load(Ordering::SeqCst), 1);
    let backlog = harness
        .inbox
        .inspect(&harness.parent_session_id)
        .await
        .unwrap();
    assert_eq!(backlog.pending + backlog.claimed, 0);
}

#[tokio::test]
async fn cancelled_manual_waiter_preserves_the_sdk_activation_token() {
    let provider_calls = Arc::new(AtomicUsize::new(0));
    let provider_entered = Arc::new(Notify::new());
    let provider_release = Arc::new(Notify::new());
    let harness = build_terminal_delivery_harness(
        Arc::new(BlockingCompletedProvider {
            calls: provider_calls.clone(),
            entered: provider_entered.clone(),
            release: provider_release.clone(),
        }),
        None,
        None,
    )
    .await;
    let admission = harness
        .messenger
        .admit(SessionMessageEnvelope::user_input(
            &harness.parent_session_id,
            "deliver after a cancelled manual registration waiter",
        ))
        .await
        .unwrap();
    harness
        .messenger
        .prepare_activation(&admission)
        .await
        .unwrap();

    let messenger = harness.messenger.clone();
    let activation = tokio::spawn(async move { messenger.activate_prepared(&admission).await });
    tokio::time::timeout(
        Duration::from_secs(5),
        harness.reservation_entered.notified(),
    )
    .await
    .expect("activation must hold its router token before checking the raw registry");

    let sender = crate::execution::get_or_create_event_sender(
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
    )
    .await;
    let mut manual = Box::pin(crate::execution::reserve_session_execution(
        &harness.ctx.agent,
        &harness.agent_runners,
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
        &sender,
    ));
    tokio::select! {
        biased;
        _ = &mut manual => panic!("manual registration must wait for the genuine SDK activation token"),
        _ = tokio::task::yield_now() => {},
    }
    assert!(
        harness
            .agent_runners
            .read()
            .await
            .get(&harness.parent_session_id)
            .is_none(),
        "a waiting manual entry must not install a raw runner"
    );
    drop(manual);
    let router = harness.ctx.agent.activation_router().unwrap().clone();
    assert!(router
        .current_run_id(&harness.parent_session_id)
        .await
        .is_none());
    assert_eq!(provider_calls.load(Ordering::SeqCst), 0);
    assert!(
        !activation.is_finished(),
        "cancelling the manual waiter must preserve the SDK token"
    );
    harness.allow_reservation.notify_one();
    let disposition = tokio::time::timeout(Duration::from_secs(5), activation)
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(
        disposition.activation,
        bamboo_domain::SessionActivationDisposition::ActivationReserved
    );
    assert_eq!(
        harness.reservations.load(Ordering::SeqCst),
        1,
        "a cancelled waiter must not manufacture an extra SDK activation"
    );
    tokio::time::timeout(Duration::from_secs(5), provider_entered.notified())
        .await
        .expect("the bounded successor must reach one real provider execution");
    let successor_run_id = {
        let runners = harness.agent_runners.read().await;
        let runner = runners
            .get(&harness.parent_session_id)
            .expect("successor raw runner must be present");
        assert!(matches!(
            runner.status,
            crate::execution::AgentStatus::Running
        ));
        runner.run_id.clone()
    };
    assert!(
        router
            .owns_run(&harness.parent_session_id, &successor_run_id)
            .await
    );
    assert_eq!(provider_calls.load(Ordering::SeqCst), 1);

    provider_release.notify_one();
    tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            let settled = harness
                .agent_runners
                .read()
                .await
                .get(&harness.parent_session_id)
                .is_some_and(|runner| {
                    runner.run_id == successor_run_id
                        && !matches!(runner.status, crate::execution::AgentStatus::Running)
                });
            let owner_released = !router
                .owns_run(&harness.parent_session_id, &successor_run_id)
                .await;
            if settled && owner_released {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("the real successor must settle without a phantom owner");
    assert_eq!(provider_calls.load(Ordering::SeqCst), 1);
    let backlog = harness
        .inbox
        .inspect(&harness.parent_session_id)
        .await
        .unwrap();
    assert_eq!(backlog.pending + backlog.claimed, 0);
}

#[tokio::test]
async fn shared_execution_reservation_collides_before_entry_side_effects() {
    let harness = build_terminal_delivery_harness(Arc::new(CompletedProvider), None, None).await;
    let router = harness
        .ctx
        .agent
        .activation_router()
        .expect("delivery harness has an activation router")
        .clone();
    let mut sdk_owner = router
        .register_run(&harness.parent_session_id, "sdk-owner")
        .await
        .unwrap();
    let sender = crate::execution::get_or_create_event_sender(
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
    )
    .await;

    let mut predecessor = crate::execution::AgentRunner::new();
    predecessor.status = crate::execution::AgentStatus::Completed;
    let previous_run = predecessor.run_id.clone();
    let previous_publication = predecessor.event_publication.clone();
    harness
        .agent_runners
        .write()
        .await
        .insert(harness.parent_session_id.clone(), predecessor);
    let outcome = crate::execution::reserve_session_execution(
        &harness.ctx.agent,
        &harness.agent_runners,
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
        &sender,
    )
    .await;
    assert!(matches!(
        outcome,
        crate::execution::SessionExecutionReserveOutcome::AlreadyRunning { run_id }
            if run_id == "sdk-owner"
    ));
    {
        let runners = harness.agent_runners.read().await;
        let retained = runners.get(&harness.parent_session_id).unwrap();
        assert!(matches!(
            retained.status,
            crate::execution::AgentStatus::Completed
        ));
        assert_eq!(retained.run_id, previous_run);
        assert!(Arc::ptr_eq(
            &retained.event_publication,
            &previous_publication
        ));
    }
    assert!(
        router
            .owns_run(&harness.parent_session_id, "sdk-owner")
            .await
    );

    sdk_owner.begin_finalization().await;
    assert_eq!(sdk_owner.finish(0).await.unwrap(), None);
}

#[tokio::test]
async fn dropped_shared_execution_reservation_releases_runner_and_router() {
    let harness = build_terminal_delivery_harness(Arc::new(CompletedProvider), None, None).await;
    let sender = crate::execution::get_or_create_event_sender(
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
    )
    .await;
    let reservation = match crate::execution::reserve_session_execution(
        &harness.ctx.agent,
        &harness.agent_runners,
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
        &sender,
    )
    .await
    {
        crate::execution::SessionExecutionReserveOutcome::Reserved(reservation) => reservation,
        crate::execution::SessionExecutionReserveOutcome::AlreadyRunning { run_id } => {
            panic!("unexpected owner {run_id}")
        }
    };
    let run_id = reservation.run_id().to_string();
    let router = harness
        .ctx
        .agent
        .activation_router()
        .expect("delivery harness has an activation router")
        .clone();
    assert!(router.owns_run(&harness.parent_session_id, &run_id).await);
    drop(reservation);

    tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            let owns = router.owns_run(&harness.parent_session_id, &run_id).await;
            let cancelled = matches!(
                harness
                    .agent_runners
                    .read()
                    .await
                    .get(&harness.parent_session_id)
                    .map(|runner| &runner.status),
                Some(crate::execution::AgentStatus::Cancelled)
            );
            if !owns && cancelled {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("Drop must clean both exact ownership planes");
}

#[tokio::test]
async fn cancelled_explicit_reservation_abandon_still_cleans_both_ownership_planes() {
    let harness = build_terminal_delivery_harness(Arc::new(CompletedProvider), None, None).await;
    let sender = crate::execution::get_or_create_event_sender(
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
    )
    .await;
    let reservation = match crate::execution::reserve_session_execution(
        &harness.ctx.agent,
        &harness.agent_runners,
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
        &sender,
    )
    .await
    {
        crate::execution::SessionExecutionReserveOutcome::Reserved(reservation) => reservation,
        crate::execution::SessionExecutionReserveOutcome::AlreadyRunning { run_id } => {
            panic!("unexpected owner {run_id}")
        }
    };
    let run_id = reservation.run_id().to_string();
    let cancel_token = reservation.cancel_token().clone();
    let router = harness
        .ctx
        .agent
        .activation_router()
        .expect("delivery harness has an activation router")
        .clone();
    let runners_guard = harness.agent_runners.write().await;
    let abandon = tokio::spawn(reservation.abandon());
    tokio::time::timeout(Duration::from_secs(5), cancel_token.cancelled())
        .await
        .expect("explicit abandon must cancel its exact runner token");
    assert!(matches!(
        runners_guard.get(&harness.parent_session_id).map(|runner| {
            assert_eq!(runner.run_id, run_id);
            &runner.status
        }),
        Some(crate::execution::AgentStatus::Running)
    ));

    abandon.abort();
    assert!(abandon.await.unwrap_err().is_cancelled());
    drop(runners_guard);

    tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            let cancelled = matches!(
                harness
                    .agent_runners
                    .read()
                    .await
                    .get(&harness.parent_session_id)
                    .map(|runner| &runner.status),
                Some(crate::execution::AgentStatus::Cancelled)
            );
            let owner_released = !router.owns_run(&harness.parent_session_id, &run_id).await;
            if cancelled && owner_released {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("caller cancellation must not cancel detached reservation cleanup");
}

#[tokio::test]
async fn cancelled_pending_registration_cleans_after_router_lock_releases() {
    let harness = build_terminal_delivery_harness(Arc::new(CompletedProvider), None, None).await;
    let router = harness
        .ctx
        .agent
        .activation_router()
        .expect("delivery harness has an activation router")
        .clone();
    let lock_entered = Arc::new(Notify::new());
    let release_lock = Arc::new(Notify::new());
    let lock_holder = {
        let router = router.clone();
        let lock_entered = lock_entered.clone();
        let release_lock = release_lock.clone();
        tokio::spawn(async move {
            router
                .hold_state_lock_for_test(lock_entered, release_lock)
                .await;
        })
    };
    lock_entered.notified().await;

    let sender = crate::execution::get_or_create_event_sender(
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
    )
    .await;
    let mut reservation = Box::pin(crate::execution::reserve_session_execution(
        &harness.ctx.agent,
        &harness.agent_runners,
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
        &sender,
    ));
    tokio::select! {
        biased;
        _ = &mut reservation => panic!("reservation must remain pending at the real router mutex"),
        _ = tokio::task::yield_now() => {},
    }
    assert!(harness
        .agent_runners
        .read()
        .await
        .get(&harness.parent_session_id)
        .is_none());
    drop(reservation);
    release_lock.notify_one();
    lock_holder.await.unwrap();
    assert!(router
        .current_run_id(&harness.parent_session_id)
        .await
        .is_none());
    assert!(harness
        .agent_runners
        .read()
        .await
        .get(&harness.parent_session_id)
        .is_none());
}

#[tokio::test]
async fn cancelled_published_placeholder_is_adopted_and_abandoned_by_raii() {
    let harness = build_terminal_delivery_harness(Arc::new(CompletedProvider), None, None).await;
    let router = harness
        .ctx
        .agent
        .activation_router()
        .expect("delivery harness has an activation router")
        .clone();
    let sender = crate::execution::get_or_create_event_sender(
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
    )
    .await;
    let raw = match crate::execution::reserve_runner_core(
        &harness.agent_runners,
        &harness.ctx.session_event_senders,
        &harness.parent_session_id,
        &sender,
    )
    .await
    {
        crate::execution::ReserveOutcome::Reserved(reservation) => reservation,
        crate::execution::ReserveOutcome::AlreadyRunning(run_id) => {
            panic!("unexpected owner {run_id}")
        }
    };
    let run_id = raw.run_id.clone();
    router
        .install_owner_placeholder_for_test(&harness.parent_session_id, &run_id)
        .await;
    let mut reservation =
        crate::execution::SessionExecutionReservation::from_activation_placeholder(
            harness.parent_session_id.clone(),
            raw,
            router.clone(),
            harness.agent_runners.clone(),
        );
    reservation.mark_activation_published();

    let lock_entered = Arc::new(Notify::new());
    let release_lock = Arc::new(Notify::new());
    let lock_holder = {
        let router = router.clone();
        let lock_entered = lock_entered.clone();
        let release_lock = release_lock.clone();
        tokio::spawn(async move {
            router
                .hold_state_lock_for_test(lock_entered, release_lock)
                .await;
        })
    };
    lock_entered.notified().await;
    let adoption = tokio::spawn(async move { reservation.ensure_registered().await });
    tokio::task::yield_now().await;
    adoption.abort();
    assert!(adoption.await.unwrap_err().is_cancelled());

    release_lock.notify_one();
    lock_holder.await.unwrap();
    tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            let owns = router.owns_run(&harness.parent_session_id, &run_id).await;
            let cancelled = matches!(
                harness
                    .agent_runners
                    .read()
                    .await
                    .get(&harness.parent_session_id)
                    .map(|runner| &runner.status),
                Some(crate::execution::AgentStatus::Cancelled)
            );
            if !owns && cancelled {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("cancelled launched placeholder must clean both ownership planes");
}

#[tokio::test]
async fn aborted_reserved_child_setup_waiter_preserves_owned_run_without_successor() {
    let provider_calls = Arc::new(AtomicUsize::new(0));
    let provider_release = Arc::new(Notify::new());
    provider_release.notify_one();
    let mut harness = build_terminal_delivery_harness(
        Arc::new(BlockingCompletedProvider {
            calls: provider_calls.clone(),
            entered: Arc::new(Notify::new()),
            release: provider_release,
        }),
        None,
        None,
    )
    .await;
    let input = SessionMessageEnvelope::user_input(
        &harness.child_session_id,
        "finish child work after its reserved setup waiter is aborted",
    );
    let admission = harness.messenger.admit(input.clone()).await.unwrap();
    harness
        .messenger
        .prepare_activation(&admission)
        .await
        .unwrap();
    let router = harness
        .ctx
        .agent
        .activation_router()
        .expect("delivery harness has an activation router")
        .clone();
    let child_sender = crate::execution::get_or_create_event_sender(
        &harness.ctx.session_event_senders,
        &harness.child_session_id,
    )
    .await;
    let raw = match crate::execution::reserve_runner_core(
        &harness.agent_runners,
        &harness.ctx.session_event_senders,
        &harness.child_session_id,
        &child_sender,
    )
    .await
    {
        crate::execution::ReserveOutcome::Reserved(reservation) => reservation,
        crate::execution::ReserveOutcome::AlreadyRunning(run_id) => {
            panic!("unexpected child owner {run_id}")
        }
    };
    let owned_run_id = raw.run_id.clone();
    router
        .install_owner_placeholder_for_test(&harness.child_session_id, &owned_run_id)
        .await;
    let mut reservation =
        crate::execution::SessionExecutionReservation::from_activation_placeholder(
            harness.child_session_id.clone(),
            raw,
            router.clone(),
            harness.agent_runners.clone(),
        );
    reservation.mark_activation_published();

    let lock_entered = Arc::new(Notify::new());
    let release_lock = Arc::new(Notify::new());
    let lock_holder = {
        let router = router.clone();
        let lock_entered = lock_entered.clone();
        let release_lock = release_lock.clone();
        tokio::spawn(async move {
            router
                .hold_state_lock_for_test(lock_entered, release_lock)
                .await;
        })
    };
    lock_entered.notified().await;
    let job = SpawnJob {
        parent_session_id: harness.parent_session_id.clone(),
        child_session_id: harness.child_session_id.clone(),
        model: "gpt-5".to_string(),
        disabled_tools: None,
    };
    let setup = harness._spawn_scheduler.launch_reserved(job, reservation);
    let expected_child_id = harness.child_session_id.clone();
    tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            match harness.parent_rx.recv().await {
                Ok(AgentEvent::SubAgentStarted {
                    child_session_id, ..
                }) if child_session_id == expected_child_id => break,
                Ok(_) | Err(broadcast::error::RecvError::Lagged(_)) => continue,
                Err(error) => panic!("parent event stream closed during child setup: {error}"),
            }
        }
    })
    .await
    .expect("reserved child setup must become runnable before its waiter is cancelled");
    setup.abort();
    match setup.await {
        Err(error) => assert!(error.is_cancelled()),
        Ok(()) => panic!("reserved child setup unexpectedly completed before cancellation"),
    }

    release_lock.notify_one();
    lock_holder.await.unwrap();
    // Cancelling the caller's wait does not abandon the owned inner setup.
    // It must hand the original reservation to exactly one real execution,
    // without asking the router to replace an owner that may have claimed work.
    tokio::time::timeout(
        Duration::from_secs(10),
        harness.first_execute_returned.notified(),
    )
    .await
    .expect("owned child setup must finish its original provider boundary");
    assert_eq!(provider_calls.load(Ordering::SeqCst), 1);
    assert_eq!(harness.reservations.load(Ordering::SeqCst), 0);
    {
        let runners = harness.agent_runners.read().await;
        let runner = runners
            .get(&harness.child_session_id)
            .expect("original owned child runner must be present");
        assert_eq!(runner.run_id, owned_run_id);
        assert!(matches!(
            runner.status,
            crate::execution::AgentStatus::Running
        ));
    }
    assert!(
        router
            .owns_run(&harness.child_session_id, &owned_run_id)
            .await
    );

    harness.release_first_terminal.notify_one();
    tokio::time::timeout(Duration::from_secs(10), async {
        loop {
            let settled = harness
                .agent_runners
                .read()
                .await
                .get(&harness.child_session_id)
                .is_some_and(|runner| {
                    runner.run_id == owned_run_id
                        && !matches!(runner.status, crate::execution::AgentStatus::Running)
                });
            let owner_released = !router
                .owns_run(&harness.child_session_id, &owned_run_id)
                .await;
            if settled && owner_released {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("original child must settle without a phantom owner");
    let events = collect_until_completed(&mut harness.parent_rx).await;
    assert_eq!(events.iter().filter(|event| matches!(event,
        AgentEvent::SubAgentCompleted { child_session_id, status, error, .. }
            if child_session_id == &harness.child_session_id && status == "completed" && error.is_none()
    )).count(), 1);
    let saved = harness
        .storage
        .load_session(&harness.child_session_id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(saved.last_run_status().as_deref(), Some("completed"));
    assert_eq!(
        saved
            .messages
            .iter()
            .filter(|message| message.id == input.id.as_str())
            .count(),
        1
    );
    let source = crate::execution::ChildCompletionSource::from_committed_session(&saved).unwrap();
    assert_eq!(source.activation_run_id, owned_run_id);
    let backlog = harness
        .inbox
        .inspect(&harness.child_session_id)
        .await
        .unwrap();
    assert_eq!(backlog.pending + backlog.claimed, 0);
    assert_eq!(harness.reservations.load(Ordering::SeqCst), 0);
    assert_eq!(provider_calls.load(Ordering::SeqCst), 1);
    assert_eq!(harness.durable_confirmations.load(Ordering::SeqCst), 1);
    while let Ok(event) = harness.parent_rx.try_recv() {
        assert!(
            !matches!(event, AgentEvent::SubAgentCompleted { .. }),
            "owned run must publish completion once"
        );
    }
}

#[tokio::test]
async fn terminal_delivery_runs_only_in_one_real_successor_execution() {
    let message_id = "terminal-production-race".to_string();
    let provider_calls = Arc::new(AtomicUsize::new(0));
    let successor_seen = Arc::new(Notify::new());
    let successor_match_count = Arc::new(AtomicUsize::new(0));
    let provider: Arc<dyn LLMProvider> = Arc::new(SuccessorRecordingProvider {
        calls: provider_calls.clone(),
        expected_message_id: message_id.clone(),
        successor_call: 2,
        successor_seen: successor_seen.clone(),
        successor_match_count: successor_match_count.clone(),
    });
    let harness = build_terminal_delivery_harness(provider, None, None).await;
    let job = SpawnJob {
        parent_session_id: harness.parent_session_id.clone(),
        child_session_id: harness.child_session_id.clone(),
        model: "gpt-5".to_string(),
        disabled_tools: None,
    };

    run_child_spawn(harness.ctx.clone(), job).await.unwrap();
    tokio::time::timeout(
        Duration::from_secs(10),
        harness.first_execute_returned.notified(),
    )
    .await
    .expect("first production run must reach terminal barrier");

    let activation_a = harness
        .agent_runners
        .read()
        .await
        .get(&harness.child_session_id)
        .unwrap()
        .run_id
        .clone();
    let mut envelope =
        SessionMessageEnvelope::user_input(&harness.child_session_id, "after final provider round");
    envelope.id = SessionMessageId::parse(message_id.clone()).unwrap();
    let receipt = harness.messenger.send(envelope.clone()).await.unwrap();
    assert_eq!(
        receipt.activation,
        bamboo_domain::SessionActivationDisposition::ActiveNotified
    );
    harness.release_first_terminal.notify_one();

    // The real sdk::spawn terminal block has entered finish_finalization and
    // asked the real coordinator for a successor. Hold that reservation long
    // enough to prove terminal cleanup did not claim or acknowledge typed work.
    tokio::time::timeout(
        Duration::from_secs(10),
        harness.reservation_entered.notified(),
    )
    .await
    .expect("terminal finalization must request a successor");
    assert_eq!(
        harness.durable_confirmations.load(Ordering::SeqCst),
        1,
        "the external ACK callback must follow the first durable terminal save"
    );
    let backlog = harness
        .inbox
        .inspect(&harness.child_session_id)
        .await
        .unwrap();
    assert_eq!(backlog.pending, 1);
    assert_eq!(backlog.claimed, 0);
    assert!(!harness
        .inbox
        .was_admitted(&harness.child_session_id, &envelope.id)
        .await
        .unwrap());
    let first_terminal_snapshot = harness
        .storage
        .load_session(&harness.child_session_id)
        .await
        .unwrap()
        .unwrap();
    let source_a =
        crate::execution::ChildCompletionSource::from_committed_session(&first_terminal_snapshot)
            .expect(
            "actual SDK finalizer must seal the canonical terminal source before successor release",
        );
    let parent = harness
        .storage
        .load_session(&harness.parent_session_id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(source_a.activation_run_id, activation_a);
    assert_eq!(
        source_a
            .result(&first_terminal_snapshot, &parent)
            .as_deref(),
        Some("done")
    );
    assert!(!first_terminal_snapshot
        .messages
        .iter()
        .any(|message| message.id == message_id));
    assert_eq!(harness.reservations.load(Ordering::SeqCst), 1);

    harness.allow_reservation.notify_one();
    tokio::time::timeout(Duration::from_secs(10), successor_seen.notified())
        .await
        .expect("real successor execution must reach its provider");
    assert_eq!(successor_match_count.load(Ordering::SeqCst), 1);

    // Wait for the successor's real terminal block to finalize its runner.
    tokio::time::timeout(Duration::from_secs(10), async {
        loop {
            let terminal = harness
                .agent_runners
                .read()
                .await
                .get(&harness.child_session_id)
                .is_some_and(|runner| {
                    !matches!(runner.status, crate::execution::AgentStatus::Running)
                });
            if terminal {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("real successor terminal block must finalize its runner");
    assert_eq!(
        provider_calls.load(Ordering::SeqCst),
        2,
        "exactly the first run and one successor may invoke the provider"
    );
    let settled = harness
        .storage
        .load_session(&harness.child_session_id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(
        settled
            .messages
            .iter()
            .filter(|message| message.id == message_id)
            .count(),
        1
    );
    let source_b = crate::execution::ChildCompletionSource::from_committed_session(&settled)
        .expect("actual resumed-child finalizer must seal its own terminal source");
    assert_eq!(
        source_b.activation_run_id,
        harness
            .agent_runners
            .read()
            .await
            .get(&harness.child_session_id)
            .unwrap()
            .run_id
    );
    assert_ne!(source_a.activation_run_id, source_b.activation_run_id);
    assert_ne!(source_a.result_message_id, source_b.result_message_id);
    assert!(source_a.result(&settled, &parent).is_none());
    assert!(source_b.result(&settled, &parent).is_some());
    let backlog = harness
        .inbox
        .inspect(&harness.child_session_id)
        .await
        .unwrap();
    assert_eq!(backlog.pending + backlog.claimed, 0);
    assert!(harness
        .inbox
        .was_admitted(&harness.child_session_id, &envelope.id)
        .await
        .unwrap());
}

#[tokio::test]
async fn legacy_boundary_checkpoint_failure_still_runs_one_successor() {
    let legacy_value =
        serde_json::json!({ "content": "legacy message whose first checkpoint fails" });
    let legacy_id = SessionMessageId::legacy("terminal-child", 0, &legacy_value).to_string();
    let provider_calls = Arc::new(AtomicUsize::new(0));
    let successor_seen = Arc::new(Notify::new());
    let successor_match_count = Arc::new(AtomicUsize::new(0));
    let provider: Arc<dyn LLMProvider> = Arc::new(SuccessorRecordingProvider {
        calls: provider_calls.clone(),
        expected_message_id: legacy_id.clone(),
        successor_call: 1,
        successor_seen: successor_seen.clone(),
        successor_match_count: successor_match_count.clone(),
    });
    let harness =
        build_terminal_delivery_harness(provider, Some(legacy_value), Some(legacy_id.clone()))
            .await;
    let job = SpawnJob {
        parent_session_id: harness.parent_session_id.clone(),
        child_session_id: harness.child_session_id.clone(),
        model: "gpt-5".to_string(),
        disabled_tools: None,
    };

    run_child_spawn(harness.ctx.clone(), job).await.unwrap();
    tokio::time::timeout(
        Duration::from_secs(10),
        harness.first_execute_returned.notified(),
    )
    .await
    .expect("first run must complete after the injected checkpoint failure");
    // The failed durable checkpoint leaves the claim recoverable and must not
    // dispatch an empty prompt before the single retained successor.
    assert_eq!(provider_calls.load(Ordering::SeqCst), 0);

    // The compatibility source is already CAS-cleared, while the typed claim
    // remains the sole recoverable copy. The boundary's observed generation
    // must therefore survive on the current run until terminal finalization.
    let first_snapshot = harness
        .storage
        .load_session(&harness.child_session_id)
        .await
        .unwrap()
        .unwrap();
    assert!(!first_snapshot.has_pending_injected_messages());
    assert!(!first_snapshot
        .messages
        .iter()
        .any(|message| message.id == legacy_id));
    let backlog = harness
        .inbox
        .inspect(&harness.child_session_id)
        .await
        .unwrap();
    assert_eq!(backlog.pending, 0);
    assert_eq!(backlog.claimed, 1);

    harness.release_first_terminal.notify_one();
    tokio::time::timeout(
        Duration::from_secs(10),
        harness.reservation_entered.notified(),
    )
    .await
    .expect("normal-boundary generation must reach terminal activation");
    assert_eq!(harness.reservations.load(Ordering::SeqCst), 1);
    assert!(!harness
        .inbox
        .was_admitted(
            &harness.child_session_id,
            &SessionMessageId::parse(legacy_id.clone()).unwrap(),
        )
        .await
        .unwrap());

    harness.allow_reservation.notify_one();
    tokio::time::timeout(Duration::from_secs(10), successor_seen.notified())
        .await
        .expect("one real successor must retry and admit the legacy envelope");
    assert_eq!(successor_match_count.load(Ordering::SeqCst), 1);

    tokio::time::timeout(Duration::from_secs(10), async {
        loop {
            let terminal = harness
                .agent_runners
                .read()
                .await
                .get(&harness.child_session_id)
                .is_some_and(|runner| {
                    !matches!(runner.status, crate::execution::AgentStatus::Running)
                });
            if terminal {
                break;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("successor must finalize");
    assert_eq!(provider_calls.load(Ordering::SeqCst), 1);
    assert_eq!(harness.reservations.load(Ordering::SeqCst), 1);
    let settled = harness
        .storage
        .load_session(&harness.child_session_id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(
        settled
            .messages
            .iter()
            .filter(|message| message.id == legacy_id)
            .count(),
        1
    );
    let backlog = harness
        .inbox
        .inspect(&harness.child_session_id)
        .await
        .unwrap();
    assert_eq!(backlog.pending + backlog.claimed, 0);
}

// ---------------------------------------------------------------------------
// S-T2.2 — Sub-agents are full agents: ChildRunner never trims tools.
//
// There are no per-role tool policies anymore; the runner always emits
// `disabled_tools: None` regardless of the live tool catalog.
// ---------------------------------------------------------------------------

#[tokio::test]
async fn s_t2_2_run_child_never_disables_tools() {
    let tool_names = vec![
        "Read".to_string(),
        "Grep".to_string(),
        "Edit".to_string(),
        "Write".to_string(),
    ];
    let harness = build_harness(Arc::new(CompletedProvider), tool_names, &[]).await;
    let runner = ChildRunner::new(harness.ctx.clone());

    let input = RunChildInput {
        child_session_id: harness.child_session_id.clone(),
        parent_session_id: harness.parent_session_id.clone(),
        model: "gpt-5".to_string(),
    };
    let job = runner.build_job(&input);
    assert!(
        job.disabled_tools.is_none(),
        "sub-agents are full agents; the runner must never trim tools"
    );
}

// ---------------------------------------------------------------------------
// S-T2.4 — model override honored over the persisted session model
// ---------------------------------------------------------------------------

#[tokio::test]
async fn s_t2_4_run_child_model_override_is_persisted() {
    let harness = build_harness(Arc::new(CompletedProvider), Vec::new(), &[]).await;
    let runner = ChildRunner::new(harness.ctx.clone());
    let mut parent_runner = crate::execution::AgentRunner::new();
    parent_runner.status = crate::execution::AgentStatus::Running;
    harness
        .ctx
        .agent_runners
        .write()
        .await
        .insert(harness.parent_session_id.clone(), parent_runner);

    // Child session was persisted with model "gpt-5"; run with a different model.
    let override_model = "claude-3-7-sonnet";
    let mut parent_rx = harness.parent_rx.resubscribe();
    runner
        .run_child(RunChildInput {
            child_session_id: harness.child_session_id.clone(),
            parent_session_id: harness.parent_session_id.clone(),
            model: override_model.to_string(),
        })
        .await
        .unwrap();

    let events = collect_until_completed(&mut parent_rx).await;
    assert!(matches!(
        events.first(),
        Some(AgentEvent::SubAgentStarted {
            child_session_id,
            ..
        }) if child_session_id == &harness.child_session_id
    ));
    assert!(matches!(
        events.last(),
        Some(AgentEvent::SubAgentCompleted {
            child_session_id,
            ..
        }) if child_session_id == &harness.child_session_id
    ));

    let runners = harness.ctx.agent_runners.read().await;
    let cached_lifecycle = runners
        .get(&harness.parent_session_id)
        .expect("parent runner retained")
        .last_critical_events
        .iter()
        .filter(|event| match event {
            AgentEvent::SubAgentStarted {
                child_session_id, ..
            }
            | AgentEvent::SubAgentCompleted {
                child_session_id, ..
            } => child_session_id == &harness.child_session_id,
            _ => false,
        })
        .collect::<Vec<_>>();
    assert_eq!(cached_lifecycle.len(), 1);
    assert!(matches!(
        cached_lifecycle[0],
        AgentEvent::SubAgentCompleted { .. }
    ));
    drop(runners);

    let persisted = harness
        .storage
        .load_session(&harness.child_session_id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(
        persisted.model, override_model,
        "child session model must reflect the runner override"
    );
}

// ---------------------------------------------------------------------------
// S-T2.5 — watchdog timeout → SubAgentCompleted status=timeout
// ---------------------------------------------------------------------------

#[tokio::test]
async fn s_t2_5_watchdog_timeout_completes_with_timeout_status() {
    // Tight watchdog: 1s total cap, checked every 1s.
    let harness = build_harness(
        Arc::new(HangingProvider),
        Vec::new(),
        // The watchdog skips its immediate tick, so the first real check lands at
        // `check_interval_secs` (1s). By then `total_secs` already meets the 1s cap,
        // so the timeout fires on that first check — well before the slow stream
        // (first chunk at +2s) could finish, making the outcome deterministic.
        &[
            ("child_watchdog.max_total_secs", "1"),
            ("child_watchdog.max_idle_secs", "1"),
            ("child_watchdog.check_interval_secs", "1"),
        ],
    )
    .await;

    let mut parent_rx = harness.parent_rx.resubscribe();
    run_child_spawn(
        harness.ctx.clone(),
        SpawnJob {
            parent_session_id: harness.parent_session_id.clone(),
            child_session_id: harness.child_session_id.clone(),
            model: "gpt-5".to_string(),
            disabled_tools: None,
        },
    )
    .await
    .unwrap();

    let events = collect_until_completed_with_budget(&mut parent_rx, Duration::from_secs(45)).await;
    match events.last().unwrap() {
        AgentEvent::SubAgentCompleted { status, .. } => {
            assert_eq!(status, "timeout", "watchdog must yield timeout status");
        }
        other => panic!("expected SubAgentCompleted, got {other:?}"),
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum TerminalDeliveryFailure {
    UnreceiptedExecutionError,
    Prepare,
    Checkpoint,
    Proof,
    Ack,
}

struct TerminalDeliveryPersistence {
    inner: Arc<dyn bamboo_domain::RuntimeSessionPersistence>,
    failure: TerminalDeliveryFailure,
}

#[async_trait]
impl bamboo_domain::RuntimeSessionPersistence for TerminalDeliveryPersistence {
    async fn save_runtime_session(&self, session: &mut Session) -> std::io::Result<()> {
        if self.failure == TerminalDeliveryFailure::UnreceiptedExecutionError
            && crate::execution::ChildCompletionSource::has_source_record(session)
        {
            return Err(std::io::Error::other(
                "injected unfenced final-save rejection",
            ));
        }
        self.inner.save_runtime_session(session).await
    }

    async fn checkpoint_runtime_session(&self, session: &mut Session) -> std::io::Result<()> {
        if self.failure == TerminalDeliveryFailure::Checkpoint
            && session.last_run_status().as_deref() == Some("running")
            && crate::execution::ChildCompletionSource::has_source_record(session)
        {
            return Err(std::io::Error::other(
                "injected terminal checkpoint failure",
            ));
        }
        self.inner.checkpoint_runtime_session(session).await
    }

    async fn load_runtime_session(&self, session_id: &str) -> std::io::Result<Option<Session>> {
        self.inner.load_runtime_session(session_id).await
    }

    async fn clear_legacy_pending_messages(
        &self,
        session_id: &str,
        expected: &[serde_json::Value],
    ) -> std::io::Result<bool> {
        self.inner
            .clear_legacy_pending_messages(session_id, expected)
            .await
    }

    async fn append_token_usage_record(
        &self,
        session_id: &str,
        json_line: &str,
    ) -> std::io::Result<()> {
        self.inner
            .append_token_usage_record(session_id, json_line)
            .await
    }
}

struct TerminalDeliveryRunner {
    storage: Arc<dyn Storage>,
    failure: TerminalDeliveryFailure,
    calls: Arc<tokio::sync::Mutex<Vec<&'static str>>>,
}

#[async_trait]
impl ExternalChildRunner for TerminalDeliveryRunner {
    async fn should_handle(&self, _session: &Session) -> bool {
        true
    }

    async fn execute_external_child(
        &self,
        session: &mut Session,
        _job: &SpawnJob,
        _event_tx: tokio::sync::mpsc::Sender<AgentEvent>,
        _cancel_token: tokio_util::sync::CancellationToken,
    ) -> crate::runtime::runner::Result<()> {
        self.calls.lock().await.push("execute");
        session.add_message(Message::assistant("durable terminal answer", None));
        if self.failure == TerminalDeliveryFailure::UnreceiptedExecutionError {
            return Err(bamboo_agent_core::AgentError::LLM(
                "actor activation fence is stale or expired".into(),
            ));
        }
        Ok(())
    }

    async fn prepare_durable_child_delivery(
        &self,
        _session: &Session,
        _activation_run_id: &str,
    ) -> Result<bool, String> {
        self.calls.lock().await.push("prepare");
        if self.failure == TerminalDeliveryFailure::Prepare {
            return Err("injected receipt preparation failure".into());
        }
        if self.failure == TerminalDeliveryFailure::UnreceiptedExecutionError {
            return Ok(false);
        }
        Ok(true)
    }

    async fn commit_durable_child_delivery(
        &self,
        session: &Session,
        activation_run_id: &str,
    ) -> Result<(), String> {
        self.calls.lock().await.push("proof");
        let durable = self
            .storage
            .load_session(&session.id)
            .await
            .map_err(|error| error.to_string())?
            .ok_or("terminal checkpoint missing")?;
        if durable.last_run_status().as_deref() != Some("running")
            || !crate::execution::ChildCompletionSource::has_source_record(&durable)
            || crate::execution::ChildCompletionSource::from_committed_session(&durable).is_some()
            || durable.messages.last().map(|message| &message.content)
                != session.messages.last().map(|message| &message.content)
        {
            return Err("proof callback did not follow the deferred transcript checkpoint".into());
        }
        if self.failure == TerminalDeliveryFailure::Proof {
            return Err("injected Host proof commitment failure".into());
        }
        // The isolated fake emulates the real actor's proof-owned canonical
        // publication. The SDK must read this back, never publish it itself.
        let mut published = session.clone();
        if !crate::execution::ChildCompletionSource::publish_after_durable_delivery(
            &mut published,
            activation_run_id,
        )? {
            return Err("pending completion source missing".into());
        }
        self.storage
            .save_session(&published)
            .await
            .map_err(|error| error.to_string())?;
        self.calls.lock().await.push("published");
        Ok(())
    }

    async fn confirm_durable_child_delivery(
        &self,
        session: &Session,
        _activation_run_id: &str,
        history_committed: bool,
    ) -> Result<(), String> {
        self.calls.lock().await.push(if history_committed {
            "ack"
        } else {
            "unconfirmed"
        });
        if history_committed {
            if crate::execution::ChildCompletionSource::from_committed_session(session).is_none() {
                return Err("ACK preceded proven source publication".into());
            }
            return Err("injected transport ACK failure".into());
        }
        Ok(())
    }
}

struct TerminalDeliveryCompletion(
    tokio::sync::mpsc::UnboundedSender<crate::execution::ChildCompletion>,
);

#[async_trait]
impl crate::execution::ChildCompletionHandler for TerminalDeliveryCompletion {
    async fn on_child_completed(&self, completion: crate::execution::ChildCompletion) {
        let _ = self.0.send(completion);
    }
}

#[tokio::test]
async fn terminal_delivery_failure_matrix_gates_success_before_transport_ack() {
    for failure in [
        TerminalDeliveryFailure::UnreceiptedExecutionError,
        TerminalDeliveryFailure::Prepare,
        TerminalDeliveryFailure::Checkpoint,
        TerminalDeliveryFailure::Proof,
        TerminalDeliveryFailure::Ack,
    ] {
        let mut harness = build_harness(Arc::new(CompletedProvider), Vec::new(), &[]).await;
        let calls = Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let persistence = Arc::new(TerminalDeliveryPersistence {
            inner: harness.ctx.agent.persistence().clone(),
            failure,
        });
        harness.ctx.agent = Arc::new(harness.ctx.agent.with_execution_persistence(persistence));
        harness.ctx.external_child_runner = Arc::new(TerminalDeliveryRunner {
            storage: harness.storage.clone(),
            failure,
            calls: calls.clone(),
        });
        let (completion_tx, mut completion_rx) = tokio::sync::mpsc::unbounded_channel();
        harness.ctx.completion_handler = Some(Arc::new(TerminalDeliveryCompletion(completion_tx)));
        let child_tx = crate::runtime::execution::session_events::get_or_create_event_sender(
            &harness.ctx.session_event_senders,
            &harness.child_session_id,
        )
        .await;
        let mut child_rx = child_tx.subscribe();
        run_child_spawn(
            harness.ctx.clone(),
            SpawnJob {
                parent_session_id: harness.parent_session_id.clone(),
                child_session_id: harness.child_session_id.clone(),
                model: "gpt-5".into(),
                disabled_tools: None,
            },
        )
        .await
        .unwrap();
        let completion = tokio::time::timeout(Duration::from_secs(10), completion_rx.recv())
            .await
            .expect("terminal callback must settle")
            .unwrap();
        if failure == TerminalDeliveryFailure::UnreceiptedExecutionError {
            // The old owned-Actor rejection is the execution outcome. An
            // unrelated generic save failure must not replace that diagnostic
            // or claim durable history/ACK for the unfenced snapshot.
            assert_eq!(completion.status, "error");
            assert!(completion
                .error
                .as_deref()
                .is_some_and(|error| error.contains("actor activation fence is stale or expired")));
            assert!(completion.source.is_none());
            assert_eq!(
                *calls.lock().await,
                vec!["execute", "prepare", "unconfirmed"]
            );
            let durable = harness
                .storage
                .load_session(&harness.child_session_id)
                .await
                .unwrap()
                .unwrap();
            assert!(!durable
                .messages
                .iter()
                .any(|message| message.content == "durable terminal answer"));
            assert!(!crate::execution::ChildCompletionSource::has_source_record(
                &durable
            ));
            while let Ok(event) = child_rx.try_recv() {
                assert!(!matches!(event, AgentEvent::SessionHistoryCommitted { .. }));
            }
            let runners = harness.ctx.agent_runners.read().await;
            assert!(
                matches!(&runners.get(&harness.child_session_id).unwrap().status,
                crate::execution::AgentStatus::Error(error) if error.contains("actor activation fence is stale or expired"))
            );
            continue;
        }
        let success = failure == TerminalDeliveryFailure::Ack;
        assert_eq!(
            completion.status,
            if success { "completed" } else { "error" },
            "{failure:?}"
        );
        assert_eq!(completion.source.is_some(), success, "{failure:?}");
        assert_eq!(completion.error.is_none(), success, "{failure:?}");
        let events = collect_until_completed(&mut harness.parent_rx).await;
        assert!(
            events.iter().any(|event| matches!(event,
                AgentEvent::SubAgentCompleted { status, .. } if status == &completion.status
            )),
            "{failure:?}"
        );
        let runners = harness.ctx.agent_runners.read().await;
        let runner = runners.get(&harness.child_session_id).unwrap();
        assert_eq!(
            matches!(runner.status, crate::execution::AgentStatus::Completed),
            success,
            "{failure:?}"
        );
        if !success {
            assert!(
                matches!(runner.status, crate::execution::AgentStatus::Error(_)),
                "{failure:?}"
            );
        }
        drop(runners);
        let durable = harness
            .storage
            .load_session(&harness.child_session_id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(
            durable.last_run_status().as_deref(),
            Some(completion.status.as_str()),
            "{failure:?}"
        );
        assert_eq!(
            crate::execution::ChildCompletionSource::from_committed_session(&durable).is_some(),
            success,
            "{failure:?}"
        );
        assert!(
            crate::execution::ChildCompletionSource::has_source_record(&durable),
            "{failure:?}"
        );
        assert_eq!(
            durable
                .messages
                .iter()
                .filter(|message| message.content == "durable terminal answer")
                .count(),
            1,
            "{failure:?}"
        );
        let mut history_barriers = 0;
        while let Ok(event) = child_rx.try_recv() {
            if matches!(event, AgentEvent::SessionHistoryCommitted { .. }) {
                history_barriers += 1;
            }
        }
        assert_eq!(history_barriers, usize::from(success), "{failure:?}");
        let expected = match failure {
            TerminalDeliveryFailure::UnreceiptedExecutionError => unreachable!(),
            TerminalDeliveryFailure::Prepare | TerminalDeliveryFailure::Checkpoint => {
                vec!["execute", "prepare", "unconfirmed"]
            }
            TerminalDeliveryFailure::Proof => vec!["execute", "prepare", "proof", "unconfirmed"],
            TerminalDeliveryFailure::Ack => vec!["execute", "prepare", "proof", "published", "ack"],
        };
        assert_eq!(*calls.lock().await, expected, "{failure:?}");
    }
}