capsule_runtime/

types.rs

use std::{
    collections::{HashMap, HashSet},
    path::PathBuf,
    sync::Arc,
};

use murmur_artifact::{
    ArtifactImplementation, ArtifactRuntime, ContextConfig, HookConfig, InferenceConfig,
    LifecycleConfig, LifecycleOverride, Registry,
};
use wasmtime::{component::Component, Engine};

use crate::{
    bindings::host::murmur::tool::run::ToolResult,
    formation::FormationId,
    formation_credentials::FormationMember,
    hooks::ShellDispatchInfo,
    limits::{CallDeadlines, EpochTicker, ExecutionLimits},
    network_policy::{HookCapabilityGrant, ToolCapabilityGrant},
    spawn_credential::{SpawnApproval, SpawnCredential},
};

pub(crate) struct DispatchOutcome {
    pub result: ToolResult,
    pub shell: Option<ShellDispatchInfo>,
    /// Set instead of [`Self::shell`] when the command outran its grace period and was demoted
    /// to the background. The two are never both `Some`: a demoted command has no exit code yet,
    /// which is exactly why `HookEvent::Shell` does not fire for one.
    pub detached: Option<crate::detached::DetachedDispatchInfo>,
    /// True when the dispatch was served by reading a skill.md file rather than running
    /// a binary or WASM component. Used by the agent loop to write a `skill_call` trace
    /// event instead of a `tool_call` event.
    pub is_skill: bool,
    /// The fence source `result.data` was wrapped under, or `None` when the result carries no
    /// fence. Set by [`crate::runtime::CapsuleStoreState::dispatch_agent_tool_async`] on the
    /// same branch that applies the fence, so the label and the wrapping cannot drift: a caller
    /// reading this never re-derives a source name by parsing the content back.
    ///
    /// `Some("tool:<name>")` on every non-skill outcome, `Some("skill:<name>")` on a skill whose
    /// `murmur.lock` origin is `runtime`, and `None` on an operator-declared skill and on every
    /// outcome that leaves the private unfenced dispatch — that is,
    /// [`crate::runtime::CapsuleStoreState::dispatch_submit_plan`]'s route, whose step output is
    /// fenced later as one field of a report.
    pub fence_source: Option<String>,
    /// Set when this dispatch failed in a way that ends the *session* rather than just this
    /// tool call — today only a `sealed` composed-root construction failure
    /// ([`crate::shell::ShellExecError::session_fatal`]). `result` still describes the failed
    /// call so the trace records what was attempted; the agent turn loop then returns this
    /// error instead of feeding the failure back to the model for another turn.
    ///
    /// A dispatch that merely failed leaves this `None`: a capsule reacting to a broken tool
    /// call is ordinary, whereas a capsule continuing to run after its declared containment
    /// boundary stopped being establishable is the silent-degradation failure mode the whole
    /// containment-class mechanism exists to prevent.
    pub fatal: Option<crate::errors::RuntimeError>,
    /// The runtime's own words about this call, appended by
    /// [`crate::runtime::CapsuleStoreState::dispatch_agent_tool_async`] after the fence's closing
    /// marker, so the model reads them as the runtime's rather than as the tool's data.
    ///
    /// `Some` on these outcomes:
    ///
    /// * a `call-member` call that started, carrying [`crate::member_call::started_note`];
    /// * an accepted `end-without-answer`;
    /// * a pull through `pull-artifact`, and a `remove-artifact` that removed something, each
    ///   saying when the tool list changes;
    /// * a shell call that finished and whose failure was attributed to a
    ///   `capabilities.resources` limit, carrying [`crate::resources::resource_limit_line`].
    ///
    /// `None` on every other outcome. The unfenced route of
    /// [`crate::runtime::CapsuleStoreState::dispatch_submit_plan`] never reads it.
    ///
    /// The session's one deadline notice is not a note: it is appended after the note, by
    /// [`crate::runtime::CapsuleStoreState::deliver_deadline_notice`], on whichever dispatched
    /// outcome is first handed back inside the notice margin.
    pub runtime_note: Option<String>,
    /// The runtime's own text, unfenced, when it could not run the call: the
    /// [`crate::shell::ShellExecError`] a native tool or shell dispatch returned, the same text
    /// `result.data` carries before the fence wraps it. The agent loop records it as the
    /// `tool_call` trace record's `error`.
    ///
    /// `None` whenever the tool ran, including when the tool itself reported failure or exited
    /// non-zero.
    pub runtime_error: Option<String>,
}

impl DispatchOutcome {
    pub(crate) fn tool(result: ToolResult) -> Self {
        Self {
            result,
            shell: None,
            detached: None,
            is_skill: false,
            fence_source: None,
            fatal: None,
            runtime_note: None,
            runtime_error: None,
        }
    }

    pub(crate) fn skill(result: ToolResult) -> Self {
        Self {
            result,
            shell: None,
            detached: None,
            is_skill: true,
            fence_source: None,
            fatal: None,
            runtime_note: None,
            runtime_error: None,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ArtifactRequest {
    pub name: String,
    pub version: String,
    pub runtime: ArtifactRuntime,
    /// Optional local source path (skill artifacts only). When set, the runtime resolves the
    /// skill's `skill.md` directly from this path and skips registry resolution entirely.
    /// Relative paths resolve against `StageRequest::manifest_dir`.
    pub source: Option<String>,
    /// Per-artifact capability grant copied verbatim from this artifact's entry in the
    /// capsule operator's own manifest (`murmur_artifact::RuntimeArtifact::capabilities`).
    /// Consumed for `runtime: hook` (default-deny baseline, `None` = deny everything) and for
    /// `runtime: tool`/`runtime: driver` (ceiling baseline, `None` = the capsule-wide policy
    /// unchanged); the manifest parser rejects the key on `runtime: skill`. Never sourced
    /// from the artifact's own bundled `murmur.yaml`, so an artifact cannot self-grant.
    pub capabilities: Option<murmur_artifact::Capabilities>,
    /// What the runtime does when this hook's async job queue is full, copied verbatim from
    /// `on_overflow:` on this artifact's entry in the capsule operator's own manifest — the
    /// same operator-only sourcing rule `capabilities` above follows. Consumed for
    /// `runtime: hook` and inert for every other role (the manifest parser rejects the key
    /// there); inert too for a hook that turns out to be `execution_mode: blocking`, which is
    /// never queued.
    pub on_overflow: murmur_artifact::HookOverflowPolicy,
    /// This artifact's operator-declared `config:` block, copied verbatim from its entry in the
    /// capsule operator's own manifest (`murmur_artifact::RuntimeArtifact::config`) — the same
    /// operator-only sourcing rule `capabilities` above follows, so an artifact pulled from a
    /// registry cannot configure itself.
    ///
    /// Lowered to compact JSON at staging by [`crate::artifact_config::lower_artifact_config`] and
    /// delivered to this artifact alone as [`crate::artifact_config::ARTIFACT_CONFIG_ENV`]. `None`
    /// means the variable is absent from the artifact's environment. Consumed for `runtime: hook`,
    /// `runtime: tool` (WASM and native) and `runtime: driver`; the manifest parser rejects the
    /// key on `runtime: skill`.
    pub config: Option<serde_norway::Value>,
    /// This artifact's operator-declared `gateway:` block, copied verbatim from its entry in the
    /// capsule operator's own manifest (`murmur_artifact::RuntimeArtifact::gateway`) — the same
    /// operator-only sourcing rule `capabilities` above follows. Built into the artifact's
    /// credential gateway at staging, which is also where the artifact's own `upstream_auth:`
    /// declaration is read; an artifact without this block is never asked for one.
    pub gateway: Option<murmur_artifact::ArtifactGateway>,
}

/// What `murmur.lock` pins one artifact at, already narrowed to this host.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LockExpectation {
    pub name: String,
    pub resolved_version: String,
    /// The hash the resolved payload must have. Selected by the caller from the lock entry's
    /// `sha256` for the platform staging runs on, so it is whichever of `any` or
    /// `platforms.<platform>` applies to this host.
    pub sha256: String,
    /// Who wrote the lock entry. A [`murmur_artifact::LockOrigin::Runtime`] pin stages only as a
    /// tool or skill without a `gateway:` block; staging refuses anything else with
    /// [`crate::RuntimeError::RuntimeOriginNotDeclarable`].
    pub origin: murmur_artifact::LockOrigin,
}

/// What staging actually resolved, for the caller to write back into `murmur.lock`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResolvedLockArtifact {
    pub name: String,
    pub resolved_version: String,
    /// Hash of the payload that was resolved on this host.
    pub sha256: String,
    /// Which platform the resolved payload is filed under, `None` for a platform-independent
    /// one. Decides whether the lock entry pins `sha256.any` or `sha256.platforms.<platform>`.
    pub platform: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct CapabilityPolicy {
    pub network_allow: Vec<String>,
    /// From `capabilities.peer_fetch.allow`: the peers this capsule may redeem a peer-file handle
    /// against. Empty by default, which is deny — and which is also why no `fetch-peer-file` tool
    /// manifest is written and the capsule's model never sees the tool exists.
    ///
    /// A separate list from [`Self::network_allow`], enforced with the same rule matcher but
    /// never merged into it. Ingesting a peer's bytes lands a file in this capsule's own workdir,
    /// so it is a prompt-injection surface and deserves its own operator control; a destination
    /// in `network.allow` is not redeemable unless it also appears here, and declaring it here
    /// does not widen `network.allow`.
    pub peer_fetch_allow: Vec<String>,
    /// From `capabilities.network.unix_sockets`. `false` by default, which makes the shell
    /// subprocess tree's `socket(AF_UNIX, ...)` calls fail with `EACCES` at the kernel level on
    /// both Linux tiers — `network_allow` above governs IP destinations only and would otherwise
    /// leave a local daemon socket (`/var/run/docker.sock` — host root) completely unmediated.
    /// `true` simply omits that seccomp rule; it is capsule-wide, not per-socket-path. See
    /// `sandbox::denied_socket_domains`.
    pub unix_sockets_allowed: bool,
    pub filesystem_scope: Option<String>,
    /// From `capabilities.filesystem.workdir_exec`. `false` by default, which withholds the
    /// Landlock `Execute` right from the workdir's own `PathBeneath` rule — see
    /// `sandbox::linux_enforce::workdir_access_rights`. With it withheld the kernel refuses to exec
    /// anything under the workdir on the resolved path itself, so `shell_allow` is enforced
    /// completely: a binary copied or compiled into the workdir cannot run under *any* name.
    ///
    /// `true` gives the right back for compile-and-run workflows and accepts, explicitly, that
    /// `shell_allow` is then unenforceable for anything inside the workdir. That is why this field
    /// is the one *grant* in this struct that also lowers the session's achieved containment class
    /// (see `crate::containment::achieved_containment_class`) — nothing else here can.
    pub workdir_exec_allowed: bool,
    /// From `capabilities.filesystem.read_only`: workdir subtrees the capsule may read but must
    /// not write, in the same vocabulary [`Self::filesystem_scope`] uses. Empty by default, which
    /// is the whole workdir writable.
    ///
    /// Carried here as the declared strings and lowered once at staging into
    /// `crate::protected_paths::ProtectedPaths`, which is what the dispatch check consults — so a
    /// malformed entry refuses the launch rather than one call at a time. Unlike every other field
    /// here this one grants nothing: it can only subtract from what the workdir preopen already
    /// allows.
    pub read_only_paths: Vec<String>,
    pub shell_allow: Vec<String>,
    pub spawn_allow: Vec<String>,
    /// From `capabilities.spawn.workdir_snapshot`: the full names in [`Self::spawn_allow`] whose
    /// every delegation starts in a copy of this session's work folder, made by this runtime
    /// before the child process starts. Empty by default, which starts every child in an empty
    /// folder.
    pub spawn_workdir_snapshot: Vec<String>,
    pub shell_strip_env: Vec<String>,
    pub shell_baseline_env: Vec<String>,
    /// Typed interpreter-runtime grants from `capabilities.shell.interpreter_runtime`: each
    /// names an already-allowlisted binary and the exact host directories outside the workdir
    /// its import machinery needs, with a per-directory `list_dir` enumerability flag. Empty
    /// unless the manifest declares them. Consumed on `KernelFull` to add narrow Landlock
    /// grants (see `sandbox::resolve_interpreter_runtime_grants`) and to fire `W-SEC-009`.
    pub shell_interpreter_runtime: Vec<murmur_artifact::InterpreterRuntimeGrant>,
    /// Typed staged-runtime grants from `capabilities.shell.staged_runtime`: each names an
    /// already-allowlisted binary, the absolute host path of a pinned runtime tree, and the `pin`
    /// identifying that tree's build. Empty unless the manifest declares them.
    ///
    /// Consumed at staging to enforce that declaring one requires an effective `sealed` floor (see
    /// `crate::staged_runtime::check_staged_runtime_floor`) and to render the `staged runtime`
    /// section of `--explain-scope`. Per-binary mutually exclusive with
    /// [`Self::shell_interpreter_runtime`], which the manifest parser already guarantees.
    pub shell_staged_runtime: Vec<murmur_artifact::StagedRuntimeGrant>,
    /// The runtime directories of each `shell_allow` binary the runtime recognises as an
    /// interpreter, as that interpreter reported them (see [`crate::interpreter_runtime`]). Not a
    /// manifest field: empty from [`capability_policy_from_runtime_manifest`], filled once by
    /// `stage_session` and by `mur run --explain-scope` through
    /// [`crate::interpreter_runtime::derive_interpreter_runtimes`]. Each becomes a readable,
    /// non-executable Landlock grant (`sandbox::resolve_derived_interpreter_grants`) and, on
    /// `sealed`, an optional composed-root bind.
    pub shell_derived_interpreter_runtime:
        Vec<crate::interpreter_runtime::DerivedInterpreterRuntime>,
    /// The host's TLS trust store, as staging found it for a capsule with a non-empty
    /// `shell_allow` and a non-empty `network_allow` (see [`crate::trust_store`]). Not a manifest
    /// field: empty from [`capability_policy_from_runtime_manifest`], filled once by
    /// `stage_session` and by `mur run --explain-scope` through
    /// [`crate::trust_store::derive_trust_store`]. Each becomes a readable, non-executable
    /// Landlock grant (`sandbox::resolve_trust_store_grants`), listable for a directory.
    pub shell_trust_store: Vec<crate::trust_store::TrustStorePath>,
    /// The host folders `~/.murmur/config.yaml`'s `host_paths:` lends every contained capsule on
    /// this host, checked on this host, in config order (see [`crate::host_paths`]). Not a
    /// manifest field: empty from [`capability_policy_from_runtime_manifest`], filled by `mur run`
    /// (and `--explain-scope`) from the global config alone. On `landlock+seccomp` each becomes a
    /// Landlock rule for commands, plan shell steps and native tools; on `sealed` also a required
    /// bind at its own path. Never executable. WASM tools' preopens do not include them.
    pub host_paths: Vec<crate::host_paths::HostPath>,
    /// Host env var names a WASM guest may observe, from `capabilities.env.allow`. Empty by
    /// default: a guest sees only the runtime's own `MURMUR_*` injection unless the manifest
    /// names a variable here. `shell_strip_env` and the credential patterns still apply on
    /// top, so a credential-shaped name declared here is stripped anyway.
    pub env_allow: Vec<String>,
    /// Execution limits applied to every guest store in the session. Always fully
    /// resolved — `capability_policy_from_runtime_manifest` substitutes a default for
    /// each field `capabilities.limits` omits, so there is no "unset" state here and a
    /// silent manifest yields [`ExecutionLimits::default`] rather than no limits.
    pub limits: ExecutionLimits,
    /// OS-level bounds applied to every *native subprocess* the session spawns — rlimits, the
    /// Linux cgroup v2 scope, and the workdir-size ceiling. Resolved on exactly the same terms
    /// as `limits` above (each omitted `capabilities.resources` field replaced by its default,
    /// no "unset" state), but a completely different subject: `limits` bounds a WASM guest
    /// inside its wasmtime store, this bounds the processes the host forks. See
    /// [`crate::resources`].
    pub resources: crate::resources::HostResourceLimits,
    /// The manifest's own `capabilities.resources` block, every omitted field still `None`.
    /// [`Self::resources`] is what a contained session applies; an unconfined session (a
    /// [`Self::containment_floor`] of `ContainmentClass::Unconfined`) applies only what this
    /// declares, creates a cgroup scope only when it declares a `cgroup_*` field and runs the
    /// workdir guard only when it declares `workdir_max_bytes`.
    pub declared_resources: murmur_artifact::ResourceCapabilities,
    /// The containment class this capsule asked for, from `capabilities.containment`.
    ///
    /// Unlike every other field here it is a *requirement*, not a grant — and unlike
    /// `StageRequest::declared_containment_floor` it is only the manifest's own vote, before the
    /// workspace config and `--containment` are folded in. `stage_session` overwrites it with the
    /// combined floor as soon as that is known, so anything reading it off a `StagedSession` sees
    /// the effective value; the manifest-only value exists so a `CapabilityPolicy` built straight
    /// from a manifest is still self-consistent.
    ///
    /// Read by `sandbox::ShellEnforcement::resolve` to decide whether a sealed-capable host
    /// actually installs a composed root for this session — a capsule that declared `scoped` must
    /// keep getting `scoped`'s mechanism. See `sandbox::applied_tier`.
    pub containment_floor: murmur_artifact::ContainmentClass,
    /// Whether the capsule's own top-level `capabilities.state` block was declared.
    ///
    /// Not a grant, and the only field here that is not: a durable state store is granted per
    /// *artifact*, so a capsule-wide declaration reaches nothing at all. It is recorded so
    /// `stage_session` can say that once (`W-SEC-014`) instead of leaving an operator to discover
    /// an empty directory, and it is read by nothing else.
    pub state_declared: bool,
    /// Recorded for the same reason [`Self::state_declared`] is, and applied just as little: the
    /// `murmur:conversation/read` grant is per artifact, so a capsule-wide declaration reaches
    /// nothing and `stage_session` says so once (`W-SEC-016`).
    pub conversation_declared: bool,
    /// From `capabilities.plan.submit`: whether this capsule's model is offered the
    /// runtime-provided `submit-plan` tool. `false` by default, which is deny — and which is also
    /// why no `submit-plan` tool manifest is written and the capsule's model never sees the tool
    /// exists.
    ///
    /// It grants no reach of its own: every step of a submitted plan runs through this same
    /// session's tool dispatch, its shell allowlist and its spawn grant, so a plan reaches exactly
    /// what the model could already call one call at a time.
    pub plan_submit: bool,
    /// From `capabilities.install.skill`: the entries naming which skill artifacts a pull
    /// (`manage.pull` or the `pull-artifact` tool) may install into this session, verbatim and in
    /// declaration order. Empty by default, which is deny. With [`Self::install_tool`] also
    /// empty, every pull is refused.
    ///
    /// A pulled skill gains no grant of its own; see `install_grant`.
    pub install_skill: Vec<String>,
    /// From `capabilities.install.tool`: the entries naming which tool artifacts, WASM or native,
    /// a pull may install, verbatim and in declaration order. Empty by default, which is deny.
    ///
    /// A pulled tool joins the session's dispatch allowlist and dispatches on this policy's own
    /// ceiling; it gains no grant of its own.
    pub install_tool: Vec<String>,
    /// From `capabilities.install.from`: `true` for `[stores, sources]`, under which a pull the
    /// stores cannot answer is fetched from the scope's GitHub source. `false` by default.
    pub install_from_sources: bool,
    /// `capabilities.limits.deadline_seconds` exactly as the manifest declared it — `None`
    /// when it declared nothing. Retained undefaulted alongside the fully-resolved `limits`
    /// above purely so hook calls can apply their own, lower default without mistaking an
    /// explicit `600` for silence. See [`Self::hook_limits`].
    pub declared_deadline_seconds: Option<u64>,
}

impl CapabilityPolicy {
    /// Execution limits for hook lifecycle calls — [`Self::limits`] with the hook-specific
    /// deadline default substituted when the manifest declared none.
    ///
    /// A hook call is one bounded piece of work, so it does not inherit the capsule-wide
    /// ten-minute budget that exists for a capsule's whole `run`: a wedged hook would
    /// otherwise stall the session for most of that, every event. An explicitly declared
    /// `deadline_seconds` still wins, for hooks as for every other guest.
    #[must_use]
    pub fn hook_limits(&self) -> ExecutionLimits {
        self.limits.for_hook_calls(self.declared_deadline_seconds)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InstalledArtifactSummary {
    pub name: String,
    pub version: String,
    pub runtime: ArtifactRuntime,
    pub implementation: Option<ArtifactImplementation>,
    /// Who pinned the artifact: its `murmur.lock` entry's origin, both for an artifact staged
    /// from the lock and for one `manage.pull()` installed mid-session. An artifact staged
    /// without a lock, and a `source:` skill, are [`murmur_artifact::LockOrigin::Operator`].
    pub origin: murmur_artifact::LockOrigin,
}

#[derive(Clone)]
pub(crate) struct StagedHookArtifact {
    pub name: String,
    /// The name this hook's `artifacts:` entry declares, scope included: what a message about a
    /// setting on that entry names it by.
    pub entry: String,
    pub version: String,
    pub component: Component,
    /// Behavioral contract (binding/execution_mode/commit_policy) from the hook artifact's
    /// *own* bundled manifest. Carries no capability information — see `grant`.
    pub config: HookConfig,
    /// What this hook is allowed to do, lowered and validated at staging time from the
    /// operator's own manifest entry. Applied identically by all three hook instantiation
    /// paths in `hooks.rs`; `HookCapabilityGrant::default()` is full default-deny.
    pub grant: HookCapabilityGrant,
    /// What to do when this hook's async job queue is full — from the operator's manifest
    /// entry (via [`ArtifactRequest::on_overflow`]), on the same terms as `grant` and never
    /// from `config`. Read only when `config.execution_mode` is `Async`; a blocking hook has
    /// no queue to overflow.
    pub on_overflow: murmur_artifact::HookOverflowPolicy,
    /// This hook's own credential gateway, when its operator entry declares `gateway:`. Attached
    /// to every store the hook is instantiated on, `on-stage` included. Filled once the session's
    /// gateways are staged, before any hook runs.
    pub gateway: Option<Arc<crate::credential_gateway::CredentialGateway>>,
}

/// How `mur run --resume` puts the loaded conversation in front of the model.
///
/// `Full` is often the cheaper of the two: a verbatim reload can land on the provider's own
/// prompt cache, while compaction changes the prefix from the first altered token, guarantees a
/// cache miss, and costs an inference call to produce the summary. `Compact` is the answer when
/// the conversation would not fit the context window at all.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum ResumeMode {
    #[default]
    Full,
    Compact,
}

/// One operator act: continue *this* conversation, whatever the capsule's own
/// `lifecycle.conversation` policy says about carrying one between tasks.
///
/// Sugar over [`StageRequest::context_id`], never a second load path: the CLI resolves a session
/// address to the context id that session ran under and fills both this and `context_id`. What
/// this adds is the override — the record is loaded even under `lifecycle.conversation:
/// stateless` — and the provenance the trace records.
#[derive(Debug, Clone)]
pub struct ResumeRequest {
    /// The prior session's id, recorded verbatim as `session_start.resumed_from`: the id its
    /// trace's own `session_start` carries, or the trace's directory name when it records none.
    pub from_session: String,
    /// The absolute directory holding the resumed trace's `trace.jsonl`: the prior session's
    /// records directory, or wherever that trace was copied. The lost-work reconciliation reads
    /// this file and appends its `shell_lost` markers to it.
    pub trace_dir: PathBuf,
    pub mode: ResumeMode,
}

#[derive(Debug, Clone)]
pub struct StageRequest {
    pub manifest_dir: PathBuf,
    pub capsule_name: String,
    pub capsule_version: String,
    /// The full `<scope>/<name>` the capsule was resolved by, for one launched by name
    /// (`mur run --capsule`). The session registers with mur-roost and reports its completion
    /// under it. `None` for a capsule run from a project directory, which is `local/<capsule_name>`.
    pub capsule_artifact: Option<String>,
    pub capsule_component_bytes: Vec<u8>,
    pub artifacts: Vec<ArtifactRequest>,
    pub allowlisted_tools: HashSet<String>,
    pub lock_expectations: Option<Vec<LockExpectation>>,
    pub capability_policy: CapabilityPolicy,
    pub inference: Option<InferenceConfig>,
    /// The global config file a `${NAME}` in an artifact's `gateway.api_key` is looked up in first, as
    /// `credentials.NAME`, and re-read from while the session runs. `None` looks in the launching
    /// environment only.
    pub credentials_file: Option<PathBuf>,
    /// Whether `inference` above had its system-prompt declaration replaced by the caller
    /// (`mur run --system-prompt`) rather than read from the manifest. Purely a provenance
    /// signal for the trace: the override itself is already applied to `inference`, which by
    /// this point is indistinguishable from a manifest that declared the same prompt inline.
    /// `true` even when the override cleared the prompt to nothing.
    pub system_prompt_overridden: bool,
    pub context: Option<ContextConfig>,
    /// Context id every `task.md` task of this launch runs under, from `mur run --context`.
    /// `None` mints a fresh one per task. Validated by `stage_session` as one path segment: it is
    /// a directory name in the conversation record path.
    pub context_id: Option<String>,
    /// `mur run --resume <session>`, already resolved: `context_id` above holds the context that
    /// session ran under, and this carries the session's own id and the mode. `None` is every
    /// ordinary launch. `stage_session` refuses a resume whose context kept no record on disk,
    /// and a `--resume-mode compact` with no hook bound to `on-compaction`.
    pub resume: Option<ResumeRequest>,
    /// `mur run --forget-session`: drop the harness session `context_id` names before this
    /// launch's first task, so that task starts a new conversation under the same context id.
    /// `false` on every ordinary launch. `stage_session` refuses it for a capsule whose transport
    /// keeps no harness session.
    pub forget_session: bool,
    /// OTLP/HTTP endpoint for span export; None = no external OTel emission.
    pub otel_endpoint: Option<String>,
    /// JSON-serialized EvalConfig injected into hook WASI env as MURMUR_EVAL_CONFIG.
    pub eval_config_json: Option<String>,
    /// Dataset case identifier injected as MURMUR_CASE_ID when running a dataset.
    pub case_id: Option<String>,
    /// Dataset identifier injected as MURMUR_DATASET_ID when running a dataset.
    pub dataset_id: Option<String>,
    /// Lifecycle config from the manifest (None = use LifecycleConfig::default()).
    pub lifecycle: Option<LifecycleConfig>,
    /// Optional override applied on top of lifecycle (e.g. from CLI flags or mur-roost).
    pub lifecycle_override: Option<LifecycleOverride>,
    /// Trace config from the manifest (None = defaults apply: input captured, output not).
    pub trace: Option<murmur_artifact::TraceConfig>,
    /// The operator's directory to use as the capsule's work folder, `mur run --workdir`; the
    /// runtime creates nothing in it. `None` uses `<manifest_dir>/workdir/<session_id>`.
    pub workdir: Option<std::path::PathBuf>,
    /// Address to bind the HTTP server on. Defaults to "127.0.0.1" for local runs.
    /// Use "0.0.0.0" when deploying to a VM so the capsule is reachable externally.
    pub bind_addr: String,
    /// Internal port from `network.internal_port` in the manifest. When `Some`, the runtime
    /// binds strictly to this port and errors if it is already in use. When `None`, the OS
    /// assigns a port.
    pub internal_port: Option<u16>,
    /// Minimum containment class this session must achieve, already combined across every
    /// source that asked for one (manifest / workspace config / `--containment`) by taking the
    /// strongest. Required rather than optional: "nobody declared anything" is
    /// [`ContainmentClass::Advisory`], an explicit floor every host clears, not an absence.
    /// `stage_session` refuses to stage when the host cannot meet it.
    pub declared_containment_floor: murmur_artifact::ContainmentClass,
    /// The manifest's top-level `exports:` block. `None` — and `Some` with no `files:` — means
    /// the resource plane is not declared and every request to it is denied.
    ///
    /// Carried beside `capability_policy` rather than inside it: an export is a disclosure the
    /// operator makes, not a capability the guest holds, and nothing derived from this field ever
    /// reaches the achieved containment class.
    pub exports: Option<murmur_artifact::Exports>,
    /// The manifest's top-level `control:` block. `None` means the session has no control
    /// surface: no token is minted and every request under `/control` is answered `404`.
    ///
    /// Beside `exports` and on the same terms: what a controller may change is not a capability
    /// the guest holds, and nothing derived from it reaches the containment class.
    pub control: Option<murmur_artifact::ControlConfig>,
    /// The manifest's `network.authentication` block. `None` is a public door, which answers every
    /// caller that reaches the port; `Some` makes staging mint this session's door key and tokens.
    pub door_authentication: Option<murmur_artifact::NetworkAuthentication>,
    /// The approval a delegated child was handed by its parent, presented once at registration.
    ///
    /// `None` for every top-level launch, which the daemon admits against its own `--spawn-allow`
    /// instead. Unlike [`StagedSession::spawn_credential`], this exists before staging: it names
    /// an artifact rather than a session, and the parent already held it when it started this
    /// process.
    pub spawn_grant: Option<SpawnApproval>,
    /// `spend.machine_tokens_per_day` from the operator's effective `config.yaml`. `Some` makes
    /// staging open the shared daily spend ledger for a `transport: http` session with a driver,
    /// and refuse when it cannot; `None` — every launch that does not read the operator config —
    /// keeps no ledger and is not counted.
    pub machine_tokens_per_day: Option<u64>,
    /// The formation this session is a member of.
    ///
    /// `None` is the ordinary case: a session nobody placed in a formation, including a
    /// standalone capsule that delegates and every child it spawns. `mur run` sets it from
    /// [`crate::formation::FormationId::from_env`], which reads what the launcher of a formation's
    /// members, or a member's runtime for its child, put in the environment.
    pub formation_id: Option<FormationId>,
    /// What this session's formation launcher handed it: its roster name, what its door checks a
    /// formation token with, and the tokens and addresses of the members it may call.
    ///
    /// `Some` only for a session `mur run --roster` launched as a member: `mur run` reads it from
    /// the channel [`crate::formation_credentials::FormationMember::from_env`] names, and refuses
    /// the launch when that cannot be read. The door verifies formation tokens with it, every WASM
    /// guest is handed its callees' names, and the runtime's egress resolves and authorizes calls
    /// to them. No native process and no delegated child is handed any of it.
    pub formation_member: Option<Arc<FormationMember>>,
    /// Where a pull the stores cannot answer is fetched from, for a capsule granted
    /// `capabilities.install.from: [stores, sources]`. `None` fetches from no source: every pull
    /// is answered by the stores alone.
    pub source_supply: Option<Arc<dyn crate::artifact_supply::SourceSupply>>,
    /// The `(st_dev, st_ino)` the accessible workdir must have, compared with `fstat` of the
    /// directory `stage_session` opens at [`Self::workdir`] before anything is written beneath
    /// it; a mismatch refuses the launch with [`crate::RuntimeError::WorkdirIdentityMismatch`].
    /// Set by a delegating parent through `mur run --workdir-identity`, for the child directory it
    /// created. `None` checks nothing.
    pub expected_workdir_identity: Option<(u64, u64)>,
    /// What a delegating parent copied into [`Self::workdir`] before starting this session, from
    /// `mur run --parent-workdir-snapshot`. Reported as `session_start.effective_grants.
    /// parent_workdir` and in `MURMUR.md`, and grants nothing. `None` for every other session.
    pub parent_workdir_snapshot: Option<crate::workdir_snapshot::WorkdirSnapshot>,
    /// The hard limit `mur run --deadline` set, counted from the moment `mur run` started. Only
    /// [`crate::launch_session_handling_sigterm`] on an agent capsule enforces one: reached, it
    /// ends the session through the termination routine a first `SIGTERM` begins. Every other
    /// launch of a session staged with one is refused with
    /// [`crate::RuntimeError::SessionDeadlineUnenforceable`]. `None` sets no limit.
    pub session_deadline: Option<crate::session_status::SessionDeadline>,
    /// The manifest file this launch read, copied into the runtime directory under its own file
    /// name: `--manifest`, or the project's `murmur.yaml`. `None` for a launch that read none from
    /// disk — an installed capsule run by name — which copies nothing.
    pub manifest_file: Option<PathBuf>,
}

/// The `transport: process` driver this session runs its harness through, compiled and described
/// once at staging.
///
/// Not a tool: it is granted nothing, never enters `tool_components`, and is reached only by the
/// process runner, which instantiates a fresh [`crate::process_driver::ProcessDriver`] from
/// `component` per harness run.
pub struct StagedProcessDriver {
    pub(crate) name: String,
    pub(crate) version: String,
    pub(crate) component: Component,
    pub(crate) description: crate::process_driver::Description,
    /// The absolute path of the executable to spawn, already resolved and checked.
    pub(crate) binary: PathBuf,
    /// Where [`Self::binary`]'s name came from, for diagnostics and the trace: `inference.command`
    /// or the process driver's `describe()`.
    pub(crate) binary_source: String,
}

/// The per-tool settings a native tool's spawn carries, lowered at staging from the capsule
/// operator's own entry for that tool, never from the artifact's bundled `murmur.yaml`, so a tool
/// pulled from a registry cannot configure itself or name a store.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub(crate) struct NativeToolSettings {
    /// The entry's `config:` block as compact JSON, from
    /// [`crate::artifact_config::lower_artifact_config`]: the value of
    /// [`crate::artifact_config::ARTIFACT_CONFIG_ENV`] in this tool's environment. `None` leaves
    /// the variable unset.
    pub(crate) config_json: Option<String>,
    /// The store `capabilities.state` names, created by
    /// [`crate::state_store::ensure_state_store`]: the value of
    /// [`crate::state_store::STATE_DIR_ENV`], and on `KernelFull` and `KernelSealed` a
    /// read-write path granted to this tool's process tree alone
    /// ([`crate::sandbox::ShellEnforcement::for_native_entrypoint`]). `None` grants nothing and
    /// leaves the variable unset.
    pub(crate) state_dir: Option<PathBuf>,
}

pub struct StagedSession {
    pub session_id: String,
    /// The work folder: the `--workdir` directory, or `<manifest-dir>/workdir/<session-id>`
    /// without one. Every command's current directory and every WASM tool's `"."`. The runtime
    /// creates nothing beneath it, and reads beneath it only through [`Self::accessible_root`].
    pub accessible_workdir: std::path::PathBuf,
    /// [`Self::accessible_workdir`], held open.
    pub accessible_root: crate::workdir_fs::WorkdirRoot,
    /// `<manifest-dir>/workdir`, which holds the work folder the runtime created for a launch
    /// without `--workdir`, and those of earlier such launches that `trace.retain` prunes. `None`
    /// under `--workdir`, whose folder is the operator's and is never pruned.
    pub(crate) created_workdirs: Option<PathBuf>,
    /// The session's runtime directory, `~/.murmur/sessions/<session-id>/runtime`: the value of
    /// `$MURMUR_RUNTIME_DIR`, holding `MURMUR.md`, `task.md`, `murmur.yaml`, `tools/` and
    /// `plans/`. Removed when the session ends; the runtime reads and writes beneath it only
    /// through [`Self::runtime_root`].
    pub runtime_dir: PathBuf,
    /// [`Self::runtime_dir`], held open, sharing [`Self::accessible_root`]'s refusal latch.
    pub runtime_root: crate::workdir_fs::WorkdirRoot,
    /// The three held directories beneath the session's runtime directory, with the records
    /// directory beside them.
    pub(crate) runtime_dirs: crate::runtime_dir::RuntimeDirRoots,
    /// The session's records directory, `~/.murmur/records/<workdir root>/<session-id>`: the
    /// value of `$MURMUR_RECORDS_DIR`, holding `trace.jsonl`, `blobs/`, `logs/`,
    /// `completion.json`, `hooks/`, `out/`, `tmp/`, `home/` and `children/`. Kept after the
    /// session ends; the runtime reads and writes beneath it only through
    /// [`Self::records_root`].
    pub records_dir: PathBuf,
    /// [`Self::records_dir`], held open, sharing [`Self::accessible_root`]'s refusal latch.
    pub records_root: crate::workdir_fs::WorkdirRoot,
    /// The file name the launched manifest was copied into the runtime directory under, which
    /// the `[Capsule]` notice and `MURMUR.md` name. `None` when the launch read no manifest file.
    pub(crate) manifest_file_name: Option<String>,
    /// What `MURMUR.md` is rendered from, for an agent session; `None` for a script capsule,
    /// which gets no `MURMUR.md`. Its identity's capsule URL is empty until launch.
    pub(crate) murmur_md: Option<crate::murmur_md::MurmurMdInputs>,
    /// What staging read for `inference.instruction_files`, once, for every task of the launch.
    /// `None` when the manifest does not set the key, and for a script capsule.
    pub(crate) instruction_files: Option<crate::instruction_files::InstructionFiles>,
    /// Owns the runtime directory: dropping it removes the directory, so it is held until the
    /// session ends — `launch` takes it and drops it on return, and a `StagedSession` dropped
    /// unlaunched drops it with itself.
    pub(crate) runtime_dir_guard: Option<crate::runtime_dir::RuntimeDir>,
    pub(crate) manifest_dir: PathBuf,
    pub(crate) capsule_name: String,
    pub(crate) capsule_version: String,
    /// [`StageRequest::capsule_artifact`], or `local/<capsule_name>` for a project run.
    pub(crate) capsule_artifact: String,
    pub(crate) capsule_url: String,
    pub resolved_lock_artifacts: Vec<ResolvedLockArtifact>,
    pub(crate) installed_artifacts: Vec<InstalledArtifactSummary>,
    /// Where staging took the facts about each allowlisted binary it examined, for
    /// `session_start.host_facts`.
    pub(crate) host_facts: Vec<crate::host_facts::HostFactsUse>,
    /// Each staged artifact's full name, by the tool name the session knows it by.
    pub(crate) artifact_full_names: std::collections::HashMap<String, String>,
    pub(crate) inference: Option<InferenceConfig>,
    /// Every credential gateway of the session, built at staging from each artifact entry's
    /// `gateway:` block and the artifact's own `upstream_auth:` declaration. Empty when no entry
    /// declares one.
    pub(crate) gateways: crate::credential_gateway::GatewayTable,
    /// This session's spend account, built at staging from `inference.max_session_tokens` and
    /// [`StageRequest::machine_tokens_per_day`]. Always present: a session with no ceiling gets a
    /// meter that admits everything, so every driver call has an admission to open.
    pub(crate) spend: Arc<crate::spend::SpendMeter>,
    /// What a controller has changed on this session, built at staging from
    /// [`StageRequest::control`]. `None` for a capsule with no `control:` block, which mints no
    /// control token and answers `404` under `/control`.
    pub(crate) control: Option<Arc<crate::control_plane::ControlState>>,
    /// The session's driver choices, the primary first, each with the credential staging left
    /// it and, under `transport: http`, its model limits. `None` for a capsule with no inference
    /// block.
    pub(crate) driver_choices: Option<crate::driver_choice::DriverChoices>,
    /// The output cap a hook's `run-inference` sends for each model it may call.
    pub(crate) hook_output_caps: crate::model_limits::HookOutputCaps,
    /// The `MURMUR_INFERENCE_DRIVER_CONFIG` a hook's `run-inference` driver sees while the hook
    /// serves `on-compaction`: `inference.driver.config` with `inference.compaction.driver_config`
    /// laid over it. `None` when the manifest sets no overlay, so the call sees the session's.
    pub(crate) compaction_driver_config: Option<String>,
    /// Copied from [`StageRequest::door_authentication`], which the cards are built from.
    pub(crate) door_authentication: Option<murmur_artifact::NetworkAuthentication>,
    /// This session's door key and tokens, minted at staging when
    /// [`Self::door_authentication`] is declared. The key never leaves this value.
    pub(crate) door_auth: Option<Arc<crate::door_auth::DoorAuth>>,
    /// Copied from [`StageRequest::system_prompt_overridden`] — the only record left that the
    /// prompt in `inference` came from `--system-prompt` and not from the manifest. Passed to
    /// `TraceWriter::open`, which turns it into `session_start.system_prompt_source`.
    pub(crate) system_prompt_overridden: bool,
    pub(crate) context: Option<ContextConfig>,
    /// Copied from [`StageRequest::context_id`] and already validated. Read by `launch_session`,
    /// which uses it in place of a freshly minted id for every `task.md` task.
    pub(crate) context_id: Option<String>,
    /// Copied from [`StageRequest::resume`] and already checked. Read by `launch_session`, which
    /// turns it into `session_start.resumed_from` and the agent loop's record-load override.
    pub(crate) resume: Option<ResumeRequest>,
    /// Copied from [`StageRequest::forget_session`] and already checked against the transport.
    /// Read by `launch_session` as a one-shot: the launch's first task carries the forget, and no
    /// later task of the same launch does.
    pub(crate) forget_session: bool,
    pub(crate) engine: Engine,
    /// The compiled-form cache handle built for [`Self::engine`] and the capsule's writable root.
    /// A mid-session pull compiles through it, so a pulled component is stored and
    /// looked up under the same name a staged one is.
    pub(crate) compiled_forms: crate::compiled_forms::CompiledForms,
    /// `None` for manifest-only agent capsules; `Some` for script capsules with a WASM component.
    pub(crate) capsule_component: Option<Component>,
    pub(crate) tool_components: HashMap<String, Component>,
    /// What each native tool installed at staging needs to start, keyed by tool name. Derived
    /// from the image the runtime wrote; see [`crate::sandbox::NativeEntrypoint`].
    pub(crate) native_entrypoints: HashMap<String, crate::sandbox::NativeEntrypoint>,
    /// What each native tool's own manifest entry hands it, keyed by tool name. Holds an entry
    /// only for a tool whose entry declares `config:` or `capabilities.state`; a tool with no
    /// entry here is spawned with neither variable and no store.
    pub(crate) native_tool_settings: HashMap<String, NativeToolSettings>,
    /// The process driver, for a `transport: process` session that named one. `None` on every
    /// other transport.
    pub(crate) process_driver: Option<Arc<StagedProcessDriver>>,
    /// Per-artifact narrowing for the tool/driver dispatch path, keyed by artifact name and
    /// lowered at staging time from the operator's own manifest entries. Holds an entry only
    /// for artifacts that actually declared a `capabilities:` block — a name absent from this
    /// map (the overwhelmingly common case) runs on the full capsule ceiling, exactly as
    /// before per-artifact narrowing existed.
    pub(crate) artifact_grants: HashMap<String, ToolCapabilityGrant>,
    pub(crate) hook_components: Vec<StagedHookArtifact>,
    pub(crate) allowlisted_tools: HashSet<String>,
    pub(crate) capability_policy: CapabilityPolicy,
    pub(crate) otel_endpoint: Option<String>,
    pub(crate) eval_config_json: Option<String>,
    pub(crate) case_id: Option<String>,
    pub(crate) dataset_id: Option<String>,
    /// Resolved lifecycle config (manifest + override already applied).
    pub(crate) lifecycle: LifecycleConfig,
    /// How much of each turn's driver request this session's trace keeps (resolved from the
    /// manifest's `trace:` block, or [`murmur_artifact::TraceCapture::default`] when it has
    /// none).
    pub(crate) trace_capture: murmur_artifact::TraceCapture,
    /// What bounds this workdir's session directories, from the manifest's `trace.retain` block.
    /// `None` — the overwhelmingly common case — prunes nothing, ever.
    pub(crate) trace_retain: Option<murmur_artifact::TraceRetainConfig>,
    /// Address the HTTP server is bound to (copied from StageRequest::bind_addr).
    pub(crate) bind_addr: String,
    /// Internal port from the manifest (copied from StageRequest::internal_port).
    pub(crate) internal_port: Option<u16>,
    /// This session's containment, decided once by `stage_session` from [`Self::host_probe`],
    /// `StageRequest::declared_containment_floor`, `workdir_exec` and where the runtime directory
    /// lies. The floor refusal, [`Self::scope_report`], [`Self::temp_mapping`], `MURMUR.md`'s exec
    /// facts and the tier `launch_session`'s `sandbox::ShellEnforcement::resolve` installs all
    /// read it, so the class and tier the session reports are the ones its commands run on.
    pub(crate) containment: crate::containment::ContainmentDecision,
    /// The complete effective grant set this session was staged with — the same report
    /// `mur run --explain-scope --json` prints, built by `containment::scope_report_for_tier`
    /// from this session's own policy and [`Self::containment`]. Written verbatim to
    /// `trace.jsonl`'s `session_start` as `effective_grants`, and the source of the
    /// `containment_declared`/`containment_achieved`/`workdir_exec` fields alongside it, so the
    /// summary and the full report cannot disagree.
    ///
    /// The session's class here, `scope_report.achieved_containment`, is copied from
    /// [`Self::containment`] when the report is built, so the two cannot drift.
    pub(crate) scope_report: crate::containment::ScopeReport,
    /// The single host reading this session was staged against, taken at the top of
    /// `runtime::stage_session`, which [`Self::containment`] was decided from. `launch_session`
    /// passes both to `sandbox::ShellEnforcement::resolve` rather than taking a new reading, so the
    /// mechanisms installed at launch are the ones [`Self::scope_report`] already claims.
    pub(crate) host_probe: crate::sandbox::HostProbe,
    /// The provider tools this session staged with and what a runtime pull of one is checked
    /// against. Its `declared` list is empty for a capsule that declares no provider tool.
    pub(crate) provider_tool_terms: crate::provider_tool::ProviderToolTerms,
    /// Whether a literal `/tmp` reaches the session's temporary folder, decided once by
    /// `stage_session` from [`Self::host_probe`]: [`Self::scope_report`]'s `temp_dir` says it and
    /// `launch_session` hands it to the enforcement that installs it.
    pub(crate) temp_mapping: crate::temp_dir::TempMapping,
    /// The declared read-only surface, lowered and validated by `stage_session` — a malformed
    /// entry refuses the launch there rather than surfacing at the first call. Empty for every
    /// capsule that declared no `capabilities.filesystem.read_only`.
    pub(crate) protected_paths: crate::protected_paths::ProtectedPaths,
    /// What each staged tool's own `input_schema` declared about where its inputs go, lowered by
    /// `stage_session` from the manifests it wrote into `<runtime dir>/tools/`. Empty for a capsule
    /// that declared no `read_only`, which reads no schema at all, and for one whose tools
    /// annotated nothing.
    pub(crate) tool_annotations: crate::tool_annotations::ToolAnnotationMap,
    /// The declared read-only file surface, with its root already checked against this session's
    /// accessible workdir by `stage_session` — a root that resolves outside it refuses the launch
    /// rather than being served. `None` means no resource plane.
    pub(crate) exports_files: Option<murmur_artifact::FileExport>,
    /// The declared peer-handoff surface, with its root already checked against this session's
    /// accessible workdir and its `max_ttl` already checked against `lifecycle.after_task` by
    /// `stage_session`. `None` means no peer plane: nothing mints, and `/resources/peer/` answers
    /// `no_peer_plane`.
    pub(crate) exports_peer_files: Option<murmur_artifact::PeerFilesExport>,
    /// Whether the door serves a task a peer capsule's runtime sends, read from the manifest's
    /// own `exports.peer_tasks` by [`murmur_artifact::Exports::accepts_peer_tasks`]. `false`
    /// answers every peer-origin request on the door with `403 peer_not_accepted`.
    pub(crate) accepts_peer_tasks: bool,
    /// Registry used to resolve this session's artifacts, retained so a pull can
    /// resolve additional artifacts at runtime after staging has completed.
    pub(crate) registry: Arc<dyn Registry>,
    /// Copied from [`StageRequest::source_supply`]: where a runtime pull the stores cannot answer
    /// is fetched from. `None` unless the capsule declared `capabilities.install.from: [stores,
    /// sources]` and the launcher supplied one.
    pub(crate) source_supply: Option<Arc<dyn crate::artifact_supply::SourceSupply>>,
    /// Keeps this session's epoch ticker running: without it no epoch deadline set on any
    /// store built from `engine` can ever fire. Held here (rather than detached) so ticking
    /// stops when the session's `StagedSession` drops. `launch_session` moves the other
    /// fields out one by one, which leaves this one in place until the session returns.
    pub(crate) _epoch_ticker: EpochTicker,
    /// The session's record of component calls the wall-clock bound stopped, created at staging
    /// beside the epoch ticker and cloned into every store that runs a component and into the
    /// session's trace writer, which drains it.
    pub(crate) call_deadlines: CallDeadlines,
    /// The credential this session presents when it asks `mur-roost` to spawn a capsule.
    ///
    /// `None` for every session that cannot delegate, which is nearly all of them. Set by the
    /// caller between `stage_session` and `launch_session` rather than carried in
    /// [`StageRequest`]: the credential names a session id, and `stage_session` is what mints one,
    /// so it cannot exist before staging returns.
    pub(crate) spawn_credential: Option<SpawnCredential>,
    /// Copied from [`StageRequest::spawn_grant`]. Read by `launch_session`, which presents it at
    /// `POST /register` and never again — an approval covers one launch.
    pub(crate) spawn_grant: Option<SpawnApproval>,
    /// The handle the spawning capsule's launcher injected, read once per process at staging.
    ///
    /// `None` for every capsule nobody delegated. Its `session_id` and `delegation_id` become
    /// `session_start.spawned_by` and `session_start.delegation_id`; its `report_to` — present
    /// only for a parent that wants a completion — is what `launch_session`'s reporting guard
    /// posts to. Read here rather than at each of those two points so one process reads
    /// `MURMUR_SPAWNER` once and both readers see the same value.
    pub(crate) spawner: Option<crate::delegation::SpawnerHandle>,
    /// Copied from [`StageRequest::formation_id`]. Hooks see it as `MURMUR_FORMATION_ID` and the
    /// delegation plane hands it to every child this session spawns.
    pub(crate) formation_id: Option<FormationId>,
    /// Copied from [`StageRequest::formation_member`].
    pub(crate) formation_member: Option<Arc<FormationMember>>,
    /// This member's formation lifeline, attached after staging by
    /// [`StagedSession::attach_formation_lifeline`]. A process property rather than a staged
    /// one, so it never travels on [`StageRequest`].
    #[cfg(unix)]
    pub(crate) formation_lifeline: Option<crate::lifeline::FormationLifeline>,
    /// This delegated child's spawner lifeline, attached after staging by
    /// [`StagedSession::attach_spawner_lifeline`]. A process property on the same terms as
    /// [`Self::formation_lifeline`].
    #[cfg(unix)]
    pub(crate) spawner_lifeline: Option<crate::lifeline::SpawnerLifeline>,
    /// [`StageRequest::session_deadline`], carried to the launch that enforces it.
    pub(crate) session_deadline: Option<crate::session_status::SessionDeadline>,
}

impl StagedSession {
    /// Writes `bytes` as the task input, `task.md` in the runtime directory, which the capsule's
    /// commands can read and not change. A launch with one finds nothing to import from the
    /// workdir.
    pub fn write_task_input(&self, bytes: &[u8]) -> Result<(), crate::workdir_fs::WorkdirError> {
        self.runtime_root.write_file(
            crate::runtime::TASK_INPUT_FILE,
            bytes,
            crate::workdir_fs::DEFAULT_FILE_MODE,
        )
    }

    /// Writes `bytes` as `rel` in the accessible workdir before launch, the folder the capsule's
    /// tools see as their working directory. A link at `rel` is refused rather than written
    /// through, and missing parent directories are created.
    pub fn write_workdir_file(
        &self,
        rel: &str,
        bytes: &[u8],
    ) -> Result<(), crate::workdir_fs::WorkdirError> {
        self.accessible_root
            .write_file(rel, bytes, crate::workdir_fs::DEFAULT_FILE_MODE)
    }

    /// Hands this session the credential it will present to the spawning daemon.
    pub fn set_spawn_credential(&mut self, credential: SpawnCredential) {
        self.spawn_credential = Some(credential);
    }

    /// The credential this session presents, if it was granted one.
    pub fn spawn_credential(&self) -> Option<&SpawnCredential> {
        self.spawn_credential.as_ref()
    }

    /// The grant set this session was staged with, as written to the trace.
    pub fn scope_report(&self) -> &crate::containment::ScopeReport {
        &self.scope_report
    }

    /// Every door token this session minted, operator first, then the declared credentials by
    /// name. Empty for a capsule that declares no `network.authentication`.
    pub fn door_tokens(&self) -> Vec<(String, crate::door_auth::DoorToken)> {
        self.door_auth
            .as_ref()
            .map(|auth| auth.tokens().to_vec())
            .unwrap_or_default()
    }

    /// The formation this session is a member of, or `None` for a session nobody placed in one.
    pub fn formation_id(&self) -> Option<&FormationId> {
        self.formation_id.as_ref()
    }

    /// Hand this session the lifeline its process was launched with. Honoured only by
    /// [`crate::launch_session_handling_sigterm`] on an agent capsule: every other launch of a
    /// session holding one is refused with [`crate::RuntimeError::FormationLifelineUnreadable`]
    /// before anything is instantiated.
    #[cfg(unix)]
    pub fn attach_formation_lifeline(&mut self, lifeline: crate::lifeline::FormationLifeline) {
        self.formation_lifeline = Some(lifeline);
    }

    /// Hand this session the spawner lifeline its process was launched with. Honoured only by
    /// [`crate::launch_session_handling_sigterm`], for an agent and a script capsule alike: every
    /// other launch of a session holding one is refused with
    /// [`crate::RuntimeError::SpawnerLifelineUnreadable`] before anything is instantiated.
    #[cfg(unix)]
    pub fn attach_spawner_lifeline(&mut self, lifeline: crate::lifeline::SpawnerLifeline) {
        self.spawner_lifeline = Some(lifeline);
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LaunchResult {
    pub session_id: String,
    /// The session's work folder, [`StagedSession::accessible_workdir`].
    pub workdir: PathBuf,
    /// The session's records directory, where its trace is, kept after the session.
    pub records_dir: PathBuf,
    /// How a launch that reports no error ended.
    pub ending: LaunchEnding,
}

/// How a launch that reports no error ended. A launch whose task did not complete for any other
/// reason is a [`crate::RuntimeError::TaskDidNotComplete`], not an ending.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LaunchEnding {
    /// Every task the launch ran completed, or it ran none: `session_end.exit_status` `ok`.
    Completed,
    /// The session's formation ended while it was running a task, and the wind-down that followed
    /// cancelled that task: `session_end.exit_status` `formation_ended`.
    FormationEnded,
    /// The session reached its `mur run --deadline`, and the wind-down that followed cancelled
    /// the task it was running, or it reached it before any task ran:
    /// `session_end.exit_status` `deadline_reached`.
    DeadlineReached,
}

impl LaunchEnding {
    /// The `exit_status` `session_end` and `on-session-end` carry for this ending.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Completed => "ok",
            Self::FormationEnded => "formation_ended",
            Self::DeadlineReached => "deadline_reached",
        }
    }
}

pub fn capability_policy_from_runtime_manifest(
    manifest: &murmur_artifact::RuntimeManifest,
) -> CapabilityPolicy {
    let caps = manifest.capabilities.as_ref();
    let network_allow = caps
        .and_then(|c| c.network.as_ref())
        .map(|network| network.allow.clone())
        .unwrap_or_default();
    // Absent `capabilities.peer_fetch` block means denied, on the same terms — and separately
    // from `network.allow`, which this deliberately does not read.
    let peer_fetch_allow = caps
        .and_then(|c| c.peer_fetch.as_ref())
        .map(|peer_fetch| peer_fetch.allow.clone())
        .unwrap_or_default();

    // Absent `capabilities.network` block, or absent key within it, both mean denied.
    let unix_sockets_allowed = caps
        .and_then(|c| c.network.as_ref())
        .is_some_and(|network| network.unix_sockets);

    // Absent `capabilities.plan` block, or `submit: false` within it, both mean denied.
    let plan_submit = caps
        .and_then(|c| c.plan.as_ref())
        .is_some_and(|plan| plan.submit);

    // Absent `capabilities.install` block means both lists are empty, which is deny.
    let install = caps.and_then(|c| c.install.as_ref());
    let install_skill = install.map(|i| i.skill.clone()).unwrap_or_default();
    let install_tool = install.map(|i| i.tool.clone()).unwrap_or_default();
    let install_from_sources = install.is_some_and(|i| i.from.sources());

    // Recorded, never applied: see `CapabilityPolicy::state_declared`.
    let state_declared = caps.is_some_and(|c| c.state.is_some());
    let conversation_declared = caps.is_some_and(|c| c.conversation.is_some());

    let filesystem_scope = caps
        .and_then(|c| c.filesystem.as_ref())
        .and_then(|filesystem| filesystem.scope.clone());
    // Absent `capabilities.filesystem` block, or absent key within it, both mean denied — the same
    // shape as `unix_sockets_allowed` above, for the same reason: the widening has to be declared.
    let workdir_exec_allowed = caps
        .and_then(|c| c.filesystem.as_ref())
        .is_some_and(|filesystem| filesystem.workdir_exec);
    // Absent block, absent key and an empty list are one declaration: nothing is protected.
    let read_only_paths = caps
        .and_then(|c| c.filesystem.as_ref())
        .map(|filesystem| filesystem.read_only.clone())
        .unwrap_or_default();

    let shell_allow = caps
        .and_then(|c| c.shell.as_ref())
        .map(|shell| shell.allow.clone())
        .unwrap_or_default();
    let shell_strip_env = caps
        .and_then(|c| c.shell.as_ref())
        .and_then(|shell| shell.strip_env.clone())
        .unwrap_or_default();
    let shell_baseline_env = caps
        .and_then(|c| c.shell.as_ref())
        .and_then(|shell| shell.baseline_env.clone())
        .unwrap_or_default();
    let shell_interpreter_runtime = caps
        .and_then(|c| c.shell.as_ref())
        .map(|shell| shell.interpreter_runtime.clone())
        .unwrap_or_default();
    let shell_staged_runtime = caps
        .and_then(|c| c.shell.as_ref())
        .map(|shell| shell.staged_runtime.clone())
        .unwrap_or_default();

    let spawn_allow = caps
        .and_then(|c| c.spawn.as_ref())
        .map(|spawn| spawn.allow.clone())
        .unwrap_or_default();
    let spawn_workdir_snapshot = caps
        .and_then(|c| c.spawn.as_ref())
        .map(|spawn| spawn.workdir_snapshot.clone())
        .unwrap_or_default();

    let env_allow = caps
        .and_then(|c| c.env.as_ref())
        .map(|env| env.allow.clone())
        .unwrap_or_default();

    let limits = ExecutionLimits::resolve(caps.and_then(|c| c.limits.as_ref()));
    let declared_deadline_seconds = caps
        .and_then(|c| c.limits.as_ref())
        .and_then(|l| l.deadline_seconds);

    let resources = crate::resources::resolve(caps.and_then(|c| c.resources.as_ref()));

    CapabilityPolicy {
        network_allow,
        peer_fetch_allow,
        unix_sockets_allowed,
        filesystem_scope,
        workdir_exec_allowed,
        read_only_paths,
        shell_allow,
        spawn_allow,
        spawn_workdir_snapshot,
        shell_strip_env,
        shell_baseline_env,
        shell_interpreter_runtime,
        shell_staged_runtime,
        shell_derived_interpreter_runtime: Vec::new(),
        shell_trust_store: Vec::new(),
        host_paths: Vec::new(),
        env_allow,
        limits,
        resources,
        declared_resources: caps.and_then(|c| c.resources).unwrap_or_default(),
        containment_floor: caps.and_then(|c| c.containment).unwrap_or_default(),
        state_declared,
        conversation_declared,
        plan_submit,
        install_skill,
        install_tool,
        install_from_sources,
        declared_deadline_seconds,
    }
}

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

    fn policy_for(manifest_yaml: &str) -> CapabilityPolicy {
        let manifest = murmur_artifact::RuntimeManifest::from_yaml_str(manifest_yaml)
            .expect("manifest fixture must parse");
        capability_policy_from_runtime_manifest(&manifest)
    }

    /// `read_only` reaches the policy in declaration order, trimmed, with empty entries already
    /// dropped by the manifest parser — and a capsule that declares none gets an empty list, which
    /// is what makes the dispatch check free for it.
    #[test]
    fn read_only_paths_are_carried_through_and_default_to_empty() {
        let declared = policy_for(
            r#"
name: cap
version: 0.1.0
capabilities:
  filesystem:
    read_only:
      - tests
      - "  bench/fixtures  "
      - "   "
"#,
        );
        assert_eq!(declared.read_only_paths, vec!["tests", "bench/fixtures"]);

        let silent = policy_for(
            "name: cap
version: 0.1.0
",
        );
        assert!(silent.read_only_paths.is_empty());

        let no_key = policy_for(
            r#"
name: cap
version: 0.1.0
capabilities:
  filesystem:
    scope: work
"#,
        );
        assert!(no_key.read_only_paths.is_empty());
    }

    /// A manifest that sets no deadline gives hook calls the hook-specific default while the
    /// capsule's own guests keep the capsule-wide one — one manifest, two budgets.
    #[test]
    fn a_silent_manifest_gives_hook_calls_the_lower_deadline() {
        let policy = policy_for(
            r#"
name: cap
version: 0.1.0
artifacts: []
"#,
        );

        assert_eq!(
            policy.limits.deadline_seconds,
            crate::limits::DEFAULT_DEADLINE_SECONDS
        );
        assert_eq!(
            policy.hook_limits().deadline_seconds,
            crate::limits::HOOK_DEFAULT_DEADLINE_SECONDS
        );
        // The default policy a hand-built fixture gets must agree with the silent-manifest one.
        assert_eq!(
            CapabilityPolicy::default().hook_limits().deadline_seconds,
            crate::limits::HOOK_DEFAULT_DEADLINE_SECONDS
        );
    }

    /// An explicit `deadline_seconds` applies to every guest, hooks included — the hook
    /// default only fills a silence.
    #[test]
    fn an_explicit_deadline_applies_to_hook_calls_too() {
        let policy = policy_for(
            r#"
name: cap
version: 0.1.0
artifacts: []
capabilities:
  limits:
    deadline_seconds: 120
"#,
        );

        assert_eq!(policy.declared_deadline_seconds, Some(120));
        assert_eq!(policy.limits.deadline_seconds, 120);
        assert_eq!(policy.hook_limits().deadline_seconds, 120);
    }

    /// `capabilities.shell.staged_runtime` lowers onto the policy verbatim — binary, source path
    /// and pin all survive, since the pin is what a human later compares across two hosts.
    #[test]
    fn staged_runtime_grants_lower_onto_the_policy() {
        let policy = policy_for(
            r#"
name: cap
version: 0.1.0
artifacts: []
capabilities:
  containment: sealed
  shell:
    allow:
      - python3
    staged_runtime:
      - binary: python3
        source_path: /opt/testbed/conda/envs/django__django
        pin: conda-4.10.3/python-3.9.19
"#,
        );

        assert_eq!(
            policy.shell_staged_runtime,
            vec![murmur_artifact::StagedRuntimeGrant {
                binary: "python3".to_string(),
                source_path: "/opt/testbed/conda/envs/django__django".to_string(),
                pin: "conda-4.10.3/python-3.9.19".to_string(),
            }]
        );
        // The two mechanisms stay separate all the way down: lowering one must not populate the
        // other.
        assert!(policy.shell_interpreter_runtime.is_empty());
        // Derived at staging, never lowered from the manifest.
        assert!(policy.shell_derived_interpreter_runtime.is_empty());
        assert!(policy.shell_trust_store.is_empty());
    }

    /// A shell block that declares no staged runtime lowers to an empty list, not to a default
    /// grant of any kind.
    #[test]
    fn absent_staged_runtime_lowers_to_empty() {
        let policy = policy_for(
            r#"
name: cap
version: 0.1.0
artifacts: []
capabilities:
  shell:
    allow:
      - bash
"#,
        );

        assert!(policy.shell_staged_runtime.is_empty());
    }

    /// The default every existing manifest gets. Three separate "nothing was declared" shapes —
    /// no `capabilities:` at all, no `filesystem:` within it, and a `filesystem:` that only sets
    /// `scope` — must all lower to denied, because each one is a real manifest in the wild and any
    /// of them silently keeping workdir `Execute` would reopen the rename-to-an-allowed-basename
    /// bypass this default exists to close.
    #[test]
    fn workdir_exec_denied_for_every_shape_of_undeclared() {
        for manifest_yaml in [
            r#"
name: cap
version: 0.1.0
artifacts: []
"#,
            r#"
name: cap
version: 0.1.0
artifacts: []
capabilities:
  network:
    allow:
      - https://api.anthropic.com
"#,
            r#"
name: cap
version: 0.1.0
artifacts: []
capabilities:
  filesystem:
    scope: workdir
"#,
        ] {
            let policy = policy_for(manifest_yaml);
            assert!(
                !policy.workdir_exec_allowed,
                "an undeclared workdir_exec must lower to denied: {manifest_yaml}"
            );
        }
    }

    #[test]
    fn workdir_exec_denied_when_declared_false() {
        let policy = policy_for(
            r#"
name: cap
version: 0.1.0
artifacts: []
capabilities:
  filesystem:
    workdir_exec: false
"#,
        );

        assert!(!policy.workdir_exec_allowed);
    }

    #[test]
    fn workdir_exec_allowed_when_declared_true() {
        let policy = policy_for(
            r#"
name: cap
version: 0.1.0
artifacts: []
capabilities:
  filesystem:
    scope: workdir
    workdir_exec: true
"#,
        );

        assert!(policy.workdir_exec_allowed);
        // The two filesystem keys stay independent: lowering one must not disturb the other.
        assert_eq!(policy.filesystem_scope.as_deref(), Some("workdir"));
    }

    /// A manifest with no `capabilities:` block at all denies unix sockets — the field must not
    /// pick up any implicit widening from the "nothing declared" path.
    #[test]
    fn unix_sockets_denied_when_capabilities_block_is_absent() {
        let policy = policy_for(
            r#"
name: cap
version: 0.1.0
artifacts: []
"#,
        );

        assert!(!policy.unix_sockets_allowed);
    }

    /// A declared `network:` block that never mentions the key still denies.
    #[test]
    fn unix_sockets_denied_when_network_block_omits_the_key() {
        let policy = policy_for(
            r#"
name: cap
version: 0.1.0
artifacts: []
capabilities:
  network:
    allow:
      - https://api.anthropic.com
"#,
        );

        assert!(!policy.unix_sockets_allowed);
        assert_eq!(
            policy.network_allow,
            vec!["https://api.anthropic.com".to_string()]
        );
    }

    #[test]
    fn unix_sockets_denied_when_declared_false() {
        let policy = policy_for(
            r#"
name: cap
version: 0.1.0
artifacts: []
capabilities:
  network:
    unix_sockets: false
"#,
        );

        assert!(!policy.unix_sockets_allowed);
    }

    #[test]
    fn unix_sockets_allowed_when_declared_true() {
        let policy = policy_for(
            r#"
name: cap
version: 0.1.0
artifacts: []
capabilities:
  network:
    unix_sockets: true
"#,
        );

        assert!(policy.unix_sockets_allowed);
    }

    /// The opt-in is independent of the IP allowlist in both directions: a non-empty `allow`
    /// does not imply unix sockets, and `unix_sockets: true` does not imply any IP destination.
    #[test]
    fn unix_sockets_opt_in_is_independent_of_the_ip_allowlist() {
        let policy = policy_for(
            r#"
name: cap
version: 0.1.0
artifacts: []
capabilities:
  network:
    unix_sockets: true
    allow:
      - https://api.anthropic.com
"#,
        );

        assert!(policy.unix_sockets_allowed);
        assert_eq!(
            policy.network_allow,
            vec!["https://api.anthropic.com".to_string()]
        );
    }

    /// `CapabilityPolicy::default()` is what the sandbox tests and every `..Default::default()`
    /// call site start from; it must deny, not inherit whatever a future field ordering implies.
    #[test]
    fn default_capability_policy_denies_unix_sockets() {
        assert!(!CapabilityPolicy::default().unix_sockets_allowed);
    }
}