bamboo-engine 2026.10.22

//! Unified agent execution runtime.
//!
//! [`AgentRuntime`] holds shared resources assembled once at server startup,
//! including the LLM provider and default tool executor.
//! [`ExecuteRequest`] captures per-request parameters.  Together they replace
//! the three duplicated `AgentLoopConfig` construction sites in the server
//! layer (HTTP handler, spawn scheduler, schedule manager).

use std::collections::{BTreeSet, HashMap};
use std::sync::Arc;

use tokio::sync::{mpsc, RwLock};
use tokio_util::sync::CancellationToken;

use bamboo_agent_core::storage::{AttachmentReader, Storage};
use bamboo_agent_core::tools::ToolExecutor;
use bamboo_agent_core::{AgentEvent, Role, Session};
use bamboo_config::PermissionMode;
use bamboo_domain::ReasoningEffort;
use bamboo_llm::Config;
use bamboo_llm::{LLMProvider, ProviderRegistry};
use bamboo_metrics::MetricsCollector;
use bamboo_skills::SkillManager;

use crate::runtime::config::{
    AgentLoopConfig, AuxiliaryModelConfig, BashCompletionSink, BashResumeHook,
    DisabledFilterResolver, GoldConfig, GuardianConfig, GuardianSpawner, ImageFallbackConfig,
    PromptMemoryFlags,
};
use crate::runtime::hooks::HookRunner;
use crate::runtime::model_roster::{ModelRoster, RoleModel};
use crate::runtime::runner::run_agent_loop_with_config;
use bamboo_domain::{RuntimeSessionPersistence, SessionInboxPort};

use crate::session_activation::SessionActivationRouter;
use crate::session_messaging::SessionMessenger;

pub(crate) type AuxiliaryModelResolver = Arc<dyn Fn() -> AuxiliaryModelConfig + Send + Sync>;
pub(crate) type AuxiliaryModelResolverFactory =
    Arc<dyn Fn(&str) -> AuxiliaryModelResolver + Send + Sync>;

// A host factory is installed only on the Schedule Agent clone. Explicitly
// absent roster keys are passed as empty, never substituted with a global key.
fn select_auxiliary_model_resolver(
    factory: Option<&AuxiliaryModelResolverFactory>,
    roster_provider: Option<&str>,
    fallback: Option<AuxiliaryModelResolver>,
) -> Option<AuxiliaryModelResolver> {
    factory
        .map(|create| create(roster_provider.unwrap_or_default()))
        .or(fallback)
}

#[cfg(test)]
mod auxiliary_resolver_factory_tests {
    use super::*;
    use std::sync::Mutex;

    #[test]
    fn factory_receives_each_roster_key_and_sdk_fallback_is_unchanged() {
        let keys = Arc::new(Mutex::new(Vec::new()));
        let recorded = keys.clone();
        let factory: AuxiliaryModelResolverFactory = Arc::new(move |key| {
            recorded.lock().unwrap().push(key.to_string());
            let name = key.to_string();
            Arc::new(move || AuxiliaryModelConfig {
                fast_model_name: Some(name.clone()),
                ..Default::default()
            })
        });
        let fallback: AuxiliaryModelResolver = Arc::new(|| AuxiliaryModelConfig {
            fast_model_name: Some("snapshot".into()),
            ..Default::default()
        });
        let a = select_auxiliary_model_resolver(
            Some(&factory),
            Some("account-a"),
            Some(fallback.clone()),
        )
        .unwrap();
        let b = select_auxiliary_model_resolver(
            Some(&factory),
            Some("account-b"),
            Some(fallback.clone()),
        )
        .unwrap();
        let missing = select_auxiliary_model_resolver(Some(&factory), None, None).unwrap();
        assert_eq!(a().fast_model_name.as_deref(), Some("account-a"));
        assert_eq!(b().fast_model_name.as_deref(), Some("account-b"));
        assert_eq!(missing().fast_model_name.as_deref(), Some(""));
        assert_eq!(*keys.lock().unwrap(), vec!["account-a", "account-b", ""]);
        let unchanged =
            select_auxiliary_model_resolver(None, Some("account-a"), Some(fallback.clone()))
                .unwrap();
        assert!(Arc::ptr_eq(&unchanged, &fallback));
        assert!(select_auxiliary_model_resolver(None, None, None).is_none());
    }
}

// Preserve the exact managed child-resume route through the optional Work
// decorator. `main` is the undecorated override and `decorated_main` was built
// from it by execute_bound. Do not trust a later resolver caller to supply the
// matching delegate, and do not silently downgrade an invariant violation.
pub(crate) fn bind_child_gold_delegate(
    config: &mut AgentLoopConfig,
    selected: Option<&Arc<dyn LLMProvider>>,
    main: &Arc<dyn LLMProvider>,
    decorated_main: &Arc<dyn LLMProvider>,
) -> crate::runtime::runner::Result<()> {
    let Some(selected) = selected else {
        return Ok(());
    };
    if !Arc::ptr_eq(selected, main) {
        return Err(bamboo_agent_core::AgentError::LLM(
            "Managed child-resume provider differs from its selected route".into(),
        ));
    }
    let resolve = config.gold_target_resolver.take().ok_or_else(|| {
        bamboo_agent_core::AgentError::LLM(
            "Managed child-resume Gold route is missing its guard".into(),
        )
    })?;
    let delegate = decorated_main.clone();
    config.gold_target_resolver = Some(Arc::new(move |model, _untrusted_delegate| {
        resolve(model, delegate.clone())
    }));
    Ok(())
}

// ---------------------------------------------------------------------------
// AgentRuntime — shared resources (assembled once)
// ---------------------------------------------------------------------------

/// Shared runtime resources assembled once at server startup.
///
/// Each field is an `Arc` or `Clone`-cheap handle so that `AgentRuntime` itself
/// is cheaply cloneable across tasks.
#[derive(Clone)]
pub struct AgentRuntime {
    pub storage: Arc<dyn Storage>,
    pub persistence: Arc<dyn RuntimeSessionPersistence>,
    /// Capturing no inherited wait is also immutable for this execution.
    pub(crate) inherited_child_wait_captured: bool,
    /// Config-derived fallback is an explicit per-run Host choice.
    pub(crate) legacy_image_from_config: bool,
    pub(crate) legacy_image_provider_source:
        Option<Arc<bamboo_llm::provider_registry::LegacyVisionProviderSource>>,
    pub(crate) sdk_skill_execution_host:
        Option<Arc<dyn crate::runtime::config::SdkSkillExecutionHost>>,
    pub(crate) workspace_execution_host:
        Option<Arc<dyn crate::runtime::config::WorkspaceExecutionHost>>,
    pub session_inbox: Option<Arc<dyn SessionInboxPort>>,
    pub(crate) live_bootstrap_host: Option<Arc<dyn crate::runtime::config::LiveBootstrapHost>>,
    pub(crate) gold_target_resolver: Option<crate::runtime::config::GoldTargetResolver>,
    pub(crate) host_gold_origin_resolver: Option<crate::runtime::host_gold::HostGoldOriginResolver>,
    // Private to managed child resume; never inferred from a registry name.
    pub(crate) child_resume_gold_selected: Option<Arc<dyn LLMProvider>>,
    pub(crate) background_model_resolver: Option<crate::runtime::config::BackgroundModelResolver>,
    pub(crate) auxiliary_model_resolver_factory: Option<AuxiliaryModelResolverFactory>,
    pub(crate) resume_auxiliary_resolver: Option<crate::runtime::config::ResumeAuxiliaryResolver>,
    pub activation_router: Option<Arc<SessionActivationRouter>>,
    pub session_messenger: Option<Arc<SessionMessenger>>,
    pub attachment_reader: Arc<dyn AttachmentReader>,
    pub skill_manager: Arc<SkillManager>,
    pub project_context_resolver: Option<Arc<crate::project_context::ProjectContextResolver>>,
    pub metrics_collector: MetricsCollector,
    pub config: Arc<RwLock<Config>>,
    pub(crate) model_limits_section:
        Option<Arc<bamboo_config::LiveSection<bamboo_config::ModelLimitsSection>>>,
    /// Live process-wide permission policy when the embedding exposes one.
    pub permission_config: Option<Arc<bamboo_tools::permission::PermissionConfig>>,
    /// SDK/static fallback used when there is no live policy handle.
    pub permission_mode: PermissionMode,

    /// Reloadable LLM provider handle (delegates to the latest provider).
    pub provider: Arc<dyn LLMProvider>,

    /// Concrete Jiandu store shared by every run from this runtime.
    pub memory_store: bamboo_memory::memory_store::MemoryStore,

    /// Default tool executor (root tools with full surface).
    /// Call sites that need a reduced tool set (child / schedule) pass their
    /// own via `ExecuteRequest::tools`.
    pub default_tools: Arc<dyn ToolExecutor>,

    /// Immutable hook registry shared by runs. Each execute call snapshots the
    /// `Arc` onto its sealed [`AgentLoopConfig`].
    pub hook_runner: Arc<HookRunner>,
}

// ---------------------------------------------------------------------------
// AgentRuntimeBuilder
// ---------------------------------------------------------------------------

/// Builder for [`AgentRuntime`].
///
/// ```
/// # use bamboo_engine::AgentRuntimeBuilder;
/// // In real code all fields are provided by the server assembly.
/// let rt = AgentRuntimeBuilder::new()
///     // .storage(...)
///     // .provider(...)
///     .build();
/// ```
pub struct AgentRuntimeBuilder {
    storage: Option<Arc<dyn Storage>>,
    persistence: Option<Arc<dyn RuntimeSessionPersistence>>,
    session_inbox: Option<Arc<dyn SessionInboxPort>>,
    activation_router: Option<Arc<SessionActivationRouter>>,
    session_messenger: Option<Arc<SessionMessenger>>,
    attachment_reader: Option<Arc<dyn AttachmentReader>>,
    skill_manager: Option<Arc<SkillManager>>,
    project_context_resolver: Option<Arc<crate::project_context::ProjectContextResolver>>,
    metrics_collector: Option<MetricsCollector>,
    config: Option<Arc<RwLock<Config>>>,
    model_limits_section:
        Option<Arc<bamboo_config::LiveSection<bamboo_config::ModelLimitsSection>>>,
    permission_config: Option<Arc<bamboo_tools::permission::PermissionConfig>>,
    permission_mode: PermissionMode,
    provider: Option<Arc<dyn LLMProvider>>,
    memory_store: bamboo_memory::memory_store::MemoryStore,
    default_tools: Option<Arc<dyn ToolExecutor>>,
    hook_runner: Arc<HookRunner>,
}

impl AgentRuntimeBuilder {
    pub fn new() -> Self {
        Self {
            storage: None,
            persistence: None,
            session_inbox: None,
            activation_router: None,
            session_messenger: None,
            attachment_reader: None,
            skill_manager: None,
            project_context_resolver: None,
            metrics_collector: None,
            config: None,
            model_limits_section: None,
            permission_config: None,
            permission_mode: PermissionMode::Default,
            provider: None,
            memory_store: bamboo_memory::memory_store::MemoryStore::with_defaults(),
            default_tools: None,
            hook_runner: Arc::new(HookRunner::new()),
        }
    }

    pub fn storage(mut self, v: Arc<dyn Storage>) -> Self {
        self.storage = Some(v);
        self
    }

    pub fn persistence(mut self, v: Arc<dyn RuntimeSessionPersistence>) -> Self {
        self.persistence = Some(v);
        self
    }

    pub fn session_inbox(mut self, v: Arc<dyn SessionInboxPort>) -> Self {
        self.session_inbox = Some(v);
        self
    }

    pub fn activation_router(mut self, v: Arc<SessionActivationRouter>) -> Self {
        self.activation_router = Some(v);
        self
    }

    pub fn session_messenger(mut self, v: Arc<SessionMessenger>) -> Self {
        self.session_messenger = Some(v);
        self
    }

    pub fn attachment_reader(mut self, v: Arc<dyn AttachmentReader>) -> Self {
        self.attachment_reader = Some(v);
        self
    }

    pub fn skill_manager(mut self, v: Arc<SkillManager>) -> Self {
        self.skill_manager = Some(v);
        self
    }

    pub fn project_context_resolver(
        mut self,
        v: Arc<crate::project_context::ProjectContextResolver>,
    ) -> Self {
        self.project_context_resolver = Some(v);
        self
    }

    pub fn metrics_collector(mut self, v: MetricsCollector) -> Self {
        self.metrics_collector = Some(v);
        self
    }

    pub fn config(mut self, v: Arc<RwLock<Config>>) -> Self {
        self.config = Some(v);
        self
    }

    /// Use a host's validated model-limit snapshots for every agent round.
    /// The host owns reloads and section health; budget resolution performs no
    /// file I/O when this handle is installed.
    pub fn model_limits_section(
        mut self,
        section: Arc<bamboo_config::LiveSection<bamboo_config::ModelLimitsSection>>,
    ) -> Self {
        self.model_limits_section = Some(section);
        self
    }

    pub fn permission_config(mut self, v: Arc<bamboo_tools::permission::PermissionConfig>) -> Self {
        self.permission_config = Some(v);
        self
    }

    pub fn permission_mode(mut self, v: PermissionMode) -> Self {
        self.permission_mode = v;
        self
    }

    pub fn provider(mut self, v: Arc<dyn LLMProvider>) -> Self {
        self.provider = Some(v);
        self
    }

    pub fn memory_store(mut self, v: bamboo_memory::memory_store::MemoryStore) -> Self {
        self.memory_store = v;
        self
    }

    pub fn default_tools(mut self, v: Arc<dyn ToolExecutor>) -> Self {
        self.default_tools = Some(v);
        self
    }

    /// Install the lifecycle hook registry used by all runs from this runtime.
    pub fn hook_runner(mut self, v: Arc<HookRunner>) -> Self {
        self.hook_runner = v;
        self
    }

    pub fn build(self) -> Result<AgentRuntime, &'static str> {
        if let (Some(router), Some(inbox)) = (&self.activation_router, &self.session_inbox) {
            router.set_inbox(inbox.clone());
        }
        Ok(AgentRuntime {
            inherited_child_wait_captured: false,
            legacy_image_from_config: false,
            legacy_image_provider_source: None,
            sdk_skill_execution_host: None,
            workspace_execution_host: None,
            storage: self.storage.ok_or_else(|| format_missing("storage"))?,
            persistence: self
                .persistence
                .ok_or_else(|| format_missing("persistence"))?,
            session_inbox: self.session_inbox,
            live_bootstrap_host: None,
            gold_target_resolver: None,
            host_gold_origin_resolver: None,
            child_resume_gold_selected: None,
            background_model_resolver: None,
            auxiliary_model_resolver_factory: None,
            resume_auxiliary_resolver: None,
            activation_router: self.activation_router,
            session_messenger: self.session_messenger,
            attachment_reader: self
                .attachment_reader
                .ok_or_else(|| format_missing("attachment_reader"))?,
            skill_manager: self
                .skill_manager
                .ok_or_else(|| format_missing("skill_manager"))?,
            project_context_resolver: self.project_context_resolver,
            metrics_collector: self
                .metrics_collector
                .ok_or_else(|| format_missing("metrics_collector"))?,
            config: self.config.ok_or_else(|| format_missing("config"))?,
            model_limits_section: self.model_limits_section,
            permission_config: self.permission_config,
            permission_mode: self.permission_mode,
            provider: self.provider.ok_or_else(|| format_missing("provider"))?,
            memory_store: self.memory_store,
            default_tools: self
                .default_tools
                .ok_or_else(|| format_missing("default_tools"))?,
            hook_runner: self.hook_runner,
        })
    }
}

fn format_missing(field: &str) -> &'static str {
    // Static strings for the common fields keep the error type `&'static str`.
    // This is good enough for a builder that only runs at startup.
    match field {
        "storage" => "AgentRuntimeBuilder: missing storage",
        "persistence" => "AgentRuntimeBuilder: missing persistence",
        "attachment_reader" => "AgentRuntimeBuilder: missing attachment_reader",
        "skill_manager" => "AgentRuntimeBuilder: missing skill_manager",
        "metrics_collector" => "AgentRuntimeBuilder: missing metrics_collector",
        "config" => "AgentRuntimeBuilder: missing config",
        "provider" => "AgentRuntimeBuilder: missing provider",
        "default_tools" => "AgentRuntimeBuilder: missing default_tools",
        _ => "AgentRuntimeBuilder: missing required field",
    }
}

impl Default for AgentRuntimeBuilder {
    fn default() -> Self {
        Self::new()
    }
}

// ---------------------------------------------------------------------------
// ExecuteRequest — per-request parameters
// ---------------------------------------------------------------------------

/// Per-request parameters for agent execution.
///
/// Required fields (`initial_message`, `event_tx`, `cancel_token`) must always
/// be provided.  The provider is taken from [`AgentRuntime::provider`]; tools
/// default to [`AgentRuntime::default_tools`] when `None`.
pub struct ExecuteRequest {
    /// Native Host attribution, never decoded from HTTP or Worker arguments.
    pub ticket_token_budget: Option<Arc<crate::ticket_token_budget::NativeWorkTokenBudget>>,
    /// Host-installed, per-run private plan capability. Never decoded from HTTP.
    pub ticket_worker_plan: Option<Arc<dyn crate::ticket_worker_plan::WorkerLocalPlan>>,
    // -- Required ----------------------------------------------------------
    pub initial_message: String,
    pub event_tx: mpsc::Sender<AgentEvent>,
    pub cancel_token: CancellationToken,

    // -- Tool override -----------------------------------------------------
    /// Override runtime's `default_tools`.  When `None`, uses the runtime's
    /// default tool executor.
    pub tools: Option<Arc<dyn ToolExecutor>>,
    /// Override the LLM provider for this execution. When `None`, uses the
    /// runtime's shared provider handle.
    pub provider_override: Option<Arc<dyn LLMProvider>>,

    // -- Model selection (None roles → config defaults) -------------------
    /// Cohesive primary + auxiliary model/provider selection. Replaces the old
    /// `model` / `provider_name` / `provider_type` / `fast_model(+provider)` /
    /// `background_model(+provider)` / `summarization_model(+provider)` clump.
    /// Per-role `None` preserves the same `None → Config::get_*` fallbacks
    /// applied during [`AgentRuntime::execute`].
    pub model_roster: ModelRoster,
    pub reasoning_effort: Option<ReasoningEffort>,
    /// Optional per-round resolver for auxiliary model settings that should be
    /// re-read from live global config between rounds.
    pub auxiliary_model_resolver: Option<Arc<dyn Fn() -> AuxiliaryModelConfig + Send + Sync>>,
    /// Optional per-round live resolver for the disabled tool/skill sets (#136).
    /// When `None`, the snapshotted `disabled_tools`/`disabled_skill_ids` are used.
    pub disabled_filter_resolver: Option<DisabledFilterResolver>,
    /// When `None`, falls back to `Config::disabled_tool_references()`.
    pub disabled_tools: Option<BTreeSet<String>>,
    /// When `None`, falls back to `Config::disabled_skill_ids()`.
    pub disabled_skill_ids: Option<BTreeSet<String>>,
    pub selected_skill_ids: Option<Vec<String>>,
    pub selected_skill_mode: Option<String>,
    pub image_fallback: Option<ImageFallbackConfig>,
    pub gold_config: Option<GoldConfig>,
    /// Optional guardian adversarial-review gate configuration.
    pub guardian_config: Option<GuardianConfig>,
    /// Late-bound spawner for the guardian reviewer child (wired by the server;
    /// the runner cannot construct a child directly).
    pub guardian_spawner: Option<Arc<dyn GuardianSpawner>>,
    /// Late-bound hook that arranges a self-resume after a background-bash
    /// suspend (issue #84 Phase 2b). Wired by the server.
    pub bash_resume_hook: Option<Arc<dyn BashResumeHook>>,
    /// Late-bound sink that pushes a completed background-bash shell's result
    /// into this loop (issue #84 Phase 2b follow-up). Wired by the server.
    pub bash_completion_sink: Option<Arc<dyn BashCompletionSink>>,
    /// Bamboo application data directory (typically `~/.bamboo`).
    pub app_data_dir: Option<std::path::PathBuf>,
    /// Per-run resource guardrail override (issue #221): token / tool-call /
    /// subagent budget for THIS execution. TIGHTEN-ONLY: per field, the
    /// effective limit is the minimum of this override and the config-level
    /// `Config::run_budget` default — a request can lower the operator's
    /// ceiling but never raise or remove it. An unset field (`None`) keeps
    /// the config default (which may itself be unlimited). See
    /// [`bamboo_config::RunBudgetConfig::merged_with_override`].
    pub run_budget: Option<bamboo_config::RunBudgetConfig>,
    /// Opt-in live counted repetition under the caller's existing execution owner.
    /// Requires the shared activation router and an already registered run.
    pub counted_loop: Option<super::counted_loop::CountedLoopConfig>,
    /// Explicit finite authority for the model to start one loop in this activation.
    pub loop_control: Option<super::loop_control::LoopControlConfig>,
}

// ---------------------------------------------------------------------------
// ExecuteRequestBuilder — ergonomic construction of ExecuteRequest
// ---------------------------------------------------------------------------

/// Fluent builder for [`ExecuteRequest`].
///
/// `ExecuteRequest` carries three required fields plus a long tail of optional
/// overrides; constructing it by hand forces callers to spell out ~20 `None`s.
/// This builder requires only `initial_message` + `event_tx` + `cancel_token`,
/// defaults every optional field to `None` (matching the runtime's spawn
/// defaults), and exposes fluent setters for the rest.
///
/// It lives in `bamboo-engine` so both the in-crate server layer (e.g. the
/// schedule manager) and the root `bamboo_agent` SDK facade (which re-exports
/// it) construct requests through one shared builder — no forked assembly.
pub struct ExecuteRequestBuilder {
    ticket_token_budget: Option<Arc<crate::ticket_token_budget::NativeWorkTokenBudget>>,
    ticket_worker_plan: Option<Arc<dyn crate::ticket_worker_plan::WorkerLocalPlan>>,
    initial_message: String,
    event_tx: mpsc::Sender<AgentEvent>,
    cancel_token: CancellationToken,

    tools: Option<Arc<dyn ToolExecutor>>,
    provider_override: Option<Arc<dyn LLMProvider>>,
    // Individual model fields are accumulated here for backward-compatible
    // fluent setters; `build()` assembles them into a `ModelRoster`. A
    // `.model_roster(..)` setter seeds all of them at once.
    model: Option<String>,
    provider_name: Option<String>,
    provider_type: Option<String>,
    fast_model: Option<String>,
    fast_model_provider: Option<Arc<dyn LLMProvider>>,
    background_model: Option<String>,
    background_model_provider: Option<Arc<dyn LLMProvider>>,
    summarization_model: Option<String>,
    summarization_model_provider: Option<Arc<dyn LLMProvider>>,
    reasoning_effort: Option<ReasoningEffort>,
    auxiliary_model_resolver: Option<Arc<dyn Fn() -> AuxiliaryModelConfig + Send + Sync>>,
    disabled_filter_resolver: Option<DisabledFilterResolver>,
    disabled_tools: Option<BTreeSet<String>>,
    disabled_skill_ids: Option<BTreeSet<String>>,
    selected_skill_ids: Option<Vec<String>>,
    selected_skill_mode: Option<String>,
    image_fallback: Option<ImageFallbackConfig>,
    gold_config: Option<GoldConfig>,
    guardian_config: Option<GuardianConfig>,
    guardian_spawner: Option<Arc<dyn GuardianSpawner>>,
    bash_resume_hook: Option<Arc<dyn BashResumeHook>>,
    bash_completion_sink: Option<Arc<dyn BashCompletionSink>>,
    app_data_dir: Option<std::path::PathBuf>,
    run_budget: Option<bamboo_config::RunBudgetConfig>,
    counted_loop: Option<super::counted_loop::CountedLoopConfig>,
    loop_control: Option<super::loop_control::LoopControlConfig>,
}

impl ExecuteRequestBuilder {
    pub(crate) fn ticket_token_budget(
        mut self,
        budget: Arc<crate::ticket_token_budget::NativeWorkTokenBudget>,
    ) -> Self {
        self.ticket_token_budget = Some(budget);
        self
    }

    /// Create a builder with the three required fields. All optional overrides
    /// default to `None`.
    pub fn new(
        initial_message: impl Into<String>,
        event_tx: mpsc::Sender<AgentEvent>,
        cancel_token: CancellationToken,
    ) -> Self {
        Self {
            initial_message: initial_message.into(),
            ticket_worker_plan: None,
            ticket_token_budget: None,
            event_tx,
            cancel_token,
            tools: None,
            provider_override: None,
            model: None,
            provider_name: None,
            provider_type: None,
            fast_model: None,
            fast_model_provider: None,
            background_model: None,
            background_model_provider: None,
            summarization_model: None,
            summarization_model_provider: None,
            reasoning_effort: None,
            auxiliary_model_resolver: None,
            disabled_filter_resolver: None,
            disabled_tools: None,
            disabled_skill_ids: None,
            selected_skill_ids: None,
            selected_skill_mode: None,
            image_fallback: None,
            gold_config: None,
            guardian_config: None,
            guardian_spawner: None,
            bash_resume_hook: None,
            bash_completion_sink: None,
            app_data_dir: None,
            run_budget: None,
            counted_loop: None,
            loop_control: None,
        }
    }

    /// Install a trusted, per-run private LocalPlan port.
    pub fn ticket_worker_plan(
        mut self,
        plan: Arc<dyn crate::ticket_worker_plan::WorkerLocalPlan>,
    ) -> Self {
        self.ticket_worker_plan = Some(plan);
        self
    }

    /// Override the tool executor for this execution.
    pub fn tools(mut self, v: Arc<dyn ToolExecutor>) -> Self {
        self.tools = Some(v);
        self
    }

    /// Override the LLM provider for this execution.
    pub fn provider_override(mut self, v: Arc<dyn LLMProvider>) -> Self {
        self.provider_override = Some(v);
        self
    }

    /// Seed the full model selection from a [`ModelRoster`].
    ///
    /// Decomposes the roster back into the builder's individual fields so it
    /// composes with the existing per-field fluent setters; later individual
    /// setters override the corresponding roster entry.
    pub fn model_roster(mut self, roster: ModelRoster) -> Self {
        self.fast_model = roster.fast_model();
        self.fast_model_provider = roster.fast_model_provider();
        self.background_model = roster.background_model();
        self.background_model_provider = roster.background_model_provider();
        self.summarization_model = roster.summarization_model();
        self.summarization_model_provider = roster.summarization_model_provider();
        self.model = roster.model;
        self.provider_name = roster.provider_name;
        self.provider_type = roster.provider_type;
        self
    }

    /// Override the primary model name.
    pub fn model(mut self, v: impl Into<String>) -> Self {
        self.model = Some(v.into());
        self
    }

    /// Override the provider name.
    pub fn provider_name(mut self, v: impl Into<String>) -> Self {
        self.provider_name = Some(v.into());
        self
    }

    /// Override the provider type.
    pub fn provider_type(mut self, v: impl Into<String>) -> Self {
        self.provider_type = Some(v.into());
        self
    }

    /// Override the fast-model name.
    pub fn fast_model(mut self, v: impl Into<String>) -> Self {
        self.fast_model = Some(v.into());
        self
    }

    /// Override the provider used for fast-model calls.
    pub fn fast_model_provider(mut self, v: Arc<dyn LLMProvider>) -> Self {
        self.fast_model_provider = Some(v);
        self
    }

    /// Override the background-model name.
    pub fn background_model(mut self, v: impl Into<String>) -> Self {
        self.background_model = Some(v.into());
        self
    }

    /// Override the provider used for background/memory model calls.
    pub fn background_model_provider(mut self, v: Arc<dyn LLMProvider>) -> Self {
        self.background_model_provider = Some(v);
        self
    }

    /// Override the summarization-model name.
    pub fn summarization_model(mut self, v: impl Into<String>) -> Self {
        self.summarization_model = Some(v.into());
        self
    }

    /// Override the provider used for summarization/compression calls.
    pub fn summarization_model_provider(mut self, v: Arc<dyn LLMProvider>) -> Self {
        self.summarization_model_provider = Some(v);
        self
    }

    /// Set the reasoning effort.
    pub fn reasoning_effort(mut self, v: ReasoningEffort) -> Self {
        self.reasoning_effort = Some(v);
        self
    }

    /// Set the per-round auxiliary-model resolver.
    pub fn auxiliary_model_resolver(
        mut self,
        v: Arc<dyn Fn() -> AuxiliaryModelConfig + Send + Sync>,
    ) -> Self {
        self.auxiliary_model_resolver = Some(v);
        self
    }

    /// Set the per-round resolver for the live disabled tool/skill sets (#136).
    pub fn disabled_filter_resolver(mut self, v: DisabledFilterResolver) -> Self {
        self.disabled_filter_resolver = Some(v);
        self
    }

    /// Set the disabled tool names (merged with config defaults at runtime).
    pub fn disabled_tools(mut self, v: BTreeSet<String>) -> Self {
        self.disabled_tools = Some(v);
        self
    }

    /// Set the disabled skill ids.
    pub fn disabled_skill_ids(mut self, v: BTreeSet<String>) -> Self {
        self.disabled_skill_ids = Some(v);
        self
    }

    /// Set the explicitly selected skill ids.
    pub fn selected_skill_ids(mut self, v: Vec<String>) -> Self {
        self.selected_skill_ids = Some(v);
        self
    }

    /// Set the skill-selection mode.
    pub fn selected_skill_mode(mut self, v: impl Into<String>) -> Self {
        self.selected_skill_mode = Some(v.into());
        self
    }

    /// Set the image fallback configuration.
    pub fn image_fallback(mut self, v: ImageFallbackConfig) -> Self {
        self.image_fallback = Some(v);
        self
    }

    /// Set the internal `gold_config` feature flag.
    ///
    /// `gold_config` is an internal feature flag (not part of the public SDK
    /// surface), so this setter is crate-visible only. Public SDK callers always
    /// leave it `None`; the in-crate spawn paths thread a resolved value through
    /// here. Defaults to `None`.
    pub(crate) fn gold_config(mut self, v: Option<GoldConfig>) -> Self {
        self.gold_config = v;
        self
    }

    /// Set the internal `guardian_config` feature flag (crate-visible, like
    /// [`Self::gold_config`]). Public SDK callers leave it `None`.
    pub(crate) fn guardian_config(mut self, v: Option<GuardianConfig>) -> Self {
        self.guardian_config = v;
        self
    }

    /// Set the late-bound guardian reviewer spawner (crate-visible; wired by the
    /// server's spawn path so the runner can create the reviewer child).
    pub(crate) fn guardian_spawner(mut self, v: Option<Arc<dyn GuardianSpawner>>) -> Self {
        self.guardian_spawner = v;
        self
    }

    /// Set the late-bound bash self-resume hook (crate-visible; wired by the
    /// server so a session suspended on background bash is always resumed).
    pub(crate) fn bash_resume_hook(mut self, v: Option<Arc<dyn BashResumeHook>>) -> Self {
        self.bash_resume_hook = v;
        self
    }

    /// Set the late-bound bash completion sink (crate-visible; wired by the
    /// server so a completed background shell's result is pushed into the loop).
    pub(crate) fn bash_completion_sink(mut self, v: Option<Arc<dyn BashCompletionSink>>) -> Self {
        self.bash_completion_sink = v;
        self
    }

    /// Set the Bamboo application data directory.
    pub fn app_data_dir(mut self, v: std::path::PathBuf) -> Self {
        self.app_data_dir = Some(v);
        self
    }

    /// Set the per-run resource guardrail override (issue #221). TIGHTEN-ONLY:
    /// per field, the effective limit is the minimum of `v` and the
    /// config-level default — this can never loosen the operator's ceiling;
    /// see [`bamboo_config::RunBudgetConfig::merged_with_override`].
    pub fn run_budget(mut self, v: bamboo_config::RunBudgetConfig) -> Self {
        self.run_budget = Some(v);
        self
    }

    /// Repeat the submitted prompt with explicit iteration and total-round bounds.
    /// This is live-only; lifecycle/model management is a separate application seam.
    pub fn counted_loop(mut self, config: super::counted_loop::CountedLoopConfig) -> Self {
        self.counted_loop = Some(config);
        self
    }

    /// Authorize live self start/status/update/stop within explicit host ceilings.
    /// This starts no loop and grants no Session ownership.
    pub fn loop_control(mut self, config: super::loop_control::LoopControlConfig) -> Self {
        self.loop_control = Some(config);
        self
    }

    /// Materialize the underlying [`ExecuteRequest`].
    ///
    /// `gold_config` is an internal feature flag with only a crate-visible
    /// setter ([`Self::gold_config`]); public SDK callers leave it `None`.
    pub fn build(self) -> ExecuteRequest {
        let model_roster = ModelRoster {
            model: self.model,
            provider_name: self.provider_name,
            provider_type: self.provider_type,
            fast: RoleModel::from_parts(self.fast_model, self.fast_model_provider),
            background: RoleModel::from_parts(
                self.background_model,
                self.background_model_provider,
            ),
            summarization: RoleModel::from_parts(
                self.summarization_model,
                self.summarization_model_provider,
            ),
        };
        ExecuteRequest {
            ticket_worker_plan: self.ticket_worker_plan,
            ticket_token_budget: self.ticket_token_budget,
            initial_message: self.initial_message,
            event_tx: self.event_tx,
            cancel_token: self.cancel_token,
            tools: self.tools,
            provider_override: self.provider_override,
            model_roster,
            reasoning_effort: self.reasoning_effort,
            auxiliary_model_resolver: self.auxiliary_model_resolver,
            disabled_filter_resolver: self.disabled_filter_resolver,
            disabled_tools: self.disabled_tools,
            disabled_skill_ids: self.disabled_skill_ids,
            selected_skill_ids: self.selected_skill_ids,
            selected_skill_mode: self.selected_skill_mode,
            image_fallback: self.image_fallback,
            gold_config: self.gold_config,
            guardian_config: self.guardian_config,
            guardian_spawner: self.guardian_spawner,
            bash_resume_hook: self.bash_resume_hook,
            bash_completion_sink: self.bash_completion_sink,
            app_data_dir: self.app_data_dir,
            run_budget: self.run_budget,
            counted_loop: self.counted_loop,
            loop_control: self.loop_control,
        }
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Extract the system prompt from session messages.
fn extract_system_prompt(session: &Session) -> Option<String> {
    session
        .messages
        .iter()
        .find(|m| matches!(m.role, Role::System))
        .map(|m| m.content.clone())
}

pub(super) fn should_observe_completed_tool_trace(
    result: &crate::runtime::runner::Result<()>,
    final_checkpoint_succeeded: bool,
    session: &Session,
) -> bool {
    final_checkpoint_succeeded
        && result.is_ok()
        && session
            .agent_runtime_state
            .as_ref()
            .is_some_and(|state| matches!(state.status, bamboo_domain::AgentStatusState::Completed))
        && !session.metadata.contains_key("runtime.suspend_reason")
        && !session.metadata.contains_key("runtime.completion_reason")
}

// ---------------------------------------------------------------------------
// Execution
// ---------------------------------------------------------------------------

impl AgentRuntime {
    /// Execute the agent loop with the given request.
    ///
    /// Builds an [`AgentLoopConfig`] from the request parameters and shared
    /// runtime resources, then delegates to [`run_agent_loop_with_config`].
    pub async fn execute(
        &self,
        session: &mut Session,
        req: ExecuteRequest,
    ) -> crate::runtime::runner::Result<()> {
        self.execute_with_inputs(session, req, None).await
    }

    /// Select an initial-request retry policy for this execution only.
    pub async fn execute_with_upstream_retry(
        &self,
        session: &mut Session,
        req: ExecuteRequest,
        upstream_retry: bamboo_llm::retry::UpstreamRetry,
    ) -> crate::runtime::runner::Result<()> {
        self.execute_with_inputs_and_upstream_retry(session, req, None, upstream_retry)
            .await
    }

    pub(crate) async fn execute_with_inputs(
        &self,
        session: &mut Session,
        req: ExecuteRequest,
        inputs: Option<crate::runtime::config::UntrustedExecutionInputs>,
    ) -> crate::runtime::runner::Result<()> {
        self.execute_with_inputs_and_upstream_retry(
            session,
            req,
            inputs,
            bamboo_llm::retry::UpstreamRetry::Bounded,
        )
        .await
    }

    pub(crate) async fn execute_with_inputs_and_upstream_retry(
        &self,
        session: &mut Session,
        req: ExecuteRequest,
        inputs: Option<crate::runtime::config::UntrustedExecutionInputs>,
        upstream_retry: bamboo_llm::retry::UpstreamRetry,
    ) -> crate::runtime::runner::Result<()> {
        struct WorkspaceFinish(Option<Arc<dyn crate::runtime::config::WorkspaceExecutionHost>>);
        impl Drop for WorkspaceFinish {
            fn drop(&mut self) {
                if let Some(host) = &self.0 {
                    host.finish();
                }
            }
        }
        // Even an admission refusal closes only this captured bridge. Do not
        // retain an apparently live Weak entry across router finalization.
        let _workspace_finish = WorkspaceFinish(self.workspace_execution_host.clone());
        session
            .check_execution_compatibility()
            .map_err(|error| bamboo_agent_core::AgentError::Tool(error.to_string()))?;
        // Refusals are not executions: never reach SessionEnd or the generic
        // final checkpoint, and never mutate another counted owner's clone.
        super::counted_loop::reject_ambiguous_reentry(session, self.persistence.as_ref()).await?;
        if let Some(control) = req.loop_control.as_ref() {
            control
                .validate_admission(session, self.activation_router.as_ref())
                .await?;
        }
        struct SdkFinish {
            host: Option<Arc<dyn crate::runtime::config::SdkSkillExecutionHost>>,
            session_id: String,
            execution_id: String,
        }
        impl Drop for SdkFinish {
            fn drop(&mut self) {
                if let Some(host) = self.host.as_ref() {
                    host.finish(&self.session_id, &self.execution_id);
                }
            }
        }
        let _sdk_finish = SdkFinish {
            host: self.sdk_skill_execution_host.clone(),
            session_id: session.id.clone(),
            execution_id: inputs
                .as_ref()
                .and_then(|inputs| inputs.sdk_execution_id())
                .unwrap_or("")
                .to_owned(),
        };
        let existing = self.persistence.inherited_child_wait();
        if let Some(inherited) = existing.as_ref() {
            inherited
                .validate_session(session)
                .map_err(|error| bamboo_agent_core::AgentError::LLM(error.to_string()))?;
        } else if !self.inherited_child_wait_captured {
            let mut runtime = self.clone();
            if let Some(inherited) = bamboo_domain::InheritedChildWait::capture(session) {
                runtime.persistence = self
                    .persistence
                    .bind_inherited_child_wait(inherited)
                    .map_err(|error| bamboo_agent_core::AgentError::LLM(error.to_string()))?;
            }
            runtime.inherited_child_wait_captured = true;
            return runtime
                .execute_bound(session, req, inputs, upstream_retry)
                .await;
        }
        self.execute_bound(session, req, inputs, upstream_retry)
            .await
    }

    async fn execute_bound(
        &self,
        session: &mut Session,
        req: ExecuteRequest,
        inputs: Option<crate::runtime::config::UntrustedExecutionInputs>,
        upstream_retry: bamboo_llm::retry::UpstreamRetry,
    ) -> crate::runtime::runner::Result<()> {
        if self.persistence.root_actor_execution_required(session) {
            return Err(bamboo_agent_core::AgentError::Tool(
                "Root execution route requires a bound Actor writer and event handoff".into(),
            ));
        }
        let session_activation_notifications = match self.activation_router.as_ref() {
            Some(router) => Some(Arc::new(parking_lot::Mutex::new(
                router.subscribe(&session.id).await,
            ))),
            None => None,
        };
        let guidance_active_run_id = match self.activation_router.as_ref() {
            Some(router) => router.current_run_id(&session.id).await,
            None => None,
        };
        if let Some(counted) = req.counted_loop.as_ref() {
            if !counted
                .owns_session(self.activation_router.as_ref(), session)
                .await
            {
                return Err(bamboo_agent_core::AgentError::Budget(
                    "counted loop requires the caller's exact Session activation owner".into(),
                ));
            }
        }
        let system_prompt = extract_system_prompt(session);
        let config = self.config.read().await;
        let ExecuteRequest {
            ticket_worker_plan,
            ticket_token_budget,
            initial_message,
            event_tx,
            cancel_token,
            tools,
            provider_override,
            model_roster,
            reasoning_effort,
            auxiliary_model_resolver,
            disabled_filter_resolver,
            disabled_tools,
            disabled_skill_ids,
            selected_skill_ids,
            selected_skill_mode,
            image_fallback,
            gold_config,
            guardian_config,
            guardian_spawner,
            bash_resume_hook,
            bash_completion_sink,
            app_data_dir,
            run_budget,
            counted_loop,
            loop_control,
        } = req;
        let tools = tools.unwrap_or_else(|| self.default_tools.clone());
        let uses_default_provider = provider_override.is_none();
        let llm = provider_override.unwrap_or_else(|| self.provider.clone());
        let configured_permission_mode = self
            .permission_config
            .as_ref()
            .map(|config| config.mode())
            .unwrap_or(self.permission_mode);
        let active_plan_gate = session
            .agent_runtime_state
            .as_ref()
            .is_some_and(|state| state.plan_mode.is_some());

        // Explicit SDK role overrides retain their supplied client/name seam.
        // Unset roles use the same strict live defaults as Host consumers.
        let explicit_roles = [
            model_roster.fast.is_some(),
            model_roster.background.is_some(),
            model_roster.summarization.is_some(),
        ];
        let fast_model = model_roster.fast_model();
        let fast_model_provider = model_roster.fast_model_provider();
        let background_model = model_roster.background_model();
        let background_model_provider = model_roster.background_model_provider();
        let summarization_model = model_roster.summarization_model();
        let summarization_model_provider = model_roster.summarization_model_provider();
        let ModelRoster {
            model,
            provider_name,
            provider_type,
            ..
        } = model_roster;
        let auxiliary_model_resolver = select_auxiliary_model_resolver(
            self.auxiliary_model_resolver_factory.as_ref(),
            provider_name.as_deref(),
            auxiliary_model_resolver,
        );
        let active_provider =
            provider_name.unwrap_or_else(|| config.effective_default_provider().to_string());
        let auxiliary_model_resolver = auxiliary_model_resolver.or_else(|| {
            // A direct SDK client authorizes this declared account only. A
            // different configured role needs an explicit Host resolver/client;
            // never borrow the main client for another account's model name.
            let registry = Arc::new(ProviderRegistry::new(
                HashMap::from([(active_provider.clone(), llm.clone())]),
                active_provider.clone(),
            ));
            let resolve = crate::model_areas::live_auxiliary_model_resolver(
                self.config.clone(),
                registry,
                active_provider.clone(),
            );
            let overrides = AuxiliaryModelConfig {
                fast_model_name: fast_model.clone().or_else(|| config.get_fast_model()),
                fast_model_provider: fast_model_provider.clone(),
                background_model_name: background_model
                    .clone()
                    .or_else(|| config.get_memory_background_model()),
                background_model_provider: background_model_provider.clone(),
                summarization_model_name: summarization_model
                    .clone()
                    .or_else(|| config.get_task_summary_model()),
                summarization_model_provider: summarization_model_provider.clone(),
                ..Default::default()
            };
            let resolver: Arc<dyn Fn() -> AuxiliaryModelConfig + Send + Sync> =
                Arc::new(move || {
                    let mut current = resolve();
                    if explicit_roles[0] {
                        current.fast_model_name = overrides.fast_model_name.clone();
                        current.fast_model_provider = overrides.fast_model_provider.clone();
                    }
                    if explicit_roles[1] {
                        current.background_model_name = overrides.background_model_name.clone();
                        current.background_model_provider =
                            overrides.background_model_provider.clone();
                    }
                    if explicit_roles[2] {
                        current.summarization_model_name =
                            overrides.summarization_model_name.clone();
                        current.summarization_model_provider =
                            overrides.summarization_model_provider.clone();
                        current.summarization_model_error = None;
                    }
                    current
                });
            Some(resolver)
        });
        let hook_runner = inputs
            .as_ref()
            .and_then(|inputs| inputs.sdk_hook_runner())
            .unwrap_or_else(|| {
                Arc::new(
                    self.hook_runner
                        .with_lifecycle_config(&config.lifecycle_hooks, app_data_dir.clone()),
                )
            });

        let mut effective_run_budget = config.run_budget.merged_with_override(run_budget.as_ref());
        let legacy_vision_model_resolver = image_fallback
            .as_ref()
            .filter(|fallback| fallback.mode == crate::ImageFallbackMode::Vision)
            .map(|fallback| {
                crate::runtime::runner::image_fallback::legacy_vision_resolver(
                    config.clone(),
                    active_provider.clone(),
                    llm.clone(),
                    fallback.vision_model.clone(),
                    app_data_dir.clone(),
                    self.legacy_image_from_config,
                    self.legacy_image_provider_source.clone(),
                )
            });
        if let Some(control) = loop_control.as_ref() {
            effective_run_budget.max_rounds = Some(
                effective_run_budget
                    .max_rounds
                    .map_or(control.max_total_rounds, |cap| {
                        cap.min(control.max_total_rounds)
                    }),
            );
        }
        let mut loop_config = AgentLoopConfig {
            ticket_worker_plan: ticket_worker_plan.clone(),
            guidance_active_run_id,
            system_prompt,
            // Snapshot the legacy model_limits from the live in-memory config so
            // resolve_token_budget never falls back to a disk-reading Config::new(). #38.
            legacy_model_limits: config.extra.get("model_limits").cloned(),
            model_limits_section: self.model_limits_section.clone(),
            disabled_skill_ids: disabled_skill_ids.unwrap_or_else(|| config.disabled_skill_ids()),
            selected_skill_ids,
            selected_skill_mode,
            initial_untrusted_inputs: inputs,
            sdk_skill_execution_host: self.sdk_skill_execution_host.clone(),
            workspace_execution_host: self.workspace_execution_host.clone(),
            skill_manager: Some(self.skill_manager.clone()),
            project_context_resolver: self.project_context_resolver.clone(),
            skip_initial_user_message: true,
            storage: Some(self.storage.clone()),
            persistence: Some(self.persistence.clone()),
            session_inbox: self.session_inbox.clone(),
            live_bootstrap_host: self.live_bootstrap_host.clone(),
            gold_target_resolver: self.gold_target_resolver.clone(),
            session_activation_notifications,
            attachment_reader: Some(self.attachment_reader.clone()),
            metrics_collector: Some(self.metrics_collector.clone()),
            model_name: model,
            fast_model_name: fast_model.or_else(|| config.get_fast_model()),
            fast_model_provider,
            auxiliary_evaluation_max_concurrency:
                crate::runtime::config::normalize_auxiliary_evaluation_max_concurrency(
                    config
                        .extra
                        .get("auxiliary_evaluation_max_concurrency")
                        .and_then(serde_json::Value::as_u64),
                ),
            background_model_name: background_model
                .or_else(|| config.get_memory_background_model()),
            planning_model_name: config
                .defaults
                .as_ref()
                .and_then(|d| d.planning.as_ref())
                .map(|r| r.model.clone()),
            search_model_name: config
                .defaults
                .as_ref()
                .and_then(|d| d.search.as_ref().or(d.fast.as_ref()))
                .map(|r| r.model.clone()),
            compression_instructions: None,
            summary_target_ratio: config
                .memory()
                .as_ref()
                .map(|memory| memory.summary_target_ratio)
                .unwrap_or(0.20),
            context_management: config.context_management.clone(),
            summary_safe_window_percent: config
                .memory()
                .as_ref()
                .map(|memory| memory.summary_safe_window_percent)
                .unwrap_or(80),
            summarization_model_name: summarization_model
                .or_else(|| config.get_task_summary_model()),
            background_model_provider,
            summarization_model_provider,
            provider_name: Some(active_provider.clone()),
            provider_type,
            reasoning_effort,
            auxiliary_model_resolver,
            background_model_resolver: self.background_model_resolver.clone(),
            resume_auxiliary_resolver: self.resume_auxiliary_resolver.clone(),
            disabled_filter_resolver,
            disabled_tools: {
                let mut merged = config.disabled_tool_references();
                if let Some(dt) = disabled_tools {
                    merged.extend(dt);
                }
                merged
            },
            image_fallback,
            legacy_vision_model_resolver,
            vision_model_resolver: Some(
                crate::runtime::runner::image_fallback::configured_vision_resolver(
                    self.config.clone(),
                    app_data_dir.clone(),
                    active_provider,
                ),
            ),
            app_data_dir,
            memory_store: self.memory_store.clone(),
            prompt_memory_flags: config
                .memory()
                .as_ref()
                .map(PromptMemoryFlags::from)
                .unwrap_or_default(),
            features_dynamic_model_routing: config.features.dynamic_model_routing,
            features_final_evidence_check: config.features.final_evidence_check,
            permission_mode: Some(if active_plan_gate {
                PermissionMode::Plan
            } else {
                configured_permission_mode
            }),
            gold_config,
            guardian_config,
            guardian_spawner,
            bash_resume_hook,
            bash_completion_sink,
            hook_runner,
            // Capture the tool executor's server-level guidance (connected MCP
            // servers' `instructions`) once, so it lands in the system prompt only
            // while those servers are loaded for this run.
            mcp_tool_guidance: tools.tool_guidance(),
            // Config-level default, tighten-only-merged with the request's
            // override (issue #221): per field the minimum wins, so no caller
            // can loosen the operator's ceiling. Merging here (rather than in
            // the HTTP layer) means every caller — HTTP, schedules, connect,
            // the in-proc SDK — gets the same clamped fallback for free.
            run_budget: effective_run_budget,
            counted_loop,
            loop_control,
            stream_timeout: config.stream_timeout,
            upstream_retry,
            ..Default::default()
        };

        drop(config);
        let undecorated_main = llm.clone();
        let llm = match ticket_token_budget.as_ref() {
            Some(budget) => budget.decorate(&mut loop_config, llm),
            None => llm,
        };
        // A Host default provider can change between main and Gold. Capture
        // each accepted main dispatch rather than asking the mutable handle
        // again when a dedicated Gold request starts.
        if uses_default_provider
            && loop_config
                .gold_config
                .as_ref()
                .is_some_and(|gold| gold.enabled && gold.model_name.is_some())
        {
            if let Some(origin_resolver) = self.host_gold_origin_resolver.clone() {
                let tracker = crate::runtime::host_gold::HostGoldRun::new(origin_resolver);
                let source_tracker = tracker.clone();
                let live_config = self.config.clone();
                let original_main = undecorated_main.clone();
                let budget = ticket_token_budget.clone();
                let limits = Arc::new(AgentLoopConfig {
                    app_data_dir: loop_config.app_data_dir.clone(),
                    model_limits_section: loop_config.model_limits_section.clone(),
                    legacy_model_limits: loop_config.legacy_model_limits.clone(),
                    ..Default::default()
                });
                let inherited = loop_config.gold_target_resolver.take();
                loop_config.gold_target_resolver = Some(Arc::new(move |model, _mutable_main| {
                    let receipt = match source_tracker.accepted() {
                        Ok(receipt) => receipt,
                        Err(reason) => {
                            return Arc::new(crate::runtime::host_gold::UnprovenGold {
                                reason,
                                refused: std::sync::atomic::AtomicBool::new(false),
                            }) as Arc<dyn LLMProvider>
                        }
                    };
                    let fixed = match &receipt.origin {
                        crate::runtime::host_gold::HostGoldOrigin::Injected => {
                            bamboo_llm::provider::fixed_injected_provider_with_identity(
                                receipt.selected.clone(),
                                original_main.clone(),
                            )
                        }
                        crate::runtime::host_gold::HostGoldOrigin::Managed { .. } => {
                            bamboo_llm::provider::fixed_provider_with_identity(
                                receipt.selected.clone(),
                                original_main.clone(),
                            )
                        }
                    };
                    let delegate = match budget.as_ref() {
                        Some(budget) => budget.wrap(fixed, &limits),
                        None => fixed,
                    };
                    match receipt.origin {
                        crate::runtime::host_gold::HostGoldOrigin::Injected => inherited
                            .as_ref()
                            .map(|policy| policy(model, delegate.clone()))
                            .unwrap_or(delegate),
                        crate::runtime::host_gold::HostGoldOrigin::Managed {
                            provider_name,
                            origin,
                        } => {
                            let selected = receipt.selected;
                            let config = live_config.clone();
                            let managed: crate::runtime::config::GoldTargetResolver =
                                Arc::new(move |model, delegate| {
                                    bamboo_llm::gold_admission::guarded_gold_target_with_origin(
                                        config.clone(),
                                        origin.clone(),
                                        bamboo_domain::ProviderModelRef::new(
                                            provider_name.as_str(),
                                            model,
                                        ),
                                        selected.clone(),
                                        delegate,
                                    )
                                });
                            crate::runtime::agent::compose_child_gold_policy(
                                inherited.clone(),
                                managed,
                            )(model, delegate)
                        }
                    }
                }));
                loop_config.host_gold_run = Some(tracker);
            }
        }
        bind_child_gold_delegate(
            &mut loop_config,
            self.child_resume_gold_selected.as_ref(),
            &undecorated_main,
            &llm,
        )?;
        #[cfg(test)]
        crate::runtime::tests::observe_untrusted_inputs(
            &session.id,
            loop_config.initial_untrusted_inputs(),
        );

        if let Some(plan) = &ticket_worker_plan {
            session
                .agent_runtime_state
                .get_or_insert_with(Default::default)
                .run_id = plan.run_id().to_owned();
            plan.bind_session(session)
                .map_err(|e| bamboo_agent_core::AgentError::LLM(e.to_string()))?;
        }

        let trace_message_start = session.messages.len();
        let session_end_runner = loop_config.hook_runner.clone();
        let session_end_event_tx = event_tx.clone();
        let sdk_mode = loop_config.sdk_skill_execution_host.is_some();
        let loop_future = Box::pin(run_agent_loop_with_config(
            session,
            initial_message,
            event_tx,
            llm,
            tools,
            cancel_token,
            loop_config,
        ));
        let result = if sdk_mode {
            crate::runtime::runner::session_setup::legacy_instruction::scope_sdk_runtime(
                loop_future,
            )
            .await
        } else {
            loop_future.await
        };

        crate::runtime::hooks::run_session_end_hooks(
            &session_end_runner,
            &result,
            session,
            &session_end_event_tx,
        )
        .await;

        // Capture only the typed terminal watchdog error from this execution.
        // The existing checkpoint and exact run fingerprint own durable state;
        // the existing bounded live projection attaches the same receipt to the
        // error event without changing terminal ordering or retry decisions.
        if let Err(bamboo_agent_core::AgentError::StreamTimeout(timeout)) = &result {
            if let Some(diagnostic) = timeout.diagnostic(session) {
                if session.record_llm_request_diagnostic(diagnostic.clone()) {
                    bamboo_llm::retry::record_stream_timeout_diagnostic(diagnostic);
                }
            }
        }

        // The runtime is the one shared execute boundary for HTTP, child-agent
        // and direct SDK callers.  Checkpoint here on every outcome so durable
        // transcript correctness never depends on a caller-specific finalize
        // wrapper (and normal completions without a TaskLoopContext are saved
        // too).  The checkpoint is append-safe: a stale live snapshot cannot
        // shrink/rewrite messages appended concurrently on disk.
        //
        // Persistence failure must not replace the execution outcome.  In
        // particular, callers need the original LLM/cancellation error for
        // retry and terminal-status mapping; the failed durability attempt is
        // recorded separately.
        let final_checkpoint_succeeded =
            match self.persistence.checkpoint_runtime_session(session).await {
                Ok(()) => true,
                Err(checkpoint_error) => {
                    match &result {
                        Ok(()) => tracing::warn!(
                            session_id = %session.id,
                            error = %checkpoint_error,
                            "failed to checkpoint session transcript after successful execution"
                        ),
                        Err(execution_error) => tracing::warn!(
                            session_id = %session.id,
                            error = %checkpoint_error,
                            execution_error = %execution_error,
                            "failed to checkpoint session transcript after execution error"
                        ),
                    }
                    false
                }
            };

        if should_observe_completed_tool_trace(&result, final_checkpoint_succeeded, session) {
            if let Err(error) = self
                .skill_manager
                .observe_completed_tool_trace(session, trace_message_start)
                .await
            {
                tracing::warn!(
                    session_id = %session.id,
                    %error,
                    "failed to observe completed tool trace for reuse draft discovery"
                );
            }
        }

        result
    }
}