sema-workflow 1.36.0

//! Run-scoped dynamic context for a workflow run.
//!
//! A workflow run installs a [`WorkflowCtx`] as a scope on the OWNING TASK via
//! [`set_workflow_scope`]; every builtin (`workflow/phase`, `checkpoint`, …) reaches the
//! live context through [`current_for`], reading the task-local [`WorkflowTaskState`]
//! extension. The scope is a token-keyed stack entry restored on drop via a panic-safe
//! RAII guard (mirrors `DynamicTaskState`'s `ScopeId` removal), so a nested run — or a
//! panic unwinding through a phase thunk — cannot leave a stale context installed, and a
//! sibling task interleaved on the same thread never observes another task's run.
//!
//! `WorkflowCtx` holds live checkpoint/memo/MCP `Value`s, so the extension is TRACED
//! (Invariant I2): the `TaskContextHandle` traces each extension, `WorkflowTaskState`
//! traces its scope stack, and each `WorkflowCtx` traces its `Value`-bearing bags.
//!
//! The `WORKFLOW` thread-local survives ONLY as the HOST-ADAPTER fallback for callers
//! outside a runtime quantum (a synchronous host `call_function`, the non-runtime
//! restricted VM); it is read only when [`sema_core::in_runtime_quantum`] is false.
//!
//! The context owns the run's monotonic `seq` counter, the wall-clock seam (`ts` /
//! `dur_ms`, both frozen under `SEMA_WORKFLOW_FIXED_TS` for byte-identical goldens),
//! the append-only [`Journal`], and a Mastra-style checkpoint/state bag.

use std::any::Any;
use std::cell::{Cell, RefCell};
use std::collections::{BTreeMap, HashMap};
use std::fmt::Write as _;
use std::io;
use std::path::Path;
use std::rc::{Rc, Weak};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc::Receiver;
use std::time::{Instant, SystemTime, UNIX_EPOCH};

use sema_core::cycle::GcEdge;
use sema_core::runtime::{IdCounter, ScopeId, TaskContextHandle, TaskLocalValue, Trace};
use sema_core::Value;

use crate::event::WorkflowEvent;
use crate::journal::Journal;
use crate::RUNS_ROOT;

/// Env var that pins the timestamp string AND forces every `dur_ms` to 0, so the
/// golden `events.jsonl` is byte-identical across runs. When set, its value is used
/// verbatim as the `ts` field of every event (e.g. `SEMA_WORKFLOW_FIXED_TS=0`).
const FIXED_TS_ENV: &str = "SEMA_WORKFLOW_FIXED_TS";

/// Env var that pins the run id (otherwise a process-derived id is generated). Used
/// both to name the run directory and to seed the journal path.
const RUN_ID_ENV: &str = "SEMA_WORKFLOW_RUN_ID";

/// Env var that overrides the run-directory base (the CLI sets it from `--run-dir`).
/// Default is [`RUNS_ROOT`] (`./.sema/runs`).
const RUN_DIR_ENV: &str = "SEMA_WORKFLOW_RUN_DIR";

/// A3 hard caps captured before any VM-thread encode. They bound the CPU/memory a single
/// leaf can spend materializing state on the quantum; the on-disk writes themselves are
/// off the VM thread (see `writer.rs`).
///
/// Max memos stored per run. Past it a leaf is simply not memoized (it re-runs on
/// resume) — the same fallback the round-trip guard already uses.
pub const MEMO_MAX_COUNT: u64 = 4096;
/// Max serialized bytes per memo. An over-cap value is NOT stored (never JSON-encoded in
/// full on the VM thread — the compact form is bounded-checked first).
pub const MEMO_FILE_MAX_BYTES: usize = 1 << 20; // 1 MiB
/// Cap for the `value_digest` bounded encode. A value larger than this gets a stable
/// marker digest instead of a full JSON materialization on the VM thread.
const DIGEST_MAX_BYTES: usize = 1 << 20; // 1 MiB

/// A `fmt::Write` sink that accepts at most `cap` bytes of a value's compact `Display`
/// form and then aborts (returns `fmt::Error`), so a huge value is never fully
/// materialized on the VM thread. Char-boundary safe.
struct CappedWriter {
    buf: String,
    cap: usize,
    truncated: bool,
}

impl std::fmt::Write for CappedWriter {
    fn write_str(&mut self, s: &str) -> std::fmt::Result {
        if self.truncated {
            return Err(std::fmt::Error);
        }
        let remaining = self.cap.saturating_sub(self.buf.len());
        if s.len() <= remaining {
            self.buf.push_str(s);
            Ok(())
        } else {
            let mut end = remaining;
            while end > 0 && !s.is_char_boundary(end) {
                end -= 1;
            }
            self.buf.push_str(&s[..end]);
            self.truncated = true;
            Err(std::fmt::Error)
        }
    }
}

/// Render `v`'s compact `Display` form into at most `cap` bytes. Returns `(text,
/// truncated)`: `truncated` ⇒ `v` exceeds `cap` and was NOT fully materialized (rendering
/// aborted at the cap). For a `v` within `cap`, `text` is its exact compact form. Shared
/// by the rendered-value / digest / memo caps so none of them can be tricked into
/// materializing an unbounded value on the quantum.
pub fn compact_capped(v: &Value, cap: usize) -> (String, bool) {
    let mut w = CappedWriter {
        buf: String::new(),
        cap,
        truncated: false,
    };
    let _ = write!(w, "{v}");
    (w.buf, w.truncated)
}

thread_local! {
    /// HOST-ADAPTER-ONLY fallback scope store for callers outside a runtime quantum.
    /// A per-thread [`WorkflowTaskState`] (same shape as the task-local extension) that
    /// only synchronous host paths install into / read from; the runtime path lives on
    /// the owning task's [`TaskContextHandle`] instead. Read only when
    /// `!sema_core::in_runtime_quantum()`.
    static WORKFLOW: Rc<WorkflowTaskState> = Rc::new(WorkflowTaskState::default());
    /// Immutable host-owned run configuration. The CLI installs this around evaluation;
    /// Sema code can mutate process environment variables but cannot reach this slot.
    static HOST_CONFIG: RefCell<Option<WorkflowHostConfig>> = const { RefCell::new(None) };
}

#[derive(Debug, Clone)]
pub struct WorkflowHostConfig {
    pub runs_root: String,
    pub explicit_run_id: Option<String>,
    pub resuming: bool,
    pub code_version: String,
    pub approval_code_version: String,
    pub args_json: String,
    pub approval_public_key: String,
    pub entry_file: String,
    pub workspace_root: String,
}

pub struct WorkflowHostConfigGuard {
    previous: Option<WorkflowHostConfig>,
}

impl Drop for WorkflowHostConfigGuard {
    fn drop(&mut self) {
        HOST_CONFIG.with(|slot| {
            *slot.borrow_mut() = self.previous.take();
        });
    }
}

pub fn install_host_config(config: WorkflowHostConfig) -> WorkflowHostConfigGuard {
    let previous = HOST_CONFIG.with(|slot| slot.borrow_mut().replace(config));
    WorkflowHostConfigGuard { previous }
}

fn host_config() -> Option<WorkflowHostConfig> {
    HOST_CONFIG.with(|slot| slot.borrow().clone())
}

pub fn host_workspace_root() -> Option<std::path::PathBuf> {
    host_config().map(|config| config.workspace_root.into())
}

/// Run-scoped dynamic context. Cheap to clone-share via `Rc`; all interior state is
/// `RefCell`/`Cell`, never `&mut self`, so the same `Rc<WorkflowCtx>` handed out by
/// [`current`] can be used while the run is still executing.
pub struct WorkflowCtx {
    /// Stable identifier for this run; names the run dir (`./.sema/runs/<run_id>/`).
    pub run_id: String,
    /// Declared `defworkflow` name. Stored separately from the run id so approval
    /// requests can bind decisions to the workflow definition that produced them.
    workflow_name: RefCell<String>,
    /// Append-only JSONL journal sink. `RefCell` because `emit` needs `&mut` access
    /// to the underlying writer while the ctx itself is shared `Rc`.
    journal: Rc<RefCell<Journal>>,
    /// Mastra-style run state / checkpoint bag, keyed by the checkpoint name. Doubles
    /// as the `(checkpoint :files)` read-back store for later phases in the same run.
    state: Rc<RefCell<BTreeMap<String, Value>>>,
    /// Monotonic event sequence counter (0-based; first `next_seq()` returns 0).
    seq: Cell<u64>,
    /// Bounded completion ledger keyed only by the frozen event vocabulary.
    event_counts: RefCell<BTreeMap<&'static str, u64>>,
    /// Wall-clock origin for `dur_ms`. Ignored when the fixed-ts seam is active.
    start: Instant,
    /// Parsed spend caps (absent ⇒ that dimension is unenforced). `usd` is best-effort
    /// (depends on the pricing table); `tokens` is deterministic from usage.
    cost_limit: Option<f64>,
    token_limit: Option<u64>,
    /// Running totals charged from each agent leaf's usage. Single-thread `Cell` is
    /// sound (the VM + scheduler are cooperative single-thread); under a concurrent
    /// fan-out the per-leaf attribution is BEST-EFFORT (the `LAST_USAGE` thread-local
    /// the snapshot reads is not swapped per task), but the cap still trips reliably.
    cost_spent: Cell<f64>,
    tokens_spent: Cell<u64>,
    /// Sticky "a cap was exceeded" latch. Set by [`Self::charge`] once a total passes
    /// its cap; checked at agent ENTRY (to refuse launching further leaves) and by
    /// `workflow/run` after the body (to force a `:failed` envelope). A latch — not
    /// `Err` propagation — because the `__fanout-tagged` engine swallows a leaf `Err`
    /// into `nil`, so an exception can't stop a concurrent batch.
    over_budget: Cell<bool>,
    /// Sticky fail-closed latch for an approval attempted from an invalid child/nested
    /// position. Shared with inherited tasks so the owning run cannot report success
    /// after a detached child tried to create a gate.
    approval_failure: RefCell<Option<String>>,
    /// `(start_seq, label)` of the currently-open marker-style phase — `start_seq` is
    /// the phase.started event's seq, so checkpoints/agents/budget events can be
    /// attributed to their phase; `label` is needed to emit the matching `phase.ended`
    /// when the next marker (or the run end) closes the phase. `None` when no phase
    /// is open.
    cur_phase: RefCell<Option<(u64, String)>>,
    /// Per-name agent invocation counter, for minting unique `agent_id`s. Run-shared
    /// (via the `Rc<WorkflowCtx>`), so ids stay unique even across concurrent tasks; the
    /// per-task ACTIVE-agent attribution slot moved to [`WorkflowTaskState::set_cur_agent`].
    agent_n: RefCell<BTreeMap<String, u64>>,
    /// Resume state. `resuming` ⇒ this run was launched with `--resume`, so leaves whose
    /// content-key is in `resume_memos` short-circuit (return the recorded value, skip
    /// the model + events). `resume_memos` is loaded from the prior run's `memo/` dir at
    /// scope open. `code_version` and the args fingerprint are folded into every
    /// content-key, so a changed workflow or changed args produce different keys ⇒ no
    /// memo hits ⇒ full re-run (automatic invalidation, no guard file). `key_seen`
    /// mints a per-base occurrence ordinal so identical-prompt repeats in source order
    /// line up across runs.
    resuming: Cell<bool>,
    /// Existing short resume fingerprint. Kept stable for memo compatibility.
    code_version: RefCell<String>,
    /// Collision-resistant source fingerprint used to bind human decisions. The CLI
    /// supplies SHA-256; library callers fall back to `code_version`.
    approval_code_version: RefCell<String>,
    /// Ed25519 public key selected by the host before evaluation. Decisions must verify
    /// against this authority; the matching private key is never exposed to Sema code.
    approval_public_key: RefCell<String>,
    resume_memos: RefCell<HashMap<String, Value>>,
    key_seen: RefCell<HashMap<String, u32>>,
    /// Number of memos stored this run, capped at [`MEMO_MAX_COUNT`] so an unbounded fan-out
    /// can't spill an unbounded number of memo sidecars.
    memo_count: Cell<u64>,
    /// The run's `--args` JSON string (for the run.started event). Empty if none.
    args_json: String,
    /// Canonical fingerprint of `--args`, folded into resume content-keys. Kept
    /// separate so `args_json` can remain the operator's original journal text.
    args_fingerprint: String,
    /// Cached fixed-ts override (read once at construction). `Some` ⇒ deterministic
    /// seam: `ts()` returns this string and `dur_ms()` returns 0.
    fixed_ts: Option<String>,
    /// Aliases declared in this run's `:mcp` meta (set once, right after the meta
    /// map's `:mcp` key parses successfully — BEFORE auth-resolution runs), so
    /// `workflow/mcp-handle` can tell "not declared" apart from "declared but this
    /// run hasn't resolved its MCP servers yet" (docs/plans/2026-06-24-workflow-mcp-auth.md
    /// §3). Empty for a workflow with no `:mcp`.
    mcp_declared: RefCell<Vec<String>>,
    /// Opaque, resolved MCP connection handles, keyed by declared alias. Populated
    /// ONCE by `workflow/run`'s auth-resolution step, after every declared server
    /// resolves to `Connected` (never partially — a `NeedsAuth`/`Failed` outcome
    /// ends the run before the body runs at all). Values are `Value`s the resolver
    /// handed back; this crate stays MCP-ignorant and never interprets them —
    /// see `crates/sema-stdlib/src/workflow_mcp.rs`'s resolver seam.
    mcp_handles: RefCell<BTreeMap<String, Value>>,
}

impl WorkflowCtx {
    /// Build a fresh context for a run.
    ///
    /// `run_id` selection (the caller resolves this, but the helper [`resolve_run_id`]
    /// implements the policy): `SEMA_WORKFLOW_RUN_ID` if set, else a generated id.
    pub fn new(
        run_id: String,
        journal: Journal,
        budget: BTreeMap<String, Value>,
    ) -> Rc<WorkflowCtx> {
        Self::new_with_args(run_id, journal, budget, String::new())
    }

    /// As [`Self::new`], plus the run's `--args` JSON string for `run.started`.
    pub fn new_with_args(
        run_id: String,
        journal: Journal,
        budget: BTreeMap<String, Value>,
        args_json: String,
    ) -> Rc<WorkflowCtx> {
        let fixed_ts = std::env::var(FIXED_TS_ENV).ok();
        let args_fingerprint = canonical_args_fingerprint(&args_json);
        // Parse spend caps from the budget submap (tolerate an int usd, e.g. `:usd 2`).
        let cost_limit = budget
            .get("usd")
            .and_then(|v| v.as_float().or_else(|| v.as_int().map(|i| i as f64)));
        // Tolerate an int OR a float token cap (`:tokens 5` or `:tokens 5.0`), so a
        // float never silently drops the cap.
        let token_limit = budget
            .get("tokens")
            .and_then(|v| v.as_int().or_else(|| v.as_float().map(|f| f as i64)))
            .map(|i| i as u64);
        Rc::new(WorkflowCtx {
            run_id,
            workflow_name: RefCell::new(String::new()),
            journal: Rc::new(RefCell::new(journal)),
            state: Rc::new(RefCell::new(BTreeMap::new())),
            seq: Cell::new(0),
            event_counts: RefCell::new(BTreeMap::new()),
            start: Instant::now(),
            cost_limit,
            token_limit,
            cost_spent: Cell::new(0.0),
            tokens_spent: Cell::new(0),
            over_budget: Cell::new(false),
            approval_failure: RefCell::new(None),
            cur_phase: RefCell::new(None),
            agent_n: RefCell::new(BTreeMap::new()),
            resuming: Cell::new(false),
            code_version: RefCell::new(String::new()),
            approval_code_version: RefCell::new(String::new()),
            approval_public_key: RefCell::new(String::new()),
            resume_memos: RefCell::new(HashMap::new()),
            key_seen: RefCell::new(HashMap::new()),
            memo_count: Cell::new(0),
            args_json,
            args_fingerprint,
            fixed_ts,
            mcp_declared: RefCell::new(Vec::new()),
            mcp_handles: RefCell::new(BTreeMap::new()),
        })
    }

    /// The run's `--args` JSON string (empty if none).
    pub fn args_json(&self) -> &str {
        &self.args_json
    }

    /// Bind this context to its declared workflow name. Called once while opening the
    /// scope, before the body can evaluate an approval gate.
    pub fn set_workflow_name(&self, name: impl Into<String>) {
        *self.workflow_name.borrow_mut() = name.into();
    }

    pub fn workflow_name(&self) -> String {
        self.workflow_name.borrow().clone()
    }

    /// Collision-resistant workflow revision used by durable approval requests.
    pub fn approval_code_version(&self) -> String {
        self.approval_code_version.borrow().clone()
    }

    pub fn approval_public_key(&self) -> String {
        self.approval_public_key.borrow().clone()
    }

    /// Full SHA-256 of canonicalized workflow arguments for approval bindings. Resume
    /// keeps its historical short content-key fingerprint; approvals use the full digest.
    pub fn approval_args_digest(&self) -> String {
        let normalized = if self.args_json.trim().is_empty() {
            String::new()
        } else {
            serde_json::from_str::<serde_json::Value>(&self.args_json)
                .ok()
                .and_then(|json| serde_json::to_string(&json).ok())
                .unwrap_or_else(|| self.args_json.clone())
        };
        crate::approval::sha256_bytes(normalized.as_bytes())
    }

    /// Run directory containing the approval authority sidecars.
    pub fn run_dir(&self) -> std::path::PathBuf {
        self.journal.borrow().dir().to_path_buf()
    }

    /// Open a marker-style phase: record its `phase.started` seq AND label so the next
    /// marker (or the run end) can emit the matching `phase.ended`. Subsequent
    /// checkpoints / agents / budget events attribute to `start_seq`.
    pub fn open_phase(&self, start_seq: u64, label: String) {
        *self.cur_phase.borrow_mut() = Some((start_seq, label));
    }

    /// Close the currently-open phase, returning its `(start_seq, label)` so the caller
    /// can emit `phase.ended`. Clears the open-phase tracking; returns `None` when no
    /// phase is open (e.g. a workflow with no `(phase …)` markers).
    pub fn take_open_phase(&self) -> Option<(u64, String)> {
        self.cur_phase.borrow_mut().take()
    }

    /// `start_seq` of the open phase, if any.
    pub fn phase_seq(&self) -> Option<u64> {
        self.cur_phase.borrow().as_ref().map(|(seq, _)| *seq)
    }

    /// Mint a unique `agent_id` for an agent of role `name` (`<name>_<n>`, 1-based).
    pub fn next_agent_id(&self, name: &str) -> String {
        let mut m = self.agent_n.borrow_mut();
        let n = m.entry(name.to_string()).or_insert(0);
        *n += 1;
        format!("{name}_{n}")
    }

    /// A stable short resume key for a checkpoint (`ck_<hex>` over key + digest).
    pub fn content_key(&self, key: &str, value_digest: &str) -> String {
        let h = format!(
            "{:x}",
            md5::compute(format!("{key}:{value_digest}").as_bytes())
        );
        format!("ck_{}", &h[..8])
    }

    /// Next monotonic sequence number (post-increment: first call yields 0).
    pub fn next_seq(&self) -> u64 {
        let n = self.seq.get();
        self.seq.set(n + 1);
        n
    }

    /// Timestamp for an event. Under the fixed-ts seam this is the verbatim env value
    /// (so goldens are byte-identical); otherwise an RFC3339 UTC instant derived from
    /// `SystemTime` (no `chrono` dependency — this crate only pulls `sema-core` +
    /// `sema-otel` + serde).
    pub fn ts(&self) -> String {
        if let Some(ref fixed) = self.fixed_ts {
            return fixed.clone();
        }
        rfc3339_now()
    }

    /// Milliseconds elapsed since `start`. Always 0 under the fixed-ts seam so the
    /// golden does not depend on real timing.
    pub fn dur_ms(&self) -> u64 {
        if self.fixed_ts.is_some() {
            return 0;
        }
        self.start.elapsed().as_millis() as u64
    }

    /// Append one event to the journal. Write errors are swallowed by the journal
    /// (same trust model as the OTel file exporter); journaling never aborts the run.
    pub fn emit(&self, event: WorkflowEvent) {
        let kind = event.kind();
        let mut counts = self.event_counts.borrow_mut();
        *counts.entry(kind).or_insert(0) += 1;
        drop(counts);
        self.journal.borrow().write(&event);
    }

    pub fn has_event(&self, kind: &str) -> bool {
        self.event_counts
            .borrow()
            .get(kind)
            .is_some_and(|count| *count > 0)
    }

    /// True under the fixed-timestamp test seam (`SEMA_WORKFLOW_FIXED_TS`). Callers
    /// that measure their own per-leaf durations force them to 0 in this mode so
    /// goldens stay byte-identical.
    pub fn deterministic(&self) -> bool {
        self.fixed_ts.is_some()
    }

    /// This run's stable identifier (also the run-dir name).
    pub fn run_id(&self) -> String {
        self.run_id.clone()
    }

    /// Store a checkpoint / run-state value under `key`, replacing any prior value.
    pub fn store_checkpoint(&self, key: &str, val: Value) {
        self.state.borrow_mut().insert(key.to_string(), val);
    }

    /// Read a checkpoint / run-state value. `None` if the key was never set in this run.
    pub fn read_checkpoint(&self, key: &str) -> Option<Value> {
        self.state.borrow().get(key).cloned()
    }

    /// Opaque, lossy digest of a checkpoint value for the event stream: the md5 hex
    /// of the value's lossy-JSON encoding. The digest is for journal compactness and
    /// diffing — NOT resume identity (resume keys on the input-derived content-key and
    /// stores the real value in `memo/`, round-trip-guarded). Stable within a process.
    pub fn value_digest(&self, v: &Value) -> String {
        // Bound the work: a value larger than the digest cap is never JSON-encoded in full
        // on the VM thread — it gets a stable marker digest over its bounded compact prefix.
        // The digest is NOT the resume identity (memo content-keys are, round-trip-guarded),
        // and the byte-identical goldens only ever digest tiny values, so a capped path here
        // never changes a golden digest.
        let (compact, truncated) = compact_capped(v, DIGEST_MAX_BYTES);
        if truncated {
            return format!("oversized_{:x}", md5::compute(compact.as_bytes()));
        }
        let json = sema_core::json::value_to_json_lossy(v);
        let bytes = serde_json::to_vec(&json).unwrap_or_default();
        format!("{:x}", md5::compute(bytes))
    }

    /// Write the final `{:status …}` envelope to `result.json` (best-effort; a write
    /// failure is swallowed like a journal write).
    pub fn write_result(&self, envelope: &Value) {
        let json = sema_core::json::value_to_json_lossy(envelope);
        self.journal.borrow().write_result(&json);
    }

    /// True when a `:budget` cap (usd and/or tokens) is in force for this run.
    pub fn has_budget(&self) -> bool {
        self.cost_limit.is_some() || self.token_limit.is_some()
    }

    /// The token cap, for the `budget_limit` field of a `Budget` event (typed `u64`).
    /// `None` for a usd-only budget (the event field is tokens; usd has no slot).
    pub fn budget_limit_for_event(&self) -> Option<u64> {
        self.token_limit
    }

    /// Add one agent leaf's usage to the running totals and, if either cap is now
    /// exceeded, set the sticky [`Self::over_budget`] latch. Returns `true` once the
    /// run is over budget. Charge AFTER the leaf's events are journaled, so the leaf
    /// that tips the cap is itself fully recorded; the NEXT leaf is the one refused.
    pub fn charge(&self, cost: Option<f64>, tokens: u64) -> bool {
        if let Some(c) = cost {
            self.cost_spent.set(self.cost_spent.get() + c);
        }
        self.tokens_spent.set(self.tokens_spent.get() + tokens);
        let over = self
            .cost_limit
            .is_some_and(|lim| self.cost_spent.get() > lim)
            || self
                .token_limit
                .is_some_and(|lim| self.tokens_spent.get() > lim);
        if over {
            self.over_budget.set(true);
        }
        over
    }

    /// Whether a cap has been exceeded this run (the sticky latch).
    pub fn over_budget(&self) -> bool {
        self.over_budget.get()
    }

    pub fn fail_approval(&self, message: impl Into<String>) {
        let mut failure = self.approval_failure.borrow_mut();
        if failure.is_none() {
            *failure = Some(message.into());
        }
    }

    pub fn approval_failure(&self) -> Option<String> {
        self.approval_failure.borrow().clone()
    }

    // ── Resume / content-key memoization ──────────────────────────────────────

    /// Set the workflow's code version (folded into every content-key alongside args).
    /// A changed workflow ⇒ different version ⇒ different keys ⇒ no memo hits ⇒ full
    /// re-run.
    pub fn set_code_version(&self, v: String) {
        *self.code_version.borrow_mut() = v;
    }

    pub fn set_approval_code_version(&self, v: String) {
        *self.approval_code_version.borrow_mut() = v;
    }

    pub fn set_approval_public_key(&self, v: String) {
        *self.approval_public_key.borrow_mut() = v;
    }

    /// Enter resume mode with the prior run's memos (content-key → value).
    pub fn enter_resume(&self, memos: HashMap<String, Value>) {
        self.resuming.set(true);
        *self.resume_memos.borrow_mut() = memos;
    }

    /// True when this run is a `--resume` continuation.
    pub fn resuming(&self) -> bool {
        self.resuming.get()
    }

    /// The label of the currently-open phase (empty outside any phase). Part of a
    /// content-key so the same leaf in different phases keys distinctly.
    pub fn cur_phase_label(&self) -> String {
        self.cur_phase
            .borrow()
            .as_ref()
            .map(|(_, label)| label.clone())
            .unwrap_or_default()
    }

    /// Next 0-based occurrence ordinal for a content-key base, so identical-input leaves
    /// repeated in body order get distinct keys that line up across runs (deterministic
    /// for a sequential body; best-effort under a concurrent fan-out).
    fn next_occurrence(&self, base: &str) -> u32 {
        let mut m = self.key_seen.borrow_mut();
        let n = m.entry(base.to_string()).or_insert(0);
        let cur = *n;
        *n += 1;
        cur
    }

    /// Content-key for an agent leaf: a stable hash over (kind, code-version, args,
    /// phase, name, prompt, schema-repr, effective-policy) plus an occurrence ordinal.
    /// Length-prefixed so `("a","bc")` and `("ab","c")` never collide.
    pub fn agent_content_key(
        &self,
        prompt: &str,
        schema_repr: &str,
        name: &str,
        phase: &str,
        policy_fingerprint: &str,
    ) -> String {
        let cv = self.code_version.borrow().clone();
        let base = hash_fields(&[
            "agent",
            &cv,
            &self.args_fingerprint,
            phase,
            name,
            prompt,
            schema_repr,
            policy_fingerprint,
        ]);
        format!("{base}_{}", self.next_occurrence(&base))
    }

    /// Content-key for a checkpoint write: hash over (kind, code-version, args, phase,
    /// key) plus an occurrence ordinal.
    pub fn checkpoint_content_key(&self, key: &str, phase: &str) -> String {
        let cv = self.code_version.borrow().clone();
        let base = hash_fields(&["checkpoint", &cv, &self.args_fingerprint, phase, key]);
        format!("{base}_{}", self.next_occurrence(&base))
    }

    /// Next occurrence for an explicit approval gate. The base binds the same inputs as
    /// the durable request, so repeated identical gates in deterministic body order get
    /// distinct request ids that line up on resume.
    pub fn approval_occurrence(&self, key: &str, subject_digest: &str, phase: &str) -> u32 {
        let cv = self.approval_code_version.borrow().clone();
        let base = crate::approval::sha256_fields(&[
            "approval",
            &cv,
            &self.args_fingerprint,
            phase,
            key,
            subject_digest,
        ]);
        self.next_occurrence(&base)
    }

    /// Look up a memoized value by content-key (only meaningful while `resuming`).
    pub fn memo_lookup(&self, content_key: &str) -> Option<Value> {
        self.resume_memos.borrow().get(content_key).cloned()
    }

    /// Persist a leaf's value as a memo sidecar AND into the in-run map — but ONLY if it
    /// round-trips through JSON identically (`value_to_json_lossy`→`json_to_value` is
    /// lossy for keyword/string keys, records, typed arrays) AND fits the A3 caps. A value
    /// that doesn't survive, or exceeds [`MEMO_MAX_COUNT`]/[`MEMO_FILE_MAX_BYTES`], is left
    /// un-memoized, so it re-runs on resume rather than resuming wrong. The whole-file
    /// memo write itself is enqueued to the writer thread (no fs on the VM thread).
    pub fn memo_store(&self, content_key: &str, v: &Value) {
        // Cap 1 — per-run memo count.
        if self.memo_count.get() >= MEMO_MAX_COUNT {
            return;
        }
        // Cap 2 (pre-encode) — bound the compact form so an oversized value is never
        // JSON-encoded in full on the VM thread. Truncated ⇒ over-cap ⇒ not stored.
        let (_, truncated) = compact_capped(v, MEMO_FILE_MAX_BYTES);
        if truncated {
            return;
        }
        let json = sema_core::json::value_to_json_lossy(v);
        // Round-trip guard: a value that doesn't survive JSON is left un-memoized.
        if sema_core::json::json_to_value(&json) != *v {
            return;
        }
        // Cap 2 (exact) — a value can be compact-small but JSON-large (deep nesting of
        // short atoms); reject on the serialized size too.
        let serialized = serde_json::to_vec(&json).unwrap_or_default();
        if serialized.len() > MEMO_FILE_MAX_BYTES {
            return;
        }
        self.memo_count.set(self.memo_count.get() + 1);
        self.journal.borrow().write_memo(content_key, &json);
        self.resume_memos
            .borrow_mut()
            .insert(content_key.to_string(), v.clone());
    }

    /// Enqueue a terminal flush barrier, returning the ack receiver WITHOUT waiting — the
    /// runtime terminal path parks on it via an External wait (see `workflow/run`'s
    /// `finish_run`).
    pub fn request_flush(&self) -> Receiver<()> {
        self.journal.borrow().request_flush()
    }

    /// Bounded blocking flush of the journal writer (host / non-quantum path). NEVER call
    /// inside a runtime quantum — park on the External flush-ack instead.
    pub fn flush(&self) {
        self.journal.borrow().flush_blocking();
    }

    // ── MCP handle registry (docs/plans/2026-06-24-workflow-mcp-auth.md §3) ────

    /// Record the aliases declared in this run's `:mcp` meta, BEFORE
    /// auth-resolution runs. `workflow/mcp-handle` uses this to distinguish an
    /// undeclared alias from one that's declared but not resolved yet.
    pub fn set_mcp_declared(&self, aliases: Vec<String>) {
        *self.mcp_declared.borrow_mut() = aliases;
    }

    /// Whether `alias` appears in this run's `:mcp` declarations.
    pub fn is_mcp_declared(&self, alias: &str) -> bool {
        self.mcp_declared.borrow().iter().any(|a| a == alias)
    }

    /// Install the resolved MCP handles for this run — called once, after every
    /// declared server resolves to `Connected` and before the body thunk runs.
    pub fn set_mcp_handles(&self, handles: BTreeMap<String, Value>) {
        *self.mcp_handles.borrow_mut() = handles;
    }

    /// The resolved handle for a declared alias, if any (`None` before
    /// resolution completes, or if `alias` was never declared).
    pub fn mcp_handle(&self, alias: &str) -> Option<Value> {
        self.mcp_handles.borrow().get(alias).cloned()
    }
}

impl Trace for WorkflowCtx {
    /// Expose every live `Value` a run holds so the CORE-2 collector never frees a
    /// checkpoint/memo/MCP handle it can still reach through the owning task (Invariant
    /// I2). A conflicting borrow means the bag is mid-mutation; report incomplete
    /// (`false`) so the collector retries rather than under-tracing.
    fn trace(&self, sink: &mut dyn FnMut(GcEdge<'_>)) -> bool {
        let (Ok(state), Ok(memos), Ok(handles)) = (
            self.state.try_borrow(),
            self.resume_memos.try_borrow(),
            self.mcp_handles.try_borrow(),
        ) else {
            return false;
        };
        for value in state.values() {
            sink(GcEdge::Value(value));
        }
        for value in memos.values() {
            sink(GcEdge::Value(value));
        }
        for value in handles.values() {
            sink(GcEdge::Value(value));
        }
        true
    }
}

/// One published workflow scope on a task's stack. A scope this task INSTALLED carries
/// `Some(token)` and is removed by that exact token (mirrors `DynamicTaskState`'s
/// `ScopeId` removal — out-of-LIFO teardown across interleaved tasks restores the exact
/// outer scope). An INHERITED scope (a spawned child observing its spawner's run) carries
/// `None`: the child sees the workflow for attribution but cannot tear down an ancestor's
/// scope. The `Rc<WorkflowCtx>` is SCOPE-SHARED, so the child journals into the same run.
struct WorkflowScope {
    token: Option<ScopeId>,
    ctx: Rc<WorkflowCtx>,
}

struct WorkflowTaskInner {
    tokens: IdCounter<ScopeId>,
    scopes: Vec<WorkflowScope>,
    /// The `agent_id` of the step currently executing ON THIS TASK, so
    /// `workflow/tool-call` attributes to it. TASK-PRIVATE: two concurrent steps on
    /// sibling tasks keep distinct active agents (no cross-attribution).
    cur_agent: Option<String>,
}

/// Task-local workflow scope: the run stack plus the per-task active-step attribution.
/// Installed on the owning task's [`TaskContextHandle`] (traced), inherited clone-shared
/// by spawned children.
pub struct WorkflowTaskState {
    inner: RefCell<WorkflowTaskInner>,
}

impl Default for WorkflowTaskState {
    fn default() -> Self {
        Self {
            inner: RefCell::new(WorkflowTaskInner {
                tokens: IdCounter::new(),
                scopes: Vec::new(),
                cur_agent: None,
            }),
        }
    }
}

impl WorkflowTaskState {
    /// Push `ctx` as the live scope, minting a fresh removal token for it.
    fn install(&self, ctx: Rc<WorkflowCtx>) -> ScopeId {
        let mut inner = self.inner.borrow_mut();
        let token = inner
            .tokens
            .allocate()
            .expect("workflow scope identity space exhausted");
        inner.scopes.push(WorkflowScope {
            token: Some(token),
            ctx,
        });
        token
    }

    /// Remove the scope carrying exactly `token`. Returns `false` if it is already gone
    /// (idempotent teardown).
    fn remove(&self, token: ScopeId) -> bool {
        let mut inner = self.inner.borrow_mut();
        match inner.scopes.iter().position(|s| s.token == Some(token)) {
            Some(pos) => {
                inner.scopes.remove(pos);
                true
            }
            None => false,
        }
    }

    /// The innermost live run scope, if any.
    fn current_ctx(&self) -> Option<Rc<WorkflowCtx>> {
        self.inner.borrow().scopes.last().map(|s| Rc::clone(&s.ctx))
    }

    fn scope_depth(&self) -> usize {
        self.inner.borrow().scopes.len()
    }

    fn current_scope_is_owned(&self) -> bool {
        self.inner
            .borrow()
            .scopes
            .last()
            .is_some_and(|scope| scope.token.is_some())
    }

    fn cur_agent(&self) -> Option<String> {
        self.inner.borrow().cur_agent.clone()
    }

    fn set_cur_agent(&self, agent_id: Option<String>) {
        self.inner.borrow_mut().cur_agent = agent_id;
    }
}

impl Trace for WorkflowTaskState {
    fn trace(&self, sink: &mut dyn FnMut(GcEdge<'_>)) -> bool {
        let Ok(inner) = self.inner.try_borrow() else {
            return false;
        };
        for scope in &inner.scopes {
            if !scope.ctx.trace(sink) {
                return false;
            }
        }
        true
    }
}

impl TaskLocalValue for WorkflowTaskState {
    /// A spawned child clone-shares the run stack (each `Rc<WorkflowCtx>` is shared, so
    /// the child journals into the SAME run) and copies the spawner's active-step
    /// attribution, but inherited scopes reset to `token: None` (the child cannot tear
    /// down an ancestor's scope) and it mints from a fresh counter.
    fn inherit(&self) -> Rc<dyn TaskLocalValue> {
        let inner = self.inner.borrow();
        let scopes = inner
            .scopes
            .iter()
            .map(|s| WorkflowScope {
                token: None,
                ctx: Rc::clone(&s.ctx),
            })
            .collect();
        Rc::new(Self {
            inner: RefCell::new(WorkflowTaskInner {
                tokens: IdCounter::new(),
                scopes,
                cur_agent: inner.cur_agent.clone(),
            }),
        })
    }

    fn as_any(&self) -> &dyn Any {
        self
    }

    fn preflight_error(&self) -> Option<sema_core::SemaError> {
        self.current_ctx()
            .and_then(|ctx| ctx.approval_failure())
            .map(|message| sema_core::SemaError::WorkflowApprovalFailed { message })
    }
}

/// Panic-safe RAII guard for one installed workflow scope. Drop removes the EXACT token
/// it minted from the task (or host) state — an `Err` short-circuit, a panic unwinding
/// through the run body, OR a continuation dropped without resume all reinstate the
/// outer scope and never leak. Holds a `Weak` so the guard never keeps the traced state
/// alive on its own (Invariant I2): the `WorkflowCtx` `Value`s are reachable ONLY through
/// the traced [`TaskContextHandle`] / host store, never through this guard.
pub struct WorkflowGuard {
    state: Weak<WorkflowTaskState>,
    token: ScopeId,
}

impl Drop for WorkflowGuard {
    fn drop(&mut self) {
        if let Some(state) = self.state.upgrade() {
            state.remove(self.token);
        }
    }
}

/// The per-thread HOST-ADAPTER fallback scope state (used only outside a runtime quantum).
fn host_state() -> Rc<WorkflowTaskState> {
    WORKFLOW.with(Rc::clone)
}

/// Resolve the [`WorkflowTaskState`] a scope installs into: the runtime path's task
/// context (get-or-create the extension) or the host fallback when there is no task
/// context.
fn resolve_state(task_context: Option<&TaskContextHandle>) -> Rc<WorkflowTaskState> {
    if let Some(handle) = task_context {
        if let Some(state) = handle.get_rc::<WorkflowTaskState>() {
            return state;
        }
        let state = Rc::new(WorkflowTaskState::default());
        handle.borrow_mut().insert(Rc::clone(&state));
        return state;
    }
    host_state()
}

/// Low-level: install an already-built `ctx` as a scope on `task_context` (or the host
/// fallback), returning a guard whose drop removes that exact scope. Used by unit tests
/// and by [`set_workflow_scope`].
pub fn install_scope(
    task_context: Option<&TaskContextHandle>,
    ctx: Rc<WorkflowCtx>,
) -> WorkflowGuard {
    let state = resolve_state(task_context);
    let token = state.install(ctx);
    WorkflowGuard {
        state: Rc::downgrade(&state),
        token,
    }
}

/// The live workflow context for `task_context`, if a run is in progress on that task.
/// Reads the task-local extension first; the `WORKFLOW` thread-local is consulted ONLY as
/// the host-adapter fallback outside a runtime quantum.
pub fn current_for(task_context: Option<&TaskContextHandle>) -> Option<Rc<WorkflowCtx>> {
    if let Some(handle) = task_context {
        if let Some(state) = handle.get_rc::<WorkflowTaskState>() {
            if let Some(ctx) = state.current_ctx() {
                return Some(ctx);
            }
        }
    }
    if !sema_core::in_runtime_quantum() {
        return host_state().current_ctx();
    }
    None
}

/// Approval gates are valid only on the owning root workflow task. Spawned children
/// inherit a read-only scope (`token: None`), and nested `workflow/run` calls have depth
/// greater than one; neither can safely suspend the outer workflow at a sequential gate.
pub fn approval_scope_is_root_owner(task_context: Option<&TaskContextHandle>) -> bool {
    let state = if let Some(handle) = task_context {
        handle.get_rc::<WorkflowTaskState>()
    } else if !sema_core::in_runtime_quantum() {
        Some(host_state())
    } else {
        None
    };
    state.is_some_and(|state| state.scope_depth() == 1 && state.current_scope_is_owned())
}

pub fn scope_depth_for(task_context: Option<&TaskContextHandle>) -> usize {
    if let Some(handle) = task_context {
        return handle
            .get_rc::<WorkflowTaskState>()
            .map_or(0, |state| state.scope_depth());
    }
    if !sema_core::in_runtime_quantum() {
        return host_state().scope_depth();
    }
    0
}

/// The `agent_id` of the step currently executing on `task_context` (TASK-PRIVATE
/// attribution), for `workflow/tool-call`.
pub fn cur_agent_for(task_context: Option<&TaskContextHandle>) -> Option<String> {
    if let Some(handle) = task_context {
        if let Some(state) = handle.get_rc::<WorkflowTaskState>() {
            return state.cur_agent();
        }
    }
    if !sema_core::in_runtime_quantum() {
        return host_state().cur_agent();
    }
    None
}

/// Set (or clear) the step currently executing on `task_context`.
pub fn set_cur_agent_for(task_context: Option<&TaskContextHandle>, agent_id: Option<String>) {
    resolve_state(task_context).set_cur_agent(agent_id);
}

/// Redact secret-bearing values out of the workflow meta map's lossy-JSON form
/// before it is written to `metadata.json`. `:mcp` declarations may carry bearer
/// tokens or API keys in `:headers` (http servers) or `:env` (stdio servers) — see
/// `docs/plans/2026-06-24-workflow-mcp-auth.md` §4 "redaction everywhere": secrets
/// must never land in the journal, `result.json`, `metadata.json`, or OTel spans.
/// Every value under `meta.mcp.<alias>.headers` and `meta.mcp.<alias>.env` is
/// replaced with the literal string `"<redacted>"`; the keys (header/env-var
/// names) are kept so the manifest still documents WHAT was configured, just not
/// its value. Everything else in `meta` — including a `meta` with no `:mcp` key at
/// all — passes through unchanged. Pure JSON shaping, no MCP semantics, which is
/// why it lives here rather than requiring a `sema-mcp` dependency (a leaf crate
/// must not gain one).
fn redact_meta_secrets(mut meta_json: serde_json::Value) -> serde_json::Value {
    let Some(mcp) = meta_json.get_mut("mcp").and_then(|v| v.as_object_mut()) else {
        return meta_json;
    };
    for spec in mcp.values_mut() {
        let Some(spec_obj) = spec.as_object_mut() else {
            continue;
        };
        for field in ["headers", "env"] {
            let Some(values) = spec_obj.get_mut(field).and_then(|v| v.as_object_mut()) else {
                continue;
            };
            for value in values.values_mut() {
                *value = serde_json::Value::String("<redacted>".to_string());
            }
        }
    }
    meta_json
}

/// High-level entry the `workflow/run` builtin calls: resolve the run id + run-dir,
/// open the journal, build the `WorkflowCtx`, write `metadata.json`, install the
/// scope, and return the guard. The journal-open error propagates so the runtime can
/// fail the run cleanly (per-event writes below are best-effort).
///
/// `meta` is the workflow's metadata map (`{:phases … :budget … :args …}`); it is
/// recorded into `metadata.json`, and `:budget` is parsed into the run's spend caps.
/// `:permissions` is enforced by the CLI before the interpreter is built.
pub fn set_workflow_scope(
    name: &str,
    doc: &str,
    meta: &Value,
    task_context: Option<&TaskContextHandle>,
) -> io::Result<WorkflowGuard> {
    let host = host_config();
    let outermost = scope_depth_for(task_context) == 0;
    let runs_root = host
        .as_ref()
        .map(|config| config.runs_root.clone())
        .unwrap_or_else(resolve_runs_root_from_env);
    let code_version = host
        .as_ref()
        .map(|config| config.code_version.clone())
        .unwrap_or_else(|| std::env::var(CODE_VERSION_ENV).unwrap_or_default());
    let approval_code_version = host
        .as_ref()
        .map(|config| config.approval_code_version.clone())
        .unwrap_or_else(|| {
            std::env::var(APPROVAL_CODE_VERSION_ENV).unwrap_or_else(|_| code_version.clone())
        });
    let approval_public_key = host
        .as_ref()
        .map(|config| config.approval_public_key.clone())
        .unwrap_or_default();
    let resuming = outermost
        && host.as_ref().map_or_else(
            || std::env::var(RESUME_ENV).map(|v| v == "1").unwrap_or(false),
            |config| config.resuming,
        );

    // An explicit run id (the `SEMA_WORKFLOW_RUN_ID` seam, or a future library caller) is
    // validated HERE as exactly one safe path component before it is ever joined into a
    // filesystem path — the library is the authoritative gate, not just the CLI.
    let configured_id = if outermost {
        host.as_ref().map_or_else(
            || std::env::var(RUN_ID_ENV).ok(),
            |config| config.explicit_run_id.clone(),
        )
    } else {
        None
    };
    let explicit_id = match configured_id {
        Some(id) if !id.is_empty() => {
            validate_explicit_run_id(&id)?;
            Some(id)
        }
        _ => None,
    };

    // Resolve the run id AND open its journal together, because the two decisions are
    // coupled: a fresh run creates its dir fresh (a pre-existing dir is an error, a
    // generated-id collision retries with a new nonce), while a resume claims a new
    // sibling `events.resume-<n>.jsonl` segment in the ALREADY-existing dir.
    let (run_id, journal) = if resuming {
        // Resume requires an existing, explicitly-named run — never a generated id.
        let id = explicit_id.ok_or_else(|| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                "workflow resume requires an explicit run id (set SEMA_WORKFLOW_RUN_ID)",
            )
        })?;
        let events = Path::new(&runs_root).join(&id).join("events.jsonl");
        if !events.exists() {
            return Err(io::Error::new(
                io::ErrorKind::NotFound,
                format!(
                    "cannot resume: no prior run journal at {}",
                    events.display()
                ),
            ));
        }
        let journal = crate::journal::next_resume_segment(&runs_root, &id)?;
        (id, journal)
    } else if let Some(id) = explicit_id {
        // Fresh run with an operator-chosen id: fail loudly if that dir already exists.
        let journal = Journal::open(&runs_root, &id).map_err(|e| annotate_fresh_open(e, &id))?;
        (id, journal)
    } else {
        // Fresh run with a generated id: retry past the (astronomically unlikely) dir
        // collision with a new nonce each time.
        open_fresh_generated(&runs_root)?
    };
    // metadata.json — self-describing run header. Best-effort; not part of the
    // byte-identical events.jsonl oracle.
    let metadata = serde_json::json!({
        "workflow": name,
        "doc": doc,
        "run_id": run_id,
        "code_version": code_version,
        "approval_code_version": approval_code_version,
        "approval_authority_public_key": approval_public_key,
        "entry_file": host.as_ref().map(|config| config.entry_file.as_str()).unwrap_or(""),
        "meta": redact_meta_secrets(sema_core::json::value_to_json_lossy(meta)),
    });
    journal.write_metadata(&metadata);
    // The `:budget` submap of meta becomes the run's enforced spend caps.
    // The CLI sets SEMA_WORKFLOW_ARGS_JSON to the verbatim `--args` string.
    let args_json = host
        .as_ref()
        .map(|config| config.args_json.clone())
        .unwrap_or_else(|| std::env::var("SEMA_WORKFLOW_ARGS_JSON").unwrap_or_default());
    let ctx = WorkflowCtx::new_with_args(run_id.clone(), journal, parse_budget(meta), args_json);
    ctx.set_workflow_name(name);
    ctx.set_code_version(code_version);
    ctx.set_approval_code_version(approval_code_version);
    ctx.set_approval_public_key(approval_public_key);
    if resuming {
        let memos: HashMap<String, Value> = crate::journal::load_memos(&runs_root, &run_id)
            .into_iter()
            .map(|(ck, json)| (ck, sema_core::json::json_to_value(&json)))
            .collect();
        ctx.enter_resume(memos);
    }
    Ok(install_scope(task_context, ctx))
}

/// Extract the `:budget` submap from a workflow `meta` map, flattening its keyword
/// keys (`:usd`, `:tokens`) to the `String` keys [`WorkflowCtx`] parses. Returns an
/// empty map when there is no (or a malformed) `:budget` — caps stay unenforced, never
/// a panic.
pub fn parse_budget(meta: &Value) -> BTreeMap<String, Value> {
    let mut out = BTreeMap::new();
    if let Some(m) = meta.as_map_rc() {
        if let Some(b) = m.get(&Value::keyword("budget")).and_then(|v| v.as_map_rc()) {
            for (k, v) in b.iter() {
                if let Some(name) = k.as_keyword() {
                    out.insert(name, v.clone());
                }
            }
        }
    }
    out
}

/// Resolve the run-directory base: the `SEMA_WORKFLOW_RUN_DIR` seam (set by the CLI
/// `--run-dir`) if present, else the project-local [`RUNS_ROOT`].
pub fn resolve_runs_root() -> String {
    host_config()
        .map(|config| config.runs_root)
        .unwrap_or_else(resolve_runs_root_from_env)
}

fn resolve_runs_root_from_env() -> String {
    std::env::var(RUN_DIR_ENV).unwrap_or_else(|_| RUNS_ROOT.to_string())
}

/// Length-prefixed md5 over a field list → short hex. Length-prefixing each field
/// (`u64` LE length then bytes) means concatenation ambiguities like `("a","bc")` vs
/// `("ab","c")` produce different digests — the separator-collision fix.
fn hash_fields(fields: &[&str]) -> String {
    let mut buf = Vec::new();
    for f in fields {
        buf.extend_from_slice(&(f.len() as u64).to_le_bytes());
        buf.extend_from_slice(f.as_bytes());
    }
    let h = format!("{:x}", md5::compute(&buf));
    h[..16].to_string()
}

fn canonical_args_fingerprint(args_json: &str) -> String {
    let normalized = if args_json.trim().is_empty() {
        String::new()
    } else {
        serde_json::from_str::<serde_json::Value>(args_json)
            .ok()
            .and_then(|json| serde_json::to_string(&json).ok())
            .unwrap_or_else(|| args_json.to_string())
    };
    hash_fields(&["args", &normalized])
}

/// Env seam: set to "1" by the CLI `--resume` path to enter resume mode.
const RESUME_ENV: &str = "SEMA_WORKFLOW_RESUME";
/// Env seam: a stable hash of the workflow source, folded into every content-key.
const CODE_VERSION_ENV: &str = "SEMA_WORKFLOW_CODE_VERSION";
/// Env seam: collision-resistant source fingerprint for durable approval binding.
const APPROVAL_CODE_VERSION_ENV: &str = "SEMA_WORKFLOW_APPROVAL_CODE_VERSION";

/// Process-wide monotonic nonce folded into every generated run id, so two runs started
/// in one process — even within the same nanosecond — never collide on a run directory.
static RUN_ID_NONCE: AtomicU64 = AtomicU64::new(0);

/// Max attempts to place a generated run into a free directory. A generated id already
/// folds in a process nonce, so a collision is astronomically unlikely; the bound only
/// stops an infinite loop if the filesystem keeps returning `AlreadyExists` for some
/// other reason.
const MAX_FRESH_ATTEMPTS: u32 = 8;

/// A freshly generated run id: `wf_<unix_secs>_<subsec_nanos>_<pid>_<nonce>`. No RNG
/// dependency, yet unique per process: the nanosecond field separates rapid runs and the
/// process nonce guarantees distinctness even at identical clock readings (two runs in
/// the same second — or nanosecond — no longer share a directory).
fn generate_run_id() -> String {
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default();
    let nonce = RUN_ID_NONCE.fetch_add(1, Ordering::Relaxed);
    format!(
        "wf_{}_{}_{}_{}",
        now.as_secs(),
        now.subsec_nanos(),
        std::process::id(),
        nonce
    )
}

/// Validate an explicit run id (the `SEMA_WORKFLOW_RUN_ID` seam or a library caller) as
/// exactly ONE safe path component: non-empty, no `/` or `\` separator, no `..` traversal,
/// not a `.`-only component, and free of NUL / control characters — it joins straight into
/// a filesystem path, so anything else is a traversal or a broken directory name. Returns
/// `InvalidInput` on rejection.
pub fn validate_explicit_run_id(id: &str) -> io::Result<()> {
    let reject = |why: &str| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            format!("workflow run id {id:?} is not a safe directory name: {why}"),
        )
    };
    if id.is_empty() {
        return Err(reject("must not be empty"));
    }
    if id.contains('/') || id.contains('\\') {
        return Err(reject("must not contain a path separator"));
    }
    if id.contains("..") {
        return Err(reject("must not contain '..'"));
    }
    if id.bytes().all(|b| b == b'.') {
        return Err(reject("must not be only '.' characters"));
    }
    if id.chars().any(|c| c == '\0' || c.is_control()) {
        return Err(reject("must not contain NUL or control characters"));
    }
    Ok(())
}

/// Resolve the run id for a NEW run: the validated `SEMA_WORKFLOW_RUN_ID` seam if set and
/// non-empty, else a freshly generated id (see [`generate_run_id`]). An invalid explicit
/// id is an error rather than a silently sanitized path. Overridable to a fixed value in
/// tests (the golden oracle sets `SEMA_WORKFLOW_RUN_ID=wf_test_0001`).
pub fn resolve_run_id() -> io::Result<String> {
    match std::env::var(RUN_ID_ENV) {
        Ok(id) if !id.is_empty() => {
            validate_explicit_run_id(&id)?;
            Ok(id)
        }
        _ => Ok(generate_run_id()),
    }
}

/// Open a fresh run under a generated id, retrying with a new id (fresh nonce) on the
/// unlikely directory collision. Returns the winning id alongside its opened journal.
fn open_fresh_generated(runs_root: &str) -> io::Result<(String, Journal)> {
    open_fresh_with(runs_root, generate_run_id)
}

/// The collision-retry core, with an injectable id source so the retry path is unit
/// testable without racing the clock. Bounded by [`MAX_FRESH_ATTEMPTS`].
fn open_fresh_with(
    runs_root: &str,
    mut next_id: impl FnMut() -> String,
) -> io::Result<(String, Journal)> {
    let mut last_err = None;
    for _ in 0..MAX_FRESH_ATTEMPTS {
        let id = next_id();
        match Journal::open(runs_root, &id) {
            Ok(journal) => return Ok((id, journal)),
            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => last_err = Some(e),
            Err(e) => return Err(e),
        }
    }
    Err(last_err.unwrap_or_else(|| {
        io::Error::new(
            io::ErrorKind::AlreadyExists,
            "could not allocate a unique workflow run directory",
        )
    }))
}

/// Turn the bare `AlreadyExists` from a fresh journal claim into an actionable message
/// (keeping the error KIND so callers can still match on it), for an operator-chosen id
/// whose journal already exists.
fn annotate_fresh_open(err: io::Error, run_id: &str) -> io::Error {
    if err.kind() == io::ErrorKind::AlreadyExists {
        io::Error::new(
            io::ErrorKind::AlreadyExists,
            format!(
                "a workflow run journal for {run_id:?} already exists; \
                 choose a fresh run id or resume it with --resume"
            ),
        )
    } else {
        err
    }
}

/// Format `SystemTime::now()` as an RFC3339 / ISO-8601 UTC string (`YYYY-MM-DDTHH:MM:SSZ`)
/// without pulling in `chrono`. Civil-date conversion via the standard
/// days-since-epoch algorithm (Howard Hinnant's `civil_from_days`).
pub(crate) fn rfc3339_now() -> String {
    let dur = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default();
    let secs = dur.as_secs();
    let days = (secs / 86_400) as i64;
    let rem = secs % 86_400;
    let (hour, min, sec) = (rem / 3600, (rem % 3600) / 60, rem % 60);
    let (y, m, d) = civil_from_days(days);
    format!("{y:04}-{m:02}-{d:02}T{hour:02}:{min:02}:{sec:02}Z")
}

/// Convert a count of days since 1970-01-01 to a (year, month, day) civil date.
/// Hinnant's algorithm; valid for the full SystemTime range we will ever journal.
fn civil_from_days(z: i64) -> (i64, u32, u32) {
    let z = z + 719_468;
    let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
    let doe = (z - era * 146_097) as u64; // [0, 146096]
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365; // [0, 399]
    let y = yoe as i64 + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); // [0, 365]
    let mp = (5 * doy + 2) / 153; // [0, 11]
    let d = (doy - (153 * mp + 2) / 5 + 1) as u32; // [1, 31]
    let m = if mp < 10 { mp + 3 } else { mp - 9 } as u32; // [1, 12]
    (if m <= 2 { y + 1 } else { y }, m, d)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn seq_is_monotonic_from_zero() {
        let ctx = WorkflowCtx::new("wf_t".into(), Journal::null(), BTreeMap::new());
        assert_eq!(ctx.next_seq(), 0);
        assert_eq!(ctx.next_seq(), 1);
        assert_eq!(ctx.next_seq(), 2);
    }

    #[test]
    fn fixed_ts_freezes_ts_and_dur() {
        std::env::set_var(FIXED_TS_ENV, "1970-01-01T00:00:00Z");
        let ctx = WorkflowCtx::new("wf_t".into(), Journal::null(), BTreeMap::new());
        assert_eq!(ctx.ts(), "1970-01-01T00:00:00Z");
        assert_eq!(ctx.dur_ms(), 0);
        std::env::remove_var(FIXED_TS_ENV);
    }

    fn ctx_with_budget(pairs: &[(&str, Value)]) -> Rc<WorkflowCtx> {
        let mut b = BTreeMap::new();
        for (k, v) in pairs {
            b.insert(k.to_string(), v.clone());
        }
        WorkflowCtx::new_with_args("wf_t".into(), Journal::null(), b, String::new())
    }

    #[test]
    fn charge_trips_usd_cap_and_latches() {
        let ctx = ctx_with_budget(&[("usd", Value::float(0.01))]);
        assert!(!ctx.charge(Some(0.005), 10), "under cap must not trip");
        assert!(!ctx.over_budget());
        assert!(ctx.charge(Some(0.02), 100), "crossing cap trips");
        assert!(ctx.over_budget(), "latch is sticky");
        // Once latched, it stays latched even on a tiny later charge.
        let _ = ctx.charge(Some(0.0), 0);
        assert!(ctx.over_budget());
    }

    #[test]
    fn charge_enforces_tokens_when_cost_unknown() {
        let ctx = ctx_with_budget(&[("tokens", Value::int(50))]);
        assert!(!ctx.charge(None, 40), "cost None still counts tokens");
        assert!(!ctx.over_budget());
        assert!(ctx.charge(None, 20), "60 > 50 trips on tokens alone");
        assert!(ctx.over_budget());
        assert_eq!(ctx.budget_limit_for_event(), Some(50));
    }

    #[test]
    fn no_budget_never_trips() {
        let ctx = WorkflowCtx::new("wf_t".into(), Journal::null(), BTreeMap::new());
        assert!(!ctx.has_budget());
        assert!(!ctx.charge(Some(9999.0), 9_999_999));
        assert!(!ctx.over_budget());
        assert_eq!(ctx.budget_limit_for_event(), None);
    }

    #[test]
    fn parse_budget_extracts_caps_and_tolerates_absence() {
        let mut bm = BTreeMap::new();
        bm.insert(Value::keyword("usd"), Value::float(2.5));
        bm.insert(Value::keyword("tokens"), Value::int(1000));
        let mut meta = BTreeMap::new();
        meta.insert(Value::keyword("budget"), Value::map(bm));
        let parsed = parse_budget(&Value::map(meta));
        assert_eq!(parsed.get("usd").and_then(|v| v.as_float()), Some(2.5));
        assert_eq!(parsed.get("tokens").and_then(|v| v.as_int()), Some(1000));
        // No :budget at all → empty (unenforced).
        assert!(parse_budget(&Value::map(BTreeMap::new())).is_empty());
    }

    #[test]
    fn content_keys_are_stable_distinct_and_length_prefixed() {
        let ctx = WorkflowCtx::new("wf_t".into(), Journal::null(), BTreeMap::new());
        ctx.set_code_version("v1".into());
        // First occurrence of each distinct input is stable; differing inputs differ.
        let k_a = ctx.agent_content_key("audit a.php", "[:list :string]", "auditor", "Audit", "");
        let k_b = ctx.agent_content_key("audit b.php", "[:list :string]", "auditor", "Audit", "");
        assert_ne!(k_a, k_b, "different prompts ⇒ different keys");
        // Length-prefixing: ('a','bc') must not collide with ('ab','c').
        let k1 = ctx.agent_content_key("a", "bc", "n", "p", "");
        let k2 = ctx.agent_content_key("ab", "c", "n", "p", "");
        assert_ne!(
            k1, k2,
            "length-prefixed fields can't collide via concatenation"
        );
        // Occurrence ordinal: a repeated identical leaf gets a distinct key.
        let r1 = ctx.checkpoint_content_key("files", "Inventory");
        let r2 = ctx.checkpoint_content_key("files", "Inventory");
        assert_ne!(
            r1, r2,
            "repeated identical checkpoint ⇒ distinct occurrence key"
        );
    }

    #[test]
    fn code_version_changes_invalidate_keys() {
        let ctx1 = WorkflowCtx::new("a".into(), Journal::null(), BTreeMap::new());
        ctx1.set_code_version("v1".into());
        let ctx2 = WorkflowCtx::new("b".into(), Journal::null(), BTreeMap::new());
        ctx2.set_code_version("v2".into());
        assert_ne!(
            ctx1.agent_content_key("p", "s", "n", "ph", ""),
            ctx2.agent_content_key("p", "s", "n", "ph", ""),
            "a changed code-version produces different content-keys (auto-invalidation)"
        );
    }

    #[test]
    fn args_changes_invalidate_keys() {
        let ctx1 = WorkflowCtx::new_with_args(
            "a".into(),
            Journal::null(),
            BTreeMap::new(),
            r#"{"batch":1}"#.into(),
        );
        ctx1.set_code_version("v1".into());
        let ctx2 = WorkflowCtx::new_with_args(
            "b".into(),
            Journal::null(),
            BTreeMap::new(),
            r#"{"batch":2}"#.into(),
        );
        ctx2.set_code_version("v1".into());
        assert_ne!(
            ctx1.checkpoint_content_key("files", "ph"),
            ctx2.checkpoint_content_key("files", "ph"),
            "changed workflow args produce different content-keys"
        );
    }

    #[test]
    fn memo_store_round_trip_guard_skips_unsurvivable_values() {
        let ctx = WorkflowCtx::new("wf_t".into(), Journal::null(), BTreeMap::new());
        // A plain string round-trips → memoized and looked up.
        ctx.memo_store("ck_text", &Value::string("hello"));
        assert_eq!(ctx.memo_lookup("ck_text"), Some(Value::string("hello")));
        // A keyword-keyed map round-trips (json_to_value rebuilds keyword keys).
        let mut m = BTreeMap::new();
        m.insert(Value::keyword("body"), Value::string("x"));
        let kw_map = Value::map(m);
        ctx.memo_store("ck_map", &kw_map);
        assert_eq!(ctx.memo_lookup("ck_map"), Some(kw_map));
        // A map with a NON-string/keyword key does NOT survive JSON round-trip (the int
        // key becomes a string), so the guard leaves it un-memoized → it re-runs on
        // resume rather than resuming a different value. This exercises the FALSE branch.
        let mut bad = BTreeMap::new();
        bad.insert(Value::int(1), Value::int(2));
        ctx.memo_store("ck_bad", &Value::map(bad));
        assert_eq!(
            ctx.memo_lookup("ck_bad"),
            None,
            "a non-round-trippable value must be left un-memoized"
        );
    }

    #[test]
    fn checkpoint_round_trips() {
        let ctx = WorkflowCtx::new("wf_t".into(), Journal::null(), BTreeMap::new());
        assert_eq!(ctx.read_checkpoint("files"), None);
        ctx.store_checkpoint("files", Value::int(3));
        assert_eq!(ctx.read_checkpoint("files"), Some(Value::int(3)));
    }

    // ── MCP handle registry ──────────────────────────────────────────────

    #[test]
    fn mcp_handle_registry_starts_empty_and_undeclared() {
        let ctx = WorkflowCtx::new("wf_t".into(), Journal::null(), BTreeMap::new());
        assert_eq!(ctx.mcp_handle("asana"), None);
        assert!(!ctx.is_mcp_declared("asana"));
    }

    #[test]
    fn mcp_declared_tracks_aliases_before_handles_resolve() {
        let ctx = WorkflowCtx::new("wf_t".into(), Journal::null(), BTreeMap::new());
        ctx.set_mcp_declared(vec!["asana".to_string(), "fs".to_string()]);
        // Declared, but resolution hasn't populated a handle yet.
        assert!(ctx.is_mcp_declared("asana"));
        assert_eq!(ctx.mcp_handle("asana"), None);
        assert!(!ctx.is_mcp_declared("zebra"));
    }

    #[test]
    fn mcp_handle_returns_resolved_handle_by_alias() {
        let ctx = WorkflowCtx::new("wf_t".into(), Journal::null(), BTreeMap::new());
        ctx.set_mcp_declared(vec!["asana".to_string(), "fs".to_string()]);
        let mut handles = BTreeMap::new();
        handles.insert("asana".to_string(), Value::string("mcp-1"));
        handles.insert("fs".to_string(), Value::string("mcp-2"));
        ctx.set_mcp_handles(handles);
        assert_eq!(ctx.mcp_handle("asana"), Some(Value::string("mcp-1")));
        assert_eq!(ctx.mcp_handle("fs"), Some(Value::string("mcp-2")));
        assert_eq!(ctx.mcp_handle("nope"), None);
    }

    #[test]
    fn workflow_ctx_traces_state_memo_and_mcp_values() {
        // Invariant I2: every live `Value` a run holds must be reachable to the
        // collector — one edge per state-bag / memo / MCP-handle value.
        let ctx = WorkflowCtx::new("t".into(), Journal::null(), BTreeMap::new());
        ctx.store_checkpoint("k", Value::int(1));
        let mut memos = HashMap::new();
        memos.insert("ck".to_string(), Value::int(2));
        ctx.enter_resume(memos);
        let mut handles = BTreeMap::new();
        handles.insert("asana".to_string(), Value::string("handle"));
        ctx.set_mcp_handles(handles);

        let mut edges = 0;
        assert!(ctx.trace(&mut |edge| {
            assert!(matches!(edge, GcEdge::Value(_)));
            edges += 1;
        }));
        assert_eq!(
            edges, 3,
            "state bag + resume memo + MCP handle each trace once"
        );
    }

    #[test]
    fn host_scope_restores_previous_on_drop() {
        // The host fallback (no task context) behaves like a scope stack: a nested run
        // reveals the outer scope again once its guard drops.
        assert!(current_for(None).is_none());
        let outer = WorkflowCtx::new("outer".into(), Journal::null(), BTreeMap::new());
        let g_outer = install_scope(None, outer);
        assert_eq!(
            current_for(None).map(|c| c.run_id.clone()).as_deref(),
            Some("outer")
        );
        {
            let inner = WorkflowCtx::new("inner".into(), Journal::null(), BTreeMap::new());
            let _g_inner = install_scope(None, inner);
            assert_eq!(
                current_for(None).map(|c| c.run_id.clone()).as_deref(),
                Some("inner")
            );
        }
        // inner guard dropped → outer reinstated
        assert_eq!(
            current_for(None).map(|c| c.run_id.clone()).as_deref(),
            Some("outer")
        );
        drop(g_outer);
        assert!(current_for(None).is_none());
    }

    #[test]
    fn task_state_removes_the_exact_token_out_of_lifo() {
        // Two scopes installed, torn down OLDEST-first (out of LIFO order): exact-token
        // removal restores the surviving inner scope, not whatever happens to be on top.
        let state = WorkflowTaskState::default();
        let outer = WorkflowCtx::new("outer".into(), Journal::null(), BTreeMap::new());
        let inner = WorkflowCtx::new("inner".into(), Journal::null(), BTreeMap::new());
        let outer_token = state.install(outer);
        let inner_token = state.install(inner);
        assert_eq!(
            state.current_ctx().map(|c| c.run_id.clone()).as_deref(),
            Some("inner")
        );

        assert!(state.remove(outer_token));
        assert!(
            !state.remove(outer_token),
            "removing the same token twice is idempotent"
        );
        assert_eq!(
            state.current_ctx().map(|c| c.run_id.clone()).as_deref(),
            Some("inner"),
            "removing the outer token leaves the inner scope live and on top"
        );
        assert!(state.remove(inner_token));
        assert!(state.current_ctx().is_none());
    }

    #[test]
    fn child_inherits_run_and_agent_but_not_removal_authority() {
        // A spawned child clone-shares the run scope + copies the active agent, but its
        // inherited scope is not removable by the child (token stripped to None).
        let state = Rc::new(WorkflowTaskState::default());
        let run = WorkflowCtx::new("shared-run".into(), Journal::null(), BTreeMap::new());
        let parent_token = state.install(run);
        state.set_cur_agent(Some("scout_1".to_string()));

        let child = state.inherit();
        let child = child
            .as_any()
            .downcast_ref::<WorkflowTaskState>()
            .expect("inherited workflow state");
        assert_eq!(
            child.current_ctx().map(|c| c.run_id.clone()).as_deref(),
            Some("shared-run"),
            "child observes the spawner's active run"
        );
        assert_eq!(child.cur_agent().as_deref(), Some("scout_1"));
        // The child cannot tear down the parent's scope with the parent's token.
        assert!(!child.remove(parent_token));
        assert_eq!(
            child.current_ctx().map(|c| c.run_id.clone()).as_deref(),
            Some("shared-run")
        );
        // The parent's own teardown still works.
        assert!(state.remove(parent_token));
        assert!(state.current_ctx().is_none());
    }

    // ── run identity (A2) ────────────────────────────────────────────────

    #[test]
    fn generated_run_id_has_secs_nanos_pid_and_nonce() {
        let a = generate_run_id();
        let b = generate_run_id();
        assert_ne!(a, b, "the process nonce makes back-to-back ids distinct");
        for id in [&a, &b] {
            assert!(id.starts_with("wf_"), "id keeps the wf_ prefix: {id}");
            let parts: Vec<&str> = id.split('_').collect();
            assert_eq!(parts.len(), 5, "wf_<secs>_<nanos>_<pid>_<nonce>: {id}");
            for field in &parts[1..] {
                assert!(
                    !field.is_empty() && field.bytes().all(|c| c.is_ascii_digit()),
                    "each generated id field is a non-empty number: {id}"
                );
            }
        }
    }

    #[test]
    fn validate_explicit_run_id_accepts_safe_names_and_rejects_unsafe() {
        for ok in ["wf_test_0001", "run-42", "abc.def", "a"] {
            assert!(validate_explicit_run_id(ok).is_ok(), "should accept {ok:?}");
        }
        for bad in [
            "",     // empty
            "a/b",  // unix separator
            "a\\b", // windows separator
            "..",   // traversal
            "a..b", // embedded traversal
            ".",    // dot-only
            "...",  // dots-only
            "a\0b", // NUL
            "a\nb", // control char
        ] {
            let err = validate_explicit_run_id(bad).expect_err(&format!("should reject {bad:?}"));
            assert_eq!(err.kind(), io::ErrorKind::InvalidInput, "for {bad:?}");
        }
    }

    #[test]
    fn open_fresh_with_retries_past_a_colliding_id() {
        let mut root = std::env::temp_dir();
        root.push(format!(
            "sema-wf-fresh-retry-{}-{}",
            std::process::id(),
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let root_str = root.to_string_lossy().to_string();
        // The first two candidate ids already have a JOURNAL (events.jsonl); the opener
        // must skip them (its create_new claim fails) and land on the first free id.
        for taken in ["taken_1", "taken_2"] {
            std::fs::create_dir_all(root.join(taken)).unwrap();
            std::fs::write(root.join(taken).join("events.jsonl"), "{}\n").unwrap();
        }
        let mut candidates = ["taken_1", "taken_2", "free_3"].into_iter();
        let (id, _journal) =
            open_fresh_with(&root_str, || candidates.next().unwrap().to_string()).unwrap();
        assert_eq!(id, "free_3", "opener retried past the colliding ids");
        std::fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn civil_date_epoch() {
        assert_eq!(civil_from_days(0), (1970, 1, 1));
        // 2026-06-24 is 20628 days after epoch.
        assert_eq!(civil_from_days(20_628), (2026, 6, 24));
    }

    // ── redact_meta_secrets ──────────────────────────────────────────────

    #[test]
    fn redacts_mcp_headers_and_env_values() {
        let meta = serde_json::json!({
            "budget": {"usd": 1.0},
            "mcp": {
                "asana": {
                    "url": "https://mcp.asana.com/mcp",
                    "headers": {"Authorization": "Bearer secret-token"},
                    "persist": "workflow"
                },
                "fs": {
                    "command": "npx",
                    "env": {"API_TOKEN": "supersecret", "PLAIN": "not-a-secret-name"}
                }
            }
        });
        let redacted = redact_meta_secrets(meta);
        assert_eq!(
            redacted["mcp"]["asana"]["headers"]["Authorization"],
            "<redacted>"
        );
        assert_eq!(redacted["mcp"]["fs"]["env"]["API_TOKEN"], "<redacted>");
        assert_eq!(redacted["mcp"]["fs"]["env"]["PLAIN"], "<redacted>");
    }

    #[test]
    fn redaction_keeps_header_and_env_keys_and_sibling_fields() {
        let meta = serde_json::json!({
            "mcp": {
                "asana": {
                    "url": "https://mcp.asana.com/mcp",
                    "headers": {"Authorization": "Bearer secret-token", "X-Trace": "abc"},
                    "tools": ["create_task"],
                    "persist": "workflow"
                }
            }
        });
        let redacted = redact_meta_secrets(meta);
        // Keys survive.
        assert!(redacted["mcp"]["asana"]["headers"]
            .as_object()
            .unwrap()
            .contains_key("Authorization"));
        assert!(redacted["mcp"]["asana"]["headers"]
            .as_object()
            .unwrap()
            .contains_key("X-Trace"));
        // Sibling fields untouched.
        assert_eq!(redacted["mcp"]["asana"]["url"], "https://mcp.asana.com/mcp");
        assert_eq!(redacted["mcp"]["asana"]["tools"][0], "create_task");
        assert_eq!(redacted["mcp"]["asana"]["persist"], "workflow");
    }

    #[test]
    fn meta_without_mcp_passes_through_unchanged() {
        let meta = serde_json::json!({
            "budget": {"usd": 1.0},
            "args": {"repo": "sema-lisp/sema"},
            "phases": ["Triage"],
        });
        let redacted = redact_meta_secrets(meta.clone());
        assert_eq!(redacted, meta);
    }

    #[test]
    fn mcp_alias_without_headers_or_env_passes_through_unchanged() {
        let meta = serde_json::json!({
            "mcp": {"asana": {"url": "https://mcp.asana.com/mcp", "persist": "workflow"}}
        });
        let redacted = redact_meta_secrets(meta.clone());
        assert_eq!(redacted, meta);
    }
}