harn-serve 0.10.71

use std::collections::{BTreeMap, VecDeque};
use std::fmt;
use std::path::Path;
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
use std::time::Instant;

use async_trait::async_trait;
use harn_vm::agent_events::{AgentEvent, AgentEventSink};
use harn_vm::event_log::{
    active_event_log, install_active_event_log, install_default_for_base_dir, AnyEventLog,
};
use harn_vm::llm::vm_value_to_json;
use harn_vm::mcp_progress::ProgressContext;
use harn_vm::trust_graph::{append_trust_record, TrustOutcome, TrustRecord};
use harn_vm::{inject_leading_authority, ActorChain, TenantId, TraceId, Vm, VmValue};
use tokio::task::LocalSet;
use tracing::Instrument;

use crate::auth::{AuthPolicy, AuthRequest, AuthenticatedPrincipal, AuthorizationDecision};
use crate::limits::{LimitContext, LimitDecision, LimitGuard, LimitRegistry};
use crate::replay::{InMemoryReplayCache, ReplayCache, ReplayCacheEntry, ReplayKey};
use crate::{BudgetSpec, DispatchError, ExportCatalog, ExportedCallableKind};

mod config;
pub use config::DispatchCoreConfig;

struct ActiveEventLogGuard {
    previous: Option<Arc<AnyEventLog>>,
}

impl Drop for ActiveEventLogGuard {
    fn drop(&mut self) {
        match self.previous.take() {
            Some(log) => {
                install_active_event_log(log);
            }
            None => {
                harn_vm::event_log::reset_active_event_log();
            }
        }
    }
}

fn install_scoped_event_log(log: Arc<AnyEventLog>) -> ActiveEventLogGuard {
    let previous = active_event_log();
    install_active_event_log(log);
    ActiveEventLogGuard { previous }
}

fn install_dispatch_vm_runtime(
    vm: &mut Vm,
    script_path: &Path,
    source: &str,
    cancel_token: Arc<AtomicBool>,
) {
    harn_vm::register_vm_stdlib(vm);
    #[cfg(feature = "hostlib")]
    {
        let _ = harn_hostlib::install_default(vm);
    }
    let store_base = script_path.parent().unwrap_or(Path::new("."));
    harn_vm::register_store_builtins(vm, store_base);
    harn_vm::register_metadata_builtins(vm, store_base);
    vm.set_source_info(&script_path.display().to_string(), source);
    vm.set_source_dir(store_base);
    vm.install_cancel_token(cancel_token);
    vm.set_harness(harn_vm::Harness::real());
}

/// Translate a VM-level error into the dispatcher's typed error.
///
/// Three signals get hoisted out of `Generic` so adapters can render
/// each correctly:
///
/// * `ErrorCategory::Cancelled` — caller-initiated cancel (HTTP 499).
/// * `ErrorCategory::BudgetExceeded` — a `@budget(...)` ceiling fired
///   (HTTP 429, `code = "budget_exceeded"`).
/// * everything else → `Execution` (HTTP 500).
fn classify_vm_error(error: harn_vm::VmError) -> DispatchError {
    let category = harn_vm::error_to_category(&error);
    let message = error.to_string();
    match category {
        harn_vm::ErrorCategory::Cancelled => DispatchError::Cancelled(message),
        harn_vm::ErrorCategory::BudgetExceeded => DispatchError::BudgetExceeded {
            category: budget_category_from_error(&error)
                .unwrap_or_else(|| "llm_cost_usd".to_string()),
            message,
        },
        _ => DispatchError::Execution(message),
    }
}

/// Best-effort attempt to recover the specific budget dimension that
/// fired (one of `llm_cost_usd`, `llm_tokens`, `mcp_calls`,
/// `pg_queries`) from a `VmError` so per-class rejection telemetry stays
/// accurate. The structured form (`VmError::Thrown(Dict)` — the
/// preflight LLM check and the mcp/pg call-count guards) carries it as
/// the `limit` field. The LLM cost/token guards raise the categorised
/// mid-call variant instead, where we disambiguate on the message.
fn budget_category_from_error(error: &harn_vm::VmError) -> Option<String> {
    match error {
        harn_vm::VmError::Thrown(harn_vm::VmValue::Dict(d)) => d
            .get("limit")
            .map(|value| value.display())
            .filter(|s| !s.is_empty()),
        harn_vm::VmError::CategorizedError { message, .. } if message.contains("LLM") => {
            if message.contains("token") {
                Some("llm_tokens".to_string())
            } else {
                Some("llm_cost_usd".to_string())
            }
        }
        _ => None,
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum CallArguments {
    Named(BTreeMap<String, serde_json::Value>),
    Positional(Vec<serde_json::Value>),
}

#[derive(Clone, Debug)]
pub struct CallRequest {
    pub adapter: String,
    pub function: String,
    pub arguments: CallArguments,
    pub auth: AuthRequest,
    pub caller: String,
    /// Stable identity for an adapter-level retry of the same logical request.
    /// `None` disables replay; a supplied key must be unique within the
    /// configured replay cache and ordinary repeated calls must not share one.
    pub replay_key: Option<String>,
    pub trace_id: Option<TraceId>,
    pub parent_span_id: Option<String>,
    pub metadata: BTreeMap<String, serde_json::Value>,
    pub cancel_token: Option<Arc<AtomicBool>>,
    /// Agent-session id to enter for the duration of the dispatch.
    /// When set, `invoke_function` / `invoke_pipeline` push this id
    /// onto the thread-local agent-session stack so worker lifecycle
    /// events fire under it. Adapters use this to scope an
    /// `AgentEventSink` to the request (e.g. A2A maps `task.id` to a
    /// session id and publishes worker/progress updates onto the task
    /// event stream).
    pub agent_session_id: Option<String>,
    /// Optional request-local event sink for live transport streams.
    ///
    /// Unlike the process-global `harn_vm::agent_events` registry, this sink is
    /// installed only for the active dispatch and is filtered to
    /// `agent_session_id`. That keeps long-lived transports such as A2A SSE
    /// streams from losing in-flight events when sibling tests or embedders
    /// reset global Harn VM state.
    pub agent_event_sink: Option<DispatchAgentEventSink>,
    /// Actor chain to bind to the active agent session for this
    /// dispatch. When unset, `DispatchCore` derives the origin from the
    /// authenticated principal after admission.
    pub actor_chain: Option<ActorChain>,
    /// Deterministic local actor to push onto the resolved chain for
    /// adapters that dispatch through a named agent hop.
    pub actor_chain_hop: Option<String>,
    /// Optional progress context — when supplied, the dispatched
    /// function can call the `mcp_report_progress` builtin to emit
    /// `notifications/progress` for the bound `progressToken`. Only
    /// the MCP transport adapter populates this today; other adapters
    /// leave it `None` and the builtin is a no-op.
    pub progress: Option<ProgressContext>,
    /// Tenant the adapter wants this dispatch to run under, overriding
    /// whatever `AuthPolicy` resolves from the credential. Set this
    /// when the transport already owns tenant resolution (e.g. an
    /// upstream cloud gateway that mapped the API key to a tenant in
    /// its own store before forwarding the call). When `None`, the
    /// tenant is sourced from the authenticated principal.
    pub tenant_id: Option<TenantId>,
    /// Request id pushed onto the ambient observability scope for the
    /// dispatched `.harn` callee. The HTTP/ACP/MCP/A2A adapters mint
    /// one per ingress (honouring `X-Request-Id` when present, falling
    /// back to [`crate::http_codec::fresh_request_id`]) so that every
    /// span/log/metric emitted under the dispatch carries the same id
    /// and the standard error envelope (A.4) round-trips it back to
    /// the caller. `None` for tests / in-process callers with no
    /// ingress to mint against.
    pub request_id: Option<String>,
    /// Opaque embedder auth context resolved at admission (e.g. by a
    /// [`crate::SiteAuth`] hook): the API-key record, session claims,
    /// or whatever else the embedder's host-call bridge needs to see
    /// for this request. harn-serve never interprets it; `invoke_*`
    /// installs it as an ambient scope on the VM thread so a
    /// [`harn_vm::HostCallBridge`] can recover it via
    /// [`crate::current_auth_context`] for the duration of the
    /// dispatch. `None` (the default) installs nothing.
    pub auth_context: Option<serde_json::Value>,
    /// Authenticated principal resolved at admission — subject, scheme,
    /// granted scopes, and optional principal kind. Unlike
    /// [`Self::auth_context`] (the opaque embedder blob surfaced only to
    /// the host-call bridge), this is the generic identity harn-serve
    /// itself vouches for; `invoke_*` installs it as the ambient
    /// `harness.auth` handle (see [`harn_vm::enter_auth_principal`]) so a
    /// `.harn` route can read scopes/subject/kind and compose its own
    /// authorization policy. `None` (the default) leaves the dispatch
    /// unauthenticated (`harness.auth.is_authenticated()` is `false`).
    pub auth_principal: Option<harn_vm::AuthPrincipal>,
}

#[derive(Clone)]
pub struct DispatchAgentEventSink {
    inner: Arc<dyn AgentEventSink>,
}

impl DispatchAgentEventSink {
    pub fn new(inner: Arc<dyn AgentEventSink>) -> Self {
        Self { inner }
    }
}

impl fmt::Debug for DispatchAgentEventSink {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("DispatchAgentEventSink(..)")
    }
}

struct SessionScopedAgentEventSink {
    session_id: String,
    inner: Arc<dyn AgentEventSink>,
}

impl AgentEventSink for SessionScopedAgentEventSink {
    fn handle_event(&self, event: &AgentEvent) {
        if event.session_id() != self.session_id {
            return;
        }
        if harn_vm::agent_events::session_has_external_sink(&self.session_id, &self.inner) {
            return;
        }
        self.inner.handle_event(event);
    }
}

fn request_event_sink(request: &CallRequest) -> Option<Arc<dyn AgentEventSink>> {
    let session_id = request.agent_session_id.as_ref()?;
    let sink = request.agent_event_sink.as_ref()?;
    Some(Arc::new(SessionScopedAgentEventSink {
        session_id: session_id.clone(),
        inner: sink.inner.clone(),
    }))
}

fn resolve_request_actor_chain(
    request: &CallRequest,
    principal: &AuthenticatedPrincipal,
) -> Option<ActorChain> {
    let mut chain = request.actor_chain.clone().or_else(|| {
        request
            .auth_principal
            .as_ref()
            .map(|principal| principal.subject.trim())
            .filter(|subject| !subject.is_empty())
            .map(ActorChain::new)
            .or_else(|| {
                let subject = principal.subject.trim();
                (!subject.is_empty()).then(|| ActorChain::new(subject))
            })
    })?;
    if let Some(actor) = request
        .actor_chain_hop
        .as_deref()
        .map(str::trim)
        .filter(|actor| !actor.is_empty())
    {
        chain.push(actor);
    }
    Some(chain)
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CallResponse {
    pub function: String,
    pub value: serde_json::Value,
    pub printed_output: String,
    pub trace_id: TraceId,
    pub cached: bool,
    pub duration_ms: u128,
}

#[async_trait(?Send)]
pub trait VmConfigurator: Send + Sync {
    fn configure(&self, _vm: &mut Vm) -> Result<(), DispatchError> {
        Ok(())
    }
}

#[derive(Clone, Default)]
pub struct NoopVmConfigurator;

#[async_trait(?Send)]
impl VmConfigurator for NoopVmConfigurator {}

pub struct DispatchCore {
    config: DispatchCoreConfig,
    catalog: ExportCatalog,
    event_log: Arc<harn_vm::event_log::AnyEventLog>,
}

impl DispatchCore {
    pub fn new(config: DispatchCoreConfig) -> Result<Self, DispatchError> {
        let catalog = ExportCatalog::from_path(&config.script_path)?;
        let event_log = install_default_for_base_dir(&config.base_dir).map_err(|error| {
            DispatchError::Io(format!(
                "failed to initialize event log for {}: {error}",
                config.base_dir.display()
            ))
        })?;
        Ok(Self {
            config,
            catalog,
            event_log,
        })
    }

    pub fn catalog(&self) -> &ExportCatalog {
        &self.catalog
    }

    pub fn auth_policy(&self) -> &AuthPolicy {
        &self.config.auth_policy
    }

    pub(crate) fn event_log(&self) -> Arc<AnyEventLog> {
        self.event_log.clone()
    }

    pub async fn dispatch(&self, mut request: CallRequest) -> Result<CallResponse, DispatchError> {
        let trace_id = request.trace_id.clone().unwrap_or_default();
        let function_scopes = self
            .catalog
            .function(&request.function)
            .map(|function| function.required_scopes.clone())
            .unwrap_or_default();
        let authorization = self
            .config
            .auth_policy
            .authorize_with_scopes(&request.auth, &function_scopes)
            .await;
        match authorization {
            AuthorizationDecision::Authorized(principal) => {
                // Adapter-supplied tenants override; otherwise the
                // authenticated principal's tenant wins. Resolving here
                // (not later inside `invoke_*`) keeps trust records and
                // span attributes consistent with the value the .harn
                // callee actually sees.
                if request.tenant_id.is_none() {
                    request.tenant_id = principal.tenant_id.clone();
                }
                // Surface the same authenticated identity to the `.harn`
                // callee as the ambient `harness.auth` handle. An adapter
                // that already resolved the principal (e.g. the site
                // adapter's `SiteAuth` hook) wins; otherwise project the
                // policy-resolved principal. The synthetic anonymous
                // principal (allow-all, no credential) binds nothing, so a
                // `.harn` route reads `harness.auth.is_authenticated() ==
                // false`. The AuthPolicy path carries no embedder-assigned
                // `kind`, so it stays `None`.
                if request.auth_principal.is_none() && !principal.is_anonymous() {
                    request.auth_principal = Some(harn_vm::AuthPrincipal {
                        subject: principal.subject.clone(),
                        scheme: principal.scheme.clone(),
                        scopes: principal.granted_scopes.clone(),
                        kind: None,
                    });
                }
                request.actor_chain = resolve_request_actor_chain(&request, &principal);
            }
            AuthorizationDecision::Rejected(message) => {
                self.record_trust(
                    &request,
                    &trace_id,
                    TrustOutcome::Denied,
                    Some(message.clone()),
                )
                .await?;
                return Err(DispatchError::Unauthorized(message));
            }
            AuthorizationDecision::MissingScope { required, granted } => {
                let error = DispatchError::Forbidden { required, granted };
                self.record_trust(
                    &request,
                    &trace_id,
                    TrustOutcome::Denied,
                    Some(error.message()),
                )
                .await?;
                return Err(error);
            }
            // MCP allowlist checks are enforced at the `harness.mcp.*`
            // dispatch boundary inside harn-vm, not on the HTTP edge;
            // surfacing the variant here would mean the policy was
            // queried with a server/tool pair, which the HTTP dispatch
            // path never does. Treat any leak as a policy bug.
            AuthorizationDecision::McpNotAllowlisted { reason, .. } => {
                self.record_trust(
                    &request,
                    &trace_id,
                    TrustOutcome::Denied,
                    Some(reason.clone()),
                )
                .await?;
                return Err(DispatchError::Unauthorized(reason));
            }
        }

        let function = self.catalog.function(&request.function).ok_or_else(|| {
            DispatchError::MissingExport(format!(
                "function '{}' is not exported by {}",
                request.function,
                self.catalog.script_path.display()
            ))
        })?;

        // Rate-limit + backpressure gate. Every dispatch attempt passes this
        // gate before replay lookup. The in-flight counter decrements when the
        // guard drops, including on cached replies and panics.
        let _limit_guard = self.check_limits(&request, function)?;

        let replay_key = request.replay_key.clone().map(ReplayKey);
        if let Some(key) = replay_key.as_ref() {
            if let Some(cached) = self.config.replay_cache.get(key).await? {
                return Ok(CallResponse {
                    function: request.function.clone(),
                    value: cached.value,
                    printed_output: cached.printed_output,
                    trace_id,
                    cached: true,
                    duration_ms: 0,
                });
            }
        }

        // Per-dispatch resource budget caps live on `function.budget`
        // and are installed inside `invoke_function` / `invoke_pipeline`
        // — the thread-local backing (`BudgetSpec::install`)
        // must be set on the same OS thread the VM runs on, which the
        // tokio `LocalSet` inside each invoker pins.

        // tenant_id is a low-cardinality routing key (one entry per
        // tenant), not PII — safe to record as a span attribute so
        // exporters can filter traces by tenant. `Empty` until populated
        // so the absent case isn't recorded as the literal string
        // `"None"`. Recorded once after the span opens, mirroring how
        // OTEL bindings expect span attributes to be set.
        let span = tracing::info_span!(
            target: "harn.serve",
            "harn_serve.dispatch",
            adapter = %request.adapter,
            function = %request.function,
            caller = %request.caller,
            trace_id = %trace_id.0,
            tenant_id = tracing::field::Empty,
        );
        if let Some(tenant) = request.tenant_id.as_ref() {
            span.record("tenant_id", tenant.0.as_str());
        }
        let _ = harn_vm::observability::otel::set_span_parent(
            &span,
            &trace_id,
            request.parent_span_id.as_deref(),
        );

        let started = Instant::now();
        let invocation = async {
            let value = match function.kind {
                ExportedCallableKind::Function => self.invoke_function(&request, function).await?,
                ExportedCallableKind::Pipeline => self.invoke_pipeline(&request, function).await?,
            };
            Ok::<_, DispatchError>(value)
        }
        .instrument(span)
        .await;

        match invocation {
            Ok((value, printed_output)) => {
                let duration_ms = started.elapsed().as_millis();
                self.record_trust(&request, &trace_id, TrustOutcome::Success, None)
                    .await?;
                if let Some(key) = replay_key {
                    self.config
                        .replay_cache
                        .put(
                            key,
                            ReplayCacheEntry {
                                value: value.clone(),
                                printed_output: printed_output.clone(),
                            },
                        )
                        .await?;
                }
                Ok(CallResponse {
                    function: request.function,
                    value,
                    printed_output,
                    trace_id,
                    cached: false,
                    duration_ms,
                })
            }
            Err(error) => {
                self.record_trust(
                    &request,
                    &trace_id,
                    TrustOutcome::Failure,
                    Some(error.to_string()),
                )
                .await?;
                Err(error)
            }
        }
    }

    /// Consult the rate-limit + backpressure registry for this dispatch.
    /// Returns a guard that decrements the in-flight counter on drop
    /// when the registry admits the call; returns
    /// `DispatchError::RateLimited` otherwise.
    fn check_limits(
        &self,
        request: &CallRequest,
        function: &crate::ExportedFunction,
    ) -> Result<LimitGuard, DispatchError> {
        let Some(registry) = self.config.limit_registry.as_ref() else {
            return Ok(LimitGuard::unbounded_for_caller());
        };
        let Some(limits) = function.limits.as_ref() else {
            return Ok(LimitGuard::unbounded_for_caller());
        };
        let ctx = LimitContext {
            route: &request.function,
            tenant_id: request.tenant_id.as_ref(),
            scopes: &function.required_scopes,
        };
        match registry.check(&ctx, limits) {
            LimitDecision::Allowed(guard) => Ok(guard),
            LimitDecision::Rejected {
                scope,
                retry_after_ms,
            } => Err(DispatchError::RateLimited {
                scope: scope.as_str().to_string(),
                retry_after_ms,
            }),
        }
    }

    async fn invoke_function(
        &self,
        request: &CallRequest,
        function: &crate::ExportedFunction,
    ) -> Result<(serde_json::Value, String), DispatchError> {
        let source = tokio::fs::read_to_string(&self.config.script_path)
            .await
            .map_err(|error| {
                DispatchError::Io(format!(
                    "failed to read {}: {error}",
                    self.config.script_path.display()
                ))
            })?;
        let script_path = self.config.script_path.clone();
        let cancel_token = request
            .cancel_token
            .clone()
            .unwrap_or_else(|| Arc::new(AtomicBool::new(false)));
        let agent_session_id = request.agent_session_id.clone();
        let agent_event_sink = request_event_sink(request);
        let actor_chain = request.actor_chain.clone();
        let progress = request.progress.clone();

        let tenant_id = request.tenant_id.clone();
        let budget = function.budget.clone();
        let request_id = request.request_id.clone();
        let auth_context = request.auth_context.clone();
        let auth_principal = request.auth_principal.clone();
        let local = LocalSet::new();
        local
            .run_until(harn_vm::mcp_progress::scope_context(progress, async move {
                harn_vm::llm::scope_agent_event_sink(agent_event_sink, async move {
                    let _event_log = install_scoped_event_log(self.event_log.clone());
                    let _session_guard = agent_session_id.as_deref().map(|session_id| {
                        harn_vm::agent_sessions::open_or_create_with_actor_chain(
                            Some(session_id.to_string()),
                            actor_chain.clone(),
                        );
                        harn_vm::agent_sessions::enter_current_session(session_id.to_string())
                    });
                    let _tenant_guard = tenant_id.map(harn_vm::enter_tenant);
                    let _budget_guard = budget.as_ref().and_then(BudgetSpec::install);
                    let _request_id_guard = request_id.map(harn_vm::enter_request_id);
                    let _auth_context_guard = auth_context.map(crate::enter_auth_context);
                    let _auth_principal_guard = auth_principal.map(harn_vm::enter_auth_principal);

                    let mut vm = Vm::new();
                    if self.config.trusted_host_dispatch {
                        vm.enable_trusted_host_dispatch()
                            .map_err(classify_vm_error)?;
                    }
                    install_dispatch_vm_runtime(&mut vm, &script_path, &source, cancel_token);
                    self.config.vm_configurator.configure(&mut vm)?;

                    let exports = vm
                        .load_module_exports(&script_path)
                        .await
                        .map_err(|error| DispatchError::Execution(error.to_string()))?;
                    let Some(closure) = exports.get(&request.function) else {
                        return Err(DispatchError::MissingExport(format!(
                            "function '{}' is not exported by {}",
                            request.function,
                            script_path.display()
                        )));
                    };
                    let mut args = inject_leading_authority(
                        &vm,
                        closure,
                        &[],
                        &format!("serve export `{}`", request.function),
                    )
                    .map_err(classify_vm_error)?;
                    let user_args = build_vm_args(&request.arguments, function, !args.is_empty())?;
                    args.extend(user_args);
                    let result = vm.call_closure_pub(closure, &args).await;

                    match result {
                        Ok(value) => Ok((vm_value_to_json(&value), vm.output().to_string())),
                        Err(error) => Err(classify_vm_error(error)),
                    }
                })
                .await
            }))
            .await
    }

    async fn invoke_pipeline(
        &self,
        request: &CallRequest,
        function: &crate::ExportedFunction,
    ) -> Result<(serde_json::Value, String), DispatchError> {
        let source = tokio::fs::read_to_string(&self.config.script_path)
            .await
            .map_err(|error| {
                DispatchError::Io(format!(
                    "failed to read {}: {error}",
                    self.config.script_path.display()
                ))
            })?;
        let arguments = request.arguments.clone();
        let function = function.clone();
        let script_path = self.config.script_path.clone();
        let cancel_token = request
            .cancel_token
            .clone()
            .unwrap_or_else(|| Arc::new(AtomicBool::new(false)));
        let agent_session_id = request.agent_session_id.clone();
        let agent_event_sink = request_event_sink(request);
        let actor_chain = request.actor_chain.clone();
        let progress = request.progress.clone();

        let tenant_id = request.tenant_id.clone();
        let budget = function.budget.clone();
        let request_id = request.request_id.clone();
        let auth_context = request.auth_context.clone();
        let auth_principal = request.auth_principal.clone();
        let local = LocalSet::new();
        local
            .run_until(harn_vm::mcp_progress::scope_context(progress, async move {
                harn_vm::llm::scope_agent_event_sink(agent_event_sink, async move {
                    let _event_log = install_scoped_event_log(self.event_log.clone());
                    let _session_guard = agent_session_id.as_deref().map(|session_id| {
                        harn_vm::agent_sessions::open_or_create_with_actor_chain(
                            Some(session_id.to_string()),
                            actor_chain.clone(),
                        );
                        harn_vm::agent_sessions::enter_current_session(session_id.to_string())
                    });
                    let _tenant_guard = tenant_id.map(harn_vm::enter_tenant);
                    let _budget_guard = budget.as_ref().and_then(BudgetSpec::install);
                    let _request_id_guard = request_id.map(harn_vm::enter_request_id);
                    let _auth_context_guard = auth_context.map(crate::enter_auth_context);
                    let _auth_principal_guard = auth_principal.map(harn_vm::enter_auth_principal);

                    let mut vm = Vm::new();
                    if self.config.trusted_host_dispatch {
                        vm.enable_trusted_host_dispatch()
                            .map_err(classify_vm_error)?;
                    }
                    install_dispatch_vm_runtime(&mut vm, &script_path, &source, cancel_token);
                    self.config.vm_configurator.configure(&mut vm)?;
                    let closure = vm
                        .load_module_callable_from_source(&script_path, &source, &function.name)
                        .await
                        .map_err(classify_vm_error)?;
                    let closure = closure
                        .ok_or_else(|| DispatchError::MissingExport(function.name.clone()))?;
                    let mut args = inject_leading_authority(
                        &vm,
                        &closure,
                        &[],
                        &format!("serve pipeline `{}`", function.name),
                    )
                    .map_err(classify_vm_error)?;
                    let user_args = build_vm_args(&arguments, &function, !args.is_empty())?;
                    args.extend(user_args);
                    let result = vm.call_closure_pub(&closure, &args).await;

                    match result {
                        Ok(_) => {
                            let output = vm.output().to_string();
                            Ok((serde_json::Value::String(output.clone()), output))
                        }
                        Err(error) => Err(classify_vm_error(error)),
                    }
                })
                .await
            }))
            .await
    }

    async fn record_trust(
        &self,
        request: &CallRequest,
        trace_id: &TraceId,
        outcome: TrustOutcome,
        error: Option<String>,
    ) -> Result<(), DispatchError> {
        let mut record = TrustRecord::new(
            self.config.service_name.clone(),
            format!("invoke.{}", request.function),
            None,
            outcome,
            trace_id.0.clone(),
            self.config.autonomy_tier,
        );
        record
            .metadata
            .insert("adapter".to_string(), serde_json::json!(request.adapter));
        record
            .metadata
            .insert("caller".to_string(), serde_json::json!(request.caller));
        record
            .metadata
            .insert("function".to_string(), serde_json::json!(request.function));
        if let Some(actor_chain) = request.actor_chain.as_ref() {
            record.set_actor_chain(Some(actor_chain.clone()));
        }
        if let Some(tenant) = request.tenant_id.as_ref() {
            record
                .metadata
                .insert("tenant_id".to_string(), serde_json::json!(tenant.0));
        }
        if let Some(error) = error {
            record
                .metadata
                .insert("error".to_string(), serde_json::json!(error));
        }
        append_trust_record(&self.event_log, &record)
            .await
            .map(|_| ())
            .map_err(|error| {
                DispatchError::Execution(format!("failed to append trust record: {error}"))
            })
    }
}

fn build_vm_args(
    arguments: &CallArguments,
    function: &crate::ExportedFunction,
    has_leading_authority: bool,
) -> Result<Vec<VmValue>, DispatchError> {
    let mut params = function.params.as_slice();
    // Authority-bearing parameters are host-injected instead of JSON-bound.
    // Exclude the leading root or nominal Harness slot from the public request
    // schema; the VM bridge remains the semantic owner of deriving the
    // declared handle from the installed root.
    if has_leading_authority {
        params = &params[1..];
    }

    let rest = match arguments {
        CallArguments::Positional(values) => {
            values.iter().map(json_to_vm_value).collect::<Vec<_>>()
        }
        CallArguments::Named(values) => {
            // Tolerate clients that emit a single-object-param tool's arguments
            // FLAT at the top level instead of nested under the parameter name
            // (harn#5039). See `lift_flat_single_object_arg`.
            let lifted = lift_flat_single_object_arg(params, values);
            let values: &BTreeMap<String, serde_json::Value> = lifted.as_ref().unwrap_or(values);
            let mut args = Vec::new();
            let mut saw_gap = false;
            for param in params {
                let value = values.get(&param.name);
                match value {
                    Some(value) => {
                        if saw_gap {
                            return Err(DispatchError::Validation(format!(
                                "named arguments for '{}' skipped '{}' before later arguments",
                                function.name, param.name
                            )));
                        }
                        args.push(json_to_vm_value(value));
                    }
                    None if param.has_default => {
                        saw_gap = true;
                    }
                    None => {
                        return Err(DispatchError::Validation(format!(
                            "missing required argument '{}' for '{}'",
                            param.name, function.name
                        )));
                    }
                }
            }
            trim_trailing_defaults(args)
        }
    };

    Ok(rest)
}

/// Tolerate MCP/JSON-RPC clients that emit a single-object-parameter tool's
/// arguments FLAT at the top level instead of nested under the parameter name
/// (harn#5039).
///
/// Harn projects `pub fn tool(params: { field: T, .. })` into an MCP
/// `inputSchema` that nests every field under `params`. Some model clients
/// (notably local/open-weight models) ignore that wrapper and emit the inner
/// fields at the top level — e.g. `{ "events_dir": ".." }` instead of
/// `{ "params": { "events_dir": ".." } }`. Bound by name, those flat keys match
/// no parameter and are dropped, so the tool silently runs on its default and
/// reports the field as "required".
///
/// When the function declares exactly one object-typed parameter and the
/// incoming named map both (a) omits that parameter's own name and (b) is
/// non-empty, wrap the whole map as that parameter's value. This is:
/// - **idempotent** — a correctly-nested `{ "params": { .. } }` call already
///   contains the parameter name, so it is passed through untouched;
/// - **strictly additive** — it only rewrites calls whose keys would otherwise
///   all be dropped (the parameter falling back to its default), never a call
///   that already binds the parameter;
/// - **shape-guarded** — it fires only for object-typed parameters, so a scalar
///   single-parameter tool still reports a clear "missing required argument"
///   instead of a spurious type error.
///
/// Returns the rewritten map when the lift applies, else `None` (bind the
/// original arguments unchanged).
fn lift_flat_single_object_arg(
    params: &[crate::ExportedParam],
    values: &BTreeMap<String, serde_json::Value>,
) -> Option<BTreeMap<String, serde_json::Value>> {
    let [only] = params else {
        return None;
    };
    if only.rest || !only.accepts_json_object() {
        return None;
    }
    if values.is_empty() || values.contains_key(&only.name) {
        return None;
    }
    let wrapped = serde_json::Value::Object(values.clone().into_iter().collect());
    Some(BTreeMap::from([(only.name.clone(), wrapped)]))
}

fn trim_trailing_defaults(mut args: Vec<VmValue>) -> Vec<VmValue> {
    let mut tail = VecDeque::from(args);
    while matches!(tail.back(), Some(VmValue::Nil)) {
        tail.pop_back();
    }
    args = tail.into_iter().collect();
    args
}

fn json_to_vm_value(value: &serde_json::Value) -> VmValue {
    match value {
        serde_json::Value::Null => VmValue::Nil,
        serde_json::Value::Bool(value) => VmValue::Bool(*value),
        serde_json::Value::Number(value) => value
            .as_i64()
            .map(VmValue::Int)
            .or_else(|| value.as_f64().map(VmValue::Float))
            .unwrap_or(VmValue::Nil),
        serde_json::Value::String(value) => VmValue::String(arcstr::ArcStr::from(value.as_str())),
        serde_json::Value::Array(items) => VmValue::List(Arc::new(
            items.iter().map(json_to_vm_value).collect::<Vec<_>>(),
        )),
        serde_json::Value::Object(map) => VmValue::dict(
            map.iter()
                .map(|(key, value)| (key.clone(), json_to_vm_value(value)))
                .collect::<harn_vm::value::DictMap>(),
        ),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    #[derive(Default)]
    struct TrackingReplayCache {
        inner: InMemoryReplayCache,
        gets: AtomicUsize,
        puts: AtomicUsize,
    }

    impl TrackingReplayCache {
        fn counts(&self) -> (usize, usize) {
            (
                self.gets.load(Ordering::SeqCst),
                self.puts.load(Ordering::SeqCst),
            )
        }
    }

    #[async_trait]
    impl ReplayCache for TrackingReplayCache {
        async fn get(&self, key: &ReplayKey) -> Result<Option<ReplayCacheEntry>, DispatchError> {
            self.gets.fetch_add(1, Ordering::SeqCst);
            self.inner.get(key).await
        }

        async fn put(&self, key: ReplayKey, value: ReplayCacheEntry) -> Result<(), DispatchError> {
            self.puts.fetch_add(1, Ordering::SeqCst);
            self.inner.put(key, value).await
        }
    }

    struct CountingVmConfigurator {
        calls: Arc<AtomicUsize>,
    }

    impl VmConfigurator for CountingVmConfigurator {
        fn configure(&self, vm: &mut Vm) -> Result<(), DispatchError> {
            let calls = self.calls.clone();
            vm.register_builtin("test_increment_call_count", move |_args, _output| {
                let count = calls.fetch_add(1, Ordering::SeqCst) + 1;
                Ok(VmValue::Int(
                    count.try_into().expect("test call count fits in i64"),
                ))
            });
            Ok(())
        }
    }

    fn replay_test_fixture() -> (
        tempfile::TempDir,
        DispatchCore,
        Arc<AtomicUsize>,
        Arc<TrackingReplayCache>,
    ) {
        let dir = tempfile::tempdir().expect("tempdir");
        let script = dir.path().join("server.harn");
        std::fs::write(
            &script,
            r"
pub fn observe_execution() -> int {
  return test_increment_call_count()
}
",
        )
        .expect("write script");

        let calls = Arc::new(AtomicUsize::new(0));
        let cache = Arc::new(TrackingReplayCache::default());
        let mut config = DispatchCoreConfig::for_script(&script);
        config.replay_cache = cache.clone();
        config.vm_configurator = Arc::new(CountingVmConfigurator {
            calls: calls.clone(),
        });
        let core = DispatchCore::new(config).expect("core");
        (dir, core, calls, cache)
    }

    fn replay_test_request(replay_key: Option<&str>) -> CallRequest {
        CallRequest {
            adapter: "mcp".to_string(),
            function: "observe_execution".to_string(),
            arguments: CallArguments::Named(BTreeMap::new()),
            auth: AuthRequest::default(),
            caller: "tester".to_string(),
            replay_key: replay_key.map(str::to_string),
            trace_id: None,
            parent_span_id: None,
            metadata: BTreeMap::new(),
            cancel_token: None,
            agent_session_id: None,
            agent_event_sink: None,
            actor_chain: None,
            actor_chain_hop: None,
            progress: None,
            tenant_id: None,
            request_id: None,
            auth_context: None,
            auth_principal: None,
        }
    }

    #[tokio::test]
    async fn dispatch_executes_exported_function() {
        let dir = tempfile::tempdir().expect("tempdir");
        let script = dir.path().join("server.harn");
        std::fs::write(
            &script,
            r"
pub fn greet(name: string) -> string {
  return name
}
",
        )
        .expect("write script");

        let core = DispatchCore::new(DispatchCoreConfig::for_script(&script)).expect("core");
        let response = core
            .dispatch(CallRequest {
                adapter: "mcp".to_string(),
                function: "greet".to_string(),
                arguments: CallArguments::Named(BTreeMap::from([(
                    "name".to_string(),
                    serde_json::json!("alice"),
                )])),
                auth: AuthRequest::default(),
                caller: "tester".to_string(),
                replay_key: None,
                trace_id: None,
                parent_span_id: None,
                metadata: BTreeMap::new(),
                cancel_token: None,
                agent_session_id: None,
                agent_event_sink: None,
                actor_chain: None,
                actor_chain_hop: None,
                progress: None,
                tenant_id: None,
                request_id: None,
                auth_context: None,
                auth_principal: None,
            })
            .await
            .expect("dispatch");

        assert_eq!(response.value, serde_json::json!("alice"));
        assert!(!response.cached);
    }

    /// harn#5039: a single-object-param tool (the `pub fn tool(params: {..})`
    /// convention) whose MCP `inputSchema` nests fields under `params` must
    /// still bind when a client emits those fields FLAT at the top level, the
    /// exact 6/6-failing shape the local model produced against
    /// `burin-harness-debugger` (`{events_dir: ".."}` rather than
    /// `{params: {events_dir: ".."}}`). The flat keys must reach the parameter
    /// instead of being dropped so the tool falls back to its default.
    async fn dispatch_echo_params(arguments: CallArguments) -> serde_json::Value {
        let dir = tempfile::tempdir().expect("tempdir");
        let script = dir.path().join("server.harn");
        std::fs::write(
            &script,
            r#"
pub fn stage_triage(
  params: {events_dir: string, verdict_path?: string} = {events_dir: ""},
) -> dict {
  return params
}
"#,
        )
        .expect("write script");

        let core = DispatchCore::new(DispatchCoreConfig::for_script(&script)).expect("core");
        core.dispatch(CallRequest {
            adapter: "mcp".to_string(),
            function: "stage_triage".to_string(),
            arguments,
            auth: AuthRequest::default(),
            caller: "tester".to_string(),
            replay_key: None,
            trace_id: None,
            parent_span_id: None,
            metadata: BTreeMap::new(),
            cancel_token: None,
            agent_session_id: None,
            agent_event_sink: None,
            actor_chain: None,
            actor_chain_hop: None,
            progress: None,
            tenant_id: None,
            request_id: None,
            auth_context: None,
            auth_principal: None,
        })
        .await
        .expect("dispatch")
        .value
    }

    #[tokio::test]
    async fn flat_single_object_arg_binds_like_nested_5039() {
        // FLAT emission (local-model shape) now reaches the `params` parameter.
        let flat = dispatch_echo_params(CallArguments::Named(BTreeMap::from([(
            "events_dir".to_string(),
            serde_json::json!("/runs/20260717-175909"),
        )])))
        .await;
        assert_eq!(
            flat,
            serde_json::json!({ "events_dir": "/runs/20260717-175909" }),
            "flat top-level args must be lifted into the single object parameter",
        );

        // Correctly-nested emission (Claude shape) is untouched — idempotent.
        let nested = dispatch_echo_params(CallArguments::Named(BTreeMap::from([(
            "params".to_string(),
            serde_json::json!({ "events_dir": "/runs/nested" }),
        )])))
        .await;
        assert_eq!(
            nested,
            serde_json::json!({ "events_dir": "/runs/nested" }),
            "a correctly-nested call must not be double-lifted",
        );

        // No arguments falls back to the declared default (no spurious lift).
        let empty = dispatch_echo_params(CallArguments::Named(BTreeMap::new())).await;
        assert_eq!(empty, serde_json::json!({ "events_dir": "" }));
    }

    #[test]
    fn flat_lift_is_scoped_to_single_object_param() {
        use crate::ExportedParam;

        let obj_param = |name: &str| ExportedParam {
            name: name.to_string(),
            type_expr: None,
            input_schema: serde_json::json!({ "type": "object", "properties": {} }),
            has_default: true,
            rest: false,
        };
        let scalar_param = |name: &str| ExportedParam {
            name: name.to_string(),
            type_expr: None,
            input_schema: serde_json::json!({ "type": "string" }),
            has_default: false,
            rest: false,
        };
        let flat = BTreeMap::from([("events_dir".to_string(), serde_json::json!("x"))]);

        // Single object param, flat args, param name absent -> lift.
        let params = [obj_param("params")];
        let lifted = lift_flat_single_object_arg(&params, &flat).expect("lift");
        assert_eq!(lifted["params"], serde_json::json!({ "events_dir": "x" }));

        // Param name already present -> no lift (idempotent).
        let nested = BTreeMap::from([("params".to_string(), serde_json::json!({ "a": 1 }))]);
        assert!(lift_flat_single_object_arg(&params, &nested).is_none());

        // Scalar single param -> no lift (preserves clear "missing required arg").
        assert!(lift_flat_single_object_arg(&[scalar_param("name")], &flat).is_none());

        // Multiple params -> no lift (normal named binding).
        assert!(lift_flat_single_object_arg(&[obj_param("a"), obj_param("b")], &flat).is_none());

        // Empty args -> no lift (default applies).
        assert!(lift_flat_single_object_arg(&params, &BTreeMap::new()).is_none());
    }

    #[cfg(feature = "hostlib")]
    #[tokio::test]
    async fn dispatch_exported_function_can_use_deterministic_tools_hostlib() {
        let dir = tempfile::tempdir().expect("tempdir");
        let script = dir.path().join("server.harn");
        std::fs::write(
            &script,
            r#"
import { command_run } from "std/command"

pub fn run_help(harness: Harness, binary: string) -> int {
  const result = command_run(
    harness.tools,
    {argv: [binary, "--help"]},
    {capture: {max_inline_bytes: 256}, timeout_ms: 5000},
  )
  return result.exit_code
}
"#,
        )
        .expect("write script");

        let core = DispatchCore::new(DispatchCoreConfig::for_script(&script)).expect("core");
        let response = core
            .dispatch(CallRequest {
                adapter: "mcp".to_string(),
                function: "run_help".to_string(),
                arguments: CallArguments::Named(BTreeMap::from([(
                    "binary".to_string(),
                    serde_json::json!(std::env::current_exe()
                        .expect("current executable")
                        .to_string_lossy()),
                )])),
                auth: AuthRequest::default(),
                caller: "tester".to_string(),
                replay_key: None,
                trace_id: None,
                parent_span_id: None,
                metadata: BTreeMap::new(),
                cancel_token: None,
                agent_session_id: None,
                agent_event_sink: None,
                actor_chain: None,
                actor_chain_hop: None,
                progress: None,
                tenant_id: None,
                request_id: None,
                auth_context: None,
                auth_principal: None,
            })
            .await
            .expect("dispatch");

        assert_eq!(response.value, serde_json::json!(0));
    }

    #[tokio::test]
    async fn dispatch_executes_legacy_pipeline_when_no_public_exports() {
        let dir = tempfile::tempdir().expect("tempdir");
        let script = dir.path().join("server.harn");
        std::fs::write(
            &script,
            r"
pipeline default(harness: Harness, task) {
  harness.stdio.println(json_stringify({task: task}))
}
",
        )
        .expect("write script");

        let core = DispatchCore::new(DispatchCoreConfig::for_script(&script)).expect("core");
        let response = core
            .dispatch(CallRequest {
                adapter: "a2a".to_string(),
                function: "default".to_string(),
                arguments: CallArguments::Named(BTreeMap::from([(
                    "task".to_string(),
                    serde_json::json!("payload"),
                )])),
                auth: AuthRequest::default(),
                caller: "tester".to_string(),
                replay_key: None,
                trace_id: None,
                parent_span_id: None,
                metadata: BTreeMap::new(),
                cancel_token: None,
                agent_session_id: None,
                agent_event_sink: None,
                actor_chain: None,
                actor_chain_hop: None,
                progress: None,
                tenant_id: None,
                request_id: None,
                auth_context: None,
                auth_principal: None,
            })
            .await
            .expect("dispatch");

        assert_eq!(
            response.value,
            serde_json::json!("{\"task\":\"payload\"}\n")
        );
        assert_eq!(response.printed_output, "{\"task\":\"payload\"}\n");
    }

    #[tokio::test]
    async fn dispatch_without_replay_key_executes_each_request_without_cache_access() {
        let (_dir, core, calls, cache) = replay_test_fixture();

        let first = core
            .dispatch(replay_test_request(None))
            .await
            .expect("first dispatch");
        let second = core
            .dispatch(replay_test_request(None))
            .await
            .expect("second dispatch");

        assert_eq!(
            [first.value, second.value],
            [serde_json::json!(1), serde_json::json!(2)]
        );
        assert_eq!([first.cached, second.cached], [false, false]);
        assert_eq!(calls.load(Ordering::SeqCst), 2);
        assert_eq!(cache.counts(), (0, 0));
    }

    #[tokio::test]
    async fn dispatch_with_same_explicit_replay_key_executes_once_and_replays_once() {
        let (_dir, core, calls, cache) = replay_test_fixture();

        let first = core
            .dispatch(replay_test_request(Some("fixed-key")))
            .await
            .expect("first dispatch");
        let second = core
            .dispatch(replay_test_request(Some("fixed-key")))
            .await
            .expect("second dispatch");

        assert_eq!(
            [first.value, second.value],
            [serde_json::json!(1), serde_json::json!(1)]
        );
        assert_eq!([first.cached, second.cached], [false, true]);
        assert_eq!(calls.load(Ordering::SeqCst), 1);
        assert_eq!(cache.counts(), (2, 1));
    }

    #[tokio::test]
    async fn dispatch_records_trust_graph_events() {
        let dir = tempfile::tempdir().expect("tempdir");
        let script = dir.path().join("server.harn");
        std::fs::write(
            &script,
            r"
pub fn greet(name: string) -> string {
  return name
}
",
        )
        .expect("write script");

        let core = DispatchCore::new(DispatchCoreConfig::for_script(&script)).expect("core");
        let response = core
            .dispatch(CallRequest {
                adapter: "mcp".to_string(),
                function: "greet".to_string(),
                arguments: CallArguments::Named(BTreeMap::from([(
                    "name".to_string(),
                    serde_json::json!("alice"),
                )])),
                auth: AuthRequest::default(),
                caller: "tester".to_string(),
                replay_key: Some("trust-key".to_string()),
                trace_id: None,
                parent_span_id: None,
                metadata: BTreeMap::new(),
                cancel_token: None,
                agent_session_id: None,
                agent_event_sink: None,
                actor_chain: None,
                actor_chain_hop: None,
                progress: None,
                tenant_id: None,
                request_id: None,
                auth_context: None,
                auth_principal: None,
            })
            .await
            .expect("dispatch");

        let records =
            harn_vm::query_trust_records(&core.event_log, &harn_vm::TrustQueryFilters::default())
                .await
                .expect("records");

        assert_eq!(records.len(), 1);
        assert_eq!(records[0].trace_id, response.trace_id.0);
        assert_eq!(records[0].metadata["adapter"], "mcp");
    }

    #[tokio::test]
    async fn dispatch_propagates_cancelled_execution() {
        let dir = tempfile::tempdir().expect("tempdir");
        let script = dir.path().join("server.harn");
        std::fs::write(
            &script,
            r#"
pub fn spin() -> string {
  while true {
    if is_cancelled() {
      return "stopped"
    }
  }
}
"#,
        )
        .expect("write script");

        let core = DispatchCore::new(DispatchCoreConfig::for_script(&script)).expect("core");
        let cancel_token = Arc::new(AtomicBool::new(true));
        let response = core
            .dispatch(CallRequest {
                adapter: "acp".to_string(),
                function: "spin".to_string(),
                arguments: CallArguments::Positional(Vec::new()),
                auth: AuthRequest::default(),
                caller: "tester".to_string(),
                replay_key: Some("cancel-key".to_string()),
                trace_id: None,
                parent_span_id: None,
                metadata: BTreeMap::new(),
                cancel_token: Some(cancel_token),
                agent_session_id: None,
                agent_event_sink: None,
                actor_chain: None,
                actor_chain_hop: None,
                progress: None,
                tenant_id: None,
                request_id: None,
                auth_context: None,
                auth_principal: None,
            })
            .await
            .expect("dispatch");

        assert_eq!(response.value, serde_json::json!("stopped"));
    }

    /// `.harn` callees see the tenant the host bound via
    /// `AuthPolicy` — the `ApiKeyEntry` was configured with a tenant,
    /// the principal carries it forward, and `DispatchCore::dispatch`
    /// installs the [`harn_vm::enter_tenant`] guard so the script's
    /// `harness.tenant.id()` returns the same id end-to-end.
    #[tokio::test]
    async fn dispatch_threads_api_key_tenant_into_harness_and_trust_record() {
        let dir = tempfile::tempdir().expect("tempdir");
        let script = dir.path().join("server.harn");
        std::fs::write(
            &script,
            r"
pub fn whoami(harness: Harness) -> string {
  return harness.tenant.id()
}
",
        )
        .expect("write script");

        let mut config = DispatchCoreConfig::for_script(&script);
        config.auth_policy = crate::auth::AuthPolicy {
            methods: vec![crate::auth::AuthMethodConfig::ApiKey(
                crate::auth::ApiKeyAuthConfig {
                    keys: vec![
                        crate::auth::ApiKeyEntry::new("alice-key", []).with_tenant("acme-corp")
                    ],
                },
            )],
            mcp_allowlist: None,
        };
        let core = DispatchCore::new(config).expect("core");

        let response = core
            .dispatch(CallRequest {
                adapter: "mcp".to_string(),
                function: "whoami".to_string(),
                arguments: CallArguments::Positional(Vec::new()),
                auth: AuthRequest {
                    headers: BTreeMap::from([(
                        "authorization".to_string(),
                        "Bearer alice-key".to_string(),
                    )]),
                    ..AuthRequest::default()
                },
                caller: "tester".to_string(),
                replay_key: Some("tenant-whoami".to_string()),
                trace_id: None,
                parent_span_id: None,
                metadata: BTreeMap::new(),
                cancel_token: None,
                agent_session_id: None,
                agent_event_sink: None,
                actor_chain: None,
                actor_chain_hop: None,
                progress: None,
                tenant_id: None,
                request_id: None,
                auth_context: None,
                auth_principal: None,
            })
            .await
            .expect("dispatch");

        assert_eq!(response.value, serde_json::json!("acme-corp"));

        let records =
            harn_vm::query_trust_records(&core.event_log, &harn_vm::TrustQueryFilters::default())
                .await
                .expect("records");
        assert_eq!(records.len(), 1);
        assert_eq!(records[0].metadata["tenant_id"], "acme-corp");
    }

    #[tokio::test]
    async fn dispatch_threads_actor_chain_into_agent_session() {
        harn_vm::reset_thread_local_state();
        let dir = tempfile::tempdir().expect("tempdir");
        let script = dir.path().join("server.harn");
        std::fs::write(
            &script,
            r"
pub fn actor_chain() -> any {
  return agent_session_actor_chain()
}
",
        )
        .expect("write script");

        let mut config = DispatchCoreConfig::for_script(&script);
        config.auth_policy = crate::auth::AuthPolicy {
            methods: vec![crate::auth::AuthMethodConfig::ApiKey(
                crate::auth::ApiKeyAuthConfig {
                    keys: vec![crate::auth::ApiKeyEntry::new("actor-key", [])],
                },
            )],
            mcp_allowlist: None,
        };
        let core = DispatchCore::new(config).expect("core");

        let response = core
            .dispatch(CallRequest {
                adapter: "a2a".to_string(),
                function: "actor_chain".to_string(),
                arguments: CallArguments::Positional(Vec::new()),
                auth: AuthRequest {
                    headers: BTreeMap::from([(
                        "authorization".to_string(),
                        "Bearer actor-key".to_string(),
                    )]),
                    ..AuthRequest::default()
                },
                caller: "tester".to_string(),
                replay_key: Some("actor-chain".to_string()),
                trace_id: None,
                parent_span_id: None,
                metadata: BTreeMap::new(),
                cancel_token: None,
                agent_session_id: Some("dispatch-actor-chain".to_string()),
                agent_event_sink: None,
                actor_chain: None,
                actor_chain_hop: Some("agent:merge-captain".to_string()),
                progress: None,
                tenant_id: None,
                request_id: None,
                auth_context: None,
                auth_principal: None,
            })
            .await
            .expect("dispatch");

        let expected = serde_json::json!({
            "sub": "api-key",
            "act": {
                "sub": "agent:merge-captain"
            }
        });
        assert_eq!(response.value, expected);
        assert_eq!(
            harn_vm::agent_sessions::actor_chain("dispatch-actor-chain")
                .map(|chain| chain.to_json_value()),
            Some(expected)
        );
    }

    /// `harness.tenant.id()` raises a typed runtime error (categorized
    /// as `auth`) when the dispatch was not bound to a tenant. The
    /// dispatch surface then maps it through the standard `Execution`
    /// error envelope so callers see the canonical message.
    #[tokio::test]
    async fn dispatch_missing_tenant_raises_typed_runtime_error() {
        let dir = tempfile::tempdir().expect("tempdir");
        let script = dir.path().join("server.harn");
        std::fs::write(
            &script,
            r"
pub fn whoami(harness: Harness) -> string {
  return harness.tenant.id()
}
",
        )
        .expect("write script");

        let core = DispatchCore::new(DispatchCoreConfig::for_script(&script)).expect("core");
        let error = core
            .dispatch(CallRequest {
                adapter: "mcp".to_string(),
                function: "whoami".to_string(),
                arguments: CallArguments::Positional(Vec::new()),
                auth: AuthRequest::default(),
                caller: "tester".to_string(),
                replay_key: Some("missing-tenant".to_string()),
                trace_id: None,
                parent_span_id: None,
                metadata: BTreeMap::new(),
                cancel_token: None,
                agent_session_id: None,
                agent_event_sink: None,
                actor_chain: None,
                actor_chain_hop: None,
                progress: None,
                tenant_id: None,
                request_id: None,
                auth_context: None,
                auth_principal: None,
            })
            .await
            .expect_err("missing tenant should error");

        let message = error.message();
        assert!(
            message.contains("harness.tenant.id()"),
            "expected typed tenant error, got: {message}"
        );
    }

    /// `CallRequest.tenant_id` overrides the principal-supplied tenant
    /// — covers the case where an upstream gateway already resolved
    /// tenancy out-of-band and hands the answer to harn-serve.
    #[tokio::test]
    async fn dispatch_request_tenant_overrides_principal_tenant() {
        let dir = tempfile::tempdir().expect("tempdir");
        let script = dir.path().join("server.harn");
        std::fs::write(
            &script,
            r"
pub fn whoami(harness: Harness) -> string {
  return harness.tenant.id()
}
",
        )
        .expect("write script");

        let mut config = DispatchCoreConfig::for_script(&script);
        config.auth_policy = crate::auth::AuthPolicy {
            methods: vec![crate::auth::AuthMethodConfig::ApiKey(
                crate::auth::ApiKeyAuthConfig {
                    keys: vec![
                        crate::auth::ApiKeyEntry::new("key", []).with_tenant("principal-tenant")
                    ],
                },
            )],
            mcp_allowlist: None,
        };
        let core = DispatchCore::new(config).expect("core");

        let response = core
            .dispatch(CallRequest {
                adapter: "mcp".to_string(),
                function: "whoami".to_string(),
                arguments: CallArguments::Positional(Vec::new()),
                auth: AuthRequest {
                    headers: BTreeMap::from([(
                        "authorization".to_string(),
                        "Bearer key".to_string(),
                    )]),
                    ..AuthRequest::default()
                },
                caller: "tester".to_string(),
                replay_key: Some("override-tenant".to_string()),
                trace_id: None,
                parent_span_id: None,
                metadata: BTreeMap::new(),
                cancel_token: None,
                agent_session_id: None,
                agent_event_sink: None,
                actor_chain: None,
                actor_chain_hop: None,
                progress: None,
                tenant_id: Some(harn_vm::TenantId::new("override-tenant")),
                request_id: None,
                auth_context: None,
                auth_principal: None,
            })
            .await
            .expect("dispatch");

        assert_eq!(response.value, serde_json::json!("override-tenant"));
    }

    #[path = "dispatch_error_tests.rs"]
    mod dispatch_error_tests;
    #[path = "trusted_host_dispatch_tests.rs"]
    mod trusted_host_dispatch_tests;
    #[path = "typed_pipeline_tests.rs"]
    mod typed_pipeline_tests;
}