supercode_harness/runtime/

hosted.rs

//! Multi-frontend host for one harness-native runtime connection.
//!
//! The native ACP/RPC/app-server process remains the single model loop. This
//! host serializes control, broadcasts the lossless native events to SDK
//! clients, and projects the same events onto Supercode's frontend contract
//! so a terminal can join an editor-created session without resuming it.

use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex as StdMutex};

use async_trait::async_trait;
use serde_json::{json, Value};
use tokio::sync::{broadcast, mpsc, oneshot};

use super::{HarnessEvent, RuntimeCapabilities, RuntimeConnection, RuntimeHandle, RuntimeInput};
use crate::frontend::{
    FrontendActions, FrontendAttachment, FrontendConnectionState, FrontendDisplayCapabilities,
    FrontendEvent, FrontendRuntime, FrontendRuntimeDescriptor, FrontendRuntimeError,
    FrontendTurnState, FRONTEND_REPLAY_CAPACITY, FRONTEND_RUNTIME_SCHEMA_VERSION,
};
use crate::server::RuntimeSubmitError;
use crate::{
    ChatMessage, DiscoveryQuery, Error, Fidelity, FrontendResponse, HarnessCatalog, Result,
};

enum HostCommand {
    Submit {
        input: RuntimeInput,
        reply: oneshot::Sender<Result<Option<String>>>,
    },
    Interrupt {
        reply: oneshot::Sender<Result<()>>,
    },
    Steer {
        text: String,
        reply: oneshot::Sender<Result<()>>,
    },
    Respond {
        request_id: Value,
        response: Value,
        reply: oneshot::Sender<Result<()>>,
    },
    Shutdown {
        reply: oneshot::Sender<Result<()>>,
    },
}

struct ProjectionState {
    next_sequence: u64,
    replay: VecDeque<FrontendEvent>,
}

/// Shared owner of a single harness-native runtime.
pub struct HostedHarnessRuntime {
    handle: RuntimeHandle,
    capabilities: RuntimeCapabilities,
    commands: mpsc::Sender<HostCommand>,
    raw_events: broadcast::Sender<HarnessEvent>,
    frontend_events: broadcast::Sender<FrontendEvent>,
    projection: StdMutex<ProjectionState>,
    /// The conversation the harness had recorded before this host took it
    /// over; every frontend event since is in `projection.replay`.
    history: Vec<ChatMessage>,
    /// Requests the translator has raised and nobody has answered yet, keyed
    /// by the canonical id a frontend sees. BOTH doors — `frontend.v2.respond`
    /// and the owner's `harness.v1.runtimes.respond` — funnel through
    /// `respond_native`, which is where an entry is dropped and
    /// `request_resolved` is published, so one request takes exactly one
    /// answer and the door that did not answer reads the resolution.
    pending_requests: StdMutex<HashMap<u64, Value>>,
    /// What the projection must remember across a session's events (`NativeProjection`).
    native: StdMutex<NativeProjection>,
    busy: AtomicBool,
    closed: AtomicBool,
    registration: StdMutex<Option<crate::LiveRuntimeRegistration>>,
}

impl HostedHarnessRuntime {
    /// Promote one exclusive native connection into a multi-frontend host and
    /// return the first SDK connection to it.
    pub fn spawn(
        runtime: Box<dyn RuntimeConnection>,
        capabilities: RuntimeCapabilities,
        fresh: bool,
    ) -> (Arc<Self>, HostedHarnessConnection) {
        let handle = runtime.handle().clone();
        let (commands, command_rx) = mpsc::channel(32);
        let (raw_events, raw_rx) = broadcast::channel(1024);
        let (frontend_events, _) = broadcast::channel(1024);
        let host = Arc::new(Self {
            handle: handle.clone(),
            capabilities,
            commands,
            raw_events,
            frontend_events,
            projection: StdMutex::new(ProjectionState {
                next_sequence: 1,
                replay: VecDeque::new(),
            }),
            history: if fresh {
                Vec::new()
            } else {
                recorded_history(&handle)
            },
            pending_requests: StdMutex::new(HashMap::new()),
            native: StdMutex::new(NativeProjection::default()),
            busy: AtomicBool::new(false),
            closed: AtomicBool::new(false),
            registration: StdMutex::new(None),
        });
        tokio::spawn(run_native_runtime(
            runtime,
            Arc::downgrade(&host),
            command_rx,
        ));
        let connection = HostedHarnessConnection {
            host: host.clone(),
            handle,
            events: raw_rx,
            closed: false,
        };
        (host, connection)
    }

    /// Keep the receipt only while the native host remains open.
    pub(crate) fn retain_registration(&self, registration: crate::LiveRuntimeRegistration) {
        let mut retained = self
            .registration
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if !self.closed.load(Ordering::SeqCst) {
            *retained = Some(registration);
        }
    }

    /// Subscribe to the canonical sequenced frontend event stream.
    pub fn frontend_sender(&self) -> broadcast::Sender<FrontendEvent> {
        self.frontend_events.clone()
    }

    /// Shut down the one native process owned by this host.
    pub async fn shutdown(&self) -> Result<()> {
        if self.closed.load(Ordering::SeqCst) {
            return Ok(());
        }
        let (reply, response) = oneshot::channel();
        self.commands
            .send(HostCommand::Shutdown { reply })
            .await
            .map_err(|_| Error::Other("hosted harness runtime is closed".into()))?;
        response
            .await
            .map_err(|_| Error::Other("hosted harness runtime stopped before shutdown".into()))?
    }

    fn claim_submit(&self) -> Result<()> {
        if self.closed.load(Ordering::SeqCst) {
            return Err(Error::Other("hosted harness runtime is closed".into()));
        }
        if self
            .busy
            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
            .is_err()
        {
            return Err(Error::Other("a harness turn is already in progress".into()));
        }
        Ok(())
    }

    async fn submit_native_claimed(&self, input: RuntimeInput) -> Result<Option<String>> {
        self.publish(json!({"type":"user_message", "text":input.text}));
        self.publish(json!({"type":"turn_started"}));
        let (reply, response) = oneshot::channel();
        if self
            .commands
            .send(HostCommand::Submit { input, reply })
            .await
            .is_err()
        {
            self.busy.store(false, Ordering::SeqCst);
            self.publish(
                json!({"type":"turn_failed", "message":"Hosted harness runtime is closed."}),
            );
            self.publish(json!({"type":"turn_completed"}));
            self.mark_closed("Harness runtime command channel closed.");
            return Err(Error::Other("hosted harness runtime is closed".into()));
        }
        match response.await {
            Ok(Ok(turn)) => Ok(turn),
            Ok(Err(error)) => {
                self.busy.store(false, Ordering::SeqCst);
                self.publish(json!({"type":"turn_failed", "message":error.to_string()}));
                self.publish(json!({"type":"turn_completed"}));
                Err(error)
            }
            Err(_) => {
                self.busy.store(false, Ordering::SeqCst);
                self.publish(json!({"type":"turn_failed", "message":"Hosted harness runtime stopped before accepting input."}));
                self.publish(json!({"type":"turn_completed"}));
                self.mark_closed("Harness runtime stopped before accepting input.");
                Err(Error::Other(
                    "hosted harness runtime stopped before accepting input".into(),
                ))
            }
        }
    }

    async fn submit_native(&self, text: String) -> Result<Option<String>> {
        self.claim_submit()?;
        self.submit_native_claimed(RuntimeInput {
            text,
            image_urls: Vec::new(),
        })
        .await
    }

    async fn interrupt_native(&self) -> Result<()> {
        let (reply, response) = oneshot::channel();
        self.commands
            .send(HostCommand::Interrupt { reply })
            .await
            .map_err(|_| Error::Other("hosted harness runtime is closed".into()))?;
        response
            .await
            .map_err(|_| Error::Other("hosted harness runtime stopped before interrupt".into()))?
    }

    async fn steer_native(&self, text: String) -> Result<()> {
        let (reply, response) = oneshot::channel();
        self.commands
            .send(HostCommand::Steer { text, reply })
            .await
            .map_err(|_| Error::Other("hosted harness runtime is closed".into()))?;
        response
            .await
            .map_err(|_| Error::Other("hosted harness runtime stopped before steering".into()))?
    }

    async fn respond_native(&self, request_id: Value, response: Value) -> Result<()> {
        self.respond_native_as(request_id, response, None).await
    }

    /// Answer one native request. `canonical` is the portable response when a
    /// frontend answered; the owner's door passes `None` and the published
    /// resolution carries the native body it actually sent.
    async fn respond_native_as(
        &self,
        request_id: Value,
        response: Value,
        canonical: Option<Value>,
    ) -> Result<()> {
        let (reply, completed) = oneshot::channel();
        self.commands
            .send(HostCommand::Respond {
                request_id: request_id.clone(),
                response: response.clone(),
                reply,
            })
            .await
            .map_err(|_| Error::Other("hosted harness runtime is closed".into()))?;
        completed
            .await
            .map_err(|_| Error::Other("hosted harness runtime stopped before response".into()))??;
        // Only a request the native runtime ACCEPTED is resolved, and only
        // once: the first door through here removes it, so a second answer
        // finds nothing to publish.
        let resolved = {
            let mut pending = self
                .pending_requests
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            let found = pending
                .iter()
                .find(|(_, native)| *native == &request_id)
                .map(|(id, _)| *id);
            found.and_then(|id| pending.remove(&id).map(|_| id))
        };
        if let Some(id) = resolved {
            self.publish(json!({
                "type": "request_resolved",
                "request_id": id,
                "response": canonical.unwrap_or(response),
            }));
        }
        Ok(())
    }

    /// The native id a canonical request maps to, while it is still open.
    fn native_request_id(&self, request_id: u64) -> Option<Value> {
        self.pending_requests
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .get(&request_id)
            .cloned()
    }

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

    fn accept_native_event(&self, event: HarnessEvent) {
        let _ = self.raw_events.send(event.clone());
        let projected = self
            .native
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .project(self.handle.harness.as_str(), &event);
        for payload in projected {
            let terminal = matches!(
                payload.get("type").and_then(Value::as_str),
                Some("turn_succeeded" | "turn_interrupted" | "turn_failed")
            );
            if terminal {
                self.busy.store(false, Ordering::SeqCst);
            }
            if payload.get("type").and_then(Value::as_str) == Some("request") {
                if let (Some(id), Some(native)) = (
                    payload.pointer("/request/id").and_then(Value::as_u64),
                    payload
                        .pointer("/request/payload/native_request_id")
                        .cloned(),
                ) {
                    self.pending_requests
                        .lock()
                        .unwrap_or_else(std::sync::PoisonError::into_inner)
                        .insert(id, native);
                }
            }
            self.publish(payload);
        }
    }

    fn mark_closed(&self, message: impl Into<String>) {
        if self.closed.swap(true, Ordering::SeqCst) {
            return;
        }
        self.registration
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .take();
        let message = message.into();
        self.busy.store(false, Ordering::SeqCst);
        // The raw SDK connection and the portable frontend must observe the
        // same terminal edge. Keeping the sender alive inside `self` otherwise
        // leaves the SDK receiver waiting forever after native EOF.
        let _ = self.raw_events.send(HarnessEvent {
            sequence: None,
            kind: "transport_closed".into(),
            payload: json!({"message":message, "terminal":true}),
        });
        self.publish(json!({"type":"runtime_disconnected", "message":message}));
    }

    fn descriptor(&self) -> FrontendRuntimeDescriptor {
        FrontendRuntimeDescriptor {
            schema_version: FRONTEND_RUNTIME_SCHEMA_VERSION,
            session_id: self.handle.runtime_id.clone(),
            source_harness: Some(self.handle.harness.as_str().to_string()),
            emulation_profile: None,
            active_modules: Vec::new(),
            commands: Vec::new(),
            operations: Vec::new(),
            actions: FrontendActions {
                submit: self.capabilities.send_input,
                interrupt: self.capabilities.interrupt,
                steer: self.capabilities.steer,
                // ANSWERABLE where this harness has a reply translation, and
                // only there. `project_native_event` turns a harness's native
                // events into the canonical stream;
                // `hosted_permission_reply` is its mirror, turning a canonical
                // decision back into that harness's own reply envelope — so a
                // portable frontend can answer a request without ever seeing
                // the native shape. Today that is Claude Code, whose
                // permission-prompt-tool protocol takes
                // `{behavior:'allow'}` / `{behavior:'deny', message}`; a
                // harness with no translation keeps `false` and its requests
                // stay observe-only, which is what every other hosted harness
                // still is. The request event states which decisions its own
                // protocol accepts, so a frontend offers exactly those.
                //
                // The owner's `harness.v1.approvals.resolve` is unchanged and
                // still authoritative: both doors funnel through
                // `respond_native_as`, which drops the pending entry and
                // publishes `request_resolved` once, so one request takes one
                // answer and whichever door answered first wins.
                respond: !hosted_answerable_decisions(self.handle.harness.as_str()).is_empty(),
                detach: true,
                close: false,
            },
            display: FrontendDisplayCapabilities {
                event_kinds: vec![
                    "user_message".into(),
                    "turn_started".into(),
                    "turn_succeeded".into(),
                    "turn_interrupted".into(),
                    "turn_failed".into(),
                    "text_delta".into(),
                    "reasoning".into(),
                    "tool_call_started".into(),
                    "tool_call_completed".into(),
                    // Observation only — `actions.respond` stays false, so a
                    // frontend shows the request and no answer control.
                    "request".into(),
                    "native_event".into(),
                    "runtime_disconnected".into(),
                ],
                opaque_fallback: true,
            },
            model: self.handle.harness.as_str().to_string(),
            turn_state: if self.busy.load(Ordering::SeqCst) {
                FrontendTurnState::Busy
            } else {
                FrontendTurnState::Idle
            },
            connection_state: if self.closed.load(Ordering::SeqCst) {
                FrontendConnectionState::ShuttingDown
            } else {
                FrontendConnectionState::Connected
            },
            extensions: Default::default(),
        }
    }
}

#[async_trait]
impl FrontendRuntime for HostedHarnessRuntime {
    async fn describe(
        &self,
    ) -> std::result::Result<FrontendRuntimeDescriptor, FrontendRuntimeError> {
        Ok(self.descriptor())
    }

    async fn attach(
        &self,
        _history_limit: usize,
    ) -> std::result::Result<FrontendAttachment, FrontendRuntimeError> {
        let live = self.frontend_events.subscribe();
        let projection = self
            .projection
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let replay = projection.replay.clone();
        if let Some(first) = replay.front() {
            if first.sequence > 1 {
                return Err(FrontendRuntimeError::ReplayGap(first.sequence - 1));
            }
        }
        Ok(FrontendAttachment::new(
            self.descriptor(),
            self.history.clone(),
            0,
            replay,
            live,
            None,
        ))
    }

    async fn send_input(
        self: Arc<Self>,
        prompt: String,
    ) -> std::result::Result<(), FrontendRuntimeError> {
        self.claim_submit().map_err(hosted_submit_error)?;
        tokio::spawn(async move {
            let _ = self
                .submit_native_claimed(RuntimeInput {
                    text: prompt,
                    image_urls: Vec::new(),
                })
                .await;
        });
        Ok(())
    }

    async fn send_input_with_images(
        self: Arc<Self>,
        prompt: String,
        image_urls: Vec<String>,
    ) -> std::result::Result<(), FrontendRuntimeError> {
        self.claim_submit().map_err(hosted_submit_error)?;
        tokio::spawn(async move {
            let _ = self
                .submit_native_claimed(RuntimeInput {
                    text: prompt,
                    image_urls,
                })
                .await;
        });
        Ok(())
    }

    async fn submit(&self, prompt: String) -> std::result::Result<String, FrontendRuntimeError> {
        self.submit_native(prompt)
            .await
            .map(|turn| turn.unwrap_or_default())
            .map_err(hosted_submit_error)
    }

    async fn interrupt(&self) -> std::result::Result<bool, FrontendRuntimeError> {
        if !self.busy.load(Ordering::SeqCst) {
            return Ok(false);
        }
        self.interrupt_native()
            .await
            .map_err(|error| FrontendRuntimeError::Transport(error.to_string()))?;
        // Acceptance of the interrupt request does not mean the native turn
        // has stopped. Keep the shared runtime busy until its terminal event,
        // so another frontend cannot race a prompt into turn teardown.
        Ok(true)
    }

    async fn steer(&self, prompt: String) -> std::result::Result<(), FrontendRuntimeError> {
        if !self.busy.load(Ordering::SeqCst) || !self.capabilities.steer {
            return Err(FrontendRuntimeError::UnsupportedAction("steer"));
        }
        self.steer_native(prompt)
            .await
            .map_err(|error| FrontendRuntimeError::Transport(error.to_string()))
    }

    async fn respond(
        &self,
        response: FrontendResponse,
    ) -> std::result::Result<(), FrontendRuntimeError> {
        let harness = self.handle.harness.as_str();
        let FrontendResponse::Approval {
            request_id,
            decision,
        } = &response
        else {
            return Err(FrontendRuntimeError::UnsupportedAction(
                "respond: a hosted harness answers approval requests only",
            ));
        };
        let Some(native_id) = self.native_request_id(*request_id) else {
            return Err(FrontendRuntimeError::UnknownRequest(*request_id));
        };
        let body = hosted_permission_reply(harness, *decision)
            .map_err(FrontendRuntimeError::InvalidResponse)?;
        let canonical = serde_json::to_value(&response).unwrap_or(Value::Null);
        self.respond_native_as(native_id, body, Some(canonical))
            .await
            .map_err(|error| FrontendRuntimeError::Execution {
                operation: crate::SdkOperation::Respond,
                message: error.to_string(),
            })
    }
}

fn hosted_submit_error(error: Error) -> FrontendRuntimeError {
    if error.to_string().contains("already in progress") {
        FrontendRuntimeError::Submit(RuntimeSubmitError::Busy)
    } else {
        FrontendRuntimeError::Transport(error.to_string())
    }
}

/// SDK connection to a hosted native runtime. Closing it shuts down the
/// owner runtime; terminal HTTP attachments are non-owning frontend leases.
pub struct HostedHarnessConnection {
    host: Arc<HostedHarnessRuntime>,
    handle: RuntimeHandle,
    events: broadcast::Receiver<HarnessEvent>,
    closed: bool,
}

#[async_trait]
impl RuntimeConnection for HostedHarnessConnection {
    fn handle(&self) -> &RuntimeHandle {
        &self.handle
    }

    async fn send_input(&mut self, input: RuntimeInput) -> Result<Option<String>> {
        self.host.claim_submit()?;
        self.host.submit_native_claimed(input).await
    }

    async fn next_event(&mut self) -> Result<Option<HarnessEvent>> {
        match self.events.recv().await {
            Ok(event) => Ok(Some(event)),
            Err(broadcast::error::RecvError::Lagged(count)) => Err(Error::Other(format!(
                "hosted harness event stream lost {count} event(s)"
            ))),
            Err(broadcast::error::RecvError::Closed) => Ok(None),
        }
    }

    async fn interrupt(&mut self) -> Result<()> {
        self.host.interrupt_native().await
    }

    async fn steer(&mut self, text: String) -> Result<()> {
        self.host.steer_native(text).await
    }

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

    async fn close(&mut self) -> Result<()> {
        if self.closed {
            return Ok(());
        }
        self.host.shutdown().await?;
        self.closed = true;
        Ok(())
    }
}

async fn run_native_runtime(
    mut runtime: Box<dyn RuntimeConnection>,
    host: std::sync::Weak<HostedHarnessRuntime>,
    mut commands: mpsc::Receiver<HostCommand>,
) {
    loop {
        tokio::select! {
            command = commands.recv() => {
                let Some(command) = command else {
                    let _ = runtime.close().await;
                    return;
                };
                match command {
                    HostCommand::Submit { input, reply } => {
                        let _ = reply.send(runtime.send_input(input).await);
                    }
                    HostCommand::Interrupt { reply } => {
                        let _ = reply.send(runtime.interrupt().await);
                    }
                    HostCommand::Steer { text, reply } => {
                        let _ = reply.send(runtime.steer(text).await);
                    }
                    HostCommand::Respond { request_id, response, reply } => {
                        let _ = reply.send(runtime.respond(request_id, response).await);
                    }
                    HostCommand::Shutdown { reply } => {
                        let result = runtime.close().await;
                        let closed = result.is_ok();
                        let _ = reply.send(result);
                        if closed {
                            if let Some(host) = host.upgrade() {
                                host.mark_closed("Harness runtime closed.");
                            }
                            return;
                        }
                    }
                }
            }
            event = runtime.next_event() => {
                let Some(host) = host.upgrade() else {
                    let _ = runtime.close().await;
                    return;
                };
                match event {
                    Ok(Some(event)) if event.kind == "transport_closed" => {
                        host.mark_closed("Harness runtime transport closed.");
                        return;
                    }
                    Ok(Some(event)) => host.accept_native_event(event),
                    Ok(None) => {
                        host.mark_closed("Harness runtime transport closed.");
                        return;
                    }
                    Err(error) => {
                        host.mark_closed(error.to_string());
                        return;
                    }
                }
            }
        }
    }
}

/// Which harnesses can answer a `request` through the PORTABLE door, and the
/// decisions each one's own protocol accepts.
///
/// This is the mirror of `project_native_event`: that side translates a
/// harness's native events into the canonical stream, this side translates a
/// canonical decision back into the harness's own reply envelope. A harness
/// with no entry here keeps `respond: false` and its request stays
/// observe-only.
pub(crate) fn hosted_answerable_decisions(harness: &str) -> &'static [&'static str] {
    match harness {
        // Claude Code's permission-prompt-tool protocol, as its own validator
        // states it: `{behavior:'allow', updatedInput?:object}` or
        // `{behavior:'deny', message:string}`
        // (docs/interop/research/orc2-claude-respond-receipt-2026-09-04.json).
        // There is no always-allow BEHAVIOR — the CLI carries session
        // permission through a separate `updatedPermissions` field — so
        // `allow_for_session` is not offered rather than quietly downgraded to
        // a once-allow the person did not choose. `crates/harness/src/approvals.rs`
        // refuses it by name on the owner's door for the same reason.
        "claude-code" => &["allow", "deny"],
        _ => &[],
    }
}

/// Translate one canonical approval decision into the harness's own envelope.
fn hosted_permission_reply(
    harness: &str,
    decision: crate::FrontendApprovalDecision,
) -> std::result::Result<Value, String> {
    use crate::FrontendApprovalDecision as Decision;
    let accepted = hosted_answerable_decisions(harness);
    let wire = match decision {
        Decision::Allow => "allow",
        Decision::AllowForSession => "allow_for_session",
        Decision::Deny => "deny",
    };
    if !accepted.contains(&wire) {
        return Err(format!(
            "`{wire}` is not a decision a hosted {harness} runtime can carry; this harness accepts \
             {}",
            if accepted.is_empty() {
                "no portable decision — its requests are observe-only".to_string()
            } else {
                accepted
                    .iter()
                    .map(|decision| format!("`{decision}`"))
                    .collect::<Vec<_>>()
                    .join(" or ")
            },
        ));
    }
    match harness {
        "claude-code" => Ok(match decision {
            Decision::Allow => json!({"behavior": "allow"}),
            // Measured: the CLI's validator rejects a `deny` with no message.
            Decision::Deny => json!({
                "behavior": "deny",
                "message": "denied through the Volter Harness frontend",
            }),
            Decision::AllowForSession => unreachable!("refused above"),
        }),
        _ => Err(format!(
            "no hosted permission reply is defined for {harness}"
        )),
    }
}

/// Read once, when the host takes a resumed session over, so an opener sees
/// the conversation before this resume as a pane's scrollback shows it. The
/// replay begins at the host's first event, so the two never overlap. A
/// fresh session has no transcript yet and is never looked for: the search is
/// a scan of every session on the machine (7-11 s on a loaded Mac, and past
/// the SDK's 30 s request bound at worse moments, so `runtimes.start` failed).
fn recorded_history(handle: &RuntimeHandle) -> Vec<ChatMessage> {
    let catalog = HarnessCatalog::new();
    let Some(locator) = claude_transcript(handle).or_else(|| {
        let query = DiscoveryQuery {
            harnesses: vec![handle.harness.clone()],
            query: Some(handle.runtime_id.clone()),
            ..DiscoveryQuery::default()
        };
        catalog.discover(&query).ok().and_then(|found| {
            found
                .into_iter()
                .find(|descriptor| descriptor.locator.session_id == handle.runtime_id)
                .map(|descriptor| descriptor.locator)
        })
    }) else {
        return Vec::new();
    };
    catalog
        .load_display_view(&locator, Fidelity::Semantic, 500)
        .map(|session| session.messages)
        .unwrap_or_default()
}

/// A Claude Code session's transcript by its file name
/// (`projects/<folder>/<session id>.jsonl`): one look in each project folder
/// instead of reading every session's header.
fn claude_transcript(handle: &RuntimeHandle) -> Option<crate::SessionLocator> {
    if handle.harness.as_str() != crate::HarnessId::CLAUDE_CODE {
        return None;
    }
    let file = format!("{}.jsonl", handle.runtime_id);
    let projects = crate::HarnessHomes::default().claude_code;
    std::fs::read_dir(projects)
        .ok()?
        .flatten()
        .map(|folder| folder.path().join(&file))
        .find(|path| path.is_file())
        .map(|path| crate::SessionLocator {
            harness: handle.harness.clone(),
            session_id: handle.runtime_id.clone(),
            storage: crate::StorageLocator::File { path },
        })
}

/// The projection of one session's events, with what it must remember between them: OpenCode streams reasoning and
/// text alike as `message.part.delta` with `field: "text"` (its processor, `reasoning-delta` and `text-delta`), and only
/// the part an earlier `message.part.updated` declared tells them apart; and a turn OpenCode failed (`session.error`)
/// still goes idle after, which closes it without saying it succeeded.
#[derive(Debug, Default)]
pub struct NativeProjection {
    opencode_reasoning_parts: HashSet<String>,
    opencode_failed: bool,
}

impl NativeProjection {
    /// `event` of `harness` as `project_native_event` projects it, with the session's memory applied.
    pub fn project(&mut self, harness: &str, event: &HarnessEvent) -> Vec<Value> {
        if harness == "opencode" {
            let key = event.kind.to_ascii_lowercase();
            let payload = &event.payload;
            match key.as_str() {
                "message.part.updated" => {
                    let part = payload
                        .pointer("/properties/part")
                        .or_else(|| payload.get("part"))
                        .unwrap_or(payload);
                    if let (Some(id), Some(kind)) = (
                        part.get("id").and_then(Value::as_str),
                        part.get("type").and_then(Value::as_str),
                    ) {
                        if kind.contains("reasoning") {
                            self.opencode_reasoning_parts.insert(id.to_string());
                        }
                    }
                }
                "message.part.delta" => {
                    let reasoning = payload
                        .pointer("/properties/partID")
                        .and_then(Value::as_str)
                        .is_some_and(|id| self.opencode_reasoning_parts.contains(id));
                    if reasoning {
                        let text = payload
                            .pointer("/properties/delta")
                            .and_then(Value::as_str)
                            .unwrap_or_default();
                        return vec![json!({"type":"reasoning", "text":text, "raw":payload})];
                    }
                }
                "session.error" => {
                    self.opencode_failed = true;
                    return vec![
                        json!({"type":"turn_failed", "message":extract_text(payload).unwrap_or_else(|| "OpenCode session failed.".into())}),
                    ];
                }
                "session.idle" if self.opencode_failed => {
                    self.opencode_failed = false;
                    self.opencode_reasoning_parts.clear();
                    return vec![json!({"type":"turn_completed"})];
                }
                "session.idle" => self.opencode_reasoning_parts.clear(),
                // a new turn starts clean: a failure no idle followed does not carry into it
                "session.status"
                    if payload
                        .pointer("/properties/status/type")
                        .or_else(|| payload.pointer("/status/type"))
                        .and_then(Value::as_str)
                        == Some("busy") =>
                {
                    self.opencode_failed = false;
                }
                _ => {}
            }
        }
        project_native_event(harness, event)
    }
}

/// One native event of `harness` as the protocol-neutral events every reader renders (`text_delta`, `reasoning`,
/// `turn_succeeded`/`turn_failed`, `turn_completed`, `runtime_disconnected`, tools, approvals): ACP's and each
/// harness's own (Codex, Claude Code, Pi, OpenCode, generic) alike.
fn project_native_event(harness: &str, event: &HarnessEvent) -> Vec<Value> {
    let key = event.kind.to_ascii_lowercase().replace('-', "_");
    let payload = &event.payload;
    if key == "transport_closed" {
        return vec![
            json!({"type":"runtime_disconnected", "message":extract_text(payload).unwrap_or_else(|| "Harness runtime transport closed.".into())}),
        ];
    }
    let retrying = payload
        .pointer("/params/willRetry")
        .or_else(|| payload.get("willRetry"))
        .and_then(Value::as_bool)
        == Some(true);
    if key == "error" && retrying {
        // the harness says it is retrying (Codex app-server's `willRetry`): the turn goes on, and ends on its own
        return vec![native_payload(&key, payload)];
    }
    if key == "transport_error" || key == "error" {
        return vec![
            json!({"type":"turn_failed", "message":extract_text(payload).unwrap_or_else(|| "Harness runtime failed.".into()), "raw":payload}),
        ];
    }
    if key == "session/update" {
        let update = payload
            .pointer("/params/update")
            .or_else(|| payload.get("update"))
            .unwrap_or(payload);
        let update_kind = update
            .get("sessionUpdate")
            .or_else(|| update.get("type"))
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_ascii_lowercase();
        if update_kind == "agent_message_chunk" {
            return vec![
                json!({"type":"text_delta", "text":extract_text(update.get("content").unwrap_or(update)).unwrap_or_default(), "raw":payload}),
            ];
        }
        if update_kind == "agent_thought_chunk" {
            return vec![
                json!({"type":"reasoning", "text":extract_text(update.get("content").unwrap_or(update)).unwrap_or_default(), "raw":payload}),
            ];
        }
        if matches!(update_kind.as_str(), "tool_call" | "tool_call_update") {
            return vec![project_tool(update, payload)];
        }
    }
    if key == "supercode/acp_request_completed" {
        let failure = payload
            .pointer("/params/error")
            .or_else(|| payload.get("error"));
        return completion(failure.and_then(extract_text));
    }
    match harness {
        "codex" => project_codex(&key, payload),
        "claude-code" => project_claude(&key, payload),
        "pi" => project_pi(&key, payload),
        "opencode" => project_opencode(&key, payload),
        _ => project_generic(&key, payload),
    }
}

fn project_codex(key: &str, payload: &Value) -> Vec<Value> {
    if key == "turn/started" {
        return vec![native_payload(key, payload)];
    }
    if key == "turn/completed" {
        let status = payload
            .pointer("/params/turn/status")
            .or_else(|| payload.pointer("/turn/status"))
            .and_then(Value::as_str)
            .unwrap_or("completed")
            .to_ascii_lowercase();
        return completion(
            (status.contains("fail") || status.contains("error") || status.contains("cancel"))
                .then(|| extract_text(payload).unwrap_or_else(|| status.clone())),
        );
    }
    if key.ends_with("/delta") && !key.contains("agentmessage") && !key.contains("reasoning") {
        // a plan's or another item's stream is not the reply (`item/plan/delta`)
        return vec![native_payload(key, payload)];
    }
    if key.ends_with("/delta") {
        let text = payload
            .pointer("/params/delta")
            .or_else(|| payload.get("delta"))
            .and_then(extract_text)
            .or_else(|| extract_text(payload))
            .unwrap_or_default();
        return vec![
            json!({"type":if key.contains("reasoning") { "reasoning" } else { "text_delta" }, "text":text, "raw":payload}),
        ];
    }
    if key.contains("commandexecution")
        || key.contains("mcptool")
        || key.contains("filechange")
        || key.contains("tool")
    {
        let source = payload
            .pointer("/params/item")
            .or_else(|| payload.get("item"))
            .unwrap_or(payload);
        return vec![project_tool(source, payload)];
    }
    vec![native_payload(key, payload)]
}

/// Claude Code's stream-json, as `ClaudeCodeBackend` launches it: `--print
/// --input-format stream-json --output-format stream-json --verbose
/// --permission-prompt-tool stdio` (`crates/harness/src/runtime/adapters.rs`).
///
/// That argv carries no `--include-partial-messages`, so nothing arrives as a
/// `stream_event`: a turn is whole `assistant` and `user` events whose
/// `message.content` is the Anthropic content-block array, plus `system`,
/// `control_request` and a final `result`. The blocks are where the turn
/// actually lives — text, thinking, `tool_use`, and `tool_result` — so a
/// projector that reads only the envelope sees a turn with no tools in it and
/// leaves every one of them to the `native_event` fallback. This walks them.
fn project_claude(key: &str, payload: &Value) -> Vec<Value> {
    if key == "result" {
        let failed = payload.get("is_error").and_then(Value::as_bool) == Some(true)
            || payload
                .get("subtype")
                .and_then(Value::as_str)
                .is_some_and(|subtype| subtype != "success");
        return completion(
            failed.then(|| extract_text(payload).unwrap_or_else(|| "Claude turn failed.".into())),
        );
    }
    if key == "stream_event" {
        // Only reachable when a caller adds `--include-partial-messages`. The
        // whole-message `assistant` event still follows, so text would arrive
        // twice; the partial path deliberately projects ONLY what the whole
        // message cannot carry on its own — nothing today.
        let stream = payload
            .get("event")
            .or_else(|| payload.get("stream_event"))
            .unwrap_or(payload);
        if stream.get("type").and_then(Value::as_str) == Some("content_block_delta") {
            return vec![native_payload(key, payload)];
        }
    }
    if key == "assistant" {
        return project_claude_blocks(payload, true);
    }
    if key == "user" {
        return project_claude_blocks(payload, false);
    }
    if key == "control_request" {
        return project_claude_control_request(payload);
    }
    if key == "system" && claude_system_is_telemetry(payload) {
        return Vec::new();
    }
    if key == "tool_progress" {
        // The CLI's heartbeat for a tool still running (`elapsed_time_seconds`
        // every 30 s). The tool's own `tool_call_started` row already shows it
        // running; Claude Code's pane shows this only as a spinner's clock.
        return Vec::new();
    }
    if key == "rate_limit_event" {
        // Quota telemetry the CLI emits mid-turn. `allowed`/`allowed_warning`
        // is nothing a reader can act on and lands in the middle of the
        // assistant's own sentences; a `rejected` status is a real refusal, so
        // it falls through and stays visible.
        let status = payload
            .pointer("/rate_limit_info/status")
            .and_then(Value::as_str)
            .unwrap_or_default();
        if status.starts_with("allowed") {
            return Vec::new();
        }
    }
    project_generic(key, payload)
}

/// The `system` events Claude Code's own pane never writes into its
/// transcript, so a frontend showing them shows more than the pane does.
///
/// Measured on a hosted orchestrator turn (1077 events, 218 tool calls): 233
/// `thinking_tokens`, 115 `task_started` and 115 `task_notification` (one pair
/// per Bash call, every one foreground with an empty `output_file`), 18
/// `task_progress`, one `task_updated`, one `vcs_state_changed`.
///
/// - `init`: the CLI announcing its tools, model and cwd; session setup.
/// - `thinking_tokens`: a running estimate the pane shows only on its spinner.
/// - `task_started` / `task_progress` / `task_updated`: the CLI's bookkeeping
///   for a tool call the `tool_call_*` rows already show.
/// - `task_notification` for a foreground task (no `output_file`): the tail
///   of that same call. A background task's notification names the file its
///   output went to and is the only word that it finished, so it stays.
/// - `hook_started` / `hook_progress`, and `hook_response` that exited 0: a
///   hook that ran and passed is invisible in the pane; a failing one is not.
/// - `vcs_state_changed`: the CLI noticing a commit or push its own tool
///   call made.
fn claude_system_is_telemetry(payload: &Value) -> bool {
    match payload
        .get("subtype")
        .and_then(Value::as_str)
        .unwrap_or_default()
    {
        "init" | "thinking_tokens" | "task_started" | "task_progress" | "task_updated"
        | "hook_started" | "hook_progress" | "vcs_state_changed" => true,
        "task_notification" => payload
            .get("output_file")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .is_empty(),
        "hook_response" => payload.get("exit_code").and_then(Value::as_i64) == Some(0),
        _ => false,
    }
}

/// The Anthropic content-block array an `assistant` or `user` event carries.
fn claude_blocks(payload: &Value) -> Option<&Vec<Value>> {
    payload
        .pointer("/message/content")
        .or_else(|| payload.get("content"))
        .and_then(Value::as_array)
}

fn project_claude_blocks(payload: &Value, assistant: bool) -> Vec<Value> {
    let Some(blocks) = claude_blocks(payload) else {
        // A content-less envelope (or a plain string content) keeps the old
        // whole-payload reading rather than vanishing.
        let text = extract_text(payload).unwrap_or_default();
        if text.is_empty() {
            return vec![native_payload(
                if assistant { "assistant" } else { "user" },
                payload,
            )];
        }
        return vec![
            json!({"type":if assistant { "text_delta" } else { "user_message" }, "text":text, "raw":payload}),
        ];
    };
    let mut projected = Vec::new();
    for block in blocks {
        match block
            .get("type")
            .and_then(Value::as_str)
            .unwrap_or_default()
        {
            "text" => {
                let text = block
                    .get("text")
                    .and_then(Value::as_str)
                    .unwrap_or_default();
                if !text.is_empty() {
                    projected.push(json!({
                        "type": if assistant { "text_delta" } else { "user_message" },
                        "text": text,
                        "raw": block,
                    }));
                }
            }
            "thinking" | "redacted_thinking" => {
                let text = extract_text(block).unwrap_or_default();
                if !text.is_empty() {
                    projected.push(json!({"type":"reasoning", "text":text, "raw":block}));
                }
            }
            "tool_use" => projected.push(json!({
                "type": "tool_call_started",
                "id": block.get("id").cloned().unwrap_or(Value::Null),
                "name": block.get("name").cloned().unwrap_or(Value::Null),
                "arguments": block.get("input").map(Value::to_string).unwrap_or_default(),
                "raw": block,
            })),
            "tool_result" => projected.push(json!({
                // The result names only the call it answers — the tool's own
                // name was stated when it started, and a frontend correlates
                // the pair by that id.
                "type": "tool_call_completed",
                "id": block.get("tool_use_id").cloned().unwrap_or(Value::Null),
                "name": Value::Null,
                "output": extract_text(block.get("content").unwrap_or(block)).unwrap_or_default(),
                "is_error": block.get("is_error").and_then(Value::as_bool).unwrap_or(false),
                "raw": block,
            })),
            _ => projected.push(native_payload(
                if assistant { "assistant" } else { "user" },
                block,
            )),
        }
    }
    projected
}

/// Claude Code's `can_use_tool` control request as the canonical `request`.
///
/// OBSERVATION ONLY, and deliberately: the hosted descriptor advertises
/// `respond: false` because the portable contract cannot carry this harness's
/// native response envelope — the owner/editor answers it through
/// `harness.v1.approvals.resolve`. Emitting it is what that same decision
/// allows ("a terminal may observe requests"), and a frontend renders controls
/// only for actions the descriptor advertises, so no answer button appears.
fn project_claude_control_request(payload: &Value) -> Vec<Value> {
    let request = payload.get("request").unwrap_or(payload);
    if request.get("subtype").and_then(Value::as_str) != Some("can_use_tool") {
        return vec![native_payload("control_request", payload)];
    }
    let native_id = payload
        .get("request_id")
        .and_then(Value::as_str)
        .unwrap_or_default();
    vec![json!({
        "type": "request",
        "request": {
            "id": claude_request_id(native_id),
            "kind": "approval",
            "payload": {
                "tool": request.get("tool_name").or_else(|| request.get("toolName")).cloned().unwrap_or(Value::Null),
                "arguments": request.get("input").cloned().unwrap_or(Value::Null),
                "native_request_id": native_id,
                // What this harness's own protocol can carry, so a frontend
                // offers exactly those answers and no button that would be
                // refused.
                "decisions": hosted_answerable_decisions("claude-code"),
            },
        },
    })]
}

/// A stable JSON-safe number for a native request id that is a string.
///
/// The contract's `FrontendRequest.id` is numeric and Claude Code's
/// `request_id` is a uuid, so the id a frontend displays and correlates on is
/// this digest. The native id travels beside it, because that is the one the
/// runtime's own resolve door needs.
fn claude_request_id(native_id: &str) -> u64 {
    let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
    for byte in native_id.as_bytes() {
        hash ^= u64::from(*byte);
        hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
    }
    // Stay inside the range JSON numbers carry exactly.
    hash & ((1_u64 << 53) - 1)
}

fn project_pi(key: &str, payload: &Value) -> Vec<Value> {
    match key {
        "agent_start" => vec![native_payload(key, payload)],
        "agent_end" => completion(payload.get("error").and_then(extract_text)),
        // a command Pi refuses (`{"type":"response","success":false,"error":…}`): a refused prompt has no turn to end
        "response"
            if payload.get("success").and_then(Value::as_bool) == Some(false)
                && payload.get("command").and_then(Value::as_str) == Some("prompt") =>
        {
            completion(Some(
                payload
                    .get("error")
                    .and_then(extract_text)
                    .unwrap_or_else(|| "Pi refused the prompt.".into()),
            ))
        }
        "message_update" => {
            // pi streams one assistant message as `text_start` → `text_delta`*
            // → `text_end` (the same for `thinking_*`). Only the `*_delta`
            // events carry NEW text; `text_end` repeats the whole block as
            // `content`, and projecting it too doubled every reply.
            let update = payload
                .get("assistantMessageEvent")
                .or_else(|| payload.get("event"))
                .unwrap_or(payload);
            let kind = update
                .get("type")
                .and_then(Value::as_str)
                .unwrap_or_default();
            match kind {
                "text_delta" => vec![
                    json!({"type":"text_delta", "text":update.get("delta").and_then(Value::as_str).unwrap_or_default(), "raw":payload}),
                ],
                "thinking_delta" => vec![
                    json!({"type":"reasoning", "text":update.get("delta").and_then(Value::as_str).unwrap_or_default(), "raw":payload}),
                ],
                _ => vec![native_payload(key, payload)],
            }
        }
        _ if key.starts_with("tool_execution_") => vec![project_tool(payload, payload)],
        _ => vec![native_payload(key, payload)],
    }
}

fn project_opencode(key: &str, payload: &Value) -> Vec<Value> {
    if key == "session.idle" {
        return completion(None);
    }
    if key == "session.status" {
        let status = payload
            .pointer("/properties/status/type")
            .or_else(|| payload.pointer("/status/type"))
            .and_then(Value::as_str)
            .unwrap_or_default();
        if status == "busy" {
            return vec![native_payload(key, payload)];
        }
        // `session.status{idle}` precedes `session.idle` in every turn: the turn ends once, on `session.idle`
        return vec![native_payload(key, payload)];
    }
    if key == "message.part.delta" {
        // OpenCode 1.2's text stream: `{field, delta}` per part; the part's final `message.part.updated` repeats it whole
        let delta = payload
            .pointer("/properties/delta")
            .or_else(|| payload.get("delta"))
            .and_then(Value::as_str)
            .unwrap_or_default();
        let field = payload
            .pointer("/properties/field")
            .and_then(Value::as_str)
            .unwrap_or("text");
        return vec![
            json!({"type":if field.contains("reasoning") { "reasoning" } else { "text_delta" }, "text":delta, "raw":payload}),
        ];
    }
    if key == "message.part.updated" {
        let part = payload
            .pointer("/properties/part")
            .or_else(|| payload.get("part"))
            .unwrap_or(payload);
        if let Some(delta) = payload
            .pointer("/properties/delta")
            .or_else(|| payload.get("delta"))
            .and_then(Value::as_str)
        {
            let reasoning = part
                .get("type")
                .and_then(Value::as_str)
                .is_some_and(|kind| kind.contains("reasoning"));
            return vec![
                json!({"type":if reasoning { "reasoning" } else { "text_delta" }, "text":delta, "raw":payload}),
            ];
        }
        if part
            .get("type")
            .and_then(Value::as_str)
            .is_some_and(|kind| kind.contains("tool"))
        {
            return vec![project_tool(part, payload)];
        }
    }
    if key == "session.error" {
        return completion(Some(
            extract_text(payload).unwrap_or_else(|| "OpenCode session failed.".into()),
        ));
    }
    vec![native_payload(key, payload)]
}

fn project_generic(key: &str, payload: &Value) -> Vec<Value> {
    match key {
        "turn_started" | "turn/started" | "agent_start" => {
            vec![native_payload(key, payload)]
        }
        "turn_completed" | "turn/completed" | "agent_end" => {
            completion(payload.get("error").and_then(extract_text))
        }
        "output_delta" | "content_delta" => vec![
            json!({"type":"text_delta", "text":extract_text(payload).unwrap_or_default(), "raw":payload}),
        ],
        "reasoning_delta" => vec![
            json!({"type":"reasoning", "text":extract_text(payload).unwrap_or_default(), "raw":payload}),
        ],
        "tool" => vec![project_tool(payload, payload)],
        _ => vec![native_payload(key, payload)],
    }
}

fn completion(error: Option<String>) -> Vec<Value> {
    match error {
        Some(message) => vec![
            json!({"type":"turn_failed", "message":message}),
            json!({"type":"turn_completed"}),
        ],
        None => vec![
            json!({"type":"turn_succeeded"}),
            json!({"type":"turn_completed"}),
        ],
    }
}

fn project_tool(source: &Value, raw: &Value) -> Value {
    let status = source
        .get("status")
        .or_else(|| source.get("state"))
        .or_else(|| source.get("sessionUpdate"))
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_ascii_lowercase();
    let completed = status.contains("complete")
        || status.contains("result")
        || status.contains("success")
        || status.contains("error")
        || status.contains("fail");
    let arguments = source
        .get("arguments")
        .or_else(|| source.get("input"))
        .or_else(|| source.get("rawInput"))
        .cloned()
        .unwrap_or(Value::Null);
    json!({
        "type": if completed { "tool_call_completed" } else { "tool_call_started" },
        "id": source.get("toolCallId").or_else(|| source.get("tool_call_id")).or_else(|| source.get("callId")).or_else(|| source.get("id")).cloned().unwrap_or(Value::Null),
        "name": source.get("title").or_else(|| source.get("name")).or_else(|| source.get("tool")).or_else(|| source.get("toolName")).cloned().unwrap_or(Value::Null),
        "arguments": if arguments.is_string() { arguments } else { Value::String(arguments.to_string()) },
        "output": extract_text(source.get("result").or_else(|| source.get("output")).or_else(|| source.get("content")).unwrap_or(&Value::Null)).unwrap_or_default(),
        "is_error": status.contains("error") || status.contains("fail") || status.contains("denied"),
        "raw": raw,
    })
}

fn native_payload(kind: &str, payload: &Value) -> Value {
    json!({"type":"native_event", "kind":kind, "raw":payload})
}

fn extract_text(value: &Value) -> Option<String> {
    match value {
        Value::String(text) => Some(text.clone()),
        Value::Array(values) => {
            let text = values
                .iter()
                .filter_map(extract_text)
                .collect::<Vec<_>>()
                .join("\n");
            (!text.is_empty()).then_some(text)
        }
        Value::Object(object) => {
            for key in ["text", "delta", "content", "message", "result", "error"] {
                if let Some(text) = object.get(key).and_then(Value::as_str) {
                    return Some(text.to_string());
                }
            }
            for key in [
                "delta",
                "content",
                "message",
                "error",
                "data",
                "part",
                "params",
                "properties",
                "update",
                "event",
            ] {
                if let Some(text) = object.get(key).and_then(extract_text) {
                    return Some(text);
                }
            }
            None
        }
        _ => None,
    }
}