supercode_harness/

frontend.rs

//! Protocol-neutral frontend contract for one SDK-owned Supercode runtime.
//!
//! Terminal, HTTP, ACP, and future browser frontends consume this contract;
//! none of them owns an [`crate::Agent`] or a second model loop.  Events keep
//! their complete JSON payload and gain a monotonic sequence so a frontend can
//! cross the history-replay/live-stream boundary without duplicates.

use std::collections::{BTreeMap, VecDeque};
#[cfg(feature = "adapter-api")]
use std::sync::atomic::AtomicBool;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
#[cfg(feature = "adapter-api")]
use std::sync::Weak;

use async_trait::async_trait;
#[cfg(feature = "adapter-api")]
use futures::StreamExt;
use serde::{Deserialize, Serialize};
#[cfg(feature = "adapter-api")]
use serde_json::json;
use serde_json::Value;
use tokio::sync::broadcast;

#[cfg(feature = "adapter-api")]
use crate::sdk::RuntimeSubmitError;
pub use crate::sdk::SdkError as FrontendRuntimeError;
pub use crate::sdk::SdkEvent as FrontendEvent;
pub use crate::sdk::SdkRuntime as FrontendRuntime;
use crate::server::RpcEngine;
use crate::ChatMessage;

/// Frontend contract schema version.
pub const FRONTEND_RUNTIME_SCHEMA_VERSION: u32 = 2;

/// Runtime lifecycle-event schema version.
///
/// Operation descriptors evolve the attach contract independently from the
/// established event payloads consumed by machine frontends.
pub(crate) const FRONTEND_EVENT_SCHEMA_VERSION: u32 = 1;

/// Maximum sequenced events retained between canonical history snapshots.
pub const FRONTEND_REPLAY_CAPACITY: usize = 4096;

/// Whether a model/tool turn currently owns the runtime.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FrontendTurnState {
    /// The runtime accepts a new turn.
    Idle,
    /// A user, scheduler, or tool turn is active.
    Busy,
}

/// Frontend-visible runtime connection state.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FrontendConnectionState {
    /// The SDK runtime is reachable.
    Connected,
    /// Graceful shutdown has been requested.
    ShuttingDown,
}

/// Actions the current runtime adapter can actually perform.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FrontendActions {
    /// Submit a new user turn.
    pub submit: bool,
    /// Interrupt an active turn.
    pub interrupt: bool,
    /// Queue a steering instruction during a turn.
    pub steer: bool,
    /// Answer an approval, elicitation, or other protocol request.
    pub respond: bool,
    /// Detach without stopping the runtime.
    pub detach: bool,
    /// Close the SDK-owned runtime.
    pub close: bool,
}

/// Display semantics emitted by the runtime.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FrontendDisplayCapabilities {
    /// Known normalized event kinds at this schema version.
    pub event_kinds: Vec<String>,
    /// Whether unknown payloads remain available for generic rendering.
    pub opaque_fallback: bool,
}

/// One runtime-provided command surfaced by a composer.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FrontendCommandDescriptor {
    /// Command name without the leading slash.
    pub name: String,
    /// Optional short help text.
    pub description: Option<String>,
    /// Optional argument usage shown beside the command.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub argument_hint: Option<String>,
}

/// Stable family for an explicitly invocable frontend operation.
///
/// Families without a production [`FrontendRuntime::invoke`] implementation
/// are never advertised. Keeping the full vocabulary here lets frontends
/// render future file/model/session/subagent/image/reduction controls from the
/// catalog without inferring them from composable modules.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FrontendOperationKind {
    /// Invoke a trusted runtime prompt template.
    Prompt,
    /// Attach or inspect a file through a typed runtime route.
    File,
    /// Inspect or switch the active model through a typed runtime route.
    Model,
    /// Perform a session operation through a typed runtime route.
    Session,
    /// Perform a subagent operation through a typed runtime route.
    Subagent,
    /// Attach an image through a typed runtime route.
    Image,
    /// Perform a reversible reduction operation through a typed runtime route.
    Reduction,
    /// BP-4 (catalog:109 "Context-usage introspection"): read the live
    /// context-window accounting for this runtime. Unlike every family
    /// above, this one takes no arguments and mutates nothing — it is the
    /// frontend's door onto cc's `/context` grid and cx's
    /// `get_context_remaining`, so a UI can render a context meter without
    /// submitting a turn.
    Context,
}

/// One operation the runtime can genuinely invoke.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FrontendOperationDescriptor {
    /// Stable runtime-scoped identifier supplied back during invocation.
    pub id: String,
    /// Typed operation family.
    pub kind: FrontendOperationKind,
    /// Optional slash-command trigger rendered by terminal composers.
    pub command: Option<FrontendCommandDescriptor>,
}

/// Typed invocation accepted by [`FrontendRuntime::invoke`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum FrontendOperationInvocation {
    /// Expand and submit one advertised trusted prompt template.
    Prompt {
        /// Identifier from [`FrontendOperationDescriptor::id`].
        operation_id: String,
        /// Free text replacing the prompt template's `{args}` placeholder.
        arguments: String,
    },
    /// BP-4: read the live context-window accounting. Takes no arguments —
    /// there is nothing for a caller to steer, and nothing it can mutate.
    Context {
        /// Identifier from [`FrontendOperationDescriptor::id`].
        operation_id: String,
    },
    /// BP-13 (catalog D9 "Mid-session model switching"): change the model
    /// this runtime sends to, WITHOUT losing the session. Advertised only
    /// by a runtime whose config allows switching
    /// (`[core.model_switch] allow_switch`), so a frontend never renders a
    /// control the runtime would refuse.
    Model {
        /// Identifier from [`FrontendOperationDescriptor::id`].
        operation_id: String,
        /// The model to switch to — an alias or a full slug, resolved
        /// through the runtime's own routing table. Empty means "report
        /// the current model", which mutates nothing.
        model: String,
    },
}

impl FrontendOperationInvocation {
    /// Identifier supplied by the runtime catalog.
    pub fn operation_id(&self) -> &str {
        match self {
            Self::Prompt { operation_id, .. }
            | Self::Context { operation_id }
            | Self::Model { operation_id, .. } => operation_id,
        }
    }
}

/// Typed result returned by [`FrontendRuntime::invoke`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum FrontendOperationResult {
    /// Reply from a prompt-template turn.
    Prompt {
        /// Final assistant reply.
        reply: String,
    },
    /// BP-4: the live context-window accounting.
    Context {
        /// Token/percentage breakdown for the runtime's current context.
        usage: crate::ContextUsage,
    },
    /// BP-13: the model in force after the operation ran.
    Model {
        /// The model the runtime now sends to (alias-resolved).
        model: String,
        /// The model it sent to before — equal to `model` when the
        /// invocation only reported the current one.
        previous: String,
    },
}

/// Source/emulation identity supplied by the session-loading surface.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct FrontendRuntimeMetadata {
    /// Source harness whose session semantics are being continued.
    pub source_harness: Option<String>,
    /// Resolved composable preset/profile name, when one was selected.
    pub emulation_profile: Option<String>,
}

/// Complete frontend-facing description of one SDK-owned runtime.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FrontendRuntimeDescriptor {
    /// Contract schema version.
    pub schema_version: u32,
    /// Stable SDK runtime/session identity.
    pub session_id: String,
    /// Source harness whose semantics are being emulated.
    pub source_harness: Option<String>,
    /// Resolved composable preset/profile name.
    pub emulation_profile: Option<String>,
    /// Active composable modules, using their stable config keys.
    pub active_modules: Vec<String>,
    /// Runtime-provided composer commands.
    pub commands: Vec<FrontendCommandDescriptor>,
    /// Explicit typed operation catalog. Missing on schema-v1 peers.
    #[serde(default)]
    pub operations: Vec<FrontendOperationDescriptor>,
    /// Supported control actions.
    pub actions: FrontendActions,
    /// Display/event capabilities.
    pub display: FrontendDisplayCapabilities,
    /// Current model label.
    pub model: String,
    /// Current turn state.
    pub turn_state: FrontendTurnState,
    /// Current connection state.
    pub connection_state: FrontendConnectionState,
    /// Compatible client/adapter metadata with no canonical-session meaning.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub extensions: BTreeMap<String, Value>,
}

/// Serializable half of an attachment returned by an out-of-process runtime.
/// The live receiver is transport-owned and joined to this snapshot locally.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct FrontendAttachSnapshot {
    /// Runtime description captured at attachment time.
    pub descriptor: FrontendRuntimeDescriptor,
    /// Bounded canonical history through `history_cursor`.
    pub history: Vec<ChatMessage>,
    /// Highest event sequence represented by `history`.
    pub history_cursor: u64,
    /// Events after the canonical history boundary and before the response.
    pub replay: VecDeque<FrontendEvent>,
}

/// Kind of interactive request surfaced by the SDK runtime.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FrontendRequestKind {
    /// A tool or sandbox action needs a policy-authorized human decision.
    Approval,
    /// An MCP server requested structured user input.
    Elicitation,
    /// Another versioned runtime request not known to this frontend build.
    /// Its complete payload remains available for a generic overlay.
    #[serde(other)]
    Other,
}

/// One pending interactive request, emitted as a sequenced frontend event.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct FrontendRequest {
    /// Runtime-scoped request identifier used exactly once by `respond`.
    pub id: u64,
    /// Typed request category.
    pub kind: FrontendRequestKind,
    /// Complete request payload, including raw tool arguments or schema.
    pub payload: Value,
}

/// Typed approval decision accepted by [`FrontendRuntime::respond`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FrontendApprovalDecision {
    /// Refuse this request.
    Deny,
    /// Allow only this request.
    Allow,
    /// Allow this request and cache the exact policy key for the session.
    AllowForSession,
}

/// MCP elicitation outcome accepted by a frontend response.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FrontendElicitationAction {
    /// Submit structured content.
    Accept,
    /// Explicitly decline the request.
    Decline,
    /// Dismiss the request without a decision.
    Cancel,
}

/// Typed response to one SDK-owned interactive request.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum FrontendResponse {
    /// Answer an approval request.
    Approval {
        /// Identifier from [`FrontendRequest::id`].
        request_id: u64,
        /// Human decision.
        decision: FrontendApprovalDecision,
    },
    /// Answer an MCP elicitation request.
    Elicitation {
        /// Identifier from [`FrontendRequest::id`].
        request_id: u64,
        /// MCP elicitation outcome.
        action: FrontendElicitationAction,
        /// Structured content for `accept`.
        content: Option<Value>,
    },
    /// Answer a generic runtime request without discarding its payload.
    Other {
        /// Identifier from [`FrontendRequest::id`].
        request_id: u64,
        /// Generic accept/decline/cancel outcome.
        action: FrontendElicitationAction,
        /// Optional structured response content.
        content: Option<Value>,
    },
}

impl FrontendResponse {
    pub(crate) fn request_id(&self) -> u64 {
        match self {
            Self::Approval { request_id, .. }
            | Self::Elicitation { request_id, .. }
            | Self::Other { request_id, .. } => *request_id,
        }
    }
}

/// Atomic history/replay/live attachment to one runtime.
pub struct FrontendAttachment {
    /// Runtime description captured at attachment time.
    pub descriptor: FrontendRuntimeDescriptor,
    /// Bounded canonical history through `history_cursor`.
    pub history: Vec<ChatMessage>,
    /// Highest event sequence already represented by `history`.
    pub history_cursor: u64,
    pub(crate) replay: VecDeque<FrontendEvent>,
    live: broadcast::Receiver<FrontendEvent>,
    delivered: u64,
    acknowledged: Option<Arc<AtomicU64>>,
    _transport_lease: Option<Arc<()>>,
}

impl FrontendAttachment {
    /// Build an in-process attachment from a serialized snapshot and a live
    /// SDK event receiver. Runtime adapters use this constructor in tests and
    /// protocol bridges without acquiring transport ownership.
    pub fn from_snapshot(
        snapshot: FrontendAttachSnapshot,
        live: broadcast::Receiver<FrontendEvent>,
    ) -> Self {
        Self::from_snapshot_after(snapshot, live, 0)
    }

    /// Build an attachment that resumes after a sequence acknowledged by a
    /// prior transport connection. Snapshot replay and any overlapping live
    /// events at or below the cursor are skipped without changing canonical
    /// history or event payloads.
    pub fn from_snapshot_after(
        snapshot: FrontendAttachSnapshot,
        live: broadcast::Receiver<FrontendEvent>,
        acknowledged_sequence: u64,
    ) -> Self {
        let delivered = snapshot.history_cursor.max(acknowledged_sequence);
        Self::new_with_delivered(
            snapshot.descriptor,
            snapshot.history,
            snapshot.history_cursor,
            snapshot.replay,
            live,
            None,
            delivered,
        )
    }

    pub(crate) fn new(
        descriptor: FrontendRuntimeDescriptor,
        history: Vec<ChatMessage>,
        history_cursor: u64,
        replay: VecDeque<FrontendEvent>,
        live: broadcast::Receiver<FrontendEvent>,
        transport_lease: Option<Arc<()>>,
    ) -> Self {
        let delivered = history_cursor;
        Self::new_with_delivered(
            descriptor,
            history,
            history_cursor,
            replay,
            live,
            transport_lease,
            delivered,
        )
    }

    fn new_with_delivered(
        descriptor: FrontendRuntimeDescriptor,
        history: Vec<ChatMessage>,
        history_cursor: u64,
        replay: VecDeque<FrontendEvent>,
        live: broadcast::Receiver<FrontendEvent>,
        transport_lease: Option<Arc<()>>,
        delivered: u64,
    ) -> Self {
        Self {
            descriptor,
            history,
            history_cursor,
            replay,
            live,
            delivered,
            acknowledged: None,
            _transport_lease: transport_lease,
        }
    }

    #[cfg(feature = "adapter-acp")]
    pub(crate) fn with_acknowledgement(mut self, acknowledged: Arc<AtomicU64>) -> Self {
        acknowledged.fetch_max(self.history_cursor, Ordering::SeqCst);
        self.acknowledged = Some(acknowledged);
        self
    }

    fn acknowledge(&self, event: &FrontendEvent) {
        if !event_advances_acknowledgement(event) {
            return;
        }
        if let Some(acknowledged) = &self.acknowledged {
            acknowledged.fetch_max(event.sequence, Ordering::SeqCst);
        }
    }

    /// Receive the next event not already represented by the history or a
    /// prior replay item. Duplicate events queued during attachment are
    /// skipped by sequence.
    pub async fn next_event(&mut self) -> Result<FrontendEvent, FrontendRuntimeError> {
        loop {
            let event = match self.next_replay_event() {
                Some(event) => return Ok(event),
                None => match self.live.recv().await {
                    Ok(event) => event,
                    Err(broadcast::error::RecvError::Lagged(count)) => {
                        return Err(FrontendRuntimeError::ReplayGap(count));
                    }
                    Err(broadcast::error::RecvError::Closed) => {
                        return Err(FrontendRuntimeError::Closed);
                    }
                },
            };
            if event.sequence <= self.delivered {
                continue;
            }
            self.delivered = event.sequence;
            self.acknowledge(&event);
            return Ok(event);
        }
    }

    /// Drain one event from the finite attachment replay without waiting for
    /// live input. Interactive frontends use this to project the complete
    /// atomic snapshot before accepting keystrokes, so a historical resolved
    /// request never appears transiently actionable.
    pub fn next_replay_event(&mut self) -> Option<FrontendEvent> {
        while let Some(event) = self.replay.pop_front() {
            if event.sequence <= self.delivered {
                continue;
            }
            self.delivered = event.sequence;
            self.acknowledge(&event);
            return Some(event);
        }
        None
    }
}

pub(crate) fn event_advances_acknowledgement(event: &FrontendEvent) -> bool {
    event
        .payload
        .pointer("/_meta/supercode/transient")
        .and_then(Value::as_bool)
        != Some(true)
}

/// State protected by `RpcEngine`'s short synchronous projection lock.
pub(crate) struct FrontendProjectionState {
    pub(crate) history: Vec<ChatMessage>,
    pub(crate) history_cursor: u64,
    pub(crate) next_sequence: u64,
    pub(crate) replay: VecDeque<FrontendEvent>,
}

#[async_trait]
impl FrontendRuntime for RpcEngine {
    async fn describe(&self) -> Result<FrontendRuntimeDescriptor, FrontendRuntimeError> {
        Ok(self.frontend_descriptor())
    }

    async fn attach(
        &self,
        history_limit: usize,
    ) -> Result<FrontendAttachment, FrontendRuntimeError> {
        self.frontend_attach(history_limit)
    }

    async fn send_input(self: Arc<Self>, prompt: String) -> Result<(), FrontendRuntimeError> {
        RpcEngine::send_input(&self, prompt)?;
        Ok(())
    }

    async fn send_input_with_images(
        self: Arc<Self>,
        prompt: String,
        image_urls: Vec<String>,
    ) -> Result<(), FrontendRuntimeError> {
        RpcEngine::send_input_with_images(&self, prompt, image_urls)?;
        Ok(())
    }

    async fn submit(&self, prompt: String) -> Result<String, FrontendRuntimeError> {
        Ok(RpcEngine::submit(self, prompt).await?)
    }

    async fn submit_with_images(
        &self,
        prompt: String,
        image_urls: Vec<String>,
    ) -> Result<String, FrontendRuntimeError> {
        Ok(RpcEngine::submit_with_images(self, prompt, image_urls).await?)
    }

    async fn interrupt(&self) -> Result<bool, FrontendRuntimeError> {
        Ok(RpcEngine::interrupt(self).await)
    }

    async fn steer(&self, prompt: String) -> Result<(), FrontendRuntimeError> {
        RpcEngine::steer(self, prompt)
    }

    async fn respond(&self, response: FrontendResponse) -> Result<(), FrontendRuntimeError> {
        RpcEngine::respond(self, response)
    }

    async fn invoke(
        &self,
        operation: FrontendOperationInvocation,
    ) -> Result<FrontendOperationResult, FrontendRuntimeError> {
        RpcEngine::invoke(self, operation).await
    }

    async fn close(&self) -> Result<(), FrontendRuntimeError> {
        RpcEngine::shutdown(self).await;
        Ok(())
    }
}

/// Authenticated HTTP implementation of [`FrontendRuntime`].
///
/// It owns only an RPC/SSE connection. The remote [`RpcEngine`] remains the
/// sole owner of the agent loop, transcript, scheduler, and persistence.
#[cfg(feature = "adapter-api")]
pub struct HttpFrontendRuntime {
    base_url: String,
    token: String,
    client_id: crate::RuntimeClientId,
    authorization: crate::RuntimeAuthorization,
    client: reqwest::Client,
    events: broadcast::Sender<FrontendEvent>,
    next_id: AtomicU64,
    lifecycle: Arc<()>,
    disconnected: AtomicBool,
}

#[cfg(feature = "adapter-api")]
/// How long a kept probe client may go unused before it is let go.
const KEPT_PROBE_IDLE: std::time::Duration = std::time::Duration::from_secs(600);

type KeptProbes =
    std::collections::HashMap<(String, String), (Arc<HttpFrontendRuntime>, std::time::Instant)>;

fn kept_probes() -> &'static std::sync::Mutex<KeptProbes> {
    static KEPT: std::sync::OnceLock<std::sync::Mutex<KeptProbes>> = std::sync::OnceLock::new();
    KEPT.get_or_init(Default::default)
}

/// The runtime every kept probe client runs on (see [`HttpFrontendRuntime::probe_described_kept`]).
fn kept_probe_runtime() -> &'static tokio::runtime::Runtime {
    static RUNTIME: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new();
    RUNTIME.get_or_init(|| {
        tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .thread_name("supercode-runtime-probes")
            .enable_all()
            .build()
            .expect("the runtime-probe runtime starts")
    })
}

impl HttpFrontendRuntime {
    /// Authenticate, verify the frontend descriptor, and establish the
    /// sequenced SSE stream before returning.
    pub async fn connect(
        base_url: impl Into<String>,
        token: impl Into<String>,
    ) -> Result<Arc<Self>, FrontendRuntimeError> {
        let mut random = [0_u8; 16];
        getrandom::getrandom(&mut random).map_err(|error| {
            FrontendRuntimeError::Transport(format!(
                "cannot generate runtime client identity: {error}"
            ))
        })?;
        let suffix = random
            .iter()
            .map(|byte| format!("{byte:02x}"))
            .collect::<String>();
        let client_id = crate::RuntimeClientId::parse(format!("http-{suffix}"))
            .map_err(|error| FrontendRuntimeError::Transport(error.to_string()))?;
        Self::connect_with_client_id(base_url, token, client_id).await
    }

    /// Connect with a caller-owned stable client identity. Reconnect tests
    /// and external bindings use this to retain deterministic lease state.
    pub async fn connect_with_client_id(
        base_url: impl Into<String>,
        token: impl Into<String>,
        client_id: crate::RuntimeClientId,
    ) -> Result<Arc<Self>, FrontendRuntimeError> {
        Self::connect_with_authorization(
            base_url,
            token,
            client_id,
            crate::RuntimeAuthorization::owner(),
        )
        .await
    }

    /// Connect while requesting an exact subset of the bearer credential's
    /// permissions. The server intersects this with the authenticated grant;
    /// this header can narrow authority but can never elevate it.
    pub async fn connect_with_authorization(
        base_url: impl Into<String>,
        token: impl Into<String>,
        client_id: crate::RuntimeClientId,
        authorization: crate::RuntimeAuthorization,
    ) -> Result<Arc<Self>, FrontendRuntimeError> {
        Self::connect_inner(base_url, token, client_id, authorization, true)
            .await
            .map(|(runtime, _)| runtime)
    }

    /// Authenticated metadata probe that does not open an event stream or
    /// register an observer, returning the descriptor the connect handshake
    /// already fetched. Used by the local runtime registry, which runs this on
    /// every `harness serve` tick for every followed session: asking the same
    /// runtime to describe itself twice for one read is pure load on that
    /// path, and each round trip costs its own loopback connection.
    pub(crate) async fn probe_described(
        base_url: impl Into<String>,
        token: impl Into<String>,
        client_id: crate::RuntimeClientId,
    ) -> Result<(Arc<Self>, FrontendRuntimeDescriptor), FrontendRuntimeError> {
        Self::connect_inner(
            base_url,
            token,
            client_id,
            crate::RuntimeAuthorization::observer(),
            false,
        )
        .await
    }

    /// [`Self::probe_described`] through a client kept per endpoint: a reader that asks every live
    /// runtime to describe itself every few seconds (the mail watcher, the registry's serve tick)
    /// reuses one kept-alive connection to each instead of opening, and leaving in TIME_WAIT, a
    /// connection per read (t_f3f102db t_4f29b43e). Every kept client lives on one long-lived tokio
    /// runtime, so its pooled connections outlive the caller's runtime; a client that fails once is
    /// dropped and the next read connects afresh, and one unused for ten minutes is let go.
    pub(crate) async fn probe_described_kept(
        base_url: String,
        token: String,
        client_id: crate::RuntimeClientId,
    ) -> Result<(Arc<Self>, FrontendRuntimeDescriptor), FrontendRuntimeError> {
        kept_probe_runtime()
            .spawn(async move {
                let key = (base_url.clone(), token.clone());
                let kept = {
                    let mut kept = kept_probes().lock().unwrap_or_else(|e| e.into_inner());
                    kept.retain(|_, (_, used)| used.elapsed() < KEPT_PROBE_IDLE);
                    kept.get(&key).map(|(runtime, _)| runtime.clone())
                };
                if let Some(runtime) = kept {
                    match FrontendRuntime::describe(runtime.as_ref()).await {
                        Ok(descriptor) => {
                            if let Some(entry) = kept_probes()
                                .lock()
                                .unwrap_or_else(|e| e.into_inner())
                                .get_mut(&key)
                            {
                                entry.1 = std::time::Instant::now();
                            }
                            return Ok((runtime, descriptor));
                        }
                        Err(_) => {
                            kept_probes()
                                .lock()
                                .unwrap_or_else(|e| e.into_inner())
                                .remove(&key);
                        }
                    }
                }
                let (runtime, descriptor) =
                    Self::probe_described(base_url, token, client_id).await?;
                kept_probes()
                    .lock()
                    .unwrap_or_else(|e| e.into_inner())
                    .insert(key, (runtime.clone(), std::time::Instant::now()));
                Ok((runtime, descriptor))
            })
            .await
            .map_err(|error| FrontendRuntimeError::Transport(error.to_string()))?
    }

    async fn connect_inner(
        base_url: impl Into<String>,
        token: impl Into<String>,
        client_id: crate::RuntimeClientId,
        authorization: crate::RuntimeAuthorization,
        stream_events: bool,
    ) -> Result<(Arc<Self>, FrontendRuntimeDescriptor), FrontendRuntimeError> {
        let runtime = Arc::new(Self {
            base_url: base_url.into().trim_end_matches('/').to_string(),
            token: token.into(),
            client_id,
            authorization,
            client: reqwest::Client::new(),
            events: broadcast::channel(1024).0,
            next_id: AtomicU64::new(1),
            lifecycle: Arc::new(()),
            disconnected: AtomicBool::new(false),
        });
        // Validate auth and schema before opening a long-lived connection.
        let descriptor: FrontendRuntimeDescriptor = runtime
            .rpc_typed(crate::FrontendFacadeMethod::Describe.wire_name(), json!({}))
            .await?;
        if stream_events {
            Self::start_event_stream(&runtime).await?;
        }
        Ok((runtime, descriptor))
    }

    async fn start_event_stream(runtime: &Arc<Self>) -> Result<(), FrontendRuntimeError> {
        let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
        let weak = Arc::downgrade(runtime);
        let lifecycle = Arc::downgrade(&runtime.lifecycle);
        tokio::spawn(async move {
            Self::run_event_stream(weak, lifecycle, ready_tx).await;
        });
        ready_rx.await.map_err(|_| {
            FrontendRuntimeError::Transport("frontend event stream exited before startup".into())
        })?
    }

    async fn run_event_stream(
        weak: Weak<Self>,
        lifecycle: Weak<()>,
        ready: tokio::sync::oneshot::Sender<Result<(), FrontendRuntimeError>>,
    ) {
        let Some(runtime) = weak.upgrade() else {
            let _ = ready.send(Err(FrontendRuntimeError::Closed));
            return;
        };
        let request = runtime
            .client
            .get(format!("{}/frontend/events", runtime.base_url))
            .bearer_auth(&runtime.token)
            .header("x-supercode-client-id", runtime.client_id.as_str())
            .header(
                "x-supercode-permissions",
                runtime.authorization.header_value(),
            );
        let events = runtime.events.clone();
        drop(runtime);
        let response = request.send().await;
        let response = match response {
            Ok(response) if response.status().is_success() => response,
            Ok(response) => {
                let _ = ready.send(Err(FrontendRuntimeError::Transport(format!(
                    "frontend event stream returned {}",
                    response.status()
                ))));
                return;
            }
            Err(error) => {
                let _ = ready.send(Err(FrontendRuntimeError::Transport(error.to_string())));
                return;
            }
        };
        let _ = ready.send(Ok(()));
        let mut stream = response.bytes_stream();
        let mut pending = Vec::<u8>::new();
        let mut liveness = tokio::time::interval(std::time::Duration::from_millis(100));
        loop {
            let chunk = tokio::select! {
                _ = liveness.tick() => {
                    if lifecycle.strong_count() == 0 {
                        break;
                    }
                    if weak
                        .upgrade()
                        .is_some_and(|runtime| runtime.disconnected.load(Ordering::SeqCst))
                    {
                        break;
                    }
                    continue;
                }
                chunk = stream.next() => chunk,
            };
            let Some(chunk) = chunk else {
                break;
            };
            let Ok(chunk) = chunk else {
                break;
            };
            pending.extend_from_slice(&chunk);
            while let Some(position) = pending.iter().position(|byte| *byte == b'\n') {
                let line = pending.drain(..=position).collect::<Vec<_>>();
                let line = String::from_utf8_lossy(&line);
                let Some(data) = line.trim_end().strip_prefix("data: ") else {
                    continue;
                };
                if let Ok(event) = serde_json::from_str::<FrontendEvent>(data) {
                    let _ = events.send(event);
                }
            }
        }
        if let Some(runtime) = weak.upgrade() {
            runtime.disconnected.store(true, Ordering::SeqCst);
            let _ = runtime.events.send(FrontendEvent::new(
                u64::MAX,
                json!({
                    "type": "runtime_disconnected",
                    "schema_version": FRONTEND_EVENT_SCHEMA_VERSION
                }),
            ));
        }
    }

    async fn rpc(&self, method: &str, params: Value) -> Result<Value, FrontendRuntimeError> {
        let id = self.next_id.fetch_add(1, Ordering::SeqCst);
        let requested_operation = params
            .pointer("/operation/operation_id")
            .and_then(Value::as_str)
            .map(str::to_owned);
        let response = self
            .client
            .post(format!("{}/rpc", self.base_url))
            .bearer_auth(&self.token)
            .header("x-supercode-client-id", self.client_id.as_str())
            .header("x-supercode-permissions", self.authorization.header_value())
            .json(&json!({"id": id, "method": method, "params": params}))
            .send()
            .await
            .map_err(|error| FrontendRuntimeError::Transport(error.to_string()))?;
        if !response.status().is_success() {
            return Err(FrontendRuntimeError::Transport(format!(
                "SDK HTTP RPC returned {}",
                response.status()
            )));
        }
        let value: Value = response
            .json()
            .await
            .map_err(|error| FrontendRuntimeError::Transport(error.to_string()))?;
        if let Some(error) = value.get("error") {
            let code = error.get("code").and_then(Value::as_i64);
            let name = error.get("name").and_then(Value::as_str);
            let operation = error
                .get("operation")
                .and_then(Value::as_str)
                .and_then(crate::SdkOperation::from_action_name);
            let message = error
                .get("message")
                .and_then(Value::as_str)
                .unwrap_or("SDK runtime request failed")
                .to_string();
            return Err(match (name, code) {
                (Some("unauthenticated"), _) | (_, Some(-32030)) => {
                    FrontendRuntimeError::Unauthenticated
                }
                (Some("unauthorized"), _) | (_, Some(-32031)) => {
                    FrontendRuntimeError::Unauthorized {
                        permission: error
                            .get("permission")
                            .and_then(Value::as_str)
                            .unwrap_or("unknown")
                            .to_string(),
                    }
                }
                (Some("controller_required"), _) | (_, Some(-32032)) => {
                    FrontendRuntimeError::ControllerRequired {
                        holder: error
                            .get("holder")
                            .and_then(Value::as_str)
                            .map(str::to_owned),
                        expires_at_ms: error.get("expiresAtMs").and_then(Value::as_u64),
                    }
                }
                (Some("lease_expired"), _) | (_, Some(-32033)) => {
                    FrontendRuntimeError::LeaseExpired
                }
                (_, Some(-32023)) => FrontendRuntimeError::UnsupportedOperation(
                    requested_operation.unwrap_or(message),
                ),
                (Some("unsupported_action"), _) => FrontendRuntimeError::UnsupportedAction(
                    operation
                        .unwrap_or_else(|| {
                            crate::SdkOperation::from_action_name(method)
                                .unwrap_or(crate::SdkOperation::Respond)
                        })
                        .action_name(),
                ),
                (Some("not_found"), Some(-32021)) => {
                    let request_id = params
                        .pointer("/response/request_id")
                        .and_then(Value::as_u64)
                        .unwrap_or_default();
                    FrontendRuntimeError::UnknownRequest(request_id)
                }
                (Some("invalid_argument"), _) => FrontendRuntimeError::InvalidResponse(message),
                (_, Some(-32000)) => RuntimeSubmitError::Busy.into(),
                (_, Some(-32001)) => RuntimeSubmitError::Interrupted.into(),
                (_, Some(-32002)) => RuntimeSubmitError::Agent(message).into(),
                (_, Some(-32020)) => FrontendRuntimeError::UnsupportedAction(
                    crate::SdkOperation::from_action_name(method)
                        .unwrap_or(crate::SdkOperation::Respond)
                        .action_name(),
                ),
                (_, Some(-32021)) => {
                    let request_id = params
                        .pointer("/response/request_id")
                        .and_then(Value::as_u64)
                        .unwrap_or_default();
                    FrontendRuntimeError::UnknownRequest(request_id)
                }
                (_, Some(-32022)) => FrontendRuntimeError::InvalidResponse(message),
                _ => FrontendRuntimeError::Transport(message),
            });
        }
        Ok(value.get("result").cloned().unwrap_or(Value::Null))
    }

    async fn rpc_typed<T: serde::de::DeserializeOwned>(
        &self,
        method: &str,
        params: Value,
    ) -> Result<T, FrontendRuntimeError> {
        serde_json::from_value(self.rpc(method, params).await?)
            .map_err(|error| FrontendRuntimeError::Transport(error.to_string()))
    }

    /// Current authenticated client identity.
    pub fn client_id(&self) -> &crate::RuntimeClientId {
        &self.client_id
    }

    /// Whether the remote event stream has already ended or this client
    /// explicitly detached/closed.
    pub fn is_disconnected(&self) -> bool {
        self.disconnected.load(Ordering::SeqCst)
    }

    /// Explicitly acquire the controller lease, displacing another
    /// interactive client only through this named operation.
    pub async fn take_control(&self) -> Result<crate::RuntimeLeaseSnapshot, FrontendRuntimeError> {
        self.rpc_typed(
            crate::FrontendFacadeMethod::TakeControl.wire_name(),
            json!({}),
        )
        .await
    }

    /// Acquire control when unowned without displacing a live controller.
    pub async fn acquire_control(
        &self,
    ) -> Result<crate::RuntimeLeaseSnapshot, FrontendRuntimeError> {
        self.rpc_typed(
            crate::FrontendFacadeMethod::AcquireControl.wire_name(),
            json!({}),
        )
        .await
    }

    /// Renew observer activity and any controller lease owned by this client.
    pub async fn heartbeat(&self) -> Result<crate::RuntimeLeaseSnapshot, FrontendRuntimeError> {
        self.rpc_typed(
            crate::FrontendFacadeMethod::Heartbeat.wire_name(),
            json!({}),
        )
        .await
    }

    /// Read the coordinated ownership state.
    pub async fn lease_snapshot(
        &self,
    ) -> Result<crate::RuntimeLeaseSnapshot, FrontendRuntimeError> {
        self.rpc_typed(crate::FrontendFacadeMethod::Lease.wire_name(), json!({}))
            .await
    }

    /// Release observer and controller state without stopping the runtime.
    pub async fn detach(&self) -> Result<crate::RuntimeLeaseSnapshot, FrontendRuntimeError> {
        let snapshot = self
            .rpc_typed(crate::FrontendFacadeMethod::Detach.wire_name(), json!({}))
            .await?;
        self.disconnected.store(true, Ordering::SeqCst);
        Ok(snapshot)
    }
}

#[async_trait]
#[cfg(feature = "adapter-api")]
impl FrontendRuntime for HttpFrontendRuntime {
    async fn describe(&self) -> Result<FrontendRuntimeDescriptor, FrontendRuntimeError> {
        self.rpc_typed(crate::FrontendFacadeMethod::Describe.wire_name(), json!({}))
            .await
    }

    async fn attach(
        &self,
        history_limit: usize,
    ) -> Result<FrontendAttachment, FrontendRuntimeError> {
        if self.disconnected.load(Ordering::SeqCst) {
            return Err(FrontendRuntimeError::Closed);
        }
        // Subscribe locally before asking the server for its atomic snapshot.
        // Anything concurrently received over SSE is either in snapshot.replay
        // or queued here; sequence filtering removes the overlap.
        let live = self.events.subscribe();
        let snapshot: FrontendAttachSnapshot = self
            .rpc_typed(
                crate::FrontendFacadeMethod::Attach.wire_name(),
                json!({"limit": history_limit}),
            )
            .await?;
        Ok(FrontendAttachment::new(
            snapshot.descriptor,
            snapshot.history,
            snapshot.history_cursor,
            snapshot.replay,
            live,
            Some(self.lifecycle.clone()),
        ))
    }

    async fn send_input(self: Arc<Self>, prompt: String) -> Result<(), FrontendRuntimeError> {
        self.send_input_with_images(prompt, Vec::new()).await
    }

    async fn send_input_with_images(
        self: Arc<Self>,
        prompt: String,
        image_urls: Vec<String>,
    ) -> Result<(), FrontendRuntimeError> {
        self.rpc(
            crate::FrontendFacadeMethod::SendInput.wire_name(),
            json!({"prompt": prompt, "image_urls": image_urls}),
        )
        .await?;
        Ok(())
    }

    async fn submit(&self, prompt: String) -> Result<String, FrontendRuntimeError> {
        let result = self
            .rpc(
                crate::FrontendFacadeMethod::Submit.wire_name(),
                json!({"prompt": prompt}),
            )
            .await?;
        Ok(result
            .get("reply")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_string())
    }

    async fn submit_with_images(
        &self,
        prompt: String,
        image_urls: Vec<String>,
    ) -> Result<String, FrontendRuntimeError> {
        let result = self
            .rpc(
                crate::FrontendFacadeMethod::Submit.wire_name(),
                json!({"prompt": prompt, "image_urls": image_urls}),
            )
            .await?;
        Ok(result
            .get("reply")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_string())
    }

    async fn interrupt(&self) -> Result<bool, FrontendRuntimeError> {
        let result = self
            .rpc(
                crate::FrontendFacadeMethod::Interrupt.wire_name(),
                json!({}),
            )
            .await?;
        Ok(result
            .get("interrupted")
            .and_then(Value::as_bool)
            .unwrap_or(false))
    }

    async fn steer(&self, prompt: String) -> Result<(), FrontendRuntimeError> {
        self.rpc(
            crate::FrontendFacadeMethod::Steer.wire_name(),
            json!({"prompt": prompt}),
        )
        .await?;
        Ok(())
    }

    async fn respond(&self, response: FrontendResponse) -> Result<(), FrontendRuntimeError> {
        self.rpc(
            crate::FrontendFacadeMethod::Respond.wire_name(),
            json!({"response": response}),
        )
        .await?;
        Ok(())
    }

    async fn invoke(
        &self,
        operation: FrontendOperationInvocation,
    ) -> Result<FrontendOperationResult, FrontendRuntimeError> {
        self.rpc_typed(
            crate::FrontendFacadeMethod::Invoke.wire_name(),
            json!({"operation": operation}),
        )
        .await
    }

    async fn lease_snapshot(&self) -> Result<crate::RuntimeLeaseSnapshot, FrontendRuntimeError> {
        HttpFrontendRuntime::lease_snapshot(self).await
    }

    async fn take_control(&self) -> Result<crate::RuntimeLeaseSnapshot, FrontendRuntimeError> {
        HttpFrontendRuntime::take_control(self).await
    }

    async fn acquire_control(&self) -> Result<crate::RuntimeLeaseSnapshot, FrontendRuntimeError> {
        HttpFrontendRuntime::acquire_control(self).await
    }

    async fn heartbeat(&self) -> Result<crate::RuntimeLeaseSnapshot, FrontendRuntimeError> {
        HttpFrontendRuntime::heartbeat(self).await
    }

    async fn detach(&self) -> Result<crate::RuntimeLeaseSnapshot, FrontendRuntimeError> {
        HttpFrontendRuntime::detach(self).await
    }

    async fn close(&self) -> Result<(), FrontendRuntimeError> {
        self.rpc(crate::FrontendFacadeMethod::Close.wire_name(), json!({}))
            .await?;
        self.disconnected.store(true, Ordering::SeqCst);
        Ok(())
    }
}