supercode_harness/

harness_service.rs

//! Versioned, language-neutral service over persisted harness sessions.
//!
//! The service is transport-agnostic: [`HarnessSessionService::handle`] accepts
//! one JSON-RPC value and [`HarnessSessionService::poll`] produces subscription
//! notifications. The CLI exposes those primitives as NDJSON over stdio.

use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use tokio::sync::Notify;

use crate::reduce;
use crate::runtime::generated_session_id;
#[cfg(feature = "adapter-api")]
use crate::runtime::{HostedHarnessConnection, HostedHarnessRuntime};
use crate::sdk::{
    discover_session_page, load_session, load_session_with_fidelity, SdkCapabilities, SdkError,
    SdkErrorCode, SdkEvent, SdkOperation, SdkRequest, SdkRuntimeEvent, SdkService,
};
use crate::Fidelity;
#[cfg(feature = "adapter-api")]
use crate::SupercodeHttpRuntimeBackend;
use crate::{
    discover_live_runtime, harness_support_registry, AcpRuntimeBackend, ClaudeCodeRuntimeBackend,
    CodexRuntimeBackend, DiscoveryQuery, HarnessCatalog, HarnessHomes, HarnessId,
    ImplementationKind, LiveRuntimeEndpoint, LiveRuntimeSource, OpenCodeRuntimeBackend,
    PiRuntimeBackend, Role, RuntimeAttachRequest, RuntimeBackend, RuntimeConnection, RuntimeInput,
    RuntimeLaunch, RuntimeStartRequest, Session, SessionDescriptor, SessionFollower, SessionFormat,
    SessionLocator, SessionSource,
};
#[cfg(feature = "adapter-api")]
use crate::{register_live_runtime_with_metadata, resolve_live_runtime};
use supercode_interchange::watch::{bound_session_view, message_json, normalized_session_json};

/// Every JSON-RPC method the harness service dispatches (`harness.v1.capabilities`
/// reports it; ORCH-4 registry tiers must cite entries of it).
pub const HARNESS_SERVICE_METHODS: &[&str] = &[
    "harness.v1.support.report",
    "harness.v1.harnesses.list",
    "harness.v1.harnesses.probe",
    "harness.v1.harnesses.settings",
    "harness.v1.harnesses.configure",
    "harness.v1.harnesses.auth.methods",
    "harness.v1.harnesses.auth.begin",
    "harness.v1.harnesses.auth.verify",
    "harness.v1.sessions.discover",
    "harness.v1.sessions.usage",
    "harness.v1.sessions.spend",
    "harness.v1.sessions.tree",
    "harness.v1.sessions.load",
    "harness.v1.sessions.recorded_config",
    "harness.v1.sessions.launch_agent",
    "harness.v1.sessions.follow",
    "harness.v1.sessions.unfollow",
    "harness.v1.sessions.activity.subscribe",
    "harness.v1.sessions.activity.unsubscribe",
    "harness.v1.sessions.activity_under",
    "harness.v1.sessions.index.subscribe",
    "harness.v1.sessions.index.resize",
    "harness.v1.sessions.index.unsubscribe",
    "harness.v1.sessions.message",
    "harness.v1.sessions.inbox",
    "harness.v1.sessions.import",
    "harness.v1.sessions.export",
    "harness.v1.sessions.translate",
    "harness.v1.sessions.reduce",
    "harness.v1.sessions.branch",
    "harness.v1.sessions.handoff",
    "harness.v1.sessions.materialize",
    "harness.v1.sessions.resume_instructions",
    "harness.v1.skills.list",
    "harness.v1.skills.install",
    "harness.v1.skills.remove",
    "harness.v1.memory.show",
    "harness.v1.memory.search",
    "harness.v1.jobs.list",
    "harness.v1.jobs.get",
    "harness.v1.jobs.create",
    "harness.v1.jobs.update",
    "harness.v1.jobs.pause",
    "harness.v1.jobs.resume",
    "harness.v1.jobs.run",
    "harness.v1.jobs.delete",
    "harness.v1.jobs.notepad",
    "harness.v1.jobs.notepad_set",
    "harness.v1.jobs.notepad_delete",
    "harness.v1.sessions.new",
    "harness.v1.sessions.reset",
    "harness.v1.sessions.archive",
    "harness.v1.sessions.delete",
    "harness.v1.runs.list",
    "harness.v1.runs.get",
    "harness.v1.approvals.list",
    "harness.v1.approvals.resolve",
    "harness.v1.runtimes.capabilities",
    "harness.v1.runtimes.start",
    "harness.v1.runtimes.resume",
    "harness.v1.runtimes.attach_existing",
    "harness.v1.runtimes.attach",
    "harness.v1.runtimes.send_input",
    "harness.v1.runtimes.interrupt",
    "harness.v1.runtimes.steer",
    "harness.v1.runtimes.respond",
    "harness.v1.runtimes.terminal_instructions",
    "harness.v1.runtimes.acquire_control",
    "harness.v1.runtimes.heartbeat",
    "harness.v1.runtimes.detach",
    "harness.v1.runtimes.close",
    "harness.v1.profiles.list",
    "harness.v1.profiles.get",
    "harness.v1.profiles.create",
    "harness.v1.profiles.delete",
    "harness.v1.channels.list",
    "harness.v1.routes.list",
    "harness.v1.triggers.list",
    "harness.v1.channels.status",
    "harness.v1.orchestration.load",
    "harness.v1.orchestration.save",
    "harness.v1.orchestration.compile",
    "harness.v1.orchestration.decompile",
    "harness.v1.orchestration.import",
    "harness.v1.orchestration.export",
    "harness.v1.workflow.load",
];

/// Protocol namespace implemented by this service.
pub const HARNESS_SERVICE_VERSION: &str = "harness.v1";
/// Notification method emitted for followed-session changes.
pub const SESSION_EVENT_METHOD: &str = "harness.v1.sessions.event";
/// Notification method emitted for normalized session-activity transitions.
pub const SESSION_ACTIVITY_EVENT_METHOD: &str = "harness.v1.sessions.activity_event";
/// Notification method emitted for revisioned session-list changes.
pub const SESSION_INDEX_EVENT_METHOD: &str = "harness.v1.sessions.index_event";
/// Notification method emitted for live runtime events.
pub const RUNTIME_EVENT_METHOD: &str = "harness.v1.runtimes.event";

/// Stateful persisted-session service. Each instance owns its follow
/// subscriptions; discovery and loading remain read-only.
pub struct HarnessSessionService {
    catalog: HarnessCatalog,
    followers: BTreeMap<String, SessionFollower>,
    followed_sources: BTreeMap<String, FollowedSource>,
    activity_subscriptions: BTreeMap<String, ActivitySubscription>,
    index_subscriptions: BTreeMap<String, crate::session_index::SessionIndexSubscription>,
    index_notifier: Arc<Notify>,
    #[cfg(feature = "adapter-api")]
    activity_monitor: crate::session_activity::SessionActivityMonitor,
    next_subscription: u64,
    runtimes: BTreeMap<String, Box<dyn RuntimeConnection>>,
    /// Connections lent to a detached call that is running right now. The
    /// runtime itself is OUT of `runtimes` for that whole call, and these
    /// names are how a second caller is told the connection is busy rather
    /// than unknown.
    runtimes_in_flight: BTreeSet<String>,
    terminal_launches: BTreeMap<String, StructuredLaunch>,
    runtime_sequences: BTreeMap<String, u64>,
    next_runtime: u64,
    reduction_store_root: Option<PathBuf>,
    /// ORCH-9: live permission/approval requests outstanding on the open
    /// runtime connections above, fed by the same event pump that publishes
    /// `harness.v1.runtimes.event`.
    approvals: crate::approvals::ApprovalRegistry,
    /// ORCH-9: supercode's own queued subagent approvals, when the host that
    /// owns this service publishes its parent queue here.
    subagent_approvals: Option<Arc<std::sync::Mutex<Vec<crate::subagents::QueuedApproval>>>>,
}

impl Default for HarnessSessionService {
    fn default() -> Self {
        Self::new()
    }
}

impl HarnessSessionService {
    /// Create an empty service instance.
    pub fn new() -> Self {
        Self {
            catalog: HarnessCatalog::new(),
            followers: BTreeMap::new(),
            followed_sources: BTreeMap::new(),
            activity_subscriptions: BTreeMap::new(),
            index_subscriptions: BTreeMap::new(),
            index_notifier: Arc::new(Notify::new()),
            #[cfg(feature = "adapter-api")]
            activity_monitor: Default::default(),
            next_subscription: 1,
            runtimes: BTreeMap::new(),
            runtimes_in_flight: BTreeSet::new(),
            terminal_launches: BTreeMap::new(),
            runtime_sequences: BTreeMap::new(),
            next_runtime: 1,
            reduction_store_root: None,
            approvals: crate::approvals::ApprovalRegistry::new(),
            subagent_approvals: None,
        }
    }

    /// Override the trusted, service-owned store used for durable reduction
    /// bundles. Embedders and tests use this to keep all writes inside an
    /// explicitly selected root; the CLI otherwise uses the normal
    /// `$SUPERCODE_HOME/sessions` location.
    pub fn with_reduction_store_root(mut self, root: impl Into<PathBuf>) -> Self {
        self.reduction_store_root = Some(root.into());
        self
    }

    /// ORCH-9: publish the parent's own subagent-approval queue into
    /// `harness.v1.approvals.list`.
    ///
    /// This is the SAME `Arc` an [`crate::Agent`] pushes into
    /// (`Agent::pending_child_approvals`), so a host that runs supercode's own
    /// loop beside this service surfaces those requests through the uniform
    /// door without copying them anywhere.
    pub fn observe_subagent_approvals(
        &mut self,
        queue: Arc<std::sync::Mutex<Vec<crate::subagents::QueuedApproval>>>,
    ) {
        self.subagent_approvals = Some(queue);
    }

    /// ORCH-9: every approval request this service can see, newest last.
    ///
    /// Two sources, both live: the requests outstanding on the open runtime
    /// connections, and supercode's own queued subagent approvals. There is
    /// no file or database source at the pinned harness versions (see
    /// [`crate::approvals`]), so a stored or proposal row is never produced.
    pub fn approvals(&self, query: &crate::approvals::ApprovalsQuery) -> Vec<crate::ApprovalRow> {
        let now = crate::approvals::now_ms();
        let mut rows = self.approvals.rows(now);
        if let Some(queue) = self.subagent_approvals.as_ref() {
            let queued = queue
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .clone();
            rows.extend(crate::approvals::subagent_rows(&queued, now));
        }
        rows.retain(|row| query.matches(row));
        rows.sort_by(|left, right| {
            left.requested_at_ms
                .cmp(&right.requested_at_ms)
                .then_with(|| left.id.cmp(&right.id))
        });
        rows
    }

    /// ORCH-20 (controlled tier): answer one listed approval request by its
    /// row id and one uniform decision.
    ///
    /// The decision is translated into the option token and reply envelope
    /// the door that raised the request already accepts
    /// ([`crate::approvals::plan_reply`]), and the answer is then sent by
    /// calling `harness.v1.runtimes.respond` itself — the same code path, the
    /// same adapter, the same bookkeeping that drops the row. This verb adds
    /// a translation and nothing else.
    async fn approvals_resolve(
        &mut self,
        params: Value,
    ) -> std::result::Result<Value, ServiceError> {
        let params = decode::<crate::approvals::ApprovalsResolveParams>(params)?;
        if params.id.trim().is_empty() {
            return Err(ServiceError::InvalidParams(
                "approvals resolve requires the `id` of a listed approval row".into(),
            ));
        }
        let choice = match (params.decision, params.option_id.as_deref()) {
            (Some(_), Some(_)) => {
                return Err(ServiceError::InvalidParams(
                    "approvals resolve takes either `decision` or `option_id`, not both".into(),
                ))
            }
            (Some(decision), None) => crate::approvals::ApprovalChoice::Decision(decision),
            (None, Some(option)) => crate::approvals::ApprovalChoice::Option(option.to_string()),
            (None, None) => {
                return Err(ServiceError::InvalidParams(format!(
                    "approvals resolve requires `decision` ({}) or an explicit `option_id`",
                    crate::approvals::ApprovalDecision::ALL
                        .map(|decision| decision.as_str())
                        .join(" | "),
                )))
            }
        };
        let resolution = self
            .approvals
            .resolution(&params.id, &choice)
            .map_err(|error| ServiceError::InvalidParams(error.to_string()))?;
        // The harness's own door, unchanged: this is the identical call
        // `harness.v1.runtimes.respond` performs for a caller who built the
        // envelope by hand, including dropping the answered row.
        self.runtime_call(
            "harness.v1.runtimes.respond",
            json!({
                "connection": resolution.connection,
                "request_id": resolution.request_id,
                "response": resolution.response,
            }),
        )
        .await?;
        Ok(json!({
            "id": params.id,
            "decision": params.decision.map(|decision| decision.as_str()),
            "option_id": resolution.option_id,
            "resolved": true,
        }))
    }

    /// Return the edge-triggered wakeup used by session-index filesystem
    /// subscriptions. Transports can await this instead of polling indexes.
    #[cfg(feature = "adapter-api")]
    pub fn session_index_notifier(&self) -> Arc<Notify> {
        Arc::clone(&self.index_notifier)
    }

    /// Handle one JSON-RPC 2.0 request and return one JSON-RPC response.
    #[cfg(feature = "adapter-api")]
    pub fn handle(&mut self, request: Value) -> Value {
        let id = request.get("id").cloned().unwrap_or(Value::Null);
        if request.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
            return rpc_error(id, -32600, "expected a JSON-RPC 2.0 request");
        }
        let Some(method) = request.get("method").and_then(Value::as_str) else {
            return rpc_error(id, -32600, "request is missing `method`");
        };
        let params = request.get("params").cloned().unwrap_or_else(|| json!({}));
        match self.call(method, params) {
            Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
            Err(ServiceError::InvalidParams(message)) => rpc_error(id, -32602, &message),
            Err(ServiceError::MethodNotFound) => rpc_error(id, -32601, "method not found"),
            Err(ServiceError::UnsupportedAction(message)) => rpc_error(id, -32020, &message),
            Err(ServiceError::Operation(message)) => rpc_error(id, -32000, &message),
            Err(ServiceError::Sdk(error)) => sdk_rpc_error(id, &error),
        }
    }

    /// Handle either a persisted-session request or an asynchronous live
    /// runtime request.
    #[cfg(feature = "adapter-api")]
    pub async fn handle_async(&mut self, request: Value) -> Value {
        let method = request
            .get("method")
            .and_then(Value::as_str)
            .unwrap_or_default();
        if matches!(
            method,
            "harness.v1.harnesses.list" | "harness.v1.harnesses.probe"
        ) {
            let id = request.get("id").cloned().unwrap_or(Value::Null);
            if request.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
                return rpc_error(id, -32600, "expected a JSON-RPC 2.0 request");
            }
            let params = request.get("params").cloned().unwrap_or_else(|| json!({}));
            return match self.inventory_call(method, params).await {
                Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
                Err(ServiceError::InvalidParams(message)) => rpc_error(id, -32602, &message),
                Err(ServiceError::MethodNotFound) => rpc_error(id, -32601, "method not found"),
                Err(ServiceError::UnsupportedAction(message)) => rpc_error(id, -32020, &message),
                Err(ServiceError::Operation(message)) => rpc_error(id, -32000, &message),
                Err(ServiceError::Sdk(error)) => sdk_rpc_error(id, &error),
            };
        }
        if matches!(
            method,
            "harness.v1.harnesses.auth.methods"
                | "harness.v1.harnesses.auth.begin"
                | "harness.v1.harnesses.auth.verify"
        ) {
            let id = request.get("id").cloned().unwrap_or(Value::Null);
            if request.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
                return rpc_error(id, -32600, "expected a JSON-RPC 2.0 request");
            }
            let params = request.get("params").cloned().unwrap_or_else(|| json!({}));
            return match self.harness_authentication_call(method, params).await {
                Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
                Err(ServiceError::InvalidParams(message)) => rpc_error(id, -32602, &message),
                Err(ServiceError::MethodNotFound) => rpc_error(id, -32601, "method not found"),
                Err(ServiceError::UnsupportedAction(message)) => rpc_error(id, -32020, &message),
                Err(ServiceError::Operation(message)) => rpc_error(id, -32000, &message),
                Err(ServiceError::Sdk(error)) => sdk_rpc_error(id, &error),
            };
        }
        // ORCH-19 controlled tier. Answered here rather than through the SDK
        // operation dispatch below so the harness's OWN refusal reaches the
        // caller: `sdk_error` collapses every `UnsupportedAction` to one
        // generic sentence, and the whole point of this tier is that a
        // refusal names which door the harness does have.
        if matches!(
            method,
            "harness.v1.sessions.new"
                | "harness.v1.sessions.reset"
                | "harness.v1.sessions.archive"
                | "harness.v1.sessions.delete"
        ) {
            let id = request.get("id").cloned().unwrap_or(Value::Null);
            if request.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
                return rpc_error(id, -32600, "expected a JSON-RPC 2.0 request");
            }
            let params = request.get("params").cloned().unwrap_or_else(|| json!({}));
            let verb = match method {
                "harness.v1.sessions.new" => crate::SessionVerb::New,
                "harness.v1.sessions.reset" => crate::SessionVerb::Reset,
                "harness.v1.sessions.archive" => crate::SessionVerb::Archive,
                _ => crate::SessionVerb::Delete,
            };
            return match self.mutate_session(verb, params).await {
                Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
                Err(ServiceError::InvalidParams(message)) => rpc_error(id, -32602, &message),
                Err(ServiceError::MethodNotFound) => rpc_error(id, -32601, "method not found"),
                Err(ServiceError::UnsupportedAction(message)) => rpc_error(id, -32020, &message),
                Err(ServiceError::Operation(message)) => rpc_error(id, -32000, &message),
                Err(ServiceError::Sdk(error)) => sdk_rpc_error(id, &error),
            };
        }
        if method == "harness.v1.sessions.message" {
            let id = request.get("id").cloned().unwrap_or(Value::Null);
            if request.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
                return rpc_error(id, -32600, "expected a JSON-RPC 2.0 request");
            }
            let params = request.get("params").cloned().unwrap_or_else(|| json!({}));
            return match self.message_call(params).await {
                Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
                Err(ServiceError::InvalidParams(message)) => rpc_error(id, -32602, &message),
                Err(ServiceError::MethodNotFound) => rpc_error(id, -32601, "method not found"),
                Err(ServiceError::UnsupportedAction(message)) => rpc_error(id, -32020, &message),
                Err(ServiceError::Operation(message)) => rpc_error(id, -32000, &message),
                Err(ServiceError::Sdk(error)) => sdk_rpc_error(id, &error),
            };
        }
        if matches!(
            method,
            "harness.v1.harnesses.settings" | "harness.v1.harnesses.configure"
        ) {
            let id = request.get("id").cloned().unwrap_or(Value::Null);
            if request.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
                return rpc_error(id, -32600, "expected a JSON-RPC 2.0 request");
            }
            let params = request.get("params").cloned().unwrap_or_else(|| json!({}));
            return match self.harness_settings_call(method, params) {
                Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
                Err(ServiceError::InvalidParams(message)) => rpc_error(id, -32602, &message),
                Err(ServiceError::MethodNotFound) => rpc_error(id, -32601, "method not found"),
                Err(ServiceError::UnsupportedAction(message)) => rpc_error(id, -32020, &message),
                Err(ServiceError::Operation(message)) => rpc_error(id, -32000, &message),
                Err(ServiceError::Sdk(error)) => sdk_rpc_error(id, &error),
            };
        }
        if method == "harness.v1.sessions.activity_under" {
            let id = request.get("id").cloned().unwrap_or(Value::Null);
            let params = request.get("params").cloned().unwrap_or_else(|| json!({}));
            return match activity_under_call(params).await {
                Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
                Err(ServiceError::InvalidParams(message)) => rpc_error(id, -32602, &message),
                Err(ServiceError::Sdk(error)) => sdk_rpc_error(id, &error),
                Err(ServiceError::Operation(message)) => rpc_error(id, -32000, &message),
                Err(_) => rpc_error(id, -32000, "sessions.activity_under failed"),
            };
        }
        if method == "harness.v1.sessions.activity.subscribe" {
            let id = request.get("id").cloned().unwrap_or(Value::Null);
            if request.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
                return rpc_error(id, -32600, "expected a JSON-RPC 2.0 request");
            }
            let params = request.get("params").cloned().unwrap_or_else(|| json!({}));
            return match self.subscribe_session_activity(params).await {
                Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
                Err(ServiceError::InvalidParams(message)) => rpc_error(id, -32602, &message),
                Err(ServiceError::MethodNotFound) => rpc_error(id, -32601, "method not found"),
                Err(ServiceError::UnsupportedAction(message)) => rpc_error(id, -32020, &message),
                Err(ServiceError::Operation(message)) => rpc_error(id, -32000, &message),
                Err(ServiceError::Sdk(error)) => sdk_rpc_error(id, &error),
            };
        }
        if let Some(operation) = SdkOperation::from_method(method) {
            let id = request.get("id").cloned().unwrap_or(Value::Null);
            if request.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
                return rpc_error(id, -32600, "expected a JSON-RPC 2.0 request");
            }
            let params = request.get("params").cloned().unwrap_or_else(|| json!({}));
            return match self.execute(SdkRequest { operation, params }).await {
                Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
                Err(error) => sdk_rpc_error(id, &error),
            };
        }
        if !method.starts_with("harness.v1.runtimes.") {
            return self.handle(request);
        }
        let id = request.get("id").cloned().unwrap_or(Value::Null);
        if request.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
            return rpc_error(id, -32600, "expected a JSON-RPC 2.0 request");
        }
        let params = request.get("params").cloned().unwrap_or_else(|| json!({}));
        match self.runtime_call(method, params).await {
            Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
            Err(ServiceError::InvalidParams(message)) => rpc_error(id, -32602, &message),
            Err(ServiceError::MethodNotFound) => rpc_error(id, -32601, "method not found"),
            Err(ServiceError::UnsupportedAction(message)) => rpc_error(id, -32020, &message),
            Err(ServiceError::Operation(message)) => rpc_error(id, -32000, &message),
            Err(ServiceError::Sdk(error)) => sdk_rpc_error(id, &error),
        }
    }

    /// Poll all active subscriptions once and return zero or more JSON-RPC
    /// notifications. Recoverable follower errors are delivered as events.
    #[cfg(feature = "adapter-api")]
    pub fn poll(&mut self) -> Vec<Value> {
        let mut notifications = Vec::new();
        for (subscription, follower) in &mut self.followers {
            match follower.poll() {
                Ok(Some(event)) => notifications.push(json!({
                    "jsonrpc": "2.0",
                    "method": SESSION_EVENT_METHOD,
                    "params": {
                        "subscription": subscription,
                        "event": event.to_json(),
                    }
                })),
                Ok(None) => {}
                Err(error) => notifications.push(json!({
                    "jsonrpc": "2.0",
                    "method": SESSION_EVENT_METHOD,
                    "params": {
                        "subscription": subscription,
                        "event": {
                            "type": "watch_error",
                            "recoverable": true,
                            "message": error.to_string(),
                        },
                    }
                })),
            }
        }
        notifications
    }

    /// Report each followed session's live-runtime lifecycle state on that
    /// session's own subscription, emitting only when the state changes.
    ///
    /// A growing transcript is not evidence that an agent is working, so the
    /// state comes from the live-runtime registry and nowhere else. A followed
    /// session with no registered Supercode runtime — a harness running outside
    /// Supercode — reports `persisted`, which says plainly that its activity is
    /// unknown rather than guessing at it. These events carry no sequence
    /// number and no transcript content; they never interleave with the
    /// content follower's sequenced stream.
    #[cfg(feature = "adapter-api")]
    pub async fn poll_session_runtime_states(&mut self) -> Vec<Value> {
        let registry = crate::LocalRuntimeRegistry::new();
        let authorization = crate::RuntimeAuthorization::observer();
        let mut notifications = Vec::new();
        for (subscription, source) in &mut self.followed_sources {
            let state = match registry
                .source_state(&source.harness, &source.session_id, &authorization)
                .await
            {
                Ok(Some(state)) => state,
                Ok(None) => crate::RuntimeRegistryState::Persisted,
                // A failed registry read is not evidence of a state change.
                Err(_) => continue,
            };
            if source.reported.as_deref() == Some(state.as_str()) {
                continue;
            }
            source.reported = Some(state.as_str().to_string());
            notifications.push(json!({
                "jsonrpc": "2.0",
                "method": SESSION_EVENT_METHOD,
                "params": {
                    "subscription": subscription,
                    "event": {"type": "runtime_state", "state": state.as_str()},
                },
            }));
        }
        notifications
    }

    /// Poll normalized activity subscriptions, emitting only proven state
    /// transitions. Every subscription is bulk-sampled so stock-harness
    /// process and registry discovery happens once per UI, not once per row.
    #[cfg(feature = "adapter-api")]
    pub async fn poll_session_activities(&mut self) -> Vec<Value> {
        let subscriptions = self
            .activity_subscriptions
            .iter()
            .map(|(id, subscription)| {
                (
                    id.clone(),
                    subscription.locators.clone(),
                    subscription.homes.clone(),
                )
            })
            .collect::<Vec<_>>();
        let mut notifications = Vec::new();
        for (subscription_id, locators, homes) in subscriptions {
            let Ok(mut activities) = self.activity_monitor.resolve(&locators, &homes).await else {
                // A failed evidence read proves no transition. Retain the last
                // good state instead of flashing every row to persisted.
                continue;
            };
            pane_prompt_override(&mut activities, &homes);
            let Some(subscription) = self.activity_subscriptions.get_mut(&subscription_id) else {
                continue;
            };
            let mut changed = Vec::new();
            for activity in activities {
                let key = activity.key();
                if subscription
                    .reported
                    .get(&key)
                    .is_some_and(|previous| previous.same_state(&activity))
                {
                    continue;
                }
                subscription.reported.insert(key, activity.clone());
                changed.push(activity);
            }
            if !changed.is_empty() {
                notifications.push(json!({
                    "jsonrpc": "2.0",
                    "method": SESSION_ACTIVITY_EVENT_METHOD,
                    "params": {
                        "subscription": subscription_id,
                        "activities": changed,
                    },
                }));
            }
        }
        notifications
    }

    /// Drain native-store invalidations and emit revisioned descriptor deltas.
    /// An idle subscription performs no catalog or transcript reads between
    /// its minute-scale recovery reconciliations.
    #[cfg(feature = "adapter-api")]
    pub fn poll_session_indexes(&mut self) -> Vec<Value> {
        let mut notifications = Vec::new();
        for (subscription, index) in &mut self.index_subscriptions {
            let homes = index.homes().clone();
            match index.poll() {
                Ok(Some(delta)) => match live_index_changes(delta.changes, &homes) {
                    Ok(changes) => notifications.push(json!({
                        "jsonrpc": "2.0",
                        "method": SESSION_INDEX_EVENT_METHOD,
                        "params": {
                            "subscription": subscription,
                            "revision": delta.revision,
                            "changes": changes,
                        },
                    })),
                    Err(error) => notifications.push(json!({
                        "jsonrpc": "2.0",
                        "method": SESSION_INDEX_EVENT_METHOD,
                        "params": {
                            "subscription": subscription,
                            "error": {"recoverable": true, "message": error_message(error)},
                        },
                    })),
                },
                Ok(None) => {}
                Err(error) => notifications.push(json!({
                    "jsonrpc": "2.0",
                    "method": SESSION_INDEX_EVENT_METHOD,
                    "params": {
                        "subscription": subscription,
                        "error": {"recoverable": true, "message": error},
                    },
                })),
            }
        }
        notifications
    }

    #[cfg(feature = "adapter-api")]
    async fn subscribe_session_activity(
        &mut self,
        params: Value,
    ) -> std::result::Result<Value, ServiceError> {
        let params = decode::<ActivitySubscribeParams>(params)?;
        if params.locators.is_empty() {
            return Err(ServiceError::InvalidParams(
                "sessions.activity.subscribe requires at least one locator".into(),
            ));
        }
        if params.locators.len() > 2_048 {
            return Err(ServiceError::InvalidParams(
                "sessions.activity.subscribe accepts at most 2048 locators".into(),
            ));
        }
        let mut initial = self
            .activity_monitor
            .resolve(&params.locators, &params.homes)
            .await
            .map_err(ServiceError::Sdk)?;
        pane_prompt_override(&mut initial, &params.homes);
        let subscription = format!("activity-sub-{}", self.next_subscription);
        self.next_subscription += 1;
        let reported = initial
            .iter()
            .cloned()
            .map(|activity| (activity.key(), activity))
            .collect();
        self.activity_subscriptions.insert(
            subscription.clone(),
            ActivitySubscription {
                locators: params.locators,
                homes: params.homes,
                reported,
            },
        );
        Ok(json!({"subscription": subscription, "initial": initial}))
    }

    /// Non-blockingly sample one event from every connected live runtime.
    #[cfg(feature = "adapter-api")]
    pub async fn poll_runtimes(&mut self) -> Vec<Value> {
        self.poll_sdk_events()
            .await
            .into_iter()
            .map(|(connection, runtime_event)| {
                json!({
                    "jsonrpc": "2.0",
                    "method": RUNTIME_EVENT_METHOD,
                    "params": {
                        "connection": connection,
                        "session_id": runtime_event.session_id,
                        "sequence": runtime_event.event.sequence,
                        "event": {
                            "kind": runtime_event.event.kind,
                            "payload": runtime_event.event.payload,
                        },
                    },
                })
            })
            .collect()
    }

    async fn poll_sdk_events(&mut self) -> Vec<(String, SdkRuntimeEvent)> {
        let mut events = Vec::new();
        let mut closed = Vec::new();
        let now_ms = crate::approvals::now_ms();
        for (connection, runtime) in &mut self.runtimes {
            let session_id = runtime.handle().runtime_id.clone();
            let harness = runtime.handle().harness.clone();
            // Drain what the runtime already has: a turn is several events
            // (updates, then the protocol's completion), and delivering one
            // per poll would cost a poll interval each. A zero timeout takes
            // only what is ready — an idle runtime costs nothing.
            for _ in 0..256 {
                match tokio::time::timeout(Duration::ZERO, runtime.next_event()).await {
                    Ok(Ok(Some(event))) => {
                        let terminal = event.kind == "transport_closed";
                        // ORCH-9: a permission/approval request arrives as an
                        // ordinary event; it becomes listable here and stops
                        // being listable when `runtimes.respond` answers it.
                        self.approvals
                            .observe(connection, &harness, &session_id, &event, now_ms);
                        let next_sequence = self
                            .runtime_sequences
                            .entry(session_id.clone())
                            .or_insert(0);
                        let sequence = event.sequence.unwrap_or_else(|| {
                            *next_sequence = next_sequence.saturating_add(1);
                            *next_sequence
                        });
                        *next_sequence = (*next_sequence).max(sequence);
                        events.push((
                            connection.clone(),
                            SdkRuntimeEvent {
                                session_id: session_id.clone(),
                                event: SdkEvent {
                                    sequence,
                                    kind: event.kind,
                                    payload: event.payload,
                                },
                            },
                        ));
                        if terminal {
                            closed.push(connection.clone());
                            break;
                        }
                    }
                    Ok(Ok(None)) => {
                        let sequence = self
                            .runtime_sequences
                            .entry(session_id.clone())
                            .or_insert(0);
                        *sequence = sequence.saturating_add(1);
                        events.push((
                        connection.clone(),
                        SdkRuntimeEvent {
                            session_id,
                            event: SdkEvent {
                                sequence: *sequence,
                                kind: "transport_closed".into(),
                                payload: json!({"message": "Harness runtime transport closed."}),
                            },
                        },
                    ));
                        closed.push(connection.clone());
                        break;
                    }
                    Err(_) => break,
                    Ok(Err(error)) => {
                        let sequence = self
                            .runtime_sequences
                            .entry(session_id.clone())
                            .or_insert(0);
                        *sequence = sequence.saturating_add(1);
                        events.push((
                        connection.clone(),
                        SdkRuntimeEvent {
                            session_id,
                            event: SdkEvent {
                                sequence: *sequence,
                                kind: "transport_error".into(),
                                payload: json!({"message": error.to_string(), "terminal": true}),
                            },
                        },
                    ));
                        closed.push(connection.clone());
                        break;
                    }
                }
            }
        }
        for connection in closed {
            if let Some(runtime) = self.runtimes.remove(&connection) {
                self.runtime_sequences.remove(&runtime.handle().runtime_id);
            }
            self.terminal_launches.remove(&connection);
            // A connection that is gone cannot answer anything it was
            // holding; those requests stop being listable with it.
            self.approvals.forget(&connection);
        }
        events
    }

    fn call(&mut self, method: &str, params: Value) -> std::result::Result<Value, ServiceError> {
        match method {
            "harness.v1.capabilities" => Ok(json!({
                "version": HARNESS_SERVICE_VERSION,
                "sdk": self.capabilities(),
                "methods": HARNESS_SERVICE_METHODS,
                "notifications": [
                    SESSION_EVENT_METHOD,
                    SESSION_ACTIVITY_EVENT_METHOD,
                    SESSION_INDEX_EVENT_METHOD,
                    RUNTIME_EVENT_METHOD
                ],
                "harnesses": harness_support_registry()
                    .harnesses
                    .into_iter()
                    .map(|harness| harness.id)
                    .collect::<Vec<_>>(),
            })),
            "harness.v1.support.report" => serde_json::to_value(harness_support_registry())
                .map_err(|error| ServiceError::Operation(error.to_string())),
            "harness.v1.profiles.list" | "harness.v1.profiles.get" => profiles_call(method, params),
            // ORCH-21 controlled tier. Each verb translates to the HARNESS'S
            // OWN profile verb and runs it (`crate::profiles_control`);
            // supercode makes and removes nothing itself. The row returned is
            // re-read through the ORCH-10 loader afterwards, and `ran`
            // narrates the exact command.
            "harness.v1.profiles.create" => {
                mutate_profile(crate::profiles_control::ProfileVerb::Create, params)
            }
            "harness.v1.profiles.delete" => {
                mutate_profile(crate::profiles_control::ProfileVerb::Delete, params)
            }
            "harness.v1.channels.list" | "harness.v1.channels.status" => {
                channels_call(method, params)
            }
            // ORCH-15 observed tier: which profile / agent a surface tuple
            // resolves to, read from each gateway harness's own config.
            "harness.v1.routes.list" => routes_call(params),
            // ORCH-16 observed tier: inbound webhook routes / hook mappings.
            "harness.v1.triggers.list" => triggers_call(params),
            // ONT-4: the orchestration doors. One home folder in, one typed orchestration
            // value out (and back). Every one of the four is
            // `crate::orchestration_doors`, which the `supercode orchestration` verbs call
            // too — the RPC adds nothing but the envelope. A vault VALUE
            // never crosses this wire: a load or a compile answers with the
            // `.env` KEY NAMES, and a caller that needs a value reads the
            // home's own `.env`.
            // the workflow layer's read door: a harness's board as one typed value,
            // the same code the `supercode workflow load` verb calls
            "harness.v1.workflow.load" => {
                let params = decode::<WorkflowLoadParams>(params)?;
                let read = crate::workflow_doors::load(
                    params.from.unwrap_or_else(|| {
                        crate::workflow_doors::WorkflowHarness::of_home(&params.home)
                    }),
                    &params.home,
                )
                .map_err(operation)?;
                serde_json::to_value(read)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            "harness.v1.orchestration.load" => {
                let params = decode::<OrchestrationLoadParams>(params)?;
                let read = crate::orchestration_doors::load(&params.root, params.flavor)
                    .map_err(operation)?;
                serde_json::to_value(read)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            "harness.v1.orchestration.save" => {
                let params = decode::<OrchestrationSaveParams>(params)?;
                let saved = crate::orchestration_doors::save(
                    &params.root,
                    params.orchestration,
                    params.vault,
                )
                .map_err(operation)?;
                serde_json::to_value(saved)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            "harness.v1.orchestration.compile" => {
                let params = decode::<OrchestrationCompileParams>(params)?;
                let read = crate::orchestration_doors::compile(params.from, &params.home)
                    .map_err(operation)?;
                serde_json::to_value(read)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            "harness.v1.orchestration.decompile" => {
                let params = decode::<OrchestrationDecompileParams>(params)?;
                let report = crate::orchestration_doors::decompile(
                    params.to,
                    params.orchestration,
                    &params.source,
                    params.source_flavor,
                    &params.dest,
                    params.vault,
                )
                .map_err(operation)?;
                serde_json::to_value(report)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            // a migration keeps the credential in this process: a compile and
            // a save (import), a load and a decompile (export), composed here
            // because composed by a client the secret would have to cross
            // the wire
            "harness.v1.orchestration.import" => {
                let params = decode::<OrchestrationImportParams>(params)?;
                let imported =
                    crate::orchestration_doors::import(params.from, &params.home, &params.into)
                        .map_err(operation)?;
                serde_json::to_value(imported)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            "harness.v1.orchestration.export" => {
                let params = decode::<OrchestrationExportParams>(params)?;
                let report =
                    crate::orchestration_doors::export(params.to, &params.root, &params.dest)
                        .map_err(operation)?;
                serde_json::to_value(report)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            // ORCH-12 observed tier: read and search the persistent memory
            // documents a harness keeps on disk. Read-only — every write
            // (`hermes memory off`, `openclaw memory forget|reset`, Claude
            // Code's `/memory`) stays the harness's own verb. A harness with
            // no memory store is refused with UnsupportedAction.
            "harness.v1.memory.show" | "harness.v1.memory.search" => memory_call(method, params),
            // ORCH-11 observed tier: read-only enumeration of every harness's
            // installed skill packages. An unknown harness id is refused with
            // UnsupportedAction — every harness supports skills, so a filter
            // that matches nothing is a caller error, never an empty listing.
            "harness.v1.skills.list" => {
                let query = decode::<crate::skills::SkillsQuery>(params)?;
                if let Some(harness) = query.harness.as_deref() {
                    if !crate::skills::SKILL_HARNESSES.contains(&harness) {
                        return Err(ServiceError::UnsupportedAction(format!(
                            "`{harness}` has no skills root Volter Harness reads"
                        )));
                    }
                }
                serde_json::to_value(crate::skills::list_skills(&query))
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            // ORCH-22 controlled tier: each verb goes through the door the
            // HARNESS publishes — `hermes skills install|uninstall`,
            // `openclaw skills install`, and for the core four the loader's
            // own directory, which is the only skills door those harnesses
            // have. supercode resolves no registry and unpacks no archive.
            // The row returned is re-read through the ORCH-11 loader
            // afterwards, and `ran` narrates exactly what was performed.
            "harness.v1.skills.install" => {
                mutate_skill(crate::skills_control::SkillVerb::Install, params)
            }
            "harness.v1.skills.remove" => {
                mutate_skill(crate::skills_control::SkillVerb::Remove, params)
            }
            // ORCH-9 observed tier: the approval requests waiting for an
            // answer. At the pinned harness versions the only uniform source
            // is a LIVE request held by an open runtime connection, plus
            // supercode's own queued subagent approvals — neither Hermes
            // 0.21.0 nor OpenClaw 2026.7.1-2 has an approvals door to read
            // (see `crate::approvals`). A harness whose runtime cannot carry
            // a protocol request at all is refused by name.
            "harness.v1.approvals.list" => {
                let query = decode::<crate::approvals::ApprovalsQuery>(params)?;
                if let Some(harness) = query.harness.as_deref() {
                    if !crate::approvals::lists_approvals(harness) {
                        return Err(ServiceError::UnsupportedAction(format!(
                            "`{harness}` has no runtime door that carries an approval request"
                        )));
                    }
                }
                serde_json::to_value(self.approvals(&query))
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            "harness.v1.sessions.discover" => {
                let query = decode::<DiscoveryQuery>(params)?;
                let mut page = discover_session_page(&query).map_err(operation)?;
                // Claude Code is the one harness that publishes its RUNNING
                // sessions. The registry is read once per discovery and joined
                // by session id; every record in it has already survived a
                // `kill(pid, 0)` liveness check inside `read_registry`.
                let doors = crate::mail_route::LiveSessions::read(&query.homes);
                // A running session's name (`twin-76`) lives only in its
                // harness's registry, not in the transcript discovery searches:
                // a query naming one also finds that session, by its id.
                if let Some(text) = query
                    .query
                    .as_deref()
                    .map(str::trim)
                    .filter(|text| !text.is_empty())
                {
                    let needle = text.to_lowercase();
                    for live in doors.all() {
                        let name = live
                            .name
                            .split('@')
                            .next()
                            .unwrap_or(&live.name)
                            .to_lowercase();
                        if !name.contains(&needle)
                            || page.sessions.iter().any(|session| {
                                session.locator.session_id == live.address.session_id
                            })
                        {
                            continue;
                        }
                        if query
                            .limit
                            .is_some_and(|limit| page.sessions.len() >= limit)
                        {
                            break;
                        }
                        // Only that session's own folder is scanned, not the whole store again.
                        let mut by_id = query.clone();
                        by_id.query = Some(live.address.session_id.clone());
                        by_id.cursor = None;
                        if let Some(cwd) = &live.cwd {
                            by_id.workspace = Some(cwd.clone());
                            by_id.workspace_subtree = false;
                            by_id.workspace_family = None;
                        }
                        if let Ok(found) = discover_session_page(&by_id) {
                            page.sessions
                                .extend(found.sessions.into_iter().filter(|session| {
                                    session.locator.session_id == live.address.session_id
                                }));
                        }
                    }
                }
                let activities = crate::session_activity::resolve_stock_session_activities(
                    &page
                        .sessions
                        .iter()
                        .map(|session| session.locator.clone())
                        .collect::<Vec<_>>(),
                    &query.homes,
                )
                .into_iter()
                .map(|activity| (activity.key(), activity))
                .collect::<BTreeMap<_, _>>();
                let sessions = page
                    .sessions
                    .into_iter()
                    .map(|session| {
                        let mut value = live_descriptor_value(&session, &doors)?;
                        let activity_key = (
                            session.locator.harness.as_str().to_string(),
                            session.locator.session_id.clone(),
                        );
                        if let Some(activity) = activities.get(&activity_key) {
                            // A Codex prompt (an approval, a choice) writes nothing to its rollout, which
                            // reads that turn as working; the live row reads its pane, and says so.
                            let mut activity = activity.clone();
                            let waiting = doors.all().iter().any(|live| {
                                live.address.harness == activity_key.0
                                    && live.address.session_id == activity_key.1
                                    && live.status == "waiting"
                            });
                            if waiting && activity.turn != crate::SessionTurnState::NeedsInput {
                                activity.turn = crate::SessionTurnState::NeedsInput;
                                activity.evidence.source = "pane_prompt".into();
                            }
                            let activity = &activity;
                            value["activity"] = serde_json::to_value(activity)
                                .map_err(|error| ServiceError::Operation(error.to_string()))?;
                            if let Some(status) = legacy_live_status(activity) {
                                value["live_status"] = json!(status);
                            }
                        }
                        Ok(value)
                    })
                    .collect::<std::result::Result<Vec<_>, ServiceError>>()?;
                let mut result = json!({"sessions": sessions, "next_cursor": page.next_cursor});
                // Preserve the metadata-only wire shape, but carry the catalog's
                // proof/counts when the caller explicitly requests preview search.
                if query.search_previews {
                    result["receipt"] = serde_json::to_value(page.receipt)
                        .map_err(|error| ServiceError::Operation(error.to_string()))?;
                }
                Ok(result)
            }
            "harness.v1.sessions.inbox" => inbox_call(decode::<InboxParams>(params)?),
            "harness.v1.sessions.usage" => {
                let params = decode::<UsageParams>(params)?;
                let since_ms = params
                    .since
                    .as_deref()
                    .map(|since| crate::token_spend::parse_since(since, now_epoch_ms() as i64))
                    .transpose()
                    .map_err(ServiceError::Operation)?;
                let session = load_session(&params.locator).map_err(operation)?;
                let usage = crate::token_spend::session_spend(
                    &session.meta.source,
                    session.meta.model.clone(),
                    &session.raw,
                    since_ms,
                );
                let mut answer = json!({"harness": params.locator.harness,
                    "session_id": session.meta.session_id, "usage": usage});
                if let Some(since_ms) = since_ms {
                    answer["since"] =
                        json!(supercode_interchange::sidecar::ms_to_rfc3339(since_ms));
                }
                Ok(answer)
            }
            "harness.v1.sessions.spend" => {
                let params = decode::<SpendParams>(params)?;
                if let Some(sort) = params.sort.as_deref().filter(|sort| !["cost", "tokens"].contains(sort)) {
                    return Err(ServiceError::InvalidParams(format!(
                        "sort `{sort}`: the spend reading sorts by `cost` or `tokens`"
                    )));
                }
                if let Some(harness) = params
                    .harnesses
                    .iter()
                    .find(|harness| ![HarnessId::CLAUDE_CODE, HarnessId::CODEX].contains(&harness.as_str()))
                {
                    return Err(ServiceError::InvalidParams(format!(
                        "harness `{harness}`: the spend reading reads `claude-code` and `codex`"
                    )));
                }
                let now_ms = now_epoch_ms() as i64;
                let since_ms = crate::token_spend::parse_since(&params.since, now_ms)
                    .map_err(ServiceError::Operation)?;
                Ok(crate::token_spend::machine_spend(
                    &params.homes,
                    since_ms,
                    now_ms,
                    &params.harnesses,
                    params.sort.as_deref().unwrap_or("cost"),
                    params.limit,
                ))
            }
            "harness.v1.sessions.tree" => {
                let params = if params.is_null() { json!({}) } else { params };
                let params = decode::<TreeParams>(params)?;
                match params.at {
                    Some(at) => {
                        let at_ms = crate::token_spend::parse_since(&at, now_epoch_ms() as i64)
                            .map_err(ServiceError::Operation)?;
                        crate::session_tree::recorded_tree(at_ms).ok_or_else(|| {
                            ServiceError::Operation(format!(
                                "nothing was recorded by {}",
                                supercode_interchange::sidecar::ms_to_rfc3339(at_ms)
                            ))
                        })
                    }
                    None => Ok(crate::session_tree::live_tree(&params.homes, true)),
                }
            }
            "harness.v1.sessions.recorded_config" => {
                let params = decode::<LocatorParams>(params)?;
                recorded_config(&params.locator).map_err(operation)
            }
            "harness.v1.sessions.launch_agent" => {
                let params = decode::<LocatorParams>(params)?;
                Ok(json!({"environment": crate::launch_agent::read(&params.locator)}))
            }
            "harness.v1.sessions.load" => {
                let params = decode::<LoadSessionParams>(params)?;
                load_session_door(params)
            }
            "harness.v1.sessions.follow" => {
                let params = decode::<LocatorParams>(params)?;
                let mut follower = self
                    .catalog
                    .follow_read_view(
                        &params.locator,
                        params.read_fidelity(),
                        params.include_subagents(),
                        params.tail_messages(),
                        params.max_message_chars(),
                        params.display_history(),
                    )
                    .map_err(operation)?;
                let initial = follower
                    .poll()
                    .map_err(operation)?
                    .map(|event| event.to_json());
                let subscription = format!("sub-{}", self.next_subscription);
                self.next_subscription += 1;
                self.followers.insert(subscription.clone(), follower);
                self.followed_sources.insert(
                    subscription.clone(),
                    FollowedSource {
                        harness: params.locator.harness.as_str().to_string(),
                        session_id: params.locator.session_id.clone(),
                        reported: None,
                    },
                );
                Ok(json!({"subscription": subscription, "initial": initial}))
            }
            "harness.v1.sessions.unfollow" => {
                let params = decode::<UnfollowParams>(params)?;
                self.followed_sources.remove(&params.subscription);
                Ok(json!({
                    "removed": self.followers.remove(&params.subscription).is_some()
                }))
            }
            "harness.v1.sessions.activity.unsubscribe" => {
                let params = decode::<UnfollowParams>(params)?;
                Ok(json!({
                    "removed": self.activity_subscriptions.remove(&params.subscription).is_some()
                }))
            }
            "harness.v1.sessions.index.subscribe" => {
                let query = decode::<DiscoveryQuery>(params)?;
                crate::session_index::validate_query(&query)
                    .map_err(ServiceError::InvalidParams)?;
                let homes = query.homes.clone();
                let (index, initial) = crate::session_index::SessionIndexSubscription::open(
                    query,
                    Arc::clone(&self.index_notifier),
                )
                .map_err(ServiceError::Operation)?;
                let doors = crate::mail_route::LiveSessions::read(&homes);
                let initial = initial
                    .iter()
                    .map(|descriptor| live_descriptor_value(descriptor, &doors))
                    .collect::<std::result::Result<Vec<_>, ServiceError>>()?;
                let subscription = format!("index-sub-{}", self.next_subscription);
                self.next_subscription += 1;
                self.index_subscriptions.insert(subscription.clone(), index);
                Ok(json!({
                    "subscription": subscription,
                    "revision": 1,
                    "initial": initial,
                }))
            }
            "harness.v1.sessions.index.resize" => {
                let params = decode::<IndexResizeParams>(params)?;
                crate::session_index::validate_limit(params.limit)
                    .map_err(ServiceError::InvalidParams)?;
                let index = self
                    .index_subscriptions
                    .get_mut(&params.subscription)
                    .ok_or_else(|| {
                        ServiceError::InvalidParams("unknown session index subscription".into())
                    })?;
                let prepared = index
                    .prepare_resize(params.limit)
                    .map_err(ServiceError::Operation)?;
                let doors = crate::mail_route::LiveSessions::read(index.homes());
                let initial = prepared
                    .page
                    .sessions
                    .iter()
                    .map(|descriptor| live_descriptor_value(descriptor, &doors))
                    .collect::<std::result::Result<Vec<_>, ServiceError>>()?;
                let response = json!({
                    "subscription": params.subscription,
                    "revision": prepared.revision,
                    "initial": initial,
                    "receipt": prepared.page.receipt,
                });
                index.commit_resize(prepared);
                Ok(response)
            }
            "harness.v1.sessions.index.unsubscribe" => {
                let params = decode::<UnfollowParams>(params)?;
                Ok(json!({
                    "removed": self.index_subscriptions.remove(&params.subscription).is_some()
                }))
            }
            "harness.v1.sessions.import" => {
                let params = decode::<ImportSessionParams>(params)?;
                let session = Session::load_str(&params.content, params.source_harness.into())
                    .map_err(operation)?;
                Ok(json!({"session": normalized_session_json(&session)}))
            }
            "harness.v1.sessions.export" | "harness.v1.sessions.translate" => {
                let params = decode::<ExportSessionParams>(params)?;
                let session = load_session(&params.locator).map_err(operation)?;
                let artifact = session_artifact(&params.locator, &session, params.target_harness)?;
                if method == "harness.v1.sessions.export"
                    && params.target_harness == TransferFormat::Hermes
                {
                    // UNI-18: write through Hermes's own door, never into its store
                    let imported = crate::hermes_import::import_into_hermes(&session, None)
                        .map_err(operation)?;
                    return Ok(json!({"artifact": artifact, "imported": imported}));
                }
                Ok(json!({"artifact": artifact}))
            }
            "harness.v1.sessions.reduce" => {
                let params = decode::<ReduceSessionParams>(params)?;
                self.reduce_session(params)
            }
            "harness.v1.sessions.branch" => {
                let params = decode::<BranchSessionParams>(params)?;
                let (session, cut) = session_before(
                    load_session(&params.locator).map_err(operation)?,
                    params.before_message_id.as_ref(),
                )?;
                let storage = params.locator.storage.path().display().to_string();
                let bootstrap_prompt = match cut {
                    Some(index) => format!(
                        "Continue as a new branch from {} session {}, forked before its message {index}. The frozen parent transcript is at {}. Read or load only what precedes message {index} for context, summarize the relevant state, then continue independently without mutating the parent session.",
                        params.locator.harness.as_str(), params.locator.session_id, storage
                    ),
                    None => format!(
                        "Continue as a new branch from {} session {}. The frozen parent transcript is at {}. Read or load that parent for context, summarize the relevant state, then continue independently without mutating the parent session.",
                        params.locator.harness.as_str(), params.locator.session_id, storage
                    ),
                };
                let artifact = params
                    .target_harness
                    .map(|target| session_artifact(&params.locator, &session, target))
                    .transpose()?;
                Ok(json!({
                    "parent": params.locator,
                    "session": normalized_session_json(&session),
                    "bootstrap_prompt": bootstrap_prompt,
                    "artifact": artifact,
                }))
            }
            "harness.v1.sessions.handoff" => {
                let params = decode::<HandoffSessionParams>(params)?;
                let (session, _) = session_before(
                    load_session(&params.locator).map_err(operation)?,
                    params.before_message_id.as_ref(),
                )?;
                let cwd = params
                    .cwd
                    .or_else(|| session.meta.cwd.clone())
                    .unwrap_or_else(|| PathBuf::from("."));
                let artifact = handoff_artifact(&params.locator, &session, params.target_harness)?;
                let target_session_id = artifact.session_id.as_deref().ok_or_else(|| {
                    ServiceError::Operation(
                        "handoff artifact omitted target session identity".into(),
                    )
                })?;
                let instructions =
                    handoff_instructions(params.target_harness, target_session_id, &cwd);
                Ok(json!({
                    "artifact": artifact,
                    "launch": instructions.launch,
                    "materialize": instructions.materialize,
                    "requires_materialization": instructions.requires_materialization,
                    "note": instructions.note,
                }))
            }
            "harness.v1.sessions.materialize" => {
                let params = decode::<MaterializeSessionParams>(params)?;
                // An artifact from another machine carries its whole source as a recovery file;
                // keeping its segments here lets a later write back to that format restore it
                // byte for byte on this machine too (docs/plans/portable-residue.md).
                for file in &params.artifact.files {
                    if file.role == "source_recovery"
                        && file.path == "recovery/source.supercode.jsonl"
                    {
                        if let Ok(source) = Session::from_native_str(&file.content) {
                            crate::residue_store::store_segments(&source);
                        }
                    }
                }
                let locator = crate::native_materialize::materialize_native_artifact(
                    params.artifact,
                    &params.cwd,
                    &params.homes,
                )
                .map_err(ServiceError::Operation)?;
                Ok(json!({"locator": locator}))
            }
            // ORCH-7 observed tier. Read-only: the handlers open the harness's
            // own job store (Claude Code's session JSONL, Hermes's and
            // OpenClaw's `cron/jobs.json`) and never write, fire, or schedule.
            "harness.v1.jobs.list" => {
                let query = decode::<crate::jobs::JobsQuery>(params)?;
                if let Some(harness) = query.harness.as_deref() {
                    refuse_harness_without_jobs(harness, "jobs.list")?;
                }
                let listing = crate::jobs::list_jobs(&query).map_err(operation)?;
                serde_json::to_value(listing)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            "harness.v1.jobs.get" => {
                let params = decode::<JobsGetParams>(params)?;
                refuse_harness_without_jobs(&params.harness, "jobs.get")?;
                match crate::jobs::get_job(&params.harness, &params.id, &params.homes)
                    .map_err(operation)?
                {
                    Some((job, source)) => Ok(json!({"job": job, "source": source})),
                    None => Err(ServiceError::Operation(format!(
                        "`{}` has no scheduled job `{}`",
                        params.harness, params.id
                    ))),
                }
            }
            // ORCH-18 controlled tier. Each verb translates to the HARNESS'S
            // OWN cron verb and runs it (`crate::jobs_control`); supercode
            // schedules nothing. The row returned is re-read from the
            // harness's store afterwards, and `ran` narrates the exact command
            // with any credential redacted.
            "harness.v1.jobs.create" => mutate_job(crate::jobs_control::JobVerb::Create, params),
            "harness.v1.jobs.update" => mutate_job(crate::jobs_control::JobVerb::Update, params),
            "harness.v1.jobs.pause" => mutate_job(crate::jobs_control::JobVerb::Pause, params),
            "harness.v1.jobs.resume" => mutate_job(crate::jobs_control::JobVerb::Resume, params),
            "harness.v1.jobs.run" => mutate_job(crate::jobs_control::JobVerb::Run, params),
            "harness.v1.jobs.delete" => mutate_job(crate::jobs_control::JobVerb::Delete, params),
            "harness.v1.jobs.notepad"
            | "harness.v1.jobs.notepad_set"
            | "harness.v1.jobs.notepad_delete" => {
                let request = decode::<crate::jobs_notepad::JobNotepadRequest>(params)?;
                refuse_harness_without_jobs(&request.harness, "jobs.notepad")?;
                let answer = match method {
                    "harness.v1.jobs.notepad_set" => crate::jobs_notepad::set(&request),
                    "harness.v1.jobs.notepad_delete" => crate::jobs_notepad::delete(&request),
                    _ => crate::jobs_notepad::read(&request),
                }
                .map_err(job_control_error)?;
                serde_json::to_value(answer)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            // ORCH-8 observed tier. Read-only: the handlers open the harness's
            // own run store (Hermes's `cron/executions.db`, OpenClaw's
            // `cron_run_logs`) and never claim, retry, or prune a fire.
            "harness.v1.runs.list" => {
                let query = decode::<crate::runs::RunsQuery>(params)?;
                if let Some(harness) = query.harness.as_deref() {
                    refuse_harness_without_runs(harness, "runs.list")?;
                }
                let listing = crate::runs::list_runs(&query).map_err(operation)?;
                serde_json::to_value(listing)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            "harness.v1.runs.get" => {
                let params = decode::<RunsGetParams>(params)?;
                refuse_harness_without_runs(&params.harness, "runs.get")?;
                match crate::runs::get_run(&params.harness, &params.id, &params.homes)
                    .map_err(operation)?
                {
                    Some((run, source)) => Ok(json!({"run": run, "source": source})),
                    None => Err(ServiceError::Operation(format!(
                        "`{}` has no run `{}`",
                        params.harness, params.id
                    ))),
                }
            }
            "harness.v1.sessions.resume_instructions" => {
                let params = decode::<ResumeInstructionsParams>(params)?;
                let session = load_session(&params.locator).map_err(operation)?;
                let cwd = params
                    .cwd
                    .or(session.meta.cwd)
                    .unwrap_or_else(|| PathBuf::from("."));
                let launch = resume_launch(
                    params.locator.harness.as_str(),
                    &params.locator.session_id,
                    &cwd,
                    params.policy,
                )?;
                Ok(json!({"launch": launch}))
            }
            _ => Err(ServiceError::MethodNotFound),
        }
    }

    fn reduce_session(
        &self,
        params: ReduceSessionParams,
    ) -> std::result::Result<Value, ServiceError> {
        let session = load_session(&params.locator).map_err(operation)?;
        if session.messages.is_empty() {
            return Err(ServiceError::InvalidParams(
                "cannot reduce an empty session".into(),
            ));
        }
        let keep_last = params.keep_last.clamp(1, 128);
        let policy = reduce::ReductionPolicy {
            clear_turns_older_than: Some(keep_last),
            ..Default::default()
        };
        let (view, log) =
            reduce::project_messages(&session.messages, &policy, &reduce::ReductionLog::default());
        if log.reductions.is_empty() {
            return Err(ServiceError::UnsupportedAction(format!(
                "session `{}` is already too small for a meaningful reversible reduction",
                params.locator.session_id
            )));
        }
        let source_tokens = supercode_runtime::estimate_view_tokens(&session.messages);
        let reduced_tokens = supercode_runtime::estimate_view_tokens(&view);
        if reduced_tokens >= source_tokens {
            return Err(ServiceError::UnsupportedAction(format!(
                "session `{}` has no token-reducing reversible projection",
                params.locator.session_id
            )));
        }

        let store_root = self
            .reduction_store_root
            .clone()
            .unwrap_or_else(default_reduction_store_root);
        let store = crate::SessionStore::open(&store_root).map_err(operation)?;
        let rescue_id = format!("rescue-{}", generated_session_id());
        let imported = session
            .imported_message_count
            .unwrap_or(session.messages.len())
            .min(session.messages.len());
        let sidecar_jsonl = session.to_native_jsonl_v2(&session.messages[imported..]);
        let view_jsonl = messages_jsonl(&view)?;
        let title = format!(
            "Reduced {} continuation from {}",
            params.target_harness.id(),
            params.locator.session_id
        );

        // Durability order is intentional: the full source of truth lands
        // before either object that can refer to it. A crash may leave an
        // unused sidecar, but can never leave a reduced view whose originals
        // were not durably written first.
        store
            .save_sidecar(&rescue_id, &sidecar_jsonl)
            .map_err(operation)?;
        store
            .save_reduction_log(&rescue_id, &log)
            .map_err(operation)?;
        store
            .save(&rescue_id, &title, &view_jsonl)
            .map_err(operation)?;

        let source_bytes = serde_json::to_vec(&session.messages)
            .map_err(|error| ServiceError::Operation(error.to_string()))?
            .len() as u64;
        let reduced_bytes = serde_json::to_vec(&view)
            .map_err(|error| ServiceError::Operation(error.to_string()))?
            .len() as u64;
        store
            .set_reduction_stats(
                &rescue_id,
                &title,
                source_bytes,
                reduced_bytes,
                log.reductions.len() as u32,
            )
            .map_err(operation)?;

        // The receipt is issued only after a real disk reload. This proves
        // the exact files another process will consume, not the convenient
        // in-memory values that produced them.
        let reloaded_sidecar = store
            .load_sidecar(&rescue_id)
            .map_err(operation)?
            .ok_or_else(|| ServiceError::Operation("reduction sidecar disappeared".into()))?;
        let reloaded_sidecar = Session::from_sidecar_str(&reloaded_sidecar).map_err(operation)?;
        let reloaded_log = store
            .load_reduction_log(&rescue_id)
            .map_err(operation)?
            .ok_or_else(|| ServiceError::Operation("reduction log disappeared".into()))?;
        let reloaded_view = parse_messages_jsonl(&store.load(&rescue_id).map_err(operation)?)?;
        reduce::verify_log(&reloaded_log, &reloaded_sidecar).map_err(operation)?;
        // `sc.reduction` is deliberately in-memory-only metadata: it must
        // never leak onto a provider-facing transcript. Reapplying the
        // durable log to the durable sidecar restores those ids. Comparing
        // its wire form with the transcript reloaded above proves that the
        // persisted view is exactly the deterministic projection before we
        // use the restamped form for inversion.
        let (restamped_view, restamped_log) =
            reduce::project_messages(&reloaded_sidecar.messages, &policy, &reloaded_log);
        if messages_jsonl(&restamped_view)? != messages_jsonl(&reloaded_view)? {
            return Err(ServiceError::Operation(
                "persisted reduction view does not match its durable log and sidecar".into(),
            ));
        }
        if restamped_log != reloaded_log {
            return Err(ServiceError::Operation(
                "reapplying the durable reduction log changed its identity".into(),
            ));
        }
        let inverted =
            reduce::invert(&restamped_view, &reloaded_log, &reloaded_sidecar).map_err(operation)?;
        if inverted != session.messages {
            return Err(ServiceError::Operation(
                "reduction inversion did not restore the source messages byte-exactly".into(),
            ));
        }

        let ratio = source_tokens as f64 / reduced_tokens.max(1) as f64;
        let sidecar_path = store.sidecar_path(&rescue_id);
        let reduction_log_path = store.reduction_log_path(&rescue_id).map_err(operation)?;
        let bootstrap_prompt = reduced_bootstrap_prompt(
            &params.locator,
            params.target_harness,
            &view_jsonl,
            &sidecar_path,
            &reduction_log_path,
        );
        let mut reduced_session = session.clone();
        reduced_session.meta.session_id = Some(rescue_id.clone());
        reduced_session.messages = view;

        Ok(json!({
            "session": normalized_session_json(&reduced_session),
            "bootstrap_prompt": bootstrap_prompt,
            "receipt": {
                "id": rescue_id,
                "sidecar_id": rescue_id,
                "source_harness": params.locator.harness,
                "target_harness": params.target_harness.id(),
                "source_tokens": source_tokens,
                "reduced_tokens": reduced_tokens,
                "ratio": ratio,
                "source_bytes": source_bytes,
                "reduced_bytes": reduced_bytes,
                "reductions": reloaded_log.reductions.len(),
                "sidecar_path": sidecar_path,
                "reduction_log_path": reduction_log_path,
                "verified": true,
                "reversible": true,
            }
        }))
    }

    /// Recognize the one request family whose waiting happens entirely
    /// outside this service's state, and hand a transport the half it can run
    /// off the task that owns the service.
    ///
    /// Opening a runtime is the only door here that waits on a foreign
    /// program: it spawns the harness's own binary and completes that
    /// program's protocol handshake, which takes as long as the program takes
    /// to answer. A transport that awaited the whole request inline would
    /// stop reading its own input for that whole time, so ONE slow launch
    /// would queue every later request on the same server — including reads
    /// like `sessions.discover` that touch no runtime at all. Splitting the
    /// request lets the transport spawn [`RuntimeOpen::open`] and keep
    /// reading, then pay only the short bookkeeping half
    /// ([`Self::register_open_runtime`]) when the runtime is up.
    ///
    /// `None` for every other method: those are answered by
    /// [`Self::handle_async`] as before.
    pub fn runtime_open(request: &Value) -> Option<RuntimeOpen> {
        if request.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
            return None;
        }
        let method = request.get("method").and_then(Value::as_str)?;
        if !RUNTIME_OPEN_METHODS.contains(&method) {
            return None;
        }
        Some(RuntimeOpen {
            id: request.get("id").cloned().unwrap_or(Value::Null),
            method: method.to_string(),
            params: request.get("params").cloned().unwrap_or_else(|| json!({})),
        })
    }

    /// Recognize a [`DETACHED_METHODS`] request and hand a transport the
    /// whole of it: the service-state half is read here and now, and what
    /// remains waits on a foreign program with nothing of this service's in
    /// hand.
    ///
    /// Same reason as [`Self::runtime_open`], different doors. Probing a
    /// harness starts it and completes its handshake; messaging a live
    /// session waits on its Claude relay's send; a conversation verb runs the
    /// harness's own CLI or calls its HTTP API. A transport that awaited any
    /// of those inline would stop reading its own input for that whole time,
    /// so one probe of an unhealthy harness would queue every later request
    /// on the same server.
    ///
    /// Unlike an opening runtime there is no bookkeeping half: the answer
    /// [`DetachedCall::run`] produces is the caller's complete response, so a
    /// transport writes it without coming back here.
    ///
    /// `None` for every other method — including the LIVE `sessions.new` /
    /// `sessions.reset` door and `runtimes.close`, which wait on a runtime
    /// connection this service owns and so are split off by
    /// [`Self::detach_runtime`] instead.
    pub fn detach(&self, request: &Value) -> Option<DetachedCall> {
        if request.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
            return None;
        }
        let method = request.get("method").and_then(Value::as_str)?;
        if !DETACHED_METHODS.contains(&method) {
            return None;
        }
        let id = request.get("id").cloned().unwrap_or(Value::Null);
        let params = request.get("params").cloned().unwrap_or_else(|| json!({}));
        let work = match method {
            "harness.v1.harnesses.list" | "harness.v1.harnesses.probe" => self
                .inventory_work(method, params)
                .map(DetachedWork::Inventory),
            "harness.v1.sessions.message" => {
                decode::<MessageSessionParams>(params).map(DetachedWork::Message)
            }
            "harness.v1.sessions.load" => {
                decode::<LoadSessionParams>(params).map(DetachedWork::Load)
            }
            _ => {
                let verb = match method {
                    "harness.v1.sessions.new" => crate::SessionVerb::New,
                    "harness.v1.sessions.reset" => crate::SessionVerb::Reset,
                    "harness.v1.sessions.archive" => crate::SessionVerb::Archive,
                    _ => crate::SessionVerb::Delete,
                };
                match decode::<crate::SessionMutation>(params) {
                    Ok(mutation) => {
                        match crate::sessions_control::door(&mutation.harness, verb) {
                            // The live door needs the open runtime connection
                            // this service owns; it stays inline.
                            Ok(crate::SessionDoor::Live(_)) => return None,
                            Ok(_) => Ok(DetachedWork::SessionMutation { verb, mutation }),
                            Err(error) => Err(session_control_error(error)),
                        }
                    }
                    Err(error) => Err(error),
                }
            }
        };
        Some(DetachedCall {
            id,
            method: method.to_string(),
            work: work.map(Work::Free),
        })
    }

    /// Recognize the two doors that wait on a runtime THIS SERVICE OWNS, and
    /// hand a transport the whole of each by lending the connection out.
    ///
    /// `runtimes.close` surrenders its runtime for good; the LIVE
    /// `sessions.new` / `sessions.reset` door borrows one for the length of
    /// the slash command and gives it back through
    /// [`Self::finish_detached`]. Both are bounded by
    /// [`RUNTIME_CONTROL_DEADLINE`], and a wedged runtime spends all of it —
    /// which is exactly as long as a transport that awaited them inline would
    /// stop reading its own input.
    ///
    /// `None` for every other method, and for the `sessions.new` /
    /// `sessions.reset` doors that are not live: [`Self::detach`] owns those.
    pub fn detach_runtime(&mut self, request: &Value) -> Option<DetachedCall> {
        if request.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
            return None;
        }
        let method = request.get("method").and_then(Value::as_str)?;
        let id = request.get("id").cloned().unwrap_or(Value::Null);
        let params = request.get("params").cloned().unwrap_or_else(|| json!({}));
        let work = match method {
            "harness.v1.runtimes.close" => decode::<RuntimeCloseParams>(params)
                .and_then(|params| self.surrender_runtime_of(&params))
                .map(|(runtime, process_group)| {
                    Work::Runtime(RuntimeWork::Close {
                        runtime,
                        process_group,
                    })
                }),
            "harness.v1.sessions.new" | "harness.v1.sessions.reset" => {
                let verb = if method == "harness.v1.sessions.new" {
                    crate::SessionVerb::New
                } else {
                    crate::SessionVerb::Reset
                };
                let mutation = decode::<crate::SessionMutation>(params).ok()?;
                // Everything but the live door — including a refusal and a
                // request naming no connection — is `detach`'s or
                // `handle_async`'s to answer.
                let Ok(crate::SessionDoor::Live(command)) =
                    crate::sessions_control::door(&mutation.harness, verb)
                else {
                    return None;
                };
                let connection = mutation
                    .connection
                    .clone()
                    .filter(|value| !value.trim().is_empty())?;
                self.lend_runtime(&connection).map(|runtime| {
                    let session = live_session_name(runtime.as_ref(), &mutation);
                    Work::Runtime(RuntimeWork::LiveCommand {
                        connection,
                        runtime,
                        verb,
                        mutation,
                        command,
                        session,
                    })
                })
            }
            _ => return None,
        };
        Some(DetachedCall {
            id,
            method: method.to_string(),
            work,
        })
    }

    /// Take back whatever a detached call borrowed and hand over the caller's
    /// response. Every answer from [`DetachedCall::run`] comes through here,
    /// so a lent-out connection is back in the service before the response
    /// that used it is written.
    pub fn finish_detached(&mut self, answer: DetachedAnswer) -> Value {
        let DetachedAnswer { response, returned } = answer;
        if let Some(ReturnedRuntime {
            connection,
            runtime,
        }) = returned
        {
            self.runtimes_in_flight.remove(&connection);
            self.runtimes.insert(connection, runtime);
        }
        response
    }

    /// Answer a request split out by [`Self::runtime_open`] and already
    /// awaited by [`RuntimeOpen::open`]: register the runtime this service now
    /// owns and build its JSON-RPC response.
    pub async fn finish_runtime_open(&mut self, opened: OpenedRuntime) -> Value {
        let OpenedRuntime { id, outcome } = opened;
        let result = match outcome {
            Ok(open) => self.register_open_runtime(open).await,
            Err(error) => Err(error),
        };
        service_response(id, result)
    }

    /// Take ownership of an opened runtime.
    async fn register_open_runtime(
        &mut self,
        open: OpenRuntime,
    ) -> std::result::Result<Value, ServiceError> {
        match open {
            OpenRuntime::Hosted {
                runtime,
                capabilities,
                workspace,
                fresh,
            } => {
                self.insert_hosted_runtime(runtime, capabilities, workspace, fresh)
                    .await
            }
            OpenRuntime::Joined { runtime } => self.insert_runtime(runtime),
        }
    }

    async fn runtime_call(
        &mut self,
        method: &str,
        params: Value,
    ) -> std::result::Result<Value, ServiceError> {
        match method {
            "harness.v1.runtimes.capabilities" => {
                let params = decode::<RuntimeBackendParams>(params)?;
                // A live endpoint is attached through the hosted-runtime backend whatever harness it
                // hosts (runtimes.attach_existing routes it there, and verifies its receipt), so it is
                // answered with that backend's capabilities, not the harness adapter's own.
                #[cfg(feature = "adapter-api")]
                if params
                    .base_url
                    .as_deref()
                    .is_some_and(|value| LiveRuntimeEndpoint::parse(value).is_ok())
                {
                    return Ok(json!({
                        "harness": &params.harness,
                        "capabilities": SupercodeHttpRuntimeBackend::live_capabilities(),
                    }));
                }
                let backend = runtime_backend(&params)?;
                Ok(json!({
                    "harness": backend.harness(),
                    "capabilities": backend.capabilities(),
                }))
            }
            method if RUNTIME_OPEN_METHODS.contains(&method) => {
                self.register_open_runtime(open_runtime(method, params).await?)
                    .await
            }
            "harness.v1.runtimes.send_input" => {
                let params = decode::<RuntimeInputParams>(params)?;
                let image_urls = validate_runtime_image_urls(params.image_urls)?;
                let runtime = self.runtime_mut(&params.connection)?;
                let turn_id = within_control_deadline(
                    method,
                    runtime.send_input(RuntimeInput {
                        text: params.text,
                        image_urls,
                    }),
                )
                .await?
                .map_err(operation)?;
                Ok(json!({"turn_id": turn_id}))
            }
            "harness.v1.runtimes.interrupt" => {
                let params = decode::<RuntimeConnectionParams>(params)?;
                within_control_deadline(method, self.runtime_mut(&params.connection)?.interrupt())
                    .await?
                    .map_err(operation)?;
                Ok(json!({}))
            }
            "harness.v1.runtimes.steer" => {
                let params = decode::<RuntimeInputParams>(params)?;
                if !params.image_urls.is_empty() {
                    return Err(ServiceError::InvalidParams(
                        "runtime steering accepts text only".into(),
                    ));
                }
                let text = params.text.trim();
                if text.is_empty() || text.chars().count() > 50_000 {
                    return Err(ServiceError::InvalidParams(
                        "runtime steering requires 1 to 50,000 text characters".into(),
                    ));
                }
                within_control_deadline(
                    method,
                    self.runtime_mut(&params.connection)?
                        .steer(text.to_string()),
                )
                .await?
                .map_err(operation)?;
                Ok(json!({}))
            }
            "harness.v1.runtimes.respond" => {
                let params = decode::<RuntimeRespondParams>(params)?;
                let request_id = params.request_id.clone();
                within_control_deadline(
                    method,
                    self.runtime_mut(&params.connection)?
                        .respond(params.request_id, params.response),
                )
                .await?
                .map_err(operation)?;
                // ORCH-9: an answered request is no longer waiting for one.
                self.approvals.answered(&params.connection, &request_id);
                Ok(json!({}))
            }
            "harness.v1.runtimes.acquire_control" => {
                let params = decode::<RuntimeConnectionParams>(params)?;
                let snapshot = within_control_deadline(
                    method,
                    self.runtime_mut(&params.connection)?.acquire_control(),
                )
                .await?
                .map_err(operation)?;
                serde_json::to_value(snapshot)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            "harness.v1.runtimes.heartbeat" => {
                let params = decode::<RuntimeConnectionParams>(params)?;
                let snapshot = within_control_deadline(
                    method,
                    self.runtime_mut(&params.connection)?.heartbeat(),
                )
                .await?
                .map_err(operation)?;
                serde_json::to_value(snapshot)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            "harness.v1.runtimes.detach" => {
                let params = decode::<RuntimeConnectionParams>(params)?;
                let snapshot =
                    within_control_deadline(method, self.runtime_mut(&params.connection)?.detach())
                        .await?
                        .map_err(operation)?;
                serde_json::to_value(snapshot)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            "harness.v1.runtimes.terminal_instructions" => {
                let params = decode::<RuntimeConnectionParams>(params)?;
                let launch = self
                    .terminal_launches
                    .get(&params.connection)
                    .ok_or_else(|| {
                        ServiceError::Operation(
                            "this runtime is not hosted for terminal attachment".into(),
                        )
                    })?;
                Ok(json!({"launch":launch}))
            }
            "harness.v1.runtimes.close" => {
                let params = decode::<RuntimeCloseParams>(params)?;
                let (runtime, process_group) = self.surrender_runtime_of(&params)?;
                close_runtime(runtime, process_group).await
            }
            _ => Err(ServiceError::MethodNotFound),
        }
    }

    /// Deliver one message into a session that is running right now.
    #[cfg(feature = "adapter-api")]
    async fn message_call(&self, params: Value) -> std::result::Result<Value, ServiceError> {
        let params = decode::<MessageSessionParams>(params)?;
        Ok(message_live_session(&params).await)
    }

    #[cfg(feature = "adapter-api")]
    fn harness_settings_call(
        &self,
        method: &str,
        params: Value,
    ) -> std::result::Result<Value, ServiceError> {
        let homes = crate::HarnessHomes::default();
        match method {
            "harness.v1.harnesses.settings" => {
                let params = decode::<HarnessSettingsParams>(params)?;
                let report = crate::inspect_harness_interop_settings(&homes, &params.harness)
                    .map_err(|error| ServiceError::Operation(error.to_string()))?;
                serde_json::to_value(report)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            "harness.v1.harnesses.configure" => {
                let params = decode::<ConfigureHarnessParams>(params)?;
                let report = crate::configure_harness_interop_settings(
                    &homes,
                    &params.harness,
                    &params.changes,
                    params.expected_revision.as_deref(),
                )
                .map_err(|error| ServiceError::Operation(error.to_string()))?;
                serde_json::to_value(report)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            _ => Err(ServiceError::MethodNotFound),
        }
    }

    fn insert_runtime(
        &mut self,
        runtime: Box<dyn RuntimeConnection>,
    ) -> std::result::Result<Value, ServiceError> {
        let connection = format!("runtime-{}", self.next_runtime);
        self.next_runtime += 1;
        let handle = runtime.handle().clone();
        self.runtime_sequences
            .entry(handle.runtime_id.clone())
            .or_insert(0);
        self.runtimes.insert(connection.clone(), runtime);
        Ok(json!({"connection": connection, "handle": handle}))
    }

    #[cfg(feature = "adapter-api")]
    async fn insert_hosted_runtime(
        &mut self,
        runtime: Box<dyn RuntimeConnection>,
        capabilities: crate::RuntimeCapabilities,
        workspace: PathBuf,
        fresh: bool,
    ) -> std::result::Result<Value, ServiceError> {
        let (host, connection) = HostedHarnessRuntime::spawn(runtime, capabilities, fresh);
        let token: std::sync::Arc<str> = crate::server::generate_token().into();
        let server = crate::server::run_frontend_http(
            host.clone(),
            host.frontend_sender(),
            "127.0.0.1:0",
            token.clone(),
            connection.handle().runtime_id.clone(),
        )
        .await
        .map_err(|error| ServiceError::Operation(error.to_string()))?;
        let source = LiveRuntimeSource {
            harness: connection.handle().harness.as_str().to_string(),
            session_id: connection.handle().runtime_id.clone(),
            workspace: workspace.clone(),
        };
        let registration = register_live_runtime_with_metadata(
            connection.handle().runtime_id.clone(),
            source.clone(),
            format!("http://{}", server.address()),
            token.to_string(),
            crate::LiveRuntimeMetadata {
                child_pid: match &connection.handle().endpoint {
                    crate::RuntimeEndpoint::LocalProcess { pid, command, .. }
                        if connection.handle().harness.as_str() != "claude-code"
                            || command.windows(2).any(|args| {
                                args[0] == "--name"
                                    && args[1].starts_with(crate::claude_relay::RELAY_NAME_PREFIX)
                            }) =>
                    {
                        *pid
                    }
                    _ => None,
                },
                endpoint_capabilities: vec!["http".into(), "acp".into()],
                ..Default::default()
            },
        )
        .map_err(|error| ServiceError::Operation(error.to_string()))?;
        let endpoint = registration.endpoint().to_string();
        let launch = StructuredLaunch {
            cwd: workspace,
            // Pin attachment to the executable hosting this runtime. A bare
            // `supercode` could resolve to an older global install whose CLI
            // does not understand the receipt it is being asked to open.
            program: std::env::current_exe()
                .ok()
                .map(|path| path.to_string_lossy().into_owned())
                .unwrap_or_else(|| "supercode".into()),
            arguments: vec![
                "open".into(),
                endpoint,
                "--harness".into(),
                source.harness,
                "--session".into(),
                source.session_id,
            ],
            env: BTreeMap::new(),
        };
        host.retain_registration(registration);
        let lease = HostedRuntimeLease {
            connection,
            _host: host,
            _server: server,
        };
        let opened = self.insert_runtime(Box::new(lease))?;
        let connection_id = opened["connection"]
            .as_str()
            .expect("insert_runtime returns a connection id")
            .to_string();
        self.terminal_launches.insert(connection_id, launch);
        Ok(opened)
    }

    #[cfg(not(feature = "adapter-api"))]
    async fn insert_hosted_runtime(
        &mut self,
        runtime: Box<dyn RuntimeConnection>,
        _capabilities: crate::RuntimeCapabilities,
        _workspace: PathBuf,
        _fresh: bool,
    ) -> std::result::Result<Value, ServiceError> {
        self.insert_runtime(runtime)
    }

    fn runtime_mut(
        &mut self,
        connection: &str,
    ) -> std::result::Result<&mut Box<dyn RuntimeConnection>, ServiceError> {
        if self.runtimes_in_flight.contains(connection) {
            return Err(self.lent_out(connection));
        }
        self.runtimes.get_mut(connection).ok_or_else(|| {
            ServiceError::InvalidParams(format!("unknown runtime connection `{connection}`"))
        })
    }

    /// What a caller is told about a connection that is out on a detached
    /// call. It is not gone and it is not free: it is mid-call, which is the
    /// same answer the runtime itself gives a second turn.
    fn lent_out(&self, connection: &str) -> ServiceError {
        ServiceError::Operation(format!(
            "runtime connection `{connection}`: a harness turn is already in progress"
        ))
    }

    /// Take a runtime OUT of the service for the duration of one detached
    /// call, leaving its name marked as lent out.
    fn lend_runtime(
        &mut self,
        connection: &str,
    ) -> std::result::Result<Box<dyn RuntimeConnection>, ServiceError> {
        if self.runtimes_in_flight.contains(connection) {
            return Err(self.lent_out(connection));
        }
        let runtime = self.runtimes.remove(connection).ok_or_else(|| {
            ServiceError::InvalidParams(format!("unknown runtime connection `{connection}`"))
        })?;
        self.runtimes_in_flight.insert(connection.to_string());
        Ok(runtime)
    }

    /// Surrender a runtime for good: the connection and everything the
    /// service hung off it are gone before its teardown is even attempted.
    ///
    /// `close` is what a caller reaches for when a runtime has stopped
    /// answering, and a runtime that has stopped answering is exactly the one
    /// whose graceful close cannot complete: a hosted runtime's own loop
    /// parks on the call the runtime never answered, so it never dequeues the
    /// shutdown either. Keeping the entry until teardown succeeded made a
    /// wedged runtime permanent — every later call on that connection, and
    /// every new turn, answered "a harness turn is already in progress" with
    /// no way to take the connection back.
    /// [`Self::surrender_runtime`] for a `runtimes.close` call, refusing a connection that no longer holds
    /// the runtime the caller named.
    fn surrender_runtime_of(
        &mut self,
        params: &RuntimeCloseParams,
    ) -> std::result::Result<(Box<dyn RuntimeConnection>, Option<u32>), ServiceError> {
        if params.connection.is_empty() {
            let expected = params.runtime_id.as_deref().ok_or_else(|| {
                ServiceError::InvalidParams("close requires connection or runtime_id".into())
            })?;
            let connection = self
                .runtimes
                .iter()
                .find(|(_, runtime)| runtime.handle().runtime_id == expected)
                .map(|(connection, _)| connection.clone())
                .ok_or_else(|| {
                    ServiceError::InvalidParams(format!("unknown runtime `{expected}`"))
                })?;
            return self.surrender_runtime(&connection);
        }
        if let Some(expected) = params.runtime_id.as_deref() {
            let held = self
                .runtimes
                .get(&params.connection)
                .map(|runtime| runtime.handle().runtime_id.clone());
            if held.as_deref() != Some(expected) {
                return Err(ServiceError::InvalidParams(format!(
                    "runtime connection `{}` does not hold runtime `{expected}`",
                    params.connection
                )));
            }
        }
        self.surrender_runtime(&params.connection)
    }

    fn surrender_runtime(
        &mut self,
        connection: &str,
    ) -> std::result::Result<(Box<dyn RuntimeConnection>, Option<u32>), ServiceError> {
        if self.runtimes_in_flight.contains(connection) {
            return Err(self.lent_out(connection));
        }
        let runtime = self.runtimes.remove(connection).ok_or_else(|| {
            ServiceError::InvalidParams(format!("unknown runtime connection `{connection}`"))
        })?;
        let process_group = runtime_process_group(runtime.handle());
        let runtime_id = runtime.handle().runtime_id.clone();
        self.terminal_launches.remove(connection);
        self.runtime_sequences.remove(&runtime_id);
        self.approvals.forget(connection);
        Ok((runtime, process_group))
    }

    /// SIGKILL the process group of every runtime this service owns, without
    /// waiting on any of them.
    ///
    /// A host leaving for good calls this BEFORE dropping the service. The
    /// handle this service holds is not the runtime's connection: a hosted
    /// runtime's real transport lives in the task driving it, so neither
    /// exiting the process nor dropping these handles reaches the harness
    /// process — while dropping them does remove each runtime's live-runtime
    /// receipt. Signalling first is what keeps a removed receipt from
    /// advertising a harness that is still running.
    pub fn kill_all_runtime_groups(&self) -> usize {
        self.runtimes
            .values()
            .filter(|runtime| kill_runtime_process_group(runtime_process_group(runtime.handle())))
            .count()
    }

    /// ORCH-19: run one conversation-lifecycle verb through the harness's own
    /// door.
    ///
    /// Two doors, one shape. A CLI / HTTP / own-store door is self-contained
    /// in [`crate::sessions_control`]. A LIVE door (Hermes's and OpenClaw's
    /// `/new` and `/reset`, which are slash commands their gateway interprets
    /// INSIDE a session) is performed here, because only the service owns the
    /// open runtime connection — the command is typed through the very same
    /// `send_input` path a human's message takes, so supercode invents no
    /// private channel.
    async fn mutate_session(
        &mut self,
        verb: crate::SessionVerb,
        params: Value,
    ) -> std::result::Result<Value, ServiceError> {
        let mutation = decode::<crate::SessionMutation>(params)?;
        let door = crate::sessions_control::door(&mutation.harness, verb)
            .map_err(session_control_error)?;
        let outcome = match door {
            // The live door types the slash command through an open hosted
            // runtime, which only exists with the `adapter-api` feature; the
            // CLI / HTTP / own-store doors below need nothing extra.
            #[cfg(not(feature = "adapter-api"))]
            crate::SessionDoor::Live(command) => {
                return Err(ServiceError::Operation(format!(
                    "`{}` performs `sessions.{}` by typing `{command}` into a live driven \
                     session, which needs this build's `adapter-api` feature",
                    mutation.harness,
                    verb.as_str()
                )));
            }
            #[cfg(feature = "adapter-api")]
            crate::SessionDoor::Live(command) => {
                let connection = mutation
                    .connection
                    .clone()
                    .filter(|value| !value.trim().is_empty())
                    .ok_or_else(|| {
                        ServiceError::InvalidParams(format!(
                            "`{}` performs `sessions.{}` by typing `{command}` into a live \
                             driven session: pass the `connection` of an open runtime \
                             (`harness.v1.runtimes.start`)",
                            mutation.harness,
                            verb.as_str()
                        ))
                    })?;
                let runtime = self.runtime_mut(&connection)?;
                let session = live_session_name(runtime.as_ref(), &mutation);
                // Typing into a live session is a control call on an open
                // runtime, and a wedged runtime never accepts one, so it is
                // bounded exactly like the other control verbs. A transport
                // with a loop of its own lends the connection out instead of
                // waiting here: see [`Self::detach_runtime`].
                return type_live_command(runtime.as_mut(), verb, &mutation, command, session)
                    .await;
            }
            _ => run_session_mutation(verb, &mutation).await?,
        };
        serde_json::to_value(outcome).map_err(|error| ServiceError::Operation(error.to_string()))
    }

    /// Answer an inventory request whole, for callers that have nowhere to
    /// put the waiting half. A transport with a loop of its own splits it
    /// instead: see [`Self::detach`].
    async fn inventory_call(
        &self,
        method: &str,
        params: Value,
    ) -> std::result::Result<Value, ServiceError> {
        run_inventory(self.inventory_work(method, params)?).await
    }

    /// The half of an inventory request that reads this service's state:
    /// resolve the selection and count the persisted sessions each row
    /// reports. What remains — finding executables, asking them their
    /// version, and (at `probe: handshake`) starting each harness and
    /// completing its protocol handshake — touches no service state at all.
    fn inventory_work(
        &self,
        method: &str,
        params: Value,
    ) -> std::result::Result<InventoryWork, ServiceError> {
        let mut params = decode::<HarnessInventoryParams>(params)?;
        if method == "harness.v1.harnesses.probe" {
            let harness = params.harness.take().ok_or_else(|| {
                ServiceError::InvalidParams("harnesses.probe requires `harness`".into())
            })?;
            params.harnesses = vec![harness];
        }
        let selected = params
            .harnesses
            .iter()
            .map(HarnessId::as_str)
            .collect::<std::collections::BTreeSet<_>>();
        let supported = harness_support_registry()
            .harnesses
            .into_iter()
            .filter(|descriptor| selected.is_empty() || selected.contains(descriptor.id.as_str()))
            .collect::<Vec<_>>();
        if !params.harnesses.is_empty() && supported.len() != selected.len() {
            let known = supported
                .iter()
                .map(|harness| harness.id.as_str())
                .collect::<std::collections::BTreeSet<_>>();
            let missing = params
                .harnesses
                .iter()
                .filter(|id| !known.contains(id.as_str()))
                .map(HarnessId::as_str)
                .collect::<Vec<_>>();
            return Err(ServiceError::InvalidParams(format!(
                "unknown harness(es): {}",
                missing.join(", ")
            )));
        }
        let global_counts = params
            .include_sessions
            .then(|| self.session_counts(None, &params.harnesses));
        let workspace_counts = params
            .include_sessions
            .then(|| {
                params
                    .workspace
                    .as_deref()
                    .map(|workspace| self.session_counts(Some(workspace), &params.harnesses))
            })
            .flatten();
        Ok(InventoryWork {
            params,
            supported,
            global_counts,
            workspace_counts,
        })
    }

    #[cfg(feature = "adapter-api")]
    async fn harness_authentication_call(
        &self,
        method: &str,
        params: Value,
    ) -> std::result::Result<Value, ServiceError> {
        match method {
            "harness.v1.harnesses.auth.methods" | "harness.v1.harnesses.auth.verify" => {
                let params = decode::<HarnessAuthenticationParams>(params)?;
                serde_json::to_value(crate::inspect_harness_authentication(&params.harness).await)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            "harness.v1.harnesses.auth.begin" => {
                let params = decode::<BeginHarnessAuthenticationParams>(params)?;
                let cwd = params
                    .cwd
                    .or_else(|| std::env::current_dir().ok())
                    .unwrap_or_else(|| PathBuf::from("."));
                let plan = crate::harness_authentication_plan(
                    &params.harness,
                    params.environment,
                    params.method,
                    &cwd,
                )
                .map_err(|error| ServiceError::UnsupportedAction(error.to_string()))?;
                serde_json::to_value(plan)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
            _ => Err(ServiceError::MethodNotFound),
        }
    }

    fn session_counts(
        &self,
        workspace: Option<&Path>,
        harnesses: &[HarnessId],
    ) -> BTreeMap<String, usize> {
        let mut counts = BTreeMap::new();
        for session in self
            .catalog
            .discover(&DiscoveryQuery {
                workspace: workspace.map(Path::to_path_buf),
                harnesses: harnesses.to_vec(),
                ..DiscoveryQuery::default()
            })
            .unwrap_or_default()
        {
            *counts
                .entry(session.locator.harness.as_str().to_string())
                .or_insert(0) += 1;
        }
        counts
    }
}

#[async_trait::async_trait]
impl SdkService for HarnessSessionService {
    fn capabilities(&self) -> SdkCapabilities {
        SdkCapabilities::default()
    }

    async fn execute(&mut self, request: SdkRequest) -> Result<Value, SdkError> {
        if request.operation == SdkOperation::Events {
            let events = self
                .poll_sdk_events()
                .await
                .into_iter()
                .map(|(_, event)| event)
                .collect::<Vec<_>>();
            return serde_json::to_value(events).map_err(|error| {
                SdkError::new(
                    SdkErrorCode::Execution,
                    request.operation,
                    error.to_string(),
                )
            });
        }
        if self.runtimes.is_empty()
            && matches!(
                request.operation,
                SdkOperation::Input
                    | SdkOperation::Interrupt
                    | SdkOperation::Steer
                    | SdkOperation::Respond
                    | SdkOperation::Close
            )
        {
            return Err(SdkError::unsupported(request.operation));
        }
        let method = request
            .operation
            .method()
            .ok_or_else(|| SdkError::unsupported(request.operation))?;
        let result = match request.operation {
            SdkOperation::Discover
            | SdkOperation::Load
            | SdkOperation::Export
            | SdkOperation::ProfilesList
            | SdkOperation::ProfilesGet
            | SdkOperation::ProfilesCreate
            | SdkOperation::ProfilesDelete
            | SdkOperation::SkillsList
            | SdkOperation::SkillsInstall
            | SdkOperation::SkillsRemove
            | SdkOperation::ChannelsList
            | SdkOperation::RoutesList
            | SdkOperation::TriggersList
            | SdkOperation::ChannelsStatus
            | SdkOperation::MemoryShow
            | SdkOperation::MemorySearch
            | SdkOperation::JobsList
            | SdkOperation::JobsGet
            | SdkOperation::JobsCreate
            | SdkOperation::JobsUpdate
            | SdkOperation::JobsPause
            | SdkOperation::JobsResume
            | SdkOperation::JobsRun
            | SdkOperation::JobsDelete
            | SdkOperation::JobsNotepad
            | SdkOperation::JobsNotepadSet
            | SdkOperation::JobsNotepadDelete
            | SdkOperation::RunsList
            | SdkOperation::RunsGet
            | SdkOperation::ApprovalsList
            | SdkOperation::OrchestrationLoad
            | SdkOperation::OrchestrationSave
            | SdkOperation::OrchestrationCompile
            | SdkOperation::OrchestrationDecompile
            | SdkOperation::OrchestrationImport
            | SdkOperation::OrchestrationExport
            | SdkOperation::WorkflowLoad => self.call(method, request.params),
            // ORCH-20: answering needs the live connection, so it takes the
            // async door and ends in `harness.v1.runtimes.respond`.
            SdkOperation::ApprovalsResolve => self.approvals_resolve(request.params).await,
            SdkOperation::Start
            | SdkOperation::Resume
            | SdkOperation::Input
            | SdkOperation::Interrupt
            | SdkOperation::Steer
            | SdkOperation::Respond
            | SdkOperation::Close => self.runtime_call(method, request.params).await,
            // ORCH-19 controlled tier. Every verb goes through the HARNESS'S
            // OWN door — its CLI, its HTTP API, or its slash command typed
            // into a live driven session — and returns the row re-read from
            // the harness's store afterwards.
            SdkOperation::SessionsNew => {
                self.mutate_session(crate::SessionVerb::New, request.params)
                    .await
            }
            SdkOperation::SessionsReset => {
                self.mutate_session(crate::SessionVerb::Reset, request.params)
                    .await
            }
            SdkOperation::SessionsArchive => {
                self.mutate_session(crate::SessionVerb::Archive, request.params)
                    .await
            }
            SdkOperation::SessionsDelete => {
                self.mutate_session(crate::SessionVerb::Delete, request.params)
                    .await
            }
            SdkOperation::Events => unreachable!("handled before method dispatch"),
        };
        result.map_err(|error| sdk_error(request.operation, error))
    }

    async fn events(&mut self) -> Result<Vec<SdkRuntimeEvent>, SdkError> {
        Ok(self
            .poll_sdk_events()
            .await
            .into_iter()
            .map(|(_, event)| event)
            .collect())
    }
}

#[cfg(feature = "adapter-api")]
struct HostedRuntimeLease {
    connection: HostedHarnessConnection,
    _host: std::sync::Arc<HostedHarnessRuntime>,
    _server: crate::server::FrontendHttpServer,
}

#[async_trait::async_trait]
#[cfg(feature = "adapter-api")]
impl RuntimeConnection for HostedRuntimeLease {
    fn handle(&self) -> &crate::RuntimeHandle {
        self.connection.handle()
    }

    async fn send_input(&mut self, input: RuntimeInput) -> crate::Result<Option<String>> {
        self.connection.send_input(input).await
    }

    async fn next_event(&mut self) -> crate::Result<Option<crate::HarnessEvent>> {
        self.connection.next_event().await
    }

    async fn interrupt(&mut self) -> crate::Result<()> {
        self.connection.interrupt().await
    }

    // the lease must forward every verb its capabilities advertise; without
    // this, steer fell to the trait default and refused a turn it claimed
    async fn steer(&mut self, text: String) -> crate::Result<()> {
        self.connection.steer(text).await
    }

    async fn respond(&mut self, request_id: Value, response: Value) -> crate::Result<()> {
        self.connection.respond(request_id, response).await
    }

    async fn close(&mut self) -> crate::Result<()> {
        self.connection.close().await
    }
}

/// One inventory request's waiting half, already separated from the service
/// state it reads. See [`HarnessSessionService::inventory_work`].
struct InventoryWork {
    params: HarnessInventoryParams,
    supported: Vec<crate::HarnessSupportDescriptor>,
    global_counts: Option<BTreeMap<String, usize>>,
    workspace_counts: Option<BTreeMap<String, usize>>,
}

/// Perform one conversation-lifecycle verb through a door that is
/// self-contained in [`crate::sessions_control`]: the harness's own CLI, its
/// HTTP API, the orchestrator daemon's socket, or supercode's own store.
/// Touches no service state, so this runs on any task. The LIVE door is not
/// here — it types its slash command through a runtime connection the service
/// owns, and is performed by [`HarnessSessionService::mutate_session`].
async fn run_session_mutation(
    verb: crate::SessionVerb,
    mutation: &crate::SessionMutation,
) -> std::result::Result<crate::SessionMutationOutcome, ServiceError> {
    // Only the HTTP door actually awaits anything. The CLI, store and daemon
    // doors run the harness's own program, or its store, with calls that
    // block the calling THREAD from start to finish — a future that never
    // yields, which no timeout around it can interrupt and which would hold a
    // runtime worker for as long as the harness takes. They go to a blocking
    // task, where blocking is what the thread is for.
    let door =
        crate::sessions_control::door(&mutation.harness, verb).map_err(session_control_error)?;
    if let crate::SessionDoor::Http = door {
        return crate::sessions_control::mutate(verb, mutation)
            .await
            .map_err(session_control_error);
    }
    let mutation = mutation.clone();
    tokio::task::spawn_blocking(move || crate::sessions_control::mutate_blocking(verb, &mutation))
        .await
        .map_err(|error| {
            ServiceError::Operation(format!("the conversation verb could not be run: {error}"))
        })?
        .map_err(session_control_error)
}

/// Probe every selected harness and assemble the report. Touches no service
/// state, so this runs on any task.
async fn run_inventory(work: InventoryWork) -> std::result::Result<Value, ServiceError> {
    let InventoryWork {
        params,
        supported,
        global_counts,
        workspace_counts,
    } = work;
    let probes = supported.into_iter().map(|descriptor| {
        let global = global_counts
            .as_ref()
            .map(|counts| counts.get(descriptor.id.as_str()).copied().unwrap_or(0));
        let workspace = workspace_counts
            .as_ref()
            .map(|counts| counts.get(descriptor.id.as_str()).copied().unwrap_or(0));
        probe_harness(descriptor, &params, global, workspace)
    });
    let harnesses = futures::future::join_all(probes).await;
    serde_json::to_value(HarnessInventoryReport {
        probe: params.probe,
        workspace: params.workspace,
        harnesses,
    })
    .map_err(|error| ServiceError::Operation(error.to_string()))
}

async fn probe_harness(
    descriptor: crate::HarnessSupportDescriptor,
    params: &HarnessInventoryParams,
    global: Option<usize>,
    workspace: Option<usize>,
) -> LocalHarness {
    let launch = descriptor.runtime.default_launch.as_ref();
    // ORC-7: the orchestrator publishes no runtime launch — it is not an
    // adapter supercode connects a turn to. What "installed" means for it
    // is that its Node daemon entry is present, so the row answers from
    // that instead of from a PATH lookup it could never satisfy.
    let orchestrator_entry = (descriptor.id.as_str() == HarnessId::ORCHESTRATOR)
        .then(crate::orchestrator::daemon_entry)
        .and_then(Result::ok);
    let executable = match &orchestrator_entry {
        Some(entry) => Some(entry.clone()),
        None => launch.and_then(|launch| find_executable(&launch.program)),
    };
    let installed = executable.is_some();
    let version = if params.skip_versions || orchestrator_entry.is_some() {
        // The orchestrator's "executable" is a Node module, not a CLI
        // with a `--version` flag; running it to ask would start a daemon.
        None
    } else {
        match executable.as_deref() {
            Some(path) => executable_version(path).await,
            None => None,
        }
    };
    let configured = auth_evidence(descriptor.id.as_str());
    let mut auth = if configured {
        HarnessAuthState::Configured
    } else if matches!(
        descriptor.id.as_str(),
        HarnessId::CLAUDE_CODE | HarnessId::CODEX
    ) {
        // These two adapters have explicit native status/login contracts
        // and complete local evidence coverage (including Claude's macOS
        // Keychain-backed oauthAccount marker). Treating absent evidence
        // as unknown advertises a start that will only fail interactively.
        HarnessAuthState::Required
    } else {
        HarnessAuthState::Unknown
    };
    let mut runtime = if installed {
        HarnessRuntimeState::Degraded
    } else {
        HarnessRuntimeState::Unavailable
    };
    let is_orchestrator = descriptor.id.as_str() == HarnessId::ORCHESTRATOR;
    let mut reason = (!installed).then(|| {
        if is_orchestrator {
            format!(
                "{} is supported but its daemon entry `{}` was not found",
                descriptor.display_name,
                crate::orchestrator::DAEMON_ENTRY
            )
        } else {
            format!(
                "{} is supported but `{}` was not found on PATH",
                descriptor.display_name,
                launch
                    .map(|launch| launch.program.as_str())
                    .unwrap_or("executable")
            )
        }
    });
    let mut repair = (!installed).then(|| {
        if is_orchestrator {
            format!(
                "Install the `supercode-orchestrator` package so `{}` resolves.",
                crate::orchestrator::DAEMON_ENTRY
            )
        } else {
            format!(
                "Install {} and ensure `{}` is on PATH.",
                descriptor.display_name,
                launch
                    .map(|launch| launch.program.as_str())
                    .unwrap_or("its executable")
            )
        }
    });

    if installed && params.probe == HarnessProbeLevel::Handshake {
        let backend_params = RuntimeBackendParams {
            harness: descriptor.id.clone(),
            protocol: None,
            launch: None,
            base_url: None,
            policy: RuntimePolicy::Default,
        };
        match runtime_backend(&backend_params) {
            Ok(backend) => {
                let cwd = params
                    .workspace
                    .clone()
                    .or_else(|| std::env::current_dir().ok())
                    .unwrap_or_else(|| PathBuf::from("."));
                let isolated = descriptor
                    .runtime
                    .default_launch
                    .clone()
                    .and_then(|launch| IsolatedProbeHome::new(descriptor.id.as_str(), launch).ok());
                let Some(isolated) = isolated else {
                    reason = Some(
                        "No-prompt runtime handshake could not create its isolated harness home."
                            .into(),
                    );
                    repair = Some(
                        "Check temporary-directory permissions, then run the handshake probe again."
                            .into(),
                    );
                    let running = probe_running_instance(descriptor.id.as_str());
                    return LocalHarness {
                        gateway: gateway_health(
                            descriptor.id.as_str(),
                            installed,
                            running.as_ref(),
                            version.as_deref(),
                        ),
                        id: descriptor.id,
                        display_name: descriptor.display_name,
                        supported: true,
                        installed,
                        executable: executable.map(|path| path.to_string_lossy().into_owned()),
                        version,
                        auth,
                        runtime,
                        protocol: descriptor.runtime.protocol,
                        capabilities: descriptor.runtime.capabilities.clone(),
                        effective_capabilities: descriptor.runtime.capabilities,
                        sessions: HarnessSessionCounts { global, workspace },
                        running,
                        reason,
                        repair,
                    };
                };
                match tokio::time::timeout(
                    Duration::from_secs(30),
                    backend.start(RuntimeStartRequest {
                        cwd,
                        launch: Some(isolated.launch.clone()),
                        mcp_servers: Vec::new(),
                        approval_policy: None,
                        model: None,
                    }),
                )
                .await
                {
                    Ok(Ok(mut connection)) => {
                        match stabilize_handshake(connection.as_mut()).await {
                            Ok(()) => {
                                auth = HarnessAuthState::Ready;
                                runtime = HarnessRuntimeState::Ready;
                                reason = Some(
                                    "No-prompt runtime handshake remained healthy through the startup stabilization window; no model request was sent."
                                        .into(),
                                );
                                repair = None;
                            }
                            Err(message) => {
                                auth = if looks_like_auth_error(&message) {
                                    HarnessAuthState::Required
                                } else if configured {
                                    HarnessAuthState::Configured
                                } else {
                                    HarnessAuthState::Unknown
                                };
                                reason = Some(format!(
                                    "No-prompt runtime handshake became unhealthy during startup: {message}"
                                ));
                                repair = Some(if auth == HarnessAuthState::Required {
                                    format!(
                                        "Run `{}` interactively once and complete sign-in, then probe again.",
                                        launch.map(|launch| launch.program.as_str()).unwrap_or("the harness")
                                    )
                                } else {
                                    "Run the harness directly to inspect its startup failure, then probe again."
                                        .into()
                                });
                            }
                        }
                        let _ =
                            tokio::time::timeout(Duration::from_secs(3), connection.close()).await;
                    }
                    Ok(Err(error)) => {
                        let message = truncate_text(&error.to_string(), 500);
                        auth = if looks_like_auth_error(&message) {
                            HarnessAuthState::Required
                        } else if configured {
                            HarnessAuthState::Configured
                        } else {
                            HarnessAuthState::Unknown
                        };
                        reason = Some(format!("No-prompt runtime handshake failed: {message}"));
                        repair = Some(if auth == HarnessAuthState::Required {
                            format!(
                                "Run `{}` interactively once and complete sign-in, then probe again.",
                                launch.map(|launch| launch.program.as_str()).unwrap_or("the harness")
                            )
                        } else {
                            "Check the harness installation and run the handshake probe again."
                                .into()
                        });
                    }
                    Err(_) => {
                        reason =
                            Some("No-prompt runtime handshake timed out after 30 seconds.".into());
                        repair = Some("Run the harness directly to check startup or authentication, then probe again.".into());
                    }
                }
                // Keep the isolated home alive through process teardown.
                // Otherwise the compiler may release the last meaningful
                // use after cloning `launch`, and a still-starting CLI can
                // recreate its state directory after Drop removed it.
                // Some Node-based launchers finish a short asynchronous
                // installation-id write just after their parent process
                // is reaped. Remove once immediately, allow that bounded
                // writer to settle, then perform the authoritative pass.
                let _ = isolated.cleanup();
                tokio::time::sleep(Duration::from_millis(250)).await;
                if let Err(error) = isolated.cleanup() {
                    auth = if configured {
                        HarnessAuthState::Configured
                    } else {
                        HarnessAuthState::Unknown
                    };
                    runtime = HarnessRuntimeState::Degraded;
                    reason = Some(format!(
                        "No-prompt runtime handshake could not remove its isolated harness home: {error}"
                    ));
                    repair = Some(
                        "Check temporary-directory permissions, remove the reported disposable probe home, then run the handshake again."
                            .into(),
                    );
                }
            }
            Err(error) => {
                reason = Some(error_message(error));
            }
        }
    } else if installed && configured {
        reason = Some("Executable and local authentication evidence found; use a handshake probe to verify readiness.".into());
    } else if installed && auth == HarnessAuthState::Required {
        reason = Some("Executable found, but no native authentication evidence is present.".into());
        repair = Some(format!(
            "Run `supercode harness login {}` to use the harness-owned sign-in flow.",
            descriptor.id.as_str()
        ));
    } else if installed {
        reason = Some("Executable found; authentication readiness is unknown until a no-prompt handshake succeeds.".into());
        repair = Some(format!(
            "Run `{}` interactively once if sign-in is required, or use `--probe handshake`.",
            launch
                .map(|launch| launch.program.as_str())
                .unwrap_or("the harness")
        ));
    }

    let effective_capabilities = if installed {
        descriptor.runtime.capabilities.clone()
    } else {
        unavailable_capabilities()
    };
    let running = probe_running_instance(descriptor.id.as_str());
    LocalHarness {
        gateway: gateway_health(
            descriptor.id.as_str(),
            installed,
            running.as_ref(),
            version.as_deref(),
        ),
        id: descriptor.id,
        display_name: descriptor.display_name,
        supported: true,
        installed,
        executable: executable.map(|path| path.to_string_lossy().into_owned()),
        version,
        auth,
        runtime,
        protocol: descriptor.runtime.protocol,
        capabilities: descriptor.runtime.capabilities,
        effective_capabilities,
        sessions: HarnessSessionCounts { global, workspace },
        running,
        reason,
        repair,
    }
}

async fn stabilize_handshake(connection: &mut dyn RuntimeConnection) -> Result<(), String> {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(3);
    loop {
        let now = tokio::time::Instant::now();
        if now >= deadline {
            return Ok(());
        }
        match tokio::time::timeout(deadline - now, connection.next_event()).await {
            Err(_) => return Ok(()),
            Ok(Ok(Some(event))) => {
                if let Some(message) = handshake_event_failure(&event) {
                    return Err(truncate_text(&message, 500));
                }
            }
            Ok(Ok(None)) => return Err("runtime transport closed during startup".into()),
            Ok(Err(error)) => return Err(error.to_string()),
        }
    }
}

fn handshake_event_failure(event: &crate::HarnessEvent) -> Option<String> {
    let detail = event
        .payload
        .get("message")
        .or_else(|| event.payload.get("line"))
        .and_then(Value::as_str)
        .unwrap_or(event.kind.as_str());
    match event.kind.as_str() {
        "transport_closed" => Some("runtime transport closed during startup".into()),
        "transport_error" => Some(format!("runtime transport error: {detail}")),
        "malformed_output" => Some(format!("runtime emitted non-protocol output: {detail}")),
        // Stderr is retained as a runtime event, but is not transport health.
        // Grok, for example, can log an AuthorizationRequired error from an
        // optional background worker while its ACP session continues to send
        // updates and complete prompts normally.
        _ => None,
    }
}

fn indexed_claude_window(
    locator: &SessionLocator,
    options: &SessionLoadOptions,
) -> std::result::Result<Option<Value>, ServiceError> {
    use supercode_interchange::session::ClaudeReadIndex;
    // Exact parent-only window: recursive/full-artifact requests retain the
    // existing owner. This is not a bounded display-history substitution.
    if locator.harness.as_str() != HarnessId::CLAUDE_CODE
        || options.include_subagents != Some(false)
    {
        return Ok(None);
    }
    let crate::StorageLocator::File { path } = &locator.storage else {
        return Ok(None);
    };
    if !ClaudeReadIndex::supports(path)
        .map_err(|error| ServiceError::Operation(error.to_string()))?
    {
        return Ok(None);
    }
    let mut index = ClaudeReadIndex::open(path, Fidelity::ByteLossless)
        .map_err(|error| ServiceError::Operation(error.to_string()))?;
    let total = index.len();
    let (offset, end) = projected_message_window(total, options);
    let session = index
        .read_messages(offset..end)
        .map_err(|error| ServiceError::Operation(error.to_string()))?;
    let summary = index
        .read_summary()
        .map_err(|error| ServiceError::Operation(error.to_string()))?;
    let selected_options = SessionLoadOptions {
        message_offset: None,
        message_limit: None,
        message_tail: None,
        ..options.clone()
    };
    let mut selected = projected_session_json(&session, &selected_options);
    selected["raw_record_count"] = json!(index.raw_record_count());
    Ok(Some(json!({
        "session": selected,
        "summary": projected_session_summary(&summary, options),
        "window": {
            "has_more": offset > 0 || end < total, "has_newer": end < total,
            "has_older": offset > 0, "newer_items": index.item_count(end..total),
            "offset": offset, "older_items": index.item_count(0..offset),
            "returned": end - offset, "total_messages": total,
        }
    })))
}

fn projected_session_result(session: &Session, options: &SessionLoadOptions) -> Value {
    let total_messages = session.messages.len();
    let (offset, end) = projected_message_window(total_messages, options);
    json!({
        "session": projected_session_json(session, options),
        "summary": projected_session_summary(session, options),
        "window": {
            "has_more": offset > 0 || end < total_messages,
            "has_newer": end < total_messages,
            "has_older": offset > 0,
            "newer_items": normalized_item_count(&session.messages[end..]),
            "offset": offset,
            "older_items": normalized_item_count(&session.messages[..offset]),
            "returned": end.saturating_sub(offset),
            "total_messages": total_messages,
        }
    })
}

fn normalized_item_count(messages: &[crate::ChatMessage]) -> usize {
    messages
        .iter()
        .map(|message| {
            let conversation = usize::from(
                matches!(message.role, Role::Assistant | Role::User)
                    && message_has_content(message),
            );
            let tool_result =
                usize::from(message.role == Role::Tool && message_has_content(message));
            conversation + tool_result + message.tool_calls().len()
        })
        .sum()
}

fn projected_session_summary(session: &Session, options: &SessionLoadOptions) -> Value {
    let mut conversational = session.messages.iter().filter(|message| {
        matches!(message.role, Role::Assistant | Role::User) && message_has_content(message)
    });
    let first_message = conversational.clone().next();
    let last_message = conversational.next_back();
    let mut assistant = session
        .messages
        .iter()
        .filter(|message| message.role == Role::Assistant && message_has_content(message));
    let first_assistant_message = assistant.clone().next();
    let last_assistant_message = assistant.next_back();
    let end_of_turn = session
        .messages
        .iter()
        .rev()
        .find(|message| message.role != Role::System)
        .is_some_and(|message| {
            message.role == Role::Assistant
                && message_has_content(message)
                && message.tool_calls().is_empty()
                // Codex narrates while it works (`phase: commentary`); only its `final_answer` ends a turn
                && message.metadata.get("phase").map(String::as_str) != Some("commentary")
        });
    let project = |message: Option<&crate::ChatMessage>| {
        message.map(|message| project_inline_media(message_json(message), options))
    };
    json!({
        "end_of_turn": end_of_turn,
        "first_assistant_message": project(first_assistant_message),
        "first_message": project(first_message),
        "last_assistant_message": project(last_assistant_message),
        "last_assistant_text": last_assistant_message.map(message_text).unwrap_or_default(),
        "last_message": project(last_message),
    })
}

fn message_has_content(message: &crate::ChatMessage) -> bool {
    message
        .content
        .as_deref()
        .is_some_and(|content| !content.trim().is_empty())
        || message
            .content_parts
            .as_ref()
            .is_some_and(|parts| !parts.is_empty())
}

fn message_text(message: &crate::ChatMessage) -> String {
    if let Some(content) = &message.content {
        return content.clone();
    }
    message
        .content_parts
        .as_ref()
        .into_iter()
        .flatten()
        .filter_map(|part| part.get("text").and_then(Value::as_str))
        .collect::<Vec<_>>()
        .join("\n")
}

fn projected_session_json(session: &Session, options: &SessionLoadOptions) -> Value {
    let (offset, end) = projected_message_window(session.messages.len(), options);
    let messages = session.messages[offset..end]
        .iter()
        .map(|message| project_inline_media(message_json(message), options))
        .collect::<Vec<_>>();
    let subagents = if options.include_subagents.unwrap_or(true) {
        // The reported window describes the top-level transcript. Applying it
        // recursively would silently truncate subagents without returning a
        // window for each child. Keep their histories complete while carrying
        // the caller's media policy through the tree.
        let subagent_options = SessionLoadOptions {
            message_limit: None,
            message_offset: None,
            message_tail: None,
            ..options.clone()
        };
        session
            .subagents
            .iter()
            .map(|subagent| projected_session_json(subagent, &subagent_options))
            .collect::<Vec<_>>()
    } else {
        Vec::new()
    };
    json!({
        "source": match session.meta.source {
            SessionSource::ClaudeCode => "claude_code",
            SessionSource::Codex => "codex",
            SessionSource::Gemini => "gemini",
            SessionSource::Goose => "goose",
            SessionSource::Grok => "grok",
            SessionSource::Native => "native",
            SessionSource::OpenClaw => "openclaw",
            SessionSource::Hermes => "hermes",
            SessionSource::OpenCode => "opencode",
            SessionSource::Pi => "pi",
        },
        "session_id": session.meta.session_id,
        "ended_at": session.meta.ended_at,
        "end_reason": session.meta.end_reason,
        "model": session.meta.model,
        "resolved_profile": recorded_session_profile(session),
        "resolved_profile_history": recorded_session_profile_history(session),
        "cwd": session.meta.cwd,
        "system_prompt": session.meta.system_prompt,
        "agent_id": session.meta.agent_id,
        "parent_tool_use_id": session.meta.parent_tool_use_id,
        "lineage": session.meta.lineage,
        "messages": messages,
        "subagents": subagents,
        "raw_record_count": session.raw.len(),
        "parse_error_lines": session.parse_error_lines,
    })
}

/// A session's recorded configuration (`harness.v1.sessions.recorded_config`, ADR 0012): what its harness recorded of
/// its directory, model and permissions, without transcript content. Native records are observations; permissions are
/// never reconstructed from current defaults.
pub fn recorded_config(locator: &SessionLocator) -> crate::Result<Value> {
    let session = load_session(locator)?;
    let mut config = json!({"harness": locator.harness, "session_id": session.meta.session_id,
        "cwd": session.meta.cwd, "model": session.meta.model,
        "approval_policy": null, "sandbox_policy": null, "permission_mode": null});
    for header in &session.meta.codex_headers {
        let payload = header.get("payload").unwrap_or(header);
        for key in ["cwd", "model", "approval_policy", "sandbox_policy"] {
            if let Some(value) = payload.get(key) {
                config[key] = value.clone();
            }
        }
    }
    // A Claude Code session's latest recorded permission mode (its transcript's permission-mode records).
    if let Ok(manifest) = crate::claude_runtime_state::ClaudeRuntimeManifest::from_session(&session)
    {
        config["permission_mode"] = json!(manifest.posture.permission_mode);
    }
    Ok(config)
}

/// Configuration recorded by the native session, independent of the caller's intended profile.
/// Unknown values remain null; neither the reader's binary nor today's folder config proves a past run.
fn recorded_session_profile(session: &Session) -> Value {
    let harness = match session.meta.source {
        SessionSource::ClaudeCode => "claude-code",
        SessionSource::Codex => "codex",
        SessionSource::Gemini => "gemini",
        SessionSource::Goose => "goose",
        SessionSource::Grok => "grok",
        SessionSource::Native => "native",
        SessionSource::OpenClaw => "openclaw",
        SessionSource::Hermes => "hermes",
        SessionSource::OpenCode => "opencode",
        SessionSource::Pi => "pi",
    };
    let mut model = session
        .messages
        .iter()
        .rev()
        .find_map(|message| message.metadata.get("model").cloned())
        .or_else(|| session.meta.model.clone());
    let mut cwd = session
        .meta
        .cwd
        .as_ref()
        .map(|path| path.to_string_lossy().into_owned());
    let mut version: Option<String> = None;
    let mut approval = Value::Null;
    let mut sandbox = Value::Null;
    for header in &session.meta.codex_headers {
        let payload = header.get("payload").unwrap_or(header);
        if let Some(value) = payload.get("cli_version").and_then(Value::as_str) {
            version = Some(value.to_owned());
        }
        if let Some(value) = payload.get("model").and_then(Value::as_str) {
            model = Some(value.to_owned());
        }
        if let Some(value) = payload.get("cwd").and_then(Value::as_str) {
            cwd = Some(value.to_owned());
        }
        if let Some(value) = payload.get("approval_policy") {
            approval = value.clone();
        }
        if let Some(value) = payload.get("sandbox_policy") {
            sandbox = value.clone();
        }
    }
    if session.meta.source == SessionSource::ClaudeCode {
        version = session
            .raw
            .iter()
            .rev()
            .filter_map(|line| serde_json::from_str::<Value>(line).ok())
            .find_map(|record| {
                record
                    .get("version")
                    .and_then(Value::as_str)
                    .map(str::to_owned)
            });
    }
    json!({"worker":{"harness":harness,"model":model,"cwd":cwd,"binary":{"version":version}},
        "permissions":{"native":{"approval":approval,"sandbox":sandbox}},"source":"native-session"})
}

/// Each native configuration change is retained even when several happened between connector captures.
/// No current model, executable or intended configuration is substituted for an earlier observation.
fn recorded_session_profile_history(session: &Session) -> Vec<Value> {
    let mut profile = json!({"worker":{"harness":if session.meta.source == SessionSource::Codex {"codex"} else {"claude-code"},
        "model":null,"cwd":null,"binary":{"version":null}},
        "permissions":{"native":{"approval":null,"sandbox":null}},"source":"native-session"});
    let records: Box<dyn Iterator<Item = (usize, Value)> + '_> =
        if session.meta.source == SessionSource::Codex {
            Box::new(session.meta.codex_headers.iter().cloned().enumerate())
        } else if session.meta.source == SessionSource::ClaudeCode {
            Box::new(session.raw.iter().enumerate().filter_map(|(index, line)| {
                serde_json::from_str::<Value>(line)
                    .ok()
                    .map(|record| (index, record))
            }))
        } else {
            return Vec::new();
        };
    let mut history = Vec::new();
    for (index, record) in records {
        let before = profile.clone();
        let payload = record.get("payload").unwrap_or(&record);
        for (native, target) in [("cwd", "cwd"), ("model", "model")] {
            if let Some(value) = payload.get(native).and_then(Value::as_str) {
                profile["worker"][target] = json!(value);
            }
        }
        let version = payload
            .get("cli_version")
            .or_else(|| record.get("version"))
            .and_then(Value::as_str);
        if let Some(value) = version {
            profile["worker"]["binary"]["version"] = json!(value);
        }
        if let Some(value) = record
            .get("message")
            .and_then(|m| m.get("model"))
            .and_then(Value::as_str)
        {
            profile["worker"]["model"] = json!(value);
        }
        for (native, target) in [
            ("approval_policy", "approval"),
            ("sandbox_policy", "sandbox"),
        ] {
            if let Some(value) = payload.get(native) {
                profile["permissions"]["native"][target] = value.clone();
            }
        }
        if profile != before {
            history.push(json!({"native_key":format!("profile-record:{index}"),"at":record.get("timestamp").and_then(Value::as_str),"profile":profile}));
        }
    }
    history
}

fn projected_message_window(total: usize, options: &SessionLoadOptions) -> (usize, usize) {
    if let Some(tail) = options.message_tail {
        return (total.saturating_sub(tail), total);
    }
    let offset = options.message_offset.unwrap_or(0).min(total);
    let end = options
        .message_limit
        .map(|limit| offset.saturating_add(limit).min(total))
        .unwrap_or(total);
    (offset, end)
}

fn project_inline_media(mut message: Value, options: &SessionLoadOptions) -> Value {
    let Some(parts) = message.get_mut("content").and_then(Value::as_array_mut) else {
        return message;
    };
    for part in parts {
        let Some(url) = part
            .get("image_url")
            .and_then(|image| image.get("url"))
            .and_then(Value::as_str)
        else {
            continue;
        };
        let Some(rest) = url.strip_prefix("data:") else {
            continue;
        };
        let Some((media_type, encoded)) = rest.split_once(";base64,") else {
            continue;
        };
        let padding = usize::from(encoded.ends_with('=')) + usize::from(encoded.ends_with("=="));
        let decoded_bytes = encoded.len().saturating_mul(3) / 4;
        let decoded_bytes = decoded_bytes.saturating_sub(padding);
        let should_elide = matches!(options.inline_media, InlineMediaMode::Metadata)
            || options
                .max_inline_media_bytes
                .is_some_and(|limit| decoded_bytes > limit);
        if should_elide {
            *part = json!({
                "type": "media_reference",
                "media_type": media_type,
                "encoding": "base64",
                "encoded_bytes": encoded.len(),
                "decoded_bytes": decoded_bytes,
                "omitted": true,
            });
        }
    }
    message
}

/// `harness.v1.sessions.usage`: one session's recorded use per UTC day and model (G3), recomputed whole from its
/// transcript ([`crate::token_spend::session_spend`]); with `since` (`10m`, `1h`, `2d`, or an RFC 3339 moment) only
/// the records from then on.
#[derive(Deserialize)]
struct UsageParams {
    locator: SessionLocator,
    #[serde(default)]
    since: Option<String>,
}

/// `harness.v1.sessions.spend`: every Claude Code and Codex session on this machine with recorded use since `since`
/// (`10m`, `1h`, `2d`, or an RFC 3339 moment), most spent first ([`crate::token_spend::machine_spend`]).
/// `harness.v1.sessions.tree`: the machine's running sessions under the sessions that started them
/// ([`crate::session_tree::live_tree`]); with `at` (`10m`, `2h` back, or an RFC 3339 moment) the tree as recorded then
/// ([`crate::session_tree::recorded_tree`]).
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct TreeParams {
    #[serde(default)]
    at: Option<String>,
    /// Same storage roots discovery accepts.
    #[serde(default)]
    homes: crate::HarnessHomes,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct SpendParams {
    since: String,
    /// `claude-code`, `codex`; empty reads both.
    #[serde(default)]
    harnesses: Vec<String>,
    /// `cost` (the default: priced sessions by cost, then the rest by tokens) or `tokens`.
    #[serde(default)]
    sort: Option<String>,
    /// The most spent sessions to list; the total counts all.
    #[serde(default)]
    limit: Option<usize>,
    /// Same storage roots discovery accepts.
    #[serde(default)]
    homes: crate::HarnessHomes,
}

#[derive(Deserialize)]
struct LocatorParams {
    locator: SessionLocator,
    /// Optional fidelity for the READ surfaces (`sessions.load`,
    /// `sessions.follow`).
    ///
    /// Omitted means [`Fidelity::Semantic`]: these two methods only ever
    /// produce a read-only view, and a compacted or resumed-across-files
    /// transcript — the everyday shape of a long Claude Code session — has no
    /// losslessly reconstructable record graph, so refusing to render it made
    /// the mirror unusable rather than accurate. A caller that intends to
    /// CONTINUE from what it reads asks for a lossless level explicitly and
    /// gets the strict refusal back. Every other method (export, translate,
    /// branch, handoff, resume_instructions) is lossless-only and has no
    /// such knob.
    #[serde(default)]
    fidelity: Option<Fidelity>,
    /// Optional bounded frontend projection. Absent preserves the historical
    /// complete-session read contract.
    #[serde(default)]
    view: Option<SessionReadView>,
}

#[derive(Deserialize)]
struct SessionReadView {
    /// Number of trailing normalized messages to return. Zero is treated as
    /// one so a caller cannot accidentally request an unbounded empty mode.
    #[serde(default)]
    tail_messages: Option<usize>,
    /// Whether Claude Code child transcripts belong in this view. The
    /// frontend default is false; the legacy no-view path remains true.
    #[serde(default)]
    include_subagents: bool,
    /// Preserve human-visible native history across model-context compaction.
    #[serde(default)]
    display_history: bool,
    /// Bound each individual text field so a single tool result cannot turn a
    /// small message window into a hundred-megabyte RPC response.
    #[serde(default)]
    max_message_chars: Option<usize>,
}

impl LocatorParams {
    fn read_fidelity(&self) -> Fidelity {
        self.fidelity.unwrap_or(Fidelity::Semantic)
    }

    fn include_subagents(&self) -> bool {
        self.view
            .as_ref()
            .map(|view| view.include_subagents)
            .unwrap_or(true)
    }

    fn tail_messages(&self) -> Option<usize> {
        self.view
            .as_ref()
            .and_then(|view| view.tail_messages)
            .map(|limit| limit.clamp(1, 5_000))
    }

    fn display_history(&self) -> bool {
        self.view.as_ref().is_some_and(|view| view.display_history)
    }

    fn max_message_chars(&self) -> Option<usize> {
        self.view
            .as_ref()
            .and_then(|view| view.max_message_chars)
            .map(|limit| limit.clamp(256, 64_000))
    }

    fn bound_session(&self, session: &mut Session) {
        bound_session_view(session, self.tail_messages(), self.max_message_chars());
    }
}

#[derive(Debug, Clone, Copy, Default, Deserialize)]
#[serde(rename_all = "snake_case")]
enum InlineMediaMode {
    #[default]
    Full,
    Metadata,
}

#[derive(Debug, Clone, Default, Deserialize)]
#[serde(default)]
struct SessionLoadOptions {
    include_subagents: Option<bool>,
    inline_media: InlineMediaMode,
    max_inline_media_bytes: Option<usize>,
    message_limit: Option<usize>,
    message_offset: Option<usize>,
    message_tail: Option<usize>,
}

impl SessionLoadOptions {
    fn validate(&self) -> std::result::Result<(), ServiceError> {
        if self.message_tail.is_some()
            && (self.message_limit.is_some() || self.message_offset.is_some())
        {
            return Err(ServiceError::InvalidParams(
                "sessions.load options.message_tail cannot be combined with message_limit or message_offset"
                    .into(),
            ));
        }
        Ok(())
    }
}

/// `sessions.load`: a read of one session's own store, sharing no service state (the catalog is stateless), so a
/// transport runs it beside its loop: a large transcript's load (seconds) held every later request on the server.
fn load_session_door(params: LoadSessionParams) -> std::result::Result<Value, ServiceError> {
    if let Some(options) = &params.options {
        options.validate()?;
        if let Some(result) = indexed_claude_window(&params.read.locator, options)? {
            return Ok(result);
        }
        return load_session(&params.read.locator)
            .map(|session| projected_session_result(&session, options))
            .map_err(operation);
    }
    let mut session = if params.read.display_history() {
        HarnessCatalog::new()
            .load_display_view(
                &params.read.locator,
                params.read.read_fidelity(),
                params.read.tail_messages().unwrap_or(500),
            )
            .map_err(crate::Error::from)
    } else if params.read.include_subagents() {
        load_session_with_fidelity(&params.read.locator, params.read.read_fidelity())
    } else {
        HarnessCatalog::new()
            .load_parent_with_fidelity(&params.read.locator, params.read.read_fidelity())
            .map_err(crate::Error::from)
    }
    .map_err(operation)?;
    params.read.bound_session(&mut session);
    // The session's own record of how it ran (model, version, cwd, policies): a capture reports it to its
    // agent's record, as the projected read carries it.
    let mut value = normalized_session_json(&session);
    value["resolved_profile"] = recorded_session_profile(&session);
    value["resolved_profile_history"] = Value::Array(recorded_session_profile_history(&session));
    Ok(json!({"session": value}))
}

#[derive(Deserialize)]
struct LoadSessionParams {
    #[serde(flatten)]
    read: LocatorParams,
    #[serde(default)]
    options: Option<SessionLoadOptions>,
}

#[derive(Deserialize)]
struct UnfollowParams {
    subscription: String,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct IndexResizeParams {
    subscription: String,
    limit: usize,
}

/// The Codex sessions whose daemon pane shows a prompt, as last read, and when; and whether a read is
/// running. Read on a thread of its own, never on serve's loop.
#[cfg(feature = "adapter-api")]
#[derive(Default)]
struct PanePrompts {
    waiting: std::collections::HashSet<String>,
    read_at: Option<std::time::Instant>,
    reading: bool,
}

#[cfg(feature = "adapter-api")]
static PANE_PROMPTS: std::sync::Mutex<Option<PanePrompts>> = std::sync::Mutex::new(None);

/// How old the pane-prompt reading may be before a new one is started.
#[cfg(feature = "adapter-api")]
const PANE_PROMPT_REFRESH: std::time::Duration = std::time::Duration::from_secs(3);

/// A Codex prompt (an approval, a choice) writes nothing to its rollout, which reads that turn as
/// working (or unknown); its daemon pane shows the prompt. A running Codex session whose live row
/// is waiting needs input, the reading discovery makes. The live rows (ps, the daemon's panes, their
/// screens) take about a second to read, and serve's loop answers every request: they are read on a
/// thread of their own at most every few seconds while some Codex session runs, and this applies
/// the last reading without waiting for it. A prompt reaches a subscription within one refresh.
#[cfg(feature = "adapter-api")]
fn pane_prompt_override(activities: &mut [crate::SessionActivity], homes: &crate::HarnessHomes) {
    let running_codex = |activity: &crate::SessionActivity| {
        activity.harness.as_str() == "codex"
            && activity.presence == crate::session_activity::SessionPresence::Running
            // An approval or a choice comes in a turn: an idle session has none to show, so idle Codex TUIs
            // never start the read.
            && matches!(
                activity.turn,
                crate::SessionTurnState::Working | crate::SessionTurnState::Unknown
            )
    };
    if !activities.iter().any(running_codex) {
        return;
    }
    let waiting = {
        let Ok(mut guard) = PANE_PROMPTS.lock() else {
            return;
        };
        let prompts = guard.get_or_insert_with(PanePrompts::default);
        let stale = prompts
            .read_at
            .is_none_or(|at| at.elapsed() >= PANE_PROMPT_REFRESH);
        if stale && !prompts.reading {
            prompts.reading = true;
            let homes = homes.clone();
            std::thread::spawn(move || {
                let doors = crate::mail_route::LiveSessions::read(&homes);
                let waiting = doors
                    .all()
                    .iter()
                    .filter(|live| live.address.harness == "codex" && live.status == "waiting")
                    .map(|live| live.address.session_id.clone())
                    .collect();
                if let Ok(mut guard) = PANE_PROMPTS.lock() {
                    let prompts = guard.get_or_insert_with(PanePrompts::default);
                    prompts.waiting = waiting;
                    prompts.read_at = Some(std::time::Instant::now());
                    prompts.reading = false;
                }
            });
        }
        prompts.waiting.clone()
    };
    for activity in activities
        .iter_mut()
        .filter(|activity| running_codex(activity))
    {
        if waiting.contains(&activity.session_id) {
            activity.turn = crate::SessionTurnState::NeedsInput;
            activity.evidence.source = "pane_prompt".into();
        }
    }
}

#[derive(Deserialize)]
struct ActivitySubscribeParams {
    locators: Vec<SessionLocator>,
    #[serde(default)]
    homes: crate::HarnessHomes,
}

/// Whom `sessions.message` writes to: a session's harness and id, as its mail address names it. A session's storage is
/// not needed to deliver mail, and an agent (`{harness: "agent", session_id: <name>}`, docs/adr/0008-agent-mailbox.md)
/// has none; a full catalog locator is accepted as well.
#[derive(Debug, Clone, Deserialize)]
struct MessageTarget {
    harness: HarnessId,
    session_id: String,
    #[serde(default)]
    #[allow(dead_code)]
    storage: Option<supercode_interchange::catalog::StorageLocator>,
}

#[derive(Deserialize)]
struct MessageSessionParams {
    locator: MessageTarget,
    text: String,
    #[serde(default)]
    subject: Option<String>,
    /// A stable caller key. Retrying delivery after a lost receipt keeps the same envelope.
    #[serde(default)]
    idempotency_key: Option<String>,
    /// Channel observations retain their trust boundary instead of becoming peer instructions.
    #[serde(default)]
    channel: bool,
    /// Name the sender is known by (`fleet-board`, `aaron`). Replies to the
    /// message are filed in this sender's mailbox, read with
    /// `sessions.inbox`. Defaults to `supercode`.
    #[serde(default)]
    from_name: Option<String>,
    /// A voice front for the receiving session, never a separate agent.
    #[serde(default)]
    voice_for: Option<crate::mailbox::MailAddress>,
    /// Human-readable sender label, independent of its return mailbox.
    #[serde(default)]
    sender_name: Option<String>,
    /// Id of the message this one answers.
    #[serde(default)]
    in_reply_to: Option<String>,
    /// A channel's own id for this line (`rh2:<message id>`), so a reply naming it
    /// by `reply_marker` joins its thread (docs/adr/0008-agent-mailbox.md).
    #[serde(default)]
    marker: Option<String>,
    /// The channel's id for the line this one answers, when the platform marks
    /// it a reply (a Slack `thread_ts`, an RH2 `parentMessageId`).
    #[serde(default)]
    reply_marker: Option<String>,
    /// Where the channel shows the thread (`rh2:<room>/<conversation>`).
    #[serde(default)]
    surface: Option<String>,
    /// File one notice in the sender's mailbox when the receiver next goes idle.
    #[serde(default)]
    notify_when_idle: bool,
    /// The text is the session's own user speaking (a voice bridge, a board
    /// the owner types in): user mail, delivered as the user's own turn.
    /// `harness.v1` is served only to the machine's owner (its daemon admits
    /// operators only), which is the authority this carries.
    #[serde(default)]
    as_user: bool,
    /// Same storage roots discovery accepts, so a caller (and a test) can
    /// point the live-session registry somewhere other than `$HOME`.
    #[serde(default)]
    homes: crate::HarnessHomes,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct InboxParams {
    /// Sender name used with `sessions.message` (its mailbox), or
    #[serde(default)]
    from_name: Option<String>,
    /// an explicit session address `sc:<machine>:<harness>:<id>`.
    #[serde(default)]
    address: Option<String>,
    /// Include messages already read.
    #[serde(default)]
    all: bool,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct HarnessSettingsParams {
    harness: String,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ConfigureHarnessParams {
    harness: String,
    #[serde(default)]
    changes: Vec<crate::HarnessSettingChange>,
    #[serde(default)]
    expected_revision: Option<String>,
}

fn claude_inbound_controls_or_error(homes: &crate::HarnessHomes) -> (Value, Value) {
    match crate::inspect_harness_interop_settings(homes, HarnessId::CLAUDE_CODE) {
        Ok(report) => (
            serde_json::to_value(report).unwrap_or(Value::Null),
            Value::Null,
        ),
        Err(error) => (
            Value::Null,
            Value::String(format!(
                "Volter Harness could not inspect Claude Code inbound controls: {error}"
            )),
        ),
    }
}

/// Deliver `text` into a session that is running right now, or say why not.
///
/// A refusal is a RESULT, not a JSON-RPC error: "that session is persisted
/// only" is an answer about the session, which a mirror renders next to the
/// transcript, and this service's error envelope carries no structured data
/// field a machine-readable reason could survive in.
///
/// The door is chosen by the one router every sender uses
/// ([`crate::mail_route`]): a session supercode controls gets it through its
/// runtime (the default tier); a Claude session it does not control through a
/// Claude relay, so the reply comes back; a Codex session through its hook;
/// anything else is stored. Replies are filed in the sender's mailbox under
/// `reply_to`, read with `sessions.inbox`.
async fn message_live_session(params: &MessageSessionParams) -> Value {
    use crate::mail_route::{Delivered, NoDoor, Refused};
    let (inbound_controls, inbound_controls_error) =
        claude_inbound_controls_or_error(&params.homes);
    let refused = |reason: &str, message: String| {
        json!({
            "delivered_to_bus": false,
            "refusal": {"reason": reason, "message": message},
            "inbound_controls": inbound_controls,
            "inbound_controls_error": inbound_controls_error,
        })
    };
    if params.text.trim().is_empty() {
        return refused(
            crate::claude_peer::ClaudePeerRefusal::DeliveryFailed.as_str(),
            "refusing to deliver an empty message".into(),
        );
    }
    let sender = match operator_address(params.from_name.as_deref()) {
        Ok(sender) => sender,
        Err(message) => return refused("invalid_sender", message),
    };
    let receiver = match crate::mailbox::MailAddress::new(
        crate::mailbox::local_machine_name(),
        params.locator.harness.as_str(),
        &params.locator.session_id,
    ) {
        Ok(receiver) => receiver,
        Err(error) => return refused("delivery_failed", error.to_string()),
    };
    if params
        .voice_for
        .as_ref()
        .is_some_and(|voice_for| voice_for != &receiver)
    {
        return refused(
            "invalid_sender",
            "a voice front must represent the receiving session".into(),
        );
    }
    // A Claude session's registry name must still name only that session. A thread's holder the mailbox launched may
    // have stopped (decision 9): its mail is the mailbox's to route, which resumes it (decision 5).
    if params.locator.harness.as_str() == HarnessId::CLAUDE_CODE
        && !crate::mail_agent::resumable(&receiver)
        && crate::runtime_mail::controlled_runtime(
            HarnessId::CLAUDE_CODE,
            &params.locator.session_id,
        )
        .is_none()
    {
        if let Err(refusal) =
            crate::claude_peer::resolve_live_session(&params.homes, &params.locator.session_id)
        {
            return refused(refusal.reason.as_str(), refusal.message);
        }
    }
    if params.as_user {
        // The owner writing to an agent types into its main session (docs/adr/0008-agent-mailbox.md, decisions 3 and
        // 12): a typed line there is the owner's, and main answers or delegates it.
        let agent_address =
            (receiver.harness == crate::mail_agent::AGENT_HARNESS).then(|| receiver.clone());
        let receiver = if receiver.harness == crate::mail_agent::AGENT_HARNESS {
            match crate::mail_agent::load(&receiver.session_id) {
                Some(agent) => agent.main_session,
                None => {
                    return refused(
                        "delivery_failed",
                        format!(
                            "no agent named {} is declared on this machine",
                            receiver.session_id
                        ),
                    )
                }
            }
        } else {
            receiver
        };
        return message_as_user(
            params,
            sender,
            receiver,
            agent_address,
            inbound_controls,
            inbound_controls_error,
        )
        .await;
    }
    // An agent's mail (a root to it, a reply in one of its threads) is routed by its
    // mailbox, which resumes a stopped receiver; anything else needs its receiver running.
    let door = crate::mail_route::door_for(&params.homes, &receiver);
    let mut envelope = match crate::mailbox::Envelope::new(
        sender.clone(),
        params
            .sender_name
            .clone()
            .unwrap_or_else(|| format!("{}@{}", sender.session_id, sender.machine)),
        if params.channel {
            crate::mailbox::MailKind::Channel
        } else {
            crate::mailbox::MailKind::Peer
        },
        crate::mailbox::ReplyVia::Command,
        params.text.clone(),
    ) {
        Ok(envelope) => envelope,
        Err(error) => return refused("delivery_failed", error.to_string()),
    };
    if let Some(key) = &params.idempotency_key {
        if key.is_empty() || key.len() > 256 {
            return refused("invalid_key", "idempotency_key needs 1–256 bytes".into());
        }
        let identity = serde_json::to_vec(&(sender.to_string(), receiver.to_string(), key))
            .expect("string tuple serializes");
        envelope.id = format!("m-{}", &blake3::hash(&identity).to_hex()[..24]);
        let mailbox = match crate::mailbox::Mailbox::open(&crate::mailbox::mail_root(), &receiver) {
            Ok(mailbox) => mailbox,
            Err(error) => return refused("delivery_failed", error.to_string()),
        };
        match mailbox.find(&envelope.id) {
            Ok(Some(previous)) => {
                let saved = previous.envelope;
                let subject = params.subject.as_deref().map(|value| {
                    value
                        .lines()
                        .next()
                        .unwrap_or("")
                        .chars()
                        .take(200)
                        .collect::<String>()
                });
                if saved.body != envelope.body
                    || saved.subject != subject
                    || saved.kind != envelope.kind
                    || saved.from_name != envelope.from_name
                    || saved.in_reply_to != params.in_reply_to
                    || saved.voice_for != params.voice_for
                {
                    return refused(
                        "idempotency_conflict",
                        "this key already names different mail".into(),
                    );
                }
            }
            Ok(None) => {}
            Err(error) => return refused("delivery_failed", error.to_string()),
        }
    }
    envelope.in_reply_to = params.in_reply_to.clone();
    envelope.thread = crate::mailbox::thread_of_reply(params.in_reply_to.as_deref());
    envelope.voice_for = params.voice_for.clone();
    envelope.subject = params.subject.as_deref().map(|value| {
        value
            .lines()
            .next()
            .unwrap_or("")
            .chars()
            .take(200)
            .collect()
    });
    let channel = crate::mail_agent::Channel {
        marker: params.marker.clone(),
        reply_marker: params.reply_marker.clone(),
        surface: params.surface.clone(),
    };
    match crate::mail_agent::plan(&mut envelope, &receiver, &channel) {
        Err(error) => return refused("delivery_failed", error.to_string()),
        Ok(Some(plan)) => {
            let caller = crate::mail_route::Caller {
                address: sender.clone(),
                name: envelope.from_name.clone(),
            };
            let outcome = crate::mail_send::deliver_planned(
                &params.homes,
                &caller,
                &envelope,
                &plan,
                false,
                params.notify_when_idle,
            )
            .await;
            return json!({
                "delivered_to_bus": outcome.code == 0,
                "message_id": envelope.id,
                "reply_to": sender.to_string(),
                "thread": plan.thread.as_ref().map(|thread| thread.id.clone()),
                "target": {
                    "harness": params.locator.harness.as_str(),
                    "session_id": params.locator.session_id,
                },
                "delivery": {"door": "agent", "how": outcome.text},
                "inbound_controls": inbound_controls,
                "inbound_controls_error": inbound_controls_error,
            });
        }
        Ok(None) => {}
    }
    let door = match door {
        Ok(door) => door,
        Err(NoDoor::NotRunning | NoDoor::OtherMachine(_)) => {
            return refused(
                crate::claude_peer::ClaudePeerRefusal::NotLive.as_str(),
                format!(
                    "no running `{}` session `{}` is reachable; its transcript is persisted only",
                    params.locator.harness.as_str(),
                    params.locator.session_id
                ),
            )
        }
    };
    let how = match crate::mail_route::deliver(
        &envelope,
        &receiver,
        &door,
        true,
        params.notify_when_idle,
    )
    .await
    {
        Err(detail) => {
            return refused(
                crate::claude_peer::ClaudePeerRefusal::DeliveryFailed.as_str(),
                detail,
            )
        }
        Ok(Err(Refused::CannotQueueNative)) => unreachable!("sessions.message always wakes"),
        Ok(Err(Refused::TooLong(bytes))) => {
            return refused(
                "too_long",
                format!(
                    "the message is {bytes} bytes; the limit is {}",
                    crate::mail_route::MAX_RELAYED_BYTES
                ),
            )
        }
        Ok(Ok(delivered)) => match delivered {
            Delivered::Steered => "steered",
            Delivered::Started => "started",
            Delivered::Native { busy: true } => "next_tool_call",
            Delivered::Native { busy: false } => "started",
            Delivered::Hooked => "hook",
            Delivered::HookWoken => "woken",
            Delivered::Queued => "queued",
            Delivered::Stored => "stored",
            Delivered::Operator => "filed",
            Delivered::Already => "already",
        },
    };
    json!({
        "delivered_to_bus": !matches!(door, crate::mail_route::Door::Stored),
        "message_id": envelope.id,
        "reply_to": sender.to_string(),
        "target": {
            "harness": params.locator.harness.as_str(),
            "session_id": params.locator.session_id,
            "name": match &door {
                crate::mail_route::Door::Native(session) => Some(session.name.clone()),
                _ => None,
            },
        },
        "delivery": {"door": door.name(), "how": how},
        "inbound_controls": inbound_controls,
        "inbound_controls_error": inbound_controls_error,
    })
}

/// `sessions.message` with `as_user`: user mail, through the one door that
/// carries the user's authority (a hosted runtime's input, or the session's
/// pane when its composer is empty; otherwise it waits in the mailbox).
async fn message_as_user(
    params: &MessageSessionParams,
    sender: crate::mailbox::MailAddress,
    receiver: crate::mailbox::MailAddress,
    agent: Option<crate::mailbox::MailAddress>,
    inbound_controls: Value,
    inbound_controls_error: Value,
) -> Value {
    let mut envelope = match crate::mailbox::Envelope::new(
        sender.clone(),
        format!("{}@{}", sender.session_id, sender.machine),
        crate::mailbox::MailKind::User,
        crate::mailbox::ReplyVia::None,
        params.text.clone(),
    ) {
        Ok(envelope) => envelope,
        Err(error) => {
            return json!({
                "delivered_to_bus": false,
                "refusal": {"reason": "delivery_failed", "message": error.to_string()},
                "inbound_controls": inbound_controls,
                "inbound_controls_error": inbound_controls_error,
            })
        }
    };
    // A line to an agent is a root (or a reply in its thread): the agent's mailbox files its thread, with the owner a
    // participant, before it is typed into main.
    // The session the plan wakes takes the typed line (main for a root, a thread's holder for a reply in it); every
    // other participant has it filed unread (main as CC, the owner's own mailbox).
    let mut receiver = receiver;
    if let Some(agent) = &agent {
        envelope.in_reply_to = params.in_reply_to.clone();
        envelope.thread = crate::mailbox::thread_of_reply(params.in_reply_to.as_deref());
        let channel = crate::mail_agent::Channel::default();
        let planned = match crate::mail_agent::plan(&mut envelope, agent, &channel) {
            Ok(planned) => planned,
            Err(error) => {
                return json!({
                    "delivered_to_bus": false,
                    "refusal": {"reason": "delivery_failed", "message": error.to_string()},
                    "inbound_controls": inbound_controls,
                    "inbound_controls_error": inbound_controls_error,
                })
            }
        };
        if let Some(planned) = planned {
            let typed_to = planned
                .recipients
                .iter()
                .find(|r| r.wake && r.address.harness != "operator")
                .map(|r| r.address.clone());
            if let Some(typed_to) = typed_to {
                receiver = typed_to;
            }
            if let Some(agent) = &planned.agent {
                // The agent's own mailbox keeps the record of what was addressed to it.
                let _ = crate::mailbox::Mailbox::open(&crate::mailbox::mail_root(), agent)
                    .and_then(|mailbox| mailbox.deliver_read(&envelope));
            }
            // A copy of the owner's line is a record of a line typed elsewhere (as decision 12's copies are), never a
            // turn of the copy's session: pending user mail would be typed into it.
            let mut copy = envelope.clone();
            copy.kind = crate::mailbox::MailKind::Typed;
            for other in planned
                .recipients
                .iter()
                .filter(|r| r.address != receiver && r.address != sender)
            {
                let _ = crate::mail_agent::file_unread(&other.address, &copy);
            }
            // Every owner line to any agent is the owner's account manager's as CC (decision 17), unless it is that
            // agent's own.
            if let Some(account_manager) = crate::mail_agent::owners_account_manager() {
                let main = account_manager.main_session.clone();
                let already =
                    main == receiver || planned.recipients.iter().any(|r| r.address == main);
                if account_manager.name != agent.session_id && !already {
                    let _ = crate::mail_agent::file_unread(&main, &copy);
                }
            }
        }
    }
    // A thread's holder the mailbox launched and that has since closed (decision 9) is resumed the way it was
    // launched, and takes the line once its door is back (decision 5); the line is never a new root.
    if agent.is_some()
        && crate::mail_route::door_for(&params.homes, &receiver).is_err()
        && crate::mail_agent::resumable(&receiver)
    {
        if let Err(error) = crate::mail_agent::resume(&receiver) {
            return json!({
                "delivered_to_bus": false,
                "refusal": {"reason": "no_user_door", "message": format!("{receiver} was not resumed: {error}")},
                "inbound_controls": inbound_controls,
                "inbound_controls_error": inbound_controls_error,
            });
        }
        let started = std::time::Instant::now();
        while crate::mail_route::door_for(&params.homes, &receiver).is_err()
            && started.elapsed() < crate::mail_agent::RESUME_WAIT
        {
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        }
    }
    match crate::mail_route::deliver_user_turn(&params.homes, &envelope, &receiver).await {
        Ok(turn) => json!({
            "delivered_to_bus": turn != crate::mail_route::UserTurn::Waiting,
            "message_id": envelope.id,
            "target": {
                "harness": params.locator.harness.as_str(),
                "session_id": params.locator.session_id,
            },
            "delivery": {
                "door": match turn {
                    crate::mail_route::UserTurn::Steered | crate::mail_route::UserTurn::Started => "runtime",
                    _ => "pane",
                },
                "how": turn.as_str(),
            },
            "inbound_controls": inbound_controls,
            "inbound_controls_error": inbound_controls_error,
        }),
        // `no_user_door` says only that the session has no door for its user's turn (a caller may then reach it as a
        // peer); a door that failed is `delivery_failed`, so the line is never sent twice.
        Err(message) => json!({
            "delivered_to_bus": false,
            "refusal": {
                "reason": if crate::mail_route::has_user_door(&params.homes, &receiver) { "delivery_failed" } else { "no_user_door" },
                "message": message,
            },
            "inbound_controls": inbound_controls,
            "inbound_controls_error": inbound_controls_error,
        }),
    }
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ActivityUnderParams {
    /// Root processes (a terminal pane's shell, say).
    pids: Vec<u32>,
    #[serde(default)]
    homes: crate::HarnessHomes,
}

/// `harness.v1.sessions.activity_under`: the harness session running beneath
/// each root process, with its activity, so a terminal's attention takes a
/// harness pane's liveness from the harness's own lifecycle.
async fn activity_under_call(params: Value) -> std::result::Result<Value, ServiceError> {
    let params = decode::<ActivityUnderParams>(params)?;
    if params.pids.len() > 1024 {
        return Err(ServiceError::InvalidParams(
            "sessions.activity_under accepts at most 1024 pids".into(),
        ));
    }
    let found = crate::session_activity::activity_under(&params.pids, &params.homes)
        .await
        .map_err(ServiceError::Sdk)?;
    Ok(json!({
        "activities": found
            .into_iter()
            .map(|(pid, activity)| json!({"pid": pid, "activity": activity}))
            .collect::<Vec<_>>(),
    }))
}

/// Mailbox address of an operator sender (a board, a person's shell) that is
/// not itself a harness session: `sc:<machine>:operator:<name>`.
fn operator_address(
    name: Option<&str>,
) -> std::result::Result<crate::mailbox::MailAddress, String> {
    let name: String = name
        .unwrap_or("supercode")
        .trim()
        .chars()
        .map(|character| {
            if character.is_whitespace() || character == '@' {
                '-'
            } else {
                character
            }
        })
        .collect();
    if name.is_empty() {
        return Err("from_name must not be empty".into());
    }
    crate::mailbox::MailAddress::new(crate::mailbox::local_machine_name(), "operator", name)
        .map_err(|error| error.to_string())
}

/// `harness.v1.sessions.inbox`: a sender's mailbox, each message with the
/// text its reader sees. An operator reading its own mailbox (`from_name`) claims and acknowledges its unread
/// messages, so a second read does not return them again. A read of any other address (`address`) only looks: no
/// third party's read ever consumes a session's mail, which its carrier still delivers.
fn inbox_call(params: InboxParams) -> std::result::Result<Value, ServiceError> {
    let address = match (&params.address, &params.from_name) {
        (Some(address), None) => crate::mailbox::MailAddress::parse(address)
            .map_err(|error| ServiceError::InvalidParams(error.to_string()))?,
        (None, name) => operator_address(name.as_deref()).map_err(ServiceError::InvalidParams)?,
        (Some(_), Some(_)) => {
            return Err(ServiceError::InvalidParams(
                "sessions.inbox takes from_name or address, not both".into(),
            ))
        }
    };
    let operation = |error: std::io::Error| ServiceError::Operation(error.to_string());
    let mailbox =
        crate::mailbox::Mailbox::open(&crate::mailbox::mail_root(), &address).map_err(operation)?;
    if params.address.is_some() {
        let messages: Vec<Value> = mailbox
            .list()
            .map_err(operation)?
            .into_iter()
            .filter(|stored| params.all || stored.state == crate::mailbox::MailState::Unread)
            .map(|stored| json!({"state": match stored.state { crate::mailbox::MailState::Read => "read", crate::mailbox::MailState::Unread => "unread" }, "envelope": stored.envelope, "rendered": stored.envelope.render()}))
            .collect();
        return Ok(json!({"address": address.to_string(), "messages": messages, "looked": true}));
    }
    let reader = address.to_string();
    let claimed = mailbox
        .claim_unread("sessions.inbox", Some(&reader))
        .map_err(operation)?;
    let mut messages: Vec<Value> = Vec::new();
    if params.all {
        for stored in mailbox.list().map_err(operation)? {
            if stored.state == crate::mailbox::MailState::Read {
                messages.push(json!({"state": "read", "envelope": stored.envelope, "rendered": stored.envelope.render()}));
            }
        }
    }
    for stored in &claimed {
        messages.push(json!({"state": "unread", "envelope": stored.envelope, "rendered": stored.envelope.render()}));
    }
    for stored in &claimed {
        mailbox.acknowledge(stored).map_err(operation)?;
    }
    Ok(json!({"address": address.to_string(), "messages": messages}))
}

/// Source identity of one follow subscription, plus the last lifecycle state
/// already reported on it. The follower itself stays purely persistence-facing.
// Only the adapter-api poll reads these; the subscription bookkeeping itself is
// shared by both builds.
#[cfg_attr(not(feature = "adapter-api"), allow(dead_code))]
struct FollowedSource {
    harness: String,
    session_id: String,
    reported: Option<String>,
}

#[cfg_attr(not(feature = "adapter-api"), allow(dead_code))]
struct ActivitySubscription {
    locators: Vec<SessionLocator>,
    homes: crate::HarnessHomes,
    reported: BTreeMap<(String, String), crate::SessionActivity>,
}

/// Add what makes an indexed row behaviorally equivalent to a discovered row: the attach
/// endpoint of a session supercode hosts, and the door a message reaches it by.
///
/// The durable index owns only persistence metadata. Both are projections: every message/attach
/// operation revalidates its authority, so publishing one here never trusts a stale browser-held
/// handle. The doors are read once per batch ([`crate::mail_route::LiveSessions`]).
fn live_descriptor_value(
    session: &SessionDescriptor,
    doors: &crate::mail_route::LiveSessions,
) -> std::result::Result<Value, ServiceError> {
    let mut value = serde_json::to_value(session)
        .map_err(|error| ServiceError::Operation(error.to_string()))?;
    // A running session with no title of its own is known by its live name.
    if value.get("title").is_none_or(Value::is_null) {
        if let Some(live) = doors.all().iter().find(|live| {
            live.address.harness == session.locator.harness.as_str()
                && live.address.session_id == session.locator.session_id
        }) {
            let name = live.name.split('@').next().unwrap_or(&live.name);
            if !name.is_empty() {
                value["title"] = json!(name);
            }
        }
    }
    if let Some(workspace) = &session.cwd {
        let source = LiveRuntimeSource {
            harness: session.locator.harness.as_str().to_string(),
            session_id: session.locator.session_id.clone(),
            workspace: workspace.clone(),
        };
        if let Some(endpoint) = discover_live_runtime(&source)
            .map_err(|error| ServiceError::Operation(error.to_string()))?
        {
            value["live_endpoint"] = json!(endpoint.as_str());
        }
    }
    // How a message reaches this session right now, chosen by the same
    // router every sender uses: `runtime` (supercode controls it), `native`,
    // `hook` or `stored`. Absent when no process is running it.
    if let Some(door) = doors.door(
        session.locator.harness.as_str(),
        &session.locator.session_id,
    ) {
        value["delivery"] = json!(door);
    }
    // What a session waiting on its user is asking, so every surface can show the question and not only the state.
    // Read as `message list` reads it: the transcript's open call, else the prompt on the session's pane.
    if let Some(live) = doors.all().iter().find(|live| {
        live.address.harness == session.locator.harness.as_str()
            && live.address.session_id == session.locator.session_id
            && live.status == "waiting"
    }) {
        if let Some(request) = live.pending_request(&crate::HarnessHomes::default()) {
            value["pending_request"] = request;
        }
    }
    Ok(value)
}

fn live_index_changes(
    changes: Vec<crate::session_index::SessionIndexChange>,
    homes: &HarnessHomes,
) -> std::result::Result<Vec<Value>, ServiceError> {
    use crate::session_index::SessionIndexChange;
    let doors = crate::mail_route::LiveSessions::read(homes);
    changes
        .into_iter()
        .map(|change| match change {
            SessionIndexChange::Added { descriptor } => Ok(json!({
                "kind": "added",
                "descriptor": live_descriptor_value(&descriptor, &doors)?,
            })),
            SessionIndexChange::Updated { descriptor } => Ok(json!({
                "kind": "updated",
                "descriptor": live_descriptor_value(&descriptor, &doors)?,
            })),
            SessionIndexChange::Removed { key } => Ok(json!({
                "kind": "removed",
                "key": key,
            })),
        })
        .collect()
}

fn legacy_live_status(activity: &crate::SessionActivity) -> Option<&'static str> {
    use crate::{SessionPresence, SessionTurnState};
    match (activity.presence, activity.turn) {
        (SessionPresence::Persisted, _) => None,
        (SessionPresence::Running, SessionTurnState::Working) => Some("busy"),
        (SessionPresence::Running, SessionTurnState::Idle) => Some("idle"),
        // Waiting on its user (a question, an approval): the word every other surface uses.
        (SessionPresence::Running, SessionTurnState::NeedsInput) => Some("waiting"),
        // The normalized activity object can honestly report a live owner even
        // when the stock harness never published a turn status. Preserve the
        // older field's stricter contract instead of guessing `running`.
        (SessionPresence::Running, SessionTurnState::Unknown)
            if activity.evidence.native_state.is_none() =>
        {
            None
        }
        (SessionPresence::Running, _) | (SessionPresence::ShuttingDown, _) => Some("running"),
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "kebab-case")]
enum TransferFormat {
    ClaudeCode,
    Codex,
    #[serde(rename = "opencode", alias = "open-code")]
    OpenCode,
    Pi,
    Grok,
    Gemini,
    Goose,
    /// UNI-18: a Hermes target. Its artifact is the Codex rollout that
    /// `hermes sessions import --from codex` reads; `sessions.export` performs
    /// that import into the Hermes home.
    Hermes,
}

impl TransferFormat {
    fn id(self) -> &'static str {
        match self {
            Self::ClaudeCode => HarnessId::CLAUDE_CODE,
            Self::Codex => HarnessId::CODEX,
            Self::OpenCode => HarnessId::OPENCODE,
            Self::Pi => HarnessId::PI,
            Self::Grok => HarnessId::GROK,
            Self::Gemini => HarnessId::GEMINI,
            Self::Goose => HarnessId::GOOSE,
            Self::Hermes => HarnessId::HERMES,
        }
    }
}

impl From<TransferFormat> for SessionFormat {
    fn from(value: TransferFormat) -> Self {
        match value {
            TransferFormat::ClaudeCode => Self::ClaudeCode,
            TransferFormat::Codex => Self::Codex,
            TransferFormat::OpenCode => Self::OpenCode,
            TransferFormat::Pi => Self::Pi,
            TransferFormat::Grok => Self::Grok,
            TransferFormat::Gemini => Self::Gemini,
            TransferFormat::Goose => Self::Goose,
            // a Hermes artifact is the Codex rollout Hermes imports
            TransferFormat::Hermes => Self::Codex,
        }
    }
}

#[derive(Deserialize)]
struct ImportSessionParams {
    source_harness: TransferFormat,
    content: String,
}

#[derive(Deserialize)]
struct ExportSessionParams {
    locator: SessionLocator,
    target_harness: TransferFormat,
}

#[derive(Deserialize)]
struct ReduceSessionParams {
    locator: SessionLocator,
    target_harness: TransferFormat,
    #[serde(default = "default_keep_last")]
    keep_last: usize,
}

fn default_keep_last() -> usize {
    6
}

#[derive(Deserialize)]
struct BranchSessionParams {
    locator: SessionLocator,
    #[serde(default)]
    target_harness: Option<TransferFormat>,
    /// Fork at a turn: the user message the branch replaces, by its native id; the branch keeps
    /// only the messages before it. See [`MessageId`].
    #[serde(default)]
    before_message_id: Option<MessageId>,
}

/// A message named by the id its harness recorded for it, as the session's message metadata carries
/// it (`key`: Claude Code's `claude_uuid`, the record's own uuid). A fork names a turn only this way:
/// never by position or text, which a client sees through bounded display tails and live appends.
/// A harness that records no per-message id (Codex keeps a `turn_id` per turn, shared with the
/// context items of that turn) cannot be forked at a turn and is refused.
#[derive(Deserialize)]
struct MessageId {
    key: String,
    value: String,
}

/// The metadata keys that carry a harness's own per-message id.
const MESSAGE_ID_KEYS: &[&str] = &["claude_uuid"];

#[derive(Deserialize)]
struct HandoffSessionParams {
    locator: SessionLocator,
    target_harness: TransferFormat,
    #[serde(default)]
    cwd: Option<PathBuf>,
    /// As for branch: continue from just before this user message, by its native id, as a new
    /// native session.
    #[serde(default)]
    before_message_id: Option<MessageId>,
}

/// The 1-based place of the user message `id` names: exactly one user message carries it.
fn message_place(session: &Session, id: &MessageId) -> std::result::Result<usize, ServiceError> {
    if !MESSAGE_ID_KEYS.contains(&id.key.as_str()) {
        return Err(ServiceError::UnsupportedAction(format!(
            "{} is not a per-message id supercode reads; a fork at a turn names a message by {}",
            id.key,
            MESSAGE_ID_KEYS.join(", ")
        )));
    }
    let mut places = session
        .messages
        .iter()
        .enumerate()
        .filter(|(_, message)| {
            matches!(message.role, Role::User)
                && message.metadata.get(&id.key).map(String::as_str) == Some(id.value.as_str())
        })
        .map(|(index, _)| index + 1);
    match (places.next(), places.next()) {
        (Some(place), None) => Ok(place),
        (None, _) => Err(ServiceError::Operation(format!(
            "no user message in this session has {} {}",
            id.key, id.value
        ))),
        (Some(_), Some(_)) => Err(ServiceError::Operation(format!(
            "more than one user message has {} {}; it does not name one turn",
            id.key, id.value
        ))),
    }
}

/// The session as loaded, or cut before the user message `id` names for a fork at a turn. Answers the
/// cut's 1-based place with it.
fn session_before(
    session: Session,
    id: Option<&MessageId>,
) -> std::result::Result<(Session, Option<usize>), ServiceError> {
    let before_message = id.map(|id| message_place(&session, id)).transpose()?;
    match before_message {
        None => Ok((session, None)),
        Some(place) => Ok((session.cut_before(place - 1), Some(place))),
    }
}

#[derive(Deserialize)]
struct MaterializeSessionParams {
    artifact: crate::native_materialize::MaterializeArtifact,
    cwd: PathBuf,
    /// Where the continuation is written; unset roots are the environment's own, as discovery reads them.
    #[serde(default)]
    homes: HarnessHomes,
}

/// A session supercode starts or resumes runs without approval prompts
/// (`yolo`) unless the caller asks for the harness's own (`default`).
#[derive(Debug, Clone, Copy, Default, Deserialize)]
#[serde(rename_all = "snake_case")]
enum ResumePolicy {
    Default,
    #[default]
    Yolo,
}

#[derive(Deserialize)]
struct ResumeInstructionsParams {
    locator: SessionLocator,
    #[serde(default)]
    cwd: Option<PathBuf>,
    #[serde(default)]
    policy: ResumePolicy,
}

/// `harness.v1.workflow.load` parameters: which harness's board, and its home.
#[derive(Deserialize)]
struct WorkflowLoadParams {
    /// The store, when the caller names it; else the one the home declares.
    #[serde(default)]
    from: Option<crate::workflow_doors::WorkflowHarness>,
    home: PathBuf,
}

/// ONT-4 `harness.v1.orchestration.load` parameters. `flavor` says which layout the
/// folder is read as; our own is the default.
#[derive(Deserialize)]
struct OrchestrationLoadParams {
    root: PathBuf,
    #[serde(default)]
    flavor: crate::orchestration_doors::HomeFlavor,
}

/// ONT-4 `harness.v1.orchestration.save` parameters. `vault` is merged into the
/// home's own secrets; a caller that sends none keeps what is on disk.
#[derive(Deserialize)]
struct OrchestrationSaveParams {
    root: PathBuf,
    orchestration: crate::orchestration::Orchestration,
    #[serde(default)]
    vault: BTreeMap<String, String>,
}

/// ONT-4 `harness.v1.orchestration.compile` parameters.
#[derive(Deserialize)]
struct OrchestrationCompileParams {
    from: crate::orchestration_doors::OrchestrationHarness,
    home: PathBuf,
}

/// ONT-4 `harness.v1.orchestration.decompile` parameters. `source` is the home the
/// orchestration was compiled from: it is re-compiled to recover the io bookkeeping
/// that byte reuse and the live-store refusal (UNI-18) are decided from.
#[derive(Deserialize)]
struct OrchestrationDecompileParams {
    to: crate::orchestration_doors::OrchestrationHarness,
    orchestration: crate::orchestration::Orchestration,
    source: PathBuf,
    #[serde(default)]
    source_flavor: crate::orchestration_doors::SourceFlavor,
    dest: PathBuf,
    #[serde(default)]
    vault: BTreeMap<String, String>,
}

/// `harness.v1.orchestration.import` parameters: another harness's home, and the
/// folder of ours it becomes.
#[derive(Deserialize)]
struct OrchestrationImportParams {
    from: crate::orchestration_doors::OrchestrationHarness,
    home: PathBuf,
    into: PathBuf,
}

/// `harness.v1.orchestration.export` parameters: a folder of ours, and the home of
/// another harness it becomes.
#[derive(Deserialize)]
struct OrchestrationExportParams {
    to: crate::orchestration_doors::OrchestrationHarness,
    root: PathBuf,
    dest: PathBuf,
}

/// `harness.v1.jobs.get` parameters.
#[derive(Deserialize)]
struct JobsGetParams {
    harness: String,
    id: String,
    #[serde(default)]
    homes: crate::HarnessHomes,
}

/// ORCH-18: run one mutating job verb through the harness's own CLI.
///
/// The refusal ladder is deliberate: a harness with no scheduled-job concept
/// at all answers with the SAME sentence `jobs.list` gives it, and a harness
/// that has jobs but publishes no client-callable verb (Claude Code, whose
/// jobs are created by the model inside a session) answers with its own
/// reason. Neither is ever a silent no-op.
fn mutate_job(
    verb: crate::jobs_control::JobVerb,
    params: Value,
) -> std::result::Result<Value, ServiceError> {
    let mutation = decode::<crate::jobs_control::JobMutation>(params)?;
    refuse_harness_without_jobs(&mutation.harness, &format!("jobs.{}", verb.as_str()))?;
    let outcome = crate::jobs_control::mutate(verb, &mutation).map_err(job_control_error)?;
    serde_json::to_value(outcome).map_err(|error| ServiceError::Operation(error.to_string()))
}

/// ORCH-22: run one mutating skills verb through the harness's own door.
///
/// The refusal ladder mirrors `jobs.*`: a harness with no skills root at all
/// answers with the same sentence `skills.list` gives it, and a harness whose
/// door does not publish this verb (OpenClaw has no `skills remove` at the
/// pin) answers with its own reason. Neither is ever a silent no-op.
fn mutate_skill(
    verb: crate::skills_control::SkillVerb,
    params: Value,
) -> std::result::Result<Value, ServiceError> {
    let mutation = decode::<crate::skills_control::SkillMutation>(params)?;
    if !crate::skills_control::supports_skill_control(&mutation.harness) {
        return Err(ServiceError::UnsupportedAction(format!(
            "`{}` has no skills root Volter Harness reads; `skills.{}` is supported for: {}",
            mutation.harness,
            verb.as_str(),
            crate::skills_control::CONTROLLED_SKILL_HARNESSES.join(", ")
        )));
    }
    let outcome =
        crate::skills_control::mutate_skill(verb, &mutation).map_err(skill_control_error)?;
    serde_json::to_value(outcome).map_err(|error| ServiceError::Operation(error.to_string()))
}

/// The skills twin of [`job_control_error`], with the same mapping rule.
fn skill_control_error(error: crate::skills_control::SkillControlError) -> ServiceError {
    match error {
        crate::skills_control::SkillControlError::Unsupported(message) => {
            ServiceError::UnsupportedAction(message)
        }
        crate::skills_control::SkillControlError::Invalid(message) => {
            ServiceError::InvalidParams(message)
        }
        crate::skills_control::SkillControlError::Failed(message) => {
            ServiceError::Operation(message)
        }
    }
}

/// ORCH-21: run one mutating profile verb through the harness's own CLI.
///
/// The refusal ladder mirrors `mutate_job`'s: a harness with no profile
/// concept at all answers with the SAME sentence `profiles.list` gives it, and
/// a harness that HAS profiles but publishes no client-callable verb (Codex's
/// file-authored `[profiles.<name>]` tables, supercode's compiled-in presets)
/// answers with its own reason. Neither is ever a silent no-op.
fn mutate_profile(
    verb: crate::profiles_control::ProfileVerb,
    params: Value,
) -> std::result::Result<Value, ServiceError> {
    let mutation = decode::<crate::profiles_control::ProfileMutation>(params)?;
    let outcome =
        crate::profiles_control::mutate(verb, &mutation).map_err(profile_control_error)?;
    serde_json::to_value(outcome).map_err(|error| ServiceError::Operation(error.to_string()))
}

/// The same mapping `job_control_error` applies, for the profile noun.
fn profile_control_error(error: crate::profiles_control::ProfileControlError) -> ServiceError {
    match error {
        crate::profiles_control::ProfileControlError::Unsupported(message) => {
            ServiceError::UnsupportedAction(message)
        }
        crate::profiles_control::ProfileControlError::Invalid(message) => {
            ServiceError::InvalidParams(message)
        }
        crate::profiles_control::ProfileControlError::Failed(message) => {
            ServiceError::Operation(message)
        }
    }
}

/// Map a controlled-tier failure onto the service's error vocabulary. A verb
/// the harness lacks is `UnsupportedAction`; a harness verb that RAN and
/// failed carries its own stderr through as the operation error.
fn job_control_error(error: crate::jobs_control::JobControlError) -> ServiceError {
    match error {
        crate::jobs_control::JobControlError::Unsupported(message) => {
            ServiceError::UnsupportedAction(message)
        }
        crate::jobs_control::JobControlError::Invalid(message) => {
            ServiceError::InvalidParams(message)
        }
        crate::jobs_control::JobControlError::Failed(message) => ServiceError::Operation(message),
    }
}

/// Map an ORCH-19 controlled-tier failure onto the service's error
/// vocabulary. A verb the harness has no door for is `UnsupportedAction`; a
/// door that RAN and failed carries the harness's own stderr / HTTP body
/// through as the operation error.
fn session_control_error(error: crate::SessionControlError) -> ServiceError {
    match error {
        crate::SessionControlError::Unsupported(message) => {
            ServiceError::UnsupportedAction(message)
        }
        crate::SessionControlError::Invalid(message) => ServiceError::InvalidParams(message),
        crate::SessionControlError::Failed(message) => ServiceError::Operation(message),
    }
}

/// A harness without a scheduled-job concept refuses the verb rather than
/// answering with an empty list — an absent capability and an empty inventory
/// are different answers (the same rule `runtimes.capabilities` applies to
/// `steer`).
fn refuse_harness_without_jobs(harness: &str, verb: &str) -> std::result::Result<(), ServiceError> {
    if crate::jobs::supports_jobs(harness) {
        return Ok(());
    }
    Err(ServiceError::UnsupportedAction(format!(
        "`{harness}` has no scheduled jobs; `{verb}` is supported for: {}",
        crate::jobs::JOB_HARNESSES.join(", ")
    )))
}

/// `harness.v1.runs.get` parameters.
#[derive(Deserialize)]
struct RunsGetParams {
    harness: String,
    id: String,
    #[serde(default)]
    homes: crate::HarnessHomes,
}

/// A harness with no run store refuses the verb rather than answering with an
/// empty history — the same rule `jobs.list` applies. Claude Code lands here
/// on purpose: its cron fires are ordinary turns inside the session that
/// created the job, so there is no fire record to list.
fn refuse_harness_without_runs(harness: &str, verb: &str) -> std::result::Result<(), ServiceError> {
    if crate::runs::supports_runs(harness) {
        return Ok(());
    }
    Err(ServiceError::UnsupportedAction(format!(
        "`{harness}` keeps no run store; `{verb}` is supported for: {}",
        crate::runs::RUN_HARNESSES.join(", ")
    )))
}

#[derive(Serialize)]
struct SessionArtifact {
    source_harness: HarnessId,
    target_harness: &'static str,
    session_id: Option<String>,
    content: String,
    suggested_filename: String,
    files: Vec<SessionArtifactFile>,
    fidelity: Fidelity,
    residue: Vec<String>,
}

#[derive(Serialize)]
struct SessionArtifactFile {
    path: String,
    content: String,
    role: ArtifactFileRole,
}

#[derive(Serialize)]
#[serde(rename_all = "snake_case")]
enum ArtifactFileRole {
    Primary,
    Subagent,
    Bundle,
    SourceRecovery,
}

#[derive(Serialize)]
struct StructuredLaunch {
    cwd: PathBuf,
    program: String,
    arguments: Vec<String>,
    env: BTreeMap<String, String>,
}

struct HandoffInstructions {
    launch: StructuredLaunch,
    materialize: Option<StructuredLaunch>,
    requires_materialization: bool,
    note: String,
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum HarnessProbeLevel {
    #[default]
    Passive,
    Handshake,
}

#[derive(Default, Deserialize)]
#[serde(default)]
struct HarnessInventoryParams {
    harness: Option<HarnessId>,
    harnesses: Vec<HarnessId>,
    workspace: Option<PathBuf>,
    probe: HarnessProbeLevel,
    include_sessions: bool,
    /// Omit subprocess-based `--version` calls when a latency-sensitive UI only needs readiness.
    skip_versions: bool,
}

#[derive(Deserialize)]
struct HarnessAuthenticationParams {
    harness: HarnessId,
}

#[derive(Deserialize)]
struct BeginHarnessAuthenticationParams {
    harness: HarnessId,
    #[serde(default = "local_browser_authentication_environment")]
    environment: crate::HarnessAuthenticationEnvironment,
    #[serde(default)]
    method: Option<crate::HarnessAuthenticationMethodId>,
    #[serde(default)]
    cwd: Option<PathBuf>,
}

fn local_browser_authentication_environment() -> crate::HarnessAuthenticationEnvironment {
    crate::HarnessAuthenticationEnvironment::LocalBrowser
}

#[derive(Serialize)]
struct HarnessInventoryReport {
    probe: HarnessProbeLevel,
    workspace: Option<PathBuf>,
    harnesses: Vec<LocalHarness>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
enum HarnessAuthState {
    Ready,
    Configured,
    Required,
    Unknown,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
enum HarnessRuntimeState {
    Ready,
    Degraded,
    Unavailable,
}

#[derive(Serialize)]
struct HarnessSessionCounts {
    global: Option<usize>,
    workspace: Option<usize>,
}

/// Receipt-backed evidence that a harness has a RUNNING instance right now,
/// distinct from being merely installed (UNI-7). Detection is passive and
/// default-on: a gateway liveness connect for daemon harnesses, a fresh
/// SQLite WAL stamp for store-writer harnesses (precedent: the opencode
/// follower's -wal/-shm freshness). Control stays behind per-connection
/// grants — this reports observations only.
/// ORCH-17: the gateway-health noun on an inventory row. Derived from the
/// UNI-7 running-instance probe (Hermes: `state.db-wal` freshness; OpenClaw:
/// a TCP connect to the gateway endpoint resolved from its OWN config) plus
/// the executable version — never by starting anything.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum GatewayState {
    Up,
    Down,
    Unknown,
}

/// ORCH-17: `gateway` on a `harness.v1.harnesses.list` row.
#[derive(Debug, Clone, Serialize)]
pub struct GatewayHealth {
    pub state: GatewayState,
    /// The endpoint supercode would connect to (OpenClaw: the gateway
    /// WebSocket resolved from `openclaw.json`; core harnesses: their
    /// declared connect address when one exists). `None` when the harness
    /// has no single endpoint (Hermes multiplexes platforms).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub endpoint: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub version: Option<String>,
    /// What the verdict rests on, or why it is `unknown`.
    pub evidence: String,
    pub checked_at_ms: u64,
}

/// OpenClaw's gateway WebSocket endpoint, resolved from its own config the
/// way the registry's connect descriptor prescribes (`gateway.url`, else
/// `gateway.port`, else the documented default).
fn openclaw_gateway_endpoint(home: &Path) -> String {
    let config_path = home.join(".openclaw/openclaw.json");
    let gateway = std::fs::read_to_string(&config_path)
        .ok()
        .and_then(|raw| serde_json::from_str::<serde_json::Value>(&raw).ok())
        .and_then(|config| config.get("gateway").cloned());
    if let Some(url) = gateway
        .as_ref()
        .and_then(|gateway| gateway.get("url"))
        .and_then(serde_json::Value::as_str)
    {
        return url.to_string();
    }
    let port = gateway
        .as_ref()
        .and_then(|gateway| gateway.get("port"))
        .and_then(serde_json::Value::as_u64)
        .unwrap_or(18789);
    format!("ws://127.0.0.1:{port}")
}

/// Ask Hermes itself (`hermes gateway status`, read-only, ~1 s) whether its
/// gateway is up. The command is per-host launchd/systemd text without a JSON
/// form at 0.19–0.21; the verdict is read from the lines it prints:
/// "supervised by launchd (PID …)" / "is running" → up, "not running" /
/// "not installed" → down, anything else → no verdict. `SUPERCODE_HERMES_BIN`
/// overrides the executable so a fake can stand in under test.
fn hermes_gateway_status() -> Option<(GatewayState, String)> {
    let program = crate::harness_command::harness_program(HarnessId::HERMES).ok()?;
    let output = std::process::Command::new(&program)
        .args(["gateway", "status"])
        .stdin(std::process::Stdio::null())
        .output()
        .ok()?;
    let text = format!(
        "{}{}",
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    let verdict = text.lines().find_map(|line| {
        let l = line.trim();
        if l.contains("supervised by launchd (PID")
            || l.contains("supervised by systemd (PID")
            || l.contains("Gateway is running")
            || l.contains("process is running")
        {
            Some((GatewayState::Up, format!("`hermes gateway status`: {l}")))
        } else if l.contains("not running") || l.contains("not installed") {
            Some((GatewayState::Down, format!("`hermes gateway status`: {l}")))
        } else {
            None
        }
    });
    verdict
}

fn gateway_health(
    id: &str,
    installed: bool,
    running: Option<&RunningInstance>,
    version: Option<&str>,
) -> GatewayHealth {
    let checked_at_ms = now_epoch_ms();
    let home = supercode_interchange::user_home()
        .map(std::path::PathBuf::into_os_string)
        .map(PathBuf::from);
    match id {
        HarnessId::HERMES | HarnessId::OPENCLAW => {
            let endpoint = (id == HarnessId::OPENCLAW)
                .then(|| home.as_deref().map(openclaw_gateway_endpoint))
                .flatten();
            let (state, evidence) = match running {
                Some(instance) => (GatewayState::Up, instance.evidence.clone()),
                None if !installed => (
                    GatewayState::Unknown,
                    format!("`{id}` is not installed; no gateway to probe"),
                ),
                None if id == HarnessId::HERMES => match hermes_gateway_status() {
                    // The harness's own door outranks the WAL heuristic: an idle
                    // gateway writes nothing for minutes yet is up.
                    Some((state, evidence)) => (state, evidence),
                    None => (
                        GatewayState::Down,
                        "no fresh state.db-wal activity under ~/.hermes and `hermes gateway status` gave no verdict".to_string(),
                    ),
                },
                None => (
                    GatewayState::Down,
                    format!(
                        "no TCP listener at {}",
                        endpoint.as_deref().unwrap_or("the gateway endpoint")
                    ),
                ),
            };
            GatewayHealth {
                state,
                endpoint,
                version: version.map(str::to_string),
                evidence,
                checked_at_ms,
            }
        }
        // ORC-7: the orchestrator's gateway IS its daemon, and the daemon's
        // own lease file is the record of it. A lease naming a live pid is
        // up; a lease whose process is gone is down and says so as a STALE
        // lease, never as "no lease"; no lease at all is down. Nothing is
        // started, and no port is guessed — the daemon multiplexes adapters
        // the way Hermes does, so it has no single endpoint either.
        HarnessId::ORCHESTRATOR => {
            let root = crate::HarnessHomes::default().orchestrator;
            let (state, evidence) = match crate::orchestrator::read_lease(&root) {
                Some(lease) if lease.is_live() => (
                    GatewayState::Up,
                    format!(
                        "`{}` names pid {} (started {}), which is live",
                        crate::orchestrator::lock_path(&root).display(),
                        lease.pid,
                        lease.started_at
                    ),
                ),
                Some(lease) => (
                    GatewayState::Down,
                    format!(
                        "stale lease `{}`: pid {} is gone",
                        crate::orchestrator::lock_path(&root).display(),
                        lease.pid
                    ),
                ),
                None => (
                    GatewayState::Down,
                    format!(
                        "no lease at `{}`; `supercode orchestrator start` writes one",
                        crate::orchestrator::lock_path(&root).display()
                    ),
                ),
            };
            GatewayHealth {
                state,
                endpoint: None,
                version: version.map(str::to_string),
                evidence,
                checked_at_ms,
            }
        }
        _ => GatewayHealth {
            state: GatewayState::Unknown,
            endpoint: None,
            version: version.map(str::to_string),
            evidence: format!("`{id}` runs per session, not as a gateway"),
            checked_at_ms,
        },
    }
}

#[derive(Debug, Clone, Serialize)]
struct RunningInstance {
    /// How the instance was detected.
    method: RunningInstanceMethod,
    /// The evidence the verdict rests on (endpoint reached / WAL path+age).
    evidence: String,
    /// Epoch-ms instant the probe executed.
    checked_at_ms: u64,
}

#[derive(Debug, Clone, Copy, Serialize)]
#[serde(rename_all = "snake_case")]
enum RunningInstanceMethod {
    /// A TCP connect to the harness's own configured gateway endpoint
    /// succeeded.
    GatewayConnect,
    /// The harness's session store has an active SQLite WAL (a live writer
    /// holds the store open and stamped it recently).
    StoreWalActivity,
}

fn now_epoch_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|elapsed| elapsed.as_millis() as u64)
        .unwrap_or(0)
}

/// OpenClaw: the gateway endpoint comes from the harness's OWN config
/// (`<home>/.openclaw/openclaw.json` — `gateway.url` or `gateway.port`,
/// default port 18789); a successful TCP connect is the running signal.
fn probe_openclaw_running(home: &Path) -> Option<RunningInstance> {
    let config_path = home.join(".openclaw/openclaw.json");
    let text = std::fs::read_to_string(&config_path).ok();
    let gateway = text
        .as_deref()
        .and_then(|raw| serde_json::from_str::<serde_json::Value>(raw).ok())
        .and_then(|config| config.get("gateway").cloned());
    let address = gateway
        .as_ref()
        .and_then(|gateway| gateway.get("url"))
        .and_then(serde_json::Value::as_str)
        .and_then(|url| {
            url.split("://").nth(1).map(|rest| {
                rest.trim_end_matches('/')
                    .split('/')
                    .next()
                    .unwrap_or(rest)
                    .to_string()
            })
        })
        .unwrap_or_else(|| {
            let port = gateway
                .as_ref()
                .and_then(|gateway| gateway.get("port"))
                .and_then(serde_json::Value::as_u64)
                .unwrap_or(18789);
            format!("127.0.0.1:{port}")
        });
    let reachable = std::net::TcpStream::connect_timeout(
        &address.parse().ok()?,
        std::time::Duration::from_millis(400),
    )
    .is_ok();
    reachable.then(|| RunningInstance {
        method: RunningInstanceMethod::GatewayConnect,
        evidence: format!(
            "gateway endpoint {address} accepted a TCP connect (from {})",
            config_path.display()
        ),
        checked_at_ms: now_epoch_ms(),
    })
}

/// Hermes: `<home>/.hermes/state.db-wal` freshly modified means a live writer
/// holds the store open (SQLite WAL exists only while a connection is open;
/// a recent stamp distinguishes an active instance from a stale crash
/// leftover).
fn probe_hermes_running(home: &Path, max_wal_age_ms: u64) -> Option<RunningInstance> {
    let wal = home.join(".hermes/state.db-wal");
    let modified = std::fs::metadata(&wal).ok()?.modified().ok()?;
    let age_ms = std::time::SystemTime::now()
        .duration_since(modified)
        .map(|age| age.as_millis() as u64)
        .unwrap_or(u64::MAX);
    (age_ms <= max_wal_age_ms).then(|| RunningInstance {
        method: RunningInstanceMethod::StoreWalActivity,
        evidence: format!(
            "{} stamped {age_ms}ms ago (threshold {max_wal_age_ms}ms)",
            wal.display()
        ),
        checked_at_ms: now_epoch_ms(),
    })
}

/// Default-on running-instance detection for the harnesses that have one.
fn probe_running_instance(id: &str) -> Option<RunningInstance> {
    let home = supercode_interchange::user_home()
        .map(std::path::PathBuf::into_os_string)
        .map(PathBuf::from)?;
    match id {
        HarnessId::OPENCLAW => probe_openclaw_running(&home),
        HarnessId::HERMES => probe_hermes_running(&home, 300_000),
        _ => None,
    }
}

#[derive(Serialize)]
struct LocalHarness {
    id: HarnessId,
    display_name: String,
    supported: bool,
    installed: bool,
    executable: Option<String>,
    version: Option<String>,
    auth: HarnessAuthState,
    runtime: HarnessRuntimeState,
    protocol: String,
    capabilities: crate::RuntimeCapabilities,
    effective_capabilities: crate::RuntimeCapabilities,
    sessions: HarnessSessionCounts,
    /// Receipt-backed running-instance detection (None = not detected or the
    /// harness has no running-instance concept). Distinct from `installed`.
    #[serde(skip_serializing_if = "Option::is_none")]
    running: Option<RunningInstance>,
    /// ORCH-17: gateway health derived from `running` + the harness's own config.
    gateway: GatewayHealth,
    reason: Option<String>,
    repair: Option<String>,
}

#[derive(Clone, Deserialize)]
struct RuntimeBackendParams {
    harness: HarnessId,
    #[serde(default)]
    protocol: Option<String>,
    #[serde(default)]
    launch: Option<RuntimeLaunch>,
    #[serde(default)]
    base_url: Option<String>,
    #[serde(default)]
    policy: RuntimePolicy,
}

/// A session supercode starts or resumes runs without approval prompts
/// (`yolo`) unless the caller asks for the harness's own (`default`).
#[derive(Debug, Clone, Copy, Default, Deserialize)]
#[serde(rename_all = "snake_case")]
enum RuntimePolicy {
    Default,
    #[default]
    Yolo,
}

#[derive(Deserialize)]
struct RuntimeStartParams {
    #[serde(flatten)]
    backend: RuntimeBackendParams,
    cwd: PathBuf,
    /// MCP servers to mount into the new session through the harness's own
    /// start door (ORC-6). Backends without such a door ignore them.
    #[serde(default)]
    mcp_servers: Vec<crate::McpServerLaunch>,
    /// The session's approval policy, where the harness's start door takes one (Codex).
    #[serde(default)]
    approval_policy: Option<String>,
    /// The model the session runs, in the harness's own naming, where its start door takes one
    /// (Claude Code, Codex); refused for a harness whose start takes none here.
    #[serde(default)]
    model: Option<String>,
}

#[derive(Deserialize)]
struct RuntimeAttachParams {
    #[serde(flatten)]
    backend: RuntimeBackendParams,
    runtime_id: String,
    #[serde(default)]
    cwd: Option<PathBuf>,
    /// MCP servers to mount into the resumed session (the start door's own
    /// field, carried again because a session's tools die with its process).
    #[serde(default)]
    mcp_servers: Vec<crate::McpServerLaunch>,
    /// The session's approval policy, carried again on resume as on start (Codex).
    #[serde(default)]
    approval_policy: Option<String>,
}

#[derive(Deserialize)]
struct RuntimeConnectionParams {
    connection: String,
}

/// `runtimes.close`: a caller that recorded `runtime_id` with the connection closes it only while the
/// connection still holds that runtime (a restarted daemon numbers its connections afresh).
#[derive(Deserialize)]
struct RuntimeCloseParams {
    #[serde(default)]
    connection: String,
    #[serde(default)]
    runtime_id: Option<String>,
}

#[derive(Deserialize)]
struct RuntimeInputParams {
    connection: String,
    text: String,
    #[serde(default)]
    image_urls: Vec<String>,
}

const MAX_RUNTIME_IMAGES: usize = 4;
const MAX_RUNTIME_IMAGE_URL_BYTES: usize = 12 * 1024 * 1024;
const MAX_RUNTIME_IMAGE_URL_BYTES_TOTAL: usize = 32 * 1024 * 1024;

fn validate_runtime_image_urls(image_urls: Vec<String>) -> Result<Vec<String>, ServiceError> {
    if image_urls.len() > MAX_RUNTIME_IMAGES {
        return Err(ServiceError::InvalidParams(format!(
            "a runtime prompt accepts at most {MAX_RUNTIME_IMAGES} images"
        )));
    }
    let mut total = 0usize;
    for url in &image_urls {
        if !(url.starts_with("data:image/")
            || url.starts_with("https://")
            || url.starts_with("http://"))
        {
            return Err(ServiceError::InvalidParams(
                "runtime images must be image data URLs or HTTP(S) URLs".into(),
            ));
        }
        if url.len() > MAX_RUNTIME_IMAGE_URL_BYTES {
            return Err(ServiceError::InvalidParams(format!(
                "one runtime image exceeds the {MAX_RUNTIME_IMAGE_URL_BYTES}-byte encoded limit"
            )));
        }
        total = total.saturating_add(url.len());
    }
    if total > MAX_RUNTIME_IMAGE_URL_BYTES_TOTAL {
        return Err(ServiceError::InvalidParams(format!(
            "runtime images exceed the {MAX_RUNTIME_IMAGE_URL_BYTES_TOTAL}-byte encoded total limit"
        )));
    }
    Ok(image_urls)
}

#[derive(Deserialize)]
struct RuntimeRespondParams {
    connection: String,
    request_id: Value,
    response: Value,
}

fn default_reduction_store_root() -> PathBuf {
    if let Some(root) = std::env::var_os("SUPERCODE_HOME") {
        return PathBuf::from(root).join("sessions");
    }
    if let Some(home) = supercode_interchange::user_home().map(std::path::PathBuf::into_os_string) {
        return PathBuf::from(home).join(".supercode").join("sessions");
    }
    PathBuf::from(".supercode").join("sessions")
}

fn messages_jsonl(messages: &[crate::ChatMessage]) -> std::result::Result<String, ServiceError> {
    let mut output = String::new();
    for message in messages {
        output.push_str(
            &serde_json::to_string(message)
                .map_err(|error| ServiceError::Operation(error.to_string()))?,
        );
        output.push('\n');
    }
    Ok(output)
}

fn parse_messages_jsonl(
    content: &str,
) -> std::result::Result<Vec<crate::ChatMessage>, ServiceError> {
    content
        .lines()
        .enumerate()
        .filter(|(_, line)| !line.trim().is_empty())
        .map(|(index, line)| {
            serde_json::from_str::<crate::ChatMessage>(line).map_err(|error| {
                ServiceError::Operation(format!(
                    "reduced transcript line {} is invalid: {error}",
                    index + 1
                ))
            })
        })
        .collect()
}

fn reduced_bootstrap_prompt(
    source: &SessionLocator,
    target: TransferFormat,
    view_jsonl: &str,
    sidecar_path: &Path,
    reduction_log_path: &Path,
) -> String {
    format!(
        "Continue the work from this losslessly reduced {source_harness} session in {target_harness}.\n\
         \n\
         The bounded working transcript is below. Treat reduction markers as transparent placeholders, not missing work. If a detail behind a marker is needed, use ordinary file-reading/search tools against the full Volter Harness sidecar at `{sidecar}` and its reduction index at `{log}`. Do not guess hidden content. Both files were reloaded and verified before this continuation was issued.\n\
         \n\
         <supercode-reduced-session source-session=\"{source_id}\">\n\
         {view_jsonl}\
         </supercode-reduced-session>\n\
         \n\
         Resume from the latest unresolved user request and preserve the source session's decisions and constraints.",
        source_harness = source.harness.as_str(),
        target_harness = target.id(),
        sidecar = sidecar_path.display(),
        log = reduction_log_path.display(),
        source_id = source.session_id,
    )
}

fn session_artifact(
    locator: &SessionLocator,
    session: &Session,
    target: TransferFormat,
) -> std::result::Result<SessionArtifact, ServiceError> {
    session_artifact_with_id(locator, session, target, None)
}

fn session_artifact_with_id(
    locator: &SessionLocator,
    session: &Session,
    target: TransferFormat,
    target_session_id: Option<&str>,
) -> std::result::Result<SessionArtifact, ServiceError> {
    let format: SessionFormat = target.into();
    let diagonal = format.source() == session.meta.source;
    crate::residue_store::store_segments(session);
    let has_appended_turns = session
        .imported_message_count
        .is_some_and(|imported| imported < session.messages.len());
    let mut restoration = None;
    let content = if let Some(id) = target_session_id {
        if diagonal && format != SessionFormat::OpenCode {
            session
                .to_jsonl_spliced(format, Some(id))
                .map_err(operation)?
        } else {
            let mut rewritten = session.clone();
            rewritten.meta.session_id = Some(id.to_string());
            rewritten.to_jsonl(format).map_err(operation)?
        }
    } else if diagonal && session.raw_is_verbatim && !has_appended_turns {
        session.raw_verbatim()
    } else if diagonal {
        session.to_jsonl_spliced(format, None).map_err(operation)?
    } else {
        // A session that came from `format` before returns its source records verbatim for the
        // prefix the residue store holds (docs/plans/portable-residue.md).
        match session
            .restore_residue(format, crate::residue_store::lookup)
            .map_err(operation)?
        {
            Some((content, report)) => {
                restoration = Some(report);
                content
            }
            None => session.to_jsonl(format).map_err(operation)?,
        }
    };
    let stem = sanitize_filename(
        target_session_id
            .or(session.meta.session_id.as_deref())
            .unwrap_or(&locator.session_id),
    );
    let suggested_filename = if diagonal && target == TransferFormat::Grok {
        "chat_history.jsonl".to_string()
    } else if target == TransferFormat::Goose {
        format!("{stem}.goose.json")
    } else {
        format!("{stem}.{}.jsonl", target.id())
    };
    let mut files = vec![SessionArtifactFile {
        path: suggested_filename.clone(),
        content: content.clone(),
        role: ArtifactFileRole::Primary,
    }];
    if target == TransferFormat::ClaudeCode {
        let bundle_stem = Path::new(&suggested_filename)
            .file_stem()
            .and_then(|stem| stem.to_str())
            .unwrap_or(&stem);
        let mut child_paths = BTreeSet::new();
        for (index, subagent) in session.subagents.iter().enumerate() {
            let agent_id = subagent
                .meta
                .agent_id
                .as_deref()
                .map(|id| id.strip_prefix("agent-").unwrap_or(id))
                .map(sanitize_filename)
                .filter(|id| !id.is_empty())
                .unwrap_or_else(|| format!("subagent-{}", index + 1));
            let child_has_appended_turns = subagent
                .imported_message_count
                .is_some_and(|imported| imported < subagent.messages.len());
            let child_content = if target_session_id.is_none()
                && subagent.meta.source == SessionSource::ClaudeCode
                && subagent.raw_is_verbatim
                && !child_has_appended_turns
            {
                subagent.raw_verbatim()
            } else if subagent.meta.source == SessionSource::ClaudeCode {
                subagent
                    .to_jsonl_spliced(SessionFormat::ClaudeCode, target_session_id)
                    .map_err(operation)?
            } else {
                let mut child = subagent.clone();
                if let Some(id) = target_session_id {
                    child.meta.session_id = Some(id.to_string());
                }
                child
                    .to_jsonl(SessionFormat::ClaudeCode)
                    .map_err(operation)?
            };
            let path = format!("{bundle_stem}/subagents/agent-{agent_id}.jsonl");
            if !child_paths.insert(path.clone()) {
                return Err(ServiceError::Operation(format!(
                    "Claude subagent ids collide at artifact path `{path}`"
                )));
            }
            files.push(SessionArtifactFile {
                path,
                content: child_content,
                role: ArtifactFileRole::Subagent,
            });
        }
    }
    if diagonal && target == TransferFormat::Grok {
        append_grok_bundle_files(locator, "", ArtifactFileRole::Bundle, &mut files)?;
    }
    if !diagonal || !session.raw_is_verbatim {
        files.push(SessionArtifactFile {
            path: "recovery/source.supercode.jsonl".into(),
            content: session.to_native_jsonl(),
            role: ArtifactFileRole::SourceRecovery,
        });
        for (index, subagent) in session.subagents.iter().enumerate() {
            let id = subagent
                .meta
                .agent_id
                .as_deref()
                .map(sanitize_filename)
                .unwrap_or_else(|| format!("subagent-{}", index + 1));
            files.push(SessionArtifactFile {
                path: format!("recovery/subagents/{id}.supercode.jsonl"),
                content: subagent.to_native_jsonl(),
                role: ArtifactFileRole::SourceRecovery,
            });
        }
    }
    if !diagonal && session.meta.source == SessionSource::Grok {
        append_grok_bundle_files(
            locator,
            "recovery/grok/",
            ArtifactFileRole::SourceRecovery,
            &mut files,
        )?;
    }
    let (fidelity, residue) = if diagonal
        && target_session_id.is_none()
        && session.raw_is_verbatim
        && !has_appended_turns
    {
        (Fidelity::ByteLossless, Vec::new())
    } else if diagonal && !(target_session_id.is_some() && target == TransferFormat::OpenCode) {
        (
            Fidelity::ValueLossless,
            vec![if target_session_id.is_some() {
                "target identity was rewritten, so the artifact intentionally differs from source bytes".into()
            } else {
                "source storage was reconstructed as a native-value-equivalent export; original container bytes were not captured".into()
            }],
        )
    } else {
        match restoration {
            Some(report) if report.rendered_messages == 0 => (
                Fidelity::ByteLossless,
                vec![format!(
                    "restored verbatim from this conversation's {} source records in the residue store",
                    target.id()
                )],
            ),
            Some(report) => (
                Fidelity::Semantic,
                vec![format!(
                    "{} of {} messages restored verbatim from the residue store; the other {} written by the {} writer",
                    report.restored_messages,
                    report.restored_messages + report.rendered_messages,
                    report.rendered_messages,
                    target.id()
                )],
            ),
            None => (
                Fidelity::Semantic,
                vec!["target schema has no portable slot for every source-native record and metadata field".into()],
            ),
        }
    };
    Ok(SessionArtifact {
        source_harness: locator.harness.clone(),
        target_harness: target.id(),
        session_id: target_session_id
            .map(str::to_string)
            .or_else(|| session.meta.session_id.clone()),
        content,
        suggested_filename,
        files,
        fidelity,
        residue,
    })
}

fn append_grok_bundle_files(
    locator: &SessionLocator,
    prefix: &str,
    role: ArtifactFileRole,
    files: &mut Vec<SessionArtifactFile>,
) -> std::result::Result<(), ServiceError> {
    let primary = locator.storage.path();
    if primary.file_name().and_then(|name| name.to_str()) != Some("chat_history.jsonl") {
        return Err(ServiceError::Operation(format!(
            "Grok bundle locator must name chat_history.jsonl, got {}",
            primary.display()
        )));
    }
    let parent = primary.parent().ok_or_else(|| {
        ServiceError::Operation("Grok chat_history.jsonl has no session directory".into())
    })?;
    for name in ["summary.json", "updates.jsonl"] {
        let path = parent.join(name);
        let metadata = match std::fs::symlink_metadata(&path) {
            Ok(metadata) => metadata,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
            Err(error) => return Err(ServiceError::Operation(error.to_string())),
        };
        if metadata.file_type().is_symlink() || !metadata.is_file() {
            return Err(ServiceError::Operation(format!(
                "refusing non-regular Grok bundle member {}",
                path.display()
            )));
        }
        let content = std::fs::read_to_string(&path).map_err(|error| {
            ServiceError::Operation(format!(
                "Grok bundle member {} is not representable as UTF-8: {error}",
                path.display()
            ))
        })?;
        files.push(SessionArtifactFile {
            path: format!("{prefix}{name}"),
            content,
            role: match role {
                ArtifactFileRole::Bundle => ArtifactFileRole::Bundle,
                _ => ArtifactFileRole::SourceRecovery,
            },
        });
    }
    Ok(())
}

fn handoff_artifact(
    locator: &SessionLocator,
    session: &Session,
    target: TransferFormat,
) -> std::result::Result<SessionArtifact, ServiceError> {
    let target_session_id = target_session_id(target);
    session_artifact_with_id(locator, session, target, Some(&target_session_id))
}

fn target_session_id(target: TransferFormat) -> String {
    let uuid = generated_session_id();
    match target {
        TransferFormat::OpenCode => format!("ses_{}", uuid.replace('-', "")),
        TransferFormat::ClaudeCode
        | TransferFormat::Codex
        | TransferFormat::Pi
        | TransferFormat::Grok
        | TransferFormat::Gemini
        | TransferFormat::Goose
        | TransferFormat::Hermes => uuid,
    }
}

fn sanitize_filename(value: &str) -> String {
    let value = value
        .chars()
        .map(|character| {
            if character.is_ascii_alphanumeric() || matches!(character, '-' | '_') {
                character
            } else {
                '-'
            }
        })
        .collect::<String>();
    let value = value.trim_matches('-');
    if value.is_empty() {
        "session".into()
    } else {
        value.chars().take(100).collect()
    }
}

fn handoff_instructions(
    target: TransferFormat,
    session_id: &str,
    cwd: &Path,
) -> HandoffInstructions {
    let launch = |program: &str, arguments: Vec<String>| StructuredLaunch {
        cwd: cwd.to_path_buf(),
        program: program.into(),
        arguments,
        env: BTreeMap::new(),
    };
    match target {
        TransferFormat::ClaudeCode => HandoffInstructions {
            launch: launch("claude", vec!["--resume".into(), session_id.into()]),
            materialize: None,
            requires_materialization: true,
            note: "Write the artifact into Claude Code's native project session store before running the resume launch; Claude Code has no general transcript-import command.".into(),
        },
        TransferFormat::Hermes => HandoffInstructions {
            launch: launch("hermes", vec!["--resume".into(), session_id.into()]),
            materialize: None,
            requires_materialization: true,
            note: "Hand the artifact (a Codex rollout) to `hermes sessions import --from codex <file>` — `sessions.export --to hermes` does exactly that — and resume the id Hermes prints: Hermes mints its own id and writes its own store.".into(),
        },
        TransferFormat::Codex => HandoffInstructions {
            launch: launch("codex", vec!["resume".into(), session_id.into()]),
            materialize: None,
            requires_materialization: true,
            note: "Write the artifact into Codex's native rollout store before running the resume launch; Codex has no general transcript-import command.".into(),
        },
        TransferFormat::OpenCode => HandoffInstructions {
            launch: launch("opencode", vec!["--session".into(), session_id.into()]),
            materialize: Some(launch(
                "opencode",
                vec!["import".into(), "{artifact_path}".into()],
            )),
            requires_materialization: true,
            note: "Write the artifact to a file, run the materialize command with its path, then launch the imported session.".into(),
        },
        TransferFormat::Pi => HandoffInstructions {
            launch: launch("pi", vec!["--session".into(), "{artifact_path}".into()]),
            materialize: None,
            requires_materialization: true,
            note: "Write the artifact to a file and replace {artifact_path} in the launch arguments; Pi can resume that file directly.".into(),
        },
        TransferFormat::Grok => HandoffInstructions {
            launch: launch(
                "grok",
                vec!["--resume".into(), "{materialized_session_id}".into()],
            ),
            materialize: None,
            requires_materialization: true,
            note: "Grok has no import command. Materialize the artifact through `harness.v1.sessions.materialize` (target `grok`, `value_lossless`, the destination cwd): it writes Grok's store entry (`chat_history.jsonl` and the `summary.json` `--resume` requires) under a fresh id; replace {materialized_session_id} with the id it returns.".into(),
        },
        TransferFormat::Gemini => HandoffInstructions {
            launch: launch(
                "gemini",
                vec!["--session-file".into(), "{artifact_path}".into()],
            ),
            materialize: None,
            requires_materialization: true,
            note: "Write the Gemini JSONL artifact to a file and replace {artifact_path}; Gemini imports it into the current project's chat store before opening the continuation.".into(),
        },
        TransferFormat::Goose => HandoffInstructions {
            launch: launch(
                "goose",
                vec![
                    "session".into(),
                    "--resume".into(),
                    "--session-id".into(),
                    "{imported_session_id}".into(),
                ],
            ),
            materialize: Some(launch(
                "goose",
                vec!["session".into(), "import".into(), "{artifact_path}".into()],
            )),
            requires_materialization: true,
            note: "Write the Goose JSON artifact to a file, run the materialize command, read the imported session id from its output, replace {imported_session_id}, then resume that native Goose session.".into(),
        },
    }
}

fn resume_launch(
    harness: &str,
    session_id: &str,
    cwd: &Path,
    policy: ResumePolicy,
) -> std::result::Result<StructuredLaunch, ServiceError> {
    let mut arguments = Vec::new();
    let program = match harness {
        HarnessId::GROK => {
            if matches!(policy, ResumePolicy::Yolo) {
                if crate::support::self_sandbox_supported() {
                    arguments.extend(["--sandbox".into(), "workspace".into()]);
                }
                arguments.push("--always-approve".into());
            }
            arguments.extend(["--resume".into(), session_id.into()]);
            "grok"
        }
        HarnessId::CODEX => {
            arguments.extend(crate::startup_prompts::startup_arguments(
                harness,
                Some(cwd),
                &[],
                matches!(policy, ResumePolicy::Yolo),
            ));
            arguments.extend(["resume".into(), session_id.into()]);
            "codex"
        }
        HarnessId::CLAUDE_CODE => {
            arguments.extend(crate::startup_prompts::startup_arguments(
                harness,
                Some(cwd),
                &[],
                matches!(policy, ResumePolicy::Yolo),
            ));
            arguments.extend(["--resume".into(), session_id.into()]);
            "claude"
        }
        HarnessId::GEMINI => {
            arguments.extend(crate::startup_prompts::startup_arguments(
                harness,
                Some(cwd),
                &[],
                matches!(policy, ResumePolicy::Yolo),
            ));
            arguments.extend(["--resume".into(), session_id.into()]);
            "gemini"
        }
        HarnessId::GOOSE => {
            arguments.extend([
                "session".into(),
                "--resume".into(),
                "--session-id".into(),
                session_id.into(),
            ]);
            "goose"
        }
        HarnessId::PI => {
            arguments.extend(crate::startup_prompts::startup_arguments(
                harness,
                Some(cwd),
                &[],
                matches!(policy, ResumePolicy::Yolo),
            ));
            arguments.extend(["--session".into(), session_id.into()]);
            "pi"
        }
        HarnessId::OPENCODE => {
            arguments.extend(["--session".into(), session_id.into()]);
            "opencode"
        }
        HarnessId::SUPERCODE => {
            if matches!(policy, ResumePolicy::Yolo) {
                arguments.push("--dangerous".into());
            }
            arguments.extend(["resume".into(), session_id.into()]);
            "supercode"
        }
        other => {
            return Err(ServiceError::InvalidParams(format!(
                "no structured resume launch is registered for harness `{other}`"
            )))
        }
    };
    Ok(StructuredLaunch {
        cwd: cwd.to_path_buf(),
        env: if program == "grok" {
            crate::support::grok_home_env()
        } else {
            BTreeMap::new()
        },
        program: program.into(),
        arguments,
    })
}

/// Stage the resolved gateway credential in a private (0600) file so the
/// bridge can read it via `--token-file` — the delivery the real `openclaw
/// acp` accepts. One stable file per endpoint (keyed by an address digest,
/// no secret material in the name), overwritten on every connect so files
/// never accumulate and a rotated token never goes stale on disk.
fn openclaw_gateway_token_file(address: &str, secret: &str) -> std::io::Result<PathBuf> {
    let digest = blake3::hash(address.as_bytes()).to_hex();
    let path = std::env::temp_dir().join(format!(
        "supercode-openclaw-gateway-token-{}",
        &digest.as_str()[..16]
    ));
    #[cfg(unix)]
    {
        use std::io::Write;
        use std::os::unix::fs::OpenOptionsExt;
        let mut file = std::fs::OpenOptions::new()
            .write(true)
            .create(true)
            .truncate(true)
            .mode(0o600)
            .open(&path)?;
        file.write_all(secret.as_bytes())?;
    }
    #[cfg(not(unix))]
    std::fs::write(&path, secret)?;
    Ok(path)
}

/// Open a connect-mode descriptor: resolve the endpoint address and
/// credential from the harness's own config file and build the backend that
/// joins the already-running endpoint. Fails closed with a specific
/// diagnostic when the config cannot be resolved or the declared protocol has
/// no connect-capable client yet.
fn open_connect_descriptor(
    descriptor: &crate::HarnessSupportDescriptor,
    home: &Path,
) -> std::result::Result<Box<dyn RuntimeBackend>, ServiceError> {
    let Some(connect) = &descriptor.runtime.connect_launch else {
        return Err(ServiceError::InvalidParams(format!(
            "harness `{}` has no registered connect-mode launch",
            descriptor.id.as_str()
        )));
    };
    let resolved = connect
        .resolve(home)
        .map_err(|error| ServiceError::UnsupportedAction(error.to_string()))?;
    match (descriptor.id.as_str(), connect.protocol.as_str()) {
        (HarnessId::OPENCODE, protocol) if protocol.starts_with("opencode-http") => {
            let mut backend = OpenCodeRuntimeBackend::connect(&resolved.address);
            if let Some(token) = resolved.auth {
                backend = backend.with_bearer(token);
            }
            Ok(Box::new(backend))
        }
        (HarnessId::OPENCLAW, protocol) if protocol.starts_with("acp") => {
            // OpenClaw's own `openclaw acp` binary is the gateway client: a
            // stdio ACP bridge that joins the RUNNING gateway at the resolved
            // endpoint. Blind-walk finding 2026-08-31: the real bridge does
            // NOT honor OPENCLAW_GATEWAY_TOKEN from the environment — the
            // credential must arrive via `--token-file` (never bare `--token`
            // on argv, where process listings could read it). The env var is
            // still set for older bridges that did read it. Requires openclaw
            // >= 2026.7: the 2026.2 bridge drops its gateway socket
            // mid-prompt and advertises no session resume (executed finding,
            // docs/interop/research/openclaw-acp-dialect-2026-08-30.json).
            let mut env = BTreeMap::new();
            let mut arguments = vec!["acp".into(), "--url".into(), resolved.address.clone()];
            if let Some(token) = resolved.auth {
                let token_path = openclaw_gateway_token_file(&resolved.address, token.secret())
                    .map_err(|error| {
                        ServiceError::UnsupportedAction(format!(
                            "could not stage the gateway credential for the bridge: {error}"
                        ))
                    })?;
                arguments.push("--token-file".into());
                arguments.push(token_path.to_string_lossy().into_owned());
                env.insert("OPENCLAW_GATEWAY_TOKEN".to_string(), token.secret().to_string());
            }
            // The bridge program comes from the descriptor's own default
            // launch (the compiled registry pins `openclaw`), so tests can
            // substitute an absolute mock-bridge path without touching
            // process-global state.
            let program = descriptor
                .runtime
                .default_launch
                .as_ref()
                .map(|launch| launch.program.clone())
                .unwrap_or_else(|| "openclaw".into());
            let launch = RuntimeLaunch {
                program,
                arguments,
                env,
            };
            Ok(Box::new(
                crate::AcpRuntimeBackend::new(descriptor.id.clone(), launch)
                    .with_resume_support(descriptor.runtime.capabilities.resume_session),
            ))
        }
        _ => Err(ServiceError::UnsupportedAction(format!(
            "connect-mode endpoint for `{}` speaks `{}`; joining it needs that protocol's gateway client",
            descriptor.id.as_str(),
            connect.protocol
        ))),
    }
}

/// The registry's connect-mode launch for this harness, honored only when the
/// caller supplied neither an explicit launch nor a base URL.
fn registry_connect_descriptor(
    params: &RuntimeBackendParams,
) -> Option<crate::HarnessSupportDescriptor> {
    if params.launch.is_some() || params.base_url.is_some() {
        return None;
    }
    harness_support_registry()
        .harnesses
        .into_iter()
        .find(|descriptor| descriptor.id == params.harness)
        .filter(|descriptor| descriptor.runtime.connect_launch.is_some())
}

fn service_home() -> std::result::Result<PathBuf, ServiceError> {
    supercode_interchange::user_home()
        .map(std::path::PathBuf::into_os_string)
        .map(PathBuf::from)
        .ok_or_else(|| {
            ServiceError::UnsupportedAction(
                "connect-mode launches need HOME to locate the harness config".into(),
            )
        })
}

/// The doors that open a runtime: each spawns or joins a program and waits on
/// that program's protocol handshake before it can answer.
pub const RUNTIME_OPEN_METHODS: &[&str] = &[
    "harness.v1.runtimes.start",
    "harness.v1.runtimes.resume",
    "harness.v1.runtimes.attach",
    "harness.v1.runtimes.attach_existing",
];

/// How long a runtime gets to finish opening before its caller is answered an
/// error instead. A program that never speaks the protocol at all — the wrong
/// binary, a shim that prints usage and waits — never answers the handshake,
/// so the wait is unbounded without this.
pub const RUNTIME_OPEN_DEADLINE: Duration = Duration::from_secs(60);

/// How long a control call on an ALREADY-open runtime — send input, interrupt,
/// steer, respond, close — gets before its caller is answered an error
/// instead. A live runtime answers these in milliseconds; a wedged one never
/// answers at all, and `close` is exactly what a caller reaches for when it
/// suspects that.
pub const RUNTIME_CONTROL_DEADLINE: Duration = Duration::from_secs(30);

/// The doors whose work happens entirely OUTSIDE this service's state once
/// its state has been read: probing harnesses, relaying a message into a
/// live session, and performing a conversation verb through a harness's own
/// CLI / HTTP / store door. Every one of them waits on a child process or a
/// network peer. See [`HarnessSessionService::detach`].
pub const DETACHED_METHODS: &[&str] = &[
    "harness.v1.harnesses.list",
    "harness.v1.sessions.load",
    "harness.v1.harnesses.probe",
    "harness.v1.sessions.message",
    "harness.v1.sessions.new",
    "harness.v1.sessions.reset",
    "harness.v1.sessions.archive",
    "harness.v1.sessions.delete",
];

/// How long a request moved off a transport's loop gets before its caller is
/// answered an error instead. Each of these already bounds its own inner
/// waits (a probe's handshake, the relay's send); this is the backstop for
/// the ones that do not — a harness CLI that never exits — so no caller waits
/// forever on a detached task no one is watching.
pub const DETACHED_CALL_DEADLINE: Duration = Duration::from_secs(120);

/// How long `sessions.discover` gets before its caller is answered an error
/// instead. Discovery reads each harness's own store, and a store on a cold
/// or unavailable mount answers at the filesystem's pace rather than its own.
///
/// Deliberately shorter than the clients' own request deadline (30s): the
/// server's answer names the store that did not answer, and it is only read
/// if it lands before the client stops listening.
pub const SESSION_DISCOVER_DEADLINE: Duration = Duration::from_secs(25);

/// Bound one control call on an open runtime by [`RUNTIME_CONTROL_DEADLINE`],
/// naming the method and the bound when it blows.
async fn within_control_deadline<F: std::future::Future>(
    method: &str,
    call: F,
) -> std::result::Result<F::Output, ServiceError> {
    tokio::time::timeout(RUNTIME_CONTROL_DEADLINE, call)
        .await
        .map_err(|_| {
            ServiceError::Operation(format!(
                "`{method}` gave up after {}s: the runtime did not answer",
                RUNTIME_CONTROL_DEADLINE.as_secs()
            ))
        })
}

/// One [`RUNTIME_OPEN_METHODS`] request, parsed but not yet started. See
/// [`HarnessSessionService::runtime_open`] for why it exists apart from
/// [`HarnessSessionService::handle_async`].
pub struct RuntimeOpen {
    id: Value,
    method: String,
    params: Value,
}

impl RuntimeOpen {
    /// Do the waiting: spawn or join the program and complete its handshake,
    /// bounded by [`RUNTIME_OPEN_DEADLINE`]. Touches no service state, so this
    /// runs on any task.
    pub async fn open(self) -> OpenedRuntime {
        let Self { id, method, params } = self;
        let outcome = open_runtime(&method, params).await;
        OpenedRuntime { id, outcome }
    }
}

/// The result of [`RuntimeOpen::open`], ready for
/// [`HarnessSessionService::finish_runtime_open`].
pub struct OpenedRuntime {
    id: Value,
    outcome: std::result::Result<OpenRuntime, ServiceError>,
}

/// One detached request: the half that reads this service's state already
/// done, and the half that waits not yet started. See
/// [`HarnessSessionService::detach`] and
/// [`HarnessSessionService::detach_runtime`].
pub struct DetachedCall {
    id: Value,
    method: String,
    work: std::result::Result<Work, ServiceError>,
}

impl DetachedCall {
    /// Do the waiting and answer. Runs on any task: whatever this call needed
    /// from the service was taken before it left.
    pub async fn run(self) -> DetachedAnswer {
        let Self { id, method, work } = self;
        match work {
            // A call holding a runtime is already bounded by
            // RUNTIME_CONTROL_DEADLINE, and its future OWNS that connection:
            // a second timeout around it would drop the connection mid-call
            // and take down a runtime its caller still has.
            Ok(Work::Runtime(work)) => {
                let (result, returned) = work.run().await;
                DetachedAnswer {
                    response: service_response(id, result),
                    returned,
                }
            }
            Ok(Work::Free(work)) => {
                let result = match tokio::time::timeout(DETACHED_CALL_DEADLINE, work.run()).await {
                    Ok(result) => result,
                    Err(_) => Err(ServiceError::Operation(format!(
                        "`{method}` gave up after {}s: the harness it waits on did not answer",
                        DETACHED_CALL_DEADLINE.as_secs()
                    ))),
                };
                DetachedAnswer {
                    response: service_response(id, result),
                    returned: None,
                }
            }
            Err(error) => DetachedAnswer {
                response: service_response(id, Err(error)),
                returned: None,
            },
        }
    }
}

/// One detached call's complete answer, plus whatever it must hand back to
/// the service before that answer is written. See
/// [`HarnessSessionService::finish_detached`].
pub struct DetachedAnswer {
    response: Value,
    returned: Option<ReturnedRuntime>,
}

impl DetachedAnswer {
    /// The caller's JSON-RPC response, for a transport that owns no service
    /// to give a borrowed connection back to.
    pub fn into_response(self) -> Value {
        self.response
    }
}

/// A connection lent to a detached call, on its way back to the service that
/// owns it.
pub struct ReturnedRuntime {
    connection: String,
    runtime: Box<dyn RuntimeConnection>,
}

/// The waiting half of one detached request: with nothing of the service's
/// in hand, or holding a connection the service lent out for the call.
enum Work {
    Free(DetachedWork),
    Runtime(RuntimeWork),
}

/// The waiting half of one detached request that holds nothing of the
/// service's.
enum DetachedWork {
    /// Probe the selected harnesses: find their executables, ask each its
    /// version, and at `probe: handshake` start each one and complete its
    /// protocol handshake.
    Inventory(InventoryWork),
    /// Relay one message into a live session.
    Message(MessageSessionParams),
    /// Read one session from its own store, on the blocking pool.
    Load(LoadSessionParams),
    /// Perform one conversation verb through the harness's own CLI, HTTP API,
    /// daemon socket, or supercode's own store.
    SessionMutation {
        verb: crate::SessionVerb,
        mutation: crate::SessionMutation,
    },
}

impl DetachedWork {
    async fn run(self) -> std::result::Result<Value, ServiceError> {
        match self {
            Self::Inventory(work) => run_inventory(work).await,
            Self::Message(params) => Ok(message_live_session(&params).await),
            Self::Load(params) => tokio::task::spawn_blocking(move || load_session_door(params))
                .await
                .map_err(|error| {
                    ServiceError::Operation(format!("the session read stopped: {error}"))
                })?,
            Self::SessionMutation { verb, mutation } => {
                let outcome = run_session_mutation(verb, &mutation).await?;
                serde_json::to_value(outcome)
                    .map_err(|error| ServiceError::Operation(error.to_string()))
            }
        }
    }
}

/// One detached call that holds a runtime connection for its whole run.
enum RuntimeWork {
    /// Tear down a runtime the service has already surrendered.
    Close {
        runtime: Box<dyn RuntimeConnection>,
        process_group: Option<u32>,
    },
    /// Type one live slash command through a borrowed connection, then give
    /// the connection back.
    LiveCommand {
        connection: String,
        runtime: Box<dyn RuntimeConnection>,
        verb: crate::SessionVerb,
        mutation: crate::SessionMutation,
        command: &'static str,
        session: String,
    },
}

/// What one [`RuntimeWork`] answers with: the caller's result, and the
/// connection to give back when the call only borrowed one.
type RuntimeWorkAnswer = (
    std::result::Result<Value, ServiceError>,
    Option<ReturnedRuntime>,
);

impl RuntimeWork {
    async fn run(self) -> RuntimeWorkAnswer {
        match self {
            Self::Close {
                runtime,
                process_group,
            } => (close_runtime(runtime, process_group).await, None),
            Self::LiveCommand {
                connection,
                mut runtime,
                verb,
                mutation,
                command,
                session,
            } => {
                let result =
                    type_live_command(runtime.as_mut(), verb, &mutation, command, session).await;
                (
                    result,
                    Some(ReturnedRuntime {
                        connection,
                        runtime,
                    }),
                )
            }
        }
    }
}

/// Tear down a runtime already out of the service, within
/// [`RUNTIME_CONTROL_DEADLINE`].
async fn close_runtime(
    mut runtime: Box<dyn RuntimeConnection>,
    process_group: Option<u32>,
) -> std::result::Result<Value, ServiceError> {
    match within_control_deadline("harness.v1.runtimes.close", runtime.close()).await {
        Ok(result) => {
            result.map_err(operation)?;
            Ok(json!({"closed": true}))
        }
        Err(deadline) => {
            // Dropping the handle is not enough: the process that stopped
            // answering is held by a task parked on it, so nothing here runs
            // its Drop. Signal the group the graceful path would have
            // signalled, then say so.
            let killed = kill_runtime_process_group(process_group);
            drop(runtime);
            Ok(json!({
                "closed": true,
                "killed": killed,
                "detail": error_message(deadline),
            }))
        }
    }
}

/// The conversation a live `sessions.new` / `sessions.reset` acts on: the one
/// the request named, or the runtime's own session.
fn live_session_name(runtime: &dyn RuntimeConnection, mutation: &crate::SessionMutation) -> String {
    mutation
        .session
        .clone()
        .filter(|value| !value.trim().is_empty())
        .unwrap_or_else(|| runtime.handle().runtime_id.clone())
}

/// Type one harness slash command into a live session through the very same
/// `send_input` path a human's message takes, within
/// [`RUNTIME_CONTROL_DEADLINE`].
async fn type_live_command(
    runtime: &mut dyn RuntimeConnection,
    verb: crate::SessionVerb,
    mutation: &crate::SessionMutation,
    command: &str,
    session: String,
) -> std::result::Result<Value, ServiceError> {
    within_control_deadline(
        &format!("sessions.{}", verb.as_str()),
        runtime.send_input(RuntimeInput {
            text: command.to_string(),
            image_urls: Vec::new(),
        }),
    )
    .await?
    .map_err(operation)?;
    let outcome = crate::sessions_control::live_outcome(verb, mutation, command, session)
        .map_err(session_control_error)?;
    serde_json::to_value(outcome).map_err(|error| ServiceError::Operation(error.to_string()))
}

/// A runtime that is up and whose handshake completed, with what the service
/// needs to take ownership of it.
enum OpenRuntime {
    /// supercode spawned this process, so it also hosts it: a frontend server,
    /// a live-runtime registration and a terminal launch of its own.
    Hosted {
        runtime: Box<dyn RuntimeConnection>,
        capabilities: crate::RuntimeCapabilities,
        workspace: PathBuf,
        /// A new session (`start`): it has no transcript yet, so there is no history to look for.
        fresh: bool,
    },
    /// `attach_existing` joined a process supercode does not own. It is
    /// registered as a bare connection and hosts nothing.
    Joined { runtime: Box<dyn RuntimeConnection> },
}

/// Open the runtime one [`RUNTIME_OPEN_METHODS`] request asks for, within
/// [`RUNTIME_OPEN_DEADLINE`]. The error a blown deadline answers names the
/// method and the bound, so a caller reads why it was cut loose instead of
/// waiting on a handshake that is never coming.
async fn open_runtime(
    method: &str,
    params: Value,
) -> std::result::Result<OpenRuntime, ServiceError> {
    match tokio::time::timeout(
        RUNTIME_OPEN_DEADLINE,
        open_runtime_unbounded(method, params),
    )
    .await
    {
        Ok(result) => result,
        Err(_) => Err(ServiceError::Operation(format!(
            "`{method}` gave up after {}s: the runtime never finished its protocol handshake",
            RUNTIME_OPEN_DEADLINE.as_secs()
        ))),
    }
}

async fn open_runtime_unbounded(
    method: &str,
    params: Value,
) -> std::result::Result<OpenRuntime, ServiceError> {
    match method {
        "harness.v1.runtimes.start" => {
            let params = decode::<RuntimeStartParams>(params)?;
            // Only the native Claude Code and Codex backends read a model; the same harness reached over
            // ACP (`protocol: acp`) would drop it, so it is refused there too.
            if params.model.is_some()
                && (params.backend.protocol.is_some()
                    || !matches!(
                        params.backend.harness.as_str(),
                        HarnessId::CLAUDE_CODE | HarnessId::CODEX
                    ))
            {
                return Err(ServiceError::UnsupportedAction(format!(
                    "{} takes no model at start through supercode",
                    params.backend.harness.as_str()
                )));
            }
            let backend = runtime_backend(&params.backend)?;
            let capabilities = backend.capabilities();
            let workspace = params.cwd.clone();
            let runtime = backend
                .start(RuntimeStartRequest {
                    cwd: params.cwd,
                    launch: runtime_launch(&params.backend),
                    mcp_servers: params.mcp_servers,
                    approval_policy: params.approval_policy,
                    model: params.model,
                })
                .await
                .map_err(operation)?;
            Ok(OpenRuntime::Hosted {
                runtime,
                capabilities,
                workspace,
                fresh: true,
            })
        }
        "harness.v1.runtimes.resume" | "harness.v1.runtimes.attach" => {
            let params = decode::<RuntimeAttachParams>(params)?;
            let backend = runtime_backend(&params.backend)?;
            let capabilities = backend.capabilities();
            let workspace = params
                .cwd
                .clone()
                .unwrap_or_else(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")));
            let runtime = backend
                .attach(RuntimeAttachRequest {
                    runtime_id: params.runtime_id,
                    cwd: params.cwd,
                    launch: runtime_launch(&params.backend),
                    mcp_servers: params.mcp_servers,
                    approval_policy: params.approval_policy,
                })
                .await
                .map_err(operation)?;
            Ok(OpenRuntime::Hosted {
                runtime,
                capabilities,
                workspace,
                fresh: false,
            })
        }
        "harness.v1.runtimes.attach_existing" => {
            let params = decode::<RuntimeAttachParams>(params)?;
            let backend: Box<dyn RuntimeBackend> = match params
                .backend
                .base_url
                .as_deref()
                .and_then(|value| LiveRuntimeEndpoint::parse(value).ok())
            {
                Some(endpoint) => {
                    #[cfg(not(feature = "adapter-api"))]
                    {
                        let _ = endpoint;
                        return Err(ServiceError::UnsupportedAction(
                            "live HTTP attachment adapter is not compiled".into(),
                        ));
                    }
                    #[cfg(feature = "adapter-api")]
                    {
                        let workspace = params.cwd.clone().ok_or_else(|| {
                            ServiceError::InvalidParams(
                                "Volter Harness live attach requires the project cwd".into(),
                            )
                        })?;
                        let source = LiveRuntimeSource {
                            harness: params.backend.harness.as_str().to_string(),
                            session_id: params.runtime_id.clone(),
                            workspace,
                        };
                        let receipt = resolve_live_runtime(&endpoint, &source)
                            .map_err(|error| ServiceError::Operation(error.to_string()))?;
                        Box::new(SupercodeHttpRuntimeBackend::new(receipt))
                    }
                }
                None => runtime_backend(&params.backend)?,
            };
            let capabilities = backend.capabilities();
            if !capabilities.attach_existing_process {
                return Err(ServiceError::Operation(format!(
                    "{} cannot attach to an already-running process; use runtimes.resume for a persisted session",
                    backend.harness().as_str()
                )));
            }
            let runtime = backend
                .attach_existing(RuntimeAttachRequest {
                    runtime_id: params.runtime_id,
                    cwd: params.cwd,
                    launch: runtime_launch(&params.backend),
                    mcp_servers: params.mcp_servers,
                    approval_policy: params.approval_policy,
                })
                .await
                .map_err(operation)?;
            Ok(OpenRuntime::Joined { runtime })
        }
        _ => Err(ServiceError::MethodNotFound),
    }
}

/// Wrap one service outcome in its JSON-RPC 2.0 envelope.
fn service_response(id: Value, result: std::result::Result<Value, ServiceError>) -> Value {
    match result {
        Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
        Err(ServiceError::InvalidParams(message)) => rpc_error(id, -32602, &message),
        Err(ServiceError::MethodNotFound) => rpc_error(id, -32601, "method not found"),
        Err(ServiceError::UnsupportedAction(message)) => rpc_error(id, -32020, &message),
        Err(ServiceError::Operation(message)) => rpc_error(id, -32000, &message),
        Err(ServiceError::Sdk(error)) => sdk_rpc_error(id, &error),
    }
}

fn runtime_backend(
    params: &RuntimeBackendParams,
) -> std::result::Result<Box<dyn RuntimeBackend>, ServiceError> {
    if let Some(descriptor) = registry_connect_descriptor(params) {
        return open_connect_descriptor(&descriptor, &service_home()?);
    }
    if params.protocol.as_deref() == Some("acp") {
        let launch = params
            .launch
            .clone()
            .or_else(|| {
                harness_support_registry()
                    .harnesses
                    .into_iter()
                    .find(|harness| harness.id == params.harness)
                    .filter(|harness| {
                        harness.runtime.implementation == ImplementationKind::GenericProtocol
                            && harness.runtime.protocol.starts_with("acp")
                    })
                    .and_then(|harness| harness.runtime.default_launch)
            })
            .ok_or_else(|| {
                ServiceError::InvalidParams(
                    "an ACP runtime requires `launch` unless the harness has a registered default"
                        .into(),
                )
            })?;
        let resume_session = harness_support_registry()
            .harnesses
            .into_iter()
            .find(|harness| harness.id == params.harness)
            .is_some_and(|harness| harness.runtime.capabilities.resume_session);
        return Ok(Box::new(
            AcpRuntimeBackend::new(params.harness.clone(), launch)
                .with_resume_support(resume_session),
        ));
    }
    let backend: Box<dyn RuntimeBackend> = match params.harness.as_str() {
        HarnessId::CODEX => Box::new(CodexRuntimeBackend::new()),
        HarnessId::CLAUDE_CODE => Box::new(ClaudeCodeRuntimeBackend::new()),
        HarnessId::PI => Box::new(PiRuntimeBackend::new()),
        HarnessId::OPENCODE => match &params.base_url {
            Some(url) => Box::new(OpenCodeRuntimeBackend::connect(url)),
            None => Box::new(OpenCodeRuntimeBackend::new()),
        },
        harness => {
            let descriptor = harness_support_registry()
                .harnesses
                .into_iter()
                .find(|descriptor| descriptor.id.as_str() == harness)
                .filter(|descriptor| {
                    descriptor.runtime.implementation == ImplementationKind::GenericProtocol
                        && descriptor.runtime.protocol.starts_with("acp")
                });
            let Some(descriptor) = descriptor else {
                return Err(ServiceError::InvalidParams(format!(
                    "no runtime adapter for harness `{harness}`; use protocol `acp` with a launch command"
                )));
            };
            let resume = descriptor.runtime.capabilities.resume_session;
            Box::new(
                AcpRuntimeBackend::new(
                    descriptor.id,
                    descriptor
                        .runtime
                        .default_launch
                        .expect("generic ACP registry entry includes its launch"),
                )
                .with_resume_support(resume),
            )
        }
    };
    Ok(backend)
}

fn runtime_launch(params: &RuntimeBackendParams) -> Option<RuntimeLaunch> {
    if let Some(launch) = &params.launch {
        return Some(launch.clone());
    }
    if !matches!(params.policy, RuntimePolicy::Yolo) {
        return None;
    }
    let launch = match params.harness.as_str() {
        HarnessId::GROK => RuntimeLaunch {
            program: "grok".into(),
            arguments: {
                let mut arguments: Vec<String> = Vec::new();
                if crate::support::self_sandbox_supported() {
                    arguments.extend(["--sandbox".into(), "workspace".into()]);
                }
                arguments.extend([
                    "--always-approve".into(),
                    "agent".into(),
                    "--no-leader".into(),
                    "stdio".into(),
                ]);
                arguments
            },
            env: crate::support::grok_env(),
        },
        HarnessId::CODEX => RuntimeLaunch {
            program: "codex".into(),
            arguments: vec![
                "--dangerously-bypass-approvals-and-sandbox".into(),
                "--dangerously-bypass-hook-trust".into(),
                "app-server".into(),
            ],
            env: BTreeMap::new(),
        },
        HarnessId::CLAUDE_CODE => RuntimeLaunch {
            program: "claude".into(),
            arguments: vec![
                "--dangerously-skip-permissions".into(),
                "--print".into(),
                "--input-format".into(),
                "stream-json".into(),
                "--output-format".into(),
                "stream-json".into(),
                "--verbose".into(),
            ],
            env: BTreeMap::new(),
        },
        HarnessId::PI => RuntimeLaunch {
            program: "pi".into(),
            arguments: vec!["--approve".into(), "--mode".into(), "rpc".into()],
            env: BTreeMap::new(),
        },
        HarnessId::OPENCODE => RuntimeLaunch {
            program: "opencode".into(),
            arguments: vec!["serve".into()],
            env: BTreeMap::new(),
        },
        HarnessId::GEMINI => RuntimeLaunch {
            program: "gemini".into(),
            arguments: vec!["--acp".into(), "--yolo".into()],
            env: BTreeMap::new(),
        },
        HarnessId::GOOSE => RuntimeLaunch {
            program: "goose".into(),
            arguments: vec!["acp".into()],
            env: BTreeMap::new(),
        },
        HarnessId::SUPERCODE => RuntimeLaunch {
            program: "supercode".into(),
            arguments: vec!["acp".into(), "--dangerous".into()],
            env: BTreeMap::new(),
        },
        _ => return None,
    };
    Some(launch)
}

/// Disposable harness state for a no-prompt readiness probe. Merely opening
/// several stock CLIs writes a session header or migrates configuration, so a
/// handshake must never point at the user's real home. Authentication files
/// are copied into the private temporary home; all writes disappear with the
/// guard after the connection closes.
struct IsolatedProbeHome {
    launch: RuntimeLaunch,
    root: PathBuf,
}

impl IsolatedProbeHome {
    fn new(harness: &str, mut launch: RuntimeLaunch) -> std::io::Result<Self> {
        let root = std::env::temp_dir().join(format!(
            "supercode-harness-probe-{harness}-{}",
            generated_session_id()
        ));
        std::fs::create_dir_all(&root)?;
        set_private_dir_permissions(&root)?;

        if let Some(source_home) = supercode_interchange::user_home()
            .map(std::path::PathBuf::into_os_string)
            .map(PathBuf::from)
        {
            for relative in probe_auth_files(harness) {
                copy_probe_file(&source_home, &root, relative)?;
            }
        }
        // supercode reads its own config home ($SUPERCODE_HOME, else
        // $XDG_CONFIG_HOME/supercode, else ~/.config/supercode), not a fixed
        // place under HOME: a login kept under XDG_CONFIG_HOME probed as
        // "no API key found" while `supercode run` answered.
        if harness == HarnessId::SUPERCODE {
            let config_home = crate::agent::global_instructions_dir();
            for file in ["config.toml", "credentials.toml"] {
                copy_probe_path(
                    &config_home.join(file),
                    &root.join(".config/supercode").join(file),
                )?;
            }
        }
        configure_isolated_probe_auth(harness, &root)?;

        let root_text = root.to_string_lossy().into_owned();
        for (key, value) in [
            ("HOME", root_text.clone()),
            (
                "XDG_CACHE_HOME",
                root.join(".cache").to_string_lossy().into_owned(),
            ),
            (
                "XDG_CONFIG_HOME",
                root.join(".config").to_string_lossy().into_owned(),
            ),
            (
                "XDG_DATA_HOME",
                root.join(".local/share").to_string_lossy().into_owned(),
            ),
        ] {
            launch.env.insert(key.into(), value);
        }
        let scoped = match harness {
            HarnessId::CLAUDE_CODE => Some(("CLAUDE_CONFIG_DIR", root.join(".claude"))),
            HarnessId::CODEX => Some(("CODEX_HOME", root.join(".codex"))),
            HarnessId::GEMINI => Some(("GEMINI_CLI_HOME", root.clone())),
            HarnessId::GROK => Some(("GROK_HOME", root.join(".grok"))),
            HarnessId::PI => Some(("PI_CODING_AGENT_DIR", root.join(".pi/agent"))),
            HarnessId::SUPERCODE => Some(("SUPERCODE_HOME", root.join(".config/supercode"))),
            _ => None,
        };
        if let Some((key, value)) = scoped {
            launch
                .env
                .insert(key.into(), value.to_string_lossy().into_owned());
        }
        Ok(Self { launch, root })
    }

    fn cleanup(&self) -> std::io::Result<()> {
        match std::fs::remove_dir_all(&self.root) {
            Ok(()) => Ok(()),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(error) => Err(error),
        }
    }
}

impl Drop for IsolatedProbeHome {
    fn drop(&mut self) {
        let _ = self.cleanup();
    }
}

fn probe_auth_files(harness: &str) -> &'static [&'static str] {
    match harness {
        HarnessId::CLAUDE_CODE => &[".claude/.credentials.json", ".claude.json"],
        // The gateway endpoint + token live in openclaw's own config; without
        // it the isolated probe dials the default endpoint unauthenticated
        // (PARITY-24 finding 2026-08-31).
        HarnessId::OPENCLAW => &[".openclaw/openclaw.json"],
        HarnessId::CODEX => &[".codex/auth.json"],
        HarnessId::GEMINI => &[
            ".gemini/google_accounts.json",
            ".gemini/oauth_creds.json",
            ".gemini/settings.json",
        ],
        HarnessId::GROK => &[".grok/auth.json", ".grok/config.toml"],
        HarnessId::OPENCODE => &[
            ".config/opencode/auth.json",
            ".local/share/opencode/auth.json",
        ],
        HarnessId::PI => &[".pi/agent/auth.json"],
        // Hermes keeps its provider selection in config.yaml, its OAuth
        // credential pool in auth.json, and API keys in .env; without them
        // the isolated probe sees "No LLM provider configured" for a
        // hermes that answers fine from the user's real home.
        HarnessId::HERMES => &[".hermes/config.yaml", ".hermes/auth.json", ".hermes/.env"],
        _ => &[],
    }
}

fn copy_probe_file(source_home: &Path, probe_home: &Path, relative: &str) -> std::io::Result<()> {
    copy_probe_path(&source_home.join(relative), &probe_home.join(relative))
}

fn copy_probe_path(source: &Path, destination: &Path) -> std::io::Result<()> {
    if !source.is_file() {
        return Ok(());
    }
    if let Some(parent) = destination.parent() {
        std::fs::create_dir_all(parent)?;
        set_private_dir_permissions(parent)?;
    }
    std::fs::copy(source, destination)?;
    set_private_file_permissions(destination)
}

fn configure_isolated_probe_auth(harness: &str, probe_home: &Path) -> std::io::Result<()> {
    if harness != HarnessId::GEMINI {
        return Ok(());
    }
    let oauth = probe_home.join(".gemini/oauth_creds.json");
    if !oauth.is_file() {
        return Ok(());
    }
    let settings_path = probe_home.join(".gemini/settings.json");
    let mut settings = std::fs::read_to_string(&settings_path)
        .ok()
        .and_then(|raw| serde_json::from_str::<Value>(&raw).ok())
        .unwrap_or_else(|| json!({}));
    settings["security"]["auth"]["selectedType"] = Value::String("oauth-personal".into());
    std::fs::write(
        &settings_path,
        serde_json::to_vec_pretty(&settings).map_err(std::io::Error::other)?,
    )?;
    set_private_file_permissions(&settings_path)
}

#[cfg(unix)]
fn set_private_dir_permissions(path: &Path) -> std::io::Result<()> {
    use std::os::unix::fs::PermissionsExt;
    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700))
}

#[cfg(not(unix))]
fn set_private_dir_permissions(_path: &Path) -> std::io::Result<()> {
    Ok(())
}

#[cfg(unix)]
fn set_private_file_permissions(path: &Path) -> std::io::Result<()> {
    use std::os::unix::fs::PermissionsExt;
    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
}

#[cfg(not(unix))]
fn set_private_file_permissions(_path: &Path) -> std::io::Result<()> {
    Ok(())
}

fn find_executable(program: &str) -> Option<PathBuf> {
    let candidate = PathBuf::from(program);
    if candidate.components().count() > 1 {
        return candidate.is_file().then_some(candidate);
    }
    let path = std::env::var_os("PATH")?;
    for directory in std::env::split_paths(&path) {
        let candidate = directory.join(program);
        if candidate.is_file() {
            return std::fs::canonicalize(&candidate).ok().or(Some(candidate));
        }
        #[cfg(windows)]
        {
            for extension in ["exe", "cmd", "bat"] {
                let candidate = directory.join(format!("{program}.{extension}"));
                if candidate.is_file() {
                    return std::fs::canonicalize(&candidate).ok().or(Some(candidate));
                }
            }
        }
    }
    None
}

async fn executable_version(executable: &Path) -> Option<String> {
    let mut command = tokio::process::Command::new(executable);
    command
        .arg("--version")
        .stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .kill_on_drop(true);
    let output = tokio::time::timeout(Duration::from_secs(3), command.output())
        .await
        .ok()?
        .ok()?;
    let stdout = String::from_utf8_lossy(&output.stdout);
    let stderr = String::from_utf8_lossy(&output.stderr);
    stdout
        .lines()
        .chain(stderr.lines())
        .map(str::trim)
        .find(|line| !line.is_empty())
        .map(|line| truncate_text(line, 200))
}

pub(crate) fn auth_evidence(harness: &str) -> bool {
    let env_names: &[&str] = match harness {
        HarnessId::CLAUDE_CODE => &["ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN"],
        HarnessId::CODEX => &["OPENAI_API_KEY"],
        HarnessId::OPENCODE => &["ANTHROPIC_API_KEY", "OPENAI_API_KEY", "OPENROUTER_API_KEY"],
        HarnessId::PI => &["ANTHROPIC_API_KEY", "OPENAI_API_KEY", "OPENROUTER_API_KEY"],
        HarnessId::GROK => &["XAI_API_KEY", "GROK_API_KEY"],
        HarnessId::GEMINI => &["GEMINI_API_KEY", "GOOGLE_API_KEY"],
        HarnessId::SUPERCODE => &["OPENROUTER_API_KEY"],
        _ => &[],
    };
    if env_names
        .iter()
        .any(|name| std::env::var_os(name).is_some_and(|value| !value.is_empty()))
    {
        return true;
    }
    let Some(home) = supercode_interchange::user_home()
        .map(std::path::PathBuf::into_os_string)
        .map(PathBuf::from)
    else {
        return false;
    };
    let files: Vec<PathBuf> = match harness {
        HarnessId::CLAUDE_CODE => vec![home.join(".claude/.credentials.json")],
        HarnessId::CODEX => vec![home.join(".codex/auth.json")],
        HarnessId::OPENCODE => vec![
            home.join(".local/share/opencode/auth.json"),
            home.join(".config/opencode/auth.json"),
        ],
        HarnessId::PI => vec![home.join(".pi/agent/auth.json")],
        HarnessId::GROK => vec![home.join(".grok/auth.json")],
        HarnessId::GEMINI => vec![
            home.join(".gemini/oauth_creds.json"),
            home.join(".gemini/google_accounts.json"),
        ],
        HarnessId::SUPERCODE => vec![home.join(".config/supercode/credentials.toml")],
        HarnessId::HERMES => vec![home.join(".hermes/auth.json"), home.join(".hermes/.env")],
        _ => Vec::new(),
    };
    if files.into_iter().any(|path| {
        std::fs::metadata(path)
            .map(|metadata| metadata.is_file() && metadata.len() > 2)
            .unwrap_or(false)
    }) {
        return true;
    }
    // macOS keeps Claude Code's OAuth login in the Keychain, so
    // `.claude/.credentials.json` never exists there and the file probe above
    // reports a signed-in install as unauthenticated forever. A completed
    // login also writes an `oauthAccount` record into `~/.claude.json` on
    // every platform — file-based, prompt-free evidence (querying the
    // Keychain itself from an unsigned daemon can raise a UI prompt).
    if harness == HarnessId::CLAUDE_CODE {
        return std::fs::read_to_string(home.join(".claude.json"))
            .map(|text| text.contains("\"oauthAccount\""))
            .unwrap_or(false);
    }
    false
}

fn looks_like_auth_error(message: &str) -> bool {
    let message = message.to_ascii_lowercase();
    [
        "auth",
        "login",
        "sign in",
        "sign-in",
        "credential",
        "unauthorized",
        "forbidden",
        "token",
    ]
    .iter()
    .any(|needle| message.contains(needle))
}

fn unavailable_capabilities() -> crate::RuntimeCapabilities {
    crate::RuntimeCapabilities {
        start_session: false,
        resume_session: false,
        attach_existing_process: false,
        send_input: false,
        stream_events: false,
        interrupt: false,
        steer: false,
        respond_to_requests: false,
    }
}

fn truncate_text(text: &str, max_chars: usize) -> String {
    let mut chars = text.chars();
    let truncated = chars.by_ref().take(max_chars).collect::<String>();
    if chars.next().is_some() {
        format!("{truncated}…")
    } else {
        truncated
    }
}

/// The process group a runtime's own handle names, when it names one.
///
/// Every adapter that spawns a local process spawns it as its own group
/// leader (`Command::process_group(0)`), so the endpoint's pid IS the group
/// id. A runtime reached over HTTP, or one supercode joined rather than
/// spawned, names no group here and is left alone.
fn runtime_process_group(handle: &crate::RuntimeHandle) -> Option<u32> {
    match &handle.endpoint {
        crate::RuntimeEndpoint::LocalProcess { pid, .. } => *pid,
        crate::RuntimeEndpoint::Http { .. } => None,
    }
}

/// SIGKILL a wedged runtime's whole process group, reporting whether there
/// was one to signal. This is the same group teardown a graceful `close`
/// performs; it runs here only when the graceful path blew its deadline,
/// because the task parked on the unanswered call still owns the process
/// handle and so no `Drop` of ours can reach it.
fn kill_runtime_process_group(process_group: Option<u32>) -> bool {
    match process_group {
        #[cfg(unix)]
        Some(pid) => {
            crate::lsp::kill_process_group(pid);
            true
        }
        #[cfg(not(unix))]
        Some(_) => false,
        None => false,
    }
}

fn error_message(error: ServiceError) -> String {
    match error {
        ServiceError::InvalidParams(message)
        | ServiceError::Operation(message)
        | ServiceError::UnsupportedAction(message) => message,
        ServiceError::MethodNotFound => "runtime adapter is not available".into(),
        ServiceError::Sdk(error) => error.to_string(),
    }
}

#[derive(Debug)]
enum ServiceError {
    InvalidParams(String),
    MethodNotFound,
    UnsupportedAction(String),
    Operation(String),
    Sdk(SdkError),
}

fn sdk_error(operation: SdkOperation, error: ServiceError) -> SdkError {
    match error {
        ServiceError::InvalidParams(message) => {
            SdkError::new(SdkErrorCode::InvalidArgument, operation, message)
        }
        ServiceError::MethodNotFound | ServiceError::UnsupportedAction(_) => {
            SdkError::unsupported(operation)
        }
        ServiceError::Operation(message) => {
            let code = if message.contains("already in progress") {
                SdkErrorCode::Busy
            } else if message.contains("not supported by this runtime") {
                SdkErrorCode::UnsupportedAction
            } else if message.contains("unknown runtime connection") {
                SdkErrorCode::NotFound
            } else {
                SdkErrorCode::Execution
            };
            SdkError::new(code, operation, message)
        }
        ServiceError::Sdk(error) => error,
    }
}

fn sdk_rpc_error(id: Value, error: &SdkError) -> Value {
    let error_code = error.code();
    let code = match error_code {
        SdkErrorCode::Unauthenticated => -32030,
        SdkErrorCode::Unauthorized => -32031,
        SdkErrorCode::ControllerRequired => -32032,
        SdkErrorCode::LeaseExpired => -32033,
        SdkErrorCode::InvalidArgument => -32602,
        SdkErrorCode::NotFound => -32004,
        SdkErrorCode::Busy => -32000,
        SdkErrorCode::UnsupportedAction => -32020,
        SdkErrorCode::Execution => -32002,
        SdkErrorCode::Transport => -32003,
    };
    json!({
        "jsonrpc": "2.0",
        "id": id,
        "error": {
            "code": code,
            "name": error_code,
            "operation": error.operation(),
            "message": error.to_string(),
        },
    })
}

fn decode<T: for<'de> Deserialize<'de>>(value: Value) -> std::result::Result<T, ServiceError> {
    serde_json::from_value(value).map_err(|error| ServiceError::InvalidParams(error.to_string()))
}

fn operation(error: impl Into<crate::Error>) -> ServiceError {
    let error = error.into();
    match error {
        crate::Error::Sdk(error) => ServiceError::Sdk(error),
        error => ServiceError::Operation(error.to_string()),
    }
}

/// ORCH-12 `harness.v1.memory.show|search` params. `homes` is the same
/// storage-root override every read-only method accepts, so a caller can
/// point the read at a fixture home without touching the real ones.
#[derive(Debug, Clone, Deserialize, Default)]
#[serde(default)]
struct MemoryRequest {
    /// Harness whose store is read. Required.
    harness: Option<String>,
    /// The needle, required by `search`.
    query: Option<String>,
    /// Hermes profile, OpenClaw agent, or Claude Code project.
    profile: Option<String>,
    /// Claude Code session id selecting a project store (`show` only).
    session: Option<String>,
    /// Include each document's whole text (`show` only).
    full: bool,
    /// Treat `query` as a regular expression (`search` only).
    regex: bool,
    /// Working tree whose project store is read.
    cwd: Option<std::path::PathBuf>,
    /// Storage roots to read.
    homes: crate::HarnessHomes,
}

/// Read the memory noun. A harness with no memory store fails with
/// `UnsupportedAction` (RPC `-32020`), never an empty list.
fn memory_call(method: &str, params: Value) -> std::result::Result<Value, ServiceError> {
    let request = decode::<MemoryRequest>(params)?;
    let harness = request
        .harness
        .clone()
        .ok_or_else(|| ServiceError::InvalidParams("`harness` is required".into()))?;
    let to_service = |error: crate::memory::MemoryError| match error {
        crate::memory::MemoryError::UnsupportedHarness { .. }
        | crate::memory::MemoryError::SessionNotScoped { .. } => {
            ServiceError::UnsupportedAction(error.to_string())
        }
        other => ServiceError::InvalidParams(other.to_string()),
    };
    match method {
        "harness.v1.memory.show" => {
            let documents = crate::memory::show_memory(&crate::memory::MemoryQuery {
                harness,
                profile: request.profile,
                session: request.session,
                full: request.full,
                cwd: request.cwd,
                homes: request.homes,
            })
            .map_err(to_service)?;
            Ok(json!({
                "schema": crate::memory::MEMORY_SCHEMA,
                "documents": documents,
            }))
        }
        "harness.v1.memory.search" => {
            let query = request
                .query
                .ok_or_else(|| ServiceError::InvalidParams("`query` is required".into()))?;
            let matches = crate::memory::search_memory(&crate::memory::MemorySearchQuery {
                harness,
                query,
                profile: request.profile,
                regex: request.regex,
                cwd: request.cwd,
                homes: request.homes,
            })
            .map_err(to_service)?;
            Ok(json!({
                "schema": crate::memory::MEMORY_SCHEMA,
                "matches": matches,
            }))
        }
        _ => Err(ServiceError::MethodNotFound),
    }
}

/// ORCH-10 `harness.v1.profiles.list|get` params. `homes` is the same
/// storage-root override every read-only method accepts, so a caller can
/// point the read at a fixture home without touching the real ones.
#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
struct ProfilesQuery {
    /// Restrict the listing to one harness. `get` requires it.
    harness: Option<String>,
    /// Profile name, required by `get`.
    name: Option<String>,
    /// Storage roots to read.
    homes: crate::HarnessHomes,
}

impl Default for ProfilesQuery {
    fn default() -> Self {
        Self {
            harness: None,
            name: None,
            homes: crate::HarnessHomes::default(),
        }
    }
}

/// Read the profile noun. A harness with no profile concept fails with
/// `UnsupportedAction` (RPC `-32020`), never an empty list.
fn profiles_call(method: &str, params: Value) -> std::result::Result<Value, ServiceError> {
    let query = decode::<ProfilesQuery>(params)?;
    let to_service = |error: crate::profiles::ProfileError| match error {
        crate::profiles::ProfileError::UnsupportedHarness { .. } => {
            ServiceError::UnsupportedAction(error.to_string())
        }
        crate::profiles::ProfileError::NotFound { .. } => {
            ServiceError::InvalidParams(error.to_string())
        }
    };
    match method {
        "harness.v1.profiles.list" => {
            let profiles = crate::profiles::list_profiles(&query.homes, query.harness.as_deref())
                .map_err(to_service)?;
            Ok(json!({
                "schema": crate::profiles::PROFILES_SCHEMA,
                "profiles": profiles,
            }))
        }
        "harness.v1.profiles.get" => {
            let harness = query
                .harness
                .ok_or_else(|| ServiceError::InvalidParams("`harness` is required".into()))?;
            let name = query
                .name
                .ok_or_else(|| ServiceError::InvalidParams("`name` is required".into()))?;
            let profile =
                crate::profiles::get_profile(&query.homes, &harness, &name).map_err(to_service)?;
            Ok(json!({
                "schema": crate::profiles::PROFILES_SCHEMA,
                "profile": profile,
            }))
        }
        _ => Err(ServiceError::MethodNotFound),
    }
}

/// ORCH-14 `harness.v1.channels.list|status` params, the same storage-root
/// override every read-only method accepts so a caller can point the read at
/// a fixture home without touching the real ones.
#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
struct ChannelsQuery {
    /// Restrict the listing to one harness. `status` requires it.
    harness: Option<String>,
    /// Channel name, required by `status`.
    name: Option<String>,
    /// Storage roots to read.
    homes: crate::HarnessHomes,
}

impl Default for ChannelsQuery {
    fn default() -> Self {
        Self {
            harness: None,
            name: None,
            homes: crate::HarnessHomes::default(),
        }
    }
}

/// Read the channel noun. A harness with no channel concept fails with
/// `UnsupportedAction` (RPC `-32020`), never an empty list. No row carries a
/// token, key or secret — see `crate::channels` "Secrecy".
#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
struct RoutesQuery {
    harness: Option<String>,
    /// Restrict to routes targeting one profile / agent.
    profile: Option<String>,
    homes: crate::HarnessHomes,
}

impl Default for RoutesQuery {
    fn default() -> Self {
        Self {
            harness: None,
            profile: None,
            homes: crate::HarnessHomes::default(),
        }
    }
}

#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
struct TriggersQuery {
    harness: Option<String>,
    homes: crate::HarnessHomes,
}

impl Default for TriggersQuery {
    fn default() -> Self {
        Self {
            harness: None,
            homes: crate::HarnessHomes::default(),
        }
    }
}

fn triggers_call(params: Value) -> std::result::Result<Value, ServiceError> {
    let query = decode::<TriggersQuery>(params)?;
    let triggers = crate::triggers::list_triggers(&query.homes, query.harness.as_deref())
        .map_err(|error| ServiceError::UnsupportedAction(error.to_string()))?;
    Ok(json!({
        "schema": crate::triggers::TRIGGERS_SCHEMA,
        "triggers": triggers,
    }))
}

fn routes_call(params: Value) -> std::result::Result<Value, ServiceError> {
    let query = decode::<RoutesQuery>(params)?;
    let routes = crate::routes::list_routes(
        &query.homes,
        query.harness.as_deref(),
        query.profile.as_deref(),
    )
    .map_err(|error| ServiceError::UnsupportedAction(error.to_string()))?;
    Ok(json!({
        "schema": crate::routes::ROUTES_SCHEMA,
        "routes": routes,
    }))
}

fn channels_call(method: &str, params: Value) -> std::result::Result<Value, ServiceError> {
    let query = decode::<ChannelsQuery>(params)?;
    let to_service = |error: crate::channels::ChannelError| match error {
        crate::channels::ChannelError::UnsupportedHarness { .. } => {
            ServiceError::UnsupportedAction(error.to_string())
        }
        crate::channels::ChannelError::NotFound { .. } => {
            ServiceError::InvalidParams(error.to_string())
        }
    };
    match method {
        "harness.v1.channels.list" => {
            let channels = crate::channels::list_channels(&query.homes, query.harness.as_deref())
                .map_err(to_service)?;
            Ok(json!({
                "schema": crate::channels::CHANNELS_SCHEMA,
                "channels": channels,
            }))
        }
        "harness.v1.channels.status" => {
            let harness = query
                .harness
                .ok_or_else(|| ServiceError::InvalidParams("`harness` is required".into()))?;
            let name = query
                .name
                .ok_or_else(|| ServiceError::InvalidParams("`name` is required".into()))?;
            let channel = crate::channels::channel_status(&query.homes, &harness, &name)
                .map_err(to_service)?;
            Ok(json!({
                "schema": crate::channels::CHANNELS_SCHEMA,
                "channel": channel,
            }))
        }
        _ => Err(ServiceError::MethodNotFound),
    }
}

fn rpc_error(id: Value, code: i64, message: &str) -> Value {
    json!({
        "jsonrpc": "2.0",
        "id": id,
        "error": {"code": code, "message": message},
    })
}