meerkat-core 0.8.8

//! Agent - the core agent orchestrator
//!
//! The Agent struct ties together all components and runs the agent loop.

mod builder;
pub mod comms_impl;
pub mod compact;
mod extraction;
mod hook_impl;
#[cfg(test)]
mod hooks_behavior_tests;
mod runner;
pub mod skills;
mod state;
#[cfg(test)]
#[doc(hidden)]
pub(crate) mod test_turn_state_handle;
use crate::budget::Budget;
use crate::comms::{
    CommsCommand, CommsTrustMutation, CommsTrustMutationResult, EventStream, PeerDirectoryEntry,
    PeerId, SendAndStreamError, SendError, SendReceipt, StreamError, StreamScope,
    TrustedPeerDescriptor,
};
use crate::compact::SessionCompactionCadence;
use crate::completion_feed::CompletionSeq;
use crate::config::{AgentConfig, HookRunOverrides};
use crate::error::AgentError;
use crate::event::ExternalToolDelta;
use crate::hooks::HookEngine;
use crate::lifecycle::RunId;
use crate::lifecycle::run_primitive::ProviderParamsOverride;
use crate::ops::OperationId;
use crate::ops_lifecycle::{OperationKind, OperationStatus, OperationTerminalOutcome};
use crate::retry::RetryPolicy;
use crate::schema::{CompiledSchema, SchemaError};
use crate::session::Session;
use crate::state::LoopState;
#[cfg(target_arch = "wasm32")]
use crate::tokio;
use crate::tool_catalog::{
    ToolCatalogCapabilities, ToolCatalogEntry, ToolCatalogMode, deferred_session_entry_count,
    select_catalog_mode_from_snapshot,
};
use crate::tool_scope::ToolScope;
use crate::turn_execution_authority::{
    ContentShape, TurnPhase, TurnPrimitiveKind, TurnTerminalCauseKind, TurnTerminalOutcome,
};
use crate::types::{
    AssistantBlock, BlockAssistantMessage, Message, OutputSchema, StopReason, ToolCallView,
    ToolDef, ToolName, ToolNameSet, Usage,
};
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;

pub use builder::{AgentBuildPolicyError, AgentBuilder, DefaultSystemPromptPolicy};
pub use runner::{AgentRunner, SnapshotProjectionError, SystemContextStateError};

/// Trait for LLM clients that can be used with the agent
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait AgentLlmClient: Send + Sync {
    /// Stream a response from the LLM
    async fn stream_response(
        &self,
        messages: &[Message],
        tools: &[Arc<ToolDef>],
        max_tokens: u32,
        temperature: Option<f32>,
        provider_params: Option<&ProviderParamsOverride>,
    ) -> Result<LlmStreamResult, AgentError>;

    /// Get the typed catalog provider identity for this client.
    ///
    /// Clients return the typed [`crate::provider::Provider`] directly so no
    /// boundary ever parses a caller-supplied string back into catalog
    /// identity. String projections are derived via
    /// [`crate::provider::Provider::as_str`].
    fn provider(&self) -> crate::provider::Provider;

    /// Get the current effective model identifier.
    ///
    /// Used by the agent loop for profile-default resolution (e.g., call timeout
    /// defaults that vary per model family). Must reflect the current model even
    /// after hot-swap.
    fn model(&self) -> &str;

    /// Prepare the next prebuilt fallback model after the generated turn
    /// authority has classified the LLM failure as recoverable.
    ///
    /// This method does not classify failures and must not call the provider.
    /// It only selects an already-constructed candidate and returns the typed
    /// state the agent loop must apply before the retry attempt.
    fn prepare_model_fallback(&self, _failure: &AgentError) -> Option<AgentLlmFallbackSwitch> {
        None
    }

    /// Move the client-local active candidate from `previous_identity` to the
    /// exact `target_identity` as one reversible transaction step.
    ///
    /// The core loop invokes this only after every target-dependent operation
    /// (including target-provider schema compilation) has been prevalidated,
    /// but before auth/session/machine state is committed. Implementations must
    /// either perform the exact switch or return an error. The default fails
    /// closed so a custom client cannot propose a fallback while silently
    /// continuing to issue requests through its old provider client.
    ///
    /// Core verifies [`AgentLlmClient::active_model_fallback_identity`] after
    /// the call and invokes this method in reverse if a later transaction step
    /// fails.
    fn commit_model_fallback(
        &self,
        _previous_identity: &crate::SessionLlmIdentity,
        target_identity: &crate::SessionLlmIdentity,
    ) -> Result<(), AgentError> {
        Err(AgentError::ConfigError(format!(
            "LLM client proposed fallback target '{}:{}' without an activation implementation",
            target_identity.provider.as_str(),
            target_identity.model
        )))
    }

    /// Exact identity of the client-local active fallback candidate.
    ///
    /// Fallback-capable clients must expose the full session identity,
    /// including auth binding and provider parameters. The default is absent,
    /// which makes fallback activation fail closed before canonical state is
    /// mutated.
    fn active_model_fallback_identity(&self) -> Option<crate::SessionLlmIdentity> {
        None
    }

    /// Compile an extraction schema against an inactive fallback target.
    ///
    /// This must delegate to the exact prebuilt target client without changing
    /// which client is active. Core calls it before auth, machine, visibility,
    /// session, or client activation state is mutated, then injects the
    /// compiled representation into the target provider request.
    fn compile_model_fallback_schema(
        &self,
        target_identity: &crate::SessionLlmIdentity,
        _output_schema: &OutputSchema,
    ) -> Result<CompiledSchema, AgentError> {
        Err(AgentError::ConfigError(format!(
            "LLM client cannot compile structured output for fallback target '{}:{}'",
            target_identity.provider.as_str(),
            target_identity.model
        )))
    }

    /// Reset per-call observation of user-visible streaming output.
    ///
    /// Adapters that emit display/reasoning deltas before returning the final
    /// stream result use this to let the retry loop distinguish a pre-stream
    /// failure from a post-partial-output failure. The default is no-op for
    /// clients that do not stream visible events outside the returned blocks.
    fn begin_stream_output_observation(&self) {}

    /// Whether the current LLM call has emitted user-visible streaming output.
    ///
    /// A `true` value suppresses model fallback for the failed call: retrying
    /// against a different model after users already saw partial output can
    /// produce duplicate assistant answers. Ordinary same-model retry policy is
    /// still governed by the generated turn recovery authority.
    fn stream_output_observed(&self) -> bool {
        false
    }

    /// Monotonic count of raw provider stream events observed by this client.
    ///
    /// This feeds the agent loop's stream-inactivity watchdog
    /// (`RetryPolicy::stream_inactivity_timeout`): the loop snapshots the
    /// count around each stream-event window and treats "no change" as a
    /// silent stream. Clients that consume a provider event stream should bump
    /// the count on every received event — including non-visible ones — so
    /// liveness is distinct from visible output
    /// ([`Self::stream_output_observed`]).
    ///
    /// `None` (the default) means this client does not report stream liveness
    /// and the watchdog is disabled for its calls; only the hard call/turn
    /// timeouts apply. This fails open on purpose: a non-streaming custom
    /// client would otherwise look permanently silent and be killed while
    /// healthy.
    fn stream_activity_count(&self) -> Option<u64> {
        None
    }

    /// Compile an output schema for this provider.
    ///
    /// Default implementation normalizes the schema without provider-specific lowering.
    /// Adapters override this to apply provider-specific transformations (e.g.,
    /// Anthropic adds `additionalProperties: false`, Gemini strips unsupported keywords).
    fn compile_schema(&self, output_schema: &OutputSchema) -> Result<CompiledSchema, SchemaError> {
        // Default passthrough: normalized clone, no provider-specific lowering
        Ok(CompiledSchema {
            schema: output_schema.schema.as_value().clone(),
            warnings: Vec::new(),
        })
    }
}

/// Hook for wrapping the final agent-facing LLM client.
///
/// Factories and runtimes apply this after provider/raw-client adaptation so
/// embedders can compose cross-cutting behavior without provider-specific
/// registry hooks.
pub type AgentLlmClientDecorator =
    Arc<dyn Fn(Arc<dyn AgentLlmClient>) -> Arc<dyn AgentLlmClient> + Send + Sync + 'static>;

/// One fallback target skipped while selecting a viable backup model.
#[derive(Debug, Clone)]
pub struct AgentLlmFallbackSkippedTarget {
    pub identity: crate::SessionLlmIdentity,
    pub reason: String,
}

/// Typed state produced when an agent-facing LLM client activates a fallback.
///
/// The client owns only prebuilt candidate selection. The agent loop owns
/// applying request policy, durable identity metadata, and tool visibility
/// before issuing the machine-authorized retry.
#[derive(Debug, Clone)]
pub struct AgentLlmFallbackSwitch {
    pub previous_identity: crate::SessionLlmIdentity,
    pub new_identity: crate::SessionLlmIdentity,
    pub request_policy: crate::SessionLlmRequestPolicy,
    /// Proposed effective-registry witness for the exact target provider/model.
    /// Core rejects foreign authority and freshly resolves all capability and
    /// token-limit facts through the agent's captured registry. The witness is
    /// required: unresolved fallback targets fail closed.
    pub target_profile: crate::ModelProfileWitness,
    pub skipped_targets: Vec<AgentLlmFallbackSkippedTarget>,
}

/// One-shot authorization for an exact sticky model-fallback activation.
///
/// There is deliberately no public constructor and the fields are private.
/// The constructor is owned by the `agent` module, so only the core agent loop
/// can mint this value after generated recovery acceptance and exact
/// effective-registry validation. A public
/// [`crate::handles::ModelRoutingHandle`] therefore cannot be driven directly
/// with a caller-minted or foreign-registry profile.
///
/// ```compile_fail
/// use meerkat_core::StickyModelFallbackActivationProof;
///
/// // Routing callers cannot fabricate an activation proof.
/// let _proof = StickyModelFallbackActivationProof::new();
/// ```
pub struct StickyModelFallbackActivationProof {
    previous_identity: crate::SessionLlmIdentity,
    target_identity: crate::SessionLlmIdentity,
    target_profile: crate::ModelProfileWitness,
    target_capability_base_filter: crate::ToolFilter,
    retry_attempt: u32,
}

impl StickyModelFallbackActivationProof {
    fn new(
        previous_identity: crate::SessionLlmIdentity,
        target_identity: crate::SessionLlmIdentity,
        target_profile: crate::ModelProfileWitness,
        retry_attempt: u32,
    ) -> Self {
        let target_capability_base_filter = crate::capability_base_filter_for_image_tool_results(
            target_profile.profile().image_tool_results,
        );
        Self {
            previous_identity,
            target_identity,
            target_profile,
            target_capability_base_filter,
            retry_attempt,
        }
    }

    /// Exact identity the generated recovery transition must still own.
    pub fn previous_identity(&self) -> &crate::SessionLlmIdentity {
        &self.previous_identity
    }

    /// Exact registry-resolved identity being activated.
    pub fn target_identity(&self) -> &crate::SessionLlmIdentity {
        &self.target_identity
    }

    /// Registry-owned target profile carried by this authorization.
    pub fn target_profile(&self) -> &crate::ModelProfileWitness {
        &self.target_profile
    }

    /// Registry-derived capability filter for the target model.
    pub fn target_capability_base_filter(&self) -> &crate::ToolFilter {
        &self.target_capability_base_filter
    }

    /// Machine-accepted retry attempt bound into this authorization.
    pub fn retry_attempt(&self) -> u32 {
        self.retry_attempt
    }
}

/// Result of streaming from the LLM
pub struct LlmStreamResult {
    blocks: Vec<AssistantBlock>,
    stop_reason: StopReason,
    usage: Usage,
}

impl LlmStreamResult {
    pub fn new(blocks: Vec<AssistantBlock>, stop_reason: StopReason, usage: Usage) -> Self {
        Self {
            blocks,
            stop_reason,
            usage,
        }
    }

    pub fn blocks(&self) -> &[AssistantBlock] {
        &self.blocks
    }
    pub fn stop_reason(&self) -> StopReason {
        self.stop_reason
    }
    pub fn usage(&self) -> &Usage {
        &self.usage
    }

    pub fn into_message(self) -> BlockAssistantMessage {
        BlockAssistantMessage::new(self.blocks, self.stop_reason)
    }

    pub fn into_parts(self) -> (Vec<AssistantBlock>, StopReason, Usage) {
        (self.blocks, self.stop_reason, self.usage)
    }
}

/// Snapshot of the core agent's live execution state.
///
/// When a runtime-backed turn-state handle is attached, this snapshots the
/// runtime-owned turn machine; otherwise it falls back to the in-process
/// standalone turn state used by core-only execution.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AgentExecutionSnapshot {
    pub loop_state: LoopState,
    pub turn_phase: TurnPhase,
    /// Machine-owned turn-terminality verdict.
    ///
    /// The `TurnTerminalityClassified.terminal` verdict emitted by the canonical
    /// MeerkatMachine `ClassifyTurnTerminality` input. Consumers mirror this bool
    /// and must not reclassify [`TurnPhase`] locally.
    pub turn_terminal: bool,
    pub active_run_id: Option<RunId>,
    pub terminal_run_id: Option<RunId>,
    pub primitive_kind: TurnPrimitiveKind,
    pub admitted_content_shape: Option<ContentShape>,
    pub vision_enabled: bool,
    pub image_tool_results_enabled: bool,
    pub tool_calls_pending: u32,
    pub pending_operation_ids: Option<Vec<OperationId>>,
    pub barrier_operation_ids: Vec<OperationId>,
    pub has_barrier_ops: bool,
    pub barrier_satisfied: bool,
    pub boundary_count: u32,
    pub cancel_after_boundary: bool,
    pub terminal_outcome: TurnTerminalOutcome,
    pub terminal_cause_kind: Option<TurnTerminalCauseKind>,
    pub extraction_attempts: u32,
    pub max_extraction_retries: u32,
    pub applied_cursor: CompletionSeq,
}

/// Result of polling for external tool updates.
///
/// Returned by [`AgentToolDispatcher::poll_external_updates`].
#[derive(Debug, Clone, Default)]
pub struct ExternalToolUpdate {
    /// Notices about completed background operations since last poll.
    pub notices: Vec<ExternalToolDelta>,
    /// Names of servers still connecting in the background.
    pub pending: Vec<String>,
}

/// Typed command requesting cancellation at the next turn boundary.
///
/// Carried over the cancel-after-boundary command channel from the surface
/// that authorized the request (e.g. `SessionService::cancel_after_boundary`)
/// to the agent loop, which observes it at the next boundary. The agent
/// resolves the request against its own live active run. The exact run witness
/// prevents a delayed request from an old executor attachment from cancelling
/// a successor run after same-session replacement.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CancelAfterBoundaryCommand {
    expected_run_id: RunId,
}

impl CancelAfterBoundaryCommand {
    /// Bind a cooperative-cancel command to one exact run incarnation.
    pub fn for_run(expected_run_id: RunId) -> Self {
        Self { expected_run_id }
    }

    /// Exact run incarnation this command is authorized to affect.
    pub fn expected_run_id(&self) -> &RunId {
        &self.expected_run_id
    }
}

/// Producer end of the cancel-after-boundary command channel.
///
/// Cloned and handed to the requesting surface via
/// [`Agent::cancel_after_boundary_handle`]; mirrors the cloneable-handle shape
/// of the session-side `interrupt_notify` so a surface can request boundary
/// cancellation without holding a reference to the agent.
pub type CancelAfterBoundarySender = tokio::sync::mpsc::UnboundedSender<CancelAfterBoundaryCommand>;

/// Typed context supplied by the agent loop when dispatching a tool call.
///
/// This is a dispatch-time projection of the already-admitted turn input. It
/// lets tool surfaces resolve typed turn-scoped references, such as a
/// `source=current_turn, index=0` image ref, without writing surface-local
/// metadata into canonical transcript history.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ToolDispatchContext {
    current_turn: Option<CurrentTurnContent>,
    turn_metadata: BTreeMap<String, serde_json::Value>,
    origin_session_id: Option<crate::types::SessionId>,
    interaction_lineage_id: Option<crate::interaction::InteractionId>,
    streaming: Option<crate::ToolStreamingDispatchContext>,
}

/// Dispatch-context key carrying the current durable objective id.
pub const TOOL_DISPATCH_OBJECTIVE_ID_KEY: &str = "meerkat.objective_id";

impl ToolDispatchContext {
    pub fn from_current_turn_input(input: &crate::types::ContentInput) -> Self {
        let blocks = match input {
            crate::types::ContentInput::Text(_) => None,
            crate::types::ContentInput::Blocks(blocks) => Some(blocks.clone()),
        };
        Self {
            current_turn: blocks.map(CurrentTurnContent::new),
            turn_metadata: BTreeMap::new(),
            origin_session_id: None,
            interaction_lineage_id: None,
            streaming: None,
        }
    }

    /// Project the typed run input into a dispatch context. The
    /// pending-tool-results continuation carries no caller content, so it
    /// projects to an empty context rather than a fabricated empty prompt.
    pub fn from_run_input(input: &crate::types::RunInput) -> Self {
        match input {
            crate::types::RunInput::Content { content } => Self::from_current_turn_input(content),
            crate::types::RunInput::PendingToolResults => Self::default(),
        }
    }

    #[must_use]
    pub fn with_turn_metadata(mut self, metadata: BTreeMap<String, serde_json::Value>) -> Self {
        self.turn_metadata = metadata;
        self
    }

    pub fn turn_metadata(&self, key: &str) -> Option<&serde_json::Value> {
        self.turn_metadata.get(key)
    }

    pub fn current_turn(&self) -> Option<&CurrentTurnContent> {
        self.current_turn.as_ref()
    }

    /// Bind the runtime-owned durable identity of the turn being dispatched.
    ///
    /// Standalone callers may leave this absent. Durable execution owners must
    /// fail closed rather than minting replacement identity at dispatch time.
    #[must_use]
    pub fn with_runtime_identity(
        mut self,
        origin_session_id: crate::types::SessionId,
        interaction_lineage_id: Option<crate::interaction::InteractionId>,
    ) -> Self {
        self.origin_session_id = Some(origin_session_id);
        self.interaction_lineage_id = interaction_lineage_id;
        self
    }

    pub fn origin_session_id(&self) -> Option<&crate::types::SessionId> {
        self.origin_session_id.as_ref()
    }

    pub const fn interaction_lineage_id(&self) -> Option<crate::interaction::InteractionId> {
        self.interaction_lineage_id
    }

    /// Streaming-only liveness surface minted by the canonical supervisor.
    ///
    /// Fast and detached dispatch contexts carry no streaming surface. A tool
    /// that declared `Streaming` must fail closed if this is absent rather than
    /// fabricating a progress sink or cancellation authority.
    pub const fn streaming(&self) -> Option<&crate::ToolStreamingDispatchContext> {
        self.streaming.as_ref()
    }

    pub(crate) fn with_streaming(mut self, streaming: crate::ToolStreamingDispatchContext) -> Self {
        self.streaming = Some(streaming);
        self
    }

    pub fn current_turn_image(
        &self,
        image_ref: CurrentTurnImageRef,
    ) -> Option<&crate::types::ContentBlock> {
        self.current_turn
            .as_ref()
            .and_then(|current_turn| current_turn.image(image_ref))
    }
}

/// Typed reference to an image in the current admitted turn.
///
/// The wrapped index addresses the turn's *filtered image stream*, not the
/// raw block list: ref `N` designates the `(N + 1)`-th image block of the
/// current turn, skipping non-image blocks (so ref `0` is the first image
/// even when text blocks precede it).
///
/// The field is private. In-process code mints refs only via
/// [`CurrentTurnContent::image_ref`], which returns a ref only when the
/// referenced image exists. Wire ingress (e.g. the comms `image_ref` tool
/// input) deserializes a bare JSON integer directly into this type via
/// `#[serde(transparent)]` — that is the sanctioned parse-at-ingress path,
/// and resolution through [`CurrentTurnContent::image`] still validates
/// existence.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(transparent)]
pub struct CurrentTurnImageRef(usize);

impl std::fmt::Display for CurrentTurnImageRef {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        std::fmt::Display::fmt(&self.0, f)
    }
}

/// Multimodal content from the currently admitted turn.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CurrentTurnContent {
    blocks: Vec<crate::types::ContentBlock>,
}

impl CurrentTurnContent {
    pub fn new(blocks: Vec<crate::types::ContentBlock>) -> Self {
        Self { blocks }
    }

    pub fn blocks(&self) -> &[crate::types::ContentBlock] {
        &self.blocks
    }

    /// Mint a typed reference to the `n`-th image of this turn's filtered
    /// image stream. Returns `Some` only when that image exists, so every
    /// in-process [`CurrentTurnImageRef`] is resolvable at mint time.
    pub fn image_ref(&self, n: usize) -> Option<CurrentTurnImageRef> {
        self.images().nth(n).map(|_| CurrentTurnImageRef(n))
    }

    pub fn image(&self, image_ref: CurrentTurnImageRef) -> Option<&crate::types::ContentBlock> {
        self.images().nth(image_ref.0)
    }

    fn images(&self) -> impl Iterator<Item = &crate::types::ContentBlock> {
        self.blocks
            .iter()
            .filter(|block| matches!(block, crate::types::ContentBlock::Image { .. }))
    }
}

/// Completion notice for a detached background operation, projected from
/// canonical ops-lifecycle terminal state plus dispatcher-owned display metadata.
///
/// This is a rebuildable projection (INV-003), not authoritative state.
/// Terminal class and timing come from `OperationLifecycleSnapshot` (INV-001).
/// Shell-projected detail is supplementary display only (INV-002).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DetachedOpCompletion {
    /// App-facing job identifier (the control noun for surfaces).
    pub job_id: String,
    /// Operation kind from canonical ops-lifecycle.
    pub kind: OperationKind,
    /// Terminal status from canonical ops-lifecycle.
    pub status: OperationStatus,
    /// Terminal outcome from canonical ops-lifecycle.
    pub terminal_outcome: Option<OperationTerminalOutcome>,
    /// Canonical display label from ops-lifecycle snapshot.
    pub display_name: String,
    /// Dispatcher-projected summary (exit code, output tail). Display only.
    pub detail: String,
    /// Monotonic elapsed millis from ops-lifecycle snapshot.
    pub elapsed_ms: Option<u64>,
}

/// Dispatcher binding capabilities — what optional bindings this dispatcher supports.
///
/// Returned by [`AgentToolDispatcher::capabilities`]. Replaces individual
/// `supports_*` boolean methods with a single structured query.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct DispatcherCapabilities {
    /// Whether `bind_ops_lifecycle` is implemented.
    pub ops_lifecycle: bool,
}

/// Result of a dispatcher binding operation.
///
/// Distinguishes "binding was applied" from "binding was skipped" so callers
/// can decide whether to wire downstream side effects (e.g. bridge tasks).
///
/// **Semantics (decision 11 — supported/best-effort/rejected):**
/// - `Ok(Bound(d))` = **supported** — binding succeeded, side effects should be wired
/// - `Ok(Skipped(d))` = **best-effort** — inner shared or incompatible, dispatcher unchanged
/// - `Err(SharedOwnership)` = **rejected** — outer wrapper is shared, caught by factory pre-check
/// - `Err(Unsupported)` = **rejected** — type doesn't support this binding, caught by `capabilities()`
pub enum BindOutcome {
    /// Binding was applied. The dispatcher was rebound.
    Bound(Arc<dyn AgentToolDispatcher>),
    /// Binding was skipped — inner dispatcher was shared or unsupported.
    /// The returned dispatcher is unchanged but safe to use.
    Skipped(Arc<dyn AgentToolDispatcher>),
}

impl BindOutcome {
    /// Extract the dispatcher, regardless of bind status.
    pub fn into_dispatcher(self) -> Arc<dyn AgentToolDispatcher> {
        match self {
            Self::Bound(d) | Self::Skipped(d) => d,
        }
    }

    /// Whether the binding was actually applied.
    pub fn was_bound(&self) -> bool {
        matches!(self, Self::Bound(_))
    }
}

/// Trait for tool dispatchers
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait AgentToolDispatcher: Send + Sync {
    /// Get available tool definitions
    fn tools(&self) -> Arc<[Arc<ToolDef>]>;

    /// Query exact catalog support for this dispatcher.
    ///
    /// Dispatchers report `exact_catalog=true` only when `tool_catalog()`
    /// returns the exact precedence-resolved winner registry for the plane
    /// they own. Wrappers that cannot prove exactness must leave this false.
    fn tool_catalog_capabilities(&self) -> ToolCatalogCapabilities {
        ToolCatalogCapabilities::default()
    }

    /// Return the precedence-resolved tool catalog for this dispatcher.
    ///
    /// The default implementation mirrors `tools()` as a visible-only inline
    /// catalog. Callers must gate any deferred-catalog behavior on
    /// `tool_catalog_capabilities().exact_catalog`.
    fn tool_catalog(&self) -> Arc<[ToolCatalogEntry]> {
        self.tools()
            .iter()
            .map(|tool| ToolCatalogEntry::session_inline(Arc::clone(tool), true))
            .collect::<Vec<_>>()
            .into()
    }

    /// Live generation for one logical tool binding.
    ///
    /// Static dispatchers keep the default zero epoch. Mutable authorities
    /// must override this and advance it for every replacement, including an
    /// identical-metadata A→B replacement.
    fn execution_binding_epoch(&self, _tool_name: &str) -> u64 {
        0
    }

    /// Snapshot the current live logical binding advertised for `tool_name`.
    fn execution_binding_fingerprint(
        &self,
        tool_name: &str,
    ) -> Result<crate::EphemeralToolBindingFingerprint, crate::ToolExecutionResolutionError> {
        let catalog = self.tool_catalog();
        let entry = catalog
            .iter()
            .find(|entry| entry.tool.name == tool_name)
            .ok_or_else(|| crate::ToolExecutionResolutionError::NotFound {
                tool_name: tool_name.to_string(),
            })?;
        Ok(crate::ephemeral_tool_catalog_binding_fingerprint(entry)
            .with_live_authority(0, self.execution_binding_epoch(tool_name)))
    }

    /// Resolve the exact execution class and deadline chain before dispatch.
    ///
    /// The default uses this dispatcher's effective catalog, so wrappers that
    /// filter or select winners apply the same decision to declaration and
    /// resolution. Hybrid tools may override this method to inspect typed
    /// arguments while preserving the catalog contract as the upper bound.
    fn resolve_execution_plan(
        &self,
        call: ToolCallView<'_>,
        _dispatch_context: &ToolDispatchContext,
        resolution_context: &crate::ToolExecutionResolutionContext,
    ) -> Result<crate::ResolvedToolExecutionPlan, crate::ToolExecutionResolutionError> {
        let catalog = self.tool_catalog();
        let entry = catalog
            .iter()
            .find(|entry| entry.tool.name == call.name)
            .ok_or_else(|| crate::ToolExecutionResolutionError::NotFound {
                tool_name: call.name.to_string(),
            })?;
        if let Some(reason) = entry.callability.unavailable_reason() {
            return Err(crate::ToolExecutionResolutionError::Unavailable {
                tool_name: call.name.to_string(),
                reason,
            });
        }
        entry
            .execution
            .resolve_default(resolution_context.deadlines().clone())
            .map_err(crate::ToolExecutionResolutionError::from)
    }

    /// Validate a resolved plan against both the caller-owned deadline prefix
    /// and this dispatcher's live advertised catalog contract.
    ///
    /// This is the mandatory root seam after argument-sensitive resolution:
    /// hybrid tools may select any advertised mode, while an override cannot
    /// return a mode or mode-derived facet absent from the effective catalog.
    fn validate_resolved_execution_plan(
        &self,
        call: ToolCallView<'_>,
        resolution_context: &crate::ToolExecutionResolutionContext,
        plan: &crate::ResolvedToolExecutionPlan,
    ) -> Result<(), crate::ToolExecutionResolutionError> {
        resolution_context.validate_resolved_plan(plan)?;
        let catalog = self.tool_catalog();
        let entry = catalog
            .iter()
            .find(|entry| entry.tool.name == call.name)
            .ok_or_else(|| crate::ToolExecutionResolutionError::NotFound {
                tool_name: call.name.to_string(),
            })?;
        if let Some(reason) = entry.callability.unavailable_reason() {
            return Err(crate::ToolExecutionResolutionError::Unavailable {
                tool_name: call.name.to_string(),
                reason,
            });
        }
        entry
            .execution
            .validate_resolved_plan(plan)
            .map_err(crate::ToolExecutionResolutionError::from)
    }

    /// Return non-draining pending source names for exact-catalog discovery.
    ///
    /// Pending sources are catalog-level discovery metadata rather than
    /// provider-visible tools. The default implementation reports none.
    fn pending_catalog_sources(&self) -> Arc<[String]> {
        Arc::from([])
    }

    /// Execute a tool call, returning the transcript result and any async operations.
    ///
    /// The `ToolDispatchOutcome` separates transcript data (`result`) from
    /// execution metadata (`async_ops`). Most tools return no async ops;
    /// use `ToolDispatchOutcome::from(result)` for synchronous tools.
    async fn dispatch(
        &self,
        call: ToolCallView<'_>,
    ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError>;

    /// Execute a tool call with the current turn's typed dispatch context.
    ///
    /// Most tools do not need turn-local context and inherit the plain
    /// `dispatch` behavior. Context-sensitive surfaces override this method
    /// rather than reaching into session history or prompt text.
    async fn dispatch_with_context(
        &self,
        call: ToolCallView<'_>,
        _context: &ToolDispatchContext,
    ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
        self.dispatch(call).await
    }

    /// Execute a previously resolved plan without re-selecting its mode.
    ///
    /// The default is deliberately a one-way lowering for Fast calls only.
    /// Streaming and Detached require an explicit mode owner; silently sending
    /// either through ordinary dispatch would erase their liveness, output,
    /// restart, and idempotency contracts.
    async fn dispatch_resolved_with_context(
        &self,
        call: ToolCallView<'_>,
        context: &ToolDispatchContext,
        plan: &crate::ResolvedToolExecutionPlan,
    ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
        match plan.mode() {
            crate::ToolExecutionMode::Fast => self.dispatch_with_context(call, context).await,
            crate::ToolExecutionMode::Streaming | crate::ToolExecutionMode::Detached => {
                Err(crate::error::ToolError::unavailable(
                    call.name,
                    crate::ToolUnavailableReason::ExecutionModeOwnerUnavailable,
                ))
            }
        }
    }

    /// Poll for external tool updates from background operations (e.g. async MCP loading).
    ///
    /// The default implementation returns an empty update. Implementations that
    /// support background tool loading (like `McpRouterAdapter`) override this
    /// to drain completed results and report pending servers.
    async fn poll_external_updates(&self) -> ExternalToolUpdate {
        ExternalToolUpdate::default()
    }

    /// Snapshot the live external tool-surface machine state, if supported.
    ///
    /// This is a hidden diagnostic surface for MeerkatMachine mapping work.
    /// Dispatchers that do not own dynamic external tool mutation should
    /// return `None`.
    fn external_tool_surface_snapshot(&self) -> Option<crate::ExternalToolSurfaceSnapshot> {
        None
    }

    /// Query which optional bindings this dispatcher supports.
    fn capabilities(&self) -> DispatcherCapabilities {
        DispatcherCapabilities::default()
    }

    /// Bind a session-canonical ops registry into this dispatcher.
    ///
    /// Dispatchers that emit session-visible `AsyncOpRef`s must route those
    /// operation IDs into the bound registry. Under the identity-first Mob
    /// regime the owner binding passed here is the canonical bridge session
    /// binding, even though many compatibility surfaces still spell it
    /// `session_id`. Default returns Unsupported.
    fn bind_ops_lifecycle(
        self: Arc<Self>,
        _registry: Arc<dyn crate::ops_lifecycle::OpsLifecycleRegistry>,
        _owner_bridge_session_id: crate::types::SessionId,
    ) -> Result<BindOutcome, OpsLifecycleBindError> {
        Err(OpsLifecycleBindError::Unsupported)
    }

    /// Return the completion enrichment provider, if available.
    ///
    /// Dispatchers with shell job management return a provider that maps
    /// operation IDs to display details (job ID, status detail string).
    fn completion_enrichment(
        &self,
    ) -> Option<Arc<dyn crate::completion_feed::CompletionEnrichmentProvider>> {
        None
    }

    /// Bind a session-scoped MCP server lifecycle handle (Phase 5G / T5g).
    ///
    /// Dispatchers that manage per-server MCP handshake lifecycle (like
    /// `McpRouterAdapter`) use the handle to mirror connection state into
    /// the session's MeerkatMachine DSL. The default implementation is a
    /// no-op for dispatchers that have no MCP handshake to route.
    fn bind_mcp_server_lifecycle_handle(
        &self,
        _handle: Arc<dyn crate::handles::McpServerLifecycleHandle>,
    ) {
    }

    /// Bind the session-canonical external tool-surface handle.
    ///
    /// MCP dispatchers use this to route add/remove/reload/call lifecycle
    /// semantics through the session's MeerkatMachine DSL instead of their
    /// standalone compatibility projection. The default implementation is a
    /// no-op for dispatchers that do not own dynamic external tool surfaces.
    fn bind_external_tool_surface_handle(
        &self,
        _handle: Arc<dyn crate::handles::ExternalToolSurfaceHandle>,
    ) {
    }
}

/// Resolve a plan against the exact live root dispatcher allocation.
///
/// The returned plan retains an ephemeral `Arc` lease to that allocation.
/// This makes reconstruction and allocator address reuse unforgeable without
/// serializing process-local authority or conflating it with durable job
/// fencing.
pub fn resolve_tool_execution_plan_fenced<T: AgentToolDispatcher + ?Sized + 'static>(
    dispatcher: &Arc<T>,
    call: ToolCallView<'_>,
    dispatch_context: &ToolDispatchContext,
    resolution_context: &crate::ToolExecutionResolutionContext,
) -> Result<crate::ResolvedToolExecutionPlan, crate::ToolExecutionResolutionError> {
    let before = dispatcher.execution_binding_fingerprint(call.name)?;
    let plan = dispatcher.resolve_execution_plan(call, dispatch_context, resolution_context)?;
    if dispatcher.execution_binding_fingerprint(call.name)? != before {
        return Err(crate::ToolExecutionResolutionError::Unavailable {
            tool_name: call.name.to_string(),
            reason: crate::ToolUnavailableReason::ExecutionOwnerChanged,
        });
    }
    let witness = crate::ToolExecutionOwnerWitness::new("root-dispatcher", call.name, before)
        .map_err(crate::ToolExecutionResolutionError::from)?;
    plan.with_owner_witness(witness)?
        .bind_root_dispatch(Arc::clone(dispatcher), call)
}

/// Dispatch a plan only through the exact root allocation and exact canonical
/// call identity used during resolution.
pub async fn dispatch_tool_execution_plan_fenced<T: AgentToolDispatcher + ?Sized + 'static>(
    dispatcher: &Arc<T>,
    call: ToolCallView<'_>,
    context: &ToolDispatchContext,
    plan: &crate::ResolvedToolExecutionPlan,
) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
    plan.validate_root_dispatch(dispatcher, call)?;
    let witness = plan.owner_witness("root-dispatcher").ok_or_else(|| {
        crate::error::ToolError::unavailable(
            call.name,
            crate::ToolUnavailableReason::ExecutionOwnerChanged,
        )
    })?;
    if witness.binding_fingerprint() != &dispatcher.execution_binding_fingerprint(call.name)? {
        return Err(crate::error::ToolError::unavailable(
            call.name,
            crate::ToolUnavailableReason::ExecutionOwnerChanged,
        ));
    }
    match plan.kind() {
        crate::ResolvedExecutionKind::Streaming(policy) => {
            let absolute_timeout = plan
                .deadlines()
                .effective_timeout()
                .unwrap_or_else(|| policy.absolute_timeout());
            crate::streaming_tool::supervise_streaming_tool(
                call.name,
                policy.inactivity_timeout(),
                absolute_timeout,
                |streaming| {
                    let streaming_context = context.clone().with_streaming(streaming);
                    async move {
                        dispatcher
                            .dispatch_resolved_with_context(call, &streaming_context, plan)
                            .await
                    }
                },
            )
            .await
        }
        crate::ResolvedExecutionKind::Fast | crate::ResolvedExecutionKind::Detached(_) => {
            dispatcher
                .dispatch_resolved_with_context(call, context, plan)
                .await
        }
    }
}

/// Compute whether the current exact catalog should stay inline or switch to deferred mode.
pub fn select_tool_catalog_mode<T>(dispatcher: &T) -> ToolCatalogMode
where
    T: AgentToolDispatcher + ?Sized,
{
    let capabilities = dispatcher.tool_catalog_capabilities();
    if !capabilities.exact_catalog {
        return ToolCatalogMode::Inline;
    }
    let pending_sources = dispatcher.pending_catalog_sources();
    let catalog = dispatcher.tool_catalog();
    select_catalog_mode_from_snapshot(
        capabilities.exact_catalog,
        catalog.as_ref(),
        pending_sources.as_ref(),
    )
}

/// Compute whether the catalog control plane should be composed for this
/// dispatcher, even if the current adaptive snapshot remains inline.
pub fn should_compose_tool_catalog_control_plane<T>(dispatcher: &T) -> bool
where
    T: AgentToolDispatcher + ?Sized,
{
    let capabilities = dispatcher.tool_catalog_capabilities();
    if !capabilities.exact_catalog {
        return false;
    }
    if capabilities.may_require_catalog_control_plane {
        return true;
    }

    let pending_sources = dispatcher.pending_catalog_sources();
    if !pending_sources.is_empty() {
        return true;
    }

    let catalog = dispatcher.tool_catalog();
    deferred_session_entry_count(catalog.as_ref()) > 0
}

/// Error from [`AgentToolDispatcher::bind_ops_lifecycle`].
#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
pub enum OpsLifecycleBindError {
    #[error("ops lifecycle binding is unsupported")]
    Unsupported,
    #[error("dispatcher has shared ownership and cannot be rebound")]
    SharedOwnership,
}

/// A tool dispatcher that filters tools based on a policy
///
/// Legacy tool lists are filtered once at construction time based on the
/// allowed_tools list. Exact-catalog dispatchers keep catalog callability live.
/// The inner dispatcher is used for actual dispatch, but only allowed tools are
/// exposed via tools() and dispatch() returns AccessDenied for filtered tools.
pub struct FilteredToolDispatcher<T: AgentToolDispatcher + ?Sized> {
    inner: Arc<T>,
    allowed_tools: ToolNameSet,
    /// Pre-computed filtered tool list for non-exact dispatchers.
    filtered_tools: Arc<[Arc<ToolDef>]>,
}

impl<T: AgentToolDispatcher + ?Sized> FilteredToolDispatcher<T> {
    pub fn new<I, N>(inner: Arc<T>, allowed_tools: I) -> Self
    where
        I: IntoIterator<Item = N>,
        N: Into<ToolName>,
    {
        let allowed_set: ToolNameSet = allowed_tools
            .into_iter()
            .map(Into::into)
            .collect::<ToolNameSet>();

        let filtered: Vec<Arc<ToolDef>> = if inner.tool_catalog_capabilities().exact_catalog {
            inner
                .tool_catalog()
                .iter()
                .filter(|entry| entry.currently_callable())
                .map(|entry| Arc::clone(&entry.tool))
                .filter(|t| allowed_set.contains(t.name.as_str()))
                .collect()
        } else {
            inner
                .tools()
                .iter()
                .filter(|t| allowed_set.contains(t.name.as_str()))
                .map(Arc::clone)
                .collect()
        };

        Self {
            inner,
            allowed_tools: allowed_set,
            filtered_tools: filtered.into(),
        }
    }
}

#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
impl<T: AgentToolDispatcher + ?Sized + 'static> AgentToolDispatcher for FilteredToolDispatcher<T> {
    fn tools(&self) -> Arc<[Arc<ToolDef>]> {
        if self.inner.tool_catalog_capabilities().exact_catalog {
            return self
                .inner
                .tool_catalog()
                .iter()
                .filter(|entry| entry.currently_callable())
                .map(|entry| Arc::clone(&entry.tool))
                .filter(|tool| self.allowed_tools.contains(tool.name.as_str()))
                .collect::<Vec<_>>()
                .into();
        }
        Arc::clone(&self.filtered_tools)
    }

    async fn dispatch(
        &self,
        call: ToolCallView<'_>,
    ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
        self.dispatch_with_context(call, &ToolDispatchContext::default())
            .await
    }

    async fn dispatch_with_context(
        &self,
        call: ToolCallView<'_>,
        context: &ToolDispatchContext,
    ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
        if !self.allowed_tools.contains(call.name) {
            let inner_knows_tool = if self.inner.tool_catalog_capabilities().exact_catalog {
                self.inner
                    .tool_catalog()
                    .iter()
                    .any(|entry| entry.tool.name == call.name)
            } else {
                self.inner.tools().iter().any(|tool| tool.name == call.name)
            };
            if !inner_knows_tool {
                return Err(crate::error::ToolError::not_found(call.name));
            }
            return Err(crate::error::ToolError::access_denied(call.name));
        }
        self.inner.dispatch_with_context(call, context).await
    }

    async fn dispatch_resolved_with_context(
        &self,
        call: ToolCallView<'_>,
        context: &ToolDispatchContext,
        plan: &crate::ResolvedToolExecutionPlan,
    ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
        if !self.allowed_tools.contains(call.name) {
            let inner_knows_tool = if self.inner.tool_catalog_capabilities().exact_catalog {
                self.inner
                    .tool_catalog()
                    .iter()
                    .any(|entry| entry.tool.name == call.name)
            } else {
                self.inner.tools().iter().any(|tool| tool.name == call.name)
            };
            if !inner_knows_tool {
                return Err(crate::error::ToolError::not_found(call.name));
            }
            return Err(crate::error::ToolError::access_denied(call.name));
        }
        self.inner
            .dispatch_resolved_with_context(call, context, plan)
            .await
    }

    fn tool_catalog_capabilities(&self) -> ToolCatalogCapabilities {
        self.inner.tool_catalog_capabilities()
    }

    fn tool_catalog(&self) -> Arc<[ToolCatalogEntry]> {
        if !self.inner.tool_catalog_capabilities().exact_catalog {
            return self
                .tools()
                .iter()
                .map(|tool| ToolCatalogEntry::session_inline(Arc::clone(tool), true))
                .collect::<Vec<_>>()
                .into();
        }
        self.inner
            .tool_catalog()
            .iter()
            .filter(|entry| self.allowed_tools.contains(entry.tool.name.as_str()))
            .cloned()
            .collect::<Vec<_>>()
            .into()
    }

    fn execution_binding_fingerprint(
        &self,
        tool_name: &str,
    ) -> Result<crate::EphemeralToolBindingFingerprint, crate::ToolExecutionResolutionError> {
        let catalog = self.tool_catalog();
        let entry = catalog
            .iter()
            .find(|entry| entry.tool.name == tool_name)
            .ok_or_else(|| crate::ToolExecutionResolutionError::NotFound {
                tool_name: tool_name.to_string(),
            })?;
        let child = self.inner.execution_binding_fingerprint(tool_name)?;
        Ok(crate::ephemeral_tool_catalog_binding_fingerprint(entry)
            .with_live_authority(0, 0)
            .with_dependency(&child))
    }

    fn resolve_execution_plan(
        &self,
        call: ToolCallView<'_>,
        dispatch_context: &ToolDispatchContext,
        resolution_context: &crate::ToolExecutionResolutionContext,
    ) -> Result<crate::ResolvedToolExecutionPlan, crate::ToolExecutionResolutionError> {
        if !self.allowed_tools.contains(call.name) {
            let inner_knows_tool = if self.inner.tool_catalog_capabilities().exact_catalog {
                self.inner
                    .tool_catalog()
                    .iter()
                    .any(|entry| entry.tool.name == call.name)
            } else {
                self.inner.tools().iter().any(|tool| tool.name == call.name)
            };
            return Err(if inner_knows_tool {
                crate::ToolExecutionResolutionError::AccessDenied {
                    tool_name: call.name.to_string(),
                }
            } else {
                crate::ToolExecutionResolutionError::NotFound {
                    tool_name: call.name.to_string(),
                }
            });
        }

        let catalog = self.tool_catalog();
        let entry = catalog
            .iter()
            .find(|entry| entry.tool.name == call.name)
            .ok_or_else(|| crate::ToolExecutionResolutionError::NotFound {
                tool_name: call.name.to_string(),
            })?;
        if let Some(reason) = entry.callability.unavailable_reason() {
            return Err(crate::ToolExecutionResolutionError::Unavailable {
                tool_name: call.name.to_string(),
                reason,
            });
        }

        self.inner
            .resolve_execution_plan(call, dispatch_context, resolution_context)
    }

    fn pending_catalog_sources(&self) -> Arc<[String]> {
        self.inner.pending_catalog_sources()
    }

    async fn poll_external_updates(&self) -> ExternalToolUpdate {
        self.inner.poll_external_updates().await
    }

    fn external_tool_surface_snapshot(&self) -> Option<crate::ExternalToolSurfaceSnapshot> {
        self.inner.external_tool_surface_snapshot()
    }

    fn capabilities(&self) -> DispatcherCapabilities {
        self.inner.capabilities()
    }

    fn bind_ops_lifecycle(
        self: Arc<Self>,
        registry: Arc<dyn crate::ops_lifecycle::OpsLifecycleRegistry>,
        owner_bridge_session_id: crate::types::SessionId,
    ) -> Result<BindOutcome, OpsLifecycleBindError> {
        let owned = Arc::try_unwrap(self).map_err(|_| OpsLifecycleBindError::SharedOwnership)?;
        if Arc::strong_count(&owned.inner) == 1 {
            let outcome = owned
                .inner
                .bind_ops_lifecycle(registry, owner_bridge_session_id)?;
            let bound = outcome.was_bound();
            let d = outcome.into_dispatcher();
            let allowed_tools = owned.allowed_tools.into_iter().collect::<Vec<_>>();
            Ok(if bound {
                BindOutcome::Bound(Arc::new(FilteredToolDispatcher::new(d, allowed_tools)))
            } else {
                BindOutcome::Skipped(Arc::new(FilteredToolDispatcher::new(d, allowed_tools)))
            })
        } else {
            Ok(BindOutcome::Skipped(Arc::new(FilteredToolDispatcher {
                inner: owned.inner,
                allowed_tools: owned.allowed_tools,
                filtered_tools: owned.filtered_tools,
            })))
        }
    }

    fn completion_enrichment(
        &self,
    ) -> Option<Arc<dyn crate::completion_feed::CompletionEnrichmentProvider>> {
        self.inner.completion_enrichment()
    }

    fn bind_mcp_server_lifecycle_handle(
        &self,
        handle: Arc<dyn crate::handles::McpServerLifecycleHandle>,
    ) {
        self.inner.bind_mcp_server_lifecycle_handle(handle);
    }

    fn bind_external_tool_surface_handle(
        &self,
        handle: Arc<dyn crate::handles::ExternalToolSurfaceHandle>,
    ) {
        self.inner.bind_external_tool_surface_handle(handle);
    }
}

/// Trait for session stores
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait AgentSessionStore: Send + Sync {
    async fn save(&self, session: &Session) -> Result<(), AgentError>;
    async fn load(&self, id: &str) -> Result<Option<Session>, AgentError>;
}

/// Runtime policy for inlining peer lifecycle updates into session context.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InlinePeerNotificationPolicy {
    /// Always inline batched peer lifecycle updates.
    Always,
    /// Never inline batched peer lifecycle updates.
    Never,
    /// Inline only when post-drain peer count is at or below this threshold.
    AtMost(usize),
}

/// Default inline threshold when no explicit value is configured.
pub const DEFAULT_MAX_INLINE_PEER_NOTIFICATIONS: usize = 50;

impl InlinePeerNotificationPolicy {
    /// Resolve policy from transport/build-layer config representation.
    pub fn try_from_raw(raw: Option<i32>) -> Result<Self, i32> {
        match raw {
            None => Ok(Self::AtMost(DEFAULT_MAX_INLINE_PEER_NOTIFICATIONS)),
            Some(-1) => Ok(Self::Always),
            Some(0) => Ok(Self::Never),
            Some(v) if v > 0 => Ok(Self::AtMost(v as usize)),
            Some(v) => Err(v),
        }
    }
}

/// Error returned when a comms runtime capability is not available.
#[derive(Debug, thiserror::Error)]
pub enum CommsCapabilityError {
    /// The runtime does not support this capability.
    #[error("comms capability not supported: {0}")]
    Unsupported(String),
}

/// Trait for comms runtime that can be used with the agent
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait CommsRuntime: Send + Sync {
    /// Canonical runtime routing identity for this peer, if available.
    ///
    /// `PeerId` is the UUID-shaped routing key used by peer directories and
    /// trust stores. Implementations that only have the legacy string carrier
    /// may return a parsed UUID-shaped `public_key()` value; implementations
    /// with Ed25519 public keys should override this and return the pubkey-
    /// derived canonical [`PeerId`].
    fn peer_id(&self) -> Option<PeerId> {
        self.public_key()
            .as_deref()
            .and_then(|public_key| PeerId::parse(public_key).ok())
    }

    /// Runtime-local transport/auth public key, if available.
    ///
    /// Returns an Ed25519 public key string in `ed25519:<base64>` format.
    /// This is not the canonical routing [`PeerId`]; use [`Self::peer_id`]
    /// for roster/projection identity and peer-directory lookups.
    fn public_key(&self) -> Option<String> {
        None
    }

    /// Runtime-local Ed25519 public key bytes, if available.
    ///
    /// This is the typed form of [`Self::public_key`]. Trust installation
    /// paths that need to verify `PeerId`/pubkey consistency should prefer
    /// this method over reparsing the string carrier.
    fn public_key_bytes(&self) -> Option<[u8; 32]> {
        None
    }

    /// Runtime-local canonical comms routing name, if available.
    ///
    /// This is the peer name used in trusted-peer descriptors and peer
    /// directories. It is separate from the advertised transport address so
    /// callers do not recover identity by parsing transport strings.
    fn comms_name(&self) -> Option<String> {
        None
    }

    /// Runtime-local advertised comms address, if available.
    ///
    /// This is the canonical address the runtime expects peers to use when
    /// constructing a [`TrustedPeerDescriptor`]. Implementations that do not
    /// expose a stable advertised address can return `None`.
    fn advertised_address(&self) -> Option<String> {
        None
    }

    /// Runtime-local bootstrap proof for the initial supervisor bind, if
    /// available.
    fn bridge_bootstrap_token(&self) -> Option<String> {
        None
    }

    /// Apply a comms trust projection mutation authorized by generated
    /// machine/composition authority.
    ///
    /// This is the only mutable trust-store seam.
    async fn apply_trust_mutation(
        &self,
        _mutation: CommsTrustMutation,
    ) -> Result<CommsTrustMutationResult, SendError> {
        Err(SendError::Unsupported(
            "apply_trust_mutation not supported for this CommsRuntime".to_string(),
        ))
    }

    /// Bind this target runtime to the generated MobMachine owner token whose
    /// trust handoffs may mutate mob-owned trust rows.
    ///
    /// Mob runtimes call this before submitting a generated mob trust mutation.
    /// Implementations must fail closed when they cannot remember and compare
    /// the owner token during [`Self::apply_trust_mutation`].
    async fn install_generated_mob_trust_owner(
        &self,
        _owner: Arc<dyn std::any::Any + Send + Sync>,
    ) -> Result<(), SendError> {
        Err(SendError::Unsupported(
            "generated mob trust owner binding not supported for this CommsRuntime".to_string(),
        ))
    }

    /// Read-only preflight for binding this target runtime to a recovered
    /// MobMachine owner token.
    ///
    /// Resume uses this to validate every generated trust repair target before
    /// mutating any trust projection row. Implementations must not change the
    /// stored owner token here; [`Self::install_recovered_generated_mob_trust_owner`]
    /// performs the actual binding after the full batch has passed preflight.
    async fn validate_recovered_generated_mob_trust_owner(
        &self,
        _owner: Arc<dyn std::any::Any + Send + Sync>,
    ) -> Result<(), SendError> {
        Err(SendError::Unsupported(
            "recovered generated mob trust owner validation not supported for this CommsRuntime"
                .to_string(),
        ))
    }

    /// Rebind this target runtime to the owner token of a recovered
    /// MobMachine authority.
    ///
    /// Recovery reconstructs generated authority from persisted machine state,
    /// which gives it a fresh process-local owner token. Implementations may
    /// bind this owner only when no generated MobMachine owner is already
    /// installed, or when it is the same owner token. They must fail closed
    /// rather than replacing a different live owner through recovery plumbing.
    async fn install_recovered_generated_mob_trust_owner(
        &self,
        _owner: Arc<dyn std::any::Any + Send + Sync>,
    ) -> Result<(), SendError> {
        Err(SendError::Unsupported(
            "recovered generated mob trust owner binding not supported for this CommsRuntime"
                .to_string(),
        ))
    }

    /// Opaque host-acceptor registration material for reverse-lane demux
    /// composition (the runtime's identity pubkey, its ack-signing keypair,
    /// and its inbox sender), encoded by the concrete comms crate.
    ///
    /// A host that composes an acceptor demux in front of this runtime (so
    /// remote peers can dial one shared listener and be routed to this
    /// identity's inbox) decodes the payload where it holds the concrete
    /// comms dependency (`meerkat_comms::HostAcceptorRegistrationMaterial`).
    /// `None` means this runtime exposes no registration material and the
    /// composer must fail closed (no acceptor registration). The default is
    /// `None`; only the concrete comms runtime overrides it — the typed
    /// trait surface itself continues to expose no signing material.
    fn host_acceptor_registration_payload(&self) -> Option<Arc<dyn std::any::Any + Send + Sync>> {
        None
    }

    /// Register a peer for admission-only trust without listing it in the
    /// directory.
    ///
    /// Used for control-plane edges — the canonical case is the supervisor
    /// bridge for session-backed mob members: lifecycle notifications
    /// (`mob.peer_added`, `mob.peer_retired`, …) must land at the member's
    /// inbox, but the supervisor must not appear as an ordinary sendable
    /// peer in `comms.peers` / REST / RPC / MCP. The admission gate consults
    /// both the public and private trust sets; `resolve_peer_directory()`
    /// consults only the public set.
    async fn add_private_trusted_peer(
        &self,
        _peer: TrustedPeerDescriptor,
    ) -> Result<(), SendError> {
        Err(SendError::Unsupported(
            "generated comms private trust mutation authority required".to_string(),
        ))
    }

    /// Remove a previously registered private-trust edge by peer ID.
    ///
    /// Returns `true` if the edge was present and removed, `false` if it
    /// was not.
    async fn remove_private_trusted_peer(&self, _peer_id: &str) -> Result<bool, SendError> {
        Err(SendError::Unsupported(
            "generated comms private trust mutation authority required".to_string(),
        ))
    }

    /// Install the host-owned outbound content-taint declaration.
    ///
    /// The declaration is host-set carrier config, not machine state: the
    /// host owns the "this session's content is tainted" fact and this
    /// runtime stamps it (inside the signed envelope region) on every
    /// outbound content-bearing send until changed. `None` clears the
    /// declaration (subsequent envelopes carry no claim — which receivers
    /// must never coalesce into `Clean`).
    ///
    /// The declaration is in-memory runtime state: a rebuilt runtime (e.g.
    /// a respawned mob member) starts with no declaration, which aligns
    /// with fresh-context taint semantics — hosts re-declare when their
    /// tracker re-marks the new context.
    ///
    /// Fails typed (never a silent no-op — silently dropping a security
    /// declaration would let tainted content ship with a clean-looking
    /// envelope) for runtimes that do not carry outbound comms.
    fn set_outbound_content_taint(
        &self,
        _taint: Option<crate::comms::SenderContentTaint>,
    ) -> Result<(), SendError> {
        Err(SendError::Unsupported(
            "outbound content-taint declaration not supported by this CommsRuntime".to_string(),
        ))
    }

    /// Dispatch a canonical comms command.
    async fn send(&self, _cmd: CommsCommand) -> Result<SendReceipt, SendError> {
        Err(SendError::Unsupported(
            "send not implemented for this CommsRuntime".to_string(),
        ))
    }

    #[doc(hidden)]
    fn stream(&self, scope: StreamScope) -> Result<EventStream, StreamError> {
        let scope_desc = match scope {
            StreamScope::Session(session_id) => format!("session {session_id}"),
            StreamScope::Interaction(interaction_id) => format!("interaction {}", interaction_id.0),
        };
        Err(StreamError::NotFound(scope_desc))
    }

    /// List peers visible to this runtime.
    async fn peers(&self) -> Vec<PeerDirectoryEntry> {
        Vec::new()
    }

    /// Count peers visible to this runtime.
    ///
    /// Implementations can override this to avoid materializing a full peer list.
    async fn peer_count(&self) -> usize {
        self.peers().await.len()
    }

    #[doc(hidden)]
    async fn send_and_stream(
        &self,
        cmd: CommsCommand,
    ) -> Result<(SendReceipt, EventStream), SendAndStreamError> {
        let receipt = self.send(cmd).await?;
        Err(SendAndStreamError::StreamAttach {
            receipt,
            error: StreamError::Internal(
                "send_and_stream is not implemented for this runtime".to_string(),
            ),
        })
    }

    /// Drain comms inbox and return messages formatted for the LLM
    async fn drain_messages(&self) -> Vec<String>;
    /// Get a notification when new messages arrive
    fn inbox_notify(&self) -> Arc<tokio::sync::Notify>;
    /// Returns true if a DISMISS signal was seen during the last `drain_messages` call.
    fn dismiss_received(&self) -> bool {
        false
    }
    /// Get an event injector for this runtime's inbox.
    ///
    /// Surfaces use this to push external events into the agent inbox.
    /// Returns `None` if the implementation doesn't support event injection.
    fn event_injector(&self) -> Option<Arc<dyn crate::EventInjector>> {
        None
    }

    /// Internal runtime seam for interaction-scoped streaming.
    #[doc(hidden)]
    fn interaction_event_injector(
        &self,
    ) -> Option<Arc<dyn crate::event_injector::SubscribableInjector>> {
        None
    }

    /// Drain comms inbox and return structured interactions.
    ///
    /// Default implementation wraps `drain_messages()` results as `InteractionContent::Message`
    /// with generated IDs.
    async fn drain_inbox_interactions(&self) -> Vec<crate::interaction::InboxInteraction> {
        self.drain_messages()
            .await
            .into_iter()
            .map(|text| crate::interaction::InboxInteraction {
                objective_id: None,
                id: crate::interaction::InteractionId(uuid::Uuid::new_v4()),
                from_route: None,
                from: "unknown".into(),
                content: crate::interaction::InteractionContent::Message {
                    body: text.clone(),
                    blocks: None,
                },
                rendered_text: text,
                handling_mode: crate::types::HandlingMode::Queue,
                render_metadata: None,
                sender_taint: None,
            })
            .collect()
    }

    /// Look up and remove a one-shot subscriber for the given interaction.
    ///
    /// Returns the event sender if a subscriber was registered (via `inject_with_subscription`).
    /// The entry is removed from the registry on lookup (one-shot).
    fn interaction_subscriber(
        &self,
        _id: &crate::interaction::InteractionId,
    ) -> Option<tokio::sync::mpsc::Sender<crate::event::AgentEvent>> {
        None
    }

    /// Take and clear the one-shot sender for an interaction-scoped stream.
    fn take_interaction_stream_sender(
        &self,
        _id: &crate::interaction::InteractionId,
    ) -> Option<tokio::sync::mpsc::Sender<crate::event::AgentEvent>> {
        self.interaction_subscriber(_id)
    }

    /// Signal that an interaction has reached a terminal state (complete or failed).
    ///
    /// Implementations should transition the reservation FSM to `Completed` and
    /// clean up registry entries. Called from the keep-alive loop after sending
    /// terminal events to the tap.
    fn mark_interaction_complete(&self, _id: &crate::interaction::InteractionId) {}

    /// Signal that an interaction stream became unusable for an explicit,
    /// typed reason. Implementations with machine-owned stream lifecycle must
    /// drive `InteractionStreamAbandoned`; transport-only implementations may
    /// clean up their local projection directly.
    fn abandon_interaction_stream(
        &self,
        _id: &crate::interaction::InteractionId,
        _reason: crate::InteractionStreamAbandonReason,
    ) {
    }

    /// Access the session's peer-interaction DSL handle (W1-A).
    ///
    /// Returns `None` for transport-only comms runtimes. A runtime that emits
    /// semantic peer request/response receipts must return `Some` after the
    /// surface installs machine authority.
    fn peer_interaction_handle(
        &self,
    ) -> Option<std::sync::Arc<dyn crate::handles::PeerInteractionHandle>> {
        None
    }

    /// Access peer request/response authority only when the runtime has the
    /// complete machine-owned lifecycle pair.
    ///
    /// Semantic peer request/response ingress requires both the peer
    /// interaction handle and the paired interaction-stream handle. The stream
    /// handle itself stays hidden behind runtime ownership; this witness lets
    /// authority boundaries fail closed instead of treating a lone peer handle
    /// as sufficient.
    fn peer_request_response_authority_handle(
        &self,
    ) -> Option<std::sync::Arc<dyn crate::handles::PeerInteractionHandle>> {
        None
    }

    /// Drain classified inbox interactions.
    ///
    /// Returns interactions with pre-computed classification from ingress.
    /// The host loop routes on the stored `PeerInputClass` instead of
    /// re-classifying after drain.
    ///
    /// Default returns `Unsupported`. Comms-enabled runtimes must override.
    async fn drain_classified_inbox_interactions(
        &self,
    ) -> Result<Vec<crate::interaction::ClassifiedInboxInteraction>, CommsCapabilityError> {
        Err(CommsCapabilityError::Unsupported(
            "drain_classified_inbox_interactions".to_string(),
        ))
    }

    /// Drain canonical peer/event ingress candidates.
    ///
    /// This remains the live runtime drain bridge for call sites that consume
    /// the `PeerInputCandidate` noun directly. The underlying drain unit is
    /// identical to `ClassifiedInboxInteraction`, so the default
    /// implementation simply forwards the classified drain path.
    async fn drain_peer_input_candidates(&self) -> Vec<crate::interaction::PeerInputCandidate> {
        self.drain_classified_inbox_interactions()
            .await
            .unwrap_or_default()
    }

    /// Snapshot the currently queued peer-ingress surface without draining it.
    ///
    /// This is a hidden diagnostic capability used while mapping the internal
    /// MeerkatMachine boundary onto existing comms ownership.
    async fn peer_ingress_queue_snapshot(
        &self,
    ) -> Result<crate::interaction::PeerIngressQueueSnapshot, CommsCapabilityError> {
        Err(CommsCapabilityError::Unsupported(
            "peer_ingress_queue_snapshot".to_string(),
        ))
    }

    /// Snapshot the current peer runtime surface for MeerkatMachine mapping.
    ///
    /// This extends the queued ingress snapshot with the local trust membership
    /// that governs peer admission.
    async fn peer_ingress_runtime_snapshot(
        &self,
    ) -> Result<crate::interaction::PeerIngressRuntimeSnapshot, CommsCapabilityError> {
        Err(CommsCapabilityError::Unsupported(
            "peer_ingress_runtime_snapshot".to_string(),
        ))
    }

    /// Snapshot only the public trust projection owned by generated public
    /// peer authority.
    ///
    /// Private/control-plane trust edges are admitted by separate generated
    /// private authority and must not be reconciled or removed by public peer
    /// projection owners.
    async fn public_trusted_peer_projection_snapshot(
        &self,
    ) -> Result<Vec<crate::comms::TrustedPeerDescriptor>, CommsCapabilityError> {
        Err(CommsCapabilityError::Unsupported(
            "public_trusted_peer_projection_snapshot".to_string(),
        ))
    }

    /// Snapshot the public trust projection owned by one generated source.
    ///
    /// This is the behavior-authority read used by generated trust
    /// reconciliation. Compatibility/public snapshots may still union public
    /// rows for display, but generated removals must diff only against rows
    /// previously installed by the same generated owner.
    async fn trusted_peer_projection_snapshot_for_source(
        &self,
        _source_kind: crate::comms::GeneratedCommsTrustAuthoritySourceKind,
    ) -> Result<Vec<crate::comms::TrustedPeerDescriptor>, CommsCapabilityError> {
        Err(CommsCapabilityError::Unsupported(
            "trusted_peer_projection_snapshot_for_source".to_string(),
        ))
    }

    /// Get a notification that fires only for actionable peer input.
    ///
    /// Default returns `Unsupported`. Comms-enabled runtimes must override.
    /// Used by the factory to bridge into `WaitTool` interrupt.
    fn actionable_input_notify(&self) -> Result<Arc<tokio::sync::Notify>, CommsCapabilityError> {
        Err(CommsCapabilityError::Unsupported(
            "actionable_input_notify".to_string(),
        ))
    }

    /// Stage a one-shot reply endpoint for a Response to a peer outside the
    /// trust store.
    ///
    /// This is the legacy uncorrelated compatibility seam. It is
    /// Response-only, one-shot, and trust-store-losing; callers may supply
    /// only a machine-authorized endpoint already held in runtime state.
    /// Neither a Request's `reply_endpoint` nor any decoded payload/sender
    /// address is authority for this method. New ingress response paths use
    /// [`Self::stage_correlated_reply_endpoint`] instead.
    ///
    /// Parameters are primitives because core cannot name the comms-crate
    /// newtypes (dependency direction). Default fails typed, not no-op:
    /// silently dropping a reply-repair staging would strand the remote
    /// sender in a timeout with no cause. Callers decide policy — reply
    /// drains treat `Unsupported` as "runtime has no staging capability" and
    /// proceed, since in-proc runtimes resolve via the ingress route anyway.
    async fn stage_declared_reply_endpoint(
        &self,
        _dest: PeerId,
        _signer_pubkey: [u8; 32],
        _declared_address: String,
    ) -> Result<(), SendError> {
        Err(SendError::Unsupported(
            "declared reply endpoint staging not supported".to_string(),
        ))
    }

    /// Stage an authenticated one-shot endpoint for the Response correlated
    /// to `in_reply_to` from `dest`.
    ///
    /// Unlike the legacy uncorrelated staging seam above, this endpoint is
    /// keyed by both peer identity and request id and therefore takes
    /// precedence over durable trust only for that exact Response. This is
    /// the only Request-ingress callback seam. `signer_pubkey` must
    /// come from a signature-verified envelope and derive `dest` in the
    /// concrete runtime. `declared_endpoint` must be the classifier's
    /// source-confined TCP projection: kernel-observed source IP plus the
    /// signed, nonzero declared port. Arbitrary payload addresses,
    /// sender-selected hosts, UDS addresses, and open-auth ingress are never
    /// callback authority.
    async fn stage_correlated_reply_endpoint(
        &self,
        _dest: PeerId,
        _in_reply_to: crate::interaction::InteractionId,
        _signer_pubkey: [u8; 32],
        _declared_endpoint: crate::comms::PeerAddress,
    ) -> Result<(), SendError> {
        Err(SendError::Unsupported(
            "correlated reply endpoint staging not supported".to_string(),
        ))
    }

    /// Idempotently discard a previously staged correlated endpoint.
    /// Responders call this when validation or response sending fails before
    /// the Router consumes the exact one-shot entry.
    async fn unstage_correlated_reply_endpoint(
        &self,
        _dest: PeerId,
        _in_reply_to: crate::interaction::InteractionId,
    ) -> Result<(), SendError> {
        Err(SendError::Unsupported(
            "correlated reply endpoint cleanup not supported".to_string(),
        ))
    }

    /// One-shot reply waiter for an agent-blocking bridge request (member
    /// upcall lane). Consulted by the comms drain BEFORE session injection: a
    /// taken waiter receives the terminal Response candidate (typed
    /// terminality intact) and the candidate never becomes session input.
    ///
    /// Returns `Some(sender)` only for a live waiter. A tombstoned (timed
    /// out) waiter entry is consumed and `None` is returned — pair with
    /// [`Self::has_bridge_reply_waiter`] to distinguish "tombstone consumed"
    /// (discard the late reply) from "never registered" (ordinary session
    /// path). Default: no registry (a query, not a capability — absence of a
    /// waiter is the universal normal case).
    fn take_bridge_reply_waiter(
        &self,
        _in_reply_to: &crate::interaction::InteractionId,
    ) -> Option<tokio::sync::oneshot::Sender<crate::interaction::PeerInputCandidate>> {
        None
    }

    /// True when a bridge-reply waiter entry (live or tombstoned) is
    /// registered for `in_reply_to`. See [`Self::take_bridge_reply_waiter`].
    fn has_bridge_reply_waiter(&self, _in_reply_to: &crate::interaction::InteractionId) -> bool {
        false
    }
}

/// The main Agent struct
pub struct Agent<C, T, S>
where
    C: AgentLlmClient + ?Sized,
    T: AgentToolDispatcher + ?Sized,
    S: AgentSessionStore + ?Sized,
{
    config: AgentConfig,
    client: Arc<C>,
    tools: Arc<T>,
    tool_scope: ToolScope,
    store: Arc<S>,
    session: Session,
    budget: Budget,
    retry_policy: RetryPolicy,
    depth: u32,
    pub(super) comms_runtime: Option<Arc<dyn CommsRuntime>>,
    pub(super) hook_engine: Option<Arc<dyn HookEngine>>,
    pub(super) hook_run_overrides: HookRunOverrides,
    /// Optional context compaction strategy.
    pub(crate) compactor: Option<Arc<dyn crate::compact::Compactor>>,
    /// Optional host-supplied compaction summary curator. When present it
    /// produces the compaction summary instead of the summarization LLM call.
    pub(crate) compaction_curator: Option<Arc<dyn crate::compact::CompactionCurator>>,
    /// Input tokens from the last LLM response (for compaction trigger).
    pub(crate) last_input_tokens: u64,
    /// Session-scoped compaction cadence tracked across runs.
    pub(crate) compaction_cadence: SessionCompactionCadence,
    /// Optional memory store for indexing compaction discards.
    pub(crate) memory_store: Option<Arc<dyn crate::memory::MemoryStore>>,
    /// Runtime-owned resultful handoff for durable transcript+memory
    /// compaction pairs. Absent on standalone paths.
    pub(crate) compaction_commit_coordinator:
        Option<Arc<dyn crate::memory::CompactionCommitCoordinator>>,
    /// Typed lifecycle for the current transcript-rewrite + staged-memory
    /// transaction. Runtime reconciliation advances this to commit-only before
    /// touching the memory store; abort is legal only while runtime commit is
    /// still pending.
    pub(crate) compaction_transaction: Option<CompactionTransaction>,
    /// Deterministic projection identity installed immediately before the
    /// durable stage await. A hard interrupt can drop that await before a
    /// receipt reaches the transaction owner, so cleanup must retain the exact
    /// identity rather than infer empty RuntimeStore authority.
    pub(crate) in_flight_compaction_stage: Option<crate::memory::CompactionProjectionId>,
    /// Optional skill engine for per-turn `/skill-ref` activation.
    pub(crate) skill_engine: Option<Arc<crate::skills::SkillRuntime>>,
    /// Skill references to resolve and inject for the next turn.
    /// Set by surfaces before calling `run()`, consumed on run start.
    pub pending_skill_references: Option<Vec<crate::skills::SkillKey>>,
    /// Per-interaction event tap for streaming events to subscribers.
    pub(crate) event_tap: crate::event_tap::EventTap,
    /// Shared control state for runtime system-context appends.
    pub(crate) system_context_state: crate::session::SystemContextStateHandle,
    /// Optional default event channel configured at build time.
    /// Used by run methods when no per-call event channel is provided.
    pub(crate) default_event_tx: Option<tokio::sync::mpsc::Sender<crate::event::AgentEvent>>,
    /// Optional session checkpointer for keep-alive persistence.
    ///
    /// Wired by `AgentBuilder::with_checkpointer`, installed by
    /// `PersistentSessionService`, and consumed by
    /// `Agent::checkpoint_current_session`.
    pub(crate) checkpointer: Option<Arc<dyn crate::checkpoint::SessionCheckpointer>>,
    /// Optional blob store used to hydrate image refs at execution seams.
    pub(crate) blob_store: Option<Arc<dyn crate::BlobStore>>,
    /// Original error detail preserved from `terminalize_fatal_error` so
    /// `build_result` can include the actual failure message (e.g. the API
    /// error body) instead of only the generic terminal-cause description.
    pub(crate) terminal_error_detail: Option<String>,
    /// Structured metadata captured from that concrete error before the
    /// public result is normalized into `AgentError::TerminalFailure`.
    pub(crate) terminal_error_metadata: Option<crate::TurnErrorMetadata>,
    /// True once the current run has accepted `RunCompleted` hooks.
    pub(crate) run_completed_hooks_applied: bool,
    /// True once the current run's public `RunCompleted` event has been
    /// emitted. Extraction may continue afterward as a separate post-run phase.
    pub(crate) run_completed_event_emitted: bool,
    /// Comms intents that should be silently injected into the session
    /// without triggering an LLM turn. Matched against `InteractionContent::Request.intent`.
    #[allow(dead_code)] // Used by comms_impl when comms feature is enabled
    pub(crate) silent_comms_intents: Vec<String>,
    /// Optional shared lifecycle registry for async operations.
    pub(crate) ops_lifecycle: Option<Arc<dyn crate::ops_lifecycle::OpsLifecycleRegistry>>,
    /// Optional completion feed for cursor-based completion delivery.
    pub(crate) completion_feed: Option<Arc<dyn crate::completion_feed::CompletionFeed>>,
    /// Shared epoch cursor state for runtime-backed cursor writeback.
    pub(crate) epoch_cursor_state: Option<Arc<crate::runtime_epoch::EpochCursorState>>,
    /// Local cursor into the completion feed — only the agent boundary advances this.
    pub(crate) applied_cursor: crate::completion_feed::CompletionSeq,
    /// Optional enrichment provider for completion display details.
    pub(crate) completion_enrichment:
        Option<Arc<dyn crate::completion_feed::CompletionEnrichmentProvider>>,
    /// Shared effective mob authority handle. Owned by the agent, passed to
    /// mob tools at construction for authorization reads. Updated by
    /// `apply_session_effects` after each tool batch as a derived projection
    /// of the canonical `session.build_state().mob_tool_authority_context`.
    pub(crate) mob_authority_handle:
        Option<Arc<std::sync::RwLock<crate::service::MobToolAuthorityContext>>>,
    /// Runtime-backed turn-state handle, provided by the session runtime bindings.
    pub(crate) turn_state_handle: Option<Arc<dyn crate::TurnStateHandle>>,
    /// Runtime-backed model-routing authority. Sticky fallback commits route
    /// through this handle in the compensated client/auth/machine transaction.
    pub(crate) model_routing_handle: Option<Arc<dyn crate::handles::ModelRoutingHandle>>,
    /// Runtime-owned durable sticky-fallback transaction coordinator.
    /// Standalone agents leave this absent and consume staged machine commits
    /// synchronously in-process.
    pub(crate) sticky_model_fallback_commit_coordinator:
        Option<Arc<dyn crate::handles::StickyModelFallbackCommitCoordinator>>,
    /// Saga state retained across cancellation while the supervised durable
    /// sticky-fallback transaction is in flight.
    pub(crate) pending_sticky_model_fallback_activation:
        Option<state::PendingStickyModelFallbackActivation>,
    /// Async operation references staged behind an external callback boundary.
    /// They are registered with the fresh continuation run before it can call
    /// the provider, preserving Barrier versus Detached semantics.
    pub(crate) pending_callback_async_ops: Option<Vec<crate::ops::AsyncOpRef>>,
    /// Effective model registry captured by the construction pipeline.
    /// Fallback profile and limit truth is freshly resolved through this exact
    /// registry before it can reach the routing machine.
    pub(crate) effective_model_registry: Option<Arc<crate::ModelRegistry>>,
    /// Registry-minted facts for the active model. This replaces client-local
    /// capability/limit projections as the durable source used by later turns.
    pub(crate) active_model_profile: Option<crate::ModelProfileWitness>,
    /// True when the runtime control plane must stamp execution kind metadata.
    pub(crate) runtime_execution_kind_required: bool,
    /// Typed execution intent for the current run, when this turn is owned by
    /// the runtime control plane rather than a direct surface call.
    pub(crate) runtime_execution_kind: Option<crate::lifecycle::RuntimeExecutionKind>,
    /// Exact per-call witness that the core turn machine admitted a runtime
    /// run. A completed future alone is not sufficient evidence: preflight
    /// failures can return before `StartConversationRun` and must never reuse
    /// the previous turn's terminal snapshot.
    pub(crate) runtime_started_run_id: Option<crate::lifecycle::RunId>,
    /// Machine-terminal failure observed for the exact runtime run above.
    /// Kept separate from the public `AgentError` so direct session surfaces
    /// preserve their original typed errors while the runtime can commit a
    /// failed-but-applied turn atomically.
    pub(crate) runtime_terminal_failure_witness:
        Option<Result<crate::TurnErrorMetadata, crate::error::AgentError>>,
    /// Stable transcript identity for the active runtime-owned turn.
    pub(crate) active_transcript_identity: Option<crate::types::TranscriptMessageIdentity>,
    /// Runtime-backed external tool-surface diagnostic handle, when provided
    /// by the session runtime bindings.
    pub(crate) external_tool_surface_handle: Option<Arc<dyn crate::ExternalToolSurfaceHandle>>,
    /// Runtime-backed auth lease handle (Phase 1.5-rev).
    pub(crate) auth_lease_handle: Option<crate::handles::GeneratedAuthLeaseHandle>,
    /// Runtime-backed MCP server lifecycle handle (Phase 5G / T5g). When set,
    /// the agent loop reads `pending_server_ids()` at each CallingLlm boundary
    /// to decide whether to emit the `[MCP_PENDING]` system notice.
    pub(crate) mcp_server_lifecycle_handle:
        Option<Arc<dyn crate::handles::McpServerLifecycleHandle>>,
    /// Producer end of the typed cancel-after-boundary command channel.
    ///
    /// Retained so [`Agent::cancel_after_boundary_handle`] can hand cloned
    /// senders to the surface that requests boundary-only cancellation. The
    /// agent never sends on this end itself; it only drains the matching
    /// receiver at turn boundaries.
    pub(crate) cancel_after_boundary_tx: CancelAfterBoundarySender,
    /// Consumer end of the typed cancel-after-boundary command channel.
    ///
    /// Drained (non-blocking) at each turn boundary by
    /// `observe_cancel_after_boundary_request`, replacing the previous
    /// `.swap`-polled `AtomicBool`. A delivered [`CancelAfterBoundaryCommand`]
    /// is observed at most once per boundary, mirroring the prior edge
    /// semantics.
    pub(crate) cancel_after_boundary_rx:
        tokio::sync::mpsc::UnboundedReceiver<CancelAfterBoundaryCommand>,
    /// Optional resolver for model-specific operational defaults (e.g., call timeout).
    /// Consulted at each LLM call for hot-swap-aware profile default resolution.
    pub(crate) model_defaults_resolver:
        Option<Arc<dyn crate::model_defaults::ModelOperationalDefaultsResolver>>,
    /// Explicit call-timeout override from the build/config composition seam.
    /// Takes precedence over profile-derived defaults.
    pub(crate) call_timeout_override: crate::config::CallTimeoutOverride,
    /// Structured-output extraction state carried into RunResult.
    pub(crate) extraction_state: extraction::ExtractionState,
    /// Last published hidden deferred-catalog names.
    pub(crate) last_hidden_deferred_catalog_names: BTreeSet<crate::types::ToolName>,
    /// Last published pending catalog sources.
    pub(crate) last_pending_catalog_sources: BTreeSet<String>,
    /// Dispatch-time projection of the current turn input for contextual tools.
    pub(crate) tool_dispatch_context: ToolDispatchContext,
    /// Runtime-owned dispatch metadata for this turn.
    pub(crate) turn_tool_dispatch_metadata: BTreeMap<String, serde_json::Value>,
    /// Typed tool-execution policy (per-call timeouts + concurrency bound)
    /// applied to the normal LLM-driven tool dispatch loop. Populated by the
    /// composition seam via `AgentBuilder::with_tools_config`; defaults to
    /// `ToolsConfig::default()` for standalone/test construction.
    pub(crate) tools_config: crate::config::ToolsConfig,
}

#[derive(Clone)]
pub(crate) struct CompactionRollbackState {
    pub(crate) rollback_session: Session,
    pub(crate) rollback_last_input_tokens: u64,
    pub(crate) rollback_compaction_cadence: SessionCompactionCadence,
}

pub(crate) enum CompactionTransactionPhase {
    AwaitingRuntimeCommit(Box<CompactionRollbackState>),
    RuntimeCommitted { bookkeeping_complete: bool },
    AbortPending { cadence_persist_pending: bool },
}

pub(crate) struct CompactionTransaction {
    pub(crate) phase: CompactionTransactionPhase,
    pub(crate) projections: Vec<crate::memory::CompactionProjectionId>,
}

#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
mod tests {
    use super::{
        AgentToolDispatcher, CommsRuntime, DEFAULT_MAX_INLINE_PEER_NOTIFICATIONS,
        FilteredToolDispatcher, InlinePeerNotificationPolicy, ToolDispatchContext,
    };
    use crate::comms::{
        PeerAddress, PeerId, PeerName, PeerTransport, SendError, TrustedPeerDescriptor,
    };
    use crate::types::{ContentBlock, ContentInput, ToolCallView, ToolDef, ToolResult};
    use async_trait::async_trait;
    use serde_json::json;
    use std::sync::Arc;
    use tokio::sync::Notify;

    struct NoopCommsRuntime {
        notify: Arc<Notify>,
    }

    struct ContextAwareToolDispatcher;

    struct ExactExecutionDispatcher {
        catalog: Arc<[crate::ToolCatalogEntry]>,
    }

    struct HybridExecutionDispatcher {
        catalog: Arc<[crate::ToolCatalogEntry]>,
    }

    struct StreamingExecutionDispatcher {
        catalog: Arc<[crate::ToolCatalogEntry]>,
        saw_streaming_context: Arc<std::sync::atomic::AtomicBool>,
    }

    struct IdenticalMutationDispatcher {
        tool: ToolDef,
        epoch: std::sync::atomic::AtomicU64,
        mutate_on_resolve: bool,
    }

    #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
    #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
    impl AgentToolDispatcher for ContextAwareToolDispatcher {
        fn tools(&self) -> Arc<[Arc<ToolDef>]> {
            Arc::from([Arc::new(ToolDef {
                name: "inspect_context".into(),
                description: "inspect context".to_string(),
                input_schema: json!({"type": "object"}),
                provenance: None,
            })])
        }

        async fn dispatch(
            &self,
            call: ToolCallView<'_>,
        ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
            Ok(ToolResult::new(
                call.id.to_string(),
                json!({"saw_context_image": false}).to_string(),
                false,
            )
            .into())
        }

        async fn dispatch_with_context(
            &self,
            call: ToolCallView<'_>,
            context: &ToolDispatchContext,
        ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
            let saw_context_image = context
                .current_turn()
                .and_then(|turn| turn.image_ref(0))
                .and_then(|image_ref| context.current_turn_image(image_ref))
                .is_some();
            Ok(ToolResult::new(
                call.id.to_string(),
                json!({"saw_context_image": saw_context_image}).to_string(),
                false,
            )
            .into())
        }
    }

    #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
    #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
    impl AgentToolDispatcher for ExactExecutionDispatcher {
        fn tools(&self) -> Arc<[Arc<ToolDef>]> {
            self.catalog
                .iter()
                .filter(|entry| entry.currently_callable())
                .map(|entry| Arc::clone(&entry.tool))
                .collect::<Vec<_>>()
                .into()
        }

        fn tool_catalog_capabilities(&self) -> crate::ToolCatalogCapabilities {
            crate::ToolCatalogCapabilities {
                exact_catalog: true,
                may_require_catalog_control_plane: false,
            }
        }

        fn tool_catalog(&self) -> Arc<[crate::ToolCatalogEntry]> {
            Arc::clone(&self.catalog)
        }

        async fn dispatch(
            &self,
            call: ToolCallView<'_>,
        ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
            Ok(ToolResult::new(call.id.to_string(), "ok".to_string(), false).into())
        }
    }

    #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
    #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
    impl AgentToolDispatcher for HybridExecutionDispatcher {
        fn tools(&self) -> Arc<[Arc<ToolDef>]> {
            self.catalog
                .iter()
                .filter(|entry| entry.currently_callable())
                .map(|entry| Arc::clone(&entry.tool))
                .collect::<Vec<_>>()
                .into()
        }

        fn tool_catalog_capabilities(&self) -> crate::ToolCatalogCapabilities {
            crate::ToolCatalogCapabilities {
                exact_catalog: true,
                may_require_catalog_control_plane: false,
            }
        }

        fn tool_catalog(&self) -> Arc<[crate::ToolCatalogEntry]> {
            Arc::clone(&self.catalog)
        }

        fn resolve_execution_plan(
            &self,
            call: ToolCallView<'_>,
            _dispatch_context: &ToolDispatchContext,
            resolution_context: &crate::ToolExecutionResolutionContext,
        ) -> Result<crate::ResolvedToolExecutionPlan, crate::ToolExecutionResolutionError> {
            let entry = self
                .catalog
                .iter()
                .find(|entry| entry.tool.name == call.name)
                .ok_or_else(|| crate::ToolExecutionResolutionError::NotFound {
                    tool_name: call.name.to_string(),
                })?;
            let arguments: serde_json::Value =
                serde_json::from_str(call.args.get()).map_err(|error| {
                    crate::ToolExecutionResolutionError::InvalidArguments {
                        tool_name: call.name.to_string(),
                        reason: error.to_string(),
                    }
                })?;
            let mode = if arguments["run_detached"] == true {
                crate::ToolExecutionMode::Detached
            } else {
                crate::ToolExecutionMode::Fast
            };
            entry
                .execution
                .resolve(mode, resolution_context.deadlines().clone())
                .map_err(crate::ToolExecutionResolutionError::from)
        }

        async fn dispatch(
            &self,
            call: ToolCallView<'_>,
        ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
            Ok(ToolResult::new(
                call.id.to_string(),
                json!({"owner": "filtered-hybrid-owner"}).to_string(),
                false,
            )
            .into())
        }

        async fn dispatch_resolved_with_context(
            &self,
            call: ToolCallView<'_>,
            _context: &ToolDispatchContext,
            plan: &crate::ResolvedToolExecutionPlan,
        ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
            if plan.mode() != crate::ToolExecutionMode::Detached {
                return Err(crate::ToolError::execution_failed(
                    "test detached owner received the wrong plan",
                ));
            }
            self.dispatch(call).await
        }
    }

    #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
    #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
    impl AgentToolDispatcher for StreamingExecutionDispatcher {
        fn tools(&self) -> Arc<[Arc<ToolDef>]> {
            self.catalog
                .iter()
                .map(|entry| Arc::clone(&entry.tool))
                .collect::<Vec<_>>()
                .into()
        }

        fn tool_catalog_capabilities(&self) -> crate::ToolCatalogCapabilities {
            crate::ToolCatalogCapabilities {
                exact_catalog: true,
                may_require_catalog_control_plane: false,
            }
        }

        fn tool_catalog(&self) -> Arc<[crate::ToolCatalogEntry]> {
            Arc::clone(&self.catalog)
        }

        async fn dispatch(
            &self,
            call: ToolCallView<'_>,
        ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
            Err(crate::ToolError::unavailable(
                call.name,
                crate::ToolUnavailableReason::ExecutionModeOwnerUnavailable,
            ))
        }

        async fn dispatch_resolved_with_context(
            &self,
            call: ToolCallView<'_>,
            context: &ToolDispatchContext,
            plan: &crate::ResolvedToolExecutionPlan,
        ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
            if plan.mode() != crate::ToolExecutionMode::Streaming {
                return Err(crate::ToolError::execution_failed(
                    "streaming owner received a non-streaming plan",
                ));
            }
            let streaming = context.streaming().ok_or_else(|| {
                crate::ToolError::unavailable(
                    call.name,
                    crate::ToolUnavailableReason::ExecutionModeOwnerUnavailable,
                )
            })?;
            streaming
                .progress()
                .try_report(
                    crate::ToolProgressFrame::message("accepted through wrapper")
                        .map_err(|error| crate::ToolError::other(error.to_string()))?,
                )
                .map_err(|error| crate::ToolError::other(error.to_string()))?;
            self.saw_streaming_context
                .store(true, std::sync::atomic::Ordering::SeqCst);
            Ok(ToolResult::new(call.id.to_string(), "stream complete".to_string(), false).into())
        }
    }

    #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
    #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
    impl AgentToolDispatcher for IdenticalMutationDispatcher {
        fn tools(&self) -> Arc<[Arc<ToolDef>]> {
            Arc::from([Arc::new(self.tool.clone())])
        }

        fn tool_catalog_capabilities(&self) -> crate::ToolCatalogCapabilities {
            crate::ToolCatalogCapabilities {
                exact_catalog: true,
                may_require_catalog_control_plane: false,
            }
        }

        fn tool_catalog(&self) -> Arc<[crate::ToolCatalogEntry]> {
            Arc::from([crate::ToolCatalogEntry::session_inline(
                Arc::new(self.tool.clone()),
                true,
            )])
        }

        fn execution_binding_epoch(&self, _tool_name: &str) -> u64 {
            self.epoch.load(std::sync::atomic::Ordering::SeqCst)
        }

        fn resolve_execution_plan(
            &self,
            _call: ToolCallView<'_>,
            _dispatch_context: &ToolDispatchContext,
            resolution_context: &crate::ToolExecutionResolutionContext,
        ) -> Result<crate::ResolvedToolExecutionPlan, crate::ToolExecutionResolutionError> {
            let plan = crate::ToolExecutionContract::default()
                .resolve_default(resolution_context.deadlines().clone())
                .map_err(crate::ToolExecutionResolutionError::from)?;
            if self.mutate_on_resolve {
                self.epoch.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            }
            Ok(plan)
        }

        async fn dispatch(
            &self,
            call: ToolCallView<'_>,
        ) -> Result<crate::ops::ToolDispatchOutcome, crate::error::ToolError> {
            Ok(ToolResult::new(call.id.to_string(), "ok".to_string(), false).into())
        }
    }

    #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
    #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
    impl CommsRuntime for NoopCommsRuntime {
        async fn drain_messages(&self) -> Vec<String> {
            Vec::new()
        }

        fn inbox_notify(&self) -> std::sync::Arc<Notify> {
            self.notify.clone()
        }
    }

    #[tokio::test]
    async fn test_comms_runtime_trait_defaults_hide_unimplemented_features() {
        let runtime = NoopCommsRuntime {
            notify: Arc::new(Notify::new()),
        };
        assert!(<NoopCommsRuntime as CommsRuntime>::public_key(&runtime).is_none());
        // The only mutable trust seam is apply_trust_mutation; without a
        // generated handoff it fails closed.
        let peer = TrustedPeerDescriptor {
            peer_id: PeerId::new(),
            name: PeerName::new("peer-a").expect("valid peer name"),
            address: PeerAddress::new(PeerTransport::Inproc, "peer-a"),
            pubkey: [0u8; 32],
        };
        let result =
            <NoopCommsRuntime as CommsRuntime>::add_private_trusted_peer(&runtime, peer).await;
        assert!(matches!(result, Err(SendError::Unsupported(_))));
    }

    /// T-12: bridge-reply waiter + declared-reply-endpoint trait defaults.
    /// `take_bridge_reply_waiter` → None (no registry),
    /// `has_bridge_reply_waiter` → false, and
    /// `stage_declared_reply_endpoint` fails typed (never a silent no-op) so
    /// a caller cannot mistake a dropped security-relevant repair for success.
    #[tokio::test]
    async fn test_comms_runtime_bridge_reply_defaults() {
        let runtime = NoopCommsRuntime {
            notify: Arc::new(Notify::new()),
        };
        let interaction_id = crate::interaction::InteractionId(uuid::Uuid::new_v4());
        assert!(
            <NoopCommsRuntime as CommsRuntime>::take_bridge_reply_waiter(&runtime, &interaction_id)
                .is_none()
        );
        assert!(
            !<NoopCommsRuntime as CommsRuntime>::has_bridge_reply_waiter(&runtime, &interaction_id)
        );
        let staged = <NoopCommsRuntime as CommsRuntime>::stage_declared_reply_endpoint(
            &runtime,
            PeerId::new(),
            [0x11u8; 32],
            "tcp://127.0.0.1:1".to_string(),
        )
        .await;
        assert!(matches!(staged, Err(SendError::Unsupported(_))));
    }

    #[tokio::test]
    async fn filtered_tool_dispatcher_preserves_dispatch_context() {
        let dispatcher =
            FilteredToolDispatcher::new(Arc::new(ContextAwareToolDispatcher), ["inspect_context"]);
        let args = serde_json::value::RawValue::from_string("{}".to_string())
            .expect("empty object should be valid JSON");
        let call = ToolCallView {
            id: "ctx-1",
            name: "inspect_context",
            args: &args,
        };
        let context = ToolDispatchContext::from_current_turn_input(&ContentInput::Blocks(vec![
            ContentBlock::Image {
                media_type: "image/png".to_string(),
                data: "abc".into(),
            },
        ]));

        let outcome = dispatcher
            .dispatch_with_context(call, &context)
            .await
            .expect("filtered wrapper should dispatch");
        let payload: serde_json::Value =
            serde_json::from_str(&outcome.result.text_content()).expect("tool result JSON");
        assert_eq!(payload["saw_context_image"], true);
    }

    #[test]
    fn default_execution_plan_resolver_uses_exact_catalog_contract() {
        use crate::{
            DetachedToolExecutionPolicy, IdempotencyScope, RestartClass, RunnerIdentity,
            ToolDeadlineChain, ToolDeadlineContributor, ToolDeadlineOwner, ToolExecutionContract,
            ToolExecutionMode, ToolExecutionResolutionContext,
        };
        use std::collections::BTreeSet;
        use std::time::Duration;

        let detached = DetachedToolExecutionPolicy::new(
            RunnerIdentity::new("homecore.security_scan", "v1").unwrap(),
            RestartClass::NonResumable,
            IdempotencyScope::InteractionAndArguments,
            Duration::from_secs(10),
        )
        .unwrap();
        let contract = ToolExecutionContract::new(
            BTreeSet::from([ToolExecutionMode::Detached]),
            ToolExecutionMode::Detached,
            None,
            Some(detached),
        )
        .unwrap();
        let dispatcher = ExactExecutionDispatcher {
            catalog: Arc::from([crate::ToolCatalogEntry::session_inline(
                Arc::new(ToolDef::new(
                    "security_scan",
                    "scan",
                    json!({"type": "object"}),
                )),
                true,
            )
            .with_execution_contract(contract)]),
        };
        let args = serde_json::value::RawValue::from_string("{}".to_string()).unwrap();
        let call = ToolCallView {
            id: "call-1",
            name: "security_scan",
            args: &args,
        };
        let resolution = ToolExecutionResolutionContext::new(
            ToolDeadlineChain::new(vec![ToolDeadlineContributor::finite(
                ToolDeadlineOwner::CoreToolDispatch,
                Duration::from_secs(600),
            )])
            .unwrap(),
        );

        let plan = dispatcher
            .resolve_execution_plan(call, &ToolDispatchContext::default(), &resolution)
            .expect("declared plan resolves");

        assert_eq!(plan.mode(), ToolExecutionMode::Detached);
        assert_eq!(
            plan.deadlines().effective_timeout(),
            Some(Duration::from_secs(10))
        );
        assert_eq!(
            plan.deadlines().winner().map(|winner| winner.owner()),
            Some(ToolDeadlineOwner::DetachedSubmission)
        );
    }

    #[tokio::test]
    async fn default_resolved_dispatch_refuses_detached_plan_before_ordinary_dispatch() {
        use crate::{
            DetachedToolExecutionPolicy, IdempotencyScope, RestartClass, RunnerIdentity,
            ToolDeadlineChain, ToolDeadlineContributor, ToolDeadlineOwner, ToolExecutionContract,
            ToolExecutionMode, ToolExecutionResolutionContext,
        };
        use std::collections::BTreeSet;
        use std::time::Duration;

        let detached = DetachedToolExecutionPolicy::new(
            RunnerIdentity::new("detached.owner", "v1").unwrap(),
            RestartClass::NonResumable,
            IdempotencyScope::ToolCall,
            Duration::from_secs(10),
        )
        .unwrap();
        let contract = ToolExecutionContract::new(
            BTreeSet::from([ToolExecutionMode::Detached]),
            ToolExecutionMode::Detached,
            None,
            Some(detached),
        )
        .unwrap();
        let dispatcher = ExactExecutionDispatcher {
            catalog: Arc::from([crate::ToolCatalogEntry::session_inline(
                Arc::new(ToolDef::new(
                    "security_scan",
                    "scan",
                    json!({"type": "object"}),
                )),
                true,
            )
            .with_execution_contract(contract)]),
        };
        let args = serde_json::value::RawValue::from_string("{}".to_string()).unwrap();
        let call = ToolCallView {
            id: "detached-call",
            name: "security_scan",
            args: &args,
        };
        let context = ToolDispatchContext::default();
        let resolution = ToolExecutionResolutionContext::new(
            ToolDeadlineChain::new(vec![ToolDeadlineContributor::finite(
                ToolDeadlineOwner::CoreToolDispatch,
                Duration::from_secs(600),
            )])
            .unwrap(),
        );
        let plan = dispatcher
            .resolve_execution_plan(call, &context, &resolution)
            .expect("detached plan resolves");

        let error = dispatcher
            .dispatch_resolved_with_context(call, &context, &plan)
            .await
            .expect_err("the default dispatcher must not lower detached work to dispatch()");

        assert!(matches!(
            error,
            crate::ToolError::Unavailable {
                reason: crate::ToolUnavailableReason::ExecutionModeOwnerUnavailable,
                ..
            }
        ));
    }

    #[tokio::test]
    async fn fenced_streaming_dispatch_mints_context_and_filtered_wrapper_preserves_it() {
        use crate::{
            StreamingToolExecutionPolicy, ToolDeadlineChain, ToolDeadlineContributor,
            ToolDeadlineOwner, ToolExecutionContract, ToolExecutionMode,
            ToolExecutionResolutionContext,
        };
        use std::collections::BTreeSet;
        use std::time::Duration;

        let contract = ToolExecutionContract::new(
            BTreeSet::from([ToolExecutionMode::Streaming]),
            ToolExecutionMode::Streaming,
            Some(
                StreamingToolExecutionPolicy::new(Duration::from_secs(5), Duration::from_secs(30))
                    .unwrap(),
            ),
            None,
        )
        .unwrap();
        let saw_streaming_context = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let owner = StreamingExecutionDispatcher {
            catalog: Arc::from([crate::ToolCatalogEntry::session_inline(
                Arc::new(ToolDef::new(
                    "stream_scan",
                    "stream scan",
                    json!({"type": "object"}),
                )),
                true,
            )
            .with_execution_contract(contract)]),
            saw_streaming_context: Arc::clone(&saw_streaming_context),
        };
        let dispatcher = Arc::new(FilteredToolDispatcher::new(
            Arc::new(owner),
            ["stream_scan"],
        ));
        let filtered_catalog = dispatcher.tool_catalog();
        assert_eq!(
            filtered_catalog[0].execution.default_mode(),
            ToolExecutionMode::Streaming
        );
        let filtered_policy = filtered_catalog[0]
            .execution
            .streaming_policy()
            .expect("wrapper preserves the streaming registration");
        assert_eq!(filtered_policy.inactivity_timeout(), Duration::from_secs(5));
        assert_eq!(filtered_policy.absolute_timeout(), Duration::from_secs(30));
        let args = serde_json::value::RawValue::from_string("{}".to_string()).unwrap();
        let call = ToolCallView {
            id: "stream-call",
            name: "stream_scan",
            args: &args,
        };
        let context = ToolDispatchContext::default();
        assert!(
            context.streaming().is_none(),
            "callers cannot pre-mint the supervised streaming context"
        );
        let resolution = ToolExecutionResolutionContext::new(
            ToolDeadlineChain::new(vec![ToolDeadlineContributor::finite(
                ToolDeadlineOwner::CoreToolDispatch,
                Duration::from_secs(60),
            )])
            .unwrap(),
        );
        let plan =
            crate::resolve_tool_execution_plan_fenced(&dispatcher, call, &context, &resolution)
                .expect("streaming plan resolves through wrapper");

        let outcome =
            crate::dispatch_tool_execution_plan_fenced(&dispatcher, call, &context, &plan)
                .await
                .expect("streaming dispatch completes");

        assert_eq!(outcome.result.text_content(), "stream complete");
        assert!(
            saw_streaming_context.load(std::sync::atomic::Ordering::SeqCst),
            "the wrapper must preserve the exact supervised context"
        );
    }

    #[tokio::test]
    async fn declared_streaming_without_a_mode_owner_fails_closed_before_plain_dispatch() {
        use crate::{
            StreamingToolExecutionPolicy, ToolDeadlineChain, ToolDeadlineContributor,
            ToolDeadlineOwner, ToolExecutionContract, ToolExecutionMode,
            ToolExecutionResolutionContext,
        };
        use std::collections::BTreeSet;
        use std::time::Duration;

        let contract = ToolExecutionContract::new(
            BTreeSet::from([ToolExecutionMode::Streaming]),
            ToolExecutionMode::Streaming,
            Some(
                StreamingToolExecutionPolicy::new(Duration::from_secs(5), Duration::from_secs(30))
                    .unwrap(),
            ),
            None,
        )
        .unwrap();
        let dispatcher = Arc::new(ExactExecutionDispatcher {
            catalog: Arc::from([crate::ToolCatalogEntry::session_inline(
                Arc::new(ToolDef::new(
                    "ownerless_stream",
                    "ownerless",
                    json!({"type": "object"}),
                )),
                true,
            )
            .with_execution_contract(contract)]),
        });
        let args = serde_json::value::RawValue::from_string("{}".to_string()).unwrap();
        let call = ToolCallView {
            id: "ownerless-call",
            name: "ownerless_stream",
            args: &args,
        };
        let context = ToolDispatchContext::default();
        let resolution = ToolExecutionResolutionContext::new(
            ToolDeadlineChain::new(vec![ToolDeadlineContributor::finite(
                ToolDeadlineOwner::CoreToolDispatch,
                Duration::from_secs(60),
            )])
            .unwrap(),
        );
        let plan =
            crate::resolve_tool_execution_plan_fenced(&dispatcher, call, &context, &resolution)
                .expect("declaration resolves");

        let error = crate::dispatch_tool_execution_plan_fenced(&dispatcher, call, &context, &plan)
            .await
            .expect_err("missing streaming owner must fail closed");
        assert!(matches!(
            error,
            crate::ToolError::Unavailable {
                reason: crate::ToolUnavailableReason::ExecutionModeOwnerUnavailable,
                ..
            }
        ));
    }

    #[test]
    fn filtered_execution_plan_resolver_rejects_policy_denied_tool() {
        use crate::{
            ToolDeadlineChain, ToolDeadlineContributor, ToolDeadlineOwner,
            ToolExecutionResolutionContext, ToolExecutionResolutionError,
        };
        use std::time::Duration;

        let dispatcher =
            FilteredToolDispatcher::new(Arc::new(ContextAwareToolDispatcher), Vec::<String>::new());
        let args = serde_json::value::RawValue::from_string("{}".to_string()).unwrap();
        let call = ToolCallView {
            id: "call-hidden",
            name: "inspect_context",
            args: &args,
        };
        let resolution = ToolExecutionResolutionContext::new(
            ToolDeadlineChain::new(vec![ToolDeadlineContributor::finite(
                ToolDeadlineOwner::CoreToolDispatch,
                Duration::from_secs(600),
            )])
            .unwrap(),
        );

        let error = dispatcher
            .resolve_execution_plan(call, &ToolDispatchContext::default(), &resolution)
            .expect_err("hidden tools must not resolve");

        assert_eq!(
            error,
            ToolExecutionResolutionError::AccessDenied {
                tool_name: "inspect_context".to_string(),
            }
        );
    }

    #[tokio::test]
    async fn filtered_execution_plan_forwards_hybrid_resolution_to_visible_owner() {
        use crate::{
            DetachedToolExecutionPolicy, IdempotencyScope, ResolvedExecutionKind, RestartClass,
            RunnerIdentity, ToolDeadlineChain, ToolDeadlineContributor, ToolDeadlineOwner,
            ToolExecutionContract, ToolExecutionMode, ToolExecutionResolutionContext,
        };
        use std::collections::BTreeSet;
        use std::time::Duration;

        let detached = DetachedToolExecutionPolicy::new(
            RunnerIdentity::new("filtered-hybrid-owner", "v1").unwrap(),
            RestartClass::NonResumable,
            IdempotencyScope::InteractionAndArguments,
            Duration::from_secs(10),
        )
        .unwrap();
        let contract = ToolExecutionContract::new(
            BTreeSet::from([ToolExecutionMode::Fast, ToolExecutionMode::Detached]),
            ToolExecutionMode::Fast,
            None,
            Some(detached),
        )
        .unwrap();
        let dispatcher = FilteredToolDispatcher::new(
            Arc::new(HybridExecutionDispatcher {
                catalog: Arc::from([crate::ToolCatalogEntry::session_inline(
                    Arc::new(ToolDef::new(
                        "hybrid_scan",
                        "filtered-hybrid-owner catalog",
                        json!({"type": "object"}),
                    )),
                    true,
                )
                .with_execution_contract(contract)]),
            }),
            ["hybrid_scan"],
        );
        let args = serde_json::value::RawValue::from_string(r#"{"run_detached":true}"#.to_string())
            .unwrap();
        let call = ToolCallView {
            id: "call-hybrid",
            name: "hybrid_scan",
            args: &args,
        };
        let resolution = ToolExecutionResolutionContext::new(
            ToolDeadlineChain::new(vec![ToolDeadlineContributor::finite(
                ToolDeadlineOwner::CoreToolDispatch,
                Duration::from_secs(600),
            )])
            .unwrap(),
        );

        let catalog = dispatcher.tool_catalog();
        assert_eq!(catalog[0].execution.default_mode(), ToolExecutionMode::Fast);
        assert_eq!(catalog[0].tool.description, "filtered-hybrid-owner catalog");

        let plan = dispatcher
            .resolve_execution_plan(call, &ToolDispatchContext::default(), &resolution)
            .expect("visible hybrid tool should delegate plan resolution");
        dispatcher
            .validate_resolved_execution_plan(call, &resolution, &plan)
            .expect("hybrid-selected advertised mode must validate");
        let ResolvedExecutionKind::Detached(policy) = plan.kind() else {
            panic!("hybrid resolver should select its non-default detached mode");
        };
        assert_eq!(policy.runner().name(), "filtered-hybrid-owner");

        let outcome = dispatcher
            .dispatch_resolved_with_context(call, &ToolDispatchContext::default(), &plan)
            .await
            .expect("visible hybrid tool should preserve resolved dispatch");
        let payload: serde_json::Value =
            serde_json::from_str(&outcome.result.text_content()).unwrap();
        assert_eq!(payload["owner"], "filtered-hybrid-owner");
    }

    #[test]
    fn root_validation_rejects_plan_outside_live_advertised_contract() {
        use crate::{
            DetachedToolExecutionPolicy, IdempotencyScope, RestartClass, RunnerIdentity,
            ToolDeadlineChain, ToolDeadlineContributor, ToolDeadlineOwner, ToolExecutionContract,
            ToolExecutionContractError, ToolExecutionMode, ToolExecutionResolutionContext,
            ToolExecutionResolutionError,
        };
        use std::collections::BTreeSet;
        use std::time::Duration;

        let dispatcher = ExactExecutionDispatcher {
            catalog: Arc::from([crate::ToolCatalogEntry::session_inline(
                Arc::new(ToolDef::new(
                    "fast_only",
                    "fast only",
                    json!({"type": "object"}),
                )),
                true,
            )]),
        };
        let detached = DetachedToolExecutionPolicy::new(
            RunnerIdentity::new("dishonest.owner", "v1").unwrap(),
            RestartClass::NonResumable,
            IdempotencyScope::ToolCall,
            Duration::from_secs(10),
        )
        .unwrap();
        let dishonest_contract = ToolExecutionContract::new(
            BTreeSet::from([ToolExecutionMode::Detached]),
            ToolExecutionMode::Detached,
            None,
            Some(detached),
        )
        .unwrap();
        let resolution = ToolExecutionResolutionContext::new(
            ToolDeadlineChain::new(vec![ToolDeadlineContributor::finite(
                ToolDeadlineOwner::CoreToolDispatch,
                Duration::from_secs(600),
            )])
            .unwrap(),
        );
        let plan = dishonest_contract
            .resolve_default(resolution.deadlines().clone())
            .unwrap();
        let args = serde_json::value::RawValue::from_string("{}".to_string()).unwrap();
        let call = ToolCallView {
            id: "dishonest-plan",
            name: "fast_only",
            args: &args,
        };

        assert_eq!(
            dispatcher.validate_resolved_execution_plan(call, &resolution, &plan),
            Err(ToolExecutionResolutionError::Contract(
                ToolExecutionContractError::RequestedModeUnsupported {
                    requested_mode: ToolExecutionMode::Detached,
                }
            ))
        );
    }

    #[tokio::test]
    async fn universal_root_fence_accepts_rebuilt_equivalent_catalog_arcs() {
        use crate::{
            ToolDeadlineChain, ToolDeadlineContributor, ToolDeadlineOwner,
            ToolExecutionResolutionContext,
        };
        use std::time::Duration;

        let dispatcher: Arc<dyn AgentToolDispatcher> = Arc::new(IdenticalMutationDispatcher {
            tool: ToolDef::new("rebuilt", "rebuilt", json!({"type": "object"})),
            epoch: std::sync::atomic::AtomicU64::new(0),
            mutate_on_resolve: false,
        });
        let args = serde_json::value::RawValue::from_string("{}".to_string()).unwrap();
        let call = ToolCallView {
            id: "rebuilt-arcs",
            name: "rebuilt",
            args: &args,
        };
        let resolution = ToolExecutionResolutionContext::new(
            ToolDeadlineChain::new(vec![ToolDeadlineContributor::finite(
                ToolDeadlineOwner::CoreToolDispatch,
                Duration::from_secs(600),
            )])
            .unwrap(),
        );

        let plan = crate::resolve_tool_execution_plan_fenced(
            &dispatcher,
            call,
            &ToolDispatchContext::default(),
            &resolution,
        )
        .expect("equivalent rebuilt catalog projections resolve");
        crate::dispatch_tool_execution_plan_fenced(
            &dispatcher,
            call,
            &ToolDispatchContext::default(),
            &plan,
        )
        .await
        .expect("equivalent rebuilt catalog projections dispatch");
    }

    #[tokio::test]
    async fn universal_root_fence_binds_canonical_call_identity() {
        use crate::{
            ToolDeadlineChain, ToolDeadlineContributor, ToolDeadlineOwner,
            ToolExecutionResolutionContext, ToolUnavailableReason,
        };
        use std::time::Duration;

        let dispatcher: Arc<dyn AgentToolDispatcher> = Arc::new(IdenticalMutationDispatcher {
            tool: ToolDef::new("bound", "bound", json!({"type": "object"})),
            epoch: std::sync::atomic::AtomicU64::new(0),
            mutate_on_resolve: false,
        });
        let resolved_args =
            serde_json::value::RawValue::from_string(r#"{"a":1,"b":2}"#.to_string()).unwrap();
        let equivalent_args =
            serde_json::value::RawValue::from_string(r#"{ "b": 2, "a": 1 }"#.to_string()).unwrap();
        let changed_args =
            serde_json::value::RawValue::from_string(r#"{"a":1,"b":3}"#.to_string()).unwrap();
        let resolved_call = ToolCallView {
            id: "bound-call",
            name: "bound",
            args: &resolved_args,
        };
        let resolution = ToolExecutionResolutionContext::new(
            ToolDeadlineChain::new(vec![ToolDeadlineContributor::finite(
                ToolDeadlineOwner::CoreToolDispatch,
                Duration::from_secs(600),
            )])
            .unwrap(),
        );
        let plan = crate::resolve_tool_execution_plan_fenced(
            &dispatcher,
            resolved_call,
            &ToolDispatchContext::default(),
            &resolution,
        )
        .unwrap();

        crate::dispatch_tool_execution_plan_fenced(
            &dispatcher,
            ToolCallView {
                args: &equivalent_args,
                ..resolved_call
            },
            &ToolDispatchContext::default(),
            &plan,
        )
        .await
        .expect("canonical JSON-equivalent arguments preserve call identity");

        let error = crate::dispatch_tool_execution_plan_fenced(
            &dispatcher,
            ToolCallView {
                args: &changed_args,
                ..resolved_call
            },
            &ToolDispatchContext::default(),
            &plan,
        )
        .await
        .expect_err("different arguments must not dispatch under the old plan");
        assert!(matches!(
            error,
            crate::ToolError::Unavailable {
                reason: ToolUnavailableReason::ExecutionOwnerChanged,
                ..
            }
        ));
    }

    #[tokio::test]
    async fn universal_root_fence_rejects_fresh_dispatcher_reconstruction() {
        use crate::{
            ToolDeadlineChain, ToolDeadlineContributor, ToolDeadlineOwner,
            ToolExecutionResolutionContext, ToolUnavailableReason,
        };
        use std::time::Duration;

        let make_dispatcher = || -> Arc<dyn AgentToolDispatcher> {
            Arc::new(IdenticalMutationDispatcher {
                tool: ToolDef::new("bound", "bound", json!({"type": "object"})),
                epoch: std::sync::atomic::AtomicU64::new(0),
                mutate_on_resolve: false,
            })
        };
        let original = make_dispatcher();
        let args = serde_json::value::RawValue::from_string("{}".to_string()).unwrap();
        let call = ToolCallView {
            id: "reconstructed",
            name: "bound",
            args: &args,
        };
        let resolution = ToolExecutionResolutionContext::new(
            ToolDeadlineChain::new(vec![ToolDeadlineContributor::finite(
                ToolDeadlineOwner::CoreToolDispatch,
                Duration::from_secs(600),
            )])
            .unwrap(),
        );
        let plan = crate::resolve_tool_execution_plan_fenced(
            &original,
            call,
            &ToolDispatchContext::default(),
            &resolution,
        )
        .unwrap();
        let reconstructed = make_dispatcher();

        let error = crate::dispatch_tool_execution_plan_fenced(
            &reconstructed,
            call,
            &ToolDispatchContext::default(),
            &plan,
        )
        .await
        .expect_err("fresh reconstruction must never reproduce ephemeral root authority");
        assert!(matches!(
            error,
            crate::ToolError::Unavailable {
                reason: ToolUnavailableReason::ExecutionOwnerChanged,
                ..
            }
        ));
    }

    #[test]
    fn universal_root_fence_rejects_direct_identical_metadata_replacement() {
        use crate::{
            ToolDeadlineChain, ToolDeadlineContributor, ToolDeadlineOwner,
            ToolExecutionResolutionContext, ToolExecutionResolutionError, ToolUnavailableReason,
        };
        use std::time::Duration;

        let dispatcher: Arc<dyn AgentToolDispatcher> = Arc::new(IdenticalMutationDispatcher {
            tool: ToolDef::new("moving", "identical metadata", json!({"type": "object"})),
            epoch: std::sync::atomic::AtomicU64::new(0),
            mutate_on_resolve: true,
        });
        let args = serde_json::value::RawValue::from_string("{}".to_string()).unwrap();
        let call = ToolCallView {
            id: "direct-identical-replacement",
            name: "moving",
            args: &args,
        };
        let resolution = ToolExecutionResolutionContext::new(
            ToolDeadlineChain::new(vec![ToolDeadlineContributor::finite(
                ToolDeadlineOwner::CoreToolDispatch,
                Duration::from_secs(600),
            )])
            .unwrap(),
        );

        assert!(matches!(
            crate::resolve_tool_execution_plan_fenced(
                &dispatcher,
                call,
                &ToolDispatchContext::default(),
                &resolution,
            ),
            Err(ToolExecutionResolutionError::Unavailable {
                reason: ToolUnavailableReason::ExecutionOwnerChanged,
                ..
            })
        ));
    }

    #[test]
    fn filtered_wrapper_composes_inner_live_binding_epoch() {
        use crate::{
            ToolDeadlineChain, ToolDeadlineContributor, ToolDeadlineOwner,
            ToolExecutionResolutionContext, ToolExecutionResolutionError, ToolUnavailableReason,
        };
        use std::time::Duration;

        let dispatcher: Arc<dyn AgentToolDispatcher> = Arc::new(FilteredToolDispatcher::new(
            Arc::new(IdenticalMutationDispatcher {
                tool: ToolDef::new("moving", "identical metadata", json!({"type": "object"})),
                epoch: std::sync::atomic::AtomicU64::new(0),
                mutate_on_resolve: true,
            }),
            ["moving"],
        ));
        let args = serde_json::value::RawValue::from_string("{}".to_string()).unwrap();
        let call = ToolCallView {
            id: "filtered-identical-replacement",
            name: "moving",
            args: &args,
        };
        let resolution = ToolExecutionResolutionContext::new(
            ToolDeadlineChain::new(vec![ToolDeadlineContributor::finite(
                ToolDeadlineOwner::CoreToolDispatch,
                Duration::from_secs(600),
            )])
            .unwrap(),
        );

        assert!(matches!(
            crate::resolve_tool_execution_plan_fenced(
                &dispatcher,
                call,
                &ToolDispatchContext::default(),
                &resolution,
            ),
            Err(ToolExecutionResolutionError::Unavailable {
                reason: ToolUnavailableReason::ExecutionOwnerChanged,
                ..
            })
        ));
    }

    #[test]
    fn test_inline_peer_notification_policy_from_raw() {
        assert_eq!(
            InlinePeerNotificationPolicy::try_from_raw(None),
            Ok(InlinePeerNotificationPolicy::AtMost(
                DEFAULT_MAX_INLINE_PEER_NOTIFICATIONS
            ))
        );
        assert_eq!(
            InlinePeerNotificationPolicy::try_from_raw(Some(-1)),
            Ok(InlinePeerNotificationPolicy::Always)
        );
        assert_eq!(
            InlinePeerNotificationPolicy::try_from_raw(Some(0)),
            Ok(InlinePeerNotificationPolicy::Never)
        );
        assert_eq!(
            InlinePeerNotificationPolicy::try_from_raw(Some(25)),
            Ok(InlinePeerNotificationPolicy::AtMost(25))
        );
        assert_eq!(
            InlinePeerNotificationPolicy::try_from_raw(Some(-42)),
            Err(-42)
        );
    }

    /// UNIT-002: DetachedOpCompletion serializes without operation_id.
    /// The app-facing control noun is job_id (CONTRACT-003).
    #[test]
    fn unit_002_detached_op_completion_has_no_operation_id() {
        use crate::agent::DetachedOpCompletion;
        use crate::ops_lifecycle::{OperationKind, OperationStatus};

        let completion = DetachedOpCompletion {
            job_id: "j_test".into(),
            kind: OperationKind::BackgroundToolOp,
            status: OperationStatus::Completed,
            terminal_outcome: None,
            display_name: "test cmd".into(),
            detail: "ok".into(),
            elapsed_ms: None,
        };
        #[allow(clippy::unwrap_used)]
        let json = serde_json::to_value(&completion).unwrap();
        assert!(
            json.get("operation_id").is_none(),
            "operation_id must not appear in serialized DetachedOpCompletion (CONTRACT-003)"
        );
        assert!(
            json.get("job_id").is_some(),
            "job_id must be the app-facing control noun"
        );
    }
}