warble_claude_code/

interactive.rs

//! Shared, deliberately small handoff contract for native interactive CLIs.

use crate::error::DispatchError;
use crate::ir::WarbleIr;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::fs;
use std::path::{Component, Path, PathBuf};
use url::{Host, Url};

pub const LAUNCH_SPEC_VERSION: &str = "1";
pub const NATIVE_SESSION_LAUNCH_SPEC_VERSION: &str = "2";
/// Native Sessions v4 adds the closed, producer-authored first prompt as the
/// final vendor argv element. The host must validate the whole argv exactly;
/// it must never write a prompt through the PTY after spawning.
pub const NATIVE_SESSION_MCP_LAUNCH_SPEC_VERSION: &str = "4";
pub const NATIVE_SCOPE_VERSION: &str = "3";
pub const NATIVE_WREN_RUNTIME_VERSION: &str = "1";
pub const NATIVE_MCP_DESCRIPTOR_VERSION: &str = "1";
pub const NATIVE_MCP_SERVER_NAME: &str = "genbi_session";
pub const NATIVE_MCP_CREDENTIAL_ENV_VAR: &str = "WARBLE_MCP_CONNECTION_CREDENTIAL";
pub const NATIVE_DASHBOARD_SAVE_TOOL: &str = "save_dashboard";
pub const NATIVE_PERSIST_ANSWER_TOOL: &str = "persist_answer";
/// Offered wherever `persist_answer` is: a read-only SQL query run through the host.
pub const NATIVE_QUERY_TOOL: &str = "query";
/// The only project-creation root a native Setup TUI may receive. GenBI sets
/// this after it revalidates the producer-authored v4 bootstrap_root; callers
/// and browsers never supply it.
pub const NATIVE_SETUP_BOOTSTRAP_ROOT_ENV_VAR: &str = "WARBLE_SETUP_BOOTSTRAP_ROOT";

const CODEX_WREN_PERMISSION_PROFILE: &str = "warble_native_wren";

const SETUP_RECOVERY_REPORT_VERSION: &str = "1";
const MAX_SAFE_SETUP_RECOVERY_SEQUENCE: u64 = 9_007_199_254_740_991;

/// The public, transport-independent input schema for the `genbi_session`
/// `report_setup_recovery` MCP tool. The host validates monotonicity against its last accepted
/// report; the schema deliberately carries neither identity nor any diagnostic text.
pub fn setup_recovery_input_schema() -> Value {
    json!({
        "$schema": "https://json-schema.org/draft/2020-12/schema",
        "title": "report_setup_recovery v1 input",
        "type": "object",
        "additionalProperties": false,
        "required": ["version", "sequence", "phase", "state", "code"],
        "properties": {
            "version": { "const": SETUP_RECOVERY_REPORT_VERSION },
            "sequence": {
                "type": "integer",
                "minimum": 1,
                "maximum": MAX_SAFE_SETUP_RECOVERY_SEQUENCE,
            },
            "phase": { "enum": ["connect", "context"] },
            "state": {
                "enum": [
                    "working",
                    "needs_input",
                    "needs_decision",
                    "retryable_failure",
                    "reported_complete",
                ]
            },
            "code": {
                "enum": [
                    "in_progress",
                    "user_action_required",
                    "continue_or_stop",
                    "retryable",
                    "completion_reported",
                ]
            },
            "decision": {
                "type": "object",
                "additionalProperties": false,
                "required": ["kind", "choices"],
                "properties": {
                    "kind": { "const": "continue_or_stop" },
                    "choices": {
                        "type": "array",
                        "prefixItems": [{ "const": "continue" }, { "const": "stop" }],
                        "items": false,
                        "minItems": 2,
                        "maxItems": 2,
                    }
                }
            }
        },
        "allOf": [
            {
                "if": { "properties": { "state": { "const": "working" } }, "required": ["state"] },
                "then": { "properties": { "code": { "const": "in_progress" } }, "required": ["code"], "not": { "required": ["decision"] } }
            },
            {
                "if": { "properties": { "state": { "const": "needs_input" } }, "required": ["state"] },
                "then": { "properties": { "code": { "const": "user_action_required" } }, "required": ["code"], "not": { "required": ["decision"] } }
            },
            {
                "if": { "properties": { "state": { "const": "needs_decision" } }, "required": ["state"] },
                "then": { "properties": { "code": { "const": "continue_or_stop" } }, "required": ["code", "decision"] }
            },
            {
                "if": { "properties": { "state": { "const": "retryable_failure" } }, "required": ["state"] },
                "then": { "properties": { "code": { "const": "retryable" } }, "required": ["code"], "not": { "required": ["decision"] } }
            },
            {
                "if": { "properties": { "state": { "const": "reported_complete" } }, "required": ["state"] },
                "then": { "properties": { "code": { "const": "completion_reported" } }, "required": ["code"], "not": { "required": ["decision"] } }
            }
        ]
    })
}

/// The generated Setup-agent instruction for the v3 `genbi_session` discovery contract.
/// This is deliberately prose plus closed examples rather than a second tool configuration.
pub fn setup_recovery_instructions() -> &'static str {
    r#"## Setup recovery reporting (v1)

When the discovered `genbi_session` MCP server exposes `report_setup_recovery`, use that exact tool to report only an honest, redacted Setup lifecycle update. Do not create another transport or attempt to report through terminal output.

Every report has exactly these fields: `version: "1"`, a positive integer `sequence`, `phase`, `state`, and `code`. Use a strictly increasing `sequence` for each report in this native session. `phase` is exactly `connect` or `context`. The only valid state/code pairs are:

- `working` / `in_progress`
- `needs_input` / `user_action_required`
- `needs_decision` / `continue_or_stop`
- `retryable_failure` / `retryable`
- `reported_complete` / `completion_reported`

Only `needs_decision` carries `decision`, and it is exactly `{ "kind": "continue_or_stop", "choices": ["continue", "stop"] }`. Do not send `decision` for another state. Do not add fields or free text: never include identity, scope, project/session/vendor details, terminal output, paths, commands, prompts, credentials, tool arguments/results, or arbitrary options.

`reported_complete` is only this agent's report; the host independently validates completion. Never infer a report from terminal bytes, an exit status, or a tool result. If a truthful report cannot be made, omit it; silence is an honest host-lifecycle outcome, not a fabricated `needs_input` or decision.
"#
}

/// Native analysis keeps the structured step values as private orchestration
/// material while giving the person in the terminal a useful answer. Headless
/// dispatch remains the owner of the JSON result contract.
pub fn native_analysis_terminal_presentation_instructions() -> &'static str {
    r#"## Native terminal presentation

For the final response to the person in this interactive session, write concise conversational Markdown. Lead with the grounded answer, then include only a small table or bullets when they make the result clearer. Mention verification or query provenance only when it helps the person understand the conclusion.

The structured values used to coordinate steps are internal. Do not print a JSON result, render envelope, step envelope, orchestration field name, driver/subagent transcript, or tool request/response dump in the terminal. In particular, do not expose fields such as `warble_step`, `produces`, `query_result`, `repaired_result`, `columns`, `rows`, or `definition` as an internal payload. Programmatic and headless callers retain their structured JSON contracts; this rule changes only the native interactive presentation."#
}

/// Native analysis persists the final typed answer before turning it into a
/// conversational terminal response. The host owns the stored bytes and the
/// opaque reference; the driver must never invent a second payload shape.
pub fn native_answer_persistence_instructions() -> &'static str {
    r#"## Persist the final answer before presentation

After the final step has produced and validated the answer, but before writing any conversational Markdown, call `genbi_session.persist_answer` once with the exact typed answer envelope. Its `blocks` array contains exactly one `table` block with `columns` and `rows`, and zero or one `definition` block with `sql`, `source_tables`, and `filters`. Keep `verified: true`. Do not add a summary, chart, raw result, or any other block or representation.

```json
{
  "version": "1",
  "idempotency_key": "<stable retry key for this exact computed answer>",
  "envelope": {
    "blocks": [
      {"type": "table", "columns": ["..."], "rows": [["..."]]},
      {"type": "definition", "sql": "...", "source_tables": ["..."], "filters": ["..."]}
    ],
    "verified": true
  }
}
```

`genbi_session.query` runs a read-only SQL query through the host and returns its rows plus a `query_id` that answers may cite.

The `definition` block is optional; omit it rather than substituting another shape. `idempotency_key` is only for retrying this same already-computed persistence request, never caller-asserted provenance. On success, retain the host-returned `answer_ref` for a later dashboard save; keep `answer_ref`, the digest, and `persisted_at` internal rather than printing them in the terminal.

If persistence ultimately fails, do not recompute, rerun `answer_query`, generate SQL, or ask the user to supply the payload again. Still present the already-computed answer conversationally, followed by a concise honest warning that it was not retained and cannot later be saved by reference. Do not claim that it was retained or saved."#
}

/// Native analysis reuses the programmatic component IR for its data-work
/// instructions, but its final terminal response is deliberately not a JSON
/// transport. Keep this narrowly scoped to the authored final-output mandate:
/// the structured per-step contracts still coordinate the native workflow.
pub fn native_analysis_prompt_fragment(prompt: &str) -> String {
    prompt.replace(
        r#"FINAL message must be a single JSON object of the form
  `{"columns": ["model"], "rows": [["<model_name>"], ...]}` listing every model in the layer
  (one row per model), so coverage can be checked deterministically. You may add a short prose
  summary after the JSON."#,
        "final response should concisely summarize the semantic coverage in conversational Markdown.",
    )
}

/// Native Context Enrichment retains the proposal data model for the host, but
/// it is not a JSON transport for the person using a terminal.  Keep this
/// target-specific override at the emitter seam: the profile and its headless
/// consumers continue to own the canonical proposal contract.
pub fn native_context_enrichment_terminal_presentation_instructions() -> &'static str {
    r#"## Native context-enrichment presentation

For the final response to the person in this interactive session, write concise conversational Markdown, never JSON. Explain the proposed change in plain language, then cover its evidence and confidence, impact/risk, and destination in the semantic context. End with a clear choice: **accept**, **edit**, or **skip**.

In Grill mode, present only the one change currently awaiting a choice. If the proposal must pause because a prerequisite is missing or the destination is ambiguous, say that it is paused, explain the missing prerequisite or ambiguity and its risk, and ask what the person wants to clarify. Do not fabricate a draft to avoid a pause.

The proposal's structured values are internal host-coordination material. Do not print an `enrichment_proposal` JSON object, YAML payload, step envelope, orchestration field name, host/session/provider detail, tool request/response dump, or raw context material. In particular, do not expose labels such as `recommended_yaml`, `relative_sink`, `requires_approval`, `autopilot_eligible`, `project_revision`, or `produces` as a machine payload. Programmatic and headless callers retain their canonical structured proposal contracts; this rule changes only the native interactive presentation."#
}

/// Remove the profile's headless-only JSON-final requirement from an emitted
/// native Context Enrichment artifact.  Earlier prompt constraints remain the
/// source of truth for how a proposal is reasoned about; this only changes how
/// the completed proposal is presented in the terminal.
pub fn native_context_enrichment_prompt_fragment(prompt: &str) -> String {
    const HEADLESS_JSON_FINAL_MANDATE: &str = r#"Produce `enrichment_proposal`; approval, canonical hashes/digests, and application are deterministic
host responsibilities. Your FINAL message must be one JSON object only. Do not include prose or
Markdown fences. The top level is `{ "enrichment_proposal": { ... } }`; for Grill it contains the
supplied `project_revision`, exactly one operation with `relative_sink` and `recommended_yaml`,
confidence/evidence locators, `impact: "high"`, `requires_approval: true`,
`autopilot_eligible: false`, and one decision whose allowed responses are exactly
`["accept", "edit", "skip"]`."#;

    prompt.replace(
        HEADLESS_JSON_FINAL_MANDATE,
        "Prepare the same safe, read-only proposal internally, then give the person a concise conversational Markdown summary. Explain the proposed change, evidence and confidence, impact/risk, and destination without exposing the proposal's JSON or field names. In Grill mode, ask for exactly one of: accept, edit, or skip. If a prerequisite is missing or the destination is ambiguous, explain that the proposal is paused instead of inventing a draft.",
    )
}

/// The host owns persistence for a native dashboard session. This instruction
/// is emitted only when the server supplied the allowlisted session MCP.
pub fn native_dashboard_save_instructions() -> &'static str {
    r#"## Save a GenBI dashboard

When the user asks to create or save a dashboard from a prior answer, call `genbi_session.save_dashboard` exactly for that persistence action. When the retained `answer_ref` is still available in this conversation, prefer the exact reference:

```json
{
  "version": "1",
  "name": "<concise dashboard name>",
  "answer_ref": "<answer_ref returned by persist_answer>",
  "idempotency_key": "<stable key for this same dashboard request>"
}
```

If the user asks to save this or the most recent answer and the opaque `answer_ref` is no longer available in the conversation, use the host-owned session selector instead of refusing or recomputing:

```json
{
  "version": "1",
  "name": "<concise dashboard name>",
  "answer_selection": "latest",
  "idempotency_key": "<stable key for this same dashboard request>"
}
```

Provide exactly one of `answer_ref` or `answer_selection`. Reuse the same idempotency key when retrying the same request; choose a new key only for a materially new dashboard request. The host reuses the exact stored bytes for the selected answer and resolves `latest` only within this native session. Do not use `latest` when the user explicitly identifies an older answer whose reference is unavailable; explain that the older reference cannot be recovered. Do not rerun `answer_query`, generate SQL, repair SQL, or otherwise recompute; do not re-supply or reconstruct the payload. If the host reports that no persisted answer is available, say that reference saving is unavailable rather than claiming a save or attempting recomputation. The saved dashboard appears on the **GenBI Artifacts page**. Do not substitute a vendor-hosted Artifact feature, artifact URL, share URL, or any external vendor-hosted link for `genbi_session.save_dashboard`. Do not claim it was saved unless that tool succeeds."#
}

/// Shared vendor-neutral instruction for Setup's two-root launch contract.
/// The private materialization cwd contains only dispatcher artifacts; the
/// explicit server-owned environment value is the sole creation authority.
pub fn setup_bootstrap_authority_instructions() -> String {
    format!(
        r#"## Setup bootstrap write authority

`{NATIVE_SETUP_BOOTSTRAP_ROOT_ENV_VAR}` is set by the host to the only server-authorized project-creation root for this Setup session. Create or modify project files only below that exact root. Do not write into the current working directory (it contains private launch artifacts), do not change cwd to broaden authority, and do not write outside the configured bootstrap root. Treat a missing, relative, or unexpected value as a host-configuration failure: do not guess a path or accept one from the user.
"#
    )
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct SetupRecoveryReport {
    version: String,
    sequence: u64,
    phase: String,
    state: String,
    code: String,
    #[serde(default)]
    decision: Option<SetupRecoveryDecision>,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct SetupRecoveryDecision {
    kind: String,
    choices: Vec<String>,
}

/// Validate the producer-owned closed vocabulary without accepting or echoing a raw report.
/// Hosts additionally own per-session authorization and monotonic sequence fencing.
pub fn validate_setup_recovery_report(value: &Value) -> Result<(), DispatchError> {
    let report: SetupRecoveryReport = serde_json::from_value(value.clone())
        .map_err(|_| DispatchError("invalid report_setup_recovery v1 input".to_string()))?;
    if report.version != SETUP_RECOVERY_REPORT_VERSION
        || report.sequence == 0
        || report.sequence > MAX_SAFE_SETUP_RECOVERY_SEQUENCE
        || !matches!(report.phase.as_str(), "connect" | "context")
    {
        return Err(DispatchError(
            "invalid report_setup_recovery v1 input".to_string(),
        ));
    }
    let valid = match report.state.as_str() {
        "working" => report.code == "in_progress" && report.decision.is_none(),
        "needs_input" => report.code == "user_action_required" && report.decision.is_none(),
        "needs_decision" => {
            report.code == "continue_or_stop"
                && matches!(
                    report.decision,
                    Some(SetupRecoveryDecision { kind, choices })
                        if kind == "continue_or_stop"
                            && choices == ["continue", "stop"]
                )
        }
        "retryable_failure" => report.code == "retryable" && report.decision.is_none(),
        "reported_complete" => report.code == "completion_reported" && report.decision.is_none(),
        _ => false,
    };
    valid
        .then_some(())
        .ok_or_else(|| DispatchError("invalid report_setup_recovery v1 input".to_string()))
}

/// A server-derived launch scope. This is deliberately a small producer input rather
/// than session state: GenBI creates and authorizes it, Warble verifies its shape/canonical cwd
/// and carries its opaque binding identity into the launch artifact, and the future runtime owns
/// comparing it to its live binding generation/revision before spawning.
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NativeSessionScope {
    pub version: String,
    pub kind: String,
    pub scope_id: String,
    pub cwd: PathBuf,
    /// Which component the session starts in, and the first thing it is told.
    ///
    /// The caller declares this rather than the dispatcher inferring it: a profile can carry
    /// several materializable components (every shipped one does), so structure alone cannot
    /// choose an entry, and a built-in table of entry names would tie this crate to whichever
    /// product authored the profile. What the dispatcher still owns is the check that the
    /// declaration is true of the IR, and the authorship of the argv built from it.
    pub entry: NativeEntry,
    /// Setup alone carries this separately authorized project-creation root.
    /// `cwd` remains the native materialization root and must equal `--out`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub bootstrap_root: Option<PathBuf>,
    /// Optional to preserve the existing cross-vendor native scope contract. Native Codex
    /// materialization requires this exact server-resolved launcher and Python runtime closure;
    /// it is never derived from the browser, environment, or PATH by the dispatcher.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub wren_runtime: Option<NativeWrenRuntime>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub binding: Option<NativeBinding>,
}

/// How a native session enters its profile.
///
/// `Agent` pins the session to one materialized component. `Scope` starts it at the scope
/// document instead, leaving every emitted component a delegable subagent and letting the
/// session's own driver choose which one owns the request.
///
/// `Agent` is the default on purpose. A descriptor written before this field existed carries a
/// `verb` and no `kind`, and must keep meaning exactly what it meant. Scope entry is the wider of
/// the two, so it has to be asked for by name and can never be arrived at by omission.
///
/// How much wider is worth stating plainly, because nothing here narrows it. A pinned session IS
/// the named agent, so its tools are that agent's declared `tools:` list. A scope session is a
/// plain vendor session with no agent frontmatter, so it carries the vendor's default surface —
/// and the emitted `.claude/settings.json` pre-approves the UNION of every mounted component's
/// grants, because the vendor loads one settings file per session rather than one per agent. A
/// scope session can therefore use a tool that no single component was granted, without being
/// prompted for it. Per-agent `tools:` still bounds each delegated subagent; it is only the
/// top-level session that is unbounded. Closing that needs the scope session to carry a declared
/// tool surface of its own, which this does not attempt.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum NativeEntryKind {
    Agent,
    Scope,
}

/// The caller-declared session entry. `kind` picks the form; `verb` names the component to
/// materialize and is present for `Agent` only; `prompt` is the single positional first turn and
/// is required by both forms. All of it is validated against the compiled IR before any of it
/// reaches an argv — see `NativePurpose::validate_profile`.
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NativeEntry {
    #[serde(default = "NativeEntry::default_kind")]
    pub kind: NativeEntryKind,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub verb: Option<String>,
    pub prompt: String,
}

impl NativeEntry {
    fn default_kind() -> NativeEntryKind {
        NativeEntryKind::Agent
    }

    /// The pinned component verb, or `None` for a scope entry.
    pub fn pinned_verb(&self) -> Option<&str> {
        self.verb.as_deref()
    }
}

#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NativeBinding {
    pub project_identity: String,
    pub generation: String,
    pub revision: String,
}

/// A closed, server-derived Wren launcher chain. The Codex sandbox's `read` filesystem grant
/// includes execution permission, so its rendered entries must be the minimum transitive closure:
/// the PATH shim, the generated console-script launcher, the venv Python symlink, both runtime
/// library roots, and the server-pinned editable Wren source root that the venv's closed `.pth`
/// resolver names.
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NativeWrenRuntime {
    pub version: String,
    pub shim: PathBuf,
    pub launcher: PathBuf,
    pub venv_python: PathBuf,
    pub tool_root: PathBuf,
    pub site_packages: PathBuf,
    pub source_root: PathBuf,
    pub interpreter: PathBuf,
    pub interpreter_root: PathBuf,
}

struct CanonicalWrenRuntime {
    shim: PathBuf,
    shim_parent: PathBuf,
    launcher: PathBuf,
    venv_python: PathBuf,
    tool_bin: PathBuf,
    tool_pyvenv: PathBuf,
    site_packages: PathBuf,
    interpreter: PathBuf,
    interpreter_bin: PathBuf,
    interpreter_lib: PathBuf,
    source_root: PathBuf,
}

/// Exact server-derived connection material for the allowlisted native-session MCP server.
///
/// The credential is deliberately opaque to Warble. It is emitted only into the vendor discovery
/// mechanism that needs it: Claude's HTTP header or Codex's dedicated credential environment
/// variable. It never appears in the launch spec, ownership record, agent Markdown, argv, or a
/// prompt. The host resolves this credential to live session/vendor/project/revision/capability
/// state when the native client connects.
#[derive(Clone, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct NativeMcpDescriptor {
    version: String,
    url: String,
    credential: String,
}

impl std::fmt::Debug for NativeMcpDescriptor {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("NativeMcpDescriptor")
            .field("version", &self.version)
            .field("url", &self.url)
            .field("credential", &"[REDACTED]")
            .finish()
    }
}

impl NativeMcpDescriptor {
    pub fn from_file(path: &Path) -> Result<Self, DispatchError> {
        let raw = fs::read_to_string(path).map_err(|e| {
            DispatchError(format!(
                "read native MCP descriptor {}: {e}",
                path.display()
            ))
        })?;
        serde_json::from_str(&raw).map_err(|e| {
            DispatchError(format!(
                "parse native MCP descriptor {}: {e}",
                path.display()
            ))
        })
    }

    fn validate(&self) -> Result<(), DispatchError> {
        if self.version != NATIVE_MCP_DESCRIPTOR_VERSION {
            return Err(DispatchError(format!(
                "unsupported native MCP descriptor version '{}' (expected: {NATIVE_MCP_DESCRIPTOR_VERSION})",
                self.version
            )));
        }
        // The generic descriptor has no redirect, DNS, or vendor-specific transport
        // semantics. Parse its authority before accepting it: HTTPS may use any exact
        // parsed host, while HTTP is confined to a literal loopback host on this machine.
        // The credential, not the URL, carries the bounded session binding.
        let url = Url::parse(&self.url).map_err(|_| {
            DispatchError(
                "native MCP descriptor URL must be HTTPS or exact loopback HTTP".to_string(),
            )
        })?;
        let loopback = match url.host() {
            Some(Host::Domain("localhost")) => true,
            Some(Host::Ipv4(address)) => address.is_loopback(),
            Some(Host::Ipv6(address)) => address.is_loopback(),
            _ => false,
        };
        let allowed_scheme = url.scheme() == "https" || url.scheme() == "http" && loopback;
        // `Url::scheme()` provides the case-normalized scheme. Retain an explicit `://`
        // shape check so a parsed relative/scheme-only URL can never stand in for a
        // network authority.
        let has_network_authority = self
            .url
            .split_once(':')
            .is_some_and(|(_, after_scheme)| after_scheme.starts_with("//"));
        if self.url.len() > 2048
            || url.host().is_none()
            || !url.username().is_empty()
            || url.password().is_some()
            || url.query().is_some()
            || url.fragment().is_some()
            || self.url.chars().any(char::is_control)
            || self.url.chars().any(char::is_whitespace)
            || !has_network_authority
            || !allowed_scheme
        {
            return Err(DispatchError(
                "native MCP descriptor URL must be HTTPS or exact loopback HTTP".to_string(),
            ));
        }
        if self.credential.is_empty()
            || self.credential.len() > 4096
            || self.credential.chars().any(char::is_control)
            || self.credential.chars().any(char::is_whitespace)
        {
            return Err(DispatchError(
                "native MCP descriptor requires a bounded opaque credential".to_string(),
            ));
        }
        Ok(())
    }

    fn ownership_digest(&self) -> String {
        let bytes = serde_json::to_vec(&json!({
            "version": self.version,
            "url": self.url,
            "credential": self.credential,
        }))
        .expect("canonical native MCP descriptor serializes");
        format!("sha256:{:x}", Sha256::digest(bytes))
    }

    pub fn claude_discovery_config(&self) -> Result<String, DispatchError> {
        serde_json::to_string_pretty(&json!({
            "mcpServers": {
                NATIVE_MCP_SERVER_NAME: {
                    "type": "http",
                    "url": self.url,
                    "headers": { "Authorization": format!("Bearer {}", self.credential) },
                }
            }
        }))
        .map(|value| format!("{value}\n"))
        .map_err(|e| DispatchError(e.to_string()))
    }

    /// pi's `mcp.json` for the agent directory. The credential is referenced through pi's own
    /// `${VAR}` interpolation of the dedicated environment variable, so the file carries the URL
    /// and the variable name only — the same posture as Codex's `bearer_token_env_var`.
    /// `exposure: "direct"` makes pi declare the server's tools to the model as ordinary tools;
    /// the contract also leaves pi's code-execution mode off, so no MCP tool reaches the model
    /// through a generated script instead of a declared tool call.
    pub fn pi_discovery_config(&self) -> Result<String, DispatchError> {
        serde_json::to_string_pretty(&json!({
            "autoEnableCodemode": false,
            "mcpServers": {
                NATIVE_MCP_SERVER_NAME: {
                    "url": self.url,
                    "headers": { "Authorization": format!("Bearer ${{{NATIVE_MCP_CREDENTIAL_ENV_VAR}}}") },
                    "exposure": "direct",
                }
            }
        }))
        .map(|value| format!("{value}\n"))
        .map_err(|e| DispatchError(e.to_string()))
    }

    pub(crate) fn codex_host_discovery_config(&self, tools: &[&str]) -> String {
        format!("[mcp_servers.{NATIVE_MCP_SERVER_NAME}]\nurl = {}\nbearer_token_env_var = {}\nenabled_tools = {}\n",
            serde_json::to_string(&self.url).expect("URL"), serde_json::to_string(NATIVE_MCP_CREDENTIAL_ENV_VAR).expect("env"),
            serde_json::to_string(tools).expect("tools"))
    }

    pub fn codex_discovery_config(
        &self,
        enable_setup_recovery_tool: bool,
        enable_dashboard_save_tool: bool,
        enable_persist_answer_tool: bool,
    ) -> String {
        // JSON strings are valid TOML basic strings and give us a single established escaping
        // primitive for the server URL. Codex reads the credential only at native-process launch.
        let mut config = format!(
            "[mcp_servers.{NATIVE_MCP_SERVER_NAME}]\nurl = {}\nbearer_token_env_var = {}\n",
            serde_json::to_string(&self.url).expect("MCP URL serializes"),
            serde_json::to_string(NATIVE_MCP_CREDENTIAL_ENV_VAR)
                .expect("credential env name serializes"),
        );
        let mut enabled_tools = Vec::new();
        if enable_setup_recovery_tool {
            enabled_tools.push("report_setup_recovery");
        }
        if enable_dashboard_save_tool {
            enabled_tools.push(NATIVE_DASHBOARD_SAVE_TOOL);
        }
        if enable_persist_answer_tool {
            enabled_tools.push(NATIVE_PERSIST_ANSWER_TOOL);
            enabled_tools.push(NATIVE_QUERY_TOOL);
        }
        if !enabled_tools.is_empty() {
            config.push_str(&format!(
                "enabled_tools = {}\n",
                serde_json::to_string(&enabled_tools).expect("MCP tool names serialize"),
            ));
        }
        config
    }
}

impl NativeSessionScope {
    pub fn from_file(path: &Path) -> Result<Self, DispatchError> {
        let raw = fs::read_to_string(path).map_err(|e| {
            DispatchError(format!("read native session scope {}: {e}", path.display()))
        })?;
        serde_json::from_str(&raw).map_err(|e| {
            DispatchError(format!(
                "parse native session scope {}: {e}",
                path.display()
            ))
        })
    }

    fn validate_preflight(&self, purpose: NativePurpose) -> Result<(), DispatchError> {
        if self.version != NATIVE_SCOPE_VERSION {
            return Err(DispatchError(format!(
                "unsupported native session scope version '{}' (expected: {NATIVE_SCOPE_VERSION})",
                self.version
            )));
        }
        if self.kind != purpose.scope_kind() {
            return Err(DispatchError(format!(
                "native session scope kind '{}' does not match purpose '{}' ({})",
                self.kind,
                purpose.as_str(),
                purpose.scope_kind()
            )));
        }
        if self.scope_id.trim().is_empty() {
            return Err(DispatchError(
                "native session scope requires a non-empty opaque scope_id".to_string(),
            ));
        }
        match (self.entry.kind, self.entry.verb.as_deref()) {
            (NativeEntryKind::Agent, None) => {
                return Err(DispatchError(
                    "native session scope entry kind 'agent' requires the component verb it pins"
                        .to_string(),
                ));
            }
            (NativeEntryKind::Agent, Some(verb)) if verb.trim().is_empty() => {
                return Err(DispatchError(
                    "native session scope requires a non-empty entry verb".to_string(),
                ));
            }
            // A verb alongside `scope` is a contradiction, not a harmless extra: the caller has
            // asked for two different sessions at once, and guessing which it meant would pick a
            // privilege level on its behalf.
            (NativeEntryKind::Scope, Some(_)) => {
                return Err(DispatchError(
                    "native session scope entry kind 'scope' must not name a verb: scope entry \
                     starts at the scope document and pins no component"
                        .to_string(),
                ));
            }
            (NativeEntryKind::Agent, Some(_)) | (NativeEntryKind::Scope, None) => {}
        }
        if self.entry.prompt.trim().is_empty() {
            return Err(DispatchError(
                "native session scope requires a non-empty entry prompt".to_string(),
            ));
        }
        // Both entry fields become positional argv elements, and neither CLI is invoked behind a
        // `--` sentinel. A dash-leading value would therefore be parsed as an option rather than
        // as the prompt or the agent name -- silently changing what was launched, since both
        // vendors expose permission-affecting flags. This surface did not exist while the prompt
        // was a constant compiled into this dispatcher; it does now that a caller supplies it, so
        // the shape has to be refused here, which is the last gate before argv is assembled.
        for (field, value) in self
            .entry
            .verb
            .iter()
            .map(|verb| ("verb", verb))
            .chain(std::iter::once(("prompt", &self.entry.prompt)))
        {
            if value.trim_start().starts_with('-') {
                return Err(DispatchError(format!(
                    "native session scope entry {field} must not begin with '-': a dash-leading \
                     value would be parsed as a vendor CLI option instead of argv content"
                )));
            }
        }
        if !self.cwd.is_absolute() {
            return Err(DispatchError(
                "native session scope cwd must be an absolute server-derived path".to_string(),
            ));
        }
        match (&self.kind[..], &self.binding, &self.bootstrap_root) {
            ("bootstrap", None, Some(bootstrap_root)) => {
                if !bootstrap_root.is_absolute() {
                    return Err(DispatchError(
                        "bootstrap native session scope root must be an absolute server-derived path".to_string(),
                    ));
                }
                let canonical_bootstrap_root = fs::canonicalize(bootstrap_root).map_err(|e| {
                    DispatchError(format!(
                        "canonicalize bootstrap native session scope root {}: {e}",
                        bootstrap_root.display()
                    ))
                })?;
                if *bootstrap_root != canonical_bootstrap_root {
                    return Err(DispatchError(
                        "bootstrap native session scope root must already be canonical".to_string(),
                    ));
                }
                Ok(())
            }
            ("bootstrap", None, None) => Err(DispatchError(
                "bootstrap native session scope requires a separately authorized bootstrap_root"
                    .to_string(),
            )),
            ("bootstrap", Some(_), _) => Err(DispatchError(
                "bootstrap native session scope must not carry a bound-project identity".to_string(),
            )),
            ("bound_project", Some(binding), None)
                if !binding.project_identity.trim().is_empty()
                    && !binding.generation.trim().is_empty()
                    && !binding.revision.trim().is_empty() =>
            {
                Ok(())
            }
            ("bound_project", _, Some(_)) => Err(DispatchError(
                "bound_project native session scope must not carry a bootstrap_root".to_string(),
            )),
            ("bound_project", _, None) => Err(DispatchError(
                "bound_project native session scope requires non-empty project_identity, generation, and revision"
                    .to_string(),
            )),
            _ => unreachable!("purpose mapping has a closed scope vocabulary"),
        }
    }

    fn validate(&self, purpose: NativePurpose, root: &Path) -> Result<(), DispatchError> {
        self.validate_preflight(purpose)?;
        let scope_cwd = match fs::canonicalize(&self.cwd) {
            Ok(canonical) => {
                if self.cwd != canonical {
                    return Err(DispatchError(
                        "native session scope cwd must already be canonical".to_string(),
                    ));
                }
                canonical
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound && self.cwd == root => {
                root.to_path_buf()
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                return Err(DispatchError(format!(
                    "native session scope cwd {} does not match canonical output root {}",
                    self.cwd.display(),
                    root.display()
                )))
            }
            Err(error) => {
                return Err(DispatchError(format!(
                    "canonicalize native session scope cwd {}: {error}",
                    self.cwd.display()
                )))
            }
        };
        if scope_cwd != root {
            return Err(DispatchError(format!(
                "native session scope cwd {} does not match canonical output root {}",
                scope_cwd.display(),
                root.display()
            )));
        }
        if let Some(bootstrap_root) = &self.bootstrap_root {
            let canonical_bootstrap_root = fs::canonicalize(bootstrap_root).map_err(|e| {
                DispatchError(format!(
                    "canonicalize bootstrap native session scope root {}: {e}",
                    bootstrap_root.display()
                ))
            })?;
            if canonical_bootstrap_root == root {
                return Err(DispatchError(
                    "bootstrap native session scope root must be distinct from the canonical output root"
                        .to_string(),
                ));
            }
        }
        Ok(())
    }

    fn ownership_digest(&self, canonical_cwd: &Path) -> String {
        // This is a collision identifier only, not a signature or provenance proof. GenBI owns
        // invocation authorization; the digest merely keeps opaque caller values out of Markdown
        // ownership markers while ensuring a changed descriptor cannot reuse an old artifact set.
        let canonical = json!({
            "version": self.version,
            "kind": self.kind,
            "scope_id": self.scope_id,
            "cwd": canonical_cwd,
            "bootstrap_root": self.bootstrap_root.as_ref().map(|root| {
                fs::canonicalize(root).expect("validated bootstrap root remains canonicalizable")
            }),
            "wren_runtime": self.wren_runtime,
            "binding": self.binding,
            // The entry is part of the descriptor, so it has to be part of the identifier the
            // docstring above promises. Omitting it let an entry-only change reuse a previous
            // artifact set's marker -- the exact reuse this digest exists to prevent.
            "entry": self.entry,
        });
        let bytes = serde_json::to_vec(&canonical).expect("canonical native scope serializes");
        format!("sha256:{:x}", Sha256::digest(bytes))
    }

    fn launch_value(&self) -> serde_json::Value {
        json!({
            "kind": self.kind,
            "scope_id": self.scope_id,
            "bootstrap_root": self.bootstrap_root.as_ref().map(|root| {
                fs::canonicalize(root).expect("validated bootstrap root remains canonicalizable")
            }),
            "binding": self.binding,
        })
    }

    /// Render the complete, server-selected Codex permission profile. It deliberately has no
    /// caller-selected filesystem, executable, PATH, browser, credential, or network input.
    pub fn codex_permission_profile(&self) -> Result<String, DispatchError> {
        let mut config = self
            .wren_runtime
            .as_ref()
            .ok_or_else(|| {
                DispatchError(
                    "native Codex purpose requires a server-derived wren_runtime closure"
                        .to_string(),
                )
            })?
            .codex_permission_profile()?;
        config.push_str(&format!(
            "\n[permissions.{CODEX_WREN_PERMISSION_PROFILE}.filesystem.\":workspace_roots\"]\n"
        ));
        if self.kind == "bootstrap" {
            let bootstrap_root = self.bootstrap_root.as_ref().ok_or_else(|| {
                DispatchError(
                    "bootstrap native session scope requires a separately authorized bootstrap_root"
                        .to_string(),
                )
            })?;
            let canonical_bootstrap_root = fs::canonicalize(bootstrap_root).map_err(|_| {
                DispatchError("bootstrap native session scope root is unavailable".to_string())
            })?;
            if *bootstrap_root != canonical_bootstrap_root {
                return Err(DispatchError(
                    "bootstrap native session scope root must already be canonical".to_string(),
                ));
            }
            // The native cwd is a private artifact root: readable for discovery, never writable
            // by Setup. Only the separately authorized bootstrap root receives write authority.
            config.push_str("\".\" = \"read\"\n");
            config.push_str(&format!(
                "{} = \"write\"\n",
                serde_json::to_string(&canonical_bootstrap_root.to_string_lossy())
                    .expect("bootstrap root serializes")
            ));
        } else {
            config.push_str("\".\" = \"write\"\n");
        }
        Ok(config)
    }

    /// Claude's permission grammar carries the same server-selected Setup
    /// root as its only Edit/Write scope. The string is generated only after
    /// NativeSessionScope::validate has canonicalized the sealed descriptor.
    pub fn claude_setup_write_permissions(&self) -> Result<[String; 2], DispatchError> {
        let bootstrap_root = self.bootstrap_root.as_ref().ok_or_else(|| {
            DispatchError(
                "bootstrap native session scope requires a separately authorized bootstrap_root"
                    .to_string(),
            )
        })?;
        let canonical_bootstrap_root = fs::canonicalize(bootstrap_root).map_err(|_| {
            DispatchError("bootstrap native session scope root is unavailable".to_string())
        })?;
        if self.kind != "bootstrap" || *bootstrap_root != canonical_bootstrap_root {
            return Err(DispatchError(
                "bootstrap native session scope root is incompatible".to_string(),
            ));
        }
        let recursive = canonical_bootstrap_root
            .join("**")
            .to_string_lossy()
            .to_string();
        Ok([format!("Edit({recursive})"), format!("Write({recursive})")])
    }
}

impl NativeWrenRuntime {
    fn validate(&self) -> Result<CanonicalWrenRuntime, DispatchError> {
        if self.version != NATIVE_WREN_RUNTIME_VERSION {
            return Err(DispatchError(format!(
                "unsupported native Wren runtime version '{}' (expected: {NATIVE_WREN_RUNTIME_VERSION})",
                self.version
            )));
        }

        let tool_root = canonical_directory(&self.tool_root, "tool_root")?;
        let interpreter_root = canonical_directory(&self.interpreter_root, "interpreter_root")?;
        let site_packages = canonical_directory(&self.site_packages, "site_packages")?;
        let source_root = canonical_directory(&self.source_root, "source_root")?;
        let launcher = canonical_regular_file(&self.launcher, "launcher")?;
        let interpreter = canonical_regular_file(&self.interpreter, "interpreter")?;
        let shim = canonical_regular_file(&self.shim, "shim")?;
        let venv_python = canonical_regular_file(&self.venv_python, "venv_python")?;
        let shim_parent = canonical_directory(
            self.shim.parent().ok_or_else(|| {
                DispatchError("native Wren runtime shim must have a parent directory".to_string())
            })?,
            "shim parent",
        )?;

        let tool_bin = tool_root.join("bin");
        let interpreter_bin = interpreter_root.join("bin");
        let tool_pyvenv = tool_root.join("pyvenv.cfg");
        let tool_lib = tool_root.join("lib");
        let interpreter_lib = interpreter_root.join("lib");
        for (path, label) in [
            (&tool_bin, "tool_root/bin"),
            (&interpreter_bin, "interpreter_root/bin"),
            (&interpreter_lib, "interpreter_root/lib"),
        ] {
            if !path.is_dir() {
                return Err(DispatchError(format!(
                    "native Wren runtime {label} must be an existing directory"
                )));
            }
        }
        if !tool_pyvenv.is_file() {
            return Err(DispatchError(
                "native Wren runtime tool_root/pyvenv.cfg must be an existing regular file"
                    .to_string(),
            ));
        }
        if !site_packages.starts_with(&tool_lib) {
            return Err(DispatchError(
                "native Wren runtime site_packages must be inside tool_root/lib".to_string(),
            ));
        }
        let editable_pth = site_packages.join("_editable_impl_wrenai.pth");
        let editable_source = fs::read_to_string(&editable_pth)
            .ok()
            .and_then(|contents| {
                let paths = contents
                    .lines()
                    .map(str::trim)
                    .filter(|line| !line.is_empty())
                    .collect::<Vec<_>>();
                let [path] = paths.as_slice() else {
                    return None;
                };
                Path::new(path).is_absolute().then(|| PathBuf::from(path))
            })
            .and_then(|path| fs::canonicalize(path).ok());
        if editable_source.as_deref() != Some(source_root.as_path())
            || !source_root.join("wren/__init__.py").is_file()
        {
            return Err(DispatchError(
                "native Wren runtime editable source must exactly match the server-approved .pth closure"
                    .to_string(),
            ));
        }

        // The launcher and venv-Python names are fixed contract points rather than flexible
        // command inputs. A server must resolve the real chain before handing it to Warble.
        if self.launcher != tool_bin.join("wren") || self.venv_python != tool_bin.join("python") {
            return Err(DispatchError(
                "native Wren runtime launcher chain must use tool_root/bin/wren and tool_root/bin/python"
                    .to_string(),
            ));
        }
        if !fs::symlink_metadata(&self.shim)
            .map_err(|e| {
                DispatchError(format!(
                    "inspect native Wren runtime shim {}: {e}",
                    self.shim.display()
                ))
            })?
            .file_type()
            .is_symlink()
        {
            return Err(DispatchError(
                "native Wren runtime shim must be a symlink to the server-approved launcher"
                    .to_string(),
            ));
        }
        if !fs::symlink_metadata(&self.venv_python)
            .map_err(|e| {
                DispatchError(format!(
                    "inspect native Wren runtime venv_python {}: {e}",
                    self.venv_python.display()
                ))
            })?
            .file_type()
            .is_symlink()
        {
            return Err(DispatchError(
                "native Wren runtime tool_root/bin/python must be a symlink to the server-approved interpreter"
                    .to_string(),
            ));
        }
        if shim != launcher
            || venv_python != interpreter
            || !interpreter.starts_with(&interpreter_bin)
        {
            return Err(DispatchError(
                "native Wren runtime shim/interpreter chain does not resolve to the declared closure"
                    .to_string(),
            ));
        }
        if !is_executable(&launcher) || !is_executable(&interpreter) {
            return Err(DispatchError(
                "native Wren runtime launcher and interpreter must be executable regular files"
                    .to_string(),
            ));
        }
        let shebang = fs::read_to_string(&launcher)
            .map_err(|e| {
                DispatchError(format!(
                    "read native Wren launcher {}: {e}",
                    launcher.display()
                ))
            })?
            .lines()
            .next()
            .and_then(|line| line.strip_prefix("#!"))
            .filter(|path| !path.is_empty() && !path.contains(char::is_whitespace))
            .map(PathBuf::from)
            .ok_or_else(|| {
                DispatchError(
                    "native Wren runtime launcher must start with an absolute interpreter shebang"
                        .to_string(),
                )
            })?;
        if !shebang.is_absolute() || shebang != self.venv_python {
            return Err(DispatchError(
                "native Wren runtime launcher shebang must exactly name tool_root/bin/python"
                    .to_string(),
            ));
        }

        Ok(CanonicalWrenRuntime {
            shim: self.shim.clone(),
            shim_parent,
            launcher: self.launcher.clone(),
            venv_python: self.venv_python.clone(),
            tool_bin,
            tool_pyvenv,
            site_packages: self.site_packages.clone(),
            interpreter: self.interpreter.clone(),
            interpreter_bin,
            interpreter_lib,
            source_root: self.source_root.clone(),
        })
    }

    fn codex_permission_profile(&self) -> Result<String, DispatchError> {
        let paths = self.validate()?;
        let entry = |path: &Path| {
            format!(
                "{} = \"read\"\n",
                serde_json::to_string(&path.to_string_lossy()).expect("runtime path serializes")
            )
        };
        let mut config = format!(
            "# Server-owned native Wren runtime closure. `read` is Codex's read-and-execute\n# filesystem grant; no PATH, browser, credential, or caller filesystem input is accepted.\ndefault_permissions = \"{CODEX_WREN_PERMISSION_PROFILE}\"\n\n[permissions.{CODEX_WREN_PERMISSION_PROFILE}]\ndescription = \"Warble native session workspace plus exact Wren runtime closure\"\n\n[permissions.{CODEX_WREN_PERMISSION_PROFILE}.filesystem]\n\":minimal\" = \"read\"\n"
        );
        for path in [
            &paths.shim,
            &paths.shim_parent,
            &paths.tool_bin,
            &paths.launcher,
            &paths.venv_python,
            &paths.tool_pyvenv,
            &paths.site_packages,
            &paths.interpreter,
            &paths.interpreter_bin,
            &paths.interpreter_lib,
            &paths.source_root,
        ] {
            config.push_str(&entry(path));
        }
        Ok(config)
    }
}

fn canonical_directory(path: &Path, label: &str) -> Result<PathBuf, DispatchError> {
    if !path.is_absolute() {
        return Err(DispatchError(format!(
            "native Wren runtime {label} must be an absolute server-derived path"
        )));
    }
    let canonical = fs::canonicalize(path).map_err(|e| {
        DispatchError(format!(
            "canonicalize native Wren runtime {label} {}: {e}",
            path.display()
        ))
    })?;
    canonical.is_dir().then_some(canonical).ok_or_else(|| {
        DispatchError(format!(
            "native Wren runtime {label} must be an existing directory"
        ))
    })
}

fn canonical_regular_file(path: &Path, label: &str) -> Result<PathBuf, DispatchError> {
    if !path.is_absolute() {
        return Err(DispatchError(format!(
            "native Wren runtime {label} must be an absolute server-derived path"
        )));
    }
    let canonical = fs::canonicalize(path).map_err(|e| {
        DispatchError(format!(
            "canonicalize native Wren runtime {label} {}: {e}",
            path.display()
        ))
    })?;
    canonical.is_file().then_some(canonical).ok_or_else(|| {
        DispatchError(format!(
            "native Wren runtime {label} must be an existing regular file"
        ))
    })
}

#[cfg(unix)]
fn is_executable(path: &Path) -> bool {
    use std::os::unix::fs::PermissionsExt;

    fs::metadata(path).is_ok_and(|metadata| metadata.permissions().mode() & 0o111 != 0)
}

#[cfg(not(unix))]
fn is_executable(_path: &Path) -> bool {
    false
}

/// The only product purposes that may opt into the native Sessions launch contract.
///
/// This enum is intentionally closed: a caller cannot use the launch artifact to select an
/// arbitrary profile, agent, cwd, or command. The GenBI runtime chooses a purpose and vendor,
/// then materializes the corresponding Warble profile in a server-owned scope.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NativePurpose {
    Analysis,
    Setup,
    ContextEnrichment,
}

impl NativePurpose {
    pub fn parse(value: &str) -> Option<Self> {
        match value {
            "analysis" => Some(Self::Analysis),
            "setup" => Some(Self::Setup),
            "context_enrichment" => Some(Self::ContextEnrichment),
            _ => None,
        }
    }

    pub fn as_str(self) -> &'static str {
        match self {
            Self::Analysis => "analysis",
            Self::Setup => "setup",
            Self::ContextEnrichment => "context_enrichment",
        }
    }

    fn scope_kind(self) -> &'static str {
        match self {
            Self::Setup => "bootstrap",
            Self::Analysis | Self::ContextEnrichment => "bound_project",
        }
    }

    pub fn codex_skill(self) -> &'static str {
        match self {
            Self::Analysis => "genbi-analysis",
            Self::Setup => "genbi-setup",
            Self::ContextEnrichment => "genbi-enrich-context",
        }
    }

    pub fn codex_description(self) -> &'static str {
        match self {
            Self::Analysis => "Analyze the server-bound semantic project using the GenBI analysis behavior.",
            Self::Setup => "Connect a source and build a semantic context only in the server-created bootstrap scope.",
            Self::ContextEnrichment => "Inspect a pinned project and draft read-only enrichment proposals; never apply an enrichment.",
        }
    }

    /// The one initial interactive prompt for this closed native purpose.
    ///
    /// This is deliberately data owned by the producer's purpose enum, rather
    /// than a browser value or terminal write. Both vendor CLIs accept one
    /// positional prompt while staying interactive, so it is passed as a
    /// single argv element without invoking a shell.
    /// Checks the caller's declared entry against the compiled IR.
    ///
    /// This replaced an allowlist of profile names. The allowlist could only answer "is this the
    /// profile I was built expecting", which says nothing about whether the thing about to be
    /// launched is safe to launch, and it tied this crate to one product's vocabulary. The
    /// question worth asking is structural, and it is asked of every profile equally: does the
    /// declared verb name exactly one component here, and is that component something a native
    /// interactive session may materialize at all?
    ///
    /// The caller therefore cannot widen what runs by naming a profile differently — only by
    /// naming a component that already satisfies these conditions.
    pub fn validate_profile(self, ir: &WarbleIr, entry: &NativeEntry) -> Result<(), DispatchError> {
        // This runs BEFORE `NativeSessionScope::validate`, so a malformed pair is reported here
        // rather than assumed away: the descriptor's own field checks have not happened yet.
        match (entry.kind, entry.verb.as_deref()) {
            (NativeEntryKind::Agent, Some(verb)) => self.validate_pinned_entry(ir, verb),
            (NativeEntryKind::Agent, None) => Err(DispatchError(format!(
                "native purpose '{}' entry kind 'agent' requires the component verb it pins",
                self.as_str()
            ))),
            (NativeEntryKind::Scope, None) => self.validate_scope_entry(ir),
            (NativeEntryKind::Scope, Some(verb)) => Err(DispatchError(format!(
                "native purpose '{}' declared scope entry but also named the verb '{verb}': scope entry must not name a verb",
                self.as_str()
            ))),
        }
    }

    /// Whether `node` can be materialized as a native interactive agent in its own right.
    ///
    /// Factored out of the pinned check so scope entry can ask the same question of a whole
    /// profile instead of one named verb. The conditions are unchanged: a component is eligible
    /// exactly when pinning to it would have been allowed.
    fn is_native_entry_eligible(node: &crate::ir::ComponentNode) -> bool {
        node.id == node.verb
            && node.realization_kind == crate::ir::RealizationKind::Skill
            && node.trigger.kind == crate::ir::TriggerKind::OneShot
            && node.effect.outcome.kind == crate::ir::OutcomeKind::None
            && !node
                .required_capabilities
                .iter()
                .any(|capability| capability == "enrichment_apply:deterministic")
    }

    fn validate_pinned_entry(self, ir: &WarbleIr, verb: &str) -> Result<(), DispatchError> {
        let entries = ir
            .components
            .iter()
            .filter(|node| node.verb == verb)
            .collect::<Vec<_>>();
        let [component] = entries.as_slice() else {
            return Err(DispatchError(format!(
                "native purpose '{}' requires exactly one component with the declared entry verb '{}', found {}",
                self.as_str(),
                verb,
                entries.len()
            )));
        };
        if !Self::is_native_entry_eligible(component) {
            return Err(DispatchError(format!(
                "native purpose '{}' entry '{}' is not materializable as a native interactive agent: it must be a one-shot skill with no outcome and must not require enrichment_apply:deterministic",
                self.as_str(),
                verb
            )));
        }
        Ok(())
    }

    /// A scope entry names no component, so what is checked instead is that the scope is worth
    /// entering: more than one component the session could actually delegate to.
    ///
    /// One eligible component makes pinning strictly better — same behavior, but the top-level
    /// session carries that agent's declared `tools:` instead of the vendor default surface. Zero
    /// makes the scope inert: agents on disk, none of them launchable, and a session with nothing
    /// to delegate to would quietly do the work itself outside any component's guardrails.
    fn validate_scope_entry(self, ir: &WarbleIr) -> Result<(), DispatchError> {
        let eligible = ir
            .components
            .iter()
            .filter(|node| Self::is_native_entry_eligible(node))
            .count();
        if eligible < 2 {
            return Err(DispatchError(format!(
                "native purpose '{}' declared scope entry, but this profile materializes {} native-eligible component(s): scope entry needs at least 2 so the session has something to choose between; pin the entry instead",
                self.as_str(),
                eligible
            )));
        }
        Ok(())
    }
}

/// What the launch spec needs from the pi emitter beyond the shared native inputs: the single
/// session model, the exact tool allowlist, and where the authored system prompt and the
/// server-owned agent directory live under the materialization root.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PiLaunch {
    pub provider: String,
    pub model: String,
    /// Fully qualified pi tool names (`mcp__<server>__<tool>`), already filtered to the host MCP
    /// tools this component may use.
    pub tools: Vec<String>,
    /// Relative to the materialization root.
    pub system_prompt: PathBuf,
    /// Relative to the materialization root.
    pub agent_dir: PathBuf,
}

pub struct InteractiveOutput {
    native_host: Option<crate::native_host::NativeHost>,
    pi: Option<PiLaunch>,
    pub root: PathBuf,
    pub launch_path: PathBuf,
    pub handoff_path: PathBuf,
    ownership_path: PathBuf,
    owned_paths: Vec<String>,
    marker: String,
    target: String,
    executable: String,
    purpose: Option<NativePurpose>,
    native_scope: Option<NativeSessionScope>,
    native_mcp: Option<NativeMcpDescriptor>,
    /// The materialized profile's id. A scope entry names it in place of a component, since the
    /// thing being entered is the profile itself.
    profile: String,
}

enum ResolvedInteractiveOutput {
    Existing(PathBuf),
    Missing(PathBuf),
}

fn resolve_interactive_output(path: &Path) -> Result<ResolvedInteractiveOutput, DispatchError> {
    let mut resolved = if path.is_absolute() {
        PathBuf::new()
    } else {
        let current = std::env::current_dir()
            .map_err(|e| DispatchError(format!("resolve current directory: {e}")))?;
        fs::canonicalize(&current).map_err(|e| {
            DispatchError(format!(
                "canonicalize current directory {}: {e}",
                current.display()
            ))
        })?
    };
    let mut missing = false;
    for component in path.components() {
        match component {
            Component::CurDir => {}
            Component::ParentDir => {
                if missing {
                    return Err(DispatchError(format!(
                        "interactive output cannot traverse '..' after a missing component: {}",
                        path.display()
                    )));
                }
                let parent = resolved.parent().unwrap_or(resolved.as_path());
                resolved = fs::canonicalize(parent).map_err(|e| {
                    DispatchError(format!(
                        "resolve interactive output parent {}: {e}",
                        parent.display()
                    ))
                })?;
            }
            Component::Prefix(_) | Component::RootDir => {
                resolved.push(component.as_os_str());
            }
            Component::Normal(name) if missing => resolved.push(name),
            Component::Normal(name) => {
                if !resolved.as_os_str().is_empty() {
                    let metadata = fs::metadata(&resolved).map_err(|e| {
                        DispatchError(format!(
                            "inspect interactive output ancestor {}: {e}",
                            resolved.display()
                        ))
                    })?;
                    if !metadata.is_dir() {
                        return Err(DispatchError(format!(
                            "interactive dispatch output has a non-directory ancestor: {}",
                            resolved.display()
                        )));
                    }
                }
                let next = resolved.join(name);
                match fs::symlink_metadata(&next) {
                    Ok(_) => {
                        resolved = fs::canonicalize(&next).map_err(|e| {
                            DispatchError(format!(
                                "resolve interactive output component {}: {e}",
                                next.display()
                            ))
                        })?;
                    }
                    Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                        resolved = next;
                        missing = true;
                    }
                    Err(error) => {
                        return Err(DispatchError(format!(
                            "inspect interactive output component {}: {error}",
                            next.display()
                        )))
                    }
                }
            }
        }
    }

    if missing {
        Ok(ResolvedInteractiveOutput::Missing(resolved))
    } else if fs::metadata(&resolved)
        .map_err(|e| {
            DispatchError(format!(
                "inspect interactive dispatch output {}: {e}",
                path.display()
            ))
        })?
        .is_dir()
    {
        Ok(ResolvedInteractiveOutput::Existing(resolved))
    } else {
        Err(DispatchError(format!(
            "interactive dispatch output exists but is not a directory: {}",
            path.display()
        )))
    }
}

#[allow(clippy::too_many_arguments)]
pub(crate) fn prepare_interactive_output_with_host(
    out_dir: &Path,
    target: &str,
    executable: &str,
    profile_signature: &str,
    profile: &str,
    owned_paths: &[PathBuf],
    purpose: Option<NativePurpose>,
    native_scope: Option<NativeSessionScope>,
    native_mcp: Option<NativeMcpDescriptor>,
    native_host: Option<crate::native_host::NativeHost>,
    pi: Option<PiLaunch>,
) -> Result<InteractiveOutput, DispatchError> {
    match (purpose, native_scope.as_ref()) {
        (Some(purpose), Some(scope)) => scope.validate_preflight(purpose)?,
        (Some(_), None) => {
            return Err(DispatchError(
                "native Sessions purpose requires a server-derived --native-scope descriptor"
                    .to_string(),
            ))
        }
        (None, Some(_)) => {
            return Err(DispatchError(
                "--native-scope requires a native Sessions --purpose".to_string(),
            ))
        }
        (None, None) => {}
    }
    match (purpose, native_mcp.as_ref()) {
        (Some(_), Some(descriptor)) => descriptor.validate()?,
        (Some(_), None) | (None, None) => {}
        (None, Some(_)) => {
            return Err(DispatchError(
                "--native-mcp requires a native Sessions --purpose".to_string(),
            ))
        }
    }
    let (prospective_root, create_root) = match resolve_interactive_output(out_dir)? {
        ResolvedInteractiveOutput::Existing(root) => (root, false),
        ResolvedInteractiveOutput::Missing(root) => (root, true),
    };
    if let (Some(purpose), Some(scope)) = (purpose, native_scope.as_ref()) {
        scope.validate(purpose, &prospective_root)?;
    }
    if create_root {
        fs::create_dir_all(out_dir).map_err(|e| {
            DispatchError(format!(
                "create interactive output directory {}: {e}",
                out_dir.display()
            ))
        })?;
    }
    let root = fs::canonicalize(out_dir).map_err(|e| {
        DispatchError(format!(
            "canonicalize interactive output {}: {e}",
            out_dir.display()
        ))
    })?;
    if root != prospective_root {
        return Err(DispatchError(format!(
            "interactive output canonical path changed during creation: expected {}, got {}",
            prospective_root.display(),
            root.display()
        )));
    }
    if let (Some(purpose), Some(scope)) = (purpose, native_scope.as_ref()) {
        scope.validate(purpose, &root)?;
    }
    let handoff_path = root.join("RUN.md");
    let launch_path = root.join(".warble/interactive-launch.json");
    ensure_inside(&root, &handoff_path)?;
    ensure_inside(&root, &launch_path)?;
    ensure_safe_path(&root, Path::new(".warble/interactive-launch.json"))?;
    ensure_safe_path(&root, Path::new(".warble/interactive-ownership.json"))?;
    let marker = format!(
        "<!-- warble-interactive-artifact target={target} profile={profile_signature}{}{}{} -->",
        native_scope
            .as_ref()
            .map(|scope| format!(" scope_digest={}", scope.ownership_digest(&root)))
            .unwrap_or_default(),
        native_mcp
            .as_ref()
            .map(|descriptor| format!(" mcp_digest={}", descriptor.ownership_digest()))
            .unwrap_or_default(),
        native_host
            .as_ref()
            .map(|host| format!(
                " host_digest={}",
                host.document()["host_plan_sha256"]
                    .as_str()
                    .expect("digest")
            ))
            .unwrap_or_default(),
    );

    let mut planned = owned_paths
        .iter()
        .map(|path| path.to_string_lossy().to_string())
        .collect::<Vec<_>>();
    planned.sort();
    planned.dedup();
    let ownership_path = root.join(".warble/interactive-ownership.json");
    ensure_inside(&root, &ownership_path)?;
    for relative in owned_paths {
        ensure_safe_path(&root, relative)?;
    }
    let any_existing = planned.iter().any(|relative| root.join(relative).exists())
        || launch_path.exists()
        || ownership_path.exists();
    if any_existing {
        let existing = fs::read_to_string(&ownership_path).map_err(|_| {
            DispatchError(format!(
                "refusing to overwrite existing interactive artifacts: {} is missing",
                ownership_path.display()
            ))
        })?;
        verify_ownership(&existing, &marker, &planned, &root)?;
    }
    for relative in owned_paths {
        let path = root.join(relative);
        if path.exists()
            && matches!(
                path.file_name().and_then(|name| name.to_str()),
                Some("RUN.md" | "AGENTS.md" | "SKILL.md" | "CLAUDE.md")
            )
        {
            let contents = fs::read_to_string(&path).map_err(|e| {
                DispatchError(format!(
                    "read existing interactive artifact {}: {e}",
                    path.display()
                ))
            })?;
            if !contents.contains(&marker) {
                return Err(DispatchError(format!(
                    "refusing to overwrite user-owned interactive artifact {}; it lacks the expected Warble ownership marker",
                    path.display()
                )));
            }
        }
    }
    if launch_path.exists() {
        let existing = fs::read_to_string(&launch_path).map_err(|e| {
            DispatchError(format!(
                "read existing launch spec {}: {e}",
                launch_path.display()
            ))
        })?;
        if existing
            != render_launch_spec(LaunchSpecInputs {
                target,
                executable,
                profile,
                root: &root,
                handoff: &handoff_path,
                purpose,
                native_scope: native_scope.as_ref(),
                native_mcp: native_mcp.as_ref(),
                native_host: native_host.as_ref(),
                pi: pi.as_ref(),
            })?
        {
            return Err(DispatchError(format!(
                "refusing to overwrite user-owned or mismatched launch spec {}",
                launch_path.display()
            )));
        }
    }
    Ok(InteractiveOutput {
        native_host,
        pi,
        root,
        launch_path,
        handoff_path,
        ownership_path,
        owned_paths: planned,
        marker,
        target: target.into(),
        executable: executable.into(),
        purpose,
        native_scope,
        native_mcp,
        profile: profile.into(),
    })
}

impl InteractiveOutput {
    pub fn marker(&self) -> &str {
        &self.marker
    }
    pub fn write_launch_spec(&self) -> Result<(), DispatchError> {
        let parent = self.launch_path.parent().expect(".warble parent");
        fs::create_dir_all(parent)
            .map_err(|e| DispatchError(format!("create {}: {e}", parent.display())))?;
        fs::write(
            &self.launch_path,
            render_launch_spec(LaunchSpecInputs {
                target: &self.target,
                executable: &self.executable,
                profile: &self.profile,
                root: &self.root,
                handoff: &self.handoff_path,
                purpose: self.purpose,
                native_scope: self.native_scope.as_ref(),
                native_mcp: self.native_mcp.as_ref(),
                native_host: self.native_host.as_ref(),
                pi: self.pi.as_ref(),
            })?,
        )
        .map_err(|e| {
            DispatchError(format!(
                "write launch spec {}: {e}",
                self.launch_path.display()
            ))
        })
    }
    pub fn write_ownership(&self) -> Result<(), DispatchError> {
        if let Some(host) = &self.native_host {
            use std::io::Write;
            fs::create_dir_all(self.root.join(".warble"))
                .map_err(|e| DispatchError(e.to_string()))?;
            let path = self.root.join(".warble/component-plans.json");
            let mut options = fs::OpenOptions::new();
            options.write(true).create(true).truncate(true);
            #[cfg(unix)]
            {
                use std::os::unix::fs::OpenOptionsExt;
                options.mode(0o600);
            }
            let mut file = options
                .open(&path)
                .map_err(|e| DispatchError(e.to_string()))?;
            #[cfg(unix)]
            {
                use std::os::unix::fs::PermissionsExt;
                // Creation mode does not narrow an existing file on repeat dispatch.
                file.set_permissions(fs::Permissions::from_mode(0o600))
                    .map_err(|e| DispatchError(e.to_string()))?;
            }
            file.write_all(host.document().to_string().as_bytes())
                .and_then(|()| file.sync_all())
                .map_err(|e| DispatchError(e.to_string()))?;
        }
        let parent = self.ownership_path.parent().expect(".warble parent");
        fs::create_dir_all(parent)
            .map_err(|e| DispatchError(format!("create {}: {e}", parent.display())))?;
        fs::write(
            &self.ownership_path,
            ownership_document(&self.marker, &self.owned_paths, &self.root)?,
        )
        .map_err(|e| {
            DispatchError(format!(
                "write ownership record {}: {e}",
                self.ownership_path.display()
            ))
        })
    }
}

fn ensure_inside(root: &Path, path: &Path) -> Result<(), DispatchError> {
    if !path.starts_with(root) {
        return Err(DispatchError(format!(
            "interactive artifact path escapes canonical output root: {}",
            path.display()
        )));
    }
    Ok(())
}

/// Reject unsafe existing path components before any output write. A lexical `starts_with` check
/// is not a containment proof when `.claude`, `.agents`, or `.warble` can redirect writes
/// elsewhere, and a regular-file parent would make creation order-dependent.
fn ensure_safe_path(root: &Path, relative: &Path) -> Result<(), DispatchError> {
    if relative.is_absolute()
        || relative
            .components()
            .any(|component| matches!(component, std::path::Component::ParentDir))
    {
        return Err(DispatchError(format!(
            "interactive artifact path escapes canonical output root: {}",
            relative.display()
        )));
    }
    let components = relative.components().collect::<Vec<_>>();
    let mut cursor = root.to_path_buf();
    for (index, component) in components.iter().enumerate() {
        cursor.push(component.as_os_str());
        if let Ok(metadata) = fs::symlink_metadata(&cursor) {
            if metadata.file_type().is_symlink() {
                return Err(DispatchError(format!(
                    "refusing interactive artifact path with symlink component: {}",
                    cursor.display()
                )));
            }
            if index + 1 < components.len() && !metadata.file_type().is_dir() {
                return Err(DispatchError(format!(
                    "refusing interactive artifact path with non-directory ancestor component: {}",
                    cursor.display()
                )));
            }
        }
    }
    Ok(())
}

/// Everything the launch spec is rendered from, grouped so the renderer keeps one parameter as it
/// grows. These are all identity or location values the emitter already holds; nothing here is
/// derived, so the struct is a bundle rather than a stage.
struct LaunchSpecInputs<'a> {
    target: &'a str,
    executable: &'a str,
    profile: &'a str,
    root: &'a Path,
    handoff: &'a Path,
    purpose: Option<NativePurpose>,
    native_scope: Option<&'a NativeSessionScope>,
    native_mcp: Option<&'a NativeMcpDescriptor>,
    native_host: Option<&'a crate::native_host::NativeHost>,
    pi: Option<&'a PiLaunch>,
}

/// pi's closed argv. Every discovery mechanism is switched off by name, the system prompt is the
/// authored file, the tools are an explicit allowlist (never `--no-tools`, which also drops MCP
/// tools), and the model is the one the dispatcher was given. JSON mode carries the first prompt
/// positionally after `--`; RPC mode takes it over stdin, so its argv ends at the model.
fn pi_argv(launch: &PiLaunch, root: &Path, mode: &str, prompt: Option<&str>) -> Value {
    let mut argv = vec![
        "--mode".to_string(),
        mode.to_string(),
        "--no-session".to_string(),
        "--no-extensions".to_string(),
        "--no-skills".to_string(),
        "--no-prompt-templates".to_string(),
        "--no-themes".to_string(),
        "--no-context-files".to_string(),
        "--no-approve".to_string(),
        "-e".to_string(),
        "builtin:mcp".to_string(),
        "--tools".to_string(),
        launch.tools.join(","),
        "--system-prompt".to_string(),
        root.join(&launch.system_prompt)
            .to_string_lossy()
            .to_string(),
        "--provider".to_string(),
        launch.provider.clone(),
        "--model".to_string(),
        launch.model.clone(),
    ];
    if let Some(prompt) = prompt {
        argv.push("--".to_string());
        argv.push(prompt.to_string());
    }
    json!(argv)
}

fn render_launch_spec(inputs: LaunchSpecInputs<'_>) -> Result<String, DispatchError> {
    let LaunchSpecInputs {
        target,
        executable,
        profile,
        root,
        handoff,
        purpose,
        native_scope,
        native_mcp,
        native_host,
        pi,
    } = inputs;
    ensure_inside(root, handoff)?;
    let pi_launch = if target == crate::targets::TargetId::PiInteractive.as_str() {
        if native_mcp.is_none() {
            return Err(DispatchError(
                "pi:interactive has no launch contract without --native-mcp".to_string(),
            ));
        }
        Some(pi.ok_or_else(|| {
            DispatchError(
                "pi:interactive launch spec requires its model and tool inputs".to_string(),
            )
        })?)
    } else {
        None
    };
    // This is intentionally the entire schema: no command string, prompt/model material, auth,
    // provider state, or session identity can be represented here.
    let mut document = match (purpose, native_mcp) {
        // v1 remains byte-for-byte schema-compatible for its enrichment consumer.
        (None, None) => json!({
            "version": LAUNCH_SPEC_VERSION,
            "target": target,
            "executable": executable,
            "argv": [],
            "cwd": root,
            "artifact_root": root,
            "handoff_path": handoff,
        }),
        (Some(purpose), None) => json!({
            "version": NATIVE_SESSION_LAUNCH_SPEC_VERSION,
            "target": target,
            "purpose": purpose.as_str(),
            "executable": executable,
            // Claude selects an agent with `--agent` when the caller pinned one, and otherwise
            // starts at the scope document with no selection flag at all — `--agent` is this
            // dispatcher's choice, never a requirement of the CLI. Codex loads the named skill
            // from its repository-scoped discovery artifacts and has no scope form. The argv is
            // still authored here and nowhere else; what it says comes from the caller's declared
            // entry, already checked against the IR by `validate_profile`.
            "argv": if target == "claude-code:interactive" {
                match native_scope.expect("v3 native scope preflighted").entry.pinned_verb() {
                    Some(verb) => json!(["--agent", verb]),
                    None => json!([]),
                }
            } else {
                json!([])
            },
            "agent": if target == "claude-code:interactive" {
                match native_scope.expect("v3 native scope preflighted").entry.pinned_verb() {
                    Some(verb) => json!({ "kind": "claude_agent", "name": verb }),
                    None => json!({ "kind": "claude_scope", "name": profile }),
                }
            } else {
                json!({ "kind": "codex_skill", "name": purpose.codex_skill() })
            },
            "scope": native_scope.expect("v3 native scope preflighted").launch_value(),
            "cwd": root,
            "artifact_root": root,
            "handoff_path": handoff,
        }),
        (Some(purpose), Some(_)) => {
            let entry = &native_scope.expect("v3 native scope preflighted").entry;
            let mut document = json!({
                "version": NATIVE_SESSION_MCP_LAUNCH_SPEC_VERSION,
                "target": target,
                "purpose": purpose.as_str(),
                "executable": executable,
            // Both interactive CLIs accept one positional prompt. Keeping it
            // in this closed argv contract makes the first turn exactly-once
            // without a shell or post-spawn PTY input injection.
                "argv": match pi_launch {
                    // pi over RPC takes the first prompt as a stdin command (see `pi.first_prompt`
                    // below); the one-shot JSON-mode argv with the prompt is `pi.json_argv`.
                    Some(launch) => pi_argv(launch, root, "rpc", None),
                    None => if target == "claude-code:interactive" {
                        match entry.pinned_verb() {
                            Some(verb) => json!(["--agent", verb, entry.prompt]),
                            // Scope entry drops only the selection flag. The prompt stays positional,
                            // so the first turn is still delivered exactly once by the same contract.
                            None => json!([entry.prompt]),
                        }
                    } else {
                        json!([entry.prompt])
                    },
                },
                "agent": if pi_launch.is_some() {
                    json!({ "kind": "pi_system_prompt", "name": entry.pinned_verb().expect("pi entry is pinned") })
                } else if target == "claude-code:interactive" {
                    match entry.pinned_verb() {
                        Some(verb) => json!({ "kind": "claude_agent", "name": verb }),
                        None => json!({ "kind": "claude_scope", "name": profile }),
                    }
                } else {
                    json!({ "kind": "codex_skill", "name": purpose.codex_skill() })
                },
                "mcp": {
                    "server_name": NATIVE_MCP_SERVER_NAME,
                    "credential_env_var": NATIVE_MCP_CREDENTIAL_ENV_VAR,
                },
                "cwd": root,
                "artifact_root": root,
                "handoff_path": handoff,
            });
            if purpose == NativePurpose::Setup {
                document
                    .as_object_mut()
                    .expect("launch spec is an object")
                    .insert(
                        "bootstrap_root".to_string(),
                        serde_json::to_value(
                            native_scope
                                .expect("v4 Setup native scope preflighted")
                                .bootstrap_root
                                .as_ref()
                                .map(|root| {
                                    fs::canonicalize(root)
                                        .expect("validated bootstrap root remains canonicalizable")
                                }),
                        )
                        .expect("bootstrap root serializes"),
                    );
            }
            document
        }
        (None, Some(_)) => {
            return Err(DispatchError(
                "--native-mcp requires a native Sessions --purpose".to_string(),
            ))
        }
    };
    if let Some(launch) = pi_launch {
        let entry = &native_scope.expect("v4 native scope preflighted").entry;
        document["pi"] = json!({
            "minimum_version": crate::pi::PI_MINIMUM_VERSION,
            "agent_dir": root.join(&launch.agent_dir),
            "system_prompt": root.join(&launch.system_prompt),
            "tools": launch.tools,
            "first_prompt": entry.prompt,
            "json_argv": pi_argv(launch, root, "json", Some(&entry.prompt)),
            "required_env": {
                "PI_CODING_AGENT_DIR": root.join(&launch.agent_dir),
                "PI_OFFLINE": "1",
                "PI_SKIP_VERSION_CHECK": "1",
                "PI_TELEMETRY": "0",
            },
        });
    }
    if let Some(host) = native_host {
        document["version"] = json!("5");
        document["component_host"] = host.launch_value();
    }
    serde_json::to_string_pretty(&document)
        .map(|v| format!("{v}\n"))
        .map_err(|e| DispatchError(e.to_string()))
}

fn ownership_document(
    marker: &str,
    paths: &[String],
    root: &Path,
) -> Result<String, DispatchError> {
    let artifacts = paths
        .iter()
        .map(|path| {
            let bytes = fs::read(root.join(path))
                .map_err(|e| DispatchError(format!("read owned artifact {path}: {e}")))?;
            Ok(json!({ "path": path, "sha256": format!("{:x}", Sha256::digest(bytes)) }))
        })
        .collect::<Result<Vec<_>, DispatchError>>()?;
    serde_json::to_string_pretty(&json!({ "marker": marker, "artifacts": artifacts }))
        .map(|v| format!("{v}\n"))
        .map_err(|e| DispatchError(e.to_string()))
}

fn verify_ownership(
    existing: &str,
    marker: &str,
    paths: &[String],
    root: &Path,
) -> Result<(), DispatchError> {
    let value: serde_json::Value = serde_json::from_str(existing).map_err(|_| {
        DispatchError("refusing to overwrite malformed interactive ownership record".to_string())
    })?;
    if value.get("marker").and_then(serde_json::Value::as_str) != Some(marker) {
        return Err(DispatchError("refusing to overwrite artifacts not owned by this exact Warble interactive emission plan".to_string()));
    }
    let artifacts = value
        .get("artifacts")
        .and_then(serde_json::Value::as_array)
        .ok_or_else(|| {
            DispatchError(
                "refusing to overwrite ownership record without per-artifact digests".to_string(),
            )
        })?;
    if artifacts.len() != paths.len() {
        return Err(DispatchError(
            "refusing to overwrite ownership record with a different artifact plan".to_string(),
        ));
    }
    for (artifact, path) in artifacts.iter().zip(paths) {
        if artifact.get("path").and_then(serde_json::Value::as_str) != Some(path.as_str()) {
            return Err(DispatchError(
                "refusing to overwrite ownership record with a different artifact plan".to_string(),
            ));
        }
        let bytes = fs::read(root.join(path)).map_err(|_| {
            DispatchError(format!(
                "refusing to overwrite missing owned artifact {path}"
            ))
        })?;
        let digest = format!("{:x}", Sha256::digest(bytes));
        if artifact.get("sha256").and_then(serde_json::Value::as_str) != Some(digest.as_str()) {
            return Err(DispatchError(format!(
                "refusing to overwrite modified owned artifact {path}"
            )));
        }
    }
    Ok(())
}

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

    #[test]
    fn native_mcp_descriptor_debug_redacts_the_opaque_credential() {
        let descriptor = NativeMcpDescriptor {
            version: "1".to_string(),
            url: "https://mcp.example.test/native".to_string(),
            credential: "sentinel-opaque-native-mcp-credential".to_string(),
        };

        let rendered = format!("{descriptor:?}");
        assert!(rendered.contains("NativeMcpDescriptor"));
        assert!(rendered.contains("version: \"1\""));
        assert!(rendered.contains("url: \"https://mcp.example.test/native\""));
        assert!(rendered.contains("credential: \"[REDACTED]\""));
        assert!(!rendered.contains("sentinel-opaque-native-mcp-credential"));
    }
}