supercode_harness/

sessions_control.rs

//! Controlled-tier conversations (Domain 11, concept 5) — `new`, `reset`,
//! `archive`, `delete` over one uniform door.
//!
//! Charter (`docs/plans/orchestration-domain-11-2026-09-02.md` §0.4):
//! **every mutation here is the harness's OWN verb.** supercode never writes
//! another harness's session store by hand; it runs the harness's CLI, calls
//! the harness's HTTP API, or types the harness's slash command into a LIVE
//! driven session, then re-reads the row the harness's own store now holds.
//!
//! The doors, per harness, at the pinned versions:
//!
//! | harness | new | reset | archive | delete |
//! |---|---|---|---|---|
//! | claude-code | refused → `runtimes.start` | refused | refused | refused |
//! | codex | refused → `runtimes.start` | refused | `codex archive <id>` | `codex delete <id>` |
//! | opencode | refused → `runtimes.start` | refused | `PATCH /session/<id>` | `DELETE /session/<id>` |
//! | hermes | refused (gateway-only) | `/reset` in a live session | refused | `hermes sessions delete <id>` |
//! | openclaw | `/new` in a live session | `/reset` in a live session | refused | refused |
//! | orchestrator | its daemon's operator door | its daemon's operator door | refused | refused |
//! | supercode | refused → `runtimes.start` | refused | own store | own store |
//!
//! Three rules the whole tier inherits from [`crate::jobs_control`]:
//!
//! 1. **The harness's answer is the answer.** After the door reports success
//!    the conversation is re-read through the ORCH-6 discovery loader and
//!    returned. A delete that leaves the row behind, or an archive the store
//!    did not record, is a FAILURE — never a silent success.
//! 2. **The door is narrated.** Every outcome carries `ran`: the exact argv,
//!    HTTP request line, slash command, or store call that was performed,
//!    with any credential rendered as `<redacted>`.
//! 3. **A verb the harness has no door for is refused**
//!    ([`SessionControlError::Unsupported`] → `UnsupportedAction`), with the
//!    reason and the door that DOES exist, never a silent no-op.
//!
//! ## Why some cells are refused at the pin
//!
//! * **`hermes sessions archive`** exists but is a BULK filter verb
//!   (`--older-than`, `--title`, `--cwd`, …) with no per-session selector, so
//!   a uniform "archive THIS conversation" cannot be expressed through it.
//!   `hermes sessions delete <id>` is per-session and IS used.
//!   (Pinned help fixture: `crates/harness/src/parity/fixtures/hermes-help.txt`,
//!   section `$ hermes sessions --help`.)
//! * **OpenClaw** registers only `sessions list | cleanup | tail |
//!   export-trajectory | compact` at v2026.7.1-2 — no `archive`, no `delete`.
//! * **Claude Code** publishes no conversation lifecycle verb at all: its
//!   sessions expire on a retention window it owns.
//! * **The orchestrator** has no archive and no delete BY MODEL: a binding
//!   (`docs/ORCHESTRATOR-IR.md` §2.5) ends, and the transcript belongs to the
//!   worker harness the binding addresses. `new` and `reset` DO exist — they
//!   are the two chat commands its reducer applies to a binding (§4.5) — and
//!   ORC-13 opened the door that reaches them from outside a chat: the
//!   daemon's operator socket, or the package's own CLI when it is down
//!   ([`crate::orchestrator_door`]). The orchestrator's conversation is a
//!   BINDING, so it is named by its SURFACE key
//!   (`platform|chat_type|chat_id|thread_id|participant_id`), never by a
//!   worker session id — `--surface`, not `--session`.
//! * **`new` / `reset` on the file-store harnesses** is not a missing verb —
//!   it is a DIFFERENT door that already exists: `harness.v1.runtimes.start`.
//!   Refusing while naming it keeps one way to do one thing.
//!
//! ## A slash command is only sent when the harness's door advertises it
//!
//! Both gateway harnesses expose `/new` and `/reset` IN CHAT, but the door
//! supercode drives is each one's ACP adapter, and the two adapters do not
//! carry the same set. Read from the harnesses themselves, 2026-09-03:
//!
//! * **OpenClaw** (`openclaw@2026.7.1-2`, `dist/commands-*.js`
//!   `BASE_AVAILABLE_COMMANDS`) advertises BOTH `new` ("Reset the session
//!   (/reset)") and `reset` on its ACP door. Both are supported.
//! * **Hermes** (`acp_adapter/server.py` `_SLASH_COMMANDS` /
//!   `_handle_slash_command`) advertises `reset` and NOT `new` — `/new` lives
//!   only in `gateway/slash_commands.py`, and the adapter comments that an
//!   unrecognized command "falls through to the LLM (the user may have typed
//!   `/something` as prose)". `sessions.new` on hermes is therefore refused:
//!   sending it would put the literal text `/new` in front of the model, which
//!   is the silent no-op this tier exists to prevent.
//!
//! The two doors also differ in EFFECT, and neither is re-interpreted here:
//! hermes's ACP `/reset` clears the conversation and keeps the session row,
//! while the gateway's `/reset` rotates the session id. supercode drives the
//! door it can reach and reports what that door did.

use std::path::{Path, PathBuf};
use std::process::Command;

use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::{DiscoveryQuery, HarnessHomes, HarnessId};

/// Environment variable overriding the `codex` executable (tests).
pub const CODEX_BIN_ENV: &str = "SUPERCODE_CODEX_BIN";
/// Environment variable overriding the `hermes` executable (tests).
pub const HERMES_BIN_ENV: &str = "SUPERCODE_HERMES_BIN";

/// One uniform conversation-lifecycle verb.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionVerb {
    /// Open a fresh conversation on the same surface.
    New,
    /// Clear the conversation while keeping the surface.
    Reset,
    /// Soft-hide the conversation, keeping its transcript.
    Archive,
    /// Permanently remove the conversation.
    Delete,
}

impl SessionVerb {
    /// Uniform spelling used in the RPC method and in outcomes.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::New => "new",
            Self::Reset => "reset",
            Self::Archive => "archive",
            Self::Delete => "delete",
        }
    }

    /// The RPC method this verb is spelled as.
    pub const fn method(self) -> &'static str {
        match self {
            Self::New => "harness.v1.sessions.new",
            Self::Reset => "harness.v1.sessions.reset",
            Self::Archive => "harness.v1.sessions.archive",
            Self::Delete => "harness.v1.sessions.delete",
        }
    }

    /// Whether the verb names an existing conversation.
    const fn needs_session(self) -> bool {
        !matches!(self, Self::New)
    }
}

/// The door one `(harness, verb)` pair goes through.
///
/// The service needs this BEFORE it acts: a [`SessionDoor::Live`] verb is
/// typed into an already-open runtime connection the service owns, while
/// every other door is self-contained in this module.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SessionDoor {
    /// The harness's own CLI verb, run as a subprocess.
    Cli,
    /// The harness's own HTTP API.
    Http,
    /// The harness's own slash command, typed into a LIVE driven session.
    /// Carries the exact command text (`/new`, `/reset`).
    Live(&'static str),
    /// supercode's own session store (the Domain 5 verb).
    Store,
    /// The orchestrator daemon's own operator door (ORC-13): its local socket
    /// while the daemon is up, its package's CLI when it is down. Both land in
    /// the reducer that owns bindings, and in the `save()` that owns the
    /// folder.
    Daemon,
}

/// One mutating request, in the uniform Domain 11 vocabulary.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionMutation {
    /// Harness that owns the conversation.
    pub harness: String,
    /// Harness-native conversation id, or supercode session name. Required
    /// for every verb but `new`.
    #[serde(default)]
    pub session: Option<String>,
    /// Working directory the new conversation belongs to (`new`).
    #[serde(default)]
    pub cwd: Option<PathBuf>,
    /// Live runtime connection id, for the slash-command doors.
    #[serde(default)]
    pub connection: Option<String>,
    /// Already-running OpenCode server this conversation lives on. Without
    /// it the HTTP door is refused rather than guessing an endpoint.
    #[serde(default)]
    pub base_url: Option<String>,
    /// Bearer credential for the OpenCode server, when it requires one. Never
    /// narrated.
    #[serde(default)]
    pub bearer: Option<String>,
    /// Hermes profile name — a profile IS a full `HERMES_HOME`. For the
    /// orchestrator it is the profile FOLDER the binding belongs to.
    #[serde(default)]
    pub profile: Option<String>,
    /// ORC-13: the surface key a conversation is bound to, in the IR's own
    /// rendering (`platform|chat_type|chat_id|thread_id|participant_id`).
    /// This is how the orchestrator's conversations are named — its binding
    /// has no id of its own, only the surface it holds.
    #[serde(default)]
    pub surface: Option<String>,
    /// Storage roots, so an isolated home is addressed the same way the read
    /// side addresses it.
    #[serde(default)]
    pub homes: HarnessHomes,
}

/// What one mutation did, with the conversation re-read afterwards.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SessionMutationOutcome {
    /// Harness whose door was used.
    pub harness: String,
    /// Uniform verb that was asked for.
    pub verb: String,
    /// The exact door that was used, credentials redacted.
    pub ran: String,
    /// Conversation the verb acted on.
    pub session: String,
    /// The conversation as the harness's own store reports it AFTER the verb.
    /// Absent for `delete`, and for a `new`/`reset` whose fresh conversation
    /// the harness has not committed to its store yet.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub row: Option<Value>,
    /// `true` on a successful `archive`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub archived: Option<bool>,
    /// `true` on a successful `delete`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub deleted: Option<bool>,
}

/// Why a mutation could not be performed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SessionControlError {
    /// The harness has no door for what was asked (refused, never faked).
    Unsupported(String),
    /// The request itself is incoherent.
    Invalid(String),
    /// The harness's door ran and failed; the message carries its own error.
    Failed(String),
}

impl std::fmt::Display for SessionControlError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Unsupported(message) | Self::Invalid(message) | Self::Failed(message) => {
                formatter.write_str(message)
            }
        }
    }
}

impl std::error::Error for SessionControlError {}

type Result<T> = std::result::Result<T, SessionControlError>;

/// Harnesses whose conversations supercode can mutate through at least one of
/// their own doors. Strictly narrower than the set it can READ.
pub const CONTROLLED_SESSION_HARNESSES: &[&str] = &[
    HarnessId::CODEX,
    HarnessId::OPENCODE,
    HarnessId::HERMES,
    HarnessId::OPENCLAW,
    HarnessId::ORCHESTRATOR,
    HarnessId::SUPERCODE,
];

/// Every harness the compiled registry carries.
///
/// Spelled out rather than read back from [`crate::harness_support_registry`]:
/// this door table is one of the INPUTS that registry is built from (the
/// `conversation` concept block reads [`controlled_methods`]), so looking the
/// registry up from here would recurse forever. A unit test below pins the two
/// lists together.
const REGISTERED_HARNESSES: &[&str] = &[
    HarnessId::CLAUDE_CODE,
    HarnessId::CODEX,
    HarnessId::PI,
    HarnessId::OPENCODE,
    HarnessId::GROK,
    HarnessId::GEMINI,
    HarnessId::GOOSE,
    HarnessId::HERMES,
    HarnessId::OPENCLAW,
    HarnessId::ORCHESTRATOR,
    HarnessId::SUPERCODE,
];

/// Whether `harness` publishes a door for at least one conversation verb.
pub fn supports_session_control(harness: &str) -> bool {
    CONTROLLED_SESSION_HARNESSES.contains(&harness)
}

/// Every uniform verb, in declaration order.
pub const ALL_SESSION_VERBS: [SessionVerb; 4] = [
    SessionVerb::New,
    SessionVerb::Reset,
    SessionVerb::Archive,
    SessionVerb::Delete,
];

/// Every uniform verb `harness` can actually perform, in declaration order.
/// Empty for a harness with no door at all.
pub fn controlled_verbs(harness: &str) -> Vec<&'static str> {
    ALL_SESSION_VERBS
        .into_iter()
        .filter(|verb| door(harness, *verb).is_ok())
        .map(SessionVerb::as_str)
        .collect()
}

/// The RPC methods `harness` actually answers for the controlled tier. This is
/// what the registry block advertises, so a method can never appear in the
/// descriptor without a door behind it.
pub fn controlled_methods(harness: &str) -> Vec<&'static str> {
    ALL_SESSION_VERBS
        .into_iter()
        .filter(|verb| door(harness, *verb).is_ok())
        .map(SessionVerb::method)
        .collect()
}

/// Which door this `(harness, verb)` pair goes through, or WHY the harness
/// refuses it.
///
/// This is the single table the whole tier is derived from: the service, the
/// registry block, and the refusal messages all read it, so a door can never
/// be advertised in one place and missing in another.
pub fn door(harness: &str, verb: SessionVerb) -> Result<SessionDoor> {
    match (harness, verb) {
        // --- codex: two native CLI verbs, no `new`/`reset` concept ---------
        (HarnessId::CODEX, SessionVerb::Archive | SessionVerb::Delete) => Ok(SessionDoor::Cli),
        // --- opencode: its own HTTP session API ---------------------------
        (HarnessId::OPENCODE, SessionVerb::Archive | SessionVerb::Delete) => Ok(SessionDoor::Http),
        // --- the live slash-command doors, each checked against what that
        //     harness's ACP door ACTUALLY advertises (module header) --------
        (HarnessId::OPENCLAW, SessionVerb::New) => Ok(SessionDoor::Live("/new")),
        (HarnessId::HERMES | HarnessId::OPENCLAW, SessionVerb::Reset) => {
            Ok(SessionDoor::Live("/reset"))
        }
        (HarnessId::HERMES, SessionVerb::New) => Err(SessionControlError::Unsupported(
            "hermes's ACP door advertises help, model, tools, context, reset, compress, steer, \
             queue and version; `/new` is a GATEWAY command \
             (`gateway/slash_commands.py::_handle_reset_command`) and hermes's ACP adapter sends \
             any UNRECOGNIZED `/word` to the model as prose. Typing `/new` there would be a \
             silent no-op dressed as a chat turn, so supercode refuses. `sessions.reset` IS \
             advertised on that door and is supported"
                .into(),
        )),
        (HarnessId::HERMES, SessionVerb::Delete) => Ok(SessionDoor::Cli),
        (HarnessId::HERMES, SessionVerb::Archive) => Err(SessionControlError::Unsupported(
            "hermes 0.21.0 registers `hermes sessions archive`, but it is a BULK filter verb \
             (--older-than / --title / --cwd / ...) with no per-session selector, so archiving \
             ONE conversation cannot be expressed through it. `sessions.delete` is per-session \
             and is supported"
                .into(),
        )),
        // --- openclaw: no lifecycle verb at the pin -----------------------
        (HarnessId::OPENCLAW, SessionVerb::Archive | SessionVerb::Delete) => {
            Err(SessionControlError::Unsupported(format!(
                "openclaw v2026.7.1-2 registers `sessions list | cleanup | tail | \
                 export-trajectory | compact` and no `archive` or `delete`, so supercode refuses \
                 `sessions.{}` rather than inventing store-maintenance semantics for it",
                verb.as_str()
            )))
        }
        // --- supercode's own store ----------------------------------------
        (HarnessId::SUPERCODE, SessionVerb::Archive | SessionVerb::Delete) => {
            Ok(SessionDoor::Store)
        }
        // --- claude-code: no lifecycle verb at all -------------------------
        (HarnessId::CLAUDE_CODE, SessionVerb::Archive | SessionVerb::Delete) => {
            Err(SessionControlError::Unsupported(format!(
                "claude-code publishes no conversation lifecycle verb: its sessions are removed \
                 by a RETENTION WINDOW the harness itself owns (`cleanupPeriodDays`), so \
                 supercode refuses `sessions.{}` rather than deleting files behind the \
                 harness's back",
                verb.as_str()
            )))
        }
        // --- the orchestrator: its lifecycle verbs are the DAEMON's -------
        (HarnessId::ORCHESTRATOR, SessionVerb::New | SessionVerb::Reset) => Ok(SessionDoor::Daemon),
        (HarnessId::ORCHESTRATOR, verb) => Err(SessionControlError::Unsupported(format!(
            "the orchestrator's conversations are BINDINGS its daemon holds \
             (`docs/ORCHESTRATOR-IR.md` §2.5): a binding is never archived or deleted — it \
             ENDS, and the transcript belongs to the WORKER harness it addresses, which is \
             where `sessions.{}` is performed. `sessions.new` and `sessions.reset` end a \
             binding through the daemon's own operator door and are supported",
            verb.as_str()
        ))),
        (other, verb) if !REGISTERED_HARNESSES.contains(&other) => {
            Err(SessionControlError::Unsupported(format!(
                "`{other}` is not a registered harness, so `sessions.{}` has no door to go \
                 through",
                verb.as_str()
            )))
        }
        // --- `new` / `reset` where the door is `runtimes.start` -----------
        (_, SessionVerb::New) => Err(SessionControlError::Unsupported(format!(
            "`{harness}` opens a conversation through `harness.v1.runtimes.start` (CLI: \
             `supercode run --harness {harness}`), not through a slash command; `sessions.new` \
             is only for the gateway harnesses whose surface outlives the conversation"
        ))),
        (_, SessionVerb::Reset) => Err(SessionControlError::Unsupported(format!(
            "`{harness}` has no conversation reset verb: a fresh conversation is a new runtime \
             (`harness.v1.runtimes.start`). `sessions.reset` is only for the gateway harnesses \
             whose surface outlives the conversation"
        ))),
        (other, verb) => Err(SessionControlError::Unsupported(format!(
            "`{other}` publishes no door for `sessions.{}`; conversation mutation is supported \
             for: {}",
            verb.as_str(),
            CONTROLLED_SESSION_HARNESSES.join(", ")
        ))),
    }
}

// ---------------------------------------------------------------------------
// Command narration (the same contract `jobs_control` established)
// ---------------------------------------------------------------------------

fn shell_quote(value: &str) -> String {
    if !value.is_empty()
        && value
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || "-_./:@=+,".contains(c))
    {
        return value.to_string();
    }
    format!("'{}'", value.replace('\'', "'\\''"))
}

/// A harness CLI invocation, ready to run and ready to narrate.
#[derive(Debug, Clone)]
struct HarnessCommand {
    program: String,
    args: Vec<String>,
    env: Vec<(String, String)>,
}

impl HarnessCommand {
    fn new(program: impl Into<String>) -> Self {
        Self {
            program: program.into(),
            args: Vec::new(),
            env: Vec::new(),
        }
    }

    fn args<I: IntoIterator<Item = S>, S: Into<String>>(&mut self, values: I) -> &mut Self {
        for value in values {
            self.args.push(value.into());
        }
        self
    }

    fn env(&mut self, key: impl Into<String>, value: impl Into<String>) -> &mut Self {
        self.env.push((key.into(), value.into()));
        self
    }

    /// The narration: exactly what ran.
    fn narrate(&self) -> String {
        let mut line = shell_quote(&self.program);
        for arg in &self.args {
            line.push(' ');
            line.push_str(&shell_quote(arg));
        }
        line
    }

    /// Run it, returning stdout on success and the harness's own stderr on
    /// failure.
    fn run(&self) -> Result<String> {
        let mut command = Command::new(&self.program);
        command.args(&self.args);
        for (key, value) in &self.env {
            command.env(key, value);
        }
        command.stdin(std::process::Stdio::null());
        let output = command.output().map_err(|error| {
            SessionControlError::Failed(format!(
                "`{}` could not be executed: {error}",
                self.narrate()
            ))
        })?;
        if output.status.success() {
            return Ok(String::from_utf8_lossy(&output.stdout).into_owned());
        }
        let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
        let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string();
        let detail = if stderr.is_empty() { stdout } else { stderr };
        Err(SessionControlError::Failed(format!(
            "`{}` failed ({}): {}",
            self.narrate(),
            output.status,
            if detail.is_empty() {
                "the harness printed nothing".to_string()
            } else {
                detail
            }
        )))
    }
}

/// The harness's own executable, with a `SUPERCODE_<HARNESS>_BIN` override so
/// a fake CLI can stand in under test without touching PATH. The registry
/// names each harness's binary family in its runtime launch; the lifecycle
/// verbs live on the base CLI, so an `-acp` bridge suffix is stripped.
pub fn harness_program(harness: &str) -> Result<String> {
    if harness == HarnessId::CODEX {
        return crate::startup_prompts::codex_program().map_err(SessionControlError::Failed);
    }
    let variable = match harness {
        HarnessId::CODEX => CODEX_BIN_ENV,
        HarnessId::HERMES => HERMES_BIN_ENV,
        other => {
            return Err(SessionControlError::Unsupported(format!(
                "`{other}` has no conversation CLI Volter Harness calls"
            )));
        }
    };
    if let Some(over) = std::env::var_os(variable) {
        let over = over.to_string_lossy().trim().to_string();
        if !over.is_empty() {
            return Ok(over);
        }
    }
    let program = crate::harness_support(harness)
        .and_then(|descriptor| descriptor.runtime.default_launch)
        .map(|launch| launch.program)
        .ok_or_else(|| {
            SessionControlError::Unsupported(format!(
                "the registry has no launch for `{harness}`, so its CLI cannot be located"
            ))
        })?;
    Ok(program.strip_suffix("-acp").unwrap_or(&program).to_string())
}

/// `HERMES_HOME` for this request: the profile's own home when one is named
/// (upstream treats a profile as a full `HERMES_HOME`), else the install root.
/// `HarnessHomes::hermes` addresses `state.db`; `HERMES_HOME` is its parent —
/// the same derivation the read side uses.
fn hermes_home(mutation: &SessionMutation) -> PathBuf {
    let root = mutation
        .homes
        .hermes
        .parent()
        .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
    match mutation.profile.as_deref() {
        Some(profile) => root.join("profiles").join(profile),
        None => root,
    }
}

/// `CODEX_HOME` for this request. `HarnessHomes::codex` addresses the
/// `sessions/` directory inside it; codex itself wants the parent.
fn codex_home(mutation: &SessionMutation) -> PathBuf {
    let root = &mutation.homes.codex;
    if root.file_name().is_some_and(|name| name == "sessions") {
        return root
            .parent()
            .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
    }
    root.clone()
}

// ---------------------------------------------------------------------------
// Re-reading the harness's own store
// ---------------------------------------------------------------------------

/// The conversation as the harness's own store reports it right now, through
/// the ORCH-6 discovery loader. `None` means the store no longer holds it.
fn read_back(mutation: &SessionMutation, session: &str) -> Result<Option<Value>> {
    if mutation.harness == HarnessId::SUPERCODE {
        let store = crate::SessionStore::open(&mutation.homes.supercode).map_err(|error| {
            SessionControlError::Failed(format!(
                "Volter Harness's session store is unreadable: {error}"
            ))
        })?;
        return Ok(store
            .list()
            .into_iter()
            .find(|info| info.name == session)
            .map(|info| serde_json::to_value(info).unwrap_or(Value::Null)));
    }
    let page = crate::discover_session_page(&DiscoveryQuery {
        harnesses: vec![HarnessId::new(mutation.harness.clone())],
        homes: mutation.homes.clone(),
        include_child_sessions: true,
        ..DiscoveryQuery::default()
    })
    .map_err(|error| {
        SessionControlError::Failed(format!(
            "the {} conversation store could not be re-read: {error}",
            mutation.harness
        ))
    })?;
    Ok(page
        .sessions
        .into_iter()
        .find(|descriptor| descriptor.locator.session_id == session)
        .map(|descriptor| serde_json::to_value(descriptor).unwrap_or(Value::Null)))
}

// ---------------------------------------------------------------------------
// The mutation itself
// ---------------------------------------------------------------------------

/// Perform one conversation mutation through the harness's own door.
///
/// [`SessionDoor::Live`] verbs are NOT handled here: they need an open runtime
/// connection, which only the service owns. Callers check [`door`] first and
/// route those to `harness.v1.runtimes.send_input`; asking for one here is an
/// [`SessionControlError::Invalid`], because it names a door this function
/// cannot open rather than a door the harness lacks.
pub async fn mutate(
    verb: SessionVerb,
    mutation: &SessionMutation,
) -> Result<SessionMutationOutcome> {
    let door = door(&mutation.harness, verb)?;
    let session = target_session(verb, &door, mutation)?;
    if let SessionDoor::Http = door {
        let ran = opencode_call(verb, mutation, &session).await?;
        // The HTTP door re-reads through OPENCODE'S OWN API, not through
        // the file/SQLite loader: the running server owns the store, and
        // its answer is the only one that can be current.
        let row = opencode_read_back(mutation, &session).await?;
        return finish(verb, mutation, session, ran, row);
    }
    perform(verb, mutation, door, session)
}

/// The same mutation as [`mutate`] for every door but the HTTP one — and
/// every one of those doors works with calls that block the calling THREAD:
/// the harness's own CLI run to completion, supercode's store, the
/// orchestrator daemon's socket, and the read-back each of them ends with.
///
/// A caller under a deadline runs this on a blocking task. Awaiting the
/// `async` [`mutate`] instead parks the calling task inside a future that
/// never yields, so no timeout wrapped around it can ever fire.
pub fn mutate_blocking(
    verb: SessionVerb,
    mutation: &SessionMutation,
) -> Result<SessionMutationOutcome> {
    let door = door(&mutation.harness, verb)?;
    let session = target_session(verb, &door, mutation)?;
    perform(verb, mutation, door, session)
}

/// The conversation a mutation names, refused when the verb needs one and the
/// request named none.
fn target_session(
    verb: SessionVerb,
    door: &SessionDoor,
    mutation: &SessionMutation,
) -> Result<String> {
    let session = mutation.session.as_deref().unwrap_or("").trim().to_string();
    // The orchestrator's conversation is named by its SURFACE, not by an id:
    // `needs_session` is about a store row, and a binding is not one.
    if verb.needs_session() && session.is_empty() && !matches!(door, SessionDoor::Daemon) {
        return Err(SessionControlError::Invalid(format!(
            "`sessions.{}` needs the conversation to act on",
            verb.as_str()
        )));
    }
    Ok(session)
}

/// Every door that answers without awaiting anything.
fn perform(
    verb: SessionVerb,
    mutation: &SessionMutation,
    door: SessionDoor,
    session: String,
) -> Result<SessionMutationOutcome> {
    match door {
        SessionDoor::Http => Err(SessionControlError::Invalid(format!(
            "`{}` performs `sessions.{}` through its own HTTP API, which is not a blocking \
             door; call [`mutate`]",
            mutation.harness,
            verb.as_str()
        ))),
        SessionDoor::Live(command) => Err(SessionControlError::Invalid(format!(
            "`{}` performs `sessions.{}` by typing `{command}` into a LIVE driven session; call \
             it with an open runtime `connection`",
            mutation.harness,
            verb.as_str()
        ))),
        SessionDoor::Cli => {
            let command = cli_command(verb, mutation, &session)?;
            let ran = command.narrate();
            command.run()?;
            let row = read_back(mutation, &session)?;
            finish(verb, mutation, session, ran, row)
        }
        SessionDoor::Store => {
            let store = crate::SessionStore::open(&mutation.homes.supercode).map_err(|error| {
                SessionControlError::Failed(format!(
                    "Volter Harness's session store is unreadable: {error}"
                ))
            })?;
            let ran = format!(
                "supercode store {} {}",
                verb.as_str(),
                shell_quote(&session)
            );
            match verb {
                SessionVerb::Archive => store.archive(&session),
                SessionVerb::Delete => store.delete(&session),
                _ => unreachable!("the door table only routes archive/delete to the store"),
            }
            .map_err(|error| SessionControlError::Failed(format!("`{ran}` failed: {error}")))?;
            let row = read_back(mutation, &session)?;
            finish(verb, mutation, session, ran, row)
        }
        SessionDoor::Daemon => orchestrator_mutate(verb, mutation),
    }
}

// ---------------------------------------------------------------------------
// The orchestrator — its own daemon's operator door (ORC-13)
// ---------------------------------------------------------------------------

/// `sessions.new|reset --harness orchestrator --surface <key>`.
///
/// The verb ends the LIVE binding on that surface, which is exactly what the
/// `/new` and `/reset` chat commands do when a human types them into the
/// conversation (`docs/ORCHESTRATOR-IR.md` §4.5) — the same reducer, reached
/// through the daemon's operator door instead of through a chat message.
/// Afterwards the binding is re-read through the ORCH-6 discovery loader, the
/// same reader `sessions list --harness orchestrator` uses.
fn orchestrator_mutate(
    verb: SessionVerb,
    mutation: &SessionMutation,
) -> Result<SessionMutationOutcome> {
    let surface = mutation
        .surface
        .as_deref()
        .map(str::trim)
        .filter(|surface| !surface.is_empty())
        .ok_or_else(|| {
            SessionControlError::Invalid(format!(
                "an orchestrator conversation is a BINDING on a surface, not a store row: \
                 `sessions.{}` needs `--surface \
                 <platform|chat_type|chat_id|thread_id|participant_id>` \
                 (`supercode sessions list --harness orchestrator` prints the surface of every \
                 binding)",
                verb.as_str()
            ))
        })?;
    let root = mutation.homes.orchestrator.clone();
    let profile = mutation
        .profile
        .as_deref()
        .map(str::trim)
        .filter(|profile| !profile.is_empty())
        .unwrap_or("default");
    let op = match verb {
        SessionVerb::New => "sessions.new",
        SessionVerb::Reset => "sessions.reset",
        other => {
            return Err(SessionControlError::Unsupported(format!(
                "the orchestrator has no door for `sessions.{}`",
                other.as_str()
            )))
        }
    };
    let args = serde_json::json!({ "surface": surface });
    let answer = crate::orchestrator_door::call(&root, op, &args, profile).map_err(|error| {
        match error {
            // The package refused: its sentence is the answer, exactly as a
            // harness's own stderr is for the CLI doors.
            crate::orchestrator_door::DoorError::Refused(message) => {
                SessionControlError::Failed(message)
            }
            crate::orchestrator_door::DoorError::Failed(message) => {
                SessionControlError::Failed(message)
            }
        }
    })?;
    let ran = format!("{} [{}]", answer.ran, answer.door.as_str());
    let session = answer
        .result
        .pointer("/binding/session_id")
        .and_then(Value::as_str)
        .filter(|id| !id.is_empty())
        .unwrap_or(surface)
        .to_string();
    // The FOLDER is the answer: the ended binding, read back through the
    // discovery loader on its own surface.
    let row = orchestrator_read_back(mutation, surface, &ran)?;
    Ok(SessionMutationOutcome {
        harness: mutation.harness.clone(),
        verb: verb.as_str().to_string(),
        ran,
        session,
        row,
        archived: None,
        deleted: None,
    })
}

/// The newest binding on `surface`, through the ORCH-6 discovery loader.
///
/// A binding is addressed by its surface, and the loader renders that surface
/// as Hermes's `agent:<profile>:<platform>:<chat_type>[:…]` key, so the match
/// is on the surface COLUMNS the descriptor carries, never on a string the
/// caller typed.
fn orchestrator_read_back(
    mutation: &SessionMutation,
    surface: &str,
    ran: &str,
) -> Result<Option<Value>> {
    let page = crate::discover_session_page(&DiscoveryQuery {
        harnesses: vec![HarnessId::new(mutation.harness.clone())],
        homes: mutation.homes.clone(),
        include_child_sessions: true,
        ..DiscoveryQuery::default()
    })
    .map_err(|error| {
        SessionControlError::Failed(format!(
            "`{ran}` succeeded but the orchestrator's binding store could not be re-read: {error}"
        ))
    })?;
    let wanted = surface_columns(surface);
    let mut best: Option<Value> = None;
    let mut best_at = 0;
    for descriptor in page.sessions {
        let key = descriptor.nouns.surface.as_ref();
        let found = [
            key.and_then(|k| k.platform.clone()).unwrap_or_default(),
            key.and_then(|k| k.kind.clone()).unwrap_or_default(),
            key.and_then(|k| k.chat_id.clone()).unwrap_or_default(),
            key.and_then(|k| k.thread_id.clone()).unwrap_or_default(),
            key.and_then(|k| k.participant_id.clone())
                .unwrap_or_default(),
        ];
        if found != wanted {
            continue;
        }
        let at = descriptor.updated_at_ms.unwrap_or_default();
        if best.is_none() || at >= best_at {
            best_at = at;
            best = Some(serde_json::to_value(&descriptor).unwrap_or(Value::Null));
        }
    }
    Ok(best)
}

/// A surface key string split into its five columns, empty for the absent
/// ones — the inverse of the IR's `surfaceKeyString`.
fn surface_columns(surface: &str) -> [String; 5] {
    let mut parts = surface.split('|');
    std::array::from_fn(|_| parts.next().unwrap_or("").to_string())
}

/// Turn a completed door into the outcome, enforcing that the harness's own
/// store agrees with what the door claimed.
fn finish(
    verb: SessionVerb,
    mutation: &SessionMutation,
    session: String,
    ran: String,
    row: Option<Value>,
) -> Result<SessionMutationOutcome> {
    let outcome = SessionMutationOutcome {
        harness: mutation.harness.clone(),
        verb: verb.as_str().to_string(),
        ran: ran.clone(),
        session: session.clone(),
        row: row.clone(),
        archived: None,
        deleted: None,
    };
    match verb {
        SessionVerb::Delete => {
            if row.is_some() {
                return Err(SessionControlError::Failed(format!(
                    "`{ran}` reported success but `{session}` is still in {}'s conversation store",
                    mutation.harness
                )));
            }
            Ok(SessionMutationOutcome {
                row: None,
                deleted: Some(true),
                ..outcome
            })
        }
        SessionVerb::Archive => {
            if !archive_took_effect(mutation, row.as_ref()) {
                return Err(SessionControlError::Failed(format!(
                    "`{ran}` reported success but {}'s store still lists `{session}` as an \
                     active conversation",
                    mutation.harness
                )));
            }
            Ok(SessionMutationOutcome {
                archived: Some(true),
                ..outcome
            })
        }
        SessionVerb::New | SessionVerb::Reset => Ok(outcome),
    }
}

/// Did the harness's own store record the archive?
///
/// Each harness answers in its own terms and none of them is guessed at:
///
/// * **supercode** publishes an `archived` flag on the row.
/// * **codex** MOVES the rollout out of `$CODEX_HOME/sessions` into
///   `archived_sessions/` (executed 2026-09-03 on an isolated `CODEX_HOME`;
///   receipt `docs/interop/research/orch19-codex-sessions-receipt-*.json`), so
///   its disappearance from the active listing IS the store's answer.
/// * **opencode** stamps `time.archived` on the session record its own API
///   returns; a `404` (the server dropped it) also counts as archived.
fn archive_took_effect(mutation: &SessionMutation, row: Option<&Value>) -> bool {
    let Some(row) = row else {
        return true;
    };
    if mutation.harness == HarnessId::SUPERCODE {
        return row
            .get("archived")
            .and_then(Value::as_bool)
            .unwrap_or(false);
    }
    row.pointer("/time/archived")
        .is_some_and(|value| !value.is_null())
}

/// Translate one verb onto the harness's own CLI invocation.
fn cli_command(
    verb: SessionVerb,
    mutation: &SessionMutation,
    session: &str,
) -> Result<HarnessCommand> {
    match (mutation.harness.as_str(), verb) {
        (HarnessId::CODEX, SessionVerb::Archive | SessionVerb::Delete) => {
            let mut command = HarnessCommand::new(harness_program(HarnessId::CODEX)?);
            command.env("CODEX_HOME", codex_home(mutation).to_string_lossy());
            command.args(["-c", "check_for_update_on_startup=false"]);
            command.args([verb.as_str(), session]);
            if matches!(verb, SessionVerb::Delete) {
                // Measured 2026-09-03 on codex 0.152: without `--force` the
                // delete refuses outright off a TTY ("cannot confirm session
                // deletion without an interactive terminal"). supercode never
                // drives a prompt it cannot see, so it passes the harness's
                // own non-interactive flag; the caller already asked for a
                // delete, and the verification read is what proves it landed.
                command.args(["--force"]);
            }
            Ok(command)
        }
        (HarnessId::HERMES, SessionVerb::Delete) => {
            let mut command = HarnessCommand::new(harness_program(HarnessId::HERMES)?);
            command.env("HERMES_HOME", hermes_home(mutation).to_string_lossy());
            // `--yes` skips hermes's own interactive confirmation; supercode
            // never drives a prompt it cannot see.
            command.args(["sessions", "delete", session, "--yes"]);
            Ok(command)
        }
        (harness, verb) => Err(SessionControlError::Unsupported(format!(
            "`{harness}` has no CLI verb for `sessions.{}`",
            verb.as_str()
        ))),
    }
}

// ---------------------------------------------------------------------------
// OpenCode — its own HTTP session API
// ---------------------------------------------------------------------------

/// OpenCode's running server, resolved the same way for the mutation and for
/// the re-read that verifies it.
///
/// `base_url` is required — an endpoint is a fact about the caller's
/// environment, and guessing one would mutate whichever OpenCode happened to
/// be listening. The bearer, when present, is sent as a sensitive header and
/// never appears in the narration.
fn opencode_endpoint(mutation: &SessionMutation) -> Result<(String, reqwest::Client)> {
    let base = mutation
        .base_url
        .as_deref()
        .map(|url| url.trim_end_matches('/').to_string())
        .ok_or_else(|| {
            SessionControlError::Invalid(
                "opencode conversations are mutated through its own running server: pass \
                 `base_url` (the address `runtimes.start` reports, or an `opencode serve` you \
                 already run)"
                    .into(),
            )
        })?;
    let mut headers = reqwest::header::HeaderMap::new();
    if let Some(bearer) = mutation.bearer.as_deref().filter(|t| !t.trim().is_empty()) {
        let mut value = reqwest::header::HeaderValue::from_str(&format!("Bearer {bearer}"))
            .map_err(|_| {
                SessionControlError::Invalid(
                    "the opencode bearer token is not a valid header value".into(),
                )
            })?;
        value.set_sensitive(true);
        headers.insert(reqwest::header::AUTHORIZATION, value);
    }
    let client = reqwest::Client::builder()
        .default_headers(headers)
        .build()
        .map_err(|error| {
            SessionControlError::Failed(format!("could not build the HTTP client: {error}"))
        })?;
    Ok((base, client))
}

/// Re-read one conversation through OPENCODE'S OWN session API.
///
/// `None` means the server no longer holds it (`404`), which is exactly what a
/// successful delete must produce. Anything else the server says — including
/// the `time.archived` stamp an archive leaves — comes back verbatim.
async fn opencode_read_back(mutation: &SessionMutation, session: &str) -> Result<Option<Value>> {
    let (base, client) = opencode_endpoint(mutation)?;
    let url = format!("{base}/session/{session}");
    let mut request = client.get(&url);
    if let Some(cwd) = mutation.cwd.as_ref() {
        request = request.query(&[("directory", cwd.to_string_lossy().into_owned())]);
    }
    let response = request.send().await.map_err(|error| {
        SessionControlError::Failed(format!("`GET {url}` could not be sent: {error}"))
    })?;
    if response.status() == reqwest::StatusCode::NOT_FOUND {
        return Ok(None);
    }
    let status = response.status();
    if !status.is_success() {
        let body = response.text().await.unwrap_or_default();
        return Err(SessionControlError::Failed(format!(
            "`GET {url}` failed ({status}): {}",
            body.trim()
        )));
    }
    response
        .json::<Value>()
        .await
        .map(|value| if value.is_null() { None } else { Some(value) })
        .map_err(|error| {
            SessionControlError::Failed(format!("`GET {url}` returned unreadable JSON: {error}"))
        })
}

/// Call OpenCode's own session API and return the narration.
///
/// The endpoint is the RUNNING server's: supercode never opens OpenCode's
/// SQLite store to archive or delete a row.
async fn opencode_call(
    verb: SessionVerb,
    mutation: &SessionMutation,
    session: &str,
) -> Result<String> {
    let (base, client) = opencode_endpoint(mutation)?;
    let url = format!("{base}/session/{session}");
    let directory = mutation
        .cwd
        .as_ref()
        .map(|cwd| cwd.to_string_lossy().into_owned());
    let (ran, request) = match verb {
        SessionVerb::Delete => (format!("DELETE {url}"), client.delete(&url)),
        SessionVerb::Archive => {
            // OpenCode records the archive as `time.archived` on the session
            // record (`packages/schema/src/v1/session.ts`), patched through
            // the same route that renames a session. The re-read in `finish`
            // is what PROVES it landed: a payload the server ignores leaves
            // `time.archived` unset and the mutation fails.
            let now = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|since| since.as_millis() as u64)
                .unwrap_or_default();
            (
                format!("PATCH {url} {{\"time\":{{\"archived\":{now}}}}}"),
                client
                    .patch(&url)
                    .json(&serde_json::json!({"time": {"archived": now}})),
            )
        }
        other => {
            return Err(SessionControlError::Unsupported(format!(
                "opencode has no HTTP door for `sessions.{}`",
                other.as_str()
            )));
        }
    };
    let request = match &directory {
        Some(directory) => request.query(&[("directory", directory)]),
        None => request,
    };
    let response = request.send().await.map_err(|error| {
        SessionControlError::Failed(format!("`{ran}` could not be sent: {error}"))
    })?;
    let status = response.status();
    if !status.is_success() {
        let body = response.text().await.unwrap_or_default();
        return Err(SessionControlError::Failed(format!(
            "`{ran}` failed ({status}): {}",
            if body.trim().is_empty() {
                "the server returned no body".to_string()
            } else {
                body.trim().to_string()
            }
        )));
    }
    Ok(ran)
}

/// Build the outcome for a slash-command door the SERVICE performed, so the
/// live path and the subprocess path publish exactly the same shape.
pub fn live_outcome(
    verb: SessionVerb,
    mutation: &SessionMutation,
    command: &str,
    session: String,
) -> Result<SessionMutationOutcome> {
    let row = read_back(mutation, &session).unwrap_or(None);
    Ok(SessionMutationOutcome {
        harness: mutation.harness.clone(),
        verb: verb.as_str().to_string(),
        ran: format!("{} live session: {command}", mutation.harness),
        session,
        row,
        archived: None,
        deleted: None,
    })
}