supercode_harness/

server.rs

//! §2 module 31 `server` (COMPOSABLE-HARNESS-DESIGN.md, D7 "full
//! programmatic RPC/HTTP server", D8 "remote attach", D10 "daemon"; §1.9
//! Obligation 9's out-of-process half — the in-process SDK already meets
//! the core commitment via [`crate::EventSink`]).
//!
//! The embedding ladder this module builds:
//!
//! 1. **`--output-format stream-json`** (the CLI's existing rung, UX-23) —
//!    already ships a JSONL [`crate::AgentEvent`] stream over stdout. This
//!    unit completes it: [`crate::AgentEvent::to_json`] is now the single
//!    canonical projection both that sink AND this module's notifications
//!    share, and it covers the FULL event set (previously
//!    [`crate::AgentEvent::BackgroundOutput`] fell into a generic
//!    "unknown" catch-all).
//! 2. **JSONL-RPC over stdio** ([`run_stdio`]) — the SDK-out-of-process
//!    surface: a parent process drives this agent's loop over stdin/stdout
//!    with `{"id","method","params"}` request lines, getting back
//!    `{"id","result"|"error"}` responses interleaved with
//!    `{"event":...}` notifications. Parent-process-trusted (same trust
//!    model as [`crate::mcp::serve_stdio`]) — no auth token.
//! 3. **The same RPC surface over HTTP** ([`run_http`], D8 "remote
//!    attach") — `POST /rpc` for request/response, `GET /events` for the
//!    event stream (SSE-shaped: `data: <json>\n\n` per line). Unlike
//!    stdio, a network client is UNTRUSTED by default, so every request
//!    must carry the bearer token (`check_auth`).
//!
//! **Security posture (this is a listener — the highest-risk module
//! class):**
//! - `[capabilities.server]` is project-forbidden (D-10) — see
//!   `crates/cli/src/userconfig.rs`'s `PROJECT_FORBIDDEN_CAPABILITY_TABLES`
//!   and this crate's `configfile::PROJECT_FORBIDDEN_CAPABILITY_TABLES`
//!   (both already listed `"server"` before this unit landed; this module
//!   is what makes the listener the strip was already guarding against
//!   real).
//! - Default-off: nothing in this module is ever reached unless a caller
//!   explicitly invokes [`run_stdio`]/[`run_http`] AND the CLI's own
//!   gate (`capabilities.server.enabled == Some(true)`, checked before
//!   either is called) passed.
//! - Loopback-only HTTP bind by default — enforced by the CALLER (the
//!   CLI's `serve` command defaults `bind` to `127.0.0.1:0` and only binds
//!   elsewhere on an explicit `bind`/`--bind` override, with a printed
//!   exposure warning); [`run_http`] itself binds whatever address it's
//!   given, since the loopback POLICY decision belongs to the config/CLI
//!   layer, not the transport.
//! - **No permission/sandbox bypass.** [`RpcEngine::new`] takes an already
//!   fully-constructed [`crate::Agent`] — the SAME `Agent` a local
//!   `run`/`chat` session would build (same `Config`, same permission
//!   rules, same sandbox). This module installs NO approval handler of its
//!   own and provides no channel for a remote/RPC caller to answer an
//!   approval prompt; combined with `Agent`'s existing fail-closed rule
//!   ("absent handler denies" — `crates/harness/src/agent.rs`'s
//!   `prepare_tool_call`), any tool call that would need interactive
//!   approval is DENIED, never silently approved, when driven through this
//!   module. See `crates/harness/tests/server_engine.rs` for a fail-on-revert
//!   proof.
//! - Bounded buffering throughout ([`SERVER_MAX_LINE_BYTES`],
//!   [`SERVER_EVENT_CHANNEL_CAPACITY`]) — same 16MiB-class discipline P5-2
//!   established for `crate::mcp`'s SSE reader, reused here rather than
//!   re-derived.
//! - Graceful shutdown: the `shutdown` RPC method stops the stdio loop and
//!   the HTTP accept loop alike (both select on the same
//!   [`RpcEngine::wait_for_shutdown`]) — no orphaned listener/accept task
//!   survives a `shutdown` call, mirroring P5-3/P5-6's drop-abort
//!   discipline for background work.

use std::collections::{BTreeMap, HashMap, VecDeque};
#[cfg(feature = "adapter-api")]
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex as StdMutex};

use async_trait::async_trait;
#[cfg(feature = "adapter-api")]
use futures::{SinkExt, StreamExt};
use serde_json::{json, Value};
use tokio::io::AsyncBufRead;
#[cfg(feature = "adapter-api")]
use tokio::io::{AsyncRead, AsyncReadExt};
#[cfg(feature = "adapter-api")]
use tokio::io::{AsyncWrite, AsyncWriteExt};
#[cfg(feature = "adapter-api")]
use tokio::net::TcpListener;
#[cfg(feature = "adapter-api")]
use tokio::sync::mpsc;
use tokio::sync::{broadcast, Mutex, Notify, RwLock};

use crate::agent::SteerInbox;
use crate::frontend::{
    FrontendActions, FrontendApprovalDecision, FrontendAttachSnapshot, FrontendAttachment,
    FrontendCommandDescriptor, FrontendConnectionState, FrontendDisplayCapabilities, FrontendEvent,
    FrontendOperationDescriptor, FrontendOperationInvocation, FrontendOperationKind,
    FrontendOperationResult, FrontendProjectionState, FrontendRequest, FrontendRequestKind,
    FrontendResponse, FrontendRuntime, FrontendRuntimeDescriptor, FrontendRuntimeError,
    FrontendRuntimeMetadata, FrontendTurnState, FRONTEND_EVENT_SCHEMA_VERSION,
    FRONTEND_REPLAY_CAPACITY, FRONTEND_RUNTIME_SCHEMA_VERSION,
};
use crate::mcp::{
    ElicitationAction, ElicitationRequest, ElicitationResponse, McpElicitationHandler,
};
use crate::message::ChatMessage;
use crate::permissions::{ApprovalOutcome, ApprovalRequest, PermissionsApprovalHandler};
pub use crate::sdk::RuntimeSubmitError;
use crate::sdk::SdkAgent;
#[cfg(feature = "adapter-api")]
use crate::{CoordinatedRuntime, CoordinatedRuntimeClient, RuntimeAuthorization, RuntimeClientId};

/// Bounded broadcast capacity for the event-notification channel — mirrors
/// `crate::mcp::MCP_SSE_CHANNEL_CAPACITY`'s bounded-buffering discipline
/// (P5-2): a slow/absent subscriber can never make the sender block or
/// grow memory unboundedly; a lagging receiver just misses old events
/// (`broadcast::error::RecvError::Lagged`) rather than stalling the agent
/// loop or accumulating unbounded backlog.
pub const SERVER_EVENT_CHANNEL_CAPACITY: usize = 1024;

/// Maximum canonical messages retained for late-attaching frontends. This
/// matches the `history` RPC limit and prevents the lock-independent snapshot
/// from duplicating an arbitrarily large agent transcript.
pub(crate) const SERVER_HISTORY_CAPACITY: usize = 200;

/// Maximum accepted line/body length (bytes) for both the stdio JSONL-RPC
/// reader and the HTTP transport's request line/headers/body — the same
/// 16MiB-class cap P5-2 established for `crate::mcp`'s SSE frame reader
/// (`MCP_MAX_SSE_FRAME_BYTES`), reused here so an adversarial or simply
/// broken client can never make either transport buffer an unbounded
/// amount of data in memory.
pub const SERVER_MAX_LINE_BYTES: usize = 16 * 1024 * 1024;

/// Cap on HTTP header line COUNT (independent of [`SERVER_MAX_LINE_BYTES`],
/// which only bounds any one line's length) — without this, a client could
/// send an unbounded NUMBER of small, individually-under-cap header lines
/// and still exhaust memory over one connection.
#[cfg(feature = "adapter-api")]
const MAX_HEADER_LINES: usize = 200;

/// One JSONL-RPC request line a client sends: `{"id", "method", "params"}`.
/// `params` defaults to `null` when omitted (a method that takes no
/// arguments, e.g. `status`/`shutdown`, never requires callers to spell out
/// `"params": null}` explicitly).
#[derive(Debug, Clone, serde::Deserialize)]
pub struct RpcRequest {
    /// Caller-chosen correlation id, echoed back verbatim on the matching
    /// response — never interpreted, so any JSON value (string, number,
    /// null) a caller likes works.
    pub id: Value,
    /// The method name (`submit` | `interrupt` | `status` | `shutdown`).
    pub method: String,
    /// Method-specific arguments; `submit` reads `params.prompt`.
    #[serde(default)]
    pub params: Value,
}

/// Build a `{"id", "result"}` response line.
fn rpc_ok(id: Value, result: Value) -> Value {
    json!({"id": id, "result": result})
}

/// Build a `{"id", "error": {"code","message"}}` response line. `code`
/// follows JSON-RPC 2.0's reserved-range convention where a natural fit
/// exists (`-32700` parse error, `-32601` method not found, `-32602`
/// invalid params) purely as a familiar, self-documenting convention — this
/// protocol does not otherwise claim JSON-RPC 2.0 compliance (no
/// `"jsonrpc":"2.0"` envelope; see the module doc's minimal wire shape).
fn rpc_error(id: Value, code: i32, message: impl Into<String>) -> Value {
    json!({"id": id, "error": {"code": code, "message": message.into()}})
}

fn sdk_runtime_rpc_error(id: Value, code: i32, error: &FrontendRuntimeError) -> Value {
    let code = match error.code() {
        crate::SdkErrorCode::Unauthenticated => -32030,
        crate::SdkErrorCode::Unauthorized => -32031,
        crate::SdkErrorCode::ControllerRequired => -32032,
        crate::SdkErrorCode::LeaseExpired => -32033,
        _ => code,
    };
    let mut envelope = json!({
        "id": id,
        "error": {
            "code": code,
            "name": error.code(),
            "operation": error.operation(),
            "message": error.to_string(),
        }
    });
    if let Some(detail) = envelope.get_mut("error").and_then(Value::as_object_mut) {
        match error {
            FrontendRuntimeError::Unauthorized { permission } => {
                detail.insert("permission".into(), Value::String(permission.clone()));
            }
            FrontendRuntimeError::ControllerRequired {
                holder,
                expires_at_ms,
            } => {
                if let Some(holder) = holder {
                    detail.insert("holder".into(), Value::String(holder.clone()));
                }
                if let Some(expires_at_ms) = expires_at_ms {
                    detail.insert("expiresAtMs".into(), json!(expires_at_ms));
                }
            }
            _ => {}
        }
    }
    envelope
}

/// Read one line (trailing `\n`/`\r\n` stripped) from `reader`, capped at
/// `cap` bytes — mirrors `crate::mcp`'s `SseLineAccumulator` bounded-
/// buffering discipline (P5-2). Returns `Ok(None)` at a clean EOF with no
/// partial line pending. On an over-cap line, the REST of that oversized
/// line is drained and discarded (up to the next `\n`) so the stream
/// resyncs at the next real line boundary instead of desyncing forever,
/// and `Err` is returned naming the cap.
async fn read_bounded_line<R>(reader: &mut R, cap: usize) -> std::io::Result<Option<String>>
where
    R: AsyncBufRead + Unpin,
{
    use tokio::io::AsyncBufReadExt;
    let mut out: Vec<u8> = Vec::new();
    loop {
        let buf = reader.fill_buf().await?;
        if buf.is_empty() {
            return Ok(if out.is_empty() {
                None
            } else {
                Some(strip_crlf(out))
            });
        }
        if let Some(pos) = buf.iter().position(|&b| b == b'\n') {
            if out.len() + pos > cap {
                reader.consume(pos + 1);
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    format!("line exceeded {cap} byte cap"),
                ));
            }
            out.extend_from_slice(&buf[..pos]);
            reader.consume(pos + 1);
            return Ok(Some(strip_crlf(out)));
        }
        let take = buf.len();
        if out.len() + take > cap {
            reader.consume(take);
            // Drain/discard the rest of this oversized line so a future
            // read starts at the next real line boundary.
            loop {
                let b = reader.fill_buf().await?;
                if b.is_empty() {
                    break;
                }
                if let Some(p) = b.iter().position(|&x| x == b'\n') {
                    reader.consume(p + 1);
                    break;
                }
                let n = b.len();
                reader.consume(n);
            }
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("line exceeded {cap} byte cap"),
            ));
        }
        out.extend_from_slice(buf);
        reader.consume(take);
    }
}

fn strip_crlf(mut v: Vec<u8>) -> String {
    if v.last() == Some(&b'\r') {
        v.pop();
    }
    String::from_utf8_lossy(&v).into_owned()
}

/// Constant-time byte-slice comparison (avoids leaking the bearer token
/// through a timing side-channel on `==`) — small, self-contained, no new
/// dependency for one comparison.
#[cfg(feature = "adapter-api")]
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff = 0u8;
    for (x, y) in a.iter().zip(b.iter()) {
        diff |= x ^ y;
    }
    diff == 0
}

/// Mint a random per-session bearer token (32 bytes, hex-encoded) for the
/// HTTP transport, when the operator hasn't configured a fixed
/// `capabilities.server.token`. Uses `getrandom` (already resolved
/// transitively via `reqwest`'s rustls/ring stack; promoted to a direct
/// dependency here so this crate can call it directly, rather than rolling
/// a hand-written PRNG for a value that must actually be unguessable).
pub fn generate_token() -> String {
    let mut bytes = [0u8; 32];
    // `getrandom::getrandom` only fails if the OS entropy source itself is
    // unavailable/misconfigured — effectively never on a real target this
    // crate supports. Falling back to a process-time/PID-derived value
    // would be a WORSE, easily-guessable token, so a failure here is
    // treated as fatal (panic) rather than silently minting a weak secret.
    getrandom::getrandom(&mut bytes).expect("OS entropy source for the server bearer token");
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

/// The `on_turn_complete` hook's type — factored out (clippy
/// `type_complexity`) rather than spelled out at both
/// [`RpcEngine`]'s field and [`RpcEngine::new`]'s parameter.
type TurnCompleteHook = Box<dyn Fn(&SdkAgent) + Send + Sync>;

/// Protocol-neutral snapshot of one SDK-owned agent runtime.
///
/// Transport adapters (CLI/RPC/HTTP/ACP) project this value into their own
/// wire shapes instead of each inventing a separate definition of "busy".
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct RuntimeStatus {
    /// Stable identity chosen by the embedding surface/session store.
    pub session_id: String,
    /// Model currently backing the agent.
    pub model: String,
    /// Whether a user or scheduler turn currently owns the agent loop.
    pub busy: bool,
    /// Whether graceful shutdown has been requested.
    pub shutting_down: bool,
}

type PendingFrontendResponses = StdMutex<
    HashMap<
        u64,
        (
            FrontendRequestKind,
            std::sync::mpsc::Sender<AcceptedFrontendResponse>,
        ),
    >,
>;

struct AcceptedFrontendResponse {
    response: FrontendResponse,
    /// The blocked handler may resume execution only after the canonical
    /// resolution event has entered the sequenced frontend projection.
    published: std::sync::mpsc::Receiver<()>,
}

/// Runtime-owned interactive request broker. It publishes complete request
/// payloads into the same sequenced event stream and resolves each id once.
struct FrontendRequestBroker {
    next_id: std::sync::atomic::AtomicU64,
    pending: PendingFrontendResponses,
    transport: StdMutex<Option<FrontendRequestTransport>>,
}

#[derive(Clone)]
struct FrontendRequestTransport {
    events: broadcast::Sender<FrontendEvent>,
    state: Arc<StdMutex<FrontendProjectionState>>,
}

impl FrontendRequestBroker {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            next_id: std::sync::atomic::AtomicU64::new(1),
            pending: StdMutex::new(HashMap::new()),
            transport: StdMutex::new(None),
        })
    }

    fn bind(
        &self,
        events: broadcast::Sender<FrontendEvent>,
        state: Arc<StdMutex<FrontendProjectionState>>,
    ) {
        *self
            .transport
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) =
            Some(FrontendRequestTransport { events, state });
    }

    fn transport(&self) -> Option<FrontendRequestTransport> {
        self.transport
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
    }

    fn publish(&self, request: &FrontendRequest) -> bool {
        self.publish_payload(json!({"type": "request", "request": request}))
    }

    fn publish_payload(&self, payload: Value) -> bool {
        let Some(transport) = self.transport() else {
            return false;
        };
        let event = {
            let mut state = transport
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            let event = FrontendEvent::new(state.next_sequence, payload);
            state.next_sequence = state.next_sequence.saturating_add(1);
            state.replay.push_back(event.clone());
            while state.replay.len() > FRONTEND_REPLAY_CAPACITY {
                state.replay.pop_front();
            }
            event
        };
        transport.events.send(event).is_ok()
    }

    fn respond(&self, response: FrontendResponse) -> Result<(), FrontendRuntimeError> {
        let request_id = response.request_id();
        let response_kind = match &response {
            FrontendResponse::Approval { .. } => FrontendRequestKind::Approval,
            FrontendResponse::Elicitation { .. } => FrontendRequestKind::Elicitation,
            FrontendResponse::Other { .. } => FrontendRequestKind::Other,
        };
        let mut pending = self
            .pending
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let expected = pending
            .get(&request_id)
            .map(|(kind, _)| *kind)
            .ok_or(FrontendRuntimeError::UnknownRequest(request_id))?;
        if expected != response_kind {
            return Err(FrontendRuntimeError::InvalidResponse(format!(
                "request {request_id} expects {expected:?}, got {response_kind:?}"
            )));
        }
        let (_, sender) = pending
            .remove(&request_id)
            .ok_or(FrontendRuntimeError::UnknownRequest(request_id))?;
        drop(pending);
        let payload = json!({
            "type": "request_resolved",
            "request_id": request_id,
            "response": &response,
        });
        let (published_tx, published_rx) = std::sync::mpsc::channel();
        sender
            .send(AcceptedFrontendResponse {
                response,
                published: published_rx,
            })
            .map_err(|_| FrontendRuntimeError::UnknownRequest(request_id))?;
        self.publish_payload(payload);
        let _ = published_tx.send(());
        Ok(())
    }

    fn ask_approval(
        &self,
        req: &ApprovalRequest<'_>,
        child: Option<(&str, &Arc<StdMutex<Vec<crate::subagents::QueuedApproval>>>)>,
    ) -> ApprovalOutcome {
        if let Some((child_agent_id, queue)) = child {
            queue
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .push(crate::subagents::QueuedApproval {
                    child_agent_id: child_agent_id.to_string(),
                    tool: req.tool.to_string(),
                    subject: req.subject.map(String::from),
                    queued_at_ms: std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .map(|duration| duration.as_millis() as i64)
                        .unwrap_or_default(),
                });
        }
        // No observer means no human can answer. Preserve the server's
        // existing fail-closed, non-blocking headless behavior.
        let Some(transport) = self.transport() else {
            return ApprovalOutcome::Deny;
        };
        if transport.events.receiver_count() == 0 {
            return ApprovalOutcome::Deny;
        }
        let id = self.next_id.fetch_add(1, Ordering::SeqCst);
        let mut payload = json!({
            "tool": req.tool,
            "subject": req.subject,
            "raw_args": req.raw_args,
        });
        if let Some((child_agent_id, _)) = child {
            payload["child_agent_id"] = Value::String(child_agent_id.to_string());
        }
        let request = FrontendRequest {
            id,
            kind: FrontendRequestKind::Approval,
            payload,
        };
        let (tx, rx) = std::sync::mpsc::channel();
        self.pending
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .insert(id, (FrontendRequestKind::Approval, tx));
        if !self.publish(&request) {
            self.pending
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .remove(&id);
            return ApprovalOutcome::Deny;
        }
        let wait_for_response = || loop {
            match rx.recv_timeout(std::time::Duration::from_millis(100)) {
                Ok(accepted) => {
                    let _ = accepted.published.recv();
                    let FrontendResponse::Approval { decision, .. } = accepted.response else {
                        return ApprovalOutcome::Deny;
                    };
                    return match decision {
                        FrontendApprovalDecision::Deny => ApprovalOutcome::Deny,
                        FrontendApprovalDecision::Allow => ApprovalOutcome::Allow,
                        FrontendApprovalDecision::AllowForSession => {
                            ApprovalOutcome::AllowForSession
                        }
                    };
                }
                Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
                    return ApprovalOutcome::Deny;
                }
                Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
                    if self
                        .transport()
                        .map(|transport| transport.events.receiver_count() == 0)
                        .unwrap_or(true)
                    {
                        self.pending
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner)
                            .remove(&id);
                        return ApprovalOutcome::Deny;
                    }
                }
            }
        };
        if tokio::runtime::Handle::try_current()
            .map(|handle| handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread)
            .unwrap_or(false)
        {
            tokio::task::block_in_place(wait_for_response)
        } else {
            wait_for_response()
        }
    }

    async fn ask_elicitation(self: Arc<Self>, req: &ElicitationRequest) -> ElicitationResponse {
        let cancel = || ElicitationResponse {
            action: ElicitationAction::Cancel,
            content: None,
        };
        let Some(transport) = self.transport() else {
            return cancel();
        };
        if transport.events.receiver_count() == 0 {
            return cancel();
        }
        let id = self.next_id.fetch_add(1, Ordering::SeqCst);
        let request = FrontendRequest {
            id,
            kind: FrontendRequestKind::Elicitation,
            payload: json!({
                "message": req.message,
                "requested_schema": req.requested_schema,
            }),
        };
        let (tx, rx) = std::sync::mpsc::channel();
        self.pending
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .insert(id, (FrontendRequestKind::Elicitation, tx));
        if !self.publish(&request) {
            self.pending
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .remove(&id);
            return cancel();
        }
        let broker = self.clone();
        tokio::task::spawn_blocking(move || loop {
            match rx.recv_timeout(std::time::Duration::from_millis(100)) {
                Ok(accepted) => {
                    let _ = accepted.published.recv();
                    let FrontendResponse::Elicitation {
                        action, content, ..
                    } = accepted.response
                    else {
                        return cancel();
                    };
                    return ElicitationResponse {
                        action: match action {
                            crate::frontend::FrontendElicitationAction::Accept => {
                                ElicitationAction::Accept
                            }
                            crate::frontend::FrontendElicitationAction::Decline => {
                                ElicitationAction::Decline
                            }
                            crate::frontend::FrontendElicitationAction::Cancel => {
                                ElicitationAction::Cancel
                            }
                        },
                        content,
                    };
                }
                Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => return cancel(),
                Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
                    if broker
                        .transport()
                        .map(|transport| transport.events.receiver_count() == 0)
                        .unwrap_or(true)
                    {
                        broker
                            .pending
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner)
                            .remove(&id);
                        return cancel();
                    }
                }
            }
        })
        .await
        .unwrap_or_else(|_| cancel())
    }
}

struct FrontendApprovalHandler(Arc<FrontendRequestBroker>);

impl PermissionsApprovalHandler for FrontendApprovalHandler {
    fn ask(&self, req: &ApprovalRequest<'_>) -> ApprovalOutcome {
        self.0.ask_approval(req, None)
    }
}

struct FrontendChildApprovalHandler {
    broker: Arc<FrontendRequestBroker>,
    child_agent_id: String,
    queue: Arc<StdMutex<Vec<crate::subagents::QueuedApproval>>>,
}

impl PermissionsApprovalHandler for FrontendChildApprovalHandler {
    fn ask(&self, req: &ApprovalRequest<'_>) -> ApprovalOutcome {
        self.broker
            .ask_approval(req, Some((&self.child_agent_id, &self.queue)))
    }
}

/// Pre-runtime bridge for MCP clients that must receive their elicitation
/// handler before they are consumed into agent tool registration.
#[derive(Clone)]
pub struct FrontendRequestBridge {
    broker: Arc<FrontendRequestBroker>,
}

impl FrontendRequestBridge {
    /// Create an unbound bridge. Pass it to
    /// [`RpcEngine::new_named_with_frontend_bridge`] after MCP registration.
    pub fn new() -> Self {
        Self {
            broker: FrontendRequestBroker::new(),
        }
    }

    /// Handler installed on every interactive MCP client before registration.
    pub fn elicitation_handler(&self) -> Arc<dyn McpElicitationHandler> {
        Arc::new(FrontendElicitationHandler(self.broker.clone()))
    }
}

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

struct FrontendElicitationHandler(Arc<FrontendRequestBroker>);

#[async_trait]
impl McpElicitationHandler for FrontendElicitationHandler {
    async fn handle(&self, request: &ElicitationRequest) -> ElicitationResponse {
        self.0.clone().ask_elicitation(request).await
    }
}

/// The out-of-process RPC driver: wraps one already-constructed
/// [`crate::Agent`] with the `submit`/`interrupt`/`status`/`shutdown`
/// method set (§ module doc). Shared by both transports ([`run_stdio`],
/// [`run_http`]) so the method semantics — including the fail-closed
/// permission behavior — can never drift between them.
pub struct RpcEngine {
    agent: Mutex<SdkAgent>,
    /// Last canonical transcript observed at a turn boundary. Frontends must
    /// be able to attach and replay prior history while `submit` holds the
    /// agent lock for an active turn, so reads use this independent snapshot.
    history_snapshot: RwLock<Vec<ChatMessage>>,
    session_id: String,
    /// The model label, captured once at construction so `status` never
    /// needs to lock `agent` (which `submit` holds for the WHOLE turn) —
    /// `status` must stay answerable while a turn is in flight.
    model: String,
    busy: Arc<AtomicBool>,
    /// The in-flight turn's cancellation handle, if any — see
    /// [`Self::handle_submit`]/[`Self::handle_interrupt`]'s doc comments
    /// for why this is a fresh [`Notify`] per turn rather than one shared
    /// instance (`notify_one`'s stored-permit semantics only give the
    /// correctness guarantee this needs when each turn gets a clean one).
    current_cancel: Arc<StdMutex<Option<Arc<Notify>>>>,
    /// Signals that the terminal event for an interrupted/completed turn is
    /// published and its canonical history snapshot is stable.
    turn_finished: Arc<Notify>,
    /// Shared control queue owned by `Agent` but writable without waiting for
    /// the active turn's long-held async agent lock.
    steer_queue: Arc<StdMutex<SteerInbox>>,
    events: broadcast::Sender<Value>,
    /// Sequenced frontend events used for atomic replay/live attachment.
    frontend_events: broadcast::Sender<FrontendEvent>,
    /// Short-held projection lock. It is never held across model/tool I/O.
    frontend_state: Arc<StdMutex<FrontendProjectionState>>,
    frontend_metadata: FrontendRuntimeMetadata,
    frontend_active_modules: Vec<String>,
    frontend_commands: Vec<FrontendCommandDescriptor>,
    frontend_operations: Vec<FrontendOperationDescriptor>,
    frontend_requests: Option<Arc<FrontendRequestBroker>>,
    shutdown: Notify,
    shutting_down: AtomicBool,
    /// Explicit owner shutdown seals new model-loop claims before it
    /// interrupts the active one. Plain transport EOF only raises
    /// `shutting_down` so already-buffered stdio requests can still flush.
    accepting_submits: AtomicBool,
    /// Serializes explicit shutdown barriers so every concurrent caller
    /// returns only after the same admitted turn and scheduler task drain.
    shutdown_barrier: Mutex<()>,
    scheduler_started: AtomicBool,
    /// The scheduler is runtime-owned work, not a detached best-effort task.
    /// Explicit shutdown takes and joins this handle before finalization may
    /// inspect or persist the agent again.
    scheduler_task: StdMutex<Option<tokio::task::JoinHandle<()>>>,
    scheduler_changed: Arc<Notify>,
    /// Fires after every SUCCESSFUL `submit` (never on an errored/
    /// interrupted turn — see the call site), with the agent still locked
    /// so the hook sees fully-consistent state (e.g. `agent.history()`).
    /// The CLI installs session persistence/auto-titling here — this
    /// module itself has no opinion on session storage.
    on_turn_complete: Option<TurnCompleteHook>,
}

/// Owns every externally visible piece of an SDK submit claim from the
/// instant the claim succeeds until the submit reaches a terminal boundary.
/// The agent loop closes the ordinary final-answer steering boundary
/// atomically with its last drain; this outer guard also restores steering,
/// cancellation, busy state, and lifecycle waiters when the public submit
/// future is dropped or fails before `Agent::run_loop` is entered.
struct SdkSubmitClaim {
    inbox: Arc<StdMutex<SteerInbox>>,
    busy: Arc<AtomicBool>,
    cancel: Arc<Notify>,
    current_cancel: Arc<StdMutex<Option<Arc<Notify>>>>,
    turn_finished: Arc<Notify>,
    scheduler_changed: Arc<Notify>,
    frontend_events: broadcast::Sender<FrontendEvent>,
    frontend_state: Arc<StdMutex<FrontendProjectionState>>,
    lifecycle_started: bool,
}

impl SdkSubmitClaim {
    fn mark_lifecycle_started(&mut self) {
        self.lifecycle_started = true;
    }

    fn mark_lifecycle_finished(&mut self) {
        self.lifecycle_started = false;
    }
}

impl Drop for SdkSubmitClaim {
    fn drop(&mut self) {
        // A caller may cancel the public `submit` future after the SDK has
        // exposed `turn_started` but before `submit_claimed` can publish its
        // ordinary terminal event. Close that exact lifecycle before making
        // the runtime idle so replay and live frontends cannot remain busy on
        // an abandoned turn. There is no await between the normal terminal
        // publication and disarming this fallback, so exactly one terminal
        // event is observable for every started claim.
        if self.lifecycle_started {
            let event = {
                let mut state = self
                    .frontend_state
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                let event = FrontendEvent::new(
                    state.next_sequence,
                    json!({
                        "type": "turn_interrupted",
                        "schema_version": FRONTEND_EVENT_SCHEMA_VERSION
                    }),
                );
                state.next_sequence = state.next_sequence.saturating_add(1);
                state.replay.push_back(event.clone());
                while state.replay.len() > FRONTEND_REPLAY_CAPACITY {
                    state.replay.pop_front();
                }
                event
            };
            let _ = self.frontend_events.send(event);
        }
        self.inbox
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .close();
        *self
            .current_cancel
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = None;
        self.busy.store(false, Ordering::SeqCst);
        self.turn_finished.notify_waiters();
        self.turn_finished.notify_one();
        self.scheduler_changed.notify_waiters();
    }
}

impl RpcEngine {
    /// Wrap `agent` (already fully built by the caller — same `Config`,
    /// same permission/sandbox posture as a local session) for out-of-
    /// process driving. Installs its OWN event sink via
    /// `Agent::set_event_sink`, overwriting whatever sink `agent` may
    /// already have had wired (callers of this module drive an agent
    /// exclusively through the RPC surface, so there is never a second,
    /// competing consumer of its events).
    pub fn new(
        agent: impl Into<SdkAgent>,
        on_turn_complete: Option<TurnCompleteHook>,
    ) -> Arc<Self> {
        let agent = agent.into();
        let session_id = agent
            .session_name()
            .map(str::to_owned)
            .unwrap_or_else(|| format!("supercode-{}", std::process::id()));
        Self::new_named(agent, session_id, on_turn_complete)
    }

    /// Construct the canonical SDK runtime with an explicit durable session
    /// identity.  Every frontend must use this identity when referring to the
    /// same live agent; transport-local connection ids are not session ids.
    pub fn new_named(
        agent: impl Into<SdkAgent>,
        session_id: impl Into<String>,
        on_turn_complete: Option<TurnCompleteHook>,
    ) -> Arc<Self> {
        Self::new_named_with_frontend_metadata(
            agent.into(),
            session_id,
            FrontendRuntimeMetadata::default(),
            on_turn_complete,
        )
    }

    /// Construct the canonical runtime with explicit source-harness and
    /// emulation-profile identity for every attached frontend.
    pub fn new_named_with_frontend_metadata(
        agent: impl Into<SdkAgent>,
        session_id: impl Into<String>,
        frontend_metadata: FrontendRuntimeMetadata,
        on_turn_complete: Option<TurnCompleteHook>,
    ) -> Arc<Self> {
        Self::build(
            agent.into(),
            session_id.into(),
            frontend_metadata,
            None,
            on_turn_complete,
        )
    }

    /// Construct a canonical runtime whose attached frontends may answer
    /// policy-authorized approval requests. Existing constructors retain the
    /// historical fail-closed headless behavior and report `respond=false`.
    pub fn new_named_with_frontend_requests(
        agent: impl Into<SdkAgent>,
        session_id: impl Into<String>,
        frontend_metadata: FrontendRuntimeMetadata,
        on_turn_complete: Option<TurnCompleteHook>,
    ) -> Arc<Self> {
        let bridge = FrontendRequestBridge::new();
        Self::new_named_with_frontend_bridge(
            agent.into(),
            session_id,
            frontend_metadata,
            bridge,
            on_turn_complete,
        )
    }

    /// Bind a pre-created request bridge after its elicitation handler has
    /// been installed on MCP clients.
    pub fn new_named_with_frontend_bridge(
        agent: impl Into<SdkAgent>,
        session_id: impl Into<String>,
        frontend_metadata: FrontendRuntimeMetadata,
        bridge: FrontendRequestBridge,
        on_turn_complete: Option<TurnCompleteHook>,
    ) -> Arc<Self> {
        Self::build(
            agent.into(),
            session_id.into(),
            frontend_metadata,
            Some(bridge.broker),
            on_turn_complete,
        )
    }

    fn build(
        mut agent: SdkAgent,
        session_id: String,
        frontend_metadata: FrontendRuntimeMetadata,
        frontend_requests: Option<Arc<FrontendRequestBroker>>,
        on_turn_complete: Option<TurnCompleteHook>,
    ) -> Arc<Self> {
        let (tx, _rx) = broadcast::channel(SERVER_EVENT_CHANNEL_CAPACITY);
        let events_tx = tx.clone();
        let (frontend_tx, _frontend_rx) = broadcast::channel(SERVER_EVENT_CHANNEL_CAPACITY);
        let frontend_events_tx = frontend_tx.clone();
        let model = agent.config().model.clone();
        let steer_queue = agent.inner().steer_queue_handle();
        let history_snapshot = bounded_history_snapshot(agent.history());
        let frontend_state = Arc::new(StdMutex::new(FrontendProjectionState {
            history: history_snapshot.clone(),
            history_cursor: 0,
            next_sequence: 1,
            replay: VecDeque::new(),
        }));
        if let Some(broker) = &frontend_requests {
            broker.bind(frontend_tx.clone(), frontend_state.clone());
            let legacy_broker = broker.clone();
            agent
                .inner_mut()
                .set_legacy_approval_handler(Box::new(move |call| {
                    let Ok(raw_args) = call.function.parsed_arguments() else {
                        return false;
                    };
                    let subject = raw_args
                        .get("command")
                        .or_else(|| raw_args.get("path"))
                        .or_else(|| raw_args.get("file_path"))
                        .or_else(|| raw_args.get("patch"))
                        .and_then(Value::as_str);
                    matches!(
                        legacy_broker.ask_approval(
                            &ApprovalRequest {
                                tool: &call.function.name,
                                subject,
                                raw_args: &raw_args,
                            },
                            None,
                        ),
                        ApprovalOutcome::Allow | ApprovalOutcome::AllowForSession
                    )
                }));
            agent
                .inner_mut()
                .set_permissions_approval_handler(FrontendApprovalHandler(broker.clone()));
            let broker = broker.clone();
            agent
                .inner_mut()
                .set_child_approval_handler_factory(move |child_agent_id, queue| {
                    Arc::new(FrontendChildApprovalHandler {
                        broker: broker.clone(),
                        child_agent_id,
                        queue,
                    }) as Arc<dyn PermissionsApprovalHandler>
                });
        }
        let event_frontend_state = frontend_state.clone();
        let frontend_active_modules = agent
            .config()
            .module_activation
            .iter()
            .map(ToString::to_string)
            .collect();
        let mut frontend_operations = agent
            .config()
            .prompts
            .keys()
            .filter(|name| valid_frontend_command_name(name))
            .map(|name| FrontendOperationDescriptor {
                id: format!("prompt:{name}"),
                kind: FrontendOperationKind::Prompt,
                command: Some(FrontendCommandDescriptor {
                    name: name.clone(),
                    description: None,
                    argument_hint: Some("[arguments]".into()),
                }),
            })
            .collect::<Vec<_>>();
        frontend_operations.sort_by(|left, right| left.id.cmp(&right.id));
        // Schema-v1 compatibility projection. New frontends use only the
        // typed operation catalog and never fall back to this list.
        let frontend_commands = frontend_operations
            .iter()
            .filter_map(|operation| operation.command.as_ref())
            .map(|command| FrontendCommandDescriptor {
                name: command.name.clone(),
                description: command.description.clone(),
                argument_hint: None,
            })
            .collect();
        agent.inner_mut().set_event_sink(Box::new(move |event| {
            // A `send` error here only means "no subscriber is currently
            // listening" (every receiver dropped) — never a reason to fail
            // the turn itself, so it's intentionally discarded.
            let payload = event.to_json();
            let _ = events_tx.send(payload.clone());
            let sequenced = {
                let mut state = event_frontend_state
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                let event = FrontendEvent::new(state.next_sequence, payload);
                state.next_sequence = state.next_sequence.saturating_add(1);
                state.replay.push_back(event.clone());
                while state.replay.len() > FRONTEND_REPLAY_CAPACITY {
                    state.replay.pop_front();
                }
                event
            };
            let _ = frontend_events_tx.send(sequenced);
        }));
        Arc::new(RpcEngine {
            agent: Mutex::new(agent),
            history_snapshot: RwLock::new(history_snapshot),
            session_id,
            model,
            busy: Arc::new(AtomicBool::new(false)),
            current_cancel: Arc::new(StdMutex::new(None)),
            turn_finished: Arc::new(Notify::new()),
            steer_queue,
            events: tx,
            frontend_events: frontend_tx,
            frontend_state,
            frontend_metadata,
            frontend_active_modules,
            frontend_commands,
            frontend_operations,
            frontend_requests,
            shutdown: Notify::new(),
            shutting_down: AtomicBool::new(false),
            accepting_submits: AtomicBool::new(true),
            shutdown_barrier: Mutex::new(()),
            scheduler_started: AtomicBool::new(false),
            scheduler_task: StdMutex::new(None),
            scheduler_changed: Arc::new(Notify::new()),
            on_turn_complete,
        })
    }

    /// Subscribe to this engine's event-notification stream (already
    /// `AgentEvent::to_json`-projected) — each subscriber gets every event
    /// emitted from this point on, independent of any other subscriber.
    pub fn subscribe(&self) -> broadcast::Receiver<Value> {
        self.events.subscribe()
    }

    /// Describe the SDK-owned runtime without locking the active agent turn.
    pub fn frontend_descriptor(&self) -> FrontendRuntimeDescriptor {
        FrontendRuntimeDescriptor {
            schema_version: FRONTEND_RUNTIME_SCHEMA_VERSION,
            session_id: self.session_id.clone(),
            source_harness: self.frontend_metadata.source_harness.clone(),
            emulation_profile: self.frontend_metadata.emulation_profile.clone(),
            active_modules: self.frontend_active_modules.clone(),
            commands: self.frontend_commands.clone(),
            operations: self.frontend_operations.clone(),
            actions: FrontendActions {
                submit: true,
                interrupt: true,
                steer: true,
                respond: self.frontend_requests.is_some(),
                detach: true,
                // The canonical runtime supports close. Coordinated client
                // projections mask this unless the authenticated grant owns
                // the independent terminate capability.
                close: true,
            },
            display: FrontendDisplayCapabilities {
                event_kinds: vec![
                    "user_message".into(),
                    "turn_started".into(),
                    "turn_succeeded".into(),
                    "turn_interrupted".into(),
                    "turn_failed".into(),
                    "text_delta".into(),
                    "turn_completed".into(),
                    "tool_call_started".into(),
                    "tool_call_completed".into(),
                    "cache_warning".into(),
                    "usage".into(),
                    "background_output".into(),
                    "request".into(),
                    "request_resolved".into(),
                    "scheduled_prompt_started".into(),
                    "scheduled_prompt_deferred".into(),
                    "scheduled_prompt_completed".into(),
                    "scheduler_error".into(),
                ],
                opaque_fallback: true,
            },
            model: self.model.clone(),
            turn_state: if self.busy.load(Ordering::SeqCst) {
                FrontendTurnState::Busy
            } else {
                FrontendTurnState::Idle
            },
            connection_state: if self.is_shutting_down() {
                FrontendConnectionState::ShuttingDown
            } else {
                FrontendConnectionState::Connected
            },
            extensions: Default::default(),
        }
    }

    /// Attach to one atomic history/replay/live boundary. The live receiver
    /// is created before the projection snapshot is locked; events racing the
    /// snapshot therefore appear either in replay or in the receiver, and
    /// [`FrontendAttachment::next_event`] removes any overlap by sequence.
    pub fn frontend_attach(
        &self,
        history_limit: usize,
    ) -> Result<FrontendAttachment, FrontendRuntimeError> {
        let live = self.frontend_subscribe();
        let snapshot = self.frontend_snapshot(history_limit)?;
        Ok(FrontendAttachment::new(
            snapshot.descriptor,
            snapshot.history,
            snapshot.history_cursor,
            snapshot.replay,
            live,
            None,
        ))
    }

    /// Subscribe to sequenced frontend events. Transport adapters subscribe
    /// before taking [`Self::frontend_snapshot`] so boundary events cannot be
    /// missed.
    pub fn frontend_subscribe(&self) -> broadcast::Receiver<FrontendEvent> {
        self.frontend_events.subscribe()
    }

    /// Capture the serializable history/replay half of a frontend attachment.
    pub fn frontend_snapshot(
        &self,
        history_limit: usize,
    ) -> Result<FrontendAttachSnapshot, FrontendRuntimeError> {
        let state = self
            .frontend_state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let limit = history_limit.min(SERVER_HISTORY_CAPACITY);
        let start = state.history.len().saturating_sub(limit);
        let replay = state
            .replay
            .iter()
            .filter(|event| event.sequence > state.history_cursor)
            .cloned()
            .collect::<VecDeque<_>>();
        if let Some(first) = replay.front() {
            let expected = state.history_cursor.saturating_add(1);
            if first.sequence > expected {
                return Err(FrontendRuntimeError::ReplayGap(first.sequence - expected));
            }
        }
        Ok(FrontendAttachSnapshot {
            descriptor: self.frontend_descriptor(),
            history: state.history[start..].to_vec(),
            history_cursor: state.history_cursor,
            replay,
        })
    }

    fn publish_frontend_payload(&self, payload: Value) {
        let event = {
            let mut state = self
                .frontend_state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            let event = FrontendEvent::new(state.next_sequence, payload);
            state.next_sequence = state.next_sequence.saturating_add(1);
            state.replay.push_back(event.clone());
            while state.replay.len() > FRONTEND_REPLAY_CAPACITY {
                state.replay.pop_front();
            }
            event
        };
        let _ = self.frontend_events.send(event);
    }

    /// Publish an SDK-owned runtime event to both the legacy RPC stream and
    /// the canonical sequenced frontend stream. Agent-originated events are
    /// bridged in `build`; scheduler-originated events use this helper so
    /// attached local and HTTP frontends observe the same lifecycle.
    fn publish_runtime_payload(&self, payload: Value) {
        let _ = self.events.send(payload.clone());
        self.publish_frontend_payload(payload);
    }

    /// Stable SDK session identity shared by every frontend.
    pub fn session_id(&self) -> &str {
        &self.session_id
    }

    fn claim_submit(&self) -> Result<SdkSubmitClaim, RuntimeSubmitError> {
        // Hold the cancellation slot across admission. Shutdown seals first,
        // then takes this same lock through `interrupt`: it therefore either
        // wins before `busy` is claimed or observes the admitted turn's
        // installed token. There is no busy-without-cancel interval.
        let mut current_cancel = self
            .current_cancel
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if !self.accepting_submits.load(Ordering::SeqCst) {
            return Err(RuntimeSubmitError::Interrupted);
        }
        if self.busy.swap(true, Ordering::SeqCst) {
            return Err(RuntimeSubmitError::Busy);
        }
        // Close the race with a concurrent shutdown between the first state
        // check and ownership of `busy`; relinquish this claim without
        // exposing a turn when the shutdown seal won.
        if !self.accepting_submits.load(Ordering::SeqCst) {
            self.busy.store(false, Ordering::SeqCst);
            self.turn_finished.notify_waiters();
            return Err(RuntimeSubmitError::Interrupted);
        }
        let cancel = Arc::new(Notify::new());
        *current_cancel = Some(cancel.clone());
        drop(current_cancel);
        self.steer_queue
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .open();
        Ok(SdkSubmitClaim {
            inbox: self.steer_queue.clone(),
            busy: self.busy.clone(),
            cancel,
            current_cancel: self.current_cancel.clone(),
            turn_finished: self.turn_finished.clone(),
            scheduler_changed: self.scheduler_changed.clone(),
            frontend_events: self.frontend_events.clone(),
            frontend_state: self.frontend_state.clone(),
            lifecycle_started: false,
        })
    }

    async fn submit_claimed(
        &self,
        prompt: String,
        image_urls: Vec<String>,
        mut submit_claim: SdkSubmitClaim,
    ) -> Result<String, RuntimeSubmitError> {
        let cancel = submit_claim.cancel.clone();
        self.publish_frontend_payload(json!({"type": "user_message", "text": &prompt}));
        self.publish_frontend_payload(json!({
            "type": "turn_started",
            "schema_version": FRONTEND_EVENT_SCHEMA_VERSION
        }));
        submit_claim.mark_lifecycle_started();
        let outcome = {
            let mut agent = self.agent.lock().await;
            let result = tokio::select! {
                biased;
                _ = cancel.notified() => Err(RuntimeSubmitError::Interrupted),
                result = async {
                    if image_urls.is_empty() {
                        agent.inner_mut().send(&prompt).await
                    } else {
                        agent.inner_mut().send_with_images(&prompt, &image_urls).await
                    }
                } => result.map_err(|error| RuntimeSubmitError::Agent(error.to_string())),
            };
            if result.is_ok() {
                if let Some(hook) = &self.on_turn_complete {
                    hook(&agent);
                }
            }
            // `Agent::send` is cancellation-safe at await boundaries. Publish
            // its latest well-formed history on success, failure, or
            // interruption without making attach readers wait on `agent`.
            let history = bounded_history_snapshot(agent.history());
            *self.history_snapshot.write().await = history.clone();
            let mut state = self
                .frontend_state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            state.history = history;
            state.history_cursor = state.next_sequence.saturating_sub(1);
            // Canonical ChatMessage history represents user/model/tool
            // content, but not interactive frontend requests or the human's
            // typed decision. Re-sequence those semantic events immediately
            // after the history boundary so later attachments retain a
            // resolved transcript without replaying ordinary turn events
            // already represented by `history`.
            let request_history = compact_frontend_request_history(&state.replay);
            state.replay.clear();
            for payload in request_history {
                let event = FrontendEvent::new(state.next_sequence, payload);
                state.next_sequence = state.next_sequence.saturating_add(1);
                state.replay.push_back(event);
                while state.replay.len() > FRONTEND_REPLAY_CAPACITY {
                    state.replay.pop_front();
                }
            }
            result
        };
        *self
            .current_cancel
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = None;
        let lifecycle = match &outcome {
            Ok(reply) => json!({
                "type": "turn_succeeded",
                "schema_version": FRONTEND_EVENT_SCHEMA_VERSION,
                "reply": reply,
            }),
            Err(RuntimeSubmitError::Interrupted) => json!({
                "type": "turn_interrupted",
                "schema_version": FRONTEND_EVENT_SCHEMA_VERSION
            }),
            Err(error) => json!({
                "type": "turn_failed",
                "schema_version": FRONTEND_EVENT_SCHEMA_VERSION,
                "message": error.to_string()
            }),
        };
        self.publish_frontend_payload(lifecycle);
        submit_claim.mark_lifecycle_finished();
        // Close before publishing idle. This is redundant with the agent's
        // normal final-boundary close, but is authoritative for pre-loop
        // validation/record failures and biased immediate interruption.
        drop(submit_claim);
        outcome
    }

    /// Submit one prompt through the canonical runtime and wait for its reply.
    pub async fn submit(&self, prompt: impl Into<String>) -> Result<String, RuntimeSubmitError> {
        let submit_claim = self.claim_submit()?;
        self.submit_claimed(prompt.into(), Vec::new(), submit_claim)
            .await
    }

    /// Submit one prompt with runtime-owned multimodal image inputs.
    pub async fn submit_with_images(
        &self,
        prompt: impl Into<String>,
        image_urls: Vec<String>,
    ) -> Result<String, RuntimeSubmitError> {
        let submit_claim = self.claim_submit()?;
        self.submit_claimed(prompt.into(), image_urls, submit_claim)
            .await
    }

    /// Atomically claim one prompt, then run it on the SDK owner while the
    /// caller consumes the canonical event stream.
    pub fn send_input(self: &Arc<Self>, prompt: String) -> Result<(), RuntimeSubmitError> {
        let submit_claim = self.claim_submit()?;
        let runtime = self.clone();
        tokio::spawn(async move {
            let _ = runtime
                .submit_claimed(prompt, Vec::new(), submit_claim)
                .await;
        });
        Ok(())
    }

    /// Queue steering for the active agent loop without waiting for its
    /// long-held async lock. The agent consumes it at the next model-loop
    /// boundary according to the configured steering mode.
    pub fn steer(&self, prompt: impl Into<String>) -> Result<(), FrontendRuntimeError> {
        let accepted = self
            .steer_queue
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .enqueue(prompt.into());
        if accepted {
            Ok(())
        } else {
            Err(FrontendRuntimeError::UnsupportedAction("steer"))
        }
    }

    /// Resolve one pending interactive request exactly once.
    pub fn respond(&self, response: FrontendResponse) -> Result<(), FrontendRuntimeError> {
        self.frontend_requests
            .as_ref()
            .ok_or(FrontendRuntimeError::UnsupportedAction("respond"))?
            .respond(response)
    }

    /// Invoke one operation after resolving its opaque identifier solely
    /// against the trusted catalog captured at runtime construction.
    pub async fn invoke(
        &self,
        operation: FrontendOperationInvocation,
    ) -> Result<FrontendOperationResult, FrontendRuntimeError> {
        match operation {
            FrontendOperationInvocation::Prompt {
                operation_id,
                arguments,
            } => {
                let prompt_name = self
                    .frontend_operations
                    .iter()
                    .find(|descriptor| {
                        descriptor.id == operation_id
                            && descriptor.kind == FrontendOperationKind::Prompt
                    })
                    .and_then(|descriptor| descriptor.command.as_ref())
                    .map(|command| command.name.as_str())
                    .ok_or_else(|| {
                        FrontendRuntimeError::UnsupportedOperation(operation_id.clone())
                    })?;
                let prompt = if arguments.is_empty() {
                    format!("/{prompt_name}")
                } else {
                    format!("/{prompt_name} {arguments}")
                };
                let reply = self.submit(prompt).await?;
                Ok(FrontendOperationResult::Prompt { reply })
            }
        }
    }

    /// Cancel the current turn through the shared runtime handle.
    pub async fn interrupt(&self) -> bool {
        let cancel = self
            .current_cancel
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone();
        match cancel {
            Some(cancel) => {
                cancel.notify_one();
                true
            }
            None => false,
        }
    }

    /// Return a lock-free runtime snapshot, including while a turn is active.
    pub fn status(&self) -> RuntimeStatus {
        RuntimeStatus {
            session_id: self.session_id.clone(),
            model: self.model.clone(),
            busy: self.busy.load(Ordering::SeqCst),
            shutting_down: self.is_shutting_down(),
        }
    }

    /// Return the tail of the canonical conversation for newly attached
    /// frontends. This is SDK state, not a transport-local replay buffer, and
    /// remains answerable while a turn owns the agent lock.
    pub async fn history(&self, limit: usize) -> Vec<ChatMessage> {
        let history = self.history_snapshot.read().await;
        let start = history.len().saturating_sub(limit);
        history[start..].to_vec()
    }

    /// Run one caller-owned finalization projection while holding the agent
    /// at a quiescent boundary. This is the persistence/inspection seam for
    /// local frontends that transfer `Agent` ownership into the SDK runtime;
    /// it does not expose a second way to drive the model loop.
    pub async fn finalize_with<R>(&self, finalize: impl FnOnce(&SdkAgent) -> R) -> R {
        let agent = self.agent.lock().await;
        finalize(&agent)
    }

    /// Request graceful runtime shutdown.  All connected frontends observe
    /// the same transition and any in-flight turn is interrupted.
    pub async fn shutdown(&self) {
        let _barrier = self.shutdown_barrier.lock().await;
        // Seal first so neither a frontend nor the scheduler can claim
        // replacement work while the active claim is interrupted. Signal
        // transports only after its terminal lifecycle is published, so SSE
        // observers receive that boundary before their connections close.
        self.accepting_submits.store(false, Ordering::SeqCst);
        self.interrupt().await;
        loop {
            let finished = self.turn_finished.notified();
            if !self.busy.load(Ordering::SeqCst) {
                break;
            }
            finished.await;
        }
        // Wake a scheduler that is between due-time checks without closing
        // frontend transports yet. `notify_one` stores a permit when it has
        // not reached its select, so the join cannot lose this wakeup.
        self.scheduler_changed.notify_one();
        let scheduler = self
            .scheduler_task
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .take();
        if let Some(scheduler) = scheduler {
            let _ = scheduler.await;
        }
        self.signal_shutdown();
    }

    /// Start the SDK-owned Claude runtime scheduler once.
    ///
    /// The scheduler is deliberately owned by the same runtime as user turns,
    /// not by a CLI/TUI input loop. Detaching every frontend therefore does
    /// not pause queued prompts, cron jobs, or wakeups while the runtime
    /// process remains alive.
    /// Returns `true` for the caller that started it and `false` thereafter.
    pub fn start_claude_scheduler(self: &Arc<Self>) -> bool {
        let mut task = self
            .scheduler_task
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if !self.accepting_submits.load(Ordering::SeqCst)
            || self.is_shutting_down()
            || self.scheduler_started.swap(true, Ordering::SeqCst)
        {
            return false;
        }
        let runtime = self.clone();
        *task = Some(tokio::spawn(
            async move { runtime.run_claude_scheduler().await },
        ));
        true
    }

    /// Whether this runtime has armed its scheduler driver.
    pub fn claude_scheduler_started(&self) -> bool {
        self.scheduler_started.load(Ordering::SeqCst)
    }

    async fn run_claude_scheduler(self: Arc<Self>) {
        use crate::claude_runtime_scheduler::ClaudeRuntimeTrigger;

        loop {
            if !self.accepting_submits.load(Ordering::SeqCst) || self.is_shutting_down() {
                return;
            }
            let now = unix_seconds();
            let next = {
                let agent = self.agent.lock().await;
                agent
                    .claude_runtime_manifest()
                    .and_then(|manifest| manifest.next_due(now).ok().flatten())
            };
            let Some(next) = next else {
                tokio::select! {
                    _ = self.wait_for_shutdown() => return,
                    _ = self.scheduler_changed.notified() => continue,
                }
            };
            let delay =
                std::time::Duration::from_secs(next.due_unix.saturating_sub(now).max(0) as u64);
            if !delay.is_zero() {
                tokio::select! {
                    _ = self.wait_for_shutdown() => return,
                    _ = self.scheduler_changed.notified() => continue,
                    _ = tokio::time::sleep(delay) => {}
                }
            }

            let claimed: Vec<ClaudeRuntimeTrigger> = {
                let mut agent = self.agent.lock().await;
                let claimed = match agent.claude_runtime_manifest_mut() {
                    Some(manifest) => manifest.claim_due(unix_seconds()),
                    None => continue,
                };
                match claimed {
                    Ok(claimed) => {
                        if !claimed.is_empty() {
                            if let Some(hook) = &self.on_turn_complete {
                                // Persist claims BEFORE provider delivery so a
                                // crash retains them for retry.
                                hook(&agent);
                            }
                        }
                        claimed
                    }
                    Err(error) => {
                        self.publish_runtime_payload(json!({
                            "type": "scheduler_error",
                            "message": error.to_string(),
                            "terminal": false
                        }));
                        Vec::new()
                    }
                }
            };

            for trigger in claimed {
                if !self.accepting_submits.load(Ordering::SeqCst) || self.is_shutting_down() {
                    return;
                }
                self.deliver_scheduled_prompt(trigger).await;
            }
        }
    }

    async fn deliver_scheduled_prompt(
        &self,
        trigger: crate::claude_runtime_scheduler::ClaudeRuntimeTrigger,
    ) {
        let kind = trigger.kind;
        let id = trigger.id.clone();
        let prompt = trigger
            .prompt
            .unwrap_or_else(|| "Scheduled wakeup".to_string());
        self.publish_runtime_payload(json!({
            "type": "scheduled_prompt_started",
            "kind": kind,
            "id": id,
            "due_unix": trigger.due_unix
        }));
        let result = self.submit(prompt).await;
        let now = unix_seconds();
        let mut agent = self.agent.lock().await;
        let update = match (agent.claude_runtime_manifest_mut(), &result) {
            (Some(manifest), Ok(_)) => manifest.complete_delivery(kind, &id),
            (Some(manifest), Err(RuntimeSubmitError::Busy)) => {
                manifest.defer_delivery(kind, &id, now.saturating_add(1))
            }
            (Some(manifest), Err(_)) => manifest.defer_delivery(kind, &id, now.saturating_add(60)),
            (None, _) => return,
        };
        if let Err(error) = update {
            self.publish_runtime_payload(json!({
                "type": "scheduler_error",
                "message": error.to_string(),
                "terminal": false
            }));
            return;
        }
        if let Some(hook) = &self.on_turn_complete {
            hook(&agent);
        }
        self.publish_runtime_payload(match result {
            Ok(_) => json!({"type": "scheduled_prompt_completed", "kind": kind, "id": id}),
            Err(error) => json!({
                "type": "scheduled_prompt_deferred",
                "kind": kind,
                "id": id,
                "message": error.to_string()
            }),
        });
        self.scheduler_changed.notify_waiters();
    }

    /// Whether `shutdown` has been requested — callers use this to stop
    /// accepting new work/connections.
    pub fn is_shutting_down(&self) -> bool {
        self.shutting_down.load(Ordering::SeqCst)
    }

    /// Resolves once `shutdown` has been requested. Cheap to call
    /// repeatedly/concurrently — every waiter is woken.
    pub async fn wait_for_shutdown(&self) {
        // `shutting_down` may already be `true` by the time a caller starts
        // waiting (e.g. a connection accepted right after `shutdown` fired)
        // — check first so this never blocks forever on a signal that
        // already happened.
        if self.is_shutting_down() {
            return;
        }
        self.shutdown.notified().await;
    }

    /// Flip `shutting_down` and wake every [`Self::wait_for_shutdown`]
    /// waiter — factored out of [`Self::handle_shutdown`] so [`run_stdio`]
    /// can raise the EXACT same signal on its OTHER termination path
    /// (`reader` hitting EOF with no explicit `shutdown` RPC) without
    /// duplicating the store-then-notify sequence. Deliberately does NOT
    /// touch `current_cancel` (unlike [`Self::handle_shutdown`], which also
    /// interrupts an in-flight turn) — a plain stdio EOF should let an
    /// already-accepted turn run to completion and flush its reply, not cut
    /// it off.
    fn signal_shutdown(&self) {
        self.shutting_down.store(true, Ordering::SeqCst);
        self.shutdown.notify_waiters();
    }

    /// Dispatch one already-parsed [`RpcRequest`] to the right method
    /// handler. Every recognized method is fully wired to real `Agent`
    /// behavior — there is no method that parses but no-ops.
    pub async fn handle_request(self: &Arc<Self>, req: RpcRequest) -> Value {
        match req.method.as_str() {
            "submit" => self.handle_submit(req).await,
            "frontend.send_input" => self.handle_frontend_send_input(req),
            "interrupt" => self.handle_interrupt(req).await,
            "steer" => self.handle_steer(req),
            "respond" => self.handle_respond(req),
            "status" => self.handle_status(req).await,
            "history" => self.handle_history(req).await,
            "frontend.describe" => self.handle_frontend_describe(req),
            "frontend.attach" => self.handle_frontend_attach(req),
            "frontend.invoke" => self.handle_frontend_invoke(req).await,
            "shutdown" => self.handle_shutdown(req).await,
            other => rpc_error(req.id, -32601, format!("unknown method `{other}`")),
        }
    }

    fn handle_frontend_send_input(self: &Arc<Self>, req: RpcRequest) -> Value {
        let Some(prompt) = req.params.get("prompt").and_then(Value::as_str) else {
            return rpc_error(
                req.id,
                -32602,
                "frontend.send_input requires a string `params.prompt`",
            );
        };
        match self.send_input(prompt.to_string()) {
            Ok(()) => rpc_ok(req.id, json!({"accepted": true})),
            Err(RuntimeSubmitError::Busy) => {
                rpc_error(req.id, -32000, "a turn is already in progress")
            }
            Err(RuntimeSubmitError::Interrupted) => rpc_error(req.id, -32001, "turn interrupted"),
            Err(RuntimeSubmitError::Agent(error)) => rpc_error(req.id, -32002, error),
        }
    }

    /// `submit`: drive one turn (`Agent::send`) with `params.prompt`.
    /// Refuses (fail, never queues) a second `submit` while one is already
    /// in flight — "no method that no-ops": a caller either gets a real
    /// answer or an explicit "busy" error, never a silently-dropped
    /// request. Races the turn against this call's own fresh cancellation
    /// handle so a concurrent `interrupt` can drop it mid-flight — the
    /// SAME cancel-safety the CLI's own `race_ctrl_c` relies on
    /// (`Agent::send`/`run_loop` only ever mutate `history`/the sidecar
    /// BETWEEN `.await` points, never during one, so dropping the future
    /// mid-poll always lands in a well-formed place).
    async fn handle_submit(&self, req: RpcRequest) -> Value {
        let Some(prompt) = req.params.get("prompt").and_then(|v| v.as_str()) else {
            return rpc_error(req.id, -32602, "submit requires a string `params.prompt`");
        };
        match self.submit(prompt).await {
            Ok(reply) => rpc_ok(req.id, json!({"reply": reply})),
            Err(RuntimeSubmitError::Busy) => rpc_error(
                req.id,
                -32000,
                "a turn is already in progress; `interrupt` it or wait for its response before submitting another",
            ),
            Err(RuntimeSubmitError::Interrupted) => {
                rpc_error(req.id, -32001, "turn interrupted")
            }
            Err(RuntimeSubmitError::Agent(error)) => rpc_error(req.id, -32002, error),
        }
    }

    /// `interrupt`: cancel the in-flight turn, if any. A no-op-but-honest
    /// `{"interrupted": false}` (never an error) when nothing is running —
    /// calling `interrupt` with no turn in flight is a normal, harmless
    /// race a client can't always avoid (it may not know yet that the
    /// previous `submit` just finished).
    async fn handle_interrupt(&self, req: RpcRequest) -> Value {
        if self.interrupt().await {
            rpc_ok(req.id, json!({"interrupted": true}))
        } else {
            rpc_ok(
                req.id,
                json!({"interrupted": false, "reason": "no turn in progress"}),
            )
        }
    }

    /// `steer`: enqueue an instruction for the next model-loop boundary.
    /// This remains responsive while `submit` owns the active agent lock.
    fn handle_steer(&self, req: RpcRequest) -> Value {
        let Some(prompt) = req.params.get("prompt").and_then(Value::as_str) else {
            return rpc_error(req.id, -32602, "steer requires a string `params.prompt`");
        };
        match self.steer(prompt) {
            Ok(()) => rpc_ok(req.id, json!({"queued": true})),
            Err(error) => rpc_error(req.id, -32020, error.to_string()),
        }
    }

    fn handle_respond(&self, req: RpcRequest) -> Value {
        let response = match req.params.get("response").cloned() {
            Some(value) => match serde_json::from_value::<FrontendResponse>(value) {
                Ok(response) => response,
                Err(error) => return rpc_error(req.id, -32602, error.to_string()),
            },
            None => return rpc_error(req.id, -32602, "respond requires `params.response`"),
        };
        match self.respond(response) {
            Ok(()) => rpc_ok(req.id, json!({"accepted": true})),
            Err(FrontendRuntimeError::UnsupportedAction(_)) => {
                rpc_error(req.id, -32020, "frontend respond is not enabled")
            }
            Err(FrontendRuntimeError::UnknownRequest(id)) => rpc_error(
                req.id,
                -32021,
                format!("frontend request {id} is not pending"),
            ),
            Err(error) => rpc_error(req.id, -32022, error.to_string()),
        }
    }

    /// `status`: current busy/idle state + the model label. Deliberately
    /// never locks `agent` (see [`Self::model`]'s doc comment) — answerable
    /// even while a turn is running.
    async fn handle_status(&self, req: RpcRequest) -> Value {
        rpc_ok(
            req.id,
            serde_json::to_value(self.status()).unwrap_or_default(),
        )
    }

    /// `history`: bounded canonical transcript replay for a frontend that
    /// attached after earlier events were emitted.
    async fn handle_history(&self, req: RpcRequest) -> Value {
        let limit = req
            .params
            .get("limit")
            .and_then(Value::as_u64)
            .unwrap_or(50)
            .clamp(1, SERVER_HISTORY_CAPACITY as u64) as usize;
        rpc_ok(req.id, json!({"messages": self.history(limit).await}))
    }

    fn handle_frontend_describe(&self, req: RpcRequest) -> Value {
        rpc_ok(
            req.id,
            serde_json::to_value(self.frontend_descriptor()).unwrap_or_default(),
        )
    }

    fn handle_frontend_attach(&self, req: RpcRequest) -> Value {
        let limit = req
            .params
            .get("limit")
            .and_then(Value::as_u64)
            .unwrap_or(50)
            .clamp(1, SERVER_HISTORY_CAPACITY as u64) as usize;
        match self.frontend_snapshot(limit) {
            Ok(snapshot) => rpc_ok(req.id, serde_json::to_value(snapshot).unwrap_or_default()),
            Err(error) => rpc_error(req.id, -32010, error.to_string()),
        }
    }

    async fn handle_frontend_invoke(&self, req: RpcRequest) -> Value {
        let operation = match req.params.get("operation").cloned() {
            Some(value) => match serde_json::from_value::<FrontendOperationInvocation>(value) {
                Ok(operation) => operation,
                Err(error) => return rpc_error(req.id, -32602, error.to_string()),
            },
            None => {
                return rpc_error(
                    req.id,
                    -32602,
                    "frontend.invoke requires `params.operation`",
                )
            }
        };
        match self.invoke(operation).await {
            Ok(result) => rpc_ok(req.id, serde_json::to_value(result).unwrap_or_default()),
            Err(FrontendRuntimeError::UnsupportedOperation(id)) => rpc_error(
                req.id,
                -32023,
                FrontendRuntimeError::UnsupportedOperation(id).to_string(),
            ),
            Err(FrontendRuntimeError::Submit(RuntimeSubmitError::Busy)) => {
                rpc_error(req.id, -32000, "a turn is already in progress")
            }
            Err(error) => rpc_error(req.id, -32022, error.to_string()),
        }
    }

    /// `shutdown`: request graceful teardown — interrupts any in-flight
    /// turn (never leaves a caller hanging on a `submit` that will now
    /// never get a transport to answer on) and wakes every
    /// [`Self::wait_for_shutdown`] waiter (both transports' accept/read
    /// loops select on it, so neither survives as an orphan).
    async fn handle_shutdown(&self, req: RpcRequest) -> Value {
        self.shutdown().await;
        rpc_ok(req.id, json!({"shutting_down": true}))
    }
}

/// Retain one canonical request and at most one resolution for every runtime
/// request id. Request history is copied across ordinary ChatMessage snapshot
/// boundaries, so blindly copying the prior replay would duplicate it on
/// every turn and could eventually leave a later attachment on a stale
/// duplicate request. Stable id order and request-before-resolution ordering
/// make the reconstructed semantic transcript deterministic.
fn compact_frontend_request_history(replay: &VecDeque<FrontendEvent>) -> Vec<Value> {
    let mut by_id: BTreeMap<u64, (Option<Value>, Option<Value>)> = BTreeMap::new();
    for event in replay {
        let (request_id, resolved) = match event.kind.as_str() {
            "request" => (
                event.payload.pointer("/request/id").and_then(Value::as_u64),
                false,
            ),
            "request_resolved" => (
                event.payload.get("request_id").and_then(Value::as_u64),
                true,
            ),
            _ => continue,
        };
        let Some(request_id) = request_id else {
            continue;
        };
        let entry = by_id.entry(request_id).or_default();
        let slot = if resolved { &mut entry.1 } else { &mut entry.0 };
        slot.get_or_insert_with(|| event.payload.clone());
    }
    by_id
        .into_values()
        .flat_map(|(request, resolution)| request.into_iter().chain(resolution))
        .collect()
}

fn valid_frontend_command_name(name: &str) -> bool {
    !name.is_empty()
        && !name.starts_with('/')
        && name
            .chars()
            .all(|character| !character.is_whitespace() && !character.is_control())
}

fn bounded_history_snapshot(history: &[ChatMessage]) -> Vec<ChatMessage> {
    let start = history.len().saturating_sub(SERVER_HISTORY_CAPACITY);
    history[start..].to_vec()
}

fn unix_seconds() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|duration| duration.as_secs().min(i64::MAX as u64) as i64)
        .unwrap_or(0)
}

/// Drive the JSONL-RPC protocol over `reader`/`writer` (the stdio rung —
/// parent-process-trusted, no auth token; see the module doc). Each
/// request line is dispatched on its OWN spawned task so a `submit`
/// in-flight never blocks the reader from picking up a subsequent
/// `interrupt`/`status` line — every outgoing line (a response OR an event
/// notification) is funneled through one mpsc channel into a single writer
/// task, so two concurrent handlers can never interleave a line's bytes.
/// Returns once `reader` hits EOF or a `shutdown` request lands.
///
/// Both termination paths raise `RpcEngine::signal_shutdown` (EOF does it
/// directly here; the `shutdown` RPC does it inside
/// `RpcEngine::handle_shutdown`), and `event_task` below SELECTS against
/// [`RpcEngine::wait_for_shutdown`] rather than merely looping on
/// `events.recv()`. This is deliberate: `events.recv()` alone only ends via
/// `RecvError::Closed`, which fires only once EVERY clone of
/// `engine.events` (the broadcast `Sender`) has dropped — and `engine`
/// itself, which keeps that `Sender` alive, is owned by THIS function for
/// its whole body. Waiting on `events.recv()` to close would therefore mean
/// waiting on `engine` to drop, which can't happen until `event_task`
/// itself finishes — a circular wait that never resolves (the bug this fn
/// exists to fix). Selecting on the shutdown signal instead lets
/// `event_task` end WITHOUT needing `engine`'s refcount to reach zero, so
/// there is no orphaned task and no leaked `engine`/`writer_task` blocking
/// on it in turn.
#[cfg(feature = "adapter-api")]
pub async fn run_stdio<R, W>(engine: Arc<RpcEngine>, reader: R, writer: W) -> std::io::Result<()>
where
    R: AsyncBufRead + Unpin + Send + 'static,
    W: AsyncWrite + Unpin + Send + 'static,
{
    let (out_tx, mut out_rx) = mpsc::unbounded_channel::<Value>();

    let writer_task = tokio::spawn(async move {
        let mut writer = writer;
        while let Some(v) = out_rx.recv().await {
            let line = format!("{v}\n");
            if writer.write_all(line.as_bytes()).await.is_err() {
                break;
            }
            if writer.flush().await.is_err() {
                break;
            }
        }
    });

    let mut events = engine.subscribe();
    let evt_tx = out_tx.clone();
    let evt_engine = engine.clone();
    let event_task = tokio::spawn(async move {
        loop {
            tokio::select! {
                biased;
                _ = evt_engine.wait_for_shutdown() => break,
                recv = events.recv() => {
                    match recv {
                        Ok(v) => {
                            if evt_tx.send(json!({"event": v})).is_err() {
                                break;
                            }
                        }
                        Err(broadcast::error::RecvError::Lagged(_)) => continue,
                        Err(broadcast::error::RecvError::Closed) => break,
                    }
                }
            }
        }
    });

    let mut reader = reader;
    loop {
        if engine.is_shutting_down() {
            break;
        }
        tokio::select! {
            biased;
            _ = engine.wait_for_shutdown() => break,
            line = read_bounded_line(&mut reader, SERVER_MAX_LINE_BYTES) => {
                match line {
                    Ok(None) => {
                        // EOF: no explicit `shutdown` RPC landed, but stdin
                        // closing is this fn's OTHER documented
                        // termination signal — raise the same shutdown
                        // signal `event_task` (and any other
                        // `wait_for_shutdown` caller) already knows how to
                        // watch for, so teardown below actually completes
                        // instead of blocking forever on `event_task`.
                        engine.signal_shutdown();
                        break;
                    }
                    Ok(Some(text)) => {
                        let text = text.trim();
                        if text.is_empty() {
                            continue;
                        }
                        match serde_json::from_str::<RpcRequest>(text) {
                            Ok(req) => {
                                let engine = engine.clone();
                                let out_tx = out_tx.clone();
                                tokio::spawn(async move {
                                    let resp = engine.handle_request(req).await;
                                    let _ = out_tx.send(resp);
                                });
                            }
                            Err(e) => {
                                let _ = out_tx.send(rpc_error(Value::Null, -32700, format!("parse error: {e}")));
                            }
                        }
                    }
                    Err(e) => {
                        let _ = out_tx.send(rpc_error(Value::Null, -32700, format!("{e}")));
                    }
                }
            }
        }
    }
    // `event_task` now ends promptly (it's shutdown-signalled above, on
    // EITHER termination path) rather than waiting on `engine`'s broadcast
    // `Sender` to drop — so awaiting it here no longer deadlocks. Dropping
    // this fn's own `out_tx` clone (plus `event_task`'s, once it exits)
    // lets `writer_task` see `out_rx.recv()` return `None` once every
    // OTHER in-flight per-request task (spawned above, each holding its own
    // `out_tx` clone) has also sent its reply and dropped its clone — so
    // any reply already accepted before shutdown is still flushed to
    // `writer` before this fn returns.
    drop(out_tx);
    let _ = event_task.await;
    let _ = writer_task.await;
    Ok(())
}

/// One parsed HTTP/1.1 request (the minimal subset this module's two
/// routes need — no keep-alive, no chunked request bodies).
#[cfg(feature = "adapter-api")]
struct HttpRequest {
    method: String,
    /// Path WITHOUT the query string (see `query` for that).
    path: String,
    query: String,
    headers: HashMap<String, String>,
    body: Vec<u8>,
}

/// Read and parse one HTTP/1.1 request from `reader`. `Ok(None)` at a
/// clean EOF before any bytes arrive (an idle keep-alive-less connection
/// closing). Bounded throughout: the request line and each header line go
/// through [`read_bounded_line`] (8KiB — generous for a request
/// line/header, far below [`SERVER_MAX_LINE_BYTES`]), the header COUNT is
/// capped at [`MAX_HEADER_LINES`], and the body is capped at
/// [`SERVER_MAX_LINE_BYTES`].
#[cfg(feature = "adapter-api")]
async fn read_http_request<R>(reader: &mut R) -> std::io::Result<Option<HttpRequest>>
where
    R: AsyncBufRead + AsyncRead + Unpin,
{
    const HEAD_LINE_CAP: usize = 8 * 1024;
    let Some(request_line) = read_bounded_line(reader, HEAD_LINE_CAP).await? else {
        return Ok(None);
    };
    let mut parts = request_line.split_whitespace();
    let method = parts.next().unwrap_or("").to_string();
    let target = parts.next().unwrap_or("").to_string();
    if method.is_empty() || target.is_empty() {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "malformed request line",
        ));
    }
    let (path, query) = match target.split_once('?') {
        Some((p, q)) => (p.to_string(), q.to_string()),
        None => (target, String::new()),
    };

    let mut headers = HashMap::new();
    let mut content_length: usize = 0;
    for _ in 0..MAX_HEADER_LINES {
        let Some(line) = read_bounded_line(reader, HEAD_LINE_CAP).await? else {
            return Ok(None);
        };
        if line.is_empty() {
            break;
        }
        if let Some((k, v)) = line.split_once(':') {
            let k = k.trim().to_ascii_lowercase();
            let v = v.trim().to_string();
            if k == "content-length" {
                content_length = v.parse().unwrap_or(0);
            }
            headers.insert(k, v);
        }
    }
    if content_length > SERVER_MAX_LINE_BYTES {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("request body exceeded {SERVER_MAX_LINE_BYTES} byte cap"),
        ));
    }
    let mut body = vec![0u8; content_length];
    if content_length > 0 {
        reader.read_exact(&mut body).await?;
    }
    Ok(Some(HttpRequest {
        method,
        path,
        query,
        headers,
        body,
    }))
}

#[cfg(feature = "adapter-api")]
async fn write_http_response<W: AsyncWrite + Unpin>(
    writer: &mut W,
    status: u16,
    reason: &str,
    content_type: &str,
    body: &[u8],
) -> std::io::Result<()> {
    let head = format!(
        "HTTP/1.1 {status} {reason}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
        body.len()
    );
    writer.write_all(head.as_bytes()).await?;
    writer.write_all(body).await?;
    writer.flush().await
}

#[cfg(feature = "adapter-api")]
fn browser_observer_asset(path: &str) -> Option<(&'static str, &'static [u8])> {
    match path {
        "/observer" | "/observer/" => Some((
            "text/html; charset=utf-8",
            include_bytes!("../embedded/frontend-browser/index.html"),
        )),
        "/observer/app.mjs" => Some((
            "text/javascript; charset=utf-8",
            include_bytes!("../embedded/frontend-browser/app.mjs"),
        )),
        "/observer/client.mjs" => Some((
            "text/javascript; charset=utf-8",
            include_bytes!("../embedded/frontend-browser/client.mjs"),
        )),
        "/observer/view.mjs" => Some((
            "text/javascript; charset=utf-8",
            include_bytes!("../embedded/frontend-browser/view.mjs"),
        )),
        "/observer/style.css" => Some((
            "text/css; charset=utf-8",
            include_bytes!("../embedded/frontend-browser/style.css"),
        )),
        "/observer/favicon.svg" | "/favicon.ico" => Some((
            "image/svg+xml",
            include_bytes!("../embedded/frontend-browser/favicon.svg"),
        )),
        "/frontend/client.mjs" => Some((
            "text/javascript; charset=utf-8",
            include_bytes!("../embedded/frontend/client.mjs"),
        )),
        "/frontend/generated-client.mjs" => Some((
            "text/javascript; charset=utf-8",
            include_bytes!("../embedded/frontend/generated-client.mjs"),
        )),
        "/frontend/generated.mjs" => Some((
            "text/javascript; charset=utf-8",
            include_bytes!("../embedded/frontend/generated.mjs"),
        )),
        _ => None,
    }
}

#[cfg(feature = "adapter-api")]
async fn write_browser_observer_asset<W: AsyncWrite + Unpin>(
    writer: &mut W,
    content_type: &str,
    body: &[u8],
) -> std::io::Result<()> {
    let head = format!(
        "HTTP/1.1 200 OK\r\n\
         Content-Type: {content_type}\r\n\
         Content-Length: {}\r\n\
         Cache-Control: no-store\r\n\
         Content-Security-Policy: default-src 'none'; script-src 'self'; style-src 'self'; connect-src 'self'; img-src 'self'; base-uri 'none'; form-action 'self'; frame-ancestors 'none'\r\n\
         Referrer-Policy: no-referrer\r\n\
         X-Content-Type-Options: nosniff\r\n\
         Connection: close\r\n\r\n",
        body.len()
    );
    writer.write_all(head.as_bytes()).await?;
    writer.write_all(body).await?;
    writer.flush().await
}

/// One bearer credential and its exact SDK authorization grant.
#[cfg(feature = "adapter-api")]
#[derive(Clone)]
pub struct RuntimeHttpCredential {
    token: Arc<str>,
    authorization: RuntimeAuthorization,
    client_id: Option<crate::RuntimeClientId>,
    bootstrap: bool,
    runtime_id: Option<Arc<str>>,
    generation: Option<[u8; 16]>,
    revocation: Option<Arc<RuntimeCredentialRevocation>>,
}

#[cfg(feature = "adapter-api")]
impl RuntimeHttpCredential {
    /// Create a scoped credential. Tokens are deliberately private and never
    /// implement `Debug` or serialization.
    pub fn new(token: impl Into<Arc<str>>, authorization: RuntimeAuthorization) -> Self {
        Self {
            token: token.into(),
            authorization,
            client_id: None,
            bootstrap: false,
            runtime_id: None,
            generation: None,
            revocation: None,
        }
    }

    /// Full owner credential preserving the historical `run_http` contract.
    ///
    /// New frontend-host paths must keep this bootstrap credential private and
    /// exchange it through the local mint endpoint for a bound frontend grant.
    pub fn owner(token: impl Into<Arc<str>>) -> Self {
        Self {
            token: token.into(),
            authorization: RuntimeAuthorization::owner(),
            client_id: None,
            bootstrap: true,
            runtime_id: None,
            generation: None,
            revocation: None,
        }
    }

    /// Read-only observer credential.
    pub fn observer(token: impl Into<Arc<str>>) -> Self {
        Self::new(token, RuntimeAuthorization::observer())
    }

    fn frontend(
        token: impl Into<Arc<str>>,
        client_id: crate::RuntimeClientId,
        authorization: RuntimeAuthorization,
        runtime_id: impl Into<Arc<str>>,
        generation: [u8; 16],
    ) -> Self {
        Self {
            token: token.into(),
            authorization,
            client_id: Some(client_id),
            bootstrap: false,
            runtime_id: Some(runtime_id.into()),
            generation: Some(generation),
            revocation: Some(Arc::new(RuntimeCredentialRevocation::new())),
        }
    }
}

#[cfg(feature = "adapter-api")]
struct AuthenticatedRuntimeHttpCredential {
    authorization: RuntimeAuthorization,
    client_id: Option<crate::RuntimeClientId>,
    bootstrap: bool,
    revocation: Option<tokio::sync::watch::Receiver<bool>>,
    attachment: Option<RuntimeCredentialAttachment>,
    via_bearer_header: bool,
}

#[cfg(feature = "adapter-api")]
struct RuntimeCredentialRevocation {
    signal: tokio::sync::watch::Sender<bool>,
    active_attachments: AtomicUsize,
    drained: tokio::sync::Notify,
}

#[cfg(feature = "adapter-api")]
impl RuntimeCredentialRevocation {
    fn new() -> Self {
        let (signal, _) = tokio::sync::watch::channel(false);
        Self {
            signal,
            active_attachments: AtomicUsize::new(0),
            drained: tokio::sync::Notify::new(),
        }
    }

    fn register(self: &Arc<Self>) -> RuntimeCredentialAttachment {
        self.active_attachments.fetch_add(1, Ordering::AcqRel);
        RuntimeCredentialAttachment {
            revocation: self.clone(),
        }
    }

    async fn revoke_and_wait(&self) {
        let _ = self.signal.send(true);
        loop {
            let drained = self.drained.notified();
            if self.active_attachments.load(Ordering::Acquire) == 0 {
                return;
            }
            drained.await;
        }
    }
}

#[cfg(feature = "adapter-api")]
struct RuntimeCredentialAttachment {
    revocation: Arc<RuntimeCredentialRevocation>,
}

#[cfg(feature = "adapter-api")]
impl Drop for RuntimeCredentialAttachment {
    fn drop(&mut self) {
        if self
            .revocation
            .active_attachments
            .fetch_sub(1, Ordering::AcqRel)
            == 1
        {
            // Registry removal makes the first revoke the sole waiter for a
            // credential; concurrent revokes find no credential. `notify_one`
            // stores a permit if this lands between the counter check and the
            // first poll of `notified()`, preventing a lost wakeup.
            self.revocation.drained.notify_one();
        }
    }
}

#[cfg(feature = "adapter-api")]
struct IssuedRuntimeHttpCredential(Arc<str>);

#[cfg(feature = "adapter-api")]
impl IssuedRuntimeHttpCredential {
    fn as_bytes(&self) -> &[u8] {
        self.0.as_bytes()
    }
}

#[cfg(feature = "adapter-api")]
struct RuntimeHttpCredentialRegistry {
    credentials: StdMutex<Vec<RuntimeHttpCredential>>,
    runtime_id: String,
    generation: [u8; 16],
}

#[cfg(feature = "adapter-api")]
impl RuntimeHttpCredentialRegistry {
    fn new(
        runtime_id: impl Into<String>,
        credentials: Vec<RuntimeHttpCredential>,
    ) -> std::io::Result<Arc<Self>> {
        let mut generation = [0_u8; 16];
        getrandom::getrandom(&mut generation).map_err(|error| {
            std::io::Error::other(format!(
                "cannot create runtime credential generation: {error}"
            ))
        })?;
        Ok(Arc::new(Self {
            credentials: StdMutex::new(credentials),
            runtime_id: runtime_id.into(),
            generation,
        }))
    }

    fn authenticate(&self, request: &HttpRequest) -> Option<AuthenticatedRuntimeHttpCredential> {
        let credentials = self
            .credentials
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        debug_assert!(credentials.iter().all(|credential| {
            credential.client_id.is_none()
                || (credential.runtime_id.as_deref() == Some(self.runtime_id.as_str())
                    && credential.generation == Some(self.generation))
        }));
        check_auth(request, &credentials)
    }

    fn issue_frontend(
        &self,
        client_id: crate::RuntimeClientId,
        observer: bool,
    ) -> std::io::Result<IssuedRuntimeHttpCredential> {
        let authorization = if observer {
            RuntimeAuthorization::observer()
        } else {
            RuntimeAuthorization::interactive()
        };
        for _ in 0..3 {
            let mut secret = [0_u8; 32];
            getrandom::getrandom(&mut secret).map_err(|error| {
                std::io::Error::other(format!("cannot mint frontend credential: {error}"))
            })?;
            let token: Arc<str> = encode_credential(&secret).into();
            secret.fill(0);
            let mut credentials = self
                .credentials
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            if credentials
                .iter()
                .any(|credential| constant_time_eq(token.as_bytes(), credential.token.as_bytes()))
            {
                continue;
            }
            credentials.push(RuntimeHttpCredential::frontend(
                token.clone(),
                client_id,
                authorization,
                self.runtime_id.clone(),
                self.generation,
            ));
            return Ok(IssuedRuntimeHttpCredential(token));
        }
        Err(std::io::Error::new(
            std::io::ErrorKind::AlreadyExists,
            "frontend credential collision limit exceeded",
        ))
    }

    async fn revoke_client(&self, client_id: &crate::RuntimeClientId) -> bool {
        // Authentication removal and attachment registration share this lock,
        // so no credential-owned channel can appear after the removal point.
        let revocations = {
            let mut credentials = self
                .credentials
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            let mut revocations = Vec::new();
            credentials.retain(|credential| {
                if credential.client_id.as_ref() == Some(client_id) {
                    if let Some(revocation) = &credential.revocation {
                        revocations.push(revocation.clone());
                    }
                    false
                } else {
                    true
                }
            });
            revocations
        };
        let revoked = !revocations.is_empty();
        for revocation in revocations {
            revocation.revoke_and_wait().await;
        }
        revoked
    }
}

#[cfg(feature = "adapter-api")]
fn encode_credential(secret: &[u8; 32]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut encoded = String::with_capacity(64);
    for byte in secret {
        encoded.push(HEX[(byte >> 4) as usize] as char);
        encoded.push(HEX[(byte & 0x0f) as usize] as char);
    }
    encoded
}

/// Does `req` carry a recognized bearer credential? Checked two ways: the
/// standard `Authorization: Bearer <token>` header, or, for unbound legacy
/// credentials only, a `?token=` query-string parameter (kept for
/// `GET /events`, since browser `EventSource` cannot set custom headers).
/// Scoped frontend credentials are header-only. Compared with
/// [`constant_time_eq`].
#[cfg(feature = "adapter-api")]
fn check_auth(
    req: &HttpRequest,
    credentials: &[RuntimeHttpCredential],
) -> Option<AuthenticatedRuntimeHttpCredential> {
    if let Some(auth) = req.headers.get("authorization") {
        if let Some(t) = auth.strip_prefix("Bearer ") {
            for credential in credentials {
                if constant_time_eq(t.as_bytes(), credential.token.as_bytes()) {
                    return Some(AuthenticatedRuntimeHttpCredential {
                        authorization: credential.authorization.clone(),
                        client_id: credential.client_id.clone(),
                        bootstrap: credential.bootstrap,
                        revocation: credential
                            .revocation
                            .as_ref()
                            .map(|revocation| revocation.signal.subscribe()),
                        attachment: credential.revocation.as_ref().and_then(|revocation| {
                            matches!(req.path.as_str(), "/events" | "/frontend/events")
                                .then(|| revocation.register())
                        }),
                        via_bearer_header: true,
                    });
                }
            }
        }
    }
    for pair in req.query.split('&') {
        if let Some((k, v)) = pair.split_once('=') {
            if k == "token" {
                for credential in credentials
                    .iter()
                    .filter(|credential| credential.client_id.is_none())
                {
                    if constant_time_eq(v.as_bytes(), credential.token.as_bytes()) {
                        return Some(AuthenticatedRuntimeHttpCredential {
                            authorization: credential.authorization.clone(),
                            client_id: credential.client_id.clone(),
                            bootstrap: credential.bootstrap,
                            revocation: None,
                            attachment: None,
                            via_bearer_header: false,
                        });
                    }
                }
            }
        }
    }
    None
}

#[cfg(feature = "adapter-api")]
fn coordinated_http_client(
    request: &HttpRequest,
    coordinator: &Arc<CoordinatedRuntime>,
    credential: AuthenticatedRuntimeHttpCredential,
) -> Result<Arc<CoordinatedRuntimeClient>, crate::RuntimeLeaseError> {
    // Old authenticated API clients predate explicit client ids. Preserve
    // them as one named compatibility controller; current SDK clients always
    // send a random stable id and therefore coordinate independently.
    let supplied_client_id = request
        .headers
        .get("x-supercode-client-id")
        .map(String::as_str);
    let client_id = match credential.client_id.as_ref() {
        Some(bound) if supplied_client_id == Some(bound.as_str()) => bound.as_str(),
        Some(_) => return Err(crate::RuntimeLeaseError::InvalidClientId),
        None => supplied_client_id.unwrap_or("legacy-owner"),
    };
    let mut authorization = credential.authorization;
    if let Some(requested) = request.headers.get("x-supercode-permissions") {
        authorization = authorization.restrict_to(&RuntimeAuthorization::parse_header(requested)?);
    }
    Ok(coordinator.client(RuntimeClientId::parse(client_id)?, authorization))
}

#[cfg(feature = "adapter-api")]
async fn coordinated_runtime_rpc(
    client: Arc<CoordinatedRuntimeClient>,
    request: RpcRequest,
) -> Value {
    let id = request.id.clone();
    let method = crate::FrontendFacadeMethod::from_wire_name(&request.method);
    let result = match method {
        Some(crate::FrontendFacadeMethod::TakeControl) => client
            .take_control()
            .and_then(|snapshot| serde_json::to_value(snapshot).map_err(json_sdk_error)),
        Some(crate::FrontendFacadeMethod::Heartbeat) => client
            .heartbeat()
            .and_then(|snapshot| serde_json::to_value(snapshot).map_err(json_sdk_error)),
        Some(crate::FrontendFacadeMethod::Lease) => client
            .lease_snapshot()
            .and_then(|snapshot| serde_json::to_value(snapshot).map_err(json_sdk_error)),
        Some(crate::FrontendFacadeMethod::Detach) => {
            serde_json::to_value(client.detach()).map_err(json_sdk_error)
        }
        Some(crate::FrontendFacadeMethod::Close) => match client.close().await {
            Ok(()) => Ok(json!({"closed":true})),
            Err(error) => Err(error),
        },
        None if request.method == "shutdown" => match client.close().await {
            Ok(()) => Ok(json!({"shutting_down":true})),
            Err(error) => Err(error),
        },
        _ => return frontend_http_rpc(client, request).await,
    };
    match result {
        Ok(value) => rpc_ok(id, value),
        Err(error) => sdk_runtime_rpc_error(id, -32002, &error),
    }
}

#[cfg(feature = "adapter-api")]
fn json_sdk_error(error: serde_json::Error) -> FrontendRuntimeError {
    FrontendRuntimeError::Transport(error.to_string())
}

#[cfg(feature = "adapter-api")]
async fn handle_http_conn(
    stream: tokio::net::TcpStream,
    engine: Arc<RpcEngine>,
    coordinator: Arc<CoordinatedRuntime>,
    credentials: Arc<RuntimeHttpCredentialRegistry>,
) -> std::io::Result<()> {
    let peer_is_loopback = stream.peer_addr()?.ip().is_loopback();
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = tokio::io::BufReader::new(read_half);
    let Some(req) = read_http_request(&mut reader).await? else {
        return Ok(());
    };

    // Static observer assets contain no runtime descriptor or session content.
    // Serve them before authentication so a user can load the credential form;
    // every SDK request made by that page remains bearer-authenticated below.
    if req.method == "GET" {
        if let Some((content_type, body)) = browser_observer_asset(&req.path) {
            return write_browser_observer_asset(&mut write_half, content_type, body).await;
        }
    }

    let Some(credential) = credentials.authenticate(&req) else {
        let body =
            sdk_runtime_rpc_error(Value::Null, -32030, &FrontendRuntimeError::Unauthenticated)
                .to_string();
        return write_http_response(
            &mut write_half,
            401,
            "Unauthorized",
            "application/json",
            body.as_bytes(),
        )
        .await;
    };

    if matches!(
        req.path.as_str(),
        "/_supercode/frontend-credentials/mint" | "/_supercode/frontend-credentials/revoke"
    ) && !credential.via_bearer_header
    {
        let body =
            sdk_runtime_rpc_error(Value::Null, -32030, &FrontendRuntimeError::Unauthenticated)
                .to_string();
        return write_http_response(
            &mut write_half,
            401,
            "Unauthorized",
            "application/json",
            body.as_bytes(),
        )
        .await;
    }

    if req.path == "/_supercode/frontend-credentials/mint" {
        if req.method != "POST" || !peer_is_loopback || !credential.bootstrap {
            let body = sdk_runtime_rpc_error(
                Value::Null,
                -32031,
                &FrontendRuntimeError::Unauthorized {
                    permission: "bootstrap".into(),
                },
            )
            .to_string();
            return write_http_response(
                &mut write_half,
                403,
                "Forbidden",
                "application/json",
                body.as_bytes(),
            )
            .await;
        }
        let request: Value = match serde_json::from_slice(&req.body) {
            Ok(request) => request,
            Err(error) => {
                return write_http_response(
                    &mut write_half,
                    400,
                    "Bad Request",
                    "application/json",
                    format!("{{\"error\":{}}}", json!(error.to_string())).as_bytes(),
                )
                .await;
            }
        };
        let Some(client_id) = request.get("clientId").and_then(Value::as_str) else {
            return write_http_response(
                &mut write_half,
                400,
                "Bad Request",
                "application/json",
                b"{\"error\":\"mint request omitted clientId\"}",
            )
            .await;
        };
        let client_id = match crate::RuntimeClientId::parse(client_id) {
            Ok(client_id) => client_id,
            Err(error) => {
                return write_http_response(
                    &mut write_half,
                    400,
                    "Bad Request",
                    "application/json",
                    format!("{{\"error\":{}}}", json!(error.to_string())).as_bytes(),
                )
                .await;
            }
        };
        let observer = match request.get("grant").and_then(Value::as_str) {
            Some("observer") => true,
            Some("interactive") => false,
            _ => {
                return write_http_response(
                    &mut write_half,
                    400,
                    "Bad Request",
                    "application/json",
                    b"{\"error\":\"grant must be observer or interactive\"}",
                )
                .await;
            }
        };
        let token = credentials.issue_frontend(client_id, observer)?;
        return write_http_response(
            &mut write_half,
            200,
            "OK",
            "application/octet-stream",
            token.as_bytes(),
        )
        .await;
    }

    if req.path == "/_supercode/frontend-credentials/revoke" {
        if req.method != "POST" || !peer_is_loopback || !credential.bootstrap {
            let body = sdk_runtime_rpc_error(
                Value::Null,
                -32031,
                &FrontendRuntimeError::Unauthorized {
                    permission: "bootstrap".into(),
                },
            )
            .to_string();
            return write_http_response(
                &mut write_half,
                403,
                "Forbidden",
                "application/json",
                body.as_bytes(),
            )
            .await;
        }
        let request: Value = match serde_json::from_slice(&req.body) {
            Ok(request) => request,
            Err(error) => {
                return write_http_response(
                    &mut write_half,
                    400,
                    "Bad Request",
                    "application/json",
                    format!("{{\"error\":{}}}", json!(error.to_string())).as_bytes(),
                )
                .await;
            }
        };
        let Some(client_id) = request.get("clientId").and_then(Value::as_str) else {
            return write_http_response(
                &mut write_half,
                400,
                "Bad Request",
                "application/json",
                b"{\"error\":\"revoke request omitted clientId\"}",
            )
            .await;
        };
        let client_id = match crate::RuntimeClientId::parse(client_id) {
            Ok(client_id) => client_id,
            Err(error) => {
                return write_http_response(
                    &mut write_half,
                    400,
                    "Bad Request",
                    "application/json",
                    format!("{{\"error\":{}}}", json!(error.to_string())).as_bytes(),
                )
                .await;
            }
        };
        let revoked = credentials.revoke_client(&client_id).await;
        if revoked {
            coordinator
                .client(client_id, RuntimeAuthorization::observer())
                .detach();
        }
        return write_http_response(
            &mut write_half,
            200,
            "OK",
            "application/json",
            if revoked {
                b"{\"revoked\":true}"
            } else {
                b"{\"revoked\":false}"
            },
        )
        .await;
    }

    let mut revocation = credential.revocation.clone();
    let mut attachment = credential.attachment;
    let credential = AuthenticatedRuntimeHttpCredential {
        authorization: credential.authorization,
        client_id: credential.client_id,
        bootstrap: credential.bootstrap,
        revocation: None,
        attachment: None,
        via_bearer_header: credential.via_bearer_header,
    };
    let client = match coordinated_http_client(&req, &coordinator, credential) {
        Ok(client) => client,
        Err(error) => {
            let permission = match error {
                crate::RuntimeLeaseError::InvalidClientId => "client_id",
                crate::RuntimeLeaseError::InvalidAuthorization => "authorization",
                _ => "runtime",
            };
            let body = sdk_runtime_rpc_error(
                Value::Null,
                -32031,
                &FrontendRuntimeError::Unauthorized {
                    permission: permission.into(),
                },
            )
            .to_string();
            return write_http_response(
                &mut write_half,
                403,
                "Forbidden",
                "application/json",
                body.as_bytes(),
            )
            .await;
        }
    };

    match (req.method.as_str(), req.path.as_str()) {
        ("POST", "/rpc") => {
            let body_text = String::from_utf8_lossy(&req.body);
            let resp = match serde_json::from_str::<RpcRequest>(&body_text) {
                Ok(rpc_req) if matches!(rpc_req.method.as_str(), "status" | "history") => {
                    engine.handle_request(rpc_req).await
                }
                Ok(rpc_req) => coordinated_runtime_rpc(client.clone(), rpc_req).await,
                Err(e) => rpc_error(Value::Null, -32700, format!("parse error: {e}")),
            };
            let body = resp.to_string();
            write_http_response(
                &mut write_half,
                200,
                "OK",
                "application/json",
                body.as_bytes(),
            )
            .await
        }
        ("GET", "/events") => {
            if let Err(error) = client.observe() {
                let body = sdk_runtime_rpc_error(Value::Null, -32002, &error).to_string();
                return write_http_response(
                    &mut write_half,
                    403,
                    "Forbidden",
                    "application/json",
                    body.as_bytes(),
                )
                .await;
            }
            // Arm the subscriber before acknowledging SSE readiness. Otherwise a
            // client can receive 200, immediately submit a turn on /rpc, and lose
            // every event emitted before this branch reaches `subscribe()`.
            let mut events = engine.subscribe();
            let head = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\nConnection: close\r\n\r\n";
            if write_half.write_all(head.as_bytes()).await.is_err() {
                client.detach();
                return Ok(());
            }
            let _ = write_half.flush().await;
            loop {
                tokio::select! {
                    biased;
                    recv = events.recv() => {
                        match recv {
                            Ok(v) => {
                                let line = format!("data: {v}\n\n");
                                if write_half.write_all(line.as_bytes()).await.is_err() {
                                    break;
                                }
                                if write_half.flush().await.is_err() {
                                    break;
                                }
                            }
                            Err(broadcast::error::RecvError::Lagged(_)) => continue,
                            Err(broadcast::error::RecvError::Closed) => break,
                        }
                    }
                    // `RpcEngine::shutdown` publishes the turn terminal
                    // event before signaling shutdown. Drain that already-
                    // buffered event first so a close racing an active turn
                    // cannot make observers miss its final state.
                    _ = engine.wait_for_shutdown() => break,
                    _ = wait_for_credential_revocation(&mut revocation) => break,
                    _ = reader.read_u8() => break,
                }
            }
            let _ = write_half.shutdown().await;
            client.detach();
            drop(attachment.take());
            Ok(())
        }
        ("GET", "/frontend/events") => {
            if let Err(error) = client.observe() {
                let body = sdk_runtime_rpc_error(Value::Null, -32002, &error).to_string();
                return write_http_response(
                    &mut write_half,
                    403,
                    "Forbidden",
                    "application/json",
                    body.as_bytes(),
                )
                .await;
            }
            // Subscribe before acknowledging the stream. Once the client has
            // received the 200 response, every later frontend event is either
            // buffered here or delivered live; there is no header/subscription
            // race at connection startup.
            let mut events = engine.frontend_subscribe();
            let head = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\nConnection: close\r\n\r\n";
            if write_half.write_all(head.as_bytes()).await.is_err() {
                client.detach();
                return Ok(());
            }
            let _ = write_half.flush().await;
            loop {
                tokio::select! {
                    biased;
                    recv = events.recv() => {
                        match recv {
                            Ok(event) => {
                                let value = serde_json::to_string(&event).unwrap_or_default();
                                let line = format!("data: {value}\n\n");
                                if write_half.write_all(line.as_bytes()).await.is_err() {
                                    break;
                                }
                                if write_half.flush().await.is_err() {
                                    break;
                                }
                            }
                            Err(broadcast::error::RecvError::Lagged(_)) => break,
                            Err(broadcast::error::RecvError::Closed) => break,
                        }
                    }
                    // Shutdown is signaled only after the terminal frontend
                    // event is published. Prefer the receiver when both are
                    // ready so every observer sees that final sequence.
                    _ = engine.wait_for_shutdown() => break,
                    _ = wait_for_credential_revocation(&mut revocation) => break,
                    _ = reader.read_u8() => break,
                }
            }
            let _ = write_half.shutdown().await;
            client.detach();
            drop(attachment.take());
            Ok(())
        }
        _ => {
            write_http_response(
                &mut write_half,
                404,
                "Not Found",
                "application/json",
                b"{\"error\":\"not found\"}",
            )
            .await
        }
    }
}

#[cfg(feature = "adapter-api")]
async fn wait_for_credential_revocation(receiver: &mut Option<tokio::sync::watch::Receiver<bool>>) {
    let Some(receiver) = receiver else {
        std::future::pending::<()>().await;
        return;
    };
    if *receiver.borrow() {
        return;
    }
    while receiver.changed().await.is_ok() {
        if *receiver.borrow() {
            return;
        }
    }
}

#[cfg(feature = "adapter-api")]
async fn handle_frontend_http_conn(
    stream: tokio::net::TcpStream,
    coordinator: Arc<CoordinatedRuntime>,
    events: broadcast::Sender<FrontendEvent>,
    credentials: Arc<[RuntimeHttpCredential]>,
) -> std::io::Result<()> {
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = tokio::io::BufReader::new(read_half);
    let Some(req) = read_http_request(&mut reader).await? else {
        return Ok(());
    };
    if req.method == "GET" {
        if let Some((content_type, body)) = browser_observer_asset(&req.path) {
            return write_browser_observer_asset(&mut write_half, content_type, body).await;
        }
    }
    let Some(credential) = check_auth(&req, &credentials) else {
        return write_http_response(
            &mut write_half,
            401,
            "Unauthorized",
            "application/json",
            b"{\"error\":\"missing or invalid bearer token\"}",
        )
        .await;
    };
    let client = match coordinated_http_client(&req, &coordinator, credential) {
        Ok(client) => client,
        Err(error) => {
            let body = json!({"error":error.to_string()}).to_string();
            return write_http_response(
                &mut write_half,
                400,
                "Bad Request",
                "application/json",
                body.as_bytes(),
            )
            .await;
        }
    };
    match (req.method.as_str(), req.path.as_str()) {
        ("POST", "/rpc") => {
            let body_text = String::from_utf8_lossy(&req.body);
            let response = match serde_json::from_str::<RpcRequest>(&body_text) {
                Ok(request) => coordinated_runtime_rpc(client.clone(), request).await,
                Err(error) => rpc_error(Value::Null, -32700, format!("parse error: {error}")),
            };
            let body = response.to_string();
            write_http_response(
                &mut write_half,
                200,
                "OK",
                "application/json",
                body.as_bytes(),
            )
            .await
        }
        ("GET", "/frontend/events") => {
            if let Err(error) = client.observe() {
                let body = sdk_runtime_rpc_error(Value::Null, -32002, &error).to_string();
                return write_http_response(
                    &mut write_half,
                    403,
                    "Forbidden",
                    "application/json",
                    body.as_bytes(),
                )
                .await;
            }
            let mut receiver = events.subscribe();
            let head = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\nConnection: close\r\n\r\n";
            if write_half.write_all(head.as_bytes()).await.is_err() {
                client.detach();
                return Ok(());
            }
            let _ = write_half.flush().await;
            loop {
                tokio::select! {
                    _ = reader.read_u8() => break,
                    event = receiver.recv() => match event {
                        Ok(event) => {
                            let value = serde_json::to_string(&event).unwrap_or_default();
                            let line = format!("data: {value}\n\n");
                            if write_half.write_all(line.as_bytes()).await.is_err() || write_half.flush().await.is_err() {
                                break;
                            }
                        }
                        Err(broadcast::error::RecvError::Lagged(_)) => break,
                        Err(broadcast::error::RecvError::Closed) => break,
                    }
                }
            }
            client.detach();
            Ok(())
        }
        _ => {
            write_http_response(
                &mut write_half,
                404,
                "Not Found",
                "application/json",
                b"{\"error\":\"not found\"}",
            )
            .await
        }
    }
}

#[cfg(feature = "adapter-api")]
async fn frontend_http_rpc(runtime: Arc<dyn FrontendRuntime>, request: RpcRequest) -> Value {
    let id = request.id;
    let Some(method) = crate::FrontendFacadeMethod::from_wire_name(&request.method) else {
        return rpc_error(id, -32601, format!("unknown method `{}`", request.method));
    };
    match method {
        crate::FrontendFacadeMethod::Describe => match runtime.describe().await {
            Ok(descriptor) => rpc_ok(id, serde_json::to_value(descriptor).unwrap_or_default()),
            Err(error) => sdk_runtime_rpc_error(id, -32010, &error),
        },
        crate::FrontendFacadeMethod::Attach => {
            let limit = request
                .params
                .get("limit")
                .and_then(Value::as_u64)
                .unwrap_or(50)
                .clamp(1, SERVER_HISTORY_CAPACITY as u64) as usize;
            match runtime.attach(limit).await {
                Ok(attachment) => rpc_ok(
                    id,
                    serde_json::to_value(FrontendAttachSnapshot {
                        descriptor: attachment.descriptor,
                        history: attachment.history,
                        history_cursor: attachment.history_cursor,
                        replay: attachment.replay,
                    })
                    .unwrap_or_default(),
                ),
                Err(error) => sdk_runtime_rpc_error(id, -32010, &error),
            }
        }
        crate::FrontendFacadeMethod::SendInput => {
            let Some(prompt) = request.params.get("prompt").and_then(Value::as_str) else {
                return rpc_error(
                    id,
                    -32602,
                    "frontend.send_input requires a string `params.prompt`",
                );
            };
            match runtime.clone().send_input(prompt.to_string()).await {
                Ok(()) => rpc_ok(id, json!({"accepted": true})),
                Err(error @ FrontendRuntimeError::Submit(RuntimeSubmitError::Busy)) => {
                    sdk_runtime_rpc_error(id, -32000, &error)
                }
                Err(error) => sdk_runtime_rpc_error(id, -32002, &error),
            }
        }
        crate::FrontendFacadeMethod::Invoke => {
            let operation = request
                .params
                .get("operation")
                .cloned()
                .ok_or("frontend.invoke requires `params.operation`")
                .and_then(|value| {
                    serde_json::from_value(value).map_err(|_| "invalid frontend operation")
                });
            match operation {
                Ok(operation) => match runtime.invoke(operation).await {
                    Ok(result) => rpc_ok(id, serde_json::to_value(result).unwrap_or_default()),
                    Err(error @ FrontendRuntimeError::UnsupportedOperation(_)) => {
                        sdk_runtime_rpc_error(id, -32023, &error)
                    }
                    Err(error @ FrontendRuntimeError::Submit(RuntimeSubmitError::Busy)) => {
                        sdk_runtime_rpc_error(id, -32000, &error)
                    }
                    Err(error @ FrontendRuntimeError::Submit(RuntimeSubmitError::Interrupted)) => {
                        sdk_runtime_rpc_error(id, -32001, &error)
                    }
                    Err(error) => sdk_runtime_rpc_error(id, -32022, &error),
                },
                Err(message) => rpc_error(id, -32602, message),
            }
        }
        crate::FrontendFacadeMethod::Submit => {
            let Some(prompt) = request.params.get("prompt").and_then(Value::as_str) else {
                return rpc_error(id, -32602, "submit requires a string `params.prompt`");
            };
            let image_urls = match request.params.get("image_urls") {
                None => Vec::new(),
                Some(Value::Array(values)) => {
                    let Some(urls) = values.iter().map(Value::as_str).collect::<Option<Vec<_>>>()
                    else {
                        return rpc_error(
                            id,
                            -32602,
                            "submit requires string entries in `params.image_urls`",
                        );
                    };
                    urls.into_iter().map(str::to_owned).collect()
                }
                Some(_) => {
                    return rpc_error(id, -32602, "submit requires array `params.image_urls`")
                }
            };
            match runtime
                .submit_with_images(prompt.to_string(), image_urls)
                .await
            {
                Ok(reply) => rpc_ok(id, json!({"reply":reply})),
                Err(error @ FrontendRuntimeError::Submit(RuntimeSubmitError::Busy)) => {
                    sdk_runtime_rpc_error(id, -32000, &error)
                }
                Err(error @ FrontendRuntimeError::Submit(RuntimeSubmitError::Interrupted)) => {
                    sdk_runtime_rpc_error(id, -32001, &error)
                }
                Err(error) => sdk_runtime_rpc_error(id, -32002, &error),
            }
        }
        crate::FrontendFacadeMethod::Interrupt => match runtime.interrupt().await {
            Ok(interrupted) => rpc_ok(id, json!({"interrupted":interrupted})),
            Err(error) => sdk_runtime_rpc_error(id, -32002, &error),
        },
        crate::FrontendFacadeMethod::Steer => {
            let Some(prompt) = request.params.get("prompt").and_then(Value::as_str) else {
                return rpc_error(id, -32602, "steer requires a string `params.prompt`");
            };
            match runtime.steer(prompt.to_string()).await {
                Ok(()) => rpc_ok(id, json!({"queued":true})),
                Err(error @ FrontendRuntimeError::UnsupportedAction(_)) => {
                    sdk_runtime_rpc_error(id, -32020, &error)
                }
                Err(error) => sdk_runtime_rpc_error(id, -32022, &error),
            }
        }
        crate::FrontendFacadeMethod::Respond => {
            let response = request
                .params
                .get("response")
                .cloned()
                .ok_or("respond requires `params.response`")
                .and_then(|value| serde_json::from_value(value).map_err(|_| "invalid response"));
            match response {
                Ok(response) => match runtime.respond(response).await {
                    Ok(()) => rpc_ok(id, json!({"accepted":true})),
                    Err(error @ FrontendRuntimeError::UnsupportedAction(_)) => {
                        sdk_runtime_rpc_error(id, -32020, &error)
                    }
                    Err(error) => sdk_runtime_rpc_error(id, -32022, &error),
                },
                Err(message) => rpc_error(id, -32602, message),
            }
        }
        crate::FrontendFacadeMethod::Lease
        | crate::FrontendFacadeMethod::TakeControl
        | crate::FrontendFacadeMethod::Heartbeat
        | crate::FrontendFacadeMethod::Detach
        | crate::FrontendFacadeMethod::Close => rpc_error(
            id,
            -32020,
            format!(
                "frontend action `{}` requires a coordinated runtime",
                method.id()
            ),
        ),
    }
}

/// Lifetime handle for a frontend-only authenticated HTTP listener.
#[cfg(feature = "adapter-api")]
pub(crate) struct FrontendHttpServer {
    address: SocketAddr,
    task: tokio::task::JoinHandle<()>,
}

#[cfg(feature = "adapter-api")]
impl FrontendHttpServer {
    /// Actually-bound loopback address.
    pub(crate) fn address(&self) -> SocketAddr {
        self.address
    }
}

#[cfg(feature = "adapter-api")]
impl Drop for FrontendHttpServer {
    fn drop(&mut self) {
        self.task.abort();
    }
}

/// Publish only the versioned frontend contract for a non-Agent runtime.
/// The caller owns the runtime and this returned listener lease.
#[cfg(feature = "adapter-api")]
pub(crate) async fn run_frontend_http(
    runtime: Arc<dyn FrontendRuntime>,
    events: broadcast::Sender<FrontendEvent>,
    bind: &str,
    token: Arc<str>,
) -> std::io::Result<FrontendHttpServer> {
    let listener = TcpListener::bind(bind).await?;
    let address = listener.local_addr()?;
    let coordinator = CoordinatedRuntime::new(runtime);
    let credentials: Arc<[RuntimeHttpCredential]> =
        vec![RuntimeHttpCredential::owner(token)].into();
    let task = tokio::spawn(async move {
        while let Ok((stream, _)) = listener.accept().await {
            let coordinator = coordinator.clone();
            let events = events.clone();
            let credentials = credentials.clone();
            tokio::spawn(async move {
                let _ = handle_frontend_http_conn(stream, coordinator, events, credentials).await;
            });
        }
    });
    Ok(FrontendHttpServer { address, task })
}

/// Lifetime handle for an authenticated `frontend.v2` WebSocket listener.
/// Dropping the handle detaches the listener without closing its SDK runtime.
#[cfg(feature = "adapter-api")]
pub struct FrontendWebSocketServer {
    address: SocketAddr,
    task: tokio::task::JoinHandle<()>,
}

#[cfg(feature = "adapter-api")]
impl FrontendWebSocketServer {
    /// Actually-bound listener address.
    pub fn address(&self) -> SocketAddr {
        self.address
    }
}

#[cfg(feature = "adapter-api")]
impl Drop for FrontendWebSocketServer {
    fn drop(&mut self) {
        self.task.abort();
    }
}

/// Publish the language-neutral facade over authenticated WebSocket RPC.
/// The endpoint accepts only `/frontend/v2`, reuses the SDK coordinator, and
/// emits canonical events as `frontend.v2.event` notifications.
#[cfg(feature = "adapter-api")]
pub async fn run_frontend_websocket(
    engine: Arc<RpcEngine>,
    bind: &str,
    credentials: Vec<RuntimeHttpCredential>,
) -> std::io::Result<FrontendWebSocketServer> {
    let runtime: Arc<dyn FrontendRuntime> = engine.clone();
    let events = engine.frontend_events.clone();
    run_frontend_websocket_runtime_inner(runtime, events, Some(engine), bind, credentials).await
}

/// Publish the authenticated WebSocket facade for any SDK runtime without
/// constructing an Agent or a second execution loop. The caller owns the
/// runtime and event sender; dropping the returned server detaches the
/// listener.
#[cfg(all(feature = "adapter-api", test))]
pub(crate) async fn run_frontend_websocket_runtime(
    runtime: Arc<dyn FrontendRuntime>,
    events: broadcast::Sender<FrontendEvent>,
    bind: &str,
    credentials: Vec<RuntimeHttpCredential>,
) -> std::io::Result<FrontendWebSocketServer> {
    run_frontend_websocket_runtime_inner(runtime, events, None, bind, credentials).await
}

#[cfg(feature = "adapter-api")]
async fn run_frontend_websocket_runtime_inner(
    runtime: Arc<dyn FrontendRuntime>,
    events: broadcast::Sender<FrontendEvent>,
    shutdown_engine: Option<Arc<RpcEngine>>,
    bind: &str,
    credentials: Vec<RuntimeHttpCredential>,
) -> std::io::Result<FrontendWebSocketServer> {
    if credentials.is_empty()
        || credentials
            .iter()
            .any(|credential| credential.token.is_empty())
    {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            "at least one non-empty runtime WebSocket credential is required",
        ));
    }
    let listener = TcpListener::bind(bind).await?;
    let address = listener.local_addr()?;
    let coordinator = CoordinatedRuntime::new(runtime);
    let credentials: Arc<[RuntimeHttpCredential]> = credentials.into();
    let task = tokio::spawn(async move {
        loop {
            tokio::select! {
                biased;
                _ = wait_for_optional_runtime_shutdown(shutdown_engine.as_ref()) => break,
                accepted = listener.accept() => {
                    let Ok((stream, _)) = accepted else { continue };
                    let coordinator = coordinator.clone();
                    let credentials = credentials.clone();
                    let events = events.clone();
                    let shutdown_engine = shutdown_engine.clone();
                    tokio::spawn(async move {
                        let _ = handle_frontend_websocket(stream, events, shutdown_engine, coordinator, credentials).await;
                    });
                }
            }
        }
    });
    Ok(FrontendWebSocketServer { address, task })
}

#[cfg(feature = "adapter-api")]
async fn wait_for_optional_runtime_shutdown(engine: Option<&Arc<RpcEngine>>) {
    match engine {
        Some(engine) => engine.wait_for_shutdown().await,
        None => std::future::pending().await,
    }
}

#[cfg(feature = "adapter-api")]
#[allow(clippy::result_large_err)] // tungstenite's handshake callback fixes this error type.
async fn handle_frontend_websocket(
    stream: tokio::net::TcpStream,
    events: broadcast::Sender<FrontendEvent>,
    shutdown_engine: Option<Arc<RpcEngine>>,
    coordinator: Arc<CoordinatedRuntime>,
    credentials: Arc<[RuntimeHttpCredential]>,
) -> Result<(), tokio_tungstenite::tungstenite::Error> {
    use std::sync::Mutex as SyncMutex;
    use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request, Response};

    let selected = Arc::new(SyncMutex::new(None::<Arc<CoordinatedRuntimeClient>>));
    let selected_by_callback = selected.clone();
    let socket = tokio_tungstenite::accept_hdr_async(
        stream,
        move |request: &Request, response: Response| -> Result<Response, ErrorResponse> {
            let reject = |status, message: &str| {
                tokio_tungstenite::tungstenite::http::Response::builder()
                    .status(status)
                    .body(Some(message.to_string()))
                    .expect("static WebSocket rejection is valid")
            };
            if request.uri().path() != "/frontend/v2" {
                return Err(reject(404, "frontend WebSocket route not found"));
            }
            let token = request
                .headers()
                .get("authorization")
                .and_then(|value| value.to_str().ok())
                .and_then(|value| value.strip_prefix("Bearer "));
            let Some(credential) = token.and_then(|token| {
                credentials.iter().find(|credential| {
                    constant_time_eq(token.as_bytes(), credential.token.as_bytes())
                })
            }) else {
                return Err(reject(401, "missing or invalid bearer token"));
            };
            let client_id = request
                .headers()
                .get("x-supercode-client-id")
                .and_then(|value| value.to_str().ok())
                .unwrap_or("legacy-websocket-owner");
            let Ok(client_id) = RuntimeClientId::parse(client_id) else {
                return Err(reject(400, "invalid runtime client id"));
            };
            let mut authorization = credential.authorization.clone();
            if let Some(requested) = request
                .headers()
                .get("x-supercode-permissions")
                .and_then(|value| value.to_str().ok())
            {
                let Ok(requested) = RuntimeAuthorization::parse_header(requested) else {
                    return Err(reject(400, "invalid runtime authorization grant"));
                };
                authorization = authorization.restrict_to(&requested);
            }
            *selected_by_callback
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner) =
                Some(coordinator.client(client_id, authorization));
            Ok(response)
        },
    )
    .await?;
    let client = selected
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .take()
        .expect("successful WebSocket handshake selects a runtime client");
    if let Err(error) = client.observe() {
        let mut socket = socket;
        let value = sdk_runtime_rpc_error(Value::Null, -32002, &error).to_string();
        socket
            .send(tokio_tungstenite::tungstenite::Message::Text(value.into()))
            .await?;
        socket.close(None).await?;
        return Ok(());
    }

    let mut events = events.subscribe();
    let (mut writer, mut reader) = socket.split();
    loop {
        tokio::select! {
            biased;
            incoming = reader.next() => match incoming {
                Some(Ok(tokio_tungstenite::tungstenite::Message::Text(text))) => {
                    let response = match serde_json::from_str::<RpcRequest>(&text) {
                        Ok(request) => coordinated_runtime_rpc(client.clone(), request).await,
                        Err(error) => rpc_error(Value::Null, -32700, format!("parse error: {error}")),
                    };
                    writer.send(tokio_tungstenite::tungstenite::Message::Text(response.to_string().into())).await?;
                }
                Some(Ok(tokio_tungstenite::tungstenite::Message::Ping(payload))) => {
                    writer.send(tokio_tungstenite::tungstenite::Message::Pong(payload)).await?;
                }
                Some(Ok(tokio_tungstenite::tungstenite::Message::Close(_))) | None => break,
                Some(Ok(_)) => {}
                Some(Err(error)) => {
                    client.detach();
                    return Err(error);
                }
            },
            event = events.recv() => match event {
                Ok(event) => {
                    let notification = json!({
                        "jsonrpc":"2.0",
                        "method":"frontend.v2.event",
                        "params":{"event":event},
                    });
                    writer.send(tokio_tungstenite::tungstenite::Message::Text(notification.to_string().into())).await?;
                }
                Err(broadcast::error::RecvError::Lagged(count)) => {
                    let notification = json!({
                        "jsonrpc":"2.0",
                        "method":"frontend.v2.event",
                        "params":{"error":{"name":"transport","message":format!("event replay gap: {count}")}},
                    });
                    writer.send(tokio_tungstenite::tungstenite::Message::Text(notification.to_string().into())).await?;
                    break;
                }
                Err(broadcast::error::RecvError::Closed) => break,
            },
            _ = wait_for_optional_runtime_shutdown(shutdown_engine.as_ref()) => break,
        }
    }
    client.detach();
    Ok(())
}

/// Bind `bind` (`host:port`; `:0` for an OS-assigned ephemeral port) and
/// serve the HTTP transport (D8 "remote attach") in a background task until
/// `engine` signals shutdown. Returns the actually-bound address (so a
/// caller that asked for port `0` can learn the real port). Every
/// connection is authenticated per-request via `token` — see
/// `check_auth`. The LOOPBACK-BY-DEFAULT policy decision is the caller's
/// (see the module doc) — this fn binds whatever address it's given.
#[cfg(feature = "adapter-api")]
pub async fn run_http(
    engine: Arc<RpcEngine>,
    bind: &str,
    token: Arc<str>,
) -> std::io::Result<SocketAddr> {
    run_http_authorized(engine, bind, vec![RuntimeHttpCredential::owner(token)]).await
}

/// Bind an SDK HTTP runtime with multiple independently scoped bearer
/// credentials. The token bytes remain server-private; each successful
/// authentication produces the exact authorization grant projected by the
/// shared runtime coordinator.
#[cfg(feature = "adapter-api")]
pub async fn run_http_authorized(
    engine: Arc<RpcEngine>,
    bind: &str,
    credentials: Vec<RuntimeHttpCredential>,
) -> std::io::Result<SocketAddr> {
    run_http_authorized_with_lease_ttl(
        engine,
        bind,
        credentials,
        crate::DEFAULT_RUNTIME_LEASE_TTL_MS,
    )
    .await
}

/// Test/embedder variant of [`run_http_authorized`] with an explicit
/// controller lease duration.
#[cfg(feature = "adapter-api")]
pub async fn run_http_authorized_with_lease_ttl(
    engine: Arc<RpcEngine>,
    bind: &str,
    credentials: Vec<RuntimeHttpCredential>,
    lease_ttl_ms: u64,
) -> std::io::Result<SocketAddr> {
    if credentials.is_empty()
        || credentials
            .iter()
            .any(|credential| credential.token.is_empty())
    {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            "at least one non-empty runtime HTTP credential is required",
        ));
    }
    if lease_ttl_ms == 0 {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            "runtime lease TTL must be non-zero",
        ));
    }
    let listener = TcpListener::bind(bind).await?;
    let local_addr = listener.local_addr()?;
    let credentials = RuntimeHttpCredentialRegistry::new(engine.session_id(), credentials)?;
    let eng = engine;
    let runtime: Arc<dyn FrontendRuntime> = eng.clone();
    let coordinator = CoordinatedRuntime::with_lease_ttl(runtime, lease_ttl_ms);
    tokio::spawn(async move {
        loop {
            tokio::select! {
                biased;
                _ = eng.wait_for_shutdown() => break,
                accepted = listener.accept() => {
                    let Ok((stream, _addr)) = accepted else { continue };
                    let eng = eng.clone();
                    let coordinator = coordinator.clone();
                    let credentials = credentials.clone();
                    tokio::spawn(async move {
                        let _ = handle_http_conn(stream, eng, coordinator, credentials).await;
                    });
                }
            }
        }
    });
    Ok(local_addr)
}

#[cfg(test)]
mod frontend_binding_conformance_tests;

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

    fn cursor(data: &[u8]) -> BufReader<std::io::Cursor<Vec<u8>>> {
        BufReader::new(std::io::Cursor::new(data.to_vec()))
    }

    #[cfg(all(feature = "adapter-api", supercode_workspace_assets))]
    #[test]
    fn packaged_observer_assets_match_the_sdk_sources() {
        let pairs: &[(&str, &[u8], &[u8])] = &[
            (
                "frontend-browser/index.html",
                include_bytes!("../embedded/frontend-browser/index.html"),
                include_bytes!("../../../sdk/frontend-browser/index.html"),
            ),
            (
                "frontend-browser/app.mjs",
                include_bytes!("../embedded/frontend-browser/app.mjs"),
                include_bytes!("../../../sdk/frontend-browser/app.mjs"),
            ),
            (
                "frontend-browser/client.mjs",
                include_bytes!("../embedded/frontend-browser/client.mjs"),
                include_bytes!("../../../sdk/frontend-browser/client.mjs"),
            ),
            (
                "frontend-browser/view.mjs",
                include_bytes!("../embedded/frontend-browser/view.mjs"),
                include_bytes!("../../../sdk/frontend-browser/view.mjs"),
            ),
            (
                "frontend-browser/style.css",
                include_bytes!("../embedded/frontend-browser/style.css"),
                include_bytes!("../../../sdk/frontend-browser/style.css"),
            ),
            (
                "frontend-browser/favicon.svg",
                include_bytes!("../embedded/frontend-browser/favicon.svg"),
                include_bytes!("../../../sdk/frontend-browser/favicon.svg"),
            ),
            (
                "frontend/client.mjs",
                include_bytes!("../embedded/frontend/client.mjs"),
                include_bytes!("../../../sdk/frontend/client.mjs"),
            ),
            (
                "frontend/generated-client.mjs",
                include_bytes!("../embedded/frontend/generated-client.mjs"),
                include_bytes!("../../../sdk/frontend/generated-client.mjs"),
            ),
            (
                "frontend/generated.mjs",
                include_bytes!("../embedded/frontend/generated.mjs"),
                include_bytes!("../../../sdk/frontend/generated.mjs"),
            ),
        ];
        for (name, packaged, source) in pairs {
            assert_eq!(packaged, source, "packaged observer asset drifted: {name}");
        }
    }

    #[tokio::test]
    async fn admitted_submit_has_a_cancel_token_before_shutdown_observes_busy() {
        let agent =
            crate::Agent::new(crate::Config::builder().api_key("test-only-key").build()).unwrap();
        let engine = RpcEngine::new(agent, None);
        let claim = engine.claim_submit().unwrap();
        assert!(engine.busy.load(Ordering::SeqCst));
        assert!(engine
            .current_cancel
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .is_some());

        let cancel = claim.cancel.clone();
        let shutdown_engine = engine.clone();
        let shutdown = tokio::spawn(async move { shutdown_engine.shutdown().await });
        tokio::time::timeout(std::time::Duration::from_secs(1), cancel.notified())
            .await
            .expect("shutdown must interrupt an admitted claim before its future starts");
        assert!(
            !shutdown.is_finished(),
            "shutdown must retain the barrier until the admitted claim drains"
        );
        drop(claim);
        tokio::time::timeout(std::time::Duration::from_secs(1), shutdown)
            .await
            .expect("claim drain must release shutdown")
            .unwrap();
    }

    #[tokio::test]
    async fn read_bounded_line_reads_a_normal_line() {
        let mut r = cursor(b"hello\nworld\n");
        assert_eq!(
            read_bounded_line(&mut r, 1024).await.unwrap(),
            Some("hello".to_string())
        );
        assert_eq!(
            read_bounded_line(&mut r, 1024).await.unwrap(),
            Some("world".to_string())
        );
        assert_eq!(read_bounded_line(&mut r, 1024).await.unwrap(), None);
    }

    #[tokio::test]
    async fn read_bounded_line_strips_trailing_cr() {
        let mut r = cursor(b"hello\r\n");
        assert_eq!(
            read_bounded_line(&mut r, 1024).await.unwrap(),
            Some("hello".to_string())
        );
    }

    #[tokio::test]
    async fn read_bounded_line_returns_final_line_without_trailing_newline() {
        let mut r = cursor(b"no newline at eof");
        assert_eq!(
            read_bounded_line(&mut r, 1024).await.unwrap(),
            Some("no newline at eof".to_string())
        );
        assert_eq!(read_bounded_line(&mut r, 1024).await.unwrap(), None);
    }

    #[tokio::test]
    async fn read_bounded_line_errors_and_resyncs_on_an_oversized_line() {
        let mut data = vec![b'x'; 20];
        data.push(b'\n');
        data.extend_from_slice(b"next\n");
        let mut r = cursor(&data);
        let err = read_bounded_line(&mut r, 10).await.unwrap_err();
        assert!(err.to_string().contains("10 byte cap"));
        // Resynced: the NEXT call sees the following real line, not more
        // of the oversized one.
        assert_eq!(
            read_bounded_line(&mut r, 1024).await.unwrap(),
            Some("next".to_string())
        );
    }

    #[test]
    fn constant_time_eq_matches_equal_slices() {
        assert!(constant_time_eq(b"abc123", b"abc123"));
    }

    #[test]
    fn constant_time_eq_rejects_different_length_or_content() {
        assert!(!constant_time_eq(b"abc123", b"abc1234"));
        assert!(!constant_time_eq(b"abc123", b"xbc123"));
    }

    #[test]
    fn generate_token_is_64_hex_chars_and_varies() {
        let a = generate_token();
        let b = generate_token();
        assert_eq!(a.len(), 64);
        assert!(a.chars().all(|c| c.is_ascii_hexdigit()));
        assert_ne!(a, b, "two calls must not mint the same token");
    }

    #[cfg(feature = "adapter-api")]
    #[tokio::test]
    async fn credential_revocation_waits_for_registered_attachment_ack() {
        let revocation = Arc::new(RuntimeCredentialRevocation::new());
        let attachment = revocation.register();
        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
        let task = tokio::spawn({
            let revocation = revocation.clone();
            async move {
                let _ = started_tx.send(());
                revocation.revoke_and_wait().await;
            }
        });

        started_rx.await.unwrap();
        tokio::task::yield_now().await;
        assert!(
            !task.is_finished(),
            "revoke must remain pending while the attachment is registered"
        );

        drop(attachment);
        tokio::time::timeout(std::time::Duration::from_secs(1), task)
            .await
            .expect("attachment acknowledgement must release revoke")
            .unwrap();
    }

    #[cfg(feature = "adapter-api")]
    #[tokio::test]
    async fn credential_revocation_has_no_check_to_wait_lost_wakeup() {
        for _ in 0..10_000 {
            let revocation = Arc::new(RuntimeCredentialRevocation::new());
            let attachment = revocation.register();
            let (started_tx, started_rx) = tokio::sync::oneshot::channel();
            let task = tokio::spawn({
                let revocation = revocation.clone();
                async move {
                    let _ = started_tx.send(());
                    revocation.revoke_and_wait().await;
                }
            });

            started_rx.await.unwrap();
            drop(attachment);
            tokio::time::timeout(std::time::Duration::from_secs(1), task)
                .await
                .expect("revoke lost its attachment-drained wakeup")
                .unwrap();
        }
    }

    #[test]
    fn request_history_compaction_deduplicates_and_orders_resolutions() {
        let request = |sequence, id| {
            FrontendEvent::new(
                sequence,
                json!({"type": "request", "request": {"id": id, "kind": "approval", "payload": {}}}),
            )
        };
        let resolved = |sequence, id| {
            FrontendEvent::new(
                sequence,
                json!({"type": "request_resolved", "request_id": id, "response": {"kind": "approval", "request_id": id, "decision": "allow"}}),
            )
        };
        let replay = VecDeque::from([
            request(1, 2),
            resolved(2, 2),
            request(3, 1),
            resolved(4, 1),
            request(5, 2),
            resolved(6, 1),
        ]);

        let compacted = compact_frontend_request_history(&replay);
        assert_eq!(compacted.len(), 4);
        assert_eq!(compacted[0]["request"]["id"], 1);
        assert_eq!(compacted[1]["request_id"], 1);
        assert_eq!(compacted[2]["request"]["id"], 2);
        assert_eq!(compacted[3]["request_id"], 2);
    }
}