bamboo-engine 2026.10.22

use crate::runtime::config::{AgentLoopConfig, AuxiliaryModelConfig};
use crate::runtime::gold_evaluation::{
    AsyncGoldEvaluationRequest, AsyncGoldEvaluationResult, GoldEvaluationTerminal,
};
use crate::runtime::runner::task_lifecycle::{
    AsyncTaskEvaluationRequest, AsyncTaskEvaluationResult,
};
use crate::runtime::task_context::TaskLoopContext;
use bamboo_agent_core::tools::ToolExecutor;
use bamboo_agent_core::{AgentEvent, Session};
use bamboo_domain::{AgentRuntimeState, AgentStatusState};
use bamboo_llm::LLMProvider;
use bamboo_metrics::MetricsCollector;
use std::sync::atomic::AtomicBool;
use std::sync::Arc;

use super::super::logging::DebugLogger;
use crate::runtime::runner::state_bridge;

const MAX_CONSECUTIVE_OVERFLOW_RECOVERIES: usize = 3;

#[derive(Debug, Clone, Default)]
pub(super) struct OverflowRecoveryState {
    pub(super) total_recoveries: usize,
    pub(super) consecutive_recoveries: usize,
    pub(super) last_recovered_round: Option<usize>,
}

impl OverflowRecoveryState {
    pub(super) fn can_attempt_recovery(&self) -> bool {
        self.consecutive_recoveries < MAX_CONSECUTIVE_OVERFLOW_RECOVERIES
    }

    pub(super) fn record_recovery(&mut self, round: usize) {
        self.total_recoveries += 1;
        self.consecutive_recoveries += 1;
        self.last_recovered_round = Some(round);
    }

    pub(super) fn reset_after_stable_round(&mut self) {
        self.consecutive_recoveries = 0;
    }
}

pub(super) struct InFlightTaskEvaluation {
    pub(super) request: AsyncTaskEvaluationRequest,
    /// Set only at the provider-dispatch boundary, after the low-priority
    /// budget has admitted this request.
    pub(super) metrics_started: Arc<AtomicBool>,
    /// Exactly-once guard shared with the completed result.
    pub(super) metrics_terminal: Arc<AtomicBool>,
    // `Option<..>` output: the spawned future `select!`s on the run's
    // `CancellationToken`, so a cancelled run yields `None` (the eval work was
    // dropped at an await point, stopping the wasted LLM spend) instead of a
    // completed result. See `spawn_task_evaluation_request`.
    pub(super) join_handle: tokio::task::JoinHandle<Option<AsyncTaskEvaluationResult>>,
}

pub(super) struct InFlightGoldEvaluation {
    pub(super) request: AsyncGoldEvaluationRequest,
    /// Shared with the async evaluator to claim one cancellation event.
    pub(super) terminal_claimed: Arc<GoldEvaluationTerminal>,
    // None is run cancellation; Err is a known local routing refusal, which
    // must never be harvested as a successful evaluator result.
    pub(super) join_handle: tokio::task::JoinHandle<
        Option<Result<AsyncGoldEvaluationResult, bamboo_agent_core::AgentError>>,
    >,
}

/// An evaluation waiting behind an in-flight one retains the exact target it
/// was prepared for. Compare underlying allocations, not budget identity;
/// account identity within a provider is a separate contract.
pub(super) struct QueuedEvaluation<R> {
    pub(super) request: R,
    pub(super) provider: Arc<dyn LLMProvider>,
    pub(super) model_name: String,
    route: bamboo_llm::provider::QueuedRouteIdentity,
    require_current: bool,
}

impl<R> QueuedEvaluation<R> {
    pub(super) fn new(
        request: R,
        provider: Arc<dyn LLMProvider>,
        model_name: String,
    ) -> Option<Self> {
        use bamboo_llm::provider::{queued_identity_for, queued_snapshot_for};
        let provider = queued_snapshot_for(provider).ok()?;
        let route = queued_identity_for(&provider);
        let require_current = provider.queue_requires_current_route();
        Some(Self {
            request,
            provider,
            model_name,
            route,
            require_current,
        })
    }

    pub(super) fn current_provider(
        &self,
        provider: Arc<dyn LLMProvider>,
        model_name: &str,
    ) -> Option<Arc<dyn LLMProvider>> {
        // An opaque SDK provider has no authoritative live registry. Preserve
        // its captured pair; never replace it with a newly returned provider.
        if !self.require_current {
            // An explicit local refusal is never treated as an unset route.
            if provider.known_pre_dispatch_error().is_some() {
                return None;
            }
            return Some(self.provider.clone());
        }
        if self.model_name != model_name {
            return None;
        }
        use bamboo_llm::provider::{queued_identity_for, queued_snapshot_for, QueuedRouteIdentity};
        let provider = queued_snapshot_for(provider).ok()?;
        if !provider.queue_requires_current_route() {
            return None;
        }
        let same = match (&self.route, queued_identity_for(&provider)) {
            (QueuedRouteIdentity::Stable(left), QueuedRouteIdentity::Stable(right)) => {
                Arc::ptr_eq(left, &right)
            }
            _ => false,
        };
        same.then_some(provider)
    }
}

#[derive(Default)]
pub(super) struct TaskEvaluationState {
    pub(super) in_flight: Option<InFlightTaskEvaluation>,
    pub(super) completed: Option<AsyncTaskEvaluationResult>,
    pub(super) queued_request: Option<QueuedEvaluation<AsyncTaskEvaluationRequest>>,
}

#[derive(Default)]
pub(super) struct GoldEvaluationState {
    pub(super) in_flight: Option<InFlightGoldEvaluation>,
    pub(super) completed: Option<AsyncGoldEvaluationResult>,
    pub(super) queued_request: Option<QueuedEvaluation<AsyncGoldEvaluationRequest>>,
}

pub(super) struct LoopRunState {
    pub(super) session_id: String,
    /// Collision-resistant namespace for metrics emitted by this logical
    /// pipeline execution. Round counters restart on resume/re-execution, so
    /// they are only unique within this private run scope.
    pub(super) execution_id: String,
    pub(super) current_inputs:
        Option<crate::runtime::managers::lifecycle::BoundedInputRequestBatch>,
    pub(super) model_name: String,
    pub(super) metrics_collector: Option<MetricsCollector>,
    pub(super) debug_logger: DebugLogger,
    pub(super) task_context: Option<TaskLoopContext>,
    pub(super) overflow_recovery: OverflowRecoveryState,
    pub(super) task_evaluation: TaskEvaluationState,
    pub(super) gold_evaluation: GoldEvaluationState,
    pub(super) auxiliary_models: AuxiliaryModelConfig,
    /// Structured runtime state persisted alongside the session.
    pub(super) runtime_state: AgentRuntimeState,
}

pub(super) fn resolve_auxiliary_models(config: &AgentLoopConfig) -> AuxiliaryModelConfig {
    config.resolve_auxiliary_models()
}

pub(super) async fn initialize_loop_state(
    session: &mut Session,
    initial_message: &str,
    config: &AgentLoopConfig,
    tools: &dyn ToolExecutor,
    event_tx: &tokio::sync::mpsc::Sender<AgentEvent>,
) -> super::super::Result<LoopRunState> {
    let debug_logger = DebugLogger::new(tracing::enabled!(tracing::Level::DEBUG));
    let session_id = session.id.clone();
    // Take before fallible startup work: failed initialization drops the value.
    let (execution_id, observation) = config
        .initial_untrusted_inputs
        .as_ref()
        .and_then(|inputs| inputs.take_startup_observation(&session_id))
        .unwrap_or_else(|| {
            (
                crate::runtime::runner::round_prelude::new_execution_id(),
                Default::default(),
            )
        });
    let metrics_collector = config.metrics_collector.clone();
    let model_name = config
        .model_name
        .clone()
        .unwrap_or_else(|| "unknown".to_string());

    super::super::metrics_lifecycle::record_session_started(
        metrics_collector.as_ref(),
        &session_id,
        &model_name,
        session.created_at,
        session.messages.len() as u32,
    );

    // A resumed User can be an admitted peer envelope. Keep diagnostics free
    // of prompt/body/routing text without changing the provider-facing input.
    tracing::debug!(
        session_id = %session_id,
        initial_message_bytes = initial_message.len(),
        "Starting agent loop"
    );
    debug_logger.log_event(
        &session_id,
        "agent_loop_start",
        serde_json::json!({
            "message_bytes": initial_message.len(),
            "round_cap": config.run_budget.max_rounds,
            "initial_message_count": session.messages.len(),
        }),
    );

    let auxiliary_models = resolve_auxiliary_models(config);

    let must_resume_pinned_activation = session
        .agent_runtime_state
        .as_ref()
        .is_some_and(|previous| matches!(previous.status, AgentStatusState::Suspended));
    let local_resources = bamboo_tools::LocalTaskResources::current();
    if let Some(resources) = &local_resources {
        let owner = resources.identity();
        if owner.session_id != session.id
            || owner.child_birth != session.created_at
            || session.kind != bamboo_domain::SessionKind::Child
        {
            return Err(bamboo_agent_core::AgentError::LLM(
                "local execution resource identity does not match canonical Child".into(),
            ));
        }
        if config
            .ticket_worker_plan
            .as_ref()
            .is_some_and(|plan| plan.run_id() != owner.run_id)
        {
            return Err(bamboo_agent_core::AgentError::LLM(
                "local Ticket execution run identity changed".into(),
            ));
        }
    }
    let mut runtime_state = AgentRuntimeState::new(
        local_resources
            .as_ref()
            .map(|scope| scope.identity().run_id.as_str())
            .or_else(|| config.ticket_worker_plan.as_ref().map(|plan| plan.run_id()))
            .unwrap_or(&session_id),
    );
    // Persist the actual pipeline UUID before its first mirror/checkpoint.
    // Ordinary executions deliberately do not inherit a prior counted owner.
    if config.effective_counted_loop().is_some() || config.loop_control.is_some() {
        runtime_state.execution_id = Some(execution_id.clone());
    }
    // Permission mode is a per-session sticky posture (set via PATCH /sessions
    // and persisted in runtime.json). Each run rebuilds a fresh runtime state,
    // so carry the exact typed mode forward instead of resetting it.
    let permission_mode = session
        .agent_runtime_state
        .as_ref()
        .map(|prev| prev.effective_permission_mode())
        .unwrap_or_default();
    runtime_state.set_permission_mode(permission_mode);
    // #73: "no interactive human approver" (headless / scheduled / deployed) is
    // likewise a sticky per-session flag; carry it forward so every run — and the
    // sub-agents it spawns (which inherit it) — route gated actions to the
    // off-loop model-reviewer instead of escalating to an absent human.
    runtime_state.no_human_approver = session
        .agent_runtime_state
        .as_ref()
        .is_some_and(|prev| prev.no_human_approver);
    // Preserve the incoming typed Child posture when rebuilding this run.
    // Legacy metadata is not authority for granting read-only mode.
    runtime_state.read_only = session
        .agent_runtime_state
        .as_ref()
        .is_some_and(|previous| previous.read_only);
    // Server-owned UserPromptSubmit runs before the engine loop and records
    // into the session state. Carry those current-turn checkpoints into the
    // fresh runner-owned state. This also preserves hook context/checkpoints
    // when a suspended run resumes without a new user prompt.
    if let Some(previous) = session.agent_runtime_state.as_ref() {
        runtime_state.checkpoints = previous.checkpoints.clone();
        runtime_state.hook_contexts = previous.hook_contexts.clone();
        runtime_state.stop_hook_forced_continuations = previous.stop_hook_forced_continuations;
        // An interrupted wait remains durable during this reasoning turn.
        // The inbox coordinator prepares an explicit interruption by clearing
        // the suspension and setting Idle while retaining the existing wait.
        // Preserve that exact lease, including an untagged safety-net wait.
        // The previous run can have ended by cancellation or error: its
        // terminal label does not undo the coordinator's prepared interruption.
        // Ordinary Suspended untagged waits still follow the startup cleanup.
        let prepared_wait_interrupt =
            previous.status == AgentStatusState::Idle && previous.suspension.is_none();
        runtime_state.waiting_for_children = previous
            .waiting_for_children
            .as_ref()
            .filter(|wait| wait.registered_by_tool_call_id.is_some() || prepared_wait_interrupt)
            .cloned();
    }
    runtime_state.llm.model_name = Some(model_name.clone());
    runtime_state.llm.provider_name = config.provider_name.clone();
    runtime_state.llm.fast_model_name = auxiliary_models.fast_model_name.clone();
    runtime_state.llm.background_model_name = auxiliary_models.background_model_name.clone();
    runtime_state.round.max_rounds = config.run_budget.max_rounds;
    state_bridge::sync_from_metadata(session, &mut runtime_state);
    runtime_state.status = AgentStatusState::Initializing;
    state_bridge::write_runtime_state(session, &runtime_state);
    runtime_state.status = AgentStatusState::Running;

    let task_context = super::super::session_setup::prepare_session_for_loop(
        session,
        initial_message,
        config,
        tools,
        metrics_collector.as_ref(),
        &session_id,
        &debug_logger,
        must_resume_pinned_activation,
        event_tx,
    )
    .await?;

    let mut current_inputs = None;
    observation.update_current(&mut current_inputs, &session_id, &execution_id);
    Ok(LoopRunState {
        session_id,
        execution_id,
        current_inputs,
        model_name,
        metrics_collector,
        debug_logger,
        task_context,
        overflow_recovery: OverflowRecoveryState::default(),
        task_evaluation: TaskEvaluationState::default(),
        gold_evaluation: GoldEvaluationState::default(),
        auxiliary_models,
        runtime_state,
    })
}

#[cfg(test)]
mod tests {
    #[tokio::test]
    async fn ql_sealed_startup_adopts_exact_measured_uuid_once_and_keeps_ie_view() {
        use bamboo_agent_core::storage::Storage;
        use bamboo_domain::{RuntimeSessionPersistence, SessionInboxPort};
        let home = tempfile::tempdir().unwrap();
        let store = Arc::new(
            bamboo_storage::SessionStoreV2::new(home.path().into())
                .await
                .unwrap(),
        );
        let storage: Arc<dyn Storage> = store.clone();
        let persistence: Arc<dyn RuntimeSessionPersistence> =
            Arc::new(bamboo_storage::LockedSessionStore::new(storage.clone()));
        let inbox: Arc<dyn SessionInboxPort> = Arc::new(bamboo_storage::FileSessionInbox::new(
            store,
            bamboo_domain::SessionInboxLimits::default(),
        ));
        let mut session = bamboo_agent_core::Session::new("ql-startup", "model");
        session.add_message(bamboo_agent_core::Message::system("system"));
        storage.save_session(&session).await.unwrap();
        let mut envelope =
            bamboo_domain::SessionMessageEnvelope::user_input(&session.id, "new queued input");
        let request = bamboo_domain::SessionSkillRequest {
            mode: Some("original-mode".into()),
            selections: vec![bamboo_domain::SessionSkillSelection {
                id: "request-outside-host-ceiling".into(),
                source: "plugin".into(),
                revision: 7,
                args: serde_json::json!({"original":[1.125,null,"原样"]}),
            }],
        };
        if let bamboo_domain::SessionMessageBody::Content(content) = &mut envelope.body {
            content.skill_request = Some(request.clone());
        }
        let receipt = inbox.deliver(&envelope).await.unwrap();
        inbox
            .mark_activation_eligible(
                &session.id,
                receipt.generation,
                bamboo_domain::SessionActivationPolicy::InterruptSpecificWait,
            )
            .await
            .unwrap();
        let (refresh, carrier) =
            crate::runtime::config::UntrustedExecutionInputs::admit_with_startup_observation(
                &mut session,
                Some(&storage),
                Some(&persistence),
                Some(&inbox),
            )
            .await;
        assert_eq!(refresh.merged, 1);
        let carrier = carrier.unwrap();
        assert_eq!(carrier.observations()[0].input_id(), envelope.id.as_str());
        let config = AgentLoopConfig {
            initial_untrusted_inputs: Some(carrier),
            selected_skill_ids: Some(Vec::new()),
            skip_initial_user_message: true,
            storage: Some(storage),
            persistence: Some(persistence),
            session_inbox: Some(inbox),
            ..Default::default()
        };
        let (tx, _rx) = tokio::sync::mpsc::channel(64);
        let first = initialize_loop_state(
            &mut session,
            "",
            &config,
            &SuccessfulLoadSkill::default(),
            &tx,
        )
        .await
        .unwrap();
        let batch = first.current_inputs.as_ref().unwrap();
        assert_eq!(batch.execution_id(), first.execution_id);
        assert_eq!(first.execution_id.len(), 32);
        assert_eq!(batch.session_id(), first.session_id);
        assert_eq!(batch.records()[0].input_id, envelope.id.as_str());
        assert_eq!(batch.records()[0].request.as_ref(), Some(&request));
        assert_eq!(
            config.initial_untrusted_inputs().unwrap().observations()[0].request(),
            Some(&request)
        );
        assert_eq!(
            config.selected_skill_ids,
            Some(Vec::new()),
            "request data never widens host ceiling"
        );
        assert_eq!(
            config.initial_untrusted_inputs().unwrap().observations()[0].input_id(),
            envelope.id.as_str()
        );
        let second = initialize_loop_state(
            &mut session,
            "",
            &config,
            &SuccessfulLoadSkill::default(),
            &tx,
        )
        .await
        .unwrap();
        assert!(
            second.current_inputs.is_none(),
            "spent initial data cannot regrant on successor/restart"
        );
        assert_ne!(first.execution_id, second.execution_id);
    }

    #[tokio::test]
    async fn ql_old_ie_and_history_start_absent_for_each_host_ceiling() {
        for ceiling in [None, Some(Vec::new()), Some(vec!["host-cap".into()])] {
            let id = "ql-old-input";
            let old = crate::runtime::config::UntrustedExecutionInputs::new(vec![
                crate::runtime::config::UntrustedInputObservation::new("unsealed-N", None).unwrap(),
            ])
            .unwrap();
            let mut session = bamboo_agent_core::Session::new(id, "model");
            let envelope =
                bamboo_domain::SessionMessageEnvelope::user_input(id, "historical canonical data");
            session.add_message(envelope.to_provider_message().unwrap());
            session
                .metadata
                .insert("selected_skill_ids".into(), "[\"old\"]".into());
            let config = AgentLoopConfig {
                initial_untrusted_inputs: Some(old),
                selected_skill_ids: ceiling.clone(),
                skip_initial_user_message: true,
                ..Default::default()
            };
            let (tx, _rx) = tokio::sync::mpsc::channel(64);
            let state = initialize_loop_state(
                &mut session,
                "",
                &config,
                &SuccessfulLoadSkill::default(),
                &tx,
            )
            .await
            .unwrap();
            assert!(state.current_inputs.is_none());
            assert_eq!(config.selected_skill_ids, ceiling);
        }
    }

    use super::{initialize_loop_state, resolve_auxiliary_models, OverflowRecoveryState};
    use crate::runtime::config::{AgentLoopConfig, AuxiliaryModelConfig};
    use async_trait::async_trait;
    use bamboo_agent_core::tools::{
        FunctionSchema, ToolCall, ToolExecutionSessionFlags, ToolExecutor, ToolResult, ToolSchema,
    };
    use bamboo_agent_core::Session;
    use bamboo_domain::{
        record_permission_audit, resolve_permission_mode, AgentRuntimeState, AgentStatusState,
        PermissionAuditSeed, PermissionAuditSnapshot, PermissionMode, SessionPermissionMode,
    };
    use bamboo_skills::runtime_metadata::{
        LOADED_SKILL_IDS_METADATA_KEY, SKILL_RUNTIME_ACTIVATION_GENERATION_KEY,
        SKILL_RUNTIME_SELECTED_SKILL_MODE_KEY, SKILL_RUNTIME_SELECTED_SKILL_REVISIONS_KEY,
        SKILL_RUNTIME_SELECTION_SOURCE_KEY,
    };
    use bamboo_skills::{SkillManager, SkillStoreConfig};
    use std::collections::BTreeSet;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{Arc, Mutex};

    #[derive(Default)]
    struct SuccessfulLoadSkill(AtomicUsize);

    #[async_trait]
    impl ToolExecutor for SuccessfulLoadSkill {
        async fn execute(
            &self,
            _call: &ToolCall,
        ) -> bamboo_agent_core::tools::executor::Result<ToolResult> {
            self.0.fetch_add(1, Ordering::SeqCst);
            Ok(ToolResult {
                success: true,
                result: serde_json::json!({"instructions": "pinned"}).to_string(),
                display_preference: None,
                images: Vec::new(),
            })
        }

        fn list_tools(&self) -> Vec<ToolSchema> {
            vec![ToolSchema {
                schema_type: "function".to_string(),
                function: FunctionSchema {
                    name: "load_skill".to_string(),
                    description: "load".to_string(),
                    parameters: serde_json::json!({"type": "object"}),
                },
            }]
        }
    }

    #[test]
    fn overflow_recovery_state_tracks_recoveries_and_resets() {
        let mut state = OverflowRecoveryState::default();
        assert!(state.can_attempt_recovery());

        state.record_recovery(0);
        assert_eq!(state.total_recoveries, 1);
        assert_eq!(state.consecutive_recoveries, 1);
        assert_eq!(state.last_recovered_round, Some(0));
        assert!(state.can_attempt_recovery());

        state.record_recovery(1);
        state.record_recovery(2);
        assert_eq!(state.total_recoveries, 3);
        assert_eq!(state.consecutive_recoveries, 3);
        assert!(!state.can_attempt_recovery());

        state.reset_after_stable_round();
        assert_eq!(state.consecutive_recoveries, 0);
        assert!(state.can_attempt_recovery());
        assert_eq!(state.total_recoveries, 3);
    }

    #[test]
    fn auxiliary_model_resolver_returns_latest_values() {
        let counter = Arc::new(Mutex::new(0usize));
        let counter_for_resolver = counter.clone();
        let config = AgentLoopConfig {
            auxiliary_model_resolver: Some(Arc::new(move || {
                let mut guard = counter_for_resolver.lock().expect("counter lock");
                *guard += 1;
                AuxiliaryModelConfig {
                    fast_model_name: Some(format!("fast-{}", *guard)),
                    background_model_name: Some(format!("bg-{}", *guard)),
                    summarization_model_name: Some(format!("sum-{}", *guard)),
                    ..Default::default()
                }
            })),
            ..Default::default()
        };

        let first = resolve_auxiliary_models(&config);
        let second = resolve_auxiliary_models(&config);

        assert_eq!(first.fast_model_name.as_deref(), Some("fast-1"));
        assert_eq!(first.background_model_name.as_deref(), Some("bg-1"));
        assert_eq!(first.summarization_model_name.as_deref(), Some("sum-1"));
        assert_eq!(second.fast_model_name.as_deref(), Some("fast-2"));
        assert_eq!(second.background_model_name.as_deref(), Some("bg-2"));
        assert_eq!(second.summarization_model_name.as_deref(), Some("sum-2"));
    }

    #[test]
    fn auxiliary_model_resolver_refreshes_summarization_model_between_calls() {
        let counter = Arc::new(Mutex::new(0usize));
        let counter_for_resolver = counter.clone();
        let config = AgentLoopConfig {
            auxiliary_model_resolver: Some(Arc::new(move || {
                let mut guard = counter_for_resolver.lock().expect("counter lock");
                *guard += 1;
                AuxiliaryModelConfig {
                    summarization_model_name: Some(format!("sum-{}", *guard)),
                    ..Default::default()
                }
            })),
            ..Default::default()
        };

        let first = resolve_auxiliary_models(&config);
        let second = resolve_auxiliary_models(&config);

        assert_eq!(first.summarization_model_name.as_deref(), Some("sum-1"));
        assert_eq!(second.summarization_model_name.as_deref(), Some("sum-2"));
    }

    #[tokio::test]
    async fn native_startup_uses_exact_owned_run_and_rejects_foreign_session() {
        let parent = Session::new("native-startup-parent", "model");
        let mut child = Session::new_child_of("native-startup-child", &parent, "model", "task");
        let resources =
            bamboo_tools::LocalTaskResources::new(&child.id, "canonical-run", child.created_at);
        let config = AgentLoopConfig::default();
        let tools = SuccessfulLoadSkill::default();
        let (tx, _rx) = tokio::sync::mpsc::channel(8);
        let state = resources
            .scope(initialize_loop_state(&mut child, "", &config, &tools, &tx))
            .await
            .unwrap();
        assert_eq!(state.runtime_state.run_id, "canonical-run");
        assert_eq!(
            child.agent_runtime_state.as_ref().unwrap().run_id,
            "canonical-run"
        );
        let mut foreign = Session::new_child_of("foreign-native-child", &parent, "model", "task");
        assert!(resources
            .scope(initialize_loop_state(
                &mut foreign,
                "",
                &config,
                &tools,
                &tx
            ))
            .await
            .is_err());
        resources.close_and_drain().await.unwrap();
    }

    #[tokio::test]
    async fn startup_preserves_typed_read_only_without_legacy_metadata_grant() {
        for prior in [Some(true), Some(false), None] {
            let mut session = Session::new("read-only-startup", "model");
            session.agent_runtime_state = prior.map(|read_only| {
                let mut runtime = AgentRuntimeState::new("previous-run");
                runtime.read_only = read_only;
                runtime
            });
            let mut legacy = AgentRuntimeState::new("legacy-run");
            legacy.read_only = true;
            session.metadata.insert(
                "agent.runtime.state".into(),
                serde_json::to_string(&legacy).unwrap(),
            );
            let tools = SuccessfulLoadSkill::default();
            let config = AgentLoopConfig::default();
            let (event_tx, _event_rx) = tokio::sync::mpsc::channel(8);

            let loop_state =
                initialize_loop_state(&mut session, "inspect only", &config, &tools, &event_tx)
                    .await
                    .expect("actual runner startup");

            for runtime in [
                &loop_state.runtime_state,
                session
                    .agent_runtime_state
                    .as_ref()
                    .expect("startup publishes typed runtime"),
            ] {
                assert_eq!(runtime.read_only, prior.unwrap_or(false), "prior={prior:?}");
                assert_eq!(runtime.run_id, session.id, "fresh run identity");
            }
        }
    }

    #[tokio::test]
    async fn startup_carries_scheduled_auto_no_human_and_audit_into_fresh_loop_state() {
        let mut session = Session::new("scheduled-startup", "model");
        let runtime = session.agent_runtime_state.get_or_insert_default();
        runtime.set_permission_mode(SessionPermissionMode::Auto);
        runtime.no_human_approver = true;
        record_permission_audit(
            &mut session.metadata,
            &PermissionAuditSeed::bamboo_runtime(
                17,
                resolve_permission_mode(SessionPermissionMode::Auto, PermissionMode::Default),
            ),
            Some("2026-08-04T00:00:00Z"),
        )
        .unwrap();
        let audit_before = PermissionAuditSnapshot::from_metadata(&session.metadata).unwrap();
        let tools = SuccessfulLoadSkill::default();
        let config = AgentLoopConfig::default();
        let (event_tx, _event_rx) = tokio::sync::mpsc::channel(8);

        let loop_state =
            initialize_loop_state(&mut session, "scheduled task", &config, &tools, &event_tx)
                .await
                .expect("scheduled startup");

        for runtime in [
            &loop_state.runtime_state,
            session
                .agent_runtime_state
                .as_ref()
                .expect("startup writes runtime state"),
        ] {
            assert_eq!(runtime.permission_mode, SessionPermissionMode::Auto);
            assert!(
                runtime.bypass_permissions,
                "legacy Auto compatibility mirror"
            );
            assert!(runtime.no_human_approver);
        }
        assert_eq!(
            ToolExecutionSessionFlags::from_session(&session),
            ToolExecutionSessionFlags {
                bypass_permissions: false,
                auto_approve_permissions: true,
                plan_read_only: false,
            }
        );
        assert_eq!(
            PermissionAuditSnapshot::from_metadata(&session.metadata).unwrap(),
            audit_before
        );
    }

    #[tokio::test]
    async fn startup_preserves_prepared_interrupt_wait_without_reviving_suspended_untagged_wait() {
        use bamboo_agent_core::storage::Storage;
        use bamboo_domain::{ChildWaitPolicy, WaitingForChildrenState};

        for (tagged, prepared_interrupt, last_status) in [
            (true, false, "suspended"),
            (false, false, "suspended"),
            (false, true, "suspended"),
            (false, true, "cancelled"),
            (false, true, "error"),
        ] {
            let directory = tempfile::tempdir().unwrap();
            let skills_dir = directory.path().join("skills");
            std::fs::create_dir_all(&skills_dir).unwrap();
            let manager = Arc::new(SkillManager::with_config(SkillStoreConfig {
                skills_dir,
                ..Default::default()
            }));
            manager.initialize().await.unwrap();
            let storage = Arc::new(
                bamboo_storage::SessionStoreV2::new(directory.path().join("sessions"))
                    .await
                    .unwrap(),
            );
            let locked = Arc::new(bamboo_storage::LockedSessionStore::new(storage.clone()));
            let repo = Arc::new(crate::SessionRepository::new(
                Arc::default(),
                storage.clone(),
                locked,
            ));
            let config = AgentLoopConfig {
                skill_manager: Some(manager),
                storage: Some(storage.clone()),
                persistence: Some(repo),
                ..Default::default()
            };
            let mut session = Session::new("startup-interrupted-wait", "model");
            let mut wait = WaitingForChildrenState::for_children(
                vec!["child".into()],
                ChildWaitPolicy::FirstError,
                chrono::Utc::now() - chrono::Duration::minutes(5),
            );
            if tagged {
                wait.registered_by_tool_call_id = Some("original-tool-call".into());
            }
            session.set_last_run_status(last_status);
            let runtime = session.agent_runtime_state.get_or_insert_default();
            runtime.status = if prepared_interrupt {
                AgentStatusState::Idle
            } else {
                AgentStatusState::Suspended
            };
            runtime.waiting_for_children = Some(wait.clone());
            storage.save_session(&session).await.unwrap();
            let tools = SuccessfulLoadSkill::default();
            let (event_tx, _event_rx) = tokio::sync::mpsc::channel(8);
            let state = initialize_loop_state(
                &mut session,
                "reason about the request",
                &config,
                &tools,
                &event_tx,
            )
            .await
            .unwrap();
            assert_eq!(
                session
                    .metadata
                    .get(SKILL_RUNTIME_SELECTION_SOURCE_KEY)
                    .map(String::as_str),
                Some("auto")
            );
            let saved = storage.load_session(&session.id).await.unwrap().unwrap();
            let expected = (tagged || prepared_interrupt).then_some(&wait);
            for runtime in [
                &state.runtime_state,
                session.agent_runtime_state.as_ref().unwrap(),
                saved.agent_runtime_state.as_ref().unwrap(),
            ] {
                assert_eq!(
                    runtime.waiting_for_children.as_ref(),
                    expected,
                    "prepared={prepared_interrupt}, tagged={tagged}, last_status={last_status}"
                );
                assert!(runtime.suspension.is_none());
            }
            assert_eq!(state.runtime_state.status, AgentStatusState::Running);
            assert_eq!(
                saved.agent_runtime_state.as_ref().unwrap().status,
                AgentStatusState::Initializing
            );
            assert!(!session.metadata.contains_key("runtime.suspend_reason"));
            assert_eq!(tools.0.load(Ordering::SeqCst), 0);
        }
    }

    #[tokio::test]
    async fn startup_reuses_retained_activation_and_explicit_selection_supersedes_it() {
        let directory = tempfile::tempdir().expect("tempdir");
        let skills_dir = directory.path().join("skills");
        for (id, prompt) in [("review-demo", "review N"), ("plan-demo", "plan N")] {
            let root = skills_dir.join(id);
            tokio::fs::create_dir_all(&root).await.expect("skill root");
            tokio::fs::write(
                root.join("SKILL.md"),
                format!("---\nname: {id}\ndescription: {id}\n---\n{prompt}\n"),
            )
            .await
            .expect("skill definition");
        }
        let manager = Arc::new(SkillManager::with_config(SkillStoreConfig {
            skills_dir: skills_dir.clone(),
            ..Default::default()
        }));
        manager.initialize().await.expect("initialize");
        let tools = SuccessfulLoadSkill::default();
        let review_config = AgentLoopConfig {
            skill_manager: Some(manager.clone()),
            selected_skill_ids: Some(vec!["review-demo".to_string()]),
            disabled_skill_ids: BTreeSet::new(),
            ..Default::default()
        };
        let mut session = Session::new("retained-session", "model");
        let (event_tx, _event_rx) = tokio::sync::mpsc::channel(8);
        initialize_loop_state(&mut session, "review", &review_config, &tools, &event_tx)
            .await
            .expect("initial selection");
        assert_eq!(
            tools.0.load(Ordering::SeqCst),
            0,
            "session setup must not execute load_skill on the model's behalf"
        );
        assert!(
            crate::runtime::runner::session_setup::skill_context::explicit_activation_pending(
                &session
            )
        );
        let pinned_generation = session
            .metadata
            .get(SKILL_RUNTIME_ACTIVATION_GENERATION_KEY)
            .cloned()
            .expect("pinned generation metadata");
        let pinned_revisions = session
            .metadata
            .get(SKILL_RUNTIME_SELECTED_SKILL_REVISIONS_KEY)
            .cloned()
            .expect("pinned revision metadata");
        assert_eq!(
            manager
                .store()
                .pinned_activation_skills("retained-session")
                .await
                .expect("review pin")
                .0[0]
                .prompt,
            "review N"
        );

        tokio::fs::write(
            skills_dir.join("review-demo/SKILL.md"),
            "---\nname: review-demo\ndescription: review N+1\n---\nreview N+1\n",
        )
        .await
        .expect("review N+1");
        manager.store().reload().await.expect("reload N+1");
        let mut suspended = AgentRuntimeState::new("retained-session");
        suspended.status = AgentStatusState::Suspended;
        session.agent_runtime_state = Some(suspended);
        session.metadata.remove(LOADED_SKILL_IDS_METADATA_KEY);
        let continuation_config = AgentLoopConfig {
            skill_manager: Some(manager.clone()),
            disabled_skill_ids: BTreeSet::new(),
            ..Default::default()
        };
        initialize_loop_state(
            &mut session,
            "plain clarification reply",
            &continuation_config,
            &tools,
            &event_tx,
        )
        .await
        .expect("suspended continuation");
        assert_eq!(
            manager
                .store()
                .pinned_activation_skills("retained-session")
                .await
                .expect("retained review pin")
                .0[0]
                .prompt,
            "review N",
            "startup overwrites Suspended with Initializing, but retained pin must still win"
        );
        assert_eq!(
            session
                .metadata
                .get(SKILL_RUNTIME_SELECTION_SOURCE_KEY)
                .map(String::as_str),
            Some("explicit")
        );
        assert_eq!(
            session
                .metadata
                .get(SKILL_RUNTIME_ACTIVATION_GENERATION_KEY),
            Some(&pinned_generation)
        );
        assert_eq!(
            session
                .metadata
                .get(SKILL_RUNTIME_SELECTED_SKILL_REVISIONS_KEY),
            Some(&pinned_revisions)
        );
        assert!(!session
            .metadata
            .contains_key(SKILL_RUNTIME_SELECTED_SKILL_MODE_KEY));
        assert!(!session.metadata.contains_key(LOADED_SKILL_IDS_METADATA_KEY));
        assert!(
            crate::runtime::runner::session_setup::skill_context::explicit_activation_pending(
                &session
            ),
            "resuming a pinned but not-yet-loaded selection must still require a model-issued activation"
        );
        assert!(crate::runtime::runner::session_setup::prompt_envelope::build_active_workflow_context_block(&session).is_none());
        assert_eq!(
            tools.0.load(Ordering::SeqCst),
            0,
            "resume must not preload or activate on the model's behalf"
        );

        let plan_config = AgentLoopConfig {
            skill_manager: Some(manager.clone()),
            selected_skill_ids: Some(vec!["plan-demo".to_string()]),
            disabled_skill_ids: BTreeSet::new(),
            ..Default::default()
        };
        initialize_loop_state(
            &mut session,
            "plan instead",
            &plan_config,
            &tools,
            &event_tx,
        )
        .await
        .expect("superseding activation");
        let (skills, descriptor) = manager
            .store()
            .pinned_activation_skills("retained-session")
            .await
            .expect("superseding plan pin");
        assert_eq!(skills.len(), 1);
        assert_eq!(skills[0].id, "plan-demo");
        assert_eq!(
            tools.0.load(Ordering::SeqCst),
            0,
            "a superseding explicit selection must also wait for the model-issued load_skill call"
        );
        assert_eq!(
            descriptor
                .skill_revisions
                .keys()
                .cloned()
                .collect::<Vec<_>>(),
            vec!["plan-demo"]
        );
        assert!(
            crate::runtime::runner::session_setup::skill_context::explicit_activation_pending(
                &session
            )
        );
        assert!(crate::runtime::runner::session_setup::prompt_envelope::build_active_workflow_context_block(&session).is_none());
    }

    #[test]
    fn rewrapped_copilot_queue_keeps_same_provider_without_network() {
        use bamboo_domain::{ProviderModelRef, ReasoningEffort};
        use bamboo_llm::{LLMProvider, ResolvedModel};
        use std::sync::Arc;

        let provider: Arc<dyn LLMProvider> = Arc::new(
            bamboo_llm::providers::copilot::CopilotProvider::with_token("synthetic-only"),
        );
        let reference =
            ProviderModelRef::new("copilot", "model-a").with_reasoning_effort(ReasoningEffort::Low);
        let first = ResolvedModel::from_ref(provider.clone(), &reference).provider;
        let second = ResolvedModel::from_ref(provider, &reference).provider;
        assert!(
            !Arc::ptr_eq(&first, &second),
            "each resolution constructs a new role wrapper"
        );
        let queued = super::QueuedEvaluation::new((), first, "model-a".into())
            .expect("normal Copilot can be queued");
        assert!(queued.current_provider(second, "model-a").is_some());
    }
}