bamboo_engine/runtime/

config.rs

use std::collections::BTreeSet;
use std::path::PathBuf;
use std::sync::Arc;

use bamboo_agent_core::storage::AttachmentReader;
use bamboo_agent_core::storage::Storage;
use bamboo_agent_core::tools::ToolSchema;
use bamboo_agent_core::GoldConfidence;
use bamboo_compression::TokenBudget;
use bamboo_config::MemoryConfig;
use bamboo_config::PermissionMode;
use bamboo_domain::ReasoningEffort;
use bamboo_domain::RuntimeSessionPersistence;
use bamboo_llm::LLMProvider;
use bamboo_metrics::MetricsCollector;
use bamboo_skills::SkillManager;
use bamboo_tools::ToolRegistry;
use serde::{Deserialize, Serialize};

use super::hooks::HookRunner;

/// Default process-wide concurrency ceiling for low-priority auxiliary
/// evaluations sharing one provider/model allocation.
pub(crate) const DEFAULT_AUXILIARY_EVALUATION_MAX_CONCURRENCY: usize = 2;
/// Defensive ceiling for operator-provided auxiliary concurrency. This stays
/// far below Tokio semaphore's implementation maximum and prevents a malformed
/// flattened config value from panicking process startup.
pub(crate) const MAX_AUXILIARY_EVALUATION_MAX_CONCURRENCY: usize = 1024;

pub(crate) fn normalize_auxiliary_evaluation_max_concurrency(value: Option<u64>) -> usize {
    value
        .filter(|value| *value > 0)
        .map(|value| value.min(MAX_AUXILIARY_EVALUATION_MAX_CONCURRENCY as u64) as usize)
        .unwrap_or(DEFAULT_AUXILIARY_EVALUATION_MAX_CONCURRENCY)
}

/// The live disabled tool and skill-id sets resolved at a round boundary.
pub type DisabledFilterSets = (BTreeSet<String>, BTreeSet<String>);

/// Late-bound resolver for disabled tool and skill-id filters.
pub type DisabledFilterResolver = Arc<dyn Fn() -> DisabledFilterSets + Send + Sync>;

#[derive(Clone, Default)]
pub struct AuxiliaryModelConfig {
    pub fast_model_name: Option<String>,
    pub fast_model_provider: Option<Arc<dyn LLMProvider>>,
    pub background_model_name: Option<String>,
    pub planning_model_name: Option<String>,
    pub search_model_name: Option<String>,
    pub summarization_model_name: Option<String>,
    pub background_model_provider: Option<Arc<dyn LLMProvider>>,
    pub summarization_model_provider: Option<Arc<dyn LLMProvider>>,
}

fn default_gold_max_output_tokens() -> u32 {
    1024
}

fn default_gold_max_auto_continuations() -> u32 {
    3
}

fn default_gold_min_confidence() -> GoldConfidence {
    GoldConfidence::Medium
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct GoldConfig {
    #[serde(default)]
    pub recovery: super::goal_recovery::GoalRecoveryPolicy,
    /// Master switch for Gold observe-only evaluation.
    #[serde(default)]
    pub enabled: bool,
    /// Independent switch for Phase 2 low-risk auto-answer.
    ///
    /// Kept separate from `enabled` so Phase 1 observe-only users do not
    /// implicitly opt into automatic clarification responses.
    #[serde(default)]
    pub auto_answer_enabled: bool,
    /// Independent switch for Phase 3 server-side auto-continue.
    ///
    /// Kept separate from both `enabled` and `auto_answer_enabled` so users can
    /// opt into terminal auto-resume explicitly without enabling other Gold
    /// automation behaviors.
    #[serde(default)]
    pub auto_continue_enabled: bool,
    /// Optional dedicated model for Gold evaluation. Falls back to fast model,
    /// then the main chat model when absent.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub model_name: Option<String>,
    /// The user's goal for this session.
    ///
    /// Unlike `evaluation_prompt` (which only tunes the *judge*), the goal is
    /// surfaced to the *main* executing agent as a persistent system-prompt
    /// block so it actively works toward it. The Gold evaluator also measures
    /// progress against this text.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub goal: Option<String>,
    /// Optional custom prompt suffix appended to the built-in Gold evaluator
    /// prompt. This tunes the judge only; it does not set the goal.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub evaluation_prompt: Option<String>,
    /// Output token limit for the Gold evaluator call.
    #[serde(default = "default_gold_max_output_tokens")]
    pub max_output_tokens: u32,
    /// Maximum number of automatic Gold continuations allowed per session.
    #[serde(default = "default_gold_max_auto_continuations")]
    pub max_auto_continuations: u32,
    /// Minimum evaluator confidence required before Gold auto-continues or
    /// auto-answers. Defaults to `medium` so the loop fires on reasonably
    /// confident verdicts rather than only `high`.
    #[serde(default = "default_gold_min_confidence")]
    pub min_auto_continue_confidence: GoldConfidence,
}

impl Default for GoldConfig {
    fn default() -> Self {
        Self {
            recovery: Default::default(),
            enabled: false,
            auto_answer_enabled: false,
            auto_continue_enabled: false,
            model_name: None,
            goal: None,
            evaluation_prompt: None,
            max_output_tokens: default_gold_max_output_tokens(),
            max_auto_continuations: default_gold_max_auto_continuations(),
            min_auto_continue_confidence: default_gold_min_confidence(),
        }
    }
}

impl GoldConfig {
    /// The session goal text, falling back to the legacy `evaluation_prompt`
    /// for sessions created before the dedicated `goal` field existed.
    ///
    /// Returns `None` when neither field holds non-empty text.
    pub fn effective_goal(&self) -> Option<&str> {
        self.goal
            .as_deref()
            .or(self.evaluation_prompt.as_deref())
            .map(str::trim)
            .filter(|value| !value.is_empty())
    }
}

fn default_guardian_max_reviews() -> u32 {
    2
}

/// Configuration for the guardian adversarial-review terminal gate.
///
/// Mirrors [`GoldConfig`]: a plain, serde-defaulting struct surfaced per run.
/// When `enabled` is false (the default) the guardian gate is inactive and the
/// terminal completion path is unchanged.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct GuardianConfig {
    /// Master switch for the guardian review gate.
    #[serde(default)]
    pub enabled: bool,
    /// Optional dedicated reviewer model. Falls back to the run's main model.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub model_name: Option<String>,
    /// Maximum guardian review passes per run (budget; mirrors
    /// [`GoldConfig::max_auto_continuations`]).
    #[serde(default = "default_guardian_max_reviews")]
    pub max_reviews: u32,
}

impl Default for GuardianConfig {
    fn default() -> Self {
        Self {
            enabled: false,
            model_name: None,
            max_reviews: default_guardian_max_reviews(),
        }
    }
}

/// Late-bound spawner for the guardian reviewer child.
///
/// The runner cannot construct a child directly: the `SpawnScheduler` is built
/// *after* the `Agent` that drives the runner (a construction-order cycle), so
/// the terminal gate spawns the reviewer through this trait object, injected
/// per-request on [`AgentLoopConfig`] exactly like `auxiliary_model_resolver`.
/// The implementation lives in the server (it captures the already-built
/// scheduler + child-session adapter); the engine holds only the trait, keeping
/// the engine free of any dependency on server/AppState types.
#[async_trait::async_trait]
pub trait GuardianSpawner: Send + Sync {
    /// Create a read-only reviewer child for `parent_session_id`, seeded with
    /// `review_prompt`, enqueue it to run, and return its session id so the
    /// caller can register a wait on it.
    async fn spawn_guardian_review(
        &self,
        parent_session: &bamboo_agent_core::Session,
        review_prompt: String,
        model: String,
        disabled_tools: Option<BTreeSet<String>>,
    ) -> Result<String, String>;
}

/// Hidden resume-message `runtime_kind` metadata value for a bash-completion
/// self-resume (issue #84 Phase 2b). Shared by the producer (the self-resume
/// task that appends the resume message) and the consumer (the suspend-
/// finalization discriminant arm that preserves it), so a typo in one cannot
/// desync from the other and silently drop the resume trigger.
pub const BASH_COMPLETION_RESUME_KIND: &str = "bash_completion_resume";

/// Re-exported so peers that already `use crate::runtime::config::{BashResumeHook, …}`
/// (the runtime/spawn threading) can name the completion sink the same way,
/// rather than reaching into `bamboo_agent_core` separately.
pub use bamboo_agent_core::BashCompletionSink;

/// Late-bound hook that arranges a self-resume for a session suspended waiting
/// on background Bash shells (issue #84 Phase 2b). Injected per-request on
/// [`AgentLoopConfig`] exactly like [`GuardianSpawner`]; the implementation
/// lives in the session-app layer (on the completion coordinator) where the
/// resume port ([`crate::session_app::resume::ResumeExecutionPort`]) is
/// reachable.
///
/// The hook spawns a detached task that **polls the live background-shell
/// registry** until every captured shell is no longer running, then clears the
/// wait and resumes the session. Polling — not the one-shot `BashCompleted`
/// event — is the liveness guarantee: even if a shell completes between the
/// suspend snapshot and the hook's first poll, or before any event subscriber
/// exists, the registry will report it as not-running and the session resumes.
pub trait BashResumeHook: Send + Sync {
    /// Arrange a detached self-resume for `session_id`, which has just been
    /// durably suspended waiting on the background shells in `bash_ids`.
    fn arrange_bash_self_resume(&self, session_id: String, bash_ids: Vec<String>);
}

/// A child sub-agent's request to have a gated tool approved by its parent.
///
/// A non-bypassed child cannot answer its own permission prompt (no human is
/// attached to a child session), so the request is delegated up to the parent.
#[derive(Debug, Clone)]
pub struct ChildApprovalRequest {
    pub child_session_id: String,
    pub parent_session_id: String,
    /// The gated tool call on the child to re-execute once approved.
    pub child_tool_call_id: String,
    pub tool_name: String,
    /// Permission type as a string (e.g. "WriteFile", "ExecuteCommand").
    pub permission_type: String,
    /// The concrete resource the permission applies to (path, command, …).
    pub resource: String,
    /// Human-facing approval question to surface on the parent.
    pub question: String,
    /// The raw `awaiting_permission_approval` payload the child's executor built,
    /// so the parent can reuse the existing grant-extraction path verbatim.
    pub approval_payload: serde_json::Value,
}

/// What the executor should do after delegating a child's approval upward.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChildApprovalOutcome {
    /// Registered on the parent; the child must SUSPEND and await the decision.
    Delegated,
    /// Parent policy auto-approved (bypass / existing grant); proceed to execute.
    AutoApproved,
    /// Parent policy auto-denied; the executor must deny the tool.
    AutoDenied,
}

/// Late-bound delegate that routes a child's approval request up to its parent.
///
/// Injected per-request on [`AgentLoopConfig`] exactly like [`GuardianSpawner`];
/// the trait lives in the engine, the implementation in the server (it owns the
/// parent session store + pending-question + notification machinery).
#[async_trait::async_trait]
pub trait ApprovalDelegate: Send + Sync {
    /// Register `request` on its parent (or auto-resolve by policy) and report
    /// what the child's executor should do next.
    async fn delegate_child_approval(
        &self,
        request: ChildApprovalRequest,
    ) -> Result<ChildApprovalOutcome, String>;
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ImageFallbackMode {
    Placeholder,
    Error,
    Ocr,
    /// Use a vision-capable LLM to describe the image, then replace the image
    /// with the textual description so that text-only models can understand
    /// the content.
    Vision,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ImageFallbackConfig {
    pub mode: ImageFallbackMode,
    /// Vision model name for `Vision` mode. Falls back to the session's main model
    /// when `None`.
    pub vision_model: Option<String>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PromptMemoryFlags {
    pub project_prompt_injection: bool,
    pub relevant_recall: bool,
    pub relevant_recall_rerank: bool,
    pub project_first_dream: bool,
    pub ledger_agenda: bool,
}

impl Default for PromptMemoryFlags {
    fn default() -> Self {
        Self {
            project_prompt_injection: true,
            relevant_recall: true,
            relevant_recall_rerank: false,
            project_first_dream: true,
            ledger_agenda: true,
        }
    }
}

impl From<&MemoryConfig> for PromptMemoryFlags {
    fn from(value: &MemoryConfig) -> Self {
        Self {
            project_prompt_injection: value.project_prompt_injection,
            relevant_recall: value.relevant_recall,
            relevant_recall_rerank: value.relevant_recall_rerank,
            project_first_dream: value.project_first_dream,
            ledger_agenda: value.ledger_agenda_injection,
        }
    }
}

/// Execution-local adapter for a registered host. Input presence is data,
/// never a Skill grant; each host owns and validates its actual caller policy.
#[async_trait::async_trait]
pub trait SkillExecutionHost: Send + Sync {
    fn observe_current_inputs(
        &self,
        session_id: &str,
        execution_id: &str,
        current: Option<&crate::runtime::managers::lifecycle::BoundedInputRequestBatch>,
    ) -> Result<(), bamboo_agent_core::AgentError>;

    async fn render_skill_prompt(
        &self,
        session: &bamboo_agent_core::Session,
        execution_id: &str,
    ) -> Result<String, bamboo_agent_core::AgentError>;

    fn finish(&self, session_id: &str, execution_id: &str);
}

/// Compatibility name for the default SDK host.
pub use SkillExecutionHost as SdkSkillExecutionHost;

/// One bounded, explicitly supplied input observation. This carries caller data,
/// never proof of currentness, a Source grant, or permission to invoke a Skill.
#[derive(Debug, PartialEq)]
pub struct UntrustedInputObservation {
    input_id: bamboo_domain::SessionMessageId,
    request: Option<bamboo_domain::SessionSkillRequest>,
}

impl UntrustedInputObservation {
    pub fn new(
        input_id: &str,
        request: Option<&bamboo_domain::SessionSkillRequest>,
    ) -> Option<Self> {
        // Bound before the owned ID/request copy. Use the existing canonical ID
        // contract, including its path rules, without trimming or normalizing.
        if input_id.is_empty() || input_id.len() > 256 {
            return None;
        }
        let input_id = bamboo_domain::SessionMessageId::parse(input_id.to_owned()).ok()?;
        if let Some(request) = request {
            request.validate().ok()?;
        }
        Some(Self {
            input_id,
            request: request.cloned(),
        })
    }

    pub fn input_id(&self) -> &str {
        self.input_id.as_str()
    }
    pub fn request(&self) -> Option<&bamboo_domain::SessionSkillRequest> {
        self.request.as_ref()
    }
}

/// Execution-owned transport, with at most one existing Inbox admission batch
/// (128 records), each retaining only an ID and an I-W-bounded request. There
/// is no whole-batch Clone/Arc or body/config/Source snapshot. This is not Q-B's
/// aggregate compact projection or its eventual current-input classification.
pub struct UntrustedExecutionInputs {
    observations: Box<[UntrustedInputObservation]>,
    startup: std::sync::Mutex<Option<StartupInputObservation>>,
    sdk_execution_id: Option<String>,
    sdk_hook_runner: Option<Arc<HookRunner>>,
    sdk_pending: std::sync::Mutex<Option<SdkPendingInput>>,
}

#[derive(Debug)]
pub(crate) struct SdkPendingInput {
    pub session_id: String,
    pub input_id: String,
    pub header_digest: [u8; 32],
    pub final_digest: Option<[u8; 32]>,
}

impl std::fmt::Debug for UntrustedExecutionInputs {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("UntrustedExecutionInputs")
            .field("observations", &self.observations)
            .field("sdk_execution_id", &self.sdk_execution_id)
            .finish_non_exhaustive()
    }
}

/// Hash one original record, without retaining a body in the input carrier.
pub(crate) fn sdk_message_digest(message: &bamboo_agent_core::Message) -> [u8; 32] {
    use sha2::{Digest, Sha256};
    let bytes = serde_json::to_vec(message).expect("Message serialization is infallible");
    Sha256::digest(bytes).into()
}

/// F may alter content and the first native Text, while identity and all other
/// original fields must remain bound to the SDK producer.
pub(crate) fn sdk_message_header_digest(message: &bamboo_agent_core::Message) -> [u8; 32] {
    let mut header = message.clone();
    header.content.clear();
    if let Some(parts) = header.content_parts.as_mut() {
        if let Some(bamboo_domain::MessagePart::Text { text }) =
            parts.iter_mut().find(|part| matches!(part, bamboo_domain::MessagePart::Text { text } if text == &message.content))
        {
            text.clear();
        }
    }
    sdk_message_digest(&header)
}

#[derive(Debug)]
struct StartupInputObservation {
    session_id: String,
    execution_id: String,
    observation: crate::runtime::managers::lifecycle::InputObservation,
}

impl UntrustedExecutionInputs {
    pub fn new(observations: Vec<UntrustedInputObservation>) -> Option<Self> {
        if observations.is_empty() || observations.len() > 128 {
            return None;
        }
        let mut ids = std::collections::HashSet::with_capacity(observations.len());
        if observations.iter().any(|item| !ids.insert(item.input_id())) {
            return None;
        }
        Some(Self {
            observations: observations.into_boxed_slice(),
            startup: std::sync::Mutex::new(None),
            sdk_execution_id: None,
            sdk_hook_runner: None,
            sdk_pending: std::sync::Mutex::new(None),
        })
    }

    /// Exact producer-reserved execution identity; generic carriers have none.
    pub fn sdk_execution_id(&self) -> Option<&str> {
        self.sdk_execution_id.as_deref()
    }

    pub(crate) fn bind_sdk_append(
        mut self,
        execution_id: String,
        pending: SdkPendingInput,
        hook_runner: Option<Arc<HookRunner>>,
    ) -> Self {
        // Reserve the real execution identity even when optional acceptance
        // fails. Identity is not a positive input observation or Source grant.
        self.startup = std::sync::Mutex::new(Some(StartupInputObservation {
            session_id: pending.session_id.clone(),
            execution_id: execution_id.clone(),
            observation: Default::default(),
        }));
        self.sdk_execution_id = Some(execution_id);
        self.sdk_hook_runner = hook_runner;
        self.sdk_pending = std::sync::Mutex::new(Some(pending));
        self
    }

    pub(crate) fn sdk_hook_runner(&self) -> Option<Arc<HookRunner>> {
        self.sdk_hook_runner.clone()
    }

    /// Consume the private append receipt after the real startup hook, then
    /// checkpoint and verify the exact final User before publishing New data.
    pub(crate) async fn seal_sdk_input(
        &self,
        session: &mut bamboo_agent_core::Session,
        persistence: Option<&Arc<dyn RuntimeSessionPersistence>>,
    ) -> Result<(), bamboo_agent_core::AgentError> {
        let pending = self
            .sdk_pending
            .lock()
            .ok()
            .and_then(|mut slot| slot.take());
        let Some(pending) = pending else {
            return Ok(());
        };
        let execution_id = self.sdk_execution_id.as_deref().unwrap_or("");
        let invalid = || {
            bamboo_agent_core::AgentError::Tool("SDK input acceptance could not be verified".into())
        };
        if session.id != pending.session_id || execution_id.is_empty() {
            return Err(invalid());
        }
        let mut matching = session
            .messages
            .iter()
            .filter(|message| message.id == pending.input_id);
        let message = matching.next().ok_or_else(invalid)?;
        if matching.next().is_some()
            || message.role != bamboo_agent_core::Role::User
            || sdk_message_header_digest(message) != pending.header_digest
            || pending
                .final_digest
                .is_some_and(|digest| sdk_message_digest(message) != digest)
        {
            return Err(invalid());
        }
        let final_digest = sdk_message_digest(message);
        let created_at = message.created_at;
        let persistence = persistence.ok_or_else(invalid)?;
        persistence
            .checkpoint_runtime_session(session)
            .await
            .map_err(|error| {
                bamboo_agent_core::AgentError::Tool(format!("SDK input checkpoint failed: {error}"))
            })?;
        let accepted = persistence
            .load_runtime_session(&session.id)
            .await
            .map_err(|error| {
                bamboo_agent_core::AgentError::Tool(format!(
                    "SDK input verification failed: {error}"
                ))
            })?
            .ok_or_else(invalid)?;
        let mut matching = accepted
            .messages
            .iter()
            .filter(|message| message.id == pending.input_id);
        let accepted_user = matching.next().ok_or_else(invalid)?;
        if matching.next().is_some()
            || accepted_user.role != bamboo_agent_core::Role::User
            || sdk_message_digest(accepted_user) != final_digest
        {
            return Err(invalid());
        }
        let observation = self
            .observations
            .iter()
            .find(|item| item.input_id() == pending.input_id)
            .ok_or_else(invalid)?;
        let source = bamboo_domain::SessionMessageSource::User;
        let record = crate::runtime::runner::state_bridge::BorrowedInputRequestRecord {
            input_id: &pending.input_id,
            source: &source,
            kind: bamboo_domain::SessionMessageKind::UserInput,
            wrapper: None,
            created_at,
            request: observation.request(),
        };
        let projection = crate::runtime::runner::state_bridge::project_input_request_batch(
            &session.id,
            execution_id,
            &[record],
        );
        let observation =
            crate::runtime::managers::lifecycle::InputObservation::projected(projection);
        *self.startup.lock().map_err(|_| invalid())? = Some(StartupInputObservation {
            session_id: session.id.clone(),
            execution_id: execution_id.to_owned(),
            observation,
        });
        Ok(())
    }

    pub fn observations(&self) -> &[UntrustedInputObservation] {
        &self.observations
    }

    /// Additive checked boundary companion. Original public refresh layout stays
    /// unchanged; the caller still performs its original error/persist checks.
    /// An unavailable projection changes only this optional data handoff.
    pub async fn admit_with_startup_observation(
        session: &mut bamboo_agent_core::Session,
        storage: Option<&Arc<dyn bamboo_agent_core::storage::Storage>>,
        persistence: Option<&Arc<dyn bamboo_domain::RuntimeSessionPersistence>>,
        inbox: Option<&Arc<dyn bamboo_domain::SessionInboxPort>>,
    ) -> (crate::runtime::runner::TurnBoundaryRefresh, Option<Self>) {
        let execution_id = crate::runtime::runner::round_prelude::new_execution_id();
        let (refresh, observation) =
            crate::runtime::runner::state_bridge::refresh_turn_boundary_with_observation(
                session,
                storage,
                persistence,
                inbox,
                None,
                &execution_id,
            )
            .await;
        // Compatibility slice is created only after complete aggregate fit.
        // It retains one extra initial IE view: bounded slots and args AST,
        // whose compact payload fits the same cap (not a heap/RSS bound).
        let carrier = observation
            .new_inputs()
            .and_then(|batch| {
                Self::new(
                    batch
                        .records()
                        .iter()
                        .map(|record| {
                            Some(UntrustedInputObservation {
                                input_id: bamboo_domain::SessionMessageId::parse(
                                    record.input_id.clone(),
                                )
                                .ok()?,
                                request: record.request.clone(),
                            })
                        })
                        .collect::<Option<Vec<_>>>()?,
                )
            })
            .map(|mut carrier| {
                carrier.startup = std::sync::Mutex::new(Some(StartupInputObservation {
                    session_id: session.id.clone(),
                    execution_id,
                    observation,
                }));
                carrier
            });
        (refresh, carrier)
    }

    pub(crate) fn take_startup_observation(
        &self,
        session_id: &str,
    ) -> Option<(
        String,
        crate::runtime::managers::lifecycle::InputObservation,
    )> {
        let startup = self.startup.lock().ok()?.take()?;
        (startup.session_id == session_id
            && (startup.observation.is_unavailable()
                || startup.observation.matches_session(session_id)))
        .then_some((startup.execution_id, startup.observation))
    }

    /// Caller supplies ONLY this call's newly checkpointed messages, after its
    /// ACK/rejection checks. Never call this on stored Session history. The
    /// existing admission producer already checked their envelope proof.
    pub fn from_committed_messages(messages: &[bamboo_agent_core::Message]) -> Option<Self> {
        if messages.len() > 128 {
            return None;
        }
        let mut observations = Vec::with_capacity(messages.len());
        for message in messages {
            if message.role != bamboo_agent_core::Role::User {
                continue;
            }
            let Some(proof) = message
                .metadata
                .as_ref()
                .and_then(|metadata| metadata.get("session_message"))
            else {
                continue;
            };
            if proof.get("id")?.as_str()? != message.id {
                return None;
            }
            let body = proof.get("body")?;
            let request = if proof.get("source")?.get("type")?.as_str()? == "user"
                && proof.get("kind")?.as_str()? == "user_input"
                && body.get("type")?.as_str()? == "content"
            {
                body.get("skill_request")
            } else if proof.get("source")?.get("type")?.as_str()? == "runtime"
                && proof.get("source")?.get("subsystem")?.as_str()? == "chat"
                && proof.get("kind")?.as_str()? == "runtime_instruction"
                && body.get("instruction")?.as_str()? == "root_chat_turn_v1"
            {
                body.get("content")?.get("skill_request")
            } else {
                continue;
            };
            let request = request
                .filter(|value| !value.is_null())
                .map(copy_checked_request)
                .transpose()
                .ok()?;
            if message.id.is_empty() || message.id.len() > 256 {
                return None;
            }
            let input_id = bamboo_domain::SessionMessageId::parse(message.id.clone()).ok()?;
            observations.push(UntrustedInputObservation { input_id, request });
        }
        Self::new(observations)
    }
}

// Read only the exact bounded request, never deserialize/copy the Message,
// images, body, proof, or publication metadata merely to transport caller data.
fn copy_checked_request(
    value: &serde_json::Value,
) -> Result<bamboo_domain::SessionSkillRequest, ()> {
    let selections = value
        .get("selections")
        .and_then(serde_json::Value::as_array)
        .ok_or(())?;
    if selections.is_empty() || selections.len() > 32 {
        return Err(());
    }
    let mode = match value.get("mode") {
        None | Some(serde_json::Value::Null) => None,
        Some(value) => {
            let mode = value.as_str().ok_or(())?;
            if mode.is_empty()
                || mode.len() > 64
                || !mode
                    .bytes()
                    .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
            {
                return Err(());
            }
            Some(mode)
        }
    };
    // Bound every original args tree before copying any selection. Existing
    // I-W validation supplies nonrecursive depth/node/compact-byte checks.
    for selection in selections {
        bamboo_domain::SessionSkillSelection::validate_borrowed(
            selection
                .get("id")
                .and_then(serde_json::Value::as_str)
                .ok_or(())?,
            selection
                .get("source")
                .and_then(serde_json::Value::as_str)
                .ok_or(())?,
            selection
                .get("revision")
                .and_then(serde_json::Value::as_u64)
                .ok_or(())?,
            selection.get("args").unwrap_or(&serde_json::Value::Null),
        )
        .map_err(|_| ())?;
    }
    let request = bamboo_domain::SessionSkillRequest {
        selections: selections
            .iter()
            .map(|selection| bamboo_domain::SessionSkillSelection {
                id: selection["id"].as_str().unwrap().to_owned(),
                source: selection["source"].as_str().unwrap().to_owned(),
                revision: selection["revision"].as_u64().unwrap(),
                args: selection.get("args").cloned().unwrap_or_default(),
            })
            .collect(),
        mode: mode.map(str::to_owned),
    };
    request.validate().map_err(|_| ())?;
    Ok(request)
}

/// Configuration for the agent loop.
///
/// # One-config-per-run invariant (#44)
///
/// These values are SNAPSHOTTED once, from the live `Config` under a brief read
/// lock, at the start of `AgentRuntime::execute()`. The entire multi-round run —
/// which can last minutes — then uses this frozen snapshot. Changing config
/// (model names, provider, `disabled_tools`/`disabled_skills`, memory flags,
/// token budget, …) while a run is in flight does NOT affect that run; the new
/// values are picked up on the NEXT execution (i.e. the next user turn / session
/// restart), not the next round of the current run.
///
/// This is intentional: a run sees a stable configuration, so its behavior can't
/// shift underneath it mid-execution. The deliberate exceptions are the
/// late-bound handles that resolve LIVE each time they're
/// used rather than being snapshotted — `auxiliary_model_resolver` (auxiliary
/// model selection), `model_limits_section` (validated model limits), and
/// `guardian_spawner` (the reviewer child). If a frozen
/// field ever needs to become live-per-round, follow that resolver pattern
/// rather than widening the snapshot.
#[non_exhaustive]
pub struct AgentLoopConfig {
    /// Keep tool guidance stable while activation updates append to model context.
    pub freeze_tool_exposure_for_cache: bool,
    /// Host-installed exact Assignment/Run permit; never taken from model args.
    pub(crate) ticket_worker_plan: Option<Arc<dyn crate::ticket_worker_plan::WorkerLocalPlan>>,
    pub(crate) system_prompt: Option<String>,
    /// Skill IDs that are disabled globally for this execution.
    pub(crate) disabled_skill_ids: BTreeSet<String>,
    /// Optional explicit skill selection for this execution.
    /// When set, only these skill IDs are considered for skill context and allowlists.
    pub(crate) selected_skill_ids: Option<Vec<String>>,
    /// Optional active skill mode for this execution.
    ///
    /// When set, skill discovery prefers `skills-<mode>` directories over generic
    /// directories for the same skill id.
    pub(crate) selected_skill_mode: Option<String>,
    /// Untrusted inputs supplied by this startup call, never inferred.
    pub(crate) initial_untrusted_inputs: Option<UntrustedExecutionInputs>,
    pub(crate) sdk_skill_execution_host: Option<Arc<dyn SdkSkillExecutionHost>>,
    pub(crate) additional_tool_schemas: Vec<ToolSchema>,
    pub(crate) tool_registry: Arc<ToolRegistry>,
    pub(crate) skill_manager: Option<Arc<SkillManager>>,
    /// Stable Project identity/resource resolver. The server wires the
    /// authoritative registry adapter once on `AgentRuntimeBuilder`.
    pub(crate) project_context_resolver:
        Option<Arc<crate::project_context::ProjectContextResolver>>,
    /// If true, skip appending the initial user message (already present in session).
    pub(crate) skip_initial_user_message: bool,
    /// Optional storage for persisting session changes
    pub(crate) storage: Option<Arc<dyn Storage>>,
    /// Optional runtime persistence for non-authoritative session saves.
    /// When set, engine save sites use this instead of `storage` for writes.
    pub(crate) persistence: Option<Arc<dyn RuntimeSessionPersistence>>,
    pub(crate) guidance_active_run_id: Option<String>,
    /// Durable logical-session inbox admitted at safe round boundaries.
    pub(crate) session_inbox: Option<Arc<dyn bamboo_domain::SessionInboxPort>>,
    /// Active-owner wake generation. The loop consumes this at the same safe
    /// boundary where it drains SessionInbox; it never interrupts an in-flight
    /// provider/tool operation.
    pub(crate) session_activation_notifications:
        Option<Arc<parking_lot::Mutex<tokio::sync::watch::Receiver<u64>>>>,
    /// Optional attachment reader for resolving `bamboo-attachment://...` references
    /// into `data:` URLs for upstream providers. This must not mutate session storage.
    pub(crate) attachment_reader: Option<Arc<dyn AttachmentReader>>,
    /// Optional asynchronous metrics collector
    pub(crate) metrics_collector: Option<MetricsCollector>,
    /// Model name used for metrics attribution
    pub(crate) model_name: Option<String>,
    /// Fast/cheap model for lightweight tasks (task evaluation, search, etc.).
    ///
    /// Call sites may fall back to `model_name` when this is unset.
    pub(crate) fast_model_name: Option<String>,
    /// Optional provider override for lightweight fast-model LLM calls.
    pub(crate) fast_model_provider: Option<Arc<dyn LLMProvider>>,
    /// Process-wide concurrency ceiling for low-priority auxiliary evaluation
    /// requests sharing the same provider allocation and model. Foreground
    /// requests do not acquire this budget.
    pub(crate) auxiliary_evaluation_max_concurrency: usize,
    /// Fast/cheap model for memory/background tasks.
    ///
    /// Resolution follows the shared dedicated → fast → chat fallback chain.
    pub(crate) background_model_name: Option<String>,

    /// Model for planning/coordination tasks (task decomposition, architecture).
    /// Falls back to `model_name` when unset.
    pub(crate) planning_model_name: Option<String>,
    /// Model for search/navigation tasks (grep, file listing, symbol resolution).
    /// Falls back to `fast_model_name` when unset.
    pub(crate) search_model_name: Option<String>,
    /// Custom instructions for conversation summarization, injected into the
    /// LLM summary prompt. Lets users control what the summary focuses on.
    ///
    /// Resolution order: session-level > config-level > built-in defaults.
    pub(crate) compression_instructions: Option<String>,
    /// Desired final summary size relative to the raw source tokens represented
    /// by it. Values are normalized at the compression boundary.
    pub(crate) summary_target_ratio: f64,
    /// Frozen host context-management strategy for this run. Missing public
    /// configuration resolves to the legacy summary behavior.
    pub(crate) context_management: bamboo_config::ContextManagementConfig,
    /// Safe request ceiling as a percentage of the selected summarization
    /// model's context window.
    pub(crate) summary_safe_window_percent: u8,
    /// Dedicated model for summarization. Falls back to `background_model_name`.
    pub(crate) summarization_model_name: Option<String>,
    /// Optional provider override for memory/background model LLM calls.
    ///
    /// When set, memory recall rerank and other memory/background tasks use this
    /// provider instead of the shared agent loop provider.
    pub(crate) background_model_provider: Option<Arc<dyn LLMProvider>>,
    /// Optional provider override for summarization / context compression calls.
    ///
    /// When set, conversation/task summarization uses this provider instead of
    /// the shared agent loop provider.
    pub(crate) summarization_model_provider: Option<Arc<dyn LLMProvider>>,
    /// Provider routing key used for provider-specific request behavior.
    ///
    /// In multi-instance mode this may be the instance id.
    pub(crate) provider_name: Option<String>,
    /// Underlying provider type (for example `openai`, `anthropic`, `copilot`).
    ///
    /// This is distinct from `provider_name` so provider-specific behavior can
    /// remain correct when routing keys are instance ids.
    pub(crate) provider_type: Option<String>,
    /// Optional request-time reasoning effort override.
    pub(crate) reasoning_effort: Option<ReasoningEffort>,
    /// Bamboo application data directory (typically `~/.bamboo`).
    ///
    /// Used by runtime features that persist auxiliary artifacts outside the
    /// session store, such as durable plan mode files under `~/.bamboo/plan`.
    pub(crate) app_data_dir: Option<PathBuf>,
    /// Jiandu memory handle for this run. The default is the independent
    /// `~/.jiandu` store; explicit construction is used by isolated tests.
    pub(crate) memory_store: bamboo_memory::memory_store::MemoryStore,
    /// Tool names that should be excluded from schemas sent to the LLM.
    pub(crate) disabled_tools: BTreeSet<String>,
    /// Token budget for context management (optional, defaults to model's limits)
    pub(crate) token_budget: Option<TokenBudget>,
    /// Host-owned model-limit section. The handle stays stable for the run;
    /// each budget resolution reads its immutable last-known-good snapshot.
    /// Standalone runtimes without this handle retain instance-local file loading.
    pub(crate) model_limits_section:
        Option<Arc<bamboo_config::LiveSection<bamboo_config::ModelLimitsSection>>>,
    /// Legacy `config.json` `model_limits` value, snapshotted from the live
    /// in-memory Config when this loop config is built. Consulted per model when
    /// the host section or standalone `model_limits.json` has no matching entry
    /// or cannot be loaded, so the engine never does a disk-reading `Config::new()`
    /// (which would also clobber the global env-var cache). #38.
    pub(crate) legacy_model_limits: Option<serde_json::Value>,
    /// Optional image fallback behavior applied to *LLM requests only* (never persisted).
    ///
    /// This is intended for text-only provider paths where image parts must be degraded
    /// (placeholder / OCR / error) without leaking into stored session history or UI.
    pub(crate) image_fallback: Option<ImageFallbackConfig>,
    /// Feature flags controlling prompt-time memory injection behavior.
    pub(crate) prompt_memory_flags: PromptMemoryFlags,
    /// Maximum tool calls allowed per round (default: 80).
    pub(crate) max_tool_calls_per_round: usize,
    /// Maximum consecutive failures per tool before circuit breaker (default: 3).
    pub(crate) max_consecutive_failures_per_tool: usize,
    /// Per-tool execution timeout in seconds (default: 120).
    pub(crate) per_tool_timeout_secs: u64,
    /// Parallel batch execution timeout in seconds (default: 300).
    pub(crate) parallel_batch_timeout_secs: u64,
    /// Resolved LLM stream transport/semantic watchdog policy. The same value
    /// is passed to main response streams and auxiliary silent streams.
    pub(crate) stream_timeout: bamboo_config::StreamTimeoutConfig,
    /// Permission mode for this execution (default: None = use PermissionConfig's mode).
    pub(crate) permission_mode: Option<PermissionMode>,
    /// Optional Gold observe-only evaluator configuration.
    ///
    /// When `None` or `enabled == false`, Gold evaluation is disabled and the
    /// existing execute/respond/resume loop remains unchanged.
    pub(crate) gold_config: Option<GoldConfig>,
    /// Optional guardian adversarial-review gate configuration. When `None` or
    /// `enabled == false`, the guardian terminal gate is inactive.
    pub(crate) guardian_config: Option<GuardianConfig>,
    /// Late-bound spawner for the guardian reviewer child. `None` (the default)
    /// leaves the guardian gate inert even when `guardian_config.enabled` is set,
    /// since the runner cannot create a child without it. Wired by the server.
    pub(crate) guardian_spawner: Option<Arc<dyn GuardianSpawner>>,
    /// Late-bound hook that arranges a self-resume for a session suspended
    /// waiting on background Bash shells (issue #84 Phase 2b). `None` (the
    /// default) leaves the bash suspend gate inert: the gate refuses to suspend
    /// without a wired hook, so a session can never strand itself without a
    /// resume path. Wired by the server (the completion coordinator impl).
    pub(crate) bash_resume_hook: Option<Arc<dyn BashResumeHook>>,
    /// Late-bound sink that pushes a completed background Bash shell's result
    /// into this session's loop (issue #84 Phase 2b follow-up) — injected at the
    /// next round boundary while the loop is actively iterating, or delivered via
    /// resume when it is idle. Threaded onto the tool dispatch context (like
    /// `can_async_resume`) so the Bash tool can hand it to the shell's
    /// completion-poll task. `None` (the default) leaves the push inert; the
    /// durable end-of-turn suspend/poll backstop (`bash_resume_hook`) still runs.
    /// Wired by the server (the completion coordinator impl).
    pub(crate) bash_completion_sink: Option<Arc<dyn bamboo_agent_core::BashCompletionSink>>,
    /// Late-bound delegate that routes a child's gated-tool approval request up
    /// to its parent (Phase 2). `None` (the default) leaves child gating on its
    /// legacy path. Wired by the server.
    pub(crate) approval_delegate: Option<Arc<dyn ApprovalDelegate>>,
    /// Frozen lifecycle-hook registry for this run. The default registry is
    /// empty, and every seam checks `has_hooks_for` before constructing payloads.
    pub(crate) hook_runner: Arc<HookRunner>,
    /// Enable dynamic per-round model routing based on task complexity.
    /// When true, the pipeline classifies complexity at each round end and
    /// stores the result in session metadata.
    pub(crate) features_dynamic_model_routing: bool,
    /// Opt-in, single auxiliary check before publishing a final answer.
    pub(crate) features_final_evidence_check: bool,
    /// Optional per-round resolver for auxiliary model settings that should
    /// follow live global config rather than stay frozen for the whole run.
    ///
    /// The main chat model remains session/request scoped; this hook is only
    /// for fast/background/planning/search/summarization helpers.
    pub(crate) auxiliary_model_resolver:
        Option<Arc<dyn Fn() -> AuxiliaryModelConfig + Send + Sync>>,
    /// Optional per-round resolver for the disabled tool/skill sets so they follow
    /// LIVE global config instead of staying frozen for the whole run. Returns the
    /// current `(disabled_tools, disabled_skill_ids)`. When `None`, the snapshotted
    /// `disabled_tools` / `disabled_skill_ids` fields below are used (#44 behavior).
    /// Re-resolved each round at the tool-schema filter, so disabling/re-enabling a
    /// tool mid-run takes effect on the next round. #136.
    pub(crate) disabled_filter_resolver: Option<DisabledFilterResolver>,
    /// Server-level usage guidance contributed by the run's tool executor —
    /// chiefly the `instructions` connected MCP servers return from `initialize`.
    /// Captured once at config construction (from `ToolExecutor::tool_guidance`)
    /// and appended to the tool-guide section of the system prompt, so a server's
    /// own how-to-use notes appear only while that server is loaded for the run.
    pub(crate) mcp_tool_guidance: Option<String>,
    /// Per-run resource guardrails (issue #221): already resolved — the
    /// per-request override merged over the config-level default (see
    /// [`AgentRuntime::execute`](crate::runtime::runtime::AgentRuntime::execute)).
    /// Checked after every round; exceeding a configured limit gracefully
    /// stops the run. Includes the run's round cap (`max_rounds`): `None`
    /// means unlimited rounds (the historical hard-coded 200 was removed);
    /// a configured cap keeps the issue #29 exhaustion contract (metadata
    /// stamp, visible notification, exactly one final summary turn).
    pub(crate) run_budget: bamboo_config::RunBudgetConfig,
    pub(crate) counted_loop: Option<super::counted_loop::CountedLoopConfig>,
}

impl AgentLoopConfig {
    /// Data-only execution transport. This accessor does not classify or grant.
    pub fn initial_untrusted_inputs(&self) -> Option<&UntrustedExecutionInputs> {
        self.initial_untrusted_inputs.as_ref()
    }
}

impl Default for AgentLoopConfig {
    fn default() -> Self {
        Self {
            freeze_tool_exposure_for_cache: true,
            ticket_worker_plan: None,
            system_prompt: None,
            disabled_skill_ids: BTreeSet::new(),
            selected_skill_ids: None,
            selected_skill_mode: None,
            initial_untrusted_inputs: None,
            sdk_skill_execution_host: None,
            additional_tool_schemas: Vec::new(),
            tool_registry: Arc::new(ToolRegistry::new()),
            skill_manager: None,
            project_context_resolver: None,
            skip_initial_user_message: false,
            storage: None,
            persistence: None,
            guidance_active_run_id: None,
            session_inbox: None,
            session_activation_notifications: None,
            attachment_reader: None,
            metrics_collector: None,
            model_name: None,
            fast_model_name: None,
            fast_model_provider: None,
            auxiliary_evaluation_max_concurrency: DEFAULT_AUXILIARY_EVALUATION_MAX_CONCURRENCY,
            background_model_name: None,
            planning_model_name: None,
            search_model_name: None,
            compression_instructions: None,
            summary_target_ratio: 0.20,
            context_management: bamboo_config::ContextManagementConfig::default(),
            summary_safe_window_percent: 80,
            summarization_model_name: None,
            background_model_provider: None,
            summarization_model_provider: None,
            provider_name: None,
            provider_type: None,
            reasoning_effort: None,
            app_data_dir: None,
            memory_store: bamboo_memory::memory_store::MemoryStore::with_defaults(),
            disabled_tools: BTreeSet::new(),
            token_budget: None,
            model_limits_section: None,
            legacy_model_limits: None,
            image_fallback: None,
            prompt_memory_flags: PromptMemoryFlags::default(),
            max_tool_calls_per_round: 80,
            max_consecutive_failures_per_tool: 3,
            per_tool_timeout_secs: 120,
            parallel_batch_timeout_secs: 300,
            stream_timeout: bamboo_config::StreamTimeoutConfig::default(),
            permission_mode: None,
            gold_config: None,
            guardian_config: None,
            guardian_spawner: None,
            bash_resume_hook: None,
            bash_completion_sink: None,
            approval_delegate: None,
            hook_runner: Arc::new(HookRunner::new()),
            features_dynamic_model_routing: false,
            features_final_evidence_check: false,
            auxiliary_model_resolver: None,
            disabled_filter_resolver: None,
            mcp_tool_guidance: None,
            run_budget: bamboo_config::RunBudgetConfig::default(),
            counted_loop: None,
        }
    }
}

impl AgentLoopConfig {
    pub fn bind_ticket_worker_plan(
        &mut self,
        plan: Arc<dyn crate::ticket_worker_plan::WorkerLocalPlan>,
    ) {
        self.ticket_worker_plan = Some(plan);
    }
    /// Live `(disabled_tools, disabled_skill_ids)` for the current round: the
    /// resolver if one is wired (#136 — follows live global config between
    /// rounds), else the per-run snapshot (#44 frozen behavior). `Cow` avoids
    /// cloning the snapshot in the common no-resolver path (SDK / tests).
    pub(crate) fn resolve_disabled_filters(
        &self,
    ) -> (
        std::borrow::Cow<'_, BTreeSet<String>>,
        std::borrow::Cow<'_, BTreeSet<String>>,
    ) {
        match &self.disabled_filter_resolver {
            Some(resolver) => {
                let (tools, skills) = resolver();
                (
                    std::borrow::Cow::Owned(tools),
                    std::borrow::Cow::Owned(skills),
                )
            }
            None => (
                std::borrow::Cow::Borrowed(&self.disabled_tools),
                std::borrow::Cow::Borrowed(&self.disabled_skill_ids),
            ),
        }
    }

    /// The active session goal to surface to the main agent, or `None` when
    /// Gold is disabled or no goal is set. Falls back to the legacy
    /// `evaluation_prompt` for back-compat via [`GoldConfig::effective_goal`].
    pub fn active_goal(&self) -> Option<&str> {
        self.gold_config
            .as_ref()
            .filter(|cfg| cfg.enabled)
            .and_then(GoldConfig::effective_goal)
    }

    /// Whether the Codex-style autonomous goal loop is active for this run.
    ///
    /// This requires Gold to be enabled, a goal to be set, AND auto-continue to
    /// be on. Only then is the `update_goal` self-report tool surfaced to the
    /// model and the terminal double-check allowed to veto a premature stop.
    /// When Gold is enabled without auto-continue, the evaluator stays purely
    /// observational (legacy behavior).
    pub fn goal_loop_active(&self) -> bool {
        self.gold_config.as_ref().is_some_and(|cfg| {
            cfg.enabled && cfg.auto_continue_enabled && cfg.effective_goal().is_some()
        })
    }

    /// Whether the guardian review gate is active for this run: a spawner is
    /// wired (so the runner can actually create the reviewer child) AND the
    /// config is present and enabled.
    pub fn guardian_active(&self) -> bool {
        self.guardian_spawner.is_some()
            && self.guardian_config.as_ref().is_some_and(|cfg| cfg.enabled)
    }

    /// Maximum guardian review passes for this run (the budget). `0` when no
    /// guardian config is set.
    pub fn guardian_max_reviews(&self) -> u32 {
        self.guardian_config
            .as_ref()
            .map_or(0, |cfg| cfg.max_reviews)
    }

    /// The reviewer model override, if a guardian config sets one.
    pub fn guardian_model(&self) -> Option<&str> {
        self.guardian_config
            .as_ref()
            .and_then(|cfg| cfg.model_name.as_deref())
    }

    /// Whether child→parent approval delegation is wired for this run.
    pub fn delegation_active(&self) -> bool {
        self.approval_delegate.is_some()
    }
}

#[cfg(test)]
mod tests;

#[cfg(test)]
mod execution_input_tests {
    use super::*;
    fn request() -> bamboo_domain::SessionSkillRequest {
        bamboo_domain::SessionSkillRequest {
            selections: vec![bamboo_domain::SessionSkillSelection {
                id: "exact Case".into(),
                source: "builtin".into(),
                revision: 1,
                args: serde_json::json!({"raw":"保持原样"}),
            }],
            mode: None,
        }
    }
    #[test]
    fn execution_input_owned_bounds_and_old_config_default() {
        assert!(AgentLoopConfig::default()
            .initial_untrusted_inputs()
            .is_none());
        let request = request();
        let input = UntrustedInputObservation::new("exact-input", Some(&request)).unwrap();
        assert_eq!(input.input_id(), "exact-input");
        assert_eq!(input.request(), Some(&request));
        for id in [
            "",
            " leading",
            "trailing ",
            "unsafe/id",
            "unsafe\\id",
            "unsafe..id",
        ] {
            assert!(UntrustedInputObservation::new(id, None).is_none());
        }
        assert!(UntrustedInputObservation::new(&"a".repeat(257), None).is_none());
        assert!(UntrustedExecutionInputs::new(Vec::new()).is_none());
        let batch = UntrustedExecutionInputs::new(
            (0..128)
                .map(|i| {
                    UntrustedInputObservation::new(&format!("input-{i}"), Some(&request)).unwrap()
                })
                .collect(),
        )
        .unwrap();
        assert_eq!(batch.observations().len(), 128);
        assert_eq!(batch.observations()[127].input_id(), "input-127");
        assert!(UntrustedExecutionInputs::new(
            (0..129)
                .map(|i| UntrustedInputObservation::new(&format!("input-{i}"), None).unwrap())
                .collect()
        )
        .is_none());
        assert!(UntrustedExecutionInputs::new(vec![
            UntrustedInputObservation::new("same", None).unwrap(),
            UntrustedInputObservation::new("same", None).unwrap()
        ])
        .is_none());
        let mut malformed = request;
        malformed.selections[0].args = (0..65).fold(serde_json::Value::Null, |v, _| {
            serde_json::Value::Array(vec![v])
        });
        assert!(UntrustedInputObservation::new("deep", Some(&malformed)).is_none());
    }
    #[test]
    fn execution_input_only_explicit_checked_commit_slice_copies_request_fields() {
        let request = request();
        let mut user = bamboo_domain::SessionMessageEnvelope::user_input("target", "ordinary");
        if let bamboo_domain::SessionMessageBody::Content(content) = &mut user.body {
            content.skill_request = Some(request.clone());
        }
        let id = user.id.to_string();
        let plain = user.to_provider_message().unwrap();
        let root = user
            .with_root_chat_prompt("system prompt is not retained".into())
            .unwrap()
            .to_provider_message()
            .unwrap();
        for message in [plain, root] {
            let batch = UntrustedExecutionInputs::from_committed_messages(&[message]).unwrap();
            assert_eq!(batch.observations().len(), 1);
            assert_eq!(batch.observations()[0].input_id(), id);
            assert_eq!(batch.observations()[0].request(), Some(&request));
        }
        let text = bamboo_agent_core::Message::user("{\"skill_request\":\"fragment\"}");
        assert!(UntrustedExecutionInputs::from_committed_messages(&[text]).is_none());
        assert!(UntrustedExecutionInputs::from_committed_messages(&[]).is_none());
    }
}

#[cfg(test)]
mod sdk_acceptance_tests {
    use super::*;
    use crate::runtime::managers::lifecycle::BoundedInputRequestBatch;
    use bamboo_agent_core::{Message, Role, Session};

    struct Repository {
        stored: std::sync::Mutex<Option<Session>>,
        reject: bool,
        corrupt: bool,
    }
    #[async_trait::async_trait]
    impl RuntimeSessionPersistence for Repository {
        async fn save_runtime_session(&self, session: &mut Session) -> std::io::Result<()> {
            if self.reject {
                return Err(std::io::Error::other("checkpoint refused"));
            }
            let mut accepted = session.clone();
            if self.corrupt {
                accepted
                    .messages
                    .iter_mut()
                    .find(|message| message.role == Role::User)
                    .unwrap()
                    .content
                    .push_str(" changed");
            }
            *self.stored.lock().unwrap() = Some(accepted);
            Ok(())
        }
        async fn checkpoint_runtime_session(&self, session: &mut Session) -> std::io::Result<()> {
            self.save_runtime_session(session).await
        }
        async fn load_runtime_session(&self, _: &str) -> std::io::Result<Option<Session>> {
            Ok(self.stored.lock().unwrap().clone())
        }
    }
    fn repository(reject: bool, corrupt: bool) -> Arc<dyn RuntimeSessionPersistence> {
        Arc::new(Repository {
            stored: Default::default(),
            reject,
            corrupt,
        })
    }
    fn producer(session: &Session, user: &Message, typed: bool) -> UntrustedExecutionInputs {
        UntrustedExecutionInputs::new(vec![UntrustedInputObservation::new(&user.id, None).unwrap()])
            .unwrap()
            .bind_sdk_append(
                "actual-sdk-execution".into(),
                SdkPendingInput {
                    session_id: session.id.clone(),
                    input_id: user.id.clone(),
                    header_digest: sdk_message_header_digest(user),
                    final_digest: typed.then(|| sdk_message_digest(user)),
                },
                None,
            )
    }
    fn current(
        inputs: &UntrustedExecutionInputs,
        session: &Session,
    ) -> Option<BoundedInputRequestBatch> {
        let (execution_id, observation) = inputs.take_startup_observation(&session.id).unwrap();
        assert_eq!(execution_id, "actual-sdk-execution");
        let mut current = None;
        observation.update_current(&mut current, &session.id, &execution_id);
        assert!(inputs.take_startup_observation(&session.id).is_none());
        current
    }
    #[tokio::test]
    async fn sdk_startup_requires_exact_checkpoint_and_preserves_request_absence_as_new() {
        let mut session = Session::new("sdk-accept", "model");
        let user = Message::user("final prepared");
        let inputs = producer(&session, &user, true);
        session.add_message(user.clone());
        inputs
            .seal_sdk_input(&mut session, Some(&repository(false, false)))
            .await
            .unwrap();
        let batch = current(&inputs, &session).unwrap();
        assert_eq!(batch.records().len(), 1);
        assert_eq!(batch.records()[0].input_id, user.id);
        assert_eq!(
            batch.records()[0].source,
            bamboo_domain::SessionMessageSource::User
        );
        assert!(batch.records()[0].request.is_none());
    }
    #[tokio::test]
    async fn sdk_startup_unknown_checkpoint_or_modified_final_user_stays_unavailable() {
        for (reject, corrupt, mutate, duplicate, wrong_session) in [
            (true, false, false, false, false),
            (false, true, false, false, false),
            (false, false, true, false, false),
            (false, false, false, true, false),
            (false, false, false, false, true),
        ] {
            let mut session = Session::new("sdk-deny", "model");
            let user = Message::user("prepared");
            let inputs = producer(&session, &user, true);
            session.add_message(user.clone());
            if mutate {
                session.messages[0].content.push_str(" changed");
            }
            if duplicate {
                session.add_message(user);
            }
            if wrong_session {
                session.id = "wrong".into();
            }
            assert!(inputs
                .seal_sdk_input(&mut session, Some(&repository(reject, corrupt)))
                .await
                .is_err());
            if wrong_session {
                assert!(inputs.take_startup_observation(&session.id).is_none());
            } else {
                assert!(current(&inputs, &session).is_none());
            }
        }
    }
    #[tokio::test]
    async fn sdk_legacy_hook_rewrite_is_bound_to_original_id_while_generic_history_is_unsealed() {
        let mut session = Session::new("sdk-string", "model");
        let user = Message::user("original");
        let inputs = producer(&session, &user, false);
        session.add_message(user.clone());
        session.messages[0].content = "real portable hook result".into();
        inputs
            .seal_sdk_input(&mut session, Some(&repository(false, false)))
            .await
            .unwrap();
        assert_eq!(
            current(&inputs, &session).unwrap().records()[0].input_id,
            user.id
        );
        let generic =
            UntrustedExecutionInputs::new(vec![
                UntrustedInputObservation::new(&user.id, None).unwrap()
            ])
            .unwrap();
        generic
            .seal_sdk_input(&mut session, Some(&repository(false, false)))
            .await
            .unwrap();
        assert!(generic.sdk_execution_id().is_none());
        assert!(generic.take_startup_observation(&session.id).is_none());
    }
    #[test]
    fn sdk_header_binding_accepts_only_the_canonical_prompt_changes() {
        let mut original = Message::user("canonical");
        original.content_parts = Some(vec![
            bamboo_domain::MessagePart::Text {
                text: "unrelated native text".into(),
            },
            bamboo_domain::MessagePart::Text {
                text: "canonical".into(),
            },
        ]);
        let digest = sdk_message_header_digest(&original);
        let mut prepared = original.clone();
        prepared.content = "canonical plus F".into();
        if let bamboo_domain::MessagePart::Text { text } =
            &mut prepared.content_parts.as_mut().unwrap()[1]
        {
            *text = prepared.content.clone();
        }
        assert_eq!(sdk_message_header_digest(&prepared), digest);
        prepared.id = "different".into();
        assert_ne!(sdk_message_header_digest(&prepared), digest);
        prepared = original;
        if let bamboo_domain::MessagePart::Text { text } =
            &mut prepared.content_parts.as_mut().unwrap()[0]
        {
            *text = "tampered".into();
        }
        assert_ne!(sdk_message_header_digest(&prepared), digest);
    }
}