supercode_harness/

acp_frontend.rs

//! First-class ACP client implementation of the canonical SDK runtime.
//!
//! This adapter joins an already-running SDK-owned runtime through an ACP
//! bridge process. The bridge is a transport lease only: dropping or
//! detaching this client never owns or stops the remote runtime.

use std::collections::BTreeSet;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;

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

use crate::frontend::{
    FrontendAttachSnapshot, FrontendAttachment, FrontendOperationInvocation,
    FrontendOperationResult, FrontendResponse, FrontendRuntimeDescriptor,
};
use crate::runtime::{JsonLineClient, RuntimeEndpoint, RuntimeLaunch};
use crate::{SdkError, SdkOperation, SdkRuntime};

const FRONTEND_EXTENSION: &str = "/agentCapabilities/_meta/supercode/frontend";

/// Connection parameters for one ACP frontend attachment.
#[derive(Debug, Clone)]
pub struct AcpFrontendConnectOptions {
    /// Command that exposes an already-running runtime as ACP over stdio.
    /// A typical value is `supercode acp --connect <url>` plus its token.
    pub launch: RuntimeLaunch,
    /// Working directory for the bridge process.
    pub cwd: Option<PathBuf>,
    /// Stable session to resume. When absent, ACP discovery opens the single
    /// runtime represented by the endpoint and adopts the returned id.
    pub session_id: Option<String>,
    /// Last canonical event fully processed by an earlier client instance.
    pub after_sequence: Option<u64>,
}

/// Durable delivery cursor for restarting an ACP frontend without replaying
/// events it already consumed.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AcpFrontendCheckpoint {
    /// Checkpoint schema version.
    pub schema_version: u32,
    /// Stable SDK runtime/session identity.
    pub session_id: String,
    /// Highest canonical event delivered through the frontend attachment.
    pub acknowledged_sequence: u64,
}

/// ACP transport implementing the same [`SdkRuntime`] consumed by local and
/// HTTP frontends.
pub struct AcpFrontendRuntime {
    client: Arc<JsonLineClient>,
    events: broadcast::Sender<crate::SdkEvent>,
    session_id: String,
    methods: BTreeSet<String>,
    event_method: String,
    extensions: Value,
    acknowledged_sequence: Arc<AtomicU64>,
    last_received_sequence: Arc<AtomicU64>,
    endpoint: RuntimeEndpoint,
    disconnected: Arc<AtomicBool>,
}

impl AcpFrontendRuntime {
    /// Negotiate ACP, discover or resume one session, and start the canonical
    /// live-event subscription.
    pub async fn connect(options: AcpFrontendConnectOptions) -> Result<Arc<Self>, SdkError> {
        let (client, mut receiver, endpoint) = JsonLineClient::spawn(
            &options.launch,
            options.cwd.as_deref(),
            true,
            "acp-v1-jsonrpc",
        )
        .await
        .map_err(transport)?;
        let initialized = client
            .request(
                "initialize",
                json!({
                    "protocolVersion": crate::acp_server::ACP_PROTOCOL_VERSION,
                    "clientCapabilities": {
                        "_meta": {"supercode": {"frontend": {"schemaVersion": 2}}}
                    },
                    "clientInfo": {
                        "name": "supercode",
                        "title": "Supercode frontend",
                        "version": env!("CARGO_PKG_VERSION")
                    }
                }),
            )
            .await
            .map_err(transport)?;
        if initialized.get("protocolVersion").and_then(Value::as_u64)
            != Some(crate::acp_server::ACP_PROTOCOL_VERSION)
        {
            return Err(SdkError::Transport(
                "ACP frontend negotiated an unsupported protocol version".into(),
            ));
        }
        let extensions = initialized
            .pointer(FRONTEND_EXTENSION)
            .cloned()
            .ok_or_else(|| {
                SdkError::Transport(
                    "ACP peer does not advertise the Supercode frontend extension".into(),
                )
            })?;
        if extensions
            .get("runtimeOwnedByClient")
            .and_then(Value::as_bool)
            != Some(false)
        {
            return Err(SdkError::Transport(
                "ACP peer does not guarantee non-owning frontend detach semantics".into(),
            ));
        }
        let methods = extensions
            .get("methods")
            .and_then(Value::as_array)
            .into_iter()
            .flatten()
            .filter_map(Value::as_str)
            .map(str::to_owned)
            .collect::<BTreeSet<_>>();
        let event_method = extensions
            .get("eventMethod")
            .and_then(Value::as_str)
            .unwrap_or("supercode/frontend/event")
            .to_string();
        for (required, legacy) in [
            (
                crate::FrontendFacadeMethod::Describe,
                "supercode/frontend/describe",
            ),
            (
                crate::FrontendFacadeMethod::Attach,
                "supercode/frontend/attach",
            ),
        ] {
            if !supports_method(&methods, required, legacy) {
                return Err(SdkError::UnsupportedAction(
                    SdkOperation::Events.action_name(),
                ));
            }
        }

        let requested_session = options.session_id.as_deref();
        let (open_method, open_params) = if let Some(session_id) = requested_session {
            let resume = initialized
                .pointer("/agentCapabilities/sessionCapabilities/resume")
                .is_some();
            let load = initialized
                .pointer("/agentCapabilities/loadSession")
                .and_then(Value::as_bool)
                .unwrap_or(false);
            let method = if resume {
                "session/resume"
            } else if load {
                "session/load"
            } else {
                return Err(SdkError::UnsupportedAction(
                    SdkOperation::Resume.action_name(),
                ));
            };
            (method, json!({"sessionId":session_id}))
        } else {
            ("session/new", json!({}))
        };
        let opened = client
            .request(open_method, open_params)
            .await
            .map_err(transport)?;
        let session_id = opened
            .get("sessionId")
            .and_then(Value::as_str)
            .ok_or_else(|| SdkError::Transport("ACP session open omitted sessionId".into()))?
            .to_string();
        if requested_session.is_some_and(|requested| requested != session_id) {
            return Err(SdkError::NotFound {
                operation: SdkOperation::Resume,
                message: format!(
                    "ACP opened `{session_id}` instead of `{}`",
                    requested_session.unwrap()
                ),
            });
        }

        let (events, _) = broadcast::channel(1024);
        let disconnected = Arc::new(AtomicBool::new(false));
        let event_sender = events.clone();
        let event_disconnected = disconnected.clone();
        let event_session_id = session_id.clone();
        let negotiated_event_method = event_method.clone();
        let last_seen = Arc::new(AtomicU64::new(options.after_sequence.unwrap_or_default()));
        let last_received_sequence = last_seen.clone();
        tokio::spawn(async move {
            while let Some(message) = receiver.recv().await {
                if message.get("method").and_then(Value::as_str)
                    == Some(negotiated_event_method.as_str())
                    && message.pointer("/params/sessionId").and_then(Value::as_str)
                        == Some(event_session_id.as_str())
                {
                    if let Ok(event) = serde_json::from_value::<crate::SdkEvent>(
                        message
                            .pointer("/params/event")
                            .cloned()
                            .unwrap_or(Value::Null),
                    ) {
                        last_seen.fetch_max(event.sequence, Ordering::SeqCst);
                        let _ = event_sender.send(event);
                    }
                    continue;
                }
                let kind = message.get("type").and_then(Value::as_str);
                if matches!(kind, Some("transport_closed" | "transport_error")) {
                    event_disconnected.store(true, Ordering::SeqCst);
                    let sequence = last_seen.fetch_add(1, Ordering::SeqCst) + 1;
                    let _ = event_sender.send(crate::SdkEvent {
                        sequence,
                        kind: "runtime_disconnected".into(),
                        payload: json!({
                            "type":"runtime_disconnected",
                            "message": message.get("message").cloned().unwrap_or_else(|| json!("ACP transport closed")),
                            "_meta":{"supercode":{"transient":true,"transport":"acp"}}
                        }),
                    });
                    break;
                }
            }
        });

        let runtime = Arc::new(Self {
            client,
            events,
            session_id,
            methods,
            event_method,
            extensions,
            acknowledged_sequence: Arc::new(AtomicU64::new(
                options.after_sequence.unwrap_or_default(),
            )),
            last_received_sequence,
            endpoint,
            disconnected,
        });
        // Validate the typed descriptor before exposing the runtime.
        let _ = runtime.describe().await?;
        Ok(runtime)
    }

    /// Stable session discovered or resumed during negotiation.
    pub fn session_id(&self) -> &str {
        &self.session_id
    }

    /// Concrete local bridge endpoint, useful for process supervision.
    pub fn endpoint(&self) -> &RuntimeEndpoint {
        &self.endpoint
    }

    /// Complete negotiated extension object. Unknown fields are retained so
    /// newer peers can be proxied or inspected without lossy decoding.
    pub fn extensions(&self) -> &Value {
        &self.extensions
    }

    /// Negotiated ACP notification carrying canonical facade events.
    pub fn event_method(&self) -> &str {
        &self.event_method
    }

    /// Restore a delivery cursor after negotiation and before attachment.
    pub fn restore_checkpoint(&self, checkpoint: AcpFrontendCheckpoint) -> Result<(), SdkError> {
        validate_checkpoint(&checkpoint, &self.session_id)?;
        self.acknowledged_sequence
            .fetch_max(checkpoint.acknowledged_sequence, Ordering::SeqCst);
        self.last_received_sequence
            .fetch_max(checkpoint.acknowledged_sequence, Ordering::SeqCst);
        Ok(())
    }

    /// Snapshot the highest event actually delivered through an attachment.
    pub fn checkpoint(&self) -> AcpFrontendCheckpoint {
        AcpFrontendCheckpoint {
            schema_version: 1,
            session_id: self.session_id.clone(),
            acknowledged_sequence: self.acknowledged_sequence.load(Ordering::SeqCst),
        }
    }

    /// Detach this transport lease without stopping the SDK-owned runtime.
    pub async fn detach(&self) -> Result<(), SdkError> {
        if self.disconnected.load(Ordering::SeqCst) {
            return self.client.close().await.map_err(transport);
        }
        if supports_method(
            &self.methods,
            crate::FrontendFacadeMethod::Detach,
            "supercode/frontend/detach",
        ) {
            let _ = SdkRuntime::detach(self).await?;
        }
        self.client.close().await.map_err(transport)
    }

    fn method(
        &self,
        method: crate::FrontendFacadeMethod,
        legacy: &'static str,
        operation: SdkOperation,
    ) -> Result<&'static str, SdkError> {
        if self.methods.contains(method.wire_name()) {
            Ok(method.wire_name())
        } else if self.methods.contains(legacy) {
            Ok(legacy)
        } else {
            Err(SdkError::UnsupportedAction(operation.action_name()))
        }
    }

    async fn request(
        &self,
        method: &str,
        params: Value,
        operation: SdkOperation,
    ) -> Result<Value, SdkError> {
        let (_id, response) = self
            .client
            .begin_request(method, params)
            .await
            .map_err(transport)?;
        match response.await {
            Ok(Ok(value)) => Ok(value),
            Ok(Err(error)) => Err(decode_sdk_error(&error, operation)),
            Err(_) => Err(SdkError::Closed),
        }
    }

    fn mask_descriptor(
        &self,
        mut descriptor: FrontendRuntimeDescriptor,
    ) -> FrontendRuntimeDescriptor {
        mask_actions(&self.methods, &mut descriptor.actions);
        descriptor
    }
}

fn supports_method(
    methods: &BTreeSet<String>,
    method: crate::FrontendFacadeMethod,
    legacy: &str,
) -> bool {
    methods.contains(method.wire_name()) || methods.contains(legacy)
}

fn validate_checkpoint(
    checkpoint: &AcpFrontendCheckpoint,
    session_id: &str,
) -> Result<(), SdkError> {
    if checkpoint.schema_version != 1 {
        return Err(SdkError::Transport(format!(
            "unsupported ACP frontend checkpoint schema {}",
            checkpoint.schema_version
        )));
    }
    if checkpoint.session_id != session_id {
        return Err(SdkError::Transport(format!(
            "ACP frontend checkpoint is for `{}`, not `{session_id}`",
            checkpoint.session_id
        )));
    }
    Ok(())
}

fn mask_actions(methods: &BTreeSet<String>, actions: &mut crate::FrontendActions) {
    actions.submit &= supports_method(
        methods,
        crate::FrontendFacadeMethod::Submit,
        "supercode/frontend/submit",
    ) && supports_method(
        methods,
        crate::FrontendFacadeMethod::SendInput,
        "supercode/frontend/send_input",
    );
    actions.interrupt &= supports_method(
        methods,
        crate::FrontendFacadeMethod::Interrupt,
        "supercode/frontend/interrupt",
    );
    actions.steer &= supports_method(methods, crate::FrontendFacadeMethod::Steer, "session/steer");
    actions.respond &= supports_method(
        methods,
        crate::FrontendFacadeMethod::Respond,
        "session/respond",
    );
    actions.detach &= supports_method(
        methods,
        crate::FrontendFacadeMethod::Detach,
        "supercode/frontend/detach",
    );
    // A future peer may advertise an explicit close route. This adapter
    // never guesses one from process ownership or transport EOF.
    actions.close &= supports_method(
        methods,
        crate::FrontendFacadeMethod::Close,
        "supercode/frontend/close",
    );
}

fn suppress_acknowledged_history(
    history: &mut Vec<crate::ChatMessage>,
    history_cursor: u64,
    acknowledged_sequence: u64,
) {
    // Canonical history is an atomic projection through `history_cursor`.
    // Once a reconnecting frontend has acknowledged that boundary, rendering
    // the projection again would duplicate already displayed messages. A
    // cursor inside the projection cannot be mapped to individual messages,
    // so retain the complete history in that case rather than risk loss.
    if acknowledged_sequence > 0 && acknowledged_sequence >= history_cursor {
        history.clear();
    }
}

#[async_trait]
impl SdkRuntime for AcpFrontendRuntime {
    async fn describe(&self) -> Result<FrontendRuntimeDescriptor, SdkError> {
        let method = self.method(
            crate::FrontendFacadeMethod::Describe,
            "supercode/frontend/describe",
            SdkOperation::Resume,
        )?;
        let descriptor = self
            .request(
                method,
                json!({"sessionId":self.session_id}),
                SdkOperation::Resume,
            )
            .await?;
        serde_json::from_value(descriptor)
            .map(|descriptor| self.mask_descriptor(descriptor))
            .map_err(|error| SdkError::Transport(error.to_string()))
    }

    async fn attach(&self, history_limit: usize) -> Result<FrontendAttachment, SdkError> {
        if self.disconnected.load(Ordering::SeqCst) {
            return Err(SdkError::Closed);
        }
        let live = self.events.subscribe();
        let method = self.method(
            crate::FrontendFacadeMethod::Attach,
            "supercode/frontend/attach",
            SdkOperation::Events,
        )?;
        let acknowledged_sequence = self.acknowledged_sequence.load(Ordering::SeqCst);
        let value = self
            .request(
                method,
                json!({
                    "sessionId":self.session_id,
                    "limit":history_limit,
                    "afterSequence":acknowledged_sequence,
                }),
                SdkOperation::Events,
            )
            .await?;
        let mut snapshot: FrontendAttachSnapshot = serde_json::from_value(value)
            .map_err(|error| SdkError::Transport(error.to_string()))?;
        snapshot.descriptor = self.mask_descriptor(snapshot.descriptor);
        suppress_acknowledged_history(
            &mut snapshot.history,
            snapshot.history_cursor,
            acknowledged_sequence,
        );
        Ok(
            FrontendAttachment::from_snapshot_after(snapshot, live, acknowledged_sequence)
                .with_acknowledgement(self.acknowledged_sequence.clone()),
        )
    }

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

    async fn send_input_with_images(
        self: Arc<Self>,
        prompt: String,
        image_urls: Vec<String>,
    ) -> Result<(), SdkError> {
        let method = self.method(
            crate::FrontendFacadeMethod::SendInput,
            "supercode/frontend/send_input",
            SdkOperation::Input,
        )?;
        self.request(
            method,
            json!({"sessionId":self.session_id,"prompt":prompt,"image_urls":image_urls}),
            SdkOperation::Input,
        )
        .await
        .map(|_| ())
    }

    async fn submit(&self, prompt: String) -> Result<String, SdkError> {
        let method = self.method(
            crate::FrontendFacadeMethod::Submit,
            "supercode/frontend/submit",
            SdkOperation::Input,
        )?;
        let result = self
            .request(
                method,
                json!({"sessionId":self.session_id,"prompt":prompt}),
                SdkOperation::Input,
            )
            .await?;
        result
            .get("reply")
            .and_then(Value::as_str)
            .map(str::to_owned)
            .ok_or_else(|| SdkError::Transport("ACP submit omitted reply".into()))
    }

    async fn submit_with_images(
        &self,
        prompt: String,
        image_urls: Vec<String>,
    ) -> Result<String, SdkError> {
        let method = self.method(
            crate::FrontendFacadeMethod::Submit,
            "supercode/frontend/submit",
            SdkOperation::Input,
        )?;
        let result = self
            .request(
                method,
                json!({"sessionId":self.session_id,"prompt":prompt,"image_urls":image_urls}),
                SdkOperation::Input,
            )
            .await?;
        result
            .get("reply")
            .and_then(Value::as_str)
            .map(str::to_owned)
            .ok_or_else(|| SdkError::Transport("ACP submit omitted reply".into()))
    }

    async fn interrupt(&self) -> Result<bool, SdkError> {
        let method = self.method(
            crate::FrontendFacadeMethod::Interrupt,
            "supercode/frontend/interrupt",
            SdkOperation::Interrupt,
        )?;
        let result = self
            .request(
                method,
                json!({"sessionId":self.session_id}),
                SdkOperation::Interrupt,
            )
            .await?;
        Ok(result
            .get("interrupted")
            .and_then(Value::as_bool)
            .unwrap_or(false))
    }

    async fn steer(&self, prompt: String) -> Result<(), SdkError> {
        let method = self.method(
            crate::FrontendFacadeMethod::Steer,
            "session/steer",
            SdkOperation::Steer,
        )?;
        self.request(
            method,
            json!({"sessionId":self.session_id,"text":prompt}),
            SdkOperation::Steer,
        )
        .await
        .map(|_| ())
    }

    async fn respond(&self, response: FrontendResponse) -> Result<(), SdkError> {
        let method = self.method(
            crate::FrontendFacadeMethod::Respond,
            "session/respond",
            SdkOperation::Respond,
        )?;
        self.request(
            method,
            json!({"sessionId":self.session_id,"response":response}),
            SdkOperation::Respond,
        )
        .await
        .map(|_| ())
    }

    async fn invoke(
        &self,
        operation: FrontendOperationInvocation,
    ) -> Result<FrontendOperationResult, SdkError> {
        let method = self.method(
            crate::FrontendFacadeMethod::Invoke,
            "supercode/frontend/invoke",
            SdkOperation::Input,
        )?;
        let result = self
            .request(
                method,
                json!({"sessionId":self.session_id,"operation":operation}),
                SdkOperation::Input,
            )
            .await?;
        serde_json::from_value(result).map_err(|error| SdkError::Transport(error.to_string()))
    }

    async fn lease_snapshot(&self) -> Result<crate::RuntimeLeaseSnapshot, SdkError> {
        let method = self.method(
            crate::FrontendFacadeMethod::Lease,
            "supercode/frontend/lease",
            SdkOperation::Events,
        )?;
        let result = self
            .request(
                method,
                json!({"sessionId":self.session_id}),
                SdkOperation::Events,
            )
            .await?;
        serde_json::from_value(result).map_err(|error| SdkError::Transport(error.to_string()))
    }

    async fn take_control(&self) -> Result<crate::RuntimeLeaseSnapshot, SdkError> {
        let method = self.method(
            crate::FrontendFacadeMethod::TakeControl,
            "supercode/frontend/take_control",
            SdkOperation::Input,
        )?;
        let result = self
            .request(
                method,
                json!({"sessionId":self.session_id}),
                SdkOperation::Input,
            )
            .await?;
        serde_json::from_value(result).map_err(|error| SdkError::Transport(error.to_string()))
    }

    async fn acquire_control(&self) -> Result<crate::RuntimeLeaseSnapshot, SdkError> {
        let method = self.method(
            crate::FrontendFacadeMethod::AcquireControl,
            "supercode/frontend/acquire_control",
            SdkOperation::Input,
        )?;
        let result = self
            .request(
                method,
                json!({"sessionId":self.session_id}),
                SdkOperation::Input,
            )
            .await?;
        serde_json::from_value(result).map_err(|error| SdkError::Transport(error.to_string()))
    }

    async fn heartbeat(&self) -> Result<crate::RuntimeLeaseSnapshot, SdkError> {
        let method = self.method(
            crate::FrontendFacadeMethod::Heartbeat,
            "supercode/frontend/heartbeat",
            SdkOperation::Events,
        )?;
        let result = self
            .request(
                method,
                json!({"sessionId":self.session_id}),
                SdkOperation::Events,
            )
            .await?;
        serde_json::from_value(result).map_err(|error| SdkError::Transport(error.to_string()))
    }

    async fn detach(&self) -> Result<crate::RuntimeLeaseSnapshot, SdkError> {
        let method = self.method(
            crate::FrontendFacadeMethod::Detach,
            "supercode/frontend/detach",
            SdkOperation::Events,
        )?;
        let result = self
            .request(
                method,
                json!({"sessionId":self.session_id}),
                SdkOperation::Events,
            )
            .await?;
        serde_json::from_value(result).map_err(|error| SdkError::Transport(error.to_string()))
    }

    async fn close(&self) -> Result<(), SdkError> {
        let descriptor = self.describe().await?;
        if !descriptor.actions.close {
            return Err(SdkError::UnsupportedAction(
                SdkOperation::Close.action_name(),
            ));
        }
        let method = self.method(
            crate::FrontendFacadeMethod::Close,
            "supercode/frontend/close",
            SdkOperation::Close,
        )?;
        self.request(
            method,
            json!({"sessionId":self.session_id}),
            SdkOperation::Close,
        )
        .await?;
        self.client.close().await.map_err(transport)
    }
}

fn transport(error: impl std::fmt::Display) -> SdkError {
    SdkError::Transport(error.to_string())
}

fn decode_sdk_error(raw: &str, operation: SdkOperation) -> SdkError {
    let value = serde_json::from_str::<Value>(raw).unwrap_or(Value::Null);
    let name = value.get("name").and_then(Value::as_str);
    let message = value
        .get("message")
        .and_then(Value::as_str)
        .unwrap_or(raw)
        .to_string();
    match name {
        Some("unauthenticated") => SdkError::Unauthenticated,
        Some("unauthorized") => SdkError::Unauthorized {
            permission: value
                .get("permission")
                .and_then(Value::as_str)
                .unwrap_or("unknown")
                .to_string(),
        },
        Some("controller_required") => SdkError::ControllerRequired {
            holder: value
                .get("holder")
                .and_then(Value::as_str)
                .map(str::to_owned),
            expires_at_ms: value.get("expiresAtMs").and_then(Value::as_u64),
        },
        Some("lease_expired") => SdkError::LeaseExpired,
        Some("invalid_argument") => SdkError::InvalidArgument { operation, message },
        Some("not_found") => SdkError::NotFound { operation, message },
        Some("busy") => crate::RuntimeSubmitError::Busy.into(),
        Some("unsupported_action") => SdkError::UnsupportedAction(operation.action_name()),
        Some("execution") => SdkError::Execution { operation, message },
        Some("transport") => SdkError::Transport(message),
        _ => SdkError::Transport(message),
    }
}

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

    #[test]
    fn negotiated_methods_only_remove_capabilities() {
        let methods = [
            "supercode/frontend/submit",
            "supercode/frontend/send_input",
            "supercode/frontend/detach",
            "session/respond",
        ]
        .into_iter()
        .map(str::to_owned)
        .collect();
        let mut actions = crate::FrontendActions {
            submit: true,
            interrupt: true,
            steer: true,
            respond: true,
            detach: true,
            close: true,
        };
        mask_actions(&methods, &mut actions);
        assert!(actions.submit);
        assert!(actions.respond);
        assert!(actions.detach);
        assert!(!actions.interrupt);
        assert!(!actions.steer);
        assert!(!actions.close);
    }

    #[test]
    fn named_sdk_errors_survive_acp_envelopes() {
        let error = decode_sdk_error(
            r#"{"name":"busy","operation":"input","message":"occupied"}"#,
            SdkOperation::Input,
        );
        assert_eq!(error.code(), crate::SdkErrorCode::Busy);
        assert_eq!(error.operation(), Some(SdkOperation::Input));
    }

    #[test]
    fn acknowledged_snapshot_history_is_not_rendered_twice() {
        let mut history = vec![crate::ChatMessage::user("already rendered")];
        suppress_acknowledged_history(&mut history, 8, 8);
        assert!(history.is_empty());

        history = vec![crate::ChatMessage::user("must not be lost")];
        suppress_acknowledged_history(&mut history, 8, 7);
        assert_eq!(history.len(), 1);
    }

    #[test]
    fn checkpoint_schema_and_session_are_fail_closed() {
        let valid = AcpFrontendCheckpoint {
            schema_version: 1,
            session_id: "runtime-a".into(),
            acknowledged_sequence: 42,
        };
        assert!(validate_checkpoint(&valid, "runtime-a").is_ok());

        let mut wrong_schema = valid.clone();
        wrong_schema.schema_version = 2;
        assert!(validate_checkpoint(&wrong_schema, "runtime-a")
            .unwrap_err()
            .to_string()
            .contains("schema 2"));

        let mut wrong_session = valid;
        wrong_session.session_id = "runtime-b".into();
        assert!(validate_checkpoint(&wrong_session, "runtime-a")
            .unwrap_err()
            .to_string()
            .contains("runtime-b"));
    }

    #[test]
    fn transient_disconnect_never_advances_the_canonical_cursor() {
        let transient = crate::SdkEvent {
            sequence: 43,
            kind: "runtime_disconnected".into(),
            payload: json!({
                "type":"runtime_disconnected",
                "_meta":{"supercode":{"transient":true,"transport":"acp"}}
            }),
        };
        assert!(!crate::frontend::event_advances_acknowledgement(&transient));

        let canonical = crate::SdkEvent {
            sequence: 43,
            kind: "text_delta".into(),
            payload: json!({"type":"text_delta","text":"next owner event"}),
        };
        assert!(crate::frontend::event_advances_acknowledgement(&canonical));
    }

    #[test]
    fn reconnect_replays_owner_event_after_transient_sequence_collision() {
        let descriptor = || crate::FrontendRuntimeDescriptor {
            schema_version: crate::FRONTEND_RUNTIME_SCHEMA_VERSION,
            session_id: "runtime-a".into(),
            source_harness: None,
            emulation_profile: None,
            active_modules: Vec::new(),
            commands: Vec::new(),
            operations: Vec::new(),
            actions: crate::FrontendActions {
                submit: false,
                interrupt: false,
                steer: false,
                respond: false,
                detach: true,
                close: false,
            },
            display: crate::FrontendDisplayCapabilities {
                event_kinds: Vec::new(),
                opaque_fallback: true,
            },
            model: "test".into(),
            turn_state: crate::FrontendTurnState::Idle,
            connection_state: crate::FrontendConnectionState::Connected,
            extensions: Default::default(),
        };
        let acknowledged = Arc::new(AtomicU64::new(42));
        let transient = crate::SdkEvent {
            sequence: 43,
            kind: "runtime_disconnected".into(),
            payload: json!({
                "type":"runtime_disconnected",
                "_meta":{"supercode":{"transient":true,"transport":"acp"}}
            }),
        };
        let (_first_sender, first_live) = broadcast::channel(1);
        let mut first = FrontendAttachment::from_snapshot_after(
            FrontendAttachSnapshot {
                descriptor: descriptor(),
                history: Vec::new(),
                history_cursor: 0,
                replay: std::collections::VecDeque::from([transient]),
            },
            first_live,
            42,
        )
        .with_acknowledgement(acknowledged.clone());
        assert_eq!(first.next_replay_event().unwrap().sequence, 43);
        assert_eq!(acknowledged.load(Ordering::SeqCst), 42);
        drop(first);

        let canonical = crate::SdkEvent {
            sequence: 43,
            kind: "text_delta".into(),
            payload: json!({"type":"text_delta","text":"next owner event"}),
        };
        let (_second_sender, second_live) = broadcast::channel(1);
        let mut second = FrontendAttachment::from_snapshot_after(
            FrontendAttachSnapshot {
                descriptor: descriptor(),
                history: Vec::new(),
                history_cursor: 0,
                replay: std::collections::VecDeque::from([canonical]),
            },
            second_live,
            acknowledged.load(Ordering::SeqCst),
        )
        .with_acknowledgement(acknowledged.clone());
        assert_eq!(second.next_replay_event().unwrap().sequence, 43);
        assert_eq!(acknowledged.load(Ordering::SeqCst), 43);
    }
}