wm_dispatch/

pipeline.rs

//! Dispatch pipeline — the request processing chain.
//!
//! Pipeline order:
//! 1. Effect check — brain-wave compatibility (zero-cost, inline)
//! 2. Dharma gate — ethical governance verdict
//! 3. Resource rules — write/spawn/network budgets, novelty, human review
//! 4. Rate limit — sliding window per-tool + global
//! 5. Circuit breaker — fault tolerance, fast-fail on repeated errors
//! 6. Tool call — execute the tool (optionally bounded by a dispatch timeout).
//!    Secret-scan sampling (6b) runs right after a successful call: warn-only
//!    credential-shape scan, deterministic 1-in-N, content never logged
//!    (P-PROV-5/B(c)).
//! 7. Karma record + write-audit journal — declared vs actual effects
//!    (confirm-gated dispatches record the confirm — the delete-confirm audit)
//! 8. Stats — success/failure and latency tracking
//!
//! Between 4 and 5 sits the firebreak (fix-queue P1.4+P1.6): the explicit
//! `confirm: true` gate for destructive tools, the promoted Jan-11
//! forbidden-command veto, the bulk-scope law, and advisory disclosure.

#[cfg(test)]
use async_trait::async_trait;
use std::sync::Arc;
use std::time::{Duration, Instant};
use wm_core::{Args, Context, CoreError, Output, Result, Tool};

use crate::capability_gate::{CapabilityGateMode, GateOutcome};
use crate::circuit_breaker::CircuitBreakerRegistry;
use crate::rate_limiter::RateLimiter;
use wm_governance::{
    ActionVerdict, DharmaGate, FirebreakOutcome, KarmaLedger, ResourceRules, ResourceVerdict,
};

/// Default dispatch timeout (300s) applied by [`DispatchPipeline::from_env`]
/// when `WM_DISPATCH_TIMEOUT_MS` is unset.
///
/// Generous enough for LLM-backed tools (research, self-play) while still
/// bounding a hung call.
pub const DEFAULT_DISPATCH_TIMEOUT: Duration = Duration::from_secs(300);

/// Stable 64-bit hash of the serialized args — drives novelty tracking so
/// identical repeated calls are recognizable across dispatches.
fn hash_args(args: &Args) -> u64 {
    use std::hash::Hasher;
    let bytes = serde_json::to_vec(args).unwrap_or_default();
    let mut hasher = ahash::AHasher::default();
    hasher.write(&bytes);
    hasher.finish()
}

/// First non-empty string found under any of the given keys.
fn first_str(v: &serde_json::Value, keys: &[&str]) -> Option<String> {
    keys.iter().find_map(|k| {
        v.get(*k)
            .and_then(serde_json::Value::as_str)
            .map(str::to_string)
    })
}

/// Append a write-audit journal entry for one dispatch.
///
/// `store_write_baseline` must be sampled at dispatch start (see
/// [`WriteAuditJournal::dispatch_baseline`]) so the entry attributes exactly
/// the mutations that happened while this dispatch ran — not whatever other
/// dispatches (or bookkeeping flushes) wrote since the previous entry.
/// `confirm_gated` is `Some(confirm)` for destructive dispatches — the
/// delete-confirm audit field (P1.6) — and `None` for everything else.
#[allow(clippy::too_many_arguments)]
fn record_write_audit(
    journal: &wm_governance::WriteAuditJournal,
    store_write_baseline: u64,
    tool: &str,
    actor: wm_governance::ActorIdentity,
    declared_writes: bool,
    args_memory_id: Option<&str>,
    args_content_hash: Option<&str>,
    args_digest: Option<String>,
    output: &serde_json::Value,
    success: bool,
    confirm_gated: Option<bool>,
) {
    // The meta-router (`wm`) mutates only through nested dispatches, which
    // journal themselves with the real tool identity; a router entry would
    // attribute the inner writes to 'wm' as an undeclared mutation — a
    // permanent false misdeclaration for every meta-routed write (first-run
    // feedback, 2026-09-13: `wm doctor` never reached a clean summary).
    if tool == "wm" {
        return;
    }
    let reported_writes = output
        .get("writes")
        .and_then(|w| w.as_array())
        .map_or(0, |a| a.len() as u32);
    let memory_id = first_str(output, &["id", "memory_id", "memory"])
        .or_else(|| args_memory_id.map(str::to_string));
    let content_hash = first_str(output, &["content_hash", "hash", "sha256"])
        .or_else(|| args_content_hash.map(str::to_string));
    let result = match confirm_gated {
        Some(confirmed) => journal.record_since_confirmed(
            store_write_baseline,
            tool,
            actor,
            memory_id.as_deref(),
            content_hash.as_deref(),
            declared_writes,
            reported_writes,
            success,
            confirmed,
            args_digest,
        ),
        None => journal.record_since(
            store_write_baseline,
            tool,
            actor,
            memory_id.as_deref(),
            content_hash.as_deref(),
            declared_writes,
            reported_writes,
            success,
            args_digest,
        ),
    };
    if let Err(e) = result {
        tracing::warn!(error = %e, "Write-audit journal record failed");
    }
}

/// The dispatch pipeline processes tool calls through governance,
/// rate limiting, circuit breaking, and karma tracking before and after
/// the actual tool execution.
pub struct DispatchPipeline {
    rate_limiter: Arc<RateLimiter>,
    circuit_breakers: Arc<CircuitBreakerRegistry>,
    dharma_gate: Arc<DharmaGate>,
    karma_ledger: Option<Arc<KarmaLedger>>,
    /// Optional ResourceRules (Yama) — write/spawn/network budgets, novelty,
    /// purpose, and human-review gates evaluated on the dispatch path.
    resource_rules: Option<Arc<ResourceRules>>,
    /// Optional write gate (V8 S5 stage 2c) — junk filter, dedup gate, and
    /// class plausibility ceilings/floors on the memory-create path.
    write_gate: Option<Arc<crate::write_gate::WriteGate>>,
    /// Optional write-audit journal — append-only record of declared vs
    /// actual store mutations per dispatch.
    write_audit: Option<Arc<wm_governance::WriteAuditJournal>>,
    /// Optional secret scanner (P-PROV-5/B(c)) — warn-only credential-shape
    /// sampling over successful dispatch outputs. `None` disables.
    secret_scan: Option<crate::secret_scan::SharedSampler>,
    /// Optional scoped-thread sandbox executor (P-SANDBOX-3, Landlock v1) —
    /// `StoreScoped` tools run confined on a fresh thread. `None` = the
    /// declared flag is inert (v0 whole-process ruleset may still apply).
    sandbox_exec: Option<Arc<crate::sandbox_exec::ScopedSandboxExecutor>>,
    /// Optional subprocess spawn sandbox registry (B2) — tools declaring
    /// `Sandbox::Subprocess` get a runner-backed
    /// [`wm_core::sandbox::SpawnPolicy`] injected into their context;
    /// counters and the active-runner disclosure ride the dispatch.
    /// `None` = the declarations are inert (Landlock v1 doctrine).
    subprocess_sandbox: Option<Arc<crate::subprocess_sandbox::SubprocessSandbox>>,
    /// Optional flight recorder (Q35b) — opt-in JSONL payload capture for
    /// replay. Captures at the same point as `args_digest` so sidecar args
    /// always hash to the journal digest (the replay identity gate).
    flight_recorder: Option<Arc<crate::flight::FlightRecorder>>,
    /// The firebreak — forbidden-command guardrail (P1.4) + bulk-scope law
    /// (P1.6). Armed by default on every construction path; see
    /// [`wm_governance::Firebreak`].
    firebreak: Option<Arc<wm_governance::Firebreak>>,
    /// Capability gate (PLAN_F F-1, dispatch half) — maps `EffectRow.invokes`
    /// onto governance capabilities and verifies any engagement credential
    /// presented under `args["_engagement"]`. Advisory by default; strict via
    /// `WM_REQUIRE_CAPABILITIES=1`.
    capability_mode: CapabilityGateMode,
    /// Optional GanaRegistry for tracking co-usage patterns (Phase 6)
    gana_registry: Option<Arc<std::sync::Mutex<wm_core::GanaRegistry>>>,
    /// Optional upper bound on tool execution. When a call exceeds it, the
    /// future is dropped and a `CoreError::Tool` timeout error is returned, so
    /// one hung tool can't wedge the server's event loop or block shutdown.
    dispatch_timeout: Option<Duration>,
}

impl DispatchPipeline {
    /// Create a new dispatch pipeline with the given components.
    ///
    /// Not `const`: the default-armed firebreak is built here (pattern
    /// sets compile once per pipeline).
    pub fn new(
        rate_limiter: Arc<RateLimiter>,
        circuit_breakers: Arc<CircuitBreakerRegistry>,
        dharma_gate: Arc<DharmaGate>,
        karma_ledger: Option<Arc<KarmaLedger>>,
    ) -> Self {
        Self {
            rate_limiter,
            circuit_breakers,
            dharma_gate,
            karma_ledger,
            resource_rules: None,
            write_gate: None,
            write_audit: None,
            flight_recorder: None,
            // The secret scanner is on by default like the firebreak: a
            // tripwire you must remember to attach is not a tripwire.
            // Warn-only at a deterministic 1-in-N cadence — it observes,
            // never blocks. Override with `with_secret_scan_option`.
            secret_scan: Some(Arc::new(crate::secret_scan::SecretSampler::from_env())),
            // The per-tool sandbox executor is attached explicitly by the
            // deployment (wm-mcp injects the Landlock callback when
            // WM_LANDLOCK_V1=1); without it, StoreScoped marks are inert.
            sandbox_exec: None,
            // Same doctrine for the B2 subprocess registry: attached
            // explicitly by the deployment (wm-mcp injects a detected
            // runner); without it, Subprocess marks are inert.
            subprocess_sandbox: None,
            // The firebreak arms by default: every construction path (server,
            // daemon, CLI, tests) inherits the veto + scope law unless it is
            // explicitly disarmed with `with_firebreak_option(None)` or the
            // `WM_FIREBREAK=0` kill-switch. A guardrail you must remember to
            // attach is not a guardrail.
            firebreak: Some(Arc::new(wm_governance::Firebreak::promoted())),
            capability_mode: CapabilityGateMode::from_env(),
            gana_registry: None,
            dispatch_timeout: None,
        }
    }

    /// Parse the dispatch timeout from `WM_DISPATCH_TIMEOUT_MS`.
    ///
    /// Unset → [`DEFAULT_DISPATCH_TIMEOUT`]; `0` → disabled; other values are
    /// milliseconds. Invalid values fall back to the default.
    #[must_use]
    pub fn timeout_from_env() -> Option<Duration> {
        match std::env::var("WM_DISPATCH_TIMEOUT_MS") {
            Ok(v) => match v.trim().parse::<u64>() {
                Ok(0) => None,
                Ok(ms) => Some(Duration::from_millis(ms)),
                Err(_) => {
                    tracing::warn!(
                        value = %v,
                        "WM_DISPATCH_TIMEOUT_MS is not a valid millisecond count — using default"
                    );
                    Some(DEFAULT_DISPATCH_TIMEOUT)
                }
            },
            Err(_) => Some(DEFAULT_DISPATCH_TIMEOUT),
        }
    }

    /// Bound tool execution with a timeout (`None` disables the bound).
    #[must_use]
    pub const fn with_dispatch_timeout(mut self, timeout: Option<Duration>) -> Self {
        self.dispatch_timeout = timeout;
        self
    }

    /// Create a pipeline with default components and no karma ledger.
    #[must_use]
    pub fn with_defaults() -> Self {
        Self::new(
            Arc::new(RateLimiter::default()),
            Arc::new(CircuitBreakerRegistry::default()),
            Arc::new(DharmaGate::default()),
            None,
        )
    }

    /// Override the capability-gate mode (tests, deliberate strict runs).
    #[must_use]
    pub const fn with_capability_mode(mut self, mode: CapabilityGateMode) -> Self {
        self.capability_mode = mode;
        self
    }

    /// Attach a GanaRegistry for co-usage tracking (Phase 6).
    #[must_use]
    pub fn with_gana_registry(
        mut self,
        registry: Arc<std::sync::Mutex<wm_core::GanaRegistry>>,
    ) -> Self {
        self.gana_registry = Some(registry);
        self
    }

    /// Attach ResourceRules (Yama) — evaluated on every dispatch.
    #[must_use]
    pub fn with_resource_rules(mut self, rules: Arc<ResourceRules>) -> Self {
        self.resource_rules = Some(rules);
        self
    }

    /// Attach the write gate (V8 S5 stage 2c) — runs between resource
    /// rules and the rate limiter: junk filter, dedup short-circuit, and
    /// class plausibility ceilings/floors on the memory-create path.
    #[must_use]
    pub fn with_write_gate(mut self, gate: Arc<crate::write_gate::WriteGate>) -> Self {
        self.write_gate = Some(gate);
        self
    }

    /// Attach a write-audit journal — every dispatch appends a journal entry
    /// recording declared vs actual store mutations.
    #[must_use]
    pub fn with_write_audit(mut self, journal: Arc<wm_governance::WriteAuditJournal>) -> Self {
        self.write_audit = Some(journal);
        self
    }

    /// Attach a flight recorder (Q35b replay capture). OFF by default.
    /// Capture point matches `args_digest` (post-gate, pre-call) so the
    /// sidecar is always digest-aligned with the journal.
    #[must_use]
    pub fn with_flight_recorder(
        mut self,
        recorder: Option<Arc<crate::flight::FlightRecorder>>,
    ) -> Self {
        self.flight_recorder = recorder;
        self
    }

    /// Replace the default secret scanner — `None` disables output
    /// sampling entirely for this pipeline (tests, special constructions).
    #[must_use]
    pub fn with_secret_scan_option(
        mut self,
        scanner: Option<crate::secret_scan::SharedSampler>,
    ) -> Self {
        self.secret_scan = scanner;
        self
    }

    /// The secret scanner attached to this pipeline (if any).
    #[must_use]
    pub fn secret_scan(&self) -> Option<&crate::secret_scan::SecretSampler> {
        self.secret_scan.as_deref()
    }

    /// Attach the scoped-thread sandbox executor (P-SANDBOX-3). When
    /// attached, tools declaring `Sandbox::StoreScoped` run on a confined
    /// fresh thread; everything else keeps the ambient path.
    #[must_use]
    pub fn with_sandbox_executor(
        mut self,
        executor: Option<Arc<crate::sandbox_exec::ScopedSandboxExecutor>>,
    ) -> Self {
        self.sandbox_exec = executor;
        self
    }

    /// The sandbox executor attached to this pipeline (if any).
    #[must_use]
    pub fn sandbox_executor(&self) -> Option<&crate::sandbox_exec::ScopedSandboxExecutor> {
        self.sandbox_exec.as_deref()
    }

    /// Attach the subprocess spawn sandbox registry (B2). When attached,
    /// `Sandbox::Subprocess` tools receive a runner-backed spawn policy in
    /// their context; anything short of an active runner loud-degrades.
    #[must_use]
    pub fn with_subprocess_sandbox(
        mut self,
        sandbox: Option<Arc<crate::subprocess_sandbox::SubprocessSandbox>>,
    ) -> Self {
        self.subprocess_sandbox = sandbox;
        self
    }

    /// The subprocess spawn sandbox registry attached to this pipeline.
    #[must_use]
    pub fn subprocess_sandbox(&self) -> Option<&crate::subprocess_sandbox::SubprocessSandbox> {
        self.subprocess_sandbox.as_deref()
    }

    /// Attach a firebreak with an explicit arm state (tests, special
    /// constructions) — see [`Self::with_firebreak_option`].
    #[must_use]
    pub fn with_firebreak(mut self, firebreak: Arc<wm_governance::Firebreak>) -> Self {
        self.firebreak = Some(firebreak);
        self
    }

    /// Replace the default-armed firebreak — `None` disarms it entirely
    /// for this pipeline (the `WM_FIREBREAK=0` env kill-switch operates
    /// inside [`wm_governance::Firebreak::promoted`] and is the normal
    /// off switch; this builder is for tests and special constructions).
    #[must_use]
    pub fn with_firebreak_option(
        mut self,
        firebreak: Option<Arc<wm_governance::Firebreak>>,
    ) -> Self {
        self.firebreak = firebreak;
        self
    }

    /// The firebreak attached to this pipeline (if any).
    #[must_use]
    pub fn firebreak(&self) -> Option<&wm_governance::Firebreak> {
        self.firebreak.as_deref()
    }

    /// Optional variant of [`Self::with_write_audit`] — read-only servers
    /// pass `None` because journaling is itself an LMDB write.
    #[must_use]
    pub fn with_write_audit_option(
        mut self,
        journal: Option<Arc<wm_governance::WriteAuditJournal>>,
    ) -> Self {
        self.write_audit = journal;
        self
    }

    /// The resource rules attached to this pipeline (if any).
    #[must_use]
    pub fn resource_rules(&self) -> Option<&ResourceRules> {
        self.resource_rules.as_deref()
    }

    /// The write-audit journal attached to this pipeline (if any).
    #[must_use]
    pub fn write_audit(&self) -> Option<&wm_governance::WriteAuditJournal> {
        self.write_audit.as_deref()
    }

    /// Dispatch a tool call through the full pipeline.
    pub async fn dispatch(&self, tool: &dyn Tool, ctx: &mut Context, args: Args) -> Result<Output> {
        let start = Instant::now();
        let mut args = args;

        // 1. Effect check — brain-wave compatibility
        // Explicit `confirm: true` (resolved here, before every gate, so
        // deliberate operator intent is visible downstream) bypasses the
        // eco-mode availability restriction: eco mode conserves autonomous
        // resources, and a confirmed destructive action is deliberate, not
        // autonomous. The coherence gate below stays absolute (9.1.6).
        let confirmed = args
            .get("confirm")
            .and_then(serde_json::Value::as_bool)
            .unwrap_or(false);
        ctx.explicit_confirm = confirmed;
        if !tool.effects().is_available_in(ctx.brain_wave) && !confirmed {
            return Err(CoreError::Governance(format!(
                "tool '{}' not available in {:?} brain-wave state",
                tool.name(),
                ctx.brain_wave
            )));
        }

        // 1b. Coherence gate — refuse writes when citta coherence is low
        const COHERENCE_THRESHOLD: f32 = 0.3;
        if !tool.effects().writes.is_empty() && ctx.citta_coherence < COHERENCE_THRESHOLD {
            return Err(CoreError::Governance(format!(
                "tool '{}' requires write access but citta coherence is {:.2} (minimum {:.2})",
                tool.name(),
                ctx.citta_coherence,
                COHERENCE_THRESHOLD
            )));
        }

        // 1c. Read-only gate — server-level `--readonly` refuses every tool
        // that declares writes, whether dispatched directly or through the
        // `wm` meta-tool.
        if ctx.readonly && !tool.effects().writes.is_empty() {
            return Err(CoreError::Governance(format!(
                "server is read-only: tool '{}' requires write access",
                tool.name()
            )));
        }

        // 1c. Self-model confidence — conservative dispatch when confidence is low
        const CONFIDENCE_THRESHOLD: f32 = 0.5;
        if ctx.self_model_confidence < CONFIDENCE_THRESHOLD {
            tracing::warn!(
                tool = tool.name(),
                confidence = ctx.self_model_confidence,
                "low self-model confidence — conservative dispatch mode"
            );
            // Block write operations when confidence is low — can't trust side effects
            if !tool.effects().writes.is_empty() {
                return Err(CoreError::Governance(format!(
                    "homeostasis limit (self-model confidence): tool '{}' requires write access but confidence is {:.2} (minimum {:.2}) — conservative dispatch blocks writes; this is load-sensitive, retry when the host settles (deterministic runs can pin WM_HOMEOSTASIS_FROZEN=1)",
                    tool.name(),
                    ctx.self_model_confidence,
                    CONFIDENCE_THRESHOLD
                )));
            }
        }

        // 1d. Drive caution gate — warn on high-caution write operations
        const DRIVE_CAUTION_THRESHOLD: f32 = 0.85;
        if !tool.effects().writes.is_empty() && ctx.drive_caution > DRIVE_CAUTION_THRESHOLD {
            tracing::warn!(
                tool = tool.name(),
                drive_caution = ctx.drive_caution,
                "high drive caution — write operation flagged for review"
            );
        }

        // 1e. Drive energy gate — warn on low-energy write operations
        const DRIVE_ENERGY_THRESHOLD: f32 = 0.15;
        if !tool.effects().writes.is_empty() && ctx.drive_energy < DRIVE_ENERGY_THRESHOLD {
            tracing::warn!(
                tool = tool.name(),
                drive_energy = ctx.drive_energy,
                "low drive energy — write operation may be resource-constrained"
            );
        }

        // 1f. Capability gate (PLAN_F F-1, dispatch half) — the tool's
        // declared `invokes` must be covered by a presented engagement
        // credential. Presenting a credential always triggers cryptographic
        // verification (signature → revocation → expiry → scope coverage);
        // missing credentials are advisory by default and refused under
        // `WM_REQUIRE_CAPABILITIES=1`. The credential key is stripped from
        // args so tokens never reach tool bodies or audit digests.
        match crate::capability_gate::evaluate(
            tool.effects(),
            &mut args,
            self.capability_mode,
            chrono::Utc::now().timestamp(),
        ) {
            Ok(GateOutcome::AdvisoryMissing { required }) => {
                tracing::debug!(
                    tool = tool.name(),
                    required = %required.labels().join(", "),
                    mode = self.capability_mode.label(),
                    "capability gate: requirement unmet (advisory)"
                );
            }
            Ok(_) => {}
            Err(reason) => {
                return Err(CoreError::Governance(format!("capability gate: {reason}")));
            }
        }

        // 2. Dharma gate — ethical governance
        // (`confirmed` was resolved at step 1; the confirm gate in 4b
        // re-uses the same value.)
        let verdict = self.dharma_gate.evaluate(tool.effects(), ctx);
        match verdict {
            ActionVerdict::Panic(reason) => {
                tracing::error!(tool = tool.name(), reason = %reason, "Dharma PANIC");
                return Err(CoreError::Governance(reason));
            }
            ActionVerdict::Intervene(reason) => {
                tracing::warn!(tool = tool.name(), reason = %reason, "Dharma INTERVENE");
                return Err(CoreError::Governance(reason));
            }
            ActionVerdict::Correct(reason) => {
                tracing::info!(tool = tool.name(), reason = %reason, "Dharma CORRECT — proceeding with restrictions");
            }
            ActionVerdict::Advise(reason) => {
                tracing::debug!(tool = tool.name(), reason = %reason, "Dharma ADVISE");
            }
            ActionVerdict::Observe => {}
        }

        // 2b. Resource rules (Yama) — budgets, novelty, purpose, human review.
        //
        // Budget violations and autonomous human-review/purpose violations
        // block the dispatch. Novelty flags are non-blocking: they are
        // attached to the response so the caller can see the repetition.
        let mut novelty_flag: Option<String> = None;
        if let Some(ref rules) = self.resource_rules {
            let effects = tool.effects();
            let is_write = !effects.writes.is_empty();
            let is_spawn = effects.spawns
                || effects
                    .writes
                    .iter()
                    .chain(effects.reads.iter())
                    .any(|r| matches!(r, wm_core::Resource::Process));
            let is_network = effects
                .writes
                .iter()
                .chain(effects.reads.iter())
                .any(|r| matches!(r, wm_core::Resource::Network));
            let has_purpose = [args.get("purpose"), ctx.meta.get("purpose")]
                .into_iter()
                .flatten()
                .filter_map(serde_json::Value::as_str)
                .any(|p| !p.trim().is_empty());
            let homeostasis = self.dharma_gate.homeostasis();
            let verdict = rules.evaluate(
                tool.name(),
                hash_args(&args),
                is_write,
                is_spawn,
                is_network,
                has_purpose,
                &homeostasis,
                ctx.brain_wave,
            );
            match verdict {
                ResourceVerdict::Allow => {}
                ResourceVerdict::NotNovel { .. } => {
                    novelty_flag = Some(verdict.reason());
                    tracing::warn!(
                        tool = tool.name(),
                        reason = %verdict.reason(),
                        "resource rules: novelty flag on response"
                    );
                }
                ResourceVerdict::BudgetExceeded { .. }
                | ResourceVerdict::RequiresHumanReview { .. }
                | ResourceVerdict::NoPurpose { .. } => {
                    tracing::warn!(
                        tool = tool.name(),
                        reason = %verdict.reason(),
                        "resource rules: dispatch blocked"
                    );
                    return Err(CoreError::Governance(format!(
                        "resource rules: {}",
                        verdict.reason()
                    )));
                }
            }
        }

        // 2c. Write gate (V8 S5, MEMORY_TYPOLOGY §3) — junk filter, dedup
        // short-circuit, and class plausibility ceilings/floors on the
        // memory-create path. Sits after Yama (budgets gate the caller's
        // rights) and before rate limiting (the gate may rewrite args or
        // short-circuit, which must not consume rate budget).
        let gate_disclosure: Option<serde_json::Value> = if let Some(ref gate) = self.write_gate {
            let outcome = gate.enforce(tool.name(), &mut args)?;
            if let Some(sc) = outcome.short_circuit {
                return Ok(sc);
            }
            outcome.disclosure
        } else {
            None
        };

        // 3. Rate limit
        //
        // Categories are named explicitly: the dispatch request-rate governor
        // is NOT a write budget, a homeostasis limit, or a circuit breaker.
        // Collapsing them all under "rate limited" made a healthy system look
        // like a broken transport (2026-09-15 audit).
        if let Err(retry_after_ms) = self.rate_limiter.try_acquire(tool.name()) {
            return Err(CoreError::RateLimited(format!(
                "request rate limit (per-tool dispatch governor): '{}' — retry after {}ms",
                tool.name(),
                retry_after_ms
            )));
        }

        // 4. Circuit breaker
        if self.circuit_breakers.is_open(tool.name()) {
            let retry_after_ms = self
                .circuit_breakers
                .remaining_cooldown(tool.name())
                .as_millis();
            return Err(CoreError::CircuitBreaker(format!(
                "{} — repeated execution failures opened the breaker; retry after {}ms",
                tool.name(),
                retry_after_ms
            )));
        }

        // 4b. Destructive tool confirmation — requires explicit `confirm: true` in args
        // (`confirmed` was resolved above, before the Dharma gate).
        let confirm_gated = if tool.effects().destructive {
            if !confirmed {
                return Err(CoreError::Governance(format!(
                    "tool '{}' is destructive — pass `\"confirm\": true` in args to proceed",
                    tool.name()
                )));
            }
            // The delete-confirm audit field (P1.6): the journal entry for
            // this dispatch records that the caller confirmed.
            Some(true)
        } else {
            None
        };

        // 4c. Firebreak — the promoted Jan-11 forbidden-command guardrail
        // (P1.4) plus the bulk-scope law (P1.6, the Jul-13 lesson). Blocks
        // before execution: forbidden patterns veto even a confirmed call;
        // dangerous patterns demand explicit confirm; destructive tools
        // must carry a scope their registry rule accepts. See
        // `wm_governance::firebreak` for the doctrine and scoping (the
        // veto gates the irreversible seam, never prose).
        let mut firebreak_advisories: Vec<String> = Vec::new();
        if let Some(ref firebreak) = self.firebreak {
            match firebreak.enforce(tool.name(), tool.effects(), &args) {
                FirebreakOutcome::Blocked(reason) => {
                    tracing::warn!(tool = tool.name(), reason = %reason, "firebreak VETO");
                    return Err(CoreError::Governance(reason));
                }
                FirebreakOutcome::Proceed { advisories } if !advisories.is_empty() => {
                    tracing::info!(tool = tool.name(), advisories = ?advisories, "firebreak advisories");
                    firebreak_advisories = advisories;
                }
                FirebreakOutcome::Proceed { .. } => {}
            }
        }

        // 4d. Compartment access control — check declared galaxy reads/writes
        //        plus runtime galaxy argument from tool args.
        //
        //        Tools like memory.read accept a `galaxy` argument at runtime that
        //        may differ from the default galaxy declared in their EffectRow.
        //        We check both the static declarations and the runtime argument
        //        to prevent compartment bypass via runtime galaxy selection.
        //
        //        When a runtime `galaxy` argument is present, the tool's galaxy
        //        effects are runtime-directed, so the static loop defers to the
        //        runtime check below — a set-covering declaration (all memory
        //        galaxies) must not require access to galaxies the call never
        //        touches.
        let has_runtime_galaxy = args
            .get("galaxy")
            .and_then(serde_json::Value::as_str)
            .is_some_and(|g| !g.is_empty());
        let mut checked_galaxies: Vec<wm_core::Galaxy> = Vec::new();

        if !has_runtime_galaxy {
            for resource in &tool.effects().reads {
                if let wm_core::Resource::Galaxy(name) = resource {
                    if let Some(galaxy) = wm_core::Galaxy::from_db_name(name) {
                        if !ctx.can_access_galaxy(galaxy) {
                            return Err(CoreError::Governance(format!(
                                "compartment '{}' cannot read galaxy '{}' (tool '{}')",
                                ctx.compartment.as_deref().unwrap_or("none"),
                                name,
                                tool.name()
                            )));
                        }
                        checked_galaxies.push(galaxy);
                    }
                }
            }
            for resource in &tool.effects().writes {
                if let wm_core::Resource::Galaxy(name) = resource {
                    if let Some(galaxy) = wm_core::Galaxy::from_db_name(name) {
                        if !ctx.can_write_galaxy(galaxy) {
                            return Err(CoreError::Governance(format!(
                                "compartment '{}' cannot write to galaxy '{}' (tool '{}')",
                                ctx.compartment.as_deref().unwrap_or("none"),
                                name,
                                tool.name()
                            )));
                        }
                        checked_galaxies.push(galaxy);
                    }
                }
            }
        }

        // Check runtime `galaxy` argument if present and not already checked
        if let Some(galaxy_str) = args.get("galaxy").and_then(serde_json::Value::as_str) {
            if !galaxy_str.is_empty() {
                if let Some(runtime_galaxy) = wm_core::Galaxy::from_db_name(galaxy_str) {
                    if !checked_galaxies.contains(&runtime_galaxy) {
                        // Determine if this is a read or write based on EffectRow writes
                        let has_writes = !tool.effects().writes.is_empty();
                        if has_writes {
                            if !ctx.can_write_galaxy(runtime_galaxy) {
                                return Err(CoreError::Governance(format!(
                                    "compartment '{}' cannot write to galaxy '{}' (tool '{}' runtime arg)",
                                    ctx.compartment.as_deref().unwrap_or("none"),
                                    galaxy_str,
                                    tool.name()
                                )));
                            }
                        } else if !ctx.can_access_galaxy(runtime_galaxy) {
                            return Err(CoreError::Governance(format!(
                                "compartment '{}' cannot read galaxy '{}' (tool '{}' runtime arg)",
                                ctx.compartment.as_deref().unwrap_or("none"),
                                galaxy_str,
                                tool.name()
                            )));
                        }
                    }
                }
            }
        }

        // 4d. Runtime Satya check — a runtime `galaxy` argument can redirect
        // a write to citta even when the static declaration doesn't name it.
        // Writing the consciousness stream without reading evidence is
        // fabrication; the static Dharma rule can't see the runtime argument,
        // so the pipeline enforces the same rule here.
        if !tool.effects().writes.is_empty()
            && let Some(galaxy_str) = args.get("galaxy").and_then(serde_json::Value::as_str)
            && galaxy_str == "citta"
            && !tool
                .effects()
                .reads
                .iter()
                .any(|r| matches!(r, wm_core::Resource::Galaxy(g) if g == "citta"))
        {
            return Err(CoreError::Governance(
                "VIOLATION_SATYA: writing to citta (runtime galaxy) without reading — memory fabrication is forbidden"
                    .to_string(),
            ));
        }

        // 5. Tool call — optionally bounded so a hung tool can't wedge the
        // server's event loop or delay graceful shutdown.
        //
        // Capture identifying args first (consumed by the call below) so the
        // write-audit journal can record which memory was touched, and
        // sample the store mutation counter so the entry covers exactly
        // this dispatch's window.
        let args_memory_id = first_str(&args, &["id", "memory_id", "memory"]);
        let args_content_hash = first_str(&args, &["content_hash", "hash", "sha256"]);
        // Q35b flight-recorder: digest the dispatch input (route identity +
        // arg keys + value hashes, no raw values) so journal entries can
        // answer "what went in" — replay verification without storing
        // untrusted payloads verbatim in the audit trail.
        let args_digest = wm_governance::args_digest(tool.name(), &args);
        // Flight capture at the SAME point (post-gate, pre-call): the
        // sidecar args must hash to the journal digest, or replay's
        // identity gate is meaningless. Recorded regardless of outcome —
        // the journal does the same, and failed dispatches are part of
        // the session being reproduced.
        if let Some(ref flight) = self.flight_recorder {
            if let Err(e) = flight.record(tool.name(), &args) {
                tracing::warn!(error = %e, "Flight recorder capture failed (replay will refuse)");
            }
        }
        let write_audit_baseline = self
            .write_audit
            .as_ref()
            .map_or(0, |j| j.dispatch_baseline());
        // 4e. B2 subprocess spawn policy. Declared `Sandbox::Subprocess`
        // tools receive a runner-backed policy on their context *before*
        // the call; a declared tool with no runner resolvable still runs
        // (availability first) but is counted and warned — and a tool that
        // declares raw `spawns` without the contract is surfaced once.
        let mut spawn_disclosure: Option<serde_json::Value> = None;
        if let Some(sb) = self.subprocess_sandbox.as_deref() {
            if crate::subprocess_sandbox::SubprocessSandbox::declared(tool.effects()) {
                let policy = sb.policy_for(tool.effects());
                if policy.is_active() {
                    let mut disclosure = serde_json::json!({
                        "net": policy.allow_net(),
                        "envelope": wm_core::sandbox::ENVELOPE_SCHEMA,
                    });
                    if let Some(runner) = policy.runner()
                        && let Some(obj) = disclosure.as_object_mut()
                    {
                        obj.insert(
                            "runner".to_string(),
                            serde_json::Value::String(runner.display().to_string()),
                        );
                    }
                    sb.note_confined();
                    spawn_disclosure = Some(disclosure);
                } else {
                    sb.note_degraded(tool.name());
                }
                ctx.spawn = policy;
            } else if tool.effects().spawns {
                sb.note_unconfined_spawn(tool.name());
            }
        }
        // P-SANDBOX-3 (Landlock v1): a `StoreScoped` tool with an executor
        // attached runs on a confined scoped thread (synchronous — see
        // `sandbox_exec` for why, and for the timeout-parity v1 gap).
        let result = if crate::sandbox_exec::ScopedSandboxExecutor::handles(tool)
            && let Some(executor) = self.sandbox_exec.as_deref()
        {
            executor.run(tool, ctx, args)
        } else if let Some(timeout) = self.dispatch_timeout {
            if let Ok(res) = tokio::time::timeout(timeout, tool.call(ctx, args)).await {
                res
            } else {
                tracing::error!(
                    tool = tool.name(),
                    timeout_ms = timeout.as_millis(),
                    "tool dispatch timed out"
                );
                self.circuit_breakers.record_failure(tool.name());
                return Err(CoreError::Tool(format!(
                    "tool '{}' timed out after {}ms",
                    tool.name(),
                    timeout.as_millis()
                )));
            }
        } else {
            tool.call(ctx, args).await
        };
        let elapsed = start.elapsed();

        // 6b. Secret-scan sampling (P-PROV-5/B(c)) — warn-only
        // credential-shape scan over successful outputs. Deterministic
        // 1-in-N inside the sampler; content never logged, dispatch never
        // blocked. Failures are not scanned (v0 scope).
        if let Some(ref scanner) = self.secret_scan {
            if let Ok(ref output) = result {
                scanner.scan(tool.name(), output);
            }
        }

        // Attach a non-blocking novelty flag so it reaches the response.
        let result = match (result, novelty_flag) {
            (Ok(mut output), Some(flag)) => {
                if let serde_json::Value::Object(ref mut map) = output {
                    match map.get_mut("resource_flags") {
                        Some(serde_json::Value::Array(arr)) => {
                            arr.push(serde_json::Value::String(flag));
                        }
                        Some(_) => {}
                        None => {
                            map.insert(
                                "resource_flags".to_string(),
                                serde_json::Value::Array(vec![serde_json::Value::String(flag)]),
                            );
                        }
                    }
                }
                Ok(output)
            }
            (result, _) => result,
        };

        // Attach the write-gate disclosure the same way — a gate that
        // acts silently is a gate nobody can audit.
        let result = match (result, gate_disclosure) {
            (Ok(mut output), Some(disclosure)) => {
                if let serde_json::Value::Object(ref mut map) = output {
                    map.insert("write_gate".to_string(), disclosure);
                }
                Ok(output)
            }
            (result, _) => result,
        };

        // Attach firebreak advisories the same way — a gate that acts
        // silently is a gate nobody can audit. Caution-class findings and
        // confirmed dangerous patterns surface under `firebreak.advisories`.
        let result = match (result, firebreak_advisories) {
            (Ok(mut output), advisories) if !advisories.is_empty() => {
                if let serde_json::Value::Object(ref mut map) = output {
                    map.insert(
                        "firebreak".to_string(),
                        serde_json::json!({ "advisories": advisories }),
                    );
                }
                Ok(output)
            }
            (result, _) => result,
        };

        // Attach the subprocess-sandbox disclosure the same way — active
        // confinement on a declared spawn tool is announced, never silent.
        let result = match (result, spawn_disclosure) {
            (Ok(mut output), Some(disclosure)) => {
                if let serde_json::Value::Object(ref mut map) = output {
                    map.insert("sandbox".to_string(), disclosure);
                }
                Ok(output)
            }
            (result, _) => result,
        };

        // 6. Stats + circuit breaker feedback + karma record + write audit
        if let Ok(output) = &result {
            tool.stats().record_success(elapsed, elapsed);
            self.circuit_breakers.record_success(tool.name());

            if let Some(ref ledger) = self.karma_ledger {
                let declared_writes = !tool.effects().writes.is_empty();
                let actual_writes = output
                    .get("writes")
                    .and_then(|w| w.as_array())
                    .map_or(0, |a| a.len() as u32);
                if let Err(e) = ledger.record(tool.name(), declared_writes, actual_writes, true) {
                    tracing::warn!(error = %e, "Karma ledger record failed");
                }
                ctx.karma_debt = ledger.total_debt();
            }

            if let Some(ref journal) = self.write_audit {
                let declared_writes = !tool.effects().writes.is_empty();
                record_write_audit(
                    journal,
                    write_audit_baseline,
                    tool.name(),
                    wm_governance::ActorIdentity::from_context(ctx),
                    declared_writes,
                    args_memory_id.as_deref(),
                    args_content_hash.as_deref(),
                    Some(args_digest),
                    output,
                    true,
                    confirm_gated,
                );
            }
        } else {
            tool.stats().record_failure(elapsed);
            // Breaker health is about the BACKEND, not the caller. A malformed
            // request that the tool correctly rejects must not fast-fail the
            // next valid request (2026-09-15 audit).
            if let Err(err) = &result {
                if err.counts_as_breaker_failure() {
                    self.circuit_breakers.record_failure(tool.name());
                }
            }

            if let Some(ref ledger) = self.karma_ledger {
                let declared_writes = !tool.effects().writes.is_empty();
                if let Err(ke) = ledger.record(tool.name(), declared_writes, 0, false) {
                    tracing::warn!(error = %ke, "Karma ledger record failed");
                }
                ctx.karma_debt = ledger.total_debt();
            }

            if let Some(ref journal) = self.write_audit {
                let declared_writes = !tool.effects().writes.is_empty();
                record_write_audit(
                    journal,
                    write_audit_baseline,
                    tool.name(),
                    wm_governance::ActorIdentity::from_context(ctx),
                    declared_writes,
                    args_memory_id.as_deref(),
                    args_content_hash.as_deref(),
                    Some(args_digest),
                    &serde_json::Value::Null,
                    false,
                    confirm_gated,
                );
            }
        }

        // 6b. GanaRegistry — record usage and co-usage (Phase 6)
        if let Some(ref registry) = self.gana_registry {
            if let Ok(mut reg) = registry.lock() {
                let gana = tool.gana();
                reg.record_usage(gana, result.is_ok());
                // Record co-usage with the last Gana seen in this context
                if let Some(prev) = ctx.last_gana {
                    reg.record_co_usage(prev, gana);
                }
                ctx.last_gana = Some(gana);
            }
        }

        result
    }

    /// Dispatch a tool by name, looking it up in a registry.
    ///
    /// Convenience method that combines registry lookup with pipeline dispatch.
    /// Returns `NotFound` if the tool isn't registered.
    pub async fn dispatch_by_name(
        &self,
        registry: &crate::ToolRegistry,
        name: &str,
        ctx: &mut Context,
        args: Args,
    ) -> Result<Output> {
        let tool = registry
            .get(name)
            .ok_or_else(|| CoreError::NotFound(format!("tool '{name}' not registered")))?;
        self.dispatch(tool.as_ref(), ctx, args).await
    }

    /// Access the rate limiter.
    #[must_use]
    pub fn rate_limiter(&self) -> &RateLimiter {
        &self.rate_limiter
    }

    /// Access the circuit breaker registry.
    #[must_use]
    pub fn circuit_breakers(&self) -> &CircuitBreakerRegistry {
        &self.circuit_breakers
    }

    /// Access the Dharma gate.
    #[must_use]
    pub fn dharma_gate(&self) -> &DharmaGate {
        &self.dharma_gate
    }

    /// Access the karma ledger (if configured).
    #[must_use]
    pub fn karma_ledger(&self) -> Option<&KarmaLedger> {
        self.karma_ledger.as_deref()
    }
}

impl Default for DispatchPipeline {
    fn default() -> Self {
        Self::with_defaults()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use wm_core::{BrainWave, EffectRow, Gana, Sandbox, ToolStats};
    use wm_governance::{ResourceRulesConfig, WriteAuditJournal};

    struct TestTool {
        name: String,
        effects: EffectRow,
        stats: ToolStats,
        should_fail: bool,
        /// When set, `call` returns a fresh error of this class (error-class tests).
        error: Option<fn() -> CoreError>,
        output: Option<Output>,
        /// When set, the tool secretly writes one memory into this store —
        /// used to simulate a misdeclaring tool for the write-audit journal.
        store: Option<Arc<wm_memory::MemoryStore>>,
    }

    impl TestTool {
        fn new(name: &str, effects: EffectRow) -> Self {
            Self {
                name: name.to_string(),
                effects,
                stats: ToolStats::default(),
                should_fail: false,
                error: None,
                output: None,
                store: None,
            }
        }

        fn returning_error(name: &str, error: fn() -> CoreError) -> Self {
            Self {
                name: name.to_string(),
                effects: EffectRow::pure(),
                stats: ToolStats::default(),
                should_fail: false,
                error: Some(error),
                output: None,
                store: None,
            }
        }

        fn with_output(mut self, output: Output) -> Self {
            self.output = Some(output);
            self
        }

        fn with_store(mut self, store: Arc<wm_memory::MemoryStore>) -> Self {
            self.store = Some(store);
            self
        }

        fn failing(name: &str) -> Self {
            Self {
                name: name.to_string(),
                effects: EffectRow::pure(),
                stats: ToolStats::default(),
                should_fail: true,
                error: None,
                output: None,
                store: None,
            }
        }
    }

    #[async_trait]
    impl Tool for TestTool {
        fn name(&self) -> &str {
            &self.name
        }
        fn gana(&self) -> Gana {
            Gana::Heart
        }
        fn effects(&self) -> &EffectRow {
            &self.effects
        }
        async fn call(&self, _ctx: &mut Context, _args: Args) -> Result<Output> {
            if let Some(store) = &self.store {
                let mem = wm_memory::Memory::new(
                    wm_core::Galaxy::Codex,
                    format!("misdeclared write from {}", self.name),
                );
                store.put(wm_core::Galaxy::Codex, &mem).ok();
            }
            if let Some(error) = self.error {
                Err(error())
            } else if self.should_fail {
                Err(CoreError::Tool(self.name.clone()))
            } else {
                Ok(self
                    .output
                    .clone()
                    .unwrap_or_else(|| serde_json::json!("ok")))
            }
        }
        fn stats(&self) -> &ToolStats {
            &self.stats
        }
    }

    #[tokio::test]
    async fn pipeline_dispatch_success() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new("test_tool", EffectRow::pure());

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    struct HangingTool {
        effects: EffectRow,
        stats: ToolStats,
    }

    impl HangingTool {
        fn new() -> Self {
            Self {
                effects: EffectRow::pure(),
                stats: ToolStats::default(),
            }
        }
    }

    #[async_trait]
    impl Tool for HangingTool {
        fn name(&self) -> &str {
            "hanging_tool"
        }
        fn gana(&self) -> Gana {
            Gana::Heart
        }
        fn effects(&self) -> &EffectRow {
            &self.effects
        }
        async fn call(&self, _ctx: &mut Context, _args: Args) -> Result<Output> {
            tokio::time::sleep(Duration::from_secs(30)).await;
            Ok(serde_json::json!("never reached"))
        }
        fn stats(&self) -> &ToolStats {
            &self.stats
        }
    }

    #[tokio::test]
    async fn pipeline_dispatch_timeout_bounds_hung_tool() {
        let pipeline = DispatchPipeline::with_defaults()
            .with_dispatch_timeout(Some(Duration::from_millis(50)));
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = HangingTool::new();

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        let msg = result.err().unwrap().to_string();
        assert!(
            msg.contains("timed out"),
            "expected timeout error, got: {msg}"
        );
    }

    #[tokio::test]
    async fn pipeline_dispatch_with_timeout_allows_fast_tool() {
        let pipeline = DispatchPipeline::with_defaults()
            .with_dispatch_timeout(Some(Duration::from_millis(500)));
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new("fast_tool", EffectRow::pure());

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_dispatch_failure_records_stats() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::failing("failing_tool");

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        assert_eq!(
            tool.stats()
                .call_count
                .load(std::sync::atomic::Ordering::Relaxed),
            1
        );
    }

    #[tokio::test]
    async fn pipeline_blocks_incompatible_brain_wave() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Delta);
        let tool = TestTool::new("test_tool", EffectRow::pure());

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(_)) => {}
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_dharma_blocks_destructive_in_strict_mode() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Theta);
        let tool = TestTool::new(
            "destructive_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Filesystem],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(_)) => {}
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_strict_refusal_is_typed_and_distinct_from_starvation() {
        // Governance refusal (strict mode via system stress, Beta brain wave
        // so the availability gate admits the call): coordination lease
        // acquisition is refused with the typed AHIMSA violation.
        let pipeline = DispatchPipeline::with_defaults();
        pipeline
            .dharma_gate()
            .update_homeostasis(wm_governance::Homeostasis {
                cpu_load: 0.95,
                memory_pressure: 0.95,
                active: true,
            });
        let mut ctx = Context::new(BrainWave::Beta);
        let tool = TestTool::new(
            "stress_probe",
            EffectRow {
                reads: vec![wm_core::Resource::Filesystem],
                writes: vec![wm_core::Resource::CoordinationLease],
                ..Default::default()
            },
        );
        let governance = pipeline
            .dispatch(&tool, &mut ctx, Args::default())
            .await
            .expect_err("strict mode must refuse coordination lease acquisition");
        let text = governance.to_string();
        assert!(text.contains("VIOLATION_AHIMSA"), "{text}");

        // First-run starvation: low self-model confidence refuses writes with
        // the typed homeostasis-limit error naming the frozen pin — a
        // different class from the governance refusal, and reads stay open.
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.self_model_confidence = 0.3;
        let write_tool = TestTool::new(
            "stress_probe",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );
        let starvation = pipeline
            .dispatch(&write_tool, &mut ctx, Args::default())
            .await
            .expect_err("low confidence must refuse writes");
        let text = starvation.to_string();
        assert!(text.contains("self-model confidence"), "{text}");
        assert!(text.contains("WM_HOMEOSTASIS_FROZEN"), "{text}");
        assert!(
            !text.contains("VIOLATION_AHIMSA"),
            "refusal classes must be distinguishable: {text}"
        );

        let read_tool = TestTool::new(
            "stress_probe_read",
            EffectRow {
                reads: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );
        assert!(
            pipeline
                .dispatch(&read_tool, &mut ctx, Args::default())
                .await
                .is_ok(),
            "starvation must not block reads"
        );
    }

    #[tokio::test]
    async fn pipeline_dharma_confirm_passes_brain_wave_strict_for_destructive() {
        // 9.1.6: explicit `confirm: true` (deliberate operator intent)
        // passes the Theta/Delta brain-wave strict arm; stressed
        // homeostasis must still block (covered by dharma_gate unit tests).
        let pipeline = DispatchPipeline::with_defaults().with_resource_rules(Arc::new(
            ResourceRules::new(ResourceRulesConfig {
                require_human_review: false,
                ..Default::default()
            }),
        ));
        let mut ctx = Context::new(BrainWave::Theta);
        let tool = TestTool::new(
            "destructive_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Filesystem],
                ..Default::default()
            },
        );
        let result = pipeline
            .dispatch(&tool, &mut ctx, serde_json::json!({"confirm": true}))
            .await;
        assert!(
            result.is_ok(),
            "confirmed destructive dispatch must pass brain-wave strict: {result:?}"
        );
    }

    #[tokio::test]
    async fn pipeline_capability_gate_strict_blocks_uncredentialed() {
        let pipeline =
            DispatchPipeline::with_defaults().with_capability_mode(CapabilityGateMode::Strict);
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "capability_tool",
            EffectRow {
                invokes: vec![wm_core::Capability::MemoryWrite],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("capability gate"), "{msg}");
                assert!(msg.contains("memory:write"), "{msg}");
            }
            other => panic!("Expected capability refusal, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_capability_gate_strict_allows_valid_token() {
        let pipeline =
            DispatchPipeline::with_defaults().with_capability_mode(CapabilityGateMode::Strict);
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "capability_tool_ok",
            EffectRow {
                invokes: vec![wm_core::Capability::MemoryWrite],
                ..Default::default()
            },
        );

        let mut issuer = wm_governance::engagement_tokens::EngagementIssuer::with_keypair(
            wm_governance::network_profile::AgentKeypair::from_seed([7u8; 32]),
        );
        let issuer_key = issuer.signer_public_key_hex();
        let token = issuer.issue(
            "tester",
            wm_governance::engagement_tokens::EngagementScope::Poc,
            "rules-hash",
            Some(3600),
        );
        let args = serde_json::json!({
            "_engagement": { "token": token, "issuer_public_key": issuer_key }
        });

        let result = pipeline.dispatch(&tool, &mut ctx, args).await;
        assert!(result.is_ok(), "valid Poc token should pass: {result:?}");
    }

    #[tokio::test]
    async fn pipeline_capability_gate_advisory_allows_uncredentialed() {
        let pipeline =
            DispatchPipeline::with_defaults().with_capability_mode(CapabilityGateMode::Advisory);
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "capability_tool_advisory",
            EffectRow {
                invokes: vec![wm_core::Capability::MemoryWrite],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok(), "advisory mode must not block: {result:?}");
    }

    #[tokio::test]
    async fn pipeline_rate_limit_blocks_excess() {
        let rate_limiter = Arc::new(RateLimiter::new(1000, 2, 0));
        let pipeline = DispatchPipeline::new(
            rate_limiter,
            Arc::new(CircuitBreakerRegistry::default()),
            Arc::new(DharmaGate::default()),
            None,
        );

        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new("limited_tool", EffectRow::pure());

        assert!(
            pipeline
                .dispatch(&tool, &mut ctx, Args::default())
                .await
                .is_ok()
        );
        assert!(
            pipeline
                .dispatch(&tool, &mut ctx, Args::default())
                .await
                .is_ok()
        );
        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::RateLimited(_)) => {}
            other => panic!("Expected RateLimited error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_circuit_breaker_opens_on_repeated_failures() {
        let breakers = Arc::new(CircuitBreakerRegistry::new(
            crate::circuit_breaker::BreakerConfig {
                failure_threshold: 3,
                window: std::time::Duration::from_secs(10),
                cooldown: std::time::Duration::from_secs(30),
            },
        ));
        let pipeline = DispatchPipeline::new(
            Arc::new(RateLimiter::new(10000, 100, 100)),
            breakers.clone(),
            Arc::new(DharmaGate::default()),
            None,
        );

        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::failing("flaky_tool");

        for _ in 0..3 {
            let _ = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        }

        assert_eq!(
            breakers.state("flaky_tool"),
            crate::circuit_breaker::BreakerState::Open
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::CircuitBreaker(_)) => {}
            other => panic!("Expected CircuitBreaker error, got {other:?}"),
        }
    }

    /// 2026-09-15 audit: a caller's malformed requests must not fast-fail the
    /// next valid request — only backend/execution failures count.
    #[tokio::test]
    async fn client_validation_errors_do_not_trip_the_breaker() {
        let breakers = Arc::new(CircuitBreakerRegistry::new(
            crate::circuit_breaker::BreakerConfig {
                failure_threshold: 3,
                window: std::time::Duration::from_secs(10),
                cooldown: std::time::Duration::from_secs(30),
            },
        ));
        let pipeline = DispatchPipeline::new(
            Arc::new(RateLimiter::new(10000, 100, 100)),
            breakers.clone(),
            Arc::new(DharmaGate::default()),
            None,
        );
        let mut ctx = Context::new(BrainWave::Gamma);

        // Five invalid-galaxy-style caller errors: the shape that used to
        // trip the breaker and block the next correct call.
        let bad = TestTool::returning_error("validated_tool", || {
            CoreError::InvalidArgs("unknown galaxy".into())
        });
        for _ in 0..5 {
            let err = pipeline
                .dispatch(&bad, &mut ctx, Args::default())
                .await
                .unwrap_err();
            assert!(matches!(err, CoreError::InvalidArgs(_)));
        }
        assert_eq!(
            breakers.state("validated_tool"),
            crate::circuit_breaker::BreakerState::Closed,
            "caller errors must not open the breaker"
        );

        // Governance refusals likewise stay caller/request-scoped.
        let governed = TestTool::returning_error("validated_tool", || {
            CoreError::Governance("budget exceeded for writes".into())
        });
        for _ in 0..5 {
            let _ = pipeline
                .dispatch(&governed, &mut ctx, Args::default())
                .await;
        }
        assert_eq!(
            breakers.state("validated_tool"),
            crate::circuit_breaker::BreakerState::Closed,
            "governance refusals must not open the breaker"
        );

        // A healthy call still succeeds immediately.
        let good = TestTool::new("validated_tool", EffectRow::pure());
        pipeline
            .dispatch(&good, &mut ctx, Args::default())
            .await
            .expect("valid call after caller errors");
    }

    #[tokio::test]
    async fn rate_limit_error_names_its_governor() {
        let pipeline = DispatchPipeline::new(
            Arc::new(RateLimiter::new(1000, 1, 0)),
            Arc::new(CircuitBreakerRegistry::default()),
            Arc::new(DharmaGate::default()),
            None,
        );
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new("bursty_tool", EffectRow::pure());
        pipeline
            .dispatch(&tool, &mut ctx, Args::default())
            .await
            .unwrap();
        let err = pipeline
            .dispatch(&tool, &mut ctx, Args::default())
            .await
            .unwrap_err();
        let text = err.to_string();
        assert!(
            text.contains("request rate limit") && text.contains("retry after"),
            "rate limit must name its category and retry hint: {text}"
        );
    }

    #[tokio::test]
    async fn pipeline_karma_ledger_records() {
        let tmp = tempfile::tempdir().unwrap();
        let store = Arc::new(wm_memory::MemoryStore::open_default(tmp.path()).unwrap());
        let ledger = Arc::new(KarmaLedger::new(store).unwrap());

        let pipeline = DispatchPipeline::new(
            Arc::new(RateLimiter::default()),
            Arc::new(CircuitBreakerRegistry::default()),
            Arc::new(DharmaGate::default()),
            Some(ledger.clone()),
        );

        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new("karma_test_tool", EffectRow::pure());

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
        assert_eq!(ledger.next_id(), 1);
        assert_eq!(ctx.karma_debt, 0.0);
    }

    #[tokio::test]
    async fn pipeline_karma_debt_updates_context() {
        let tmp = tempfile::tempdir().unwrap();
        let store = Arc::new(wm_memory::MemoryStore::open_default(tmp.path()).unwrap());
        let ledger = Arc::new(KarmaLedger::new(store).unwrap());

        let pipeline = DispatchPipeline::new(
            Arc::new(RateLimiter::default()),
            Arc::new(CircuitBreakerRegistry::default()),
            Arc::new(DharmaGate::default()),
            Some(ledger),
        );

        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "wasteful_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
        assert!(
            (ctx.karma_debt - 0.2).abs() < 0.001,
            "Context karma_debt should be 0.2, got {}",
            ctx.karma_debt
        );
    }

    #[tokio::test]
    async fn pipeline_karma_batched_e2e() {
        // E2E: Full dispatch cycle with batched karma writes produces
        // correct total_debt() and chain integrity after flush.
        let tmp = tempfile::tempdir().unwrap();
        let store = Arc::new(wm_memory::MemoryStore::open_default(tmp.path()).unwrap());
        let ledger = Arc::new(KarmaLedger::with_flush_threshold(store.clone(), 100).unwrap());

        let pipeline = DispatchPipeline::new(
            Arc::new(RateLimiter::default()),
            Arc::new(CircuitBreakerRegistry::default()),
            Arc::new(DharmaGate::default()),
            Some(ledger.clone()),
        );

        let mut ctx = Context::new(BrainWave::Gamma);

        // Dispatch 10 honest tools (no debt) and 10 wasteful tools (0.2 debt each)
        let honest_tool = TestTool::new("honest_tool", EffectRow::pure());
        let wasteful_tool = TestTool::new(
            "wasteful_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        for _ in 0..10 {
            let result = pipeline
                .dispatch(&honest_tool, &mut ctx, Args::default())
                .await;
            assert!(result.is_ok());
        }
        for _ in 0..10 {
            let result = pipeline
                .dispatch(&wasteful_tool, &mut ctx, Args::default())
                .await;
            assert!(result.is_ok());
        }

        // 20 entries should be buffered (not yet in LMDB)
        assert_eq!(ledger.next_id(), 20);
        assert_eq!(
            ledger.pending_count(),
            20,
            "All 20 entries should be pending before flush"
        );

        // total_debt() reads from in-memory chain state — should reflect all 20
        let debt = ledger.total_debt();
        assert!(
            (debt - 2.0).abs() < 0.001,
            "Total debt should be 2.0 (10 x 0.2), got {debt}"
        );

        // Flush to persist all entries in one batch transaction
        ledger.flush().unwrap();
        assert_eq!(ledger.pending_count(), 0);

        // Verify chain integrity after batched flush
        let result = ledger.verify_integrity().unwrap();
        assert!(
            result.valid,
            "Chain should be valid after batched flush: {:?}",
            result.violation
        );
        assert_eq!(result.entries_verified, 20);

        // Verify entries are persisted by creating a new ledger from same store
        let ledger2 = KarmaLedger::new(store).unwrap();
        assert_eq!(
            ledger2.next_id(),
            20,
            "Next ID should persist across instances"
        );
        let entries = ledger2.scan_entries().unwrap();
        assert_eq!(
            entries.len(),
            20,
            "All 20 entries should be persisted in LMDB"
        );

        // Verify total debt persisted
        let debt2 = ledger2.total_debt();
        assert!(
            (debt2 - 2.0).abs() < 0.001,
            "Total debt should persist as 2.0, got {debt2}"
        );

        // Verify chain integrity on the reloaded ledger
        let result2 = ledger2.verify_integrity().unwrap();
        assert!(result2.valid, "Chain should be valid on reloaded ledger");
        assert_eq!(result2.entries_verified, 20);
    }

    #[tokio::test]
    async fn pipeline_coherence_gate_blocks_writes() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.citta_coherence = 0.1; // Below 0.3 threshold
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("coherence"));
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_coherence_gate_allows_reads() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.citta_coherence = 0.1; // Below threshold, but no writes
        let tool = TestTool::new("read_tool", EffectRow::pure());

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_coherence_gate_allows_writes_when_coherent() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.citta_coherence = 0.5; // Above threshold
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_low_confidence_blocks_writes() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.self_model_confidence = 0.3; // Below 0.5 threshold
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("confidence"));
                assert!(msg.contains("conservative"));
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_low_confidence_allows_reads() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.self_model_confidence = 0.3; // Below threshold, but no writes
        let tool = TestTool::new("read_tool", EffectRow::pure());

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_high_confidence_allows_writes() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.self_model_confidence = 0.8; // Above threshold
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_high_caution_warns_on_writes() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.drive_caution = 0.9; // Above 0.85 threshold
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        // Should still succeed — caution is a warning, not a block
        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_low_energy_warns_on_writes() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.drive_energy = 0.1; // Below 0.15 threshold
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        // Should still succeed — low energy is a warning, not a block
        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_drive_gates_dont_affect_reads() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.drive_caution = 0.95;
        ctx.drive_energy = 0.05;
        let tool = TestTool::new("read_tool", EffectRow::pure());

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_destructive_blocked_without_confirm() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "destructive_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                destructive: true,
                ..Default::default()
            },
        );

        let result = pipeline
            .dispatch(&tool, &mut ctx, serde_json::json!({}))
            .await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("destructive"));
                assert!(msg.contains("confirm"));
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_destructive_allowed_with_confirm() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "destructive_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                destructive: true,
                ..Default::default()
            },
        );

        let result = pipeline
            .dispatch(&tool, &mut ctx, serde_json::json!({"confirm": true}))
            .await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_destructive_blocked_with_false_confirm() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "destructive_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                destructive: true,
                ..Default::default()
            },
        );

        let result = pipeline
            .dispatch(&tool, &mut ctx, serde_json::json!({"confirm": false}))
            .await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn pipeline_compartment_no_restriction_allows_all() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        // No compartment set — full access
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_compartment_sandbox_blocks_write_to_codex() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("sandbox".into());
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("sandbox"));
                assert!(msg.contains("codex"));
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_asserted_user_id_confers_no_authority() {
        // P-DEPUTY-2 (2026-09-10, Glama confused-deputy series): the
        // client-asserted `_meta.user_id` label is attribution only — it
        // must never widen compartment authority. A sandbox dispatch
        // labeled as any privileged user is still a sandbox dispatch.
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("sandbox".into());
        ctx.user_id = Some("ceo".into());
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("sandbox"));
                assert!(msg.contains("codex"));
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_routes_store_scoped_tools_through_executor() {
        // P-SANDBOX-3 (Landlock v1): `StoreScoped` marks route through the
        // executor when one is attached; plain tools keep the ambient path.
        use crate::sandbox_exec::ScopedSandboxExecutor;
        use std::sync::atomic::{AtomicU64, Ordering};
        let calls = Arc::new(AtomicU64::new(0));
        let counter = Arc::clone(&calls);
        let executor = Arc::new(ScopedSandboxExecutor::new(move || {
            counter.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }));
        let pipeline =
            DispatchPipeline::with_defaults().with_sandbox_executor(Some(Arc::clone(&executor)));
        let mut ctx = Context::new(BrainWave::Gamma);

        let scoped = TestTool::new(
            "scoped_tool",
            EffectRow {
                sandbox: Sandbox::StoreScoped,
                ..Default::default()
            },
        );
        assert!(
            pipeline
                .dispatch(&scoped, &mut ctx, Args::default())
                .await
                .is_ok()
        );
        assert_eq!(calls.load(Ordering::SeqCst), 1, "scoped tool must confine");

        let plain = TestTool::new("plain_tool", EffectRow::pure());
        assert!(
            pipeline
                .dispatch(&plain, &mut ctx, Args::default())
                .await
                .is_ok()
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "plain tools must not ride the sandbox path"
        );
        assert_eq!(executor.stats(), (1, 0, 0));

        // A scoped tool with no executor attached is inert (v0 behavior).
        let bare = DispatchPipeline::with_defaults();
        let scoped2 = TestTool::new(
            "scoped_tool",
            EffectRow {
                sandbox: Sandbox::StoreScoped,
                ..Default::default()
            },
        );
        assert!(
            bare.dispatch(&scoped2, &mut ctx, Args::default())
                .await
                .is_ok()
        );
    }

    #[tokio::test]
    async fn pipeline_injects_subprocess_policy_and_discloses() {
        // B2: declared `Sandbox::Subprocess` tools get a runner-backed
        // policy on their context before the call, and the active runner is
        // disclosed on the response.
        use crate::subprocess_sandbox::SubprocessSandbox;
        use std::path::PathBuf;
        use wm_core::sandbox::RunnerSource;
        let sandbox = Arc::new(SubprocessSandbox::with_runner(Some(
            wm_core::sandbox::RunnerInfo {
                path: PathBuf::from("/opt/mandala-sandbox"),
                source: RunnerSource::Env,
            },
        )));
        let pipeline =
            DispatchPipeline::with_defaults().with_subprocess_sandbox(Some(Arc::clone(&sandbox)));
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "spawn_tool",
            EffectRow {
                reads: vec![wm_core::Resource::Network, wm_core::Resource::Process],
                spawns: true,
                sandbox: Sandbox::Subprocess,
                ..Default::default()
            },
        )
        .with_output(serde_json::json!({"ok": true}));

        let out = pipeline
            .dispatch(&tool, &mut ctx, Args::default())
            .await
            .expect("declared spawn tool dispatches");
        assert!(ctx.spawn.is_active(), "policy must ride the context");
        assert!(ctx.spawn.allow_net(), "network read grants the runner net");
        assert_eq!(out["sandbox"]["runner"], "/opt/mandala-sandbox");
        assert_eq!(out["sandbox"]["net"], true);
        assert_eq!(
            out["sandbox"]["envelope"],
            wm_core::sandbox::ENVELOPE_SCHEMA
        );
        assert_eq!(sandbox.status()["dispatches"], 1);
        assert_eq!(sandbox.status()["degraded"], 0);
    }

    #[tokio::test]
    async fn pipeline_degrades_loudly_when_runner_missing() {
        // No runner resolvable: the declared tool still runs (availability
        // first), the dispatch is counted, and no confinement is claimed.
        use crate::subprocess_sandbox::SubprocessSandbox;
        let sandbox = Arc::new(SubprocessSandbox::with_runner(None));
        let pipeline =
            DispatchPipeline::with_defaults().with_subprocess_sandbox(Some(Arc::clone(&sandbox)));
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "spawn_tool",
            EffectRow {
                reads: vec![wm_core::Resource::Process],
                spawns: true,
                sandbox: Sandbox::Subprocess,
                ..Default::default()
            },
        )
        .with_output(serde_json::json!({"ok": true}));

        let out = pipeline
            .dispatch(&tool, &mut ctx, Args::default())
            .await
            .expect("degrade keeps availability up");
        assert!(!ctx.spawn.is_active());
        assert!(
            out.get("sandbox").is_none(),
            "no runner means no confinement claim"
        );
        assert_eq!(sandbox.status()["dispatches"], 1);
        assert_eq!(sandbox.status()["degraded"], 1);
    }

    #[tokio::test]
    async fn pipeline_surfaces_unmigrated_spawn_tools() {
        // A tool that declares raw `spawns` without adopting the
        // `Sandbox::Subprocess` contract is counted and warned — the seam
        // must not silently pretend coverage it does not have.
        use crate::subprocess_sandbox::SubprocessSandbox;
        use std::path::PathBuf;
        use wm_core::sandbox::RunnerSource;
        let sandbox = Arc::new(SubprocessSandbox::with_runner(Some(
            wm_core::sandbox::RunnerInfo {
                path: PathBuf::from("/opt/mandala-sandbox"),
                source: RunnerSource::Env,
            },
        )));
        let pipeline =
            DispatchPipeline::with_defaults().with_subprocess_sandbox(Some(Arc::clone(&sandbox)));
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "legacy_git_tool",
            EffectRow {
                reads: vec![wm_core::Resource::Process],
                spawns: true,
                ..Default::default()
            },
        );

        assert!(
            pipeline
                .dispatch(&tool, &mut ctx, Args::default())
                .await
                .is_ok()
        );
        assert!(!ctx.spawn.is_active());
        assert_eq!(sandbox.status()["unconfined_spawns"], 1);
        assert_eq!(sandbox.status()["dispatches"], 0);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn declared_spawn_executes_through_the_runner_envelope() {
        // End-to-end wrap proof: a tool builds its command through
        // `ctx.spawn.command(...)`, the fake runner receives the JSON
        // envelope on argv, and the envelope carries program/args/net.
        use crate::subprocess_sandbox::SubprocessSandbox;
        use std::os::unix::fs::PermissionsExt;
        use wm_core::sandbox::{RunnerInfo, RunnerSource};

        let dir = tempfile::tempdir().expect("tempdir");
        let marker = dir.path().join("envelope.json");
        let runner = dir.path().join("fake-runner");
        std::fs::write(
            &runner,
            format!(
                "#!/bin/sh\nprintf '%s' \"$2\" > '{}'\nexit 0\n",
                marker.display()
            ),
        )
        .expect("write fake runner");
        std::fs::set_permissions(&runner, std::fs::Permissions::from_mode(0o755))
            .expect("chmod fake runner");

        struct SpawnProbeTool {
            effects: EffectRow,
            stats: ToolStats,
        }
        #[async_trait]
        impl Tool for SpawnProbeTool {
            fn name(&self) -> &str {
                "spawn_probe"
            }
            fn gana(&self) -> Gana {
                Gana::Heart
            }
            fn effects(&self) -> &EffectRow {
                &self.effects
            }
            async fn call(&self, ctx: &mut Context, _args: Args) -> Result<Output> {
                let out = ctx
                    .spawn
                    .command("printf", &["%s", "hi"])
                    .output()
                    .map_err(|e| CoreError::Tool(format!("spawn failed: {e}")))?;
                if !out.status.success() {
                    return Err(CoreError::Tool("wrapped command failed".into()));
                }
                Ok(serde_json::json!({"ok": true}))
            }
            fn stats(&self) -> &ToolStats {
                &self.stats
            }
        }

        let sandbox = Arc::new(SubprocessSandbox::with_runner(Some(RunnerInfo {
            path: runner,
            source: RunnerSource::Env,
        })));
        let pipeline =
            DispatchPipeline::with_defaults().with_subprocess_sandbox(Some(Arc::clone(&sandbox)));
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = SpawnProbeTool {
            effects: EffectRow {
                reads: vec![wm_core::Resource::Network, wm_core::Resource::Process],
                spawns: true,
                sandbox: Sandbox::Subprocess,
                ..Default::default()
            },
            stats: ToolStats::default(),
        };
        let out = pipeline
            .dispatch(&tool, &mut ctx, Args::default())
            .await
            .expect("wrapped spawn succeeds");
        assert_eq!(out["ok"], true);
        assert_eq!(out["sandbox"]["net"], true);

        let captured = std::fs::read_to_string(&marker).expect("runner captured the envelope");
        let envelope: serde_json::Value = serde_json::from_str(&captured).expect("envelope JSON");
        assert_eq!(envelope["schema"], wm_core::sandbox::ENVELOPE_SCHEMA);
        assert_eq!(envelope["program"], "printf");
        assert_eq!(envelope["args"], serde_json::json!(["%s", "hi"]));
        assert_eq!(envelope["net"], true);
    }

    #[tokio::test]
    async fn pipeline_secret_scan_warns_without_blocking() {
        // P-PROV-5/B(c): the output sampler observes but never governs.
        // A credential-shaped successful output dispatches fine and
        // records exactly one hit on the attached sampler.
        use crate::secret_scan::SecretSampler;
        let sampler = Arc::new(SecretSampler::new(1));
        let pipeline =
            DispatchPipeline::with_defaults().with_secret_scan_option(Some(Arc::clone(&sampler)));
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new("key_tool", EffectRow::pure())
            .with_output(serde_json::json!({"data": "key=AKIAIOSFODNN7EXAMPLE"}));
        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok(), "warn-only scan must never block");
        assert_eq!(sampler.stats(), (1, 1, 1));

        // Clean outputs scan without hits.
        let clean = TestTool::new("clean_tool", EffectRow::pure())
            .with_output(serde_json::json!({"results": []}));
        assert!(
            pipeline
                .dispatch(&clean, &mut ctx, Args::default())
                .await
                .is_ok()
        );
        assert_eq!(sampler.stats(), (2, 2, 1));
    }

    #[tokio::test]
    async fn pipeline_compartment_sandbox_blocks_read_from_karma() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("sandbox".into());
        let tool = TestTool::new(
            "read_tool",
            EffectRow::read_only(vec![wm_core::Resource::Galaxy("karma".into())]),
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("sandbox"));
                assert!(msg.contains("karma"));
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_compartment_sandbox_allows_write_to_tutorial() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("sandbox".into());
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("tutorial".into())],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_compartment_sandbox_allows_read_from_research() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("sandbox".into());
        let tool = TestTool::new(
            "read_tool",
            EffectRow::read_only(vec![wm_core::Resource::Galaxy("research".into())]),
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_compartment_production_blocks_read_from_karma() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("production".into());
        let tool = TestTool::new(
            "read_tool",
            EffectRow::read_only(vec![wm_core::Resource::Galaxy("karma".into())]),
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("production"));
                assert!(msg.contains("karma"));
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_compartment_production_allows_write_to_codex() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("production".into());
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_compartment_secure_allows_write_to_codex() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("secure".into());
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_compartment_secure_blocks_read_from_karma() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("secure".into());
        let tool = TestTool::new(
            "read_tool",
            EffectRow::read_only(vec![wm_core::Resource::Galaxy("karma".into())]),
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("secure"));
                assert!(msg.contains("karma"));
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    // ── Resource rules (Yama) pipeline tests ──────────────────────────

    fn rules_with(max_writes: u32, max_repeats: u32) -> Arc<ResourceRules> {
        Arc::new(ResourceRules::new(ResourceRulesConfig {
            max_writes_per_minute: max_writes,
            max_spawns_per_minute: 100,
            max_network_per_minute: 100,
            novelty_window: 50,
            max_repeats,
            require_human_review: false,
        }))
    }

    #[tokio::test]
    async fn pipeline_resource_rules_budget_exceeding_write_refused() {
        let pipeline = DispatchPipeline::with_defaults().with_resource_rules(rules_with(2, 1000));
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        assert!(
            pipeline
                .dispatch(&tool, &mut ctx, Args::default())
                .await
                .is_ok(),
            "first write within budget"
        );
        assert!(
            pipeline
                .dispatch(&tool, &mut ctx, Args::default())
                .await
                .is_ok(),
            "second write within budget"
        );
        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err(), "third write must exceed the budget");
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("resource rules"), "got: {msg}");
                assert!(msg.contains("writes"), "got: {msg}");
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_resource_rules_novelty_flag_reaches_response() {
        let pipeline = DispatchPipeline::with_defaults().with_resource_rules(rules_with(1000, 1));
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new("read_tool", EffectRow::pure())
            .with_output(serde_json::json!({"status": "ok"}));

        let first = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(first.is_ok());
        assert!(
            first.unwrap().get("resource_flags").is_none(),
            "first call is novel — no flag"
        );

        let second = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        let output = second.expect("repeated call must still succeed (flag, not block)");
        let flags = output
            .get("resource_flags")
            .and_then(|f| f.as_array())
            .expect("novelty flag must reach the response");
        assert_eq!(flags.len(), 1);
        assert!(flags[0].as_str().unwrap().contains("not novel"));
    }

    #[tokio::test]
    async fn pipeline_resource_rules_blocks_unapproved_autonomous() {
        let rules = Arc::new(ResourceRules::default());
        rules.set_user_initiated(false);
        let pipeline = DispatchPipeline::with_defaults().with_resource_rules(rules);
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "memory.consolidate",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("human review"), "got: {msg}");
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_resource_rules_allows_approved_autonomous() {
        let rules = Arc::new(ResourceRules::default());
        rules.set_user_initiated(false);
        rules.set_human_approved(true);
        let pipeline = DispatchPipeline::with_defaults().with_resource_rules(rules);
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "memory.consolidate",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline
            .dispatch(
                &tool,
                &mut ctx,
                serde_json::json!({"purpose": "consolidate codex"}),
            )
            .await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_resource_rules_user_initiated_writes_allowed_by_default() {
        // Default rules: user-initiated actions are not gated by human review.
        let pipeline = DispatchPipeline::with_defaults()
            .with_resource_rules(Arc::new(ResourceRules::default()));
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "write_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());
    }

    // ── Runtime Satya (fabrication) tests ─────────────────────────────

    #[tokio::test]
    async fn pipeline_runtime_satya_blocks_citta_write_without_read() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "memory.create",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline
            .dispatch(&tool, &mut ctx, serde_json::json!({"galaxy": "citta"}))
            .await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("VIOLATION_SATYA"), "got: {msg}");
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_runtime_satya_allows_citta_write_with_read_evidence() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "consolidate_tool",
            EffectRow {
                reads: vec![wm_core::Resource::Galaxy("citta".into())],
                writes: vec![wm_core::Resource::Galaxy("citta".into())],
                ..Default::default()
            },
        );

        let result = pipeline
            .dispatch(&tool, &mut ctx, serde_json::json!({"galaxy": "citta"}))
            .await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_runtime_satya_allows_non_citta_runtime_galaxy() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "memory.create",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline
            .dispatch(&tool, &mut ctx, serde_json::json!({"galaxy": "research"}))
            .await;
        assert!(result.is_ok());
    }

    // ── Write-audit journal pipeline tests ────────────────────────────

    #[tokio::test]
    async fn pipeline_write_audit_detects_misdeclaring_tool() {
        let tmp = tempfile::tempdir().unwrap();
        let store = Arc::new(wm_memory::MemoryStore::open_default(tmp.path()).unwrap());
        let journal = Arc::new(WriteAuditJournal::with_flush_threshold(store.clone(), 0).unwrap());
        let pipeline = DispatchPipeline::with_defaults().with_write_audit(journal.clone());
        let mut ctx = Context::new(BrainWave::Gamma);

        // Declares a pure effect row but actually writes to the store.
        let tool = TestTool::new("sneaky_tool", EffectRow::pure()).with_store(store);

        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());

        let mis = journal.misdeclarations().unwrap();
        assert!(!mis.is_empty(), "misdeclaring tool must be detected");
        assert_eq!(mis.last().unwrap().tool, "sneaky_tool");
        assert!(mis.last().unwrap().undeclared_mutation());
    }

    #[tokio::test]
    async fn pipeline_write_audit_skips_meta_router() {
        let tmp = tempfile::tempdir().unwrap();
        let store = Arc::new(wm_memory::MemoryStore::open_default(tmp.path()).unwrap());
        let journal = Arc::new(WriteAuditJournal::with_flush_threshold(store.clone(), 0).unwrap());
        let pipeline = DispatchPipeline::with_defaults().with_write_audit(journal.clone());
        let mut ctx = Context::new(BrainWave::Gamma);

        // The meta-router mutates through nested dispatches (which journal
        // the real tool); its own entry must never be flagged as an
        // undeclared mutation (first-run feedback regression, 2026-09-13).
        let tool = TestTool::new("wm", EffectRow::pure()).with_store(store);
        let result = pipeline.dispatch(&tool, &mut ctx, Args::default()).await;
        assert!(result.is_ok());

        let mis = journal.misdeclarations().unwrap();
        assert!(
            mis.iter().all(|m| m.tool != "wm"),
            "meta router must not appear as a misdeclaration: {mis:?}"
        );
    }

    #[tokio::test]
    async fn pipeline_write_audit_records_declared_writes_with_identity() {
        let tmp = tempfile::tempdir().unwrap();
        let store = Arc::new(wm_memory::MemoryStore::open_default(tmp.path()).unwrap());
        let journal = Arc::new(WriteAuditJournal::with_flush_threshold(store.clone(), 0).unwrap());
        let pipeline = DispatchPipeline::with_defaults().with_write_audit(journal.clone());
        let mut ctx = Context::new(BrainWave::Gamma);

        let tool = TestTool::new(
            "honest_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        )
        .with_store(store);

        let args = serde_json::json!({"id": "abc-123", "content_hash": "hash-xyz"});
        let result = pipeline.dispatch(&tool, &mut ctx, args).await;
        assert!(result.is_ok());

        let entries = journal.scan_entries().unwrap();
        assert_eq!(entries.len(), 1);
        let entry = &entries[0];
        assert!(entry.declared_writes);
        assert!(entry.store_write_delta >= 1);
        assert_eq!(entry.memory_id.as_deref(), Some("abc-123"));
        assert_eq!(entry.content_hash.as_deref(), Some("hash-xyz"));
        assert!(journal.misdeclarations().unwrap().is_empty());
    }

    #[tokio::test]
    async fn pipeline_write_audit_captures_actor_identity() {
        // S11b: the journal answers "which agent did this" — identity rides
        // the Context (_meta-derived) into every entry.
        let tmp = tempfile::tempdir().unwrap();
        let store = Arc::new(wm_memory::MemoryStore::open_default(tmp.path()).unwrap());
        let journal = Arc::new(WriteAuditJournal::with_flush_threshold(store.clone(), 0).unwrap());
        let pipeline = DispatchPipeline::with_defaults().with_write_audit(journal.clone());
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.session_id = Some(uuid::Uuid::nil());
        ctx.user_id = Some("agent-b".to_string());
        ctx.compartment = Some("production".to_string());

        let tool = TestTool::new(
            "honest_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        )
        .with_store(store);

        let result = pipeline
            .dispatch(&tool, &mut ctx, serde_json::json!({"id": "abc-123"}))
            .await;
        assert!(result.is_ok());

        let entries = journal.scan_entries().unwrap();
        assert_eq!(entries.len(), 1);
        let entry = &entries[0];
        assert_eq!(
            entry.actor_session.as_deref(),
            Some(uuid::Uuid::nil().to_string().as_str())
        );
        assert_eq!(entry.actor_user.as_deref(), Some("agent-b"));
        assert_eq!(entry.actor_compartment.as_deref(), Some("production"));
    }

    #[tokio::test]
    async fn pipeline_write_audit_read_dispatch_not_flagged_after_external_writes() {
        // The 2026-08-28 restore-drill false positive: a parallel session's
        // writes land before (or while) an honest read-only dispatch runs;
        // the old since-last-entry attribution flagged the read tool with
        // the other dispatch's write count. Per-dispatch baselines close it.
        let tmp = tempfile::tempdir().unwrap();
        let store = Arc::new(wm_memory::MemoryStore::open_default(tmp.path()).unwrap());
        let journal = Arc::new(WriteAuditJournal::with_flush_threshold(store.clone(), 0).unwrap());
        let pipeline = DispatchPipeline::with_defaults().with_write_audit(journal.clone());
        let mut ctx = Context::new(BrainWave::Gamma);

        // The other session's traffic lands before this dispatch starts.
        for i in 0..3 {
            let mem = wm_memory::Memory::new(wm_core::Galaxy::Codex, format!("other session {i}"));
            store.put(wm_core::Galaxy::Codex, &mem).unwrap();
        }

        let read_tool = TestTool::new("memory.search", EffectRow::pure());
        let result = pipeline
            .dispatch(&read_tool, &mut ctx, Args::default())
            .await;
        assert!(result.is_ok());

        let mis = journal.misdeclarations().unwrap();
        assert!(
            mis.is_empty(),
            "read-only dispatch must not inherit the other session's writes: {mis:?}"
        );
        let entries = journal.scan_entries().unwrap();
        assert_eq!(entries.last().unwrap().store_write_delta, 0);
    }

    // ── Firebreak (P1.4 forbidden-command veto + P1.6 bulk-scope law) ──

    #[tokio::test]
    async fn pipeline_firebreak_forbidden_blocks_even_with_confirm() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "destructive_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                destructive: true,
                ..Default::default()
            },
        );

        let result = pipeline
            .dispatch(
                &tool,
                &mut ctx,
                serde_json::json!({"confirm": true, "cmd": "rm -rf /"}),
            )
            .await;
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("FORBIDDEN"), "got: {msg}");
                assert!(msg.contains("never allowed"), "got: {msg}");
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_firebreak_scope_law_blocks_unscoped_destructive() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        // Named like the real tool so the scope registry entry applies.
        let tool = TestTool::new(
            "memory.delete",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                destructive: true,
                ..Default::default()
            },
        );

        let result = pipeline
            .dispatch(&tool, &mut ctx, serde_json::json!({"confirm": true}))
            .await;
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("no explicit scope"), "got: {msg}");
                assert!(msg.contains("id"), "names the scope field: {msg}");
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_firebreak_scope_law_allows_scoped_destructive() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "memory.delete",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                destructive: true,
                ..Default::default()
            },
        );

        let result = pipeline
            .dispatch(
                &tool,
                &mut ctx,
                serde_json::json!({"confirm": true, "id": "0f0e0d0c-0000-0000-0000-000000000000"}),
            )
            .await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_firebreak_caution_disclosed_in_response() {
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "galaxy.transfer",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                destructive: true,
                ..Default::default()
            },
        )
        .with_output(serde_json::json!({"status": "success"}));

        let result = pipeline
            .dispatch(
                &tool,
                &mut ctx,
                serde_json::json!({"confirm": true, "from_galaxy": "codex", "note": "mv old new"}),
            )
            .await;
        let output = result.expect("caution must not block");
        let advisories = output
            .get("firebreak")
            .and_then(|f| f.get("advisories"))
            .and_then(|a| a.as_array())
            .expect("advisories must reach the response");
        assert_eq!(advisories.len(), 1);
    }

    #[tokio::test]
    async fn pipeline_firebreak_dangerous_escalates_off_confirm_gate() {
        // A spawn-class seam tool that is NOT destructive-flagged: the
        // confirm gate (4b) never fires, but a dangerous payload in args
        // must still demand explicit confirm — the confirm-gate hardening.
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "spawn_tool",
            EffectRow {
                spawns: true,
                ..Default::default()
            },
        );

        let blocked = pipeline
            .dispatch(
                &tool,
                &mut ctx,
                serde_json::json!({"cmd": "sudo rm -r /tmp/build"}),
            )
            .await;
        match blocked {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("dangerous"), "got: {msg}");
                assert!(msg.contains("confirm"), "got: {msg}");
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }

        let allowed = pipeline
            .dispatch(
                &tool,
                &mut ctx,
                serde_json::json!({"cmd": "sudo rm -r /tmp/build", "confirm": true}),
            )
            .await;
        assert!(allowed.is_ok());
    }

    #[tokio::test]
    async fn pipeline_firebreak_never_scans_prose() {
        // The seam is irreversible dispatches — a memory-create-style tool
        // recording an incident note quoting a forbidden command must pass.
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "memory.create",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let result = pipeline
            .dispatch(
                &tool,
                &mut ctx,
                serde_json::json!({"content": "incident: operator ran rm -rf / on the store"}),
            )
            .await;
        assert!(result.is_ok(), "prose is never vetoed");
    }

    #[tokio::test]
    async fn pipeline_firebreak_disarmable_per_pipeline() {
        let pipeline = DispatchPipeline::with_defaults()
            .with_firebreak_option(None::<Arc<wm_governance::Firebreak>>);
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "destructive_tool",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                destructive: true,
                ..Default::default()
            },
        );

        // Confirm gate still fires (it is outside the firebreak).
        let result = pipeline
            .dispatch(&tool, &mut ctx, serde_json::json!({}))
            .await;
        assert!(result.is_err());

        // But the forbidden-command veto is gone.
        let result = pipeline
            .dispatch(
                &tool,
                &mut ctx,
                serde_json::json!({"confirm": true, "cmd": "rm -rf /"}),
            )
            .await;
        assert!(result.is_ok(), "disarmed pipeline must not veto");
    }

    #[tokio::test]
    async fn pipeline_write_audit_records_destructive_confirm() {
        // The delete-confirm audit (P1.6): a destructive dispatch's journal
        // entry answers "was this confirmed?".
        let tmp = tempfile::tempdir().unwrap();
        let store = Arc::new(wm_memory::MemoryStore::open_default(tmp.path()).unwrap());
        let journal = Arc::new(WriteAuditJournal::with_flush_threshold(store.clone(), 0).unwrap());
        let pipeline = DispatchPipeline::with_defaults().with_write_audit(journal.clone());
        let mut ctx = Context::new(BrainWave::Gamma);

        let tool = TestTool::new(
            "memory.delete",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                destructive: true,
                ..Default::default()
            },
        )
        .with_store(store);

        let result = pipeline
            .dispatch(
                &tool,
                &mut ctx,
                serde_json::json!({"confirm": true, "id": "abc-123"}),
            )
            .await;
        assert!(result.is_ok());

        let entries = journal.scan_entries().unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(
            entries[0].confirmed,
            Some(true),
            "destructive entry must record the confirm"
        );
    }

    // ── Runtime galaxy argument enforcement tests ──────────────────────

    #[tokio::test]
    async fn pipeline_compartment_production_blocks_runtime_galaxy_write_bypass() {
        // Tool declares writes to "codex" (allowed for production) but runtime
        // galaxy arg is "karma" — production should be blocked from writing karma.
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("production".into());
        let tool = TestTool::new(
            "memory_update",
            EffectRow {
                writes: vec![wm_core::Resource::Galaxy("codex".into())],
                ..Default::default()
            },
        );

        let args = serde_json::json!({"galaxy": "karma"});
        let result = pipeline.dispatch(&tool, &mut ctx, args).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("production"));
                assert!(msg.contains("karma"));
                assert!(msg.contains("runtime"));
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_compartment_production_blocks_runtime_galaxy_read_bypass() {
        // Tool declares reads from "codex" (allowed for production) but runtime
        // galaxy arg is "karma" — production should be blocked from reading karma.
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("production".into());
        let tool = TestTool::new(
            "memory_read",
            EffectRow::read_only(vec![wm_core::Resource::Galaxy("codex".into())]),
        );

        let args = serde_json::json!({"galaxy": "karma"});
        let result = pipeline.dispatch(&tool, &mut ctx, args).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("production"));
                assert!(msg.contains("karma"));
                assert!(msg.contains("runtime"));
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn pipeline_compartment_production_allows_runtime_galaxy_same_as_declared() {
        // Tool declares reads from "codex" and runtime galaxy arg is also "codex"
        // — production should allow this (no duplicate check needed).
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("production".into());
        let tool = TestTool::new(
            "memory_read",
            EffectRow::read_only(vec![wm_core::Resource::Galaxy("codex".into())]),
        );

        let args = serde_json::json!({"galaxy": "codex"});
        let result = pipeline.dispatch(&tool, &mut ctx, args).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_compartment_no_restriction_allows_runtime_galaxy() {
        // No compartment — runtime galaxy arg should be allowed regardless.
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        let tool = TestTool::new(
            "memory_read",
            EffectRow::read_only(vec![wm_core::Resource::Galaxy("codex".into())]),
        );

        let args = serde_json::json!({"galaxy": "karma"});
        let result = pipeline.dispatch(&tool, &mut ctx, args).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_compartment_production_allows_runtime_memory_galaxy() {
        // Production compartment — runtime galaxy arg "codex" should be allowed
        // since production can access all memory galaxies.
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("production".into());
        let tool = TestTool::new(
            "memory_read",
            EffectRow::read_only(vec![wm_core::Resource::Galaxy("codex".into())]),
        );

        let args = serde_json::json!({"galaxy": "research"});
        let result = pipeline.dispatch(&tool, &mut ctx, args).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn pipeline_compartment_production_blocks_runtime_system_galaxy() {
        // Production compartment — runtime galaxy arg "karma" should be blocked
        // since production can't access system galaxies.
        let pipeline = DispatchPipeline::with_defaults();
        let mut ctx = Context::new(BrainWave::Gamma);
        ctx.compartment = Some("production".into());
        let tool = TestTool::new(
            "memory_read",
            EffectRow::read_only(vec![wm_core::Resource::Galaxy("codex".into())]),
        );

        let args = serde_json::json!({"galaxy": "karma"});
        let result = pipeline.dispatch(&tool, &mut ctx, args).await;
        assert!(result.is_err());
        match result {
            Err(CoreError::Governance(msg)) => {
                assert!(msg.contains("production"));
                assert!(msg.contains("karma"));
                assert!(msg.contains("runtime"));
            }
            other => panic!("Expected Governance error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn benchmark_pipeline_overhead() {
        let pipeline = DispatchPipeline::with_defaults();
        let tool = TestTool::new("bench_tool", EffectRow::pure());
        let args = Args::default();

        // Warm up
        for _ in 0..100 {
            let mut ctx = Context::new(BrainWave::Gamma);
            let _ = pipeline.dispatch(&tool, &mut ctx, args.clone()).await;
        }

        // Measure pipeline dispatch
        let n = 10_000;
        let start = std::time::Instant::now();
        for _ in 0..n {
            let mut ctx = Context::new(BrainWave::Gamma);
            let _ = pipeline.dispatch(&tool, &mut ctx, args.clone()).await;
        }
        let pipeline_ns = start.elapsed().as_nanos() / n;

        // Measure direct tool call (no pipeline)
        let start = std::time::Instant::now();
        for _ in 0..n {
            let mut ctx = Context::new(BrainWave::Gamma);
            let _ = tool.call(&mut ctx, args.clone()).await;
        }
        let direct_ns = start.elapsed().as_nanos() / n;

        let overhead_ns = pipeline_ns.saturating_sub(direct_ns);
        println!(
            "\n  Pipeline: {pipeline_ns} ns/call | Direct: {direct_ns} ns/call | Overhead: {overhead_ns} ns/call"
        );

        // Pipeline overhead should be under 5µs per call (5000 ns) in release builds.
        // Debug builds have unoptimized async/await overhead, so we only assert
        // when compiled with optimizations.
        #[cfg(not(debug_assertions))]
        assert!(
            overhead_ns < 5_000,
            "Pipeline overhead {overhead_ns} ns/call exceeds 5µs budget"
        );
    }
}