supercode-cli 0.4.6

use std::collections::BTreeMap;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use serde_json::json;
use supercode::reduce::{export_session, invert, project_messages, verify_log, ReductionPolicy};
use supercode::sidecar::SidecarWriter;
use supercode::store::SessionStore;
use supercode::{
    Agent, ChatMessage, ChatRequest, Config, CoordinatedRuntime, FrontendAttachment, FrontendEvent,
    FrontendRuntime, FrontendRuntimeError, HttpFrontendRuntime, Provider, RpcEngine,
    RuntimeAuthorization, RuntimeClientId, RuntimeSubmitError, SdkOperation, Session,
    SessionFormat, Usage,
};
use supercode_codex_frontend::{CodexAppServerAdapter, CodexEndpoint, CodexEndpointKind};
use supercode_frontend_tui::runtime::TerminalRuntimeView;
use supercode_opencode_frontend::{OpenCodeAdapter, OpenCodeRequest, ResponseBody};
use tokio::io::{
    AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader,
};

const SESSION_ID: &str = "sup60-cross-surface";
const SESSION_NAME: &str = "cross-surface";
const PROMPT: &str = "continue the same reduced session";
const REPLY: &str = "identical continuation reply";
const FIXED_SIDECAR_TIMESTAMP: &str = "2026-01-01T00:00:00.000Z";

struct BlockingProvider {
    entered: tokio::sync::Notify,
    release: tokio::sync::Notify,
    actions: Mutex<Vec<SdkActionRecord>>,
}

#[derive(Debug, Clone, PartialEq)]
struct SdkActionRecord {
    operation: SdkOperation,
    prompt: String,
    request_messages: Vec<ChatMessage>,
}

impl BlockingProvider {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            entered: tokio::sync::Notify::new(),
            release: tokio::sync::Notify::new(),
            actions: Mutex::new(Vec::new()),
        })
    }
}

struct SharedProvider(Arc<BlockingProvider>);

#[async_trait]
impl Provider for SharedProvider {
    async fn complete(
        &self,
        request: &ChatRequest,
        on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode::Result<(ChatMessage, Usage)> {
        let prompt = request
            .messages
            .last()
            .and_then(|message| message.content.clone())
            .expect("accepted SDK input must reach the provider as a user prompt");
        self.0
            .actions
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .push(SdkActionRecord {
                operation: SdkOperation::Input,
                prompt,
                request_messages: request.messages.clone(),
            });
        self.0.entered.notify_one();
        self.0.release.notified().await;
        on_delta(REPLY);
        Ok((ChatMessage::assistant(REPLY), Usage::default()))
    }
}

#[derive(Clone, Default)]
struct ImageRequestCapture(Arc<Mutex<Vec<ChatRequest>>>);

struct ImageCaptureProvider(ImageRequestCapture);

#[async_trait]
impl Provider for ImageCaptureProvider {
    async fn complete(
        &self,
        request: &ChatRequest,
        on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode::Result<(ChatMessage, Usage)> {
        self.0
             .0
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .push(request.clone());
        on_delta("multimodal-ok");
        Ok((ChatMessage::assistant("multimodal-ok"), Usage::default()))
    }
}

fn image_runtime(surface: &str) -> (PathBuf, Arc<RpcEngine>, ImageRequestCapture) {
    let root = std::env::temp_dir().join(format!(
        "supercode-sup62-multimodal-{surface}-{}",
        std::process::id()
    ));
    std::fs::remove_dir_all(&root).ok();
    std::fs::create_dir_all(&root).unwrap();
    let capture = ImageRequestCapture::default();
    let agent = Agent::with_provider(
        Config::builder().cwd(&root).build(),
        Box::new(ImageCaptureProvider(capture.clone())),
    );
    (
        root,
        RpcEngine::new_named(agent, format!("sup62-multimodal-{surface}"), None),
        capture,
    )
}

fn assert_exact_multimodal_request(
    surface: &str,
    capture: &ImageRequestCapture,
    prompt: &str,
    image_urls: &[String],
) {
    let requests = capture
        .0
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    assert_eq!(requests.len(), 1, "{surface} provider request count");
    assert_eq!(
        requests[0].messages.last(),
        Some(&ChatMessage::user_with_images(prompt, image_urls)),
        "{surface} changed the canonical multimodal payload"
    );
}

fn source_session() -> Session {
    let duplicate = (0..640)
        .map(|index| (b'a' + (index % 26) as u8) as char)
        .collect::<String>();
    let session_id = "11111111-1111-4111-8111-111111111111";
    let records = [
        json!({"type":"user","sessionId":session_id,"cwd":"/workspace","timestamp":"2020-01-01T00:00:00Z","message":{"role":"user","content":"first imported run"}}),
        json!({"type":"assistant","sessionId":session_id,"cwd":"/workspace","timestamp":"2020-01-01T00:00:01Z","message":{"role":"assistant","content":[{"type":"tool_use","id":"call-1","name":"bash","input":{"command":"cargo test"}}]}}),
        json!({"type":"user","sessionId":session_id,"cwd":"/workspace","timestamp":"2020-01-01T00:00:02Z","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"call-1","content":duplicate.clone()}]}}),
        json!({"type":"user","sessionId":session_id,"cwd":"/workspace","timestamp":"2020-01-01T00:00:03Z","message":{"role":"user","content":"second imported run"}}),
        json!({"type":"assistant","sessionId":session_id,"cwd":"/workspace","timestamp":"2020-01-01T00:00:04Z","message":{"role":"assistant","content":[{"type":"tool_use","id":"call-2","name":"bash","input":{"command":"cargo test"}}]}}),
        json!({"type":"user","sessionId":session_id,"cwd":"/workspace","timestamp":"2020-01-01T00:00:05Z","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"call-2","content":duplicate}]}}),
        json!({"type":"assistant","sessionId":session_id,"cwd":"/workspace","timestamp":"2020-01-01T00:00:06Z","message":{"role":"assistant","content":"imported work complete"}}),
    ];
    let raw = records
        .iter()
        .map(serde_json::Value::to_string)
        .collect::<Vec<_>>()
        .join("\n")
        + "\n";
    let session = Session::from_claude_code_str(&raw).unwrap();
    assert_eq!(session.messages.len(), 7);
    assert_eq!(
        session.messages[0].content.as_deref(),
        Some("first imported run")
    );
    assert_eq!(session.messages[1].tool_calls()[0].id, "call-1");
    assert_eq!(
        session.messages[2].content.as_deref(),
        Some(duplicate.as_str())
    );
    assert_eq!(session.messages[4].tool_calls()[0].id, "call-2");
    assert_eq!(
        session.messages[5].content.as_deref(),
        Some(duplicate.as_str())
    );
    session
}

fn reduction_policy() -> ReductionPolicy {
    ReductionPolicy {
        protect_last_n_tool_results: 0,
        supersede_enabled: false,
        ..ReductionPolicy::default()
    }
}

struct Harness {
    root: PathBuf,
    store_root: PathBuf,
    provider: Arc<BlockingProvider>,
    engine: Arc<RpcEngine>,
}

fn harness(surface: &str) -> Harness {
    let root = std::env::temp_dir().join(format!(
        "supercode-sup60-cross-surface-{surface}-{}",
        std::process::id()
    ));
    std::fs::remove_dir_all(&root).ok();
    std::fs::create_dir_all(&root).unwrap();
    let store_root = root.join("store");
    let store = SessionStore::open(&store_root).unwrap();
    let provider = BlockingProvider::new();
    let mut agent = Agent::with_provider(
        Config::builder()
            .cwd(&root)
            .system_prompt("SUP-60 cross-surface conformance")
            .build(),
        Box::new(SharedProvider(provider.clone())),
    );
    let source = source_session();
    let recorder = SidecarWriter::create_with_timestamp(
        &store.sidecar_path(SESSION_NAME),
        &source,
        FIXED_SIDECAR_TIMESTAMP,
    )
    .unwrap();
    agent.load_session(source);
    agent.set_recorder(recorder);
    agent.set_reduction_policy(reduction_policy());

    let persistence_root = store_root.clone();
    let engine = RpcEngine::new_named(
        agent,
        SESSION_ID,
        Some(Box::new(move |agent: &supercode::SdkAgent| {
            let store = SessionStore::open(&persistence_root).unwrap();
            let transcript = agent
                .history()
                .iter()
                .map(|message| serde_json::to_string(message).unwrap())
                .collect::<Vec<_>>()
                .join("\n")
                + "\n";
            store
                .save(SESSION_NAME, "cross-surface", &transcript)
                .unwrap();
            store
                .save_reduction_log(SESSION_NAME, agent.reduction_log())
                .unwrap();
        })),
    );
    Harness {
        root,
        store_root,
        provider,
        engine,
    }
}

#[derive(Debug, PartialEq)]
struct Observation {
    session_id: String,
    sdk_actions: Vec<SdkActionRecord>,
    history_cursor: u64,
    events: Vec<FrontendEvent>,
    history: Vec<ChatMessage>,
    transcript: Vec<u8>,
    sidecar: Vec<u8>,
    reduction_log: Vec<u8>,
    claude_export: Vec<u8>,
}

async fn observe_turn(
    mut observer: FrontendAttachment,
    provider: &BlockingProvider,
) -> Vec<FrontendEvent> {
    provider.entered.notified().await;
    provider.release.notify_waiters();
    collect_turn_events(&mut observer).await
}

async fn collect_turn_events(observer: &mut FrontendAttachment) -> Vec<FrontendEvent> {
    let mut events = Vec::new();
    loop {
        let event = tokio::time::timeout(std::time::Duration::from_secs(5), observer.next_event())
            .await
            .expect("surface continuation timed out")
            .unwrap();
        let terminal = event.kind == "turn_succeeded";
        events.push(event);
        if terminal {
            break;
        }
    }
    events
}

async fn read_observation(harness: &Harness, events: Vec<FrontendEvent>) -> Observation {
    let store = SessionStore::open(&harness.store_root).unwrap();
    let transcript = store.load(SESSION_NAME).unwrap().into_bytes();
    let sidecar = store.load_sidecar(SESSION_NAME).unwrap().unwrap();
    let log = store.load_reduction_log(SESSION_NAME).unwrap().unwrap();
    assert_eq!(log.reductions.len(), 1, "fixture must mint one reduction");
    let sidecar_session = Session::from_sidecar_str(&sidecar).unwrap();
    verify_log(&log, &sidecar_session).unwrap();
    let (view, _) = project_messages(&sidecar_session.messages, &reduction_policy(), &log);
    assert_eq!(
        invert(&view, &log, &sidecar_session).unwrap(),
        sidecar_session.messages
    );
    let claude_export = export_session(&sidecar, SessionFormat::ClaudeCode)
        .unwrap()
        .into_bytes();
    let attachment = harness.engine.attach(100).await.unwrap();
    let sdk_actions = harness
        .provider
        .actions
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .clone();
    Observation {
        session_id: SESSION_ID.into(),
        sdk_actions,
        history_cursor: attachment.history_cursor,
        events,
        history: attachment.history,
        transcript,
        sidecar: normalize_sidecar_framing(&sidecar),
        reduction_log: serde_json::to_vec(&log).unwrap(),
        claude_export,
    }
}

fn normalize_sidecar_framing(sidecar: &str) -> Vec<u8> {
    // Independent executions necessarily have different wall-clock framing.
    // Remove only the native-v2 header/append timestamps; every canonical
    // field, imported raw record, message, tool call, and byte of content
    // remains in the equality comparison below.
    let mut output = String::new();
    for line in sidecar.lines() {
        let mut value: serde_json::Value = serde_json::from_str(line).unwrap();
        if value.get("supercode_native").is_some() {
            value.as_object_mut().unwrap().remove("created");
        }
        if value.get("supercode_turn").is_some() {
            value.as_object_mut().unwrap().remove("ts");
        }
        output.push_str(&value.to_string());
        output.push('\n');
    }
    output.into_bytes()
}

async fn embedded_observation() -> Observation {
    let harness = harness("embedded");
    let observer = harness.engine.attach(100).await.unwrap();
    let view = TerminalRuntimeView::attach(harness.engine.clone(), 100)
        .await
        .unwrap();
    assert_eq!(view.descriptor().session_id, SESSION_ID);
    FrontendRuntime::send_input(view.controller(), PROMPT.into())
        .await
        .unwrap();
    assert!(matches!(
        FrontendRuntime::send_input(view.controller(), "competing".into()).await,
        Err(FrontendRuntimeError::Submit(RuntimeSubmitError::Busy))
    ));
    let event_kinds = observe_turn(observer, &harness.provider).await;
    let observation = read_observation(&harness, event_kinds).await;
    harness.engine.shutdown().await;
    std::fs::remove_dir_all(harness.root).ok();
    observation
}

async fn http_observation() -> Observation {
    let harness = harness("http");
    let observer = harness.engine.attach(100).await.unwrap();
    let token: Arc<str> = "sup60-http-token".into();
    let address = supercode::server::run_http(harness.engine.clone(), "127.0.0.1:0", token.clone())
        .await
        .unwrap();
    let runtime = HttpFrontendRuntime::connect(format!("http://{address}"), token.to_string())
        .await
        .unwrap();
    assert_eq!(runtime.describe().await.unwrap().session_id, SESSION_ID);
    FrontendRuntime::send_input(runtime.clone(), PROMPT.into())
        .await
        .unwrap();
    assert!(matches!(
        FrontendRuntime::send_input(runtime.clone(), "competing".into()).await,
        Err(FrontendRuntimeError::Submit(RuntimeSubmitError::Busy))
    ));
    let event_kinds = observe_turn(observer, &harness.provider).await;
    runtime.detach().await.unwrap();
    let observation = read_observation(&harness, event_kinds).await;
    harness.engine.shutdown().await;
    std::fs::remove_dir_all(harness.root).ok();
    observation
}

async fn pi_observation() -> Observation {
    use std::process::Stdio;

    let driver = std::env::var_os("SUPERCODE_PI_CONFORMANCE_DRIVER")
        .expect("SUPERCODE_PI_CONFORMANCE_DRIVER must name the built Pi conformance driver");
    let node = std::env::var_os("SUPERCODE_PI_NODE")
        .expect("SUPERCODE_PI_NODE must name the exact Node 22.19+ executable");
    let harness = harness("pi");
    let observer = harness.engine.attach(100).await.unwrap();
    let expected_history = observer.history.clone();
    let token = "sup65-pi-conformance-token";
    let address = supercode::server::run_http(
        harness.engine.clone(),
        "127.0.0.1:0",
        Arc::<str>::from(token),
    )
    .await
    .unwrap();
    let mut child = tokio::process::Command::new(node)
        .arg(driver)
        .current_dir(&harness.root)
        .env_clear()
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .kill_on_drop(true)
        .spawn()
        .expect("Node 22.19+ must launch the built Pi conformance driver");
    let input = json!({
        "baseUrl":format!("http://{address}"),
        "token":token,
        "prompt":PROMPT,
        "competingPrompt":"competing",
    });
    child
        .stdin
        .take()
        .unwrap()
        .write_all(input.to_string().as_bytes())
        .await
        .unwrap();
    let mut stdout = BufReader::new(child.stdout.take().unwrap());
    harness.provider.entered.notified().await;
    let mut ready_line = String::new();
    tokio::time::timeout(
        std::time::Duration::from_secs(5),
        stdout.read_line(&mut ready_line),
    )
    .await
    .expect("Pi conformance driver did not check the busy boundary")
    .unwrap();
    let ready: serde_json::Value = serde_json::from_str(&ready_line).unwrap();
    assert_eq!(ready["stage"], "busy_checked");
    let idle_actions = json!({
        "submit":true,
        "steer":false,
        "interrupt":false,
        "respond":false,
        "detach":true,
        "close":false,
        "takeover":true,
    });
    let busy_actions = json!({
        "submit":false,
        "steer":true,
        "interrupt":true,
        "respond":false,
        "detach":true,
        "close":false,
        "takeover":true,
    });
    assert_eq!(ready["idleActions"], idle_actions);
    assert_eq!(ready["busyActions"], busy_actions);
    assert_eq!(
        ready["busyActionError"],
        "frontend action submit is unavailable"
    );
    assert_eq!(ready["busyTransportError"], "busy");
    harness.provider.release.notify_waiters();
    let mut observer = observer;
    let event_kinds = collect_turn_events(&mut observer).await;
    let mut receipt_line = String::new();
    tokio::time::timeout(
        std::time::Duration::from_secs(10),
        stdout.read_line(&mut receipt_line),
    )
    .await
    .expect("Pi conformance driver did not detach")
    .unwrap();
    let status = tokio::time::timeout(std::time::Duration::from_secs(5), child.wait())
        .await
        .expect("Pi conformance driver did not detach")
        .unwrap();
    let mut stderr = String::new();
    child
        .stderr
        .take()
        .unwrap()
        .read_to_string(&mut stderr)
        .await
        .unwrap();
    assert!(status.success(), "Pi conformance driver failed: {stderr}");
    let receipt: serde_json::Value = serde_json::from_str(&receipt_line).unwrap();
    assert_eq!(receipt["sessionId"], SESSION_ID);
    let projected_history: Vec<ChatMessage> =
        serde_json::from_value(receipt["history"].clone()).unwrap();
    let project_display_history = |mut messages: Vec<ChatMessage>| {
        for message in &mut messages {
            message.metadata.clear();
        }
        messages
    };
    assert_eq!(projected_history, project_display_history(expected_history));
    assert!(projected_history.ends_with(&project_display_history(source_session().messages)));
    assert_eq!(receipt["historyCursor"], 0);
    assert_eq!(receipt["cursor"], 6);
    assert_eq!(
        receipt["events"],
        json!([
            [1, "user_message"],
            [2, "turn_started"],
            [3, "text_delta"],
            [4, "usage"],
            [5, "turn_completed"],
            [6, "turn_succeeded"],
        ])
    );
    assert_eq!(receipt["completedActions"], idle_actions);
    assert_eq!(receipt["detachedConnection"], "disconnected");
    assert_eq!(
        receipt["leaseAfterDetach"]["controller"],
        serde_json::Value::Null
    );
    assert_eq!(receipt["leaseAfterDetach"]["observers"], json!([]));
    let rendered = format!("{ready_line}{receipt_line}{stderr}");
    assert!(!rendered.contains(token));
    let observation = read_observation(&harness, event_kinds).await;
    harness.engine.shutdown().await;
    std::fs::remove_dir_all(harness.root).ok();
    observation
}

async fn acp_observation() -> Observation {
    let harness = harness("acp");
    let observer = harness.engine.attach(100).await.unwrap();
    let token = "sup60-acp-token";
    let address = supercode::server::run_http(
        harness.engine.clone(),
        "127.0.0.1:0",
        Arc::<str>::from(token),
    )
    .await
    .unwrap();
    let mut env = BTreeMap::new();
    env.insert("SUPERCODE_SERVER_TOKEN".into(), token.into());
    let runtime = supercode::AcpFrontendRuntime::connect(supercode::AcpFrontendConnectOptions {
        launch: supercode::RuntimeLaunch {
            program: env!("CARGO_BIN_EXE_supercode").into(),
            arguments: vec![
                "acp".into(),
                "--connect".into(),
                format!("http://{address}"),
            ],
            env,
        },
        cwd: Some(harness.root.clone()),
        session_id: None,
        after_sequence: None,
    })
    .await
    .unwrap();
    assert_eq!(runtime.session_id(), SESSION_ID);
    FrontendRuntime::send_input(runtime.clone(), PROMPT.into())
        .await
        .unwrap();
    assert!(matches!(
        FrontendRuntime::send_input(runtime.clone(), "competing".into()).await,
        Err(FrontendRuntimeError::Submit(RuntimeSubmitError::Busy))
    ));
    let event_kinds = observe_turn(observer, &harness.provider).await;
    runtime.detach().await.unwrap();
    let observation = read_observation(&harness, event_kinds).await;
    harness.engine.shutdown().await;
    std::fs::remove_dir_all(harness.root).ok();
    observation
}

async fn write_acp_message<W: AsyncWrite + Unpin>(writer: &mut W, message: serde_json::Value) {
    writer
        .write_all(format!("{message}\n").as_bytes())
        .await
        .unwrap();
    writer.flush().await.unwrap();
}

async fn read_acp_message<R: AsyncBufRead + Unpin>(reader: &mut R) -> serde_json::Value {
    let mut line = String::new();
    let bytes = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        reader.read_line(&mut line),
    )
    .await
    .expect("Goose ACP response timed out")
    .unwrap();
    assert!(bytes > 0, "Goose ACP stream closed before its response");
    serde_json::from_str(&line).unwrap()
}

async fn goose_observation() -> Observation {
    use supercode::acp_server::{self, AcpCompatibilityProfile, AcpServer};

    let harness = harness("goose");
    let observer = harness.engine.attach(100).await.unwrap();
    let server = AcpServer::new_with_profile(
        harness.engine.clone(),
        AcpCompatibilityProfile::GooseAcd3c135,
    )
    .await
    .unwrap();
    let (server_io, client_io) = tokio::io::duplex(128 * 1024);
    let (server_read, server_write) = tokio::io::split(server_io);
    let (client_read, mut client_write) = tokio::io::split(client_io);
    let server_task = tokio::spawn(acp_server::run_stdio(
        server,
        BufReader::new(server_read),
        server_write,
    ));
    let mut client_read = BufReader::new(client_read);

    write_acp_message(
        &mut client_write,
        json!({
            "jsonrpc":"2.0", "id":1, "method":"initialize",
            "params":{
                "protocolVersion":1,
                "clientInfo":{"name":"goose-text","version":"0.1.0"},
                "clientCapabilities":{}
            }
        }),
    )
    .await;
    let initialized = read_acp_message(&mut client_read).await;
    assert_eq!(initialized["result"]["agentInfo"]["name"], "supercode");

    write_acp_message(
        &mut client_write,
        json!({
            "jsonrpc":"2.0", "id":2,
            "method":"_goose/unstable/defaults/read", "params":{}
        }),
    )
    .await;
    let defaults = read_acp_message(&mut client_read).await;
    assert_eq!(defaults["result"]["providerId"], "supercode");

    write_acp_message(
        &mut client_write,
        json!({
            "jsonrpc":"2.0", "id":3, "method":"session/new",
            "params":{"cwd":"/client-only/path", "mcpServers":[]}
        }),
    )
    .await;
    let mut history_updates = Vec::new();
    loop {
        let message = read_acp_message(&mut client_read).await;
        if message.get("id") == Some(&json!(3)) {
            assert_eq!(message["result"]["sessionId"], SESSION_ID);
            break;
        }
        history_updates.push(message);
    }
    assert!(history_updates.iter().any(|message| {
        message
            .pointer("/params/update/content/text")
            .and_then(serde_json::Value::as_str)
            == Some("first imported run")
    }));
    assert!(history_updates.iter().any(|message| {
        message
            .pointer("/params/update/toolCallId")
            .and_then(serde_json::Value::as_str)
            == Some("call-2")
    }));

    write_acp_message(
        &mut client_write,
        json!({
            "jsonrpc":"2.0", "id":4, "method":"session/prompt",
            "params":{
                "sessionId":SESSION_ID,
                "prompt":[{"type":"text", "text":PROMPT}]
            }
        }),
    )
    .await;
    write_acp_message(
        &mut client_write,
        json!({
            "jsonrpc":"2.0", "id":5, "method":"session/prompt",
            "params":{
                "sessionId":SESSION_ID,
                "prompt":[{"type":"text", "text":"competing"}]
            }
        }),
    )
    .await;
    let event_kinds = observe_turn(observer, &harness.provider).await;

    let mut completed = false;
    let mut rejected_busy = false;
    while !completed || !rejected_busy {
        let message = read_acp_message(&mut client_read).await;
        match message.get("id").and_then(serde_json::Value::as_u64) {
            Some(4) => {
                assert_eq!(message["result"]["stopReason"], "end_turn");
                completed = true;
            }
            Some(5) => {
                assert_eq!(message["error"]["name"], "busy");
                rejected_busy = true;
            }
            _ => {}
        }
    }

    client_write.shutdown().await.unwrap();
    tokio::time::timeout(std::time::Duration::from_secs(2), server_task)
        .await
        .expect("Goose ACP projection did not detach")
        .unwrap()
        .unwrap();
    let observation = read_observation(&harness, event_kinds).await;
    harness.engine.shutdown().await;
    std::fs::remove_dir_all(harness.root).ok();
    observation
}

async fn codex_observation() -> Observation {
    let harness = harness("codex");
    let observer = harness.engine.attach(100).await.unwrap();
    let coordinated = CoordinatedRuntime::new(harness.engine.clone());
    let runtime = coordinated.client(
        RuntimeClientId::parse("stock-codex-conformance").unwrap(),
        RuntimeAuthorization::owner(),
    );
    let adapter = CodexAppServerAdapter::new(runtime, &harness.root);
    let mut connection = adapter.connection();
    connection
        .handle(json!({"id":1,"method":"initialize","params":{"clientInfo":{"version":"0.144.4"}}}))
        .await;
    connection.handle(json!({"method":"initialized"})).await;
    let started = connection
        .handle(json!({"id":2,"method":"thread/start","params":{}}))
        .await;
    let thread = started[0]["result"]["thread"]["id"]
        .as_str()
        .unwrap()
        .to_owned();
    assert!(started[0]["result"]["thread"]
        .to_string()
        .contains("first imported run"));
    connection
        .handle(json!({"id":3,"method":"turn/start","params":{"threadId":thread,"input":[{"type":"text","text":PROMPT}]}}))
        .await;
    let competing = connection
        .handle(json!({"id":4,"method":"turn/start","params":{"threadId":thread,"input":[{"type":"text","text":"competing"}]}}))
        .await;
    assert_eq!(competing[0]["error"]["data"]["name"], "busy");
    let event_kinds = observe_turn(observer, &harness.provider).await;
    let detached = connection
        .handle(json!({"id":5,"method":"thread/unsubscribe","params":{"threadId":thread}}))
        .await;
    assert_eq!(detached[0]["result"]["status"], "unsubscribed");
    let observation = read_observation(&harness, event_kinds).await;
    harness.engine.shutdown().await;
    std::fs::remove_dir_all(harness.root).ok();
    observation
}

async fn opencode_observation() -> Observation {
    let harness = harness("opencode");
    let observer = harness.engine.attach(100).await.unwrap();
    let coordinated = CoordinatedRuntime::new(harness.engine.clone());
    let runtime = coordinated.client(
        RuntimeClientId::parse("stock-opencode-conformance").unwrap(),
        RuntimeAuthorization::owner(),
    );
    let adapter = OpenCodeAdapter::new(runtime.clone(), SESSION_ID, &harness.root);
    let session_id = adapter.session_id().to_owned();
    let request =
        OpenCodeRequest::new("POST", format!("/session/{session_id}/message")).with_body(json!({
            "messageID": "msg_sup62_conformance",
            "agent": "build",
            "model": {
                "providerID": "anthropic",
                "modelID": "claude-opus-4-8"
            },
            "parts": [{
                "id": "prt_sup62_conformance",
                "type": "text",
                "text": PROMPT
            }]
        }));
    let turn_adapter = adapter.clone();
    let turn = tokio::spawn(async move { turn_adapter.handle(request).await });
    let event_kinds = observe_turn(observer, &harness.provider).await;
    let response = turn.await.unwrap();
    assert_eq!(response.status, 200);
    let body = match response.body {
        ResponseBody::Json(body) => body,
        ResponseBody::EventStream(_) => panic!("message submission returned an event stream"),
    };
    assert_eq!(body["info"]["role"], "assistant");
    assert_eq!(body["parts"][0]["text"], REPLY);
    runtime.detach();
    let observation = read_observation(&harness, event_kinds).await;
    harness.engine.shutdown().await;
    std::fs::remove_dir_all(harness.root).ok();
    observation
}

#[cfg(unix)]
fn kill_stock_process_group(child: &mut tokio::process::Child) {
    if let Some(pid) = child.id() {
        // SAFETY: each proof launch uses `process_group(0)`, so its pgid is
        // the direct child's pid. A negative pid targets the entire isolated
        // launcher/native-client group and cannot hit the test process.
        unsafe {
            libc::kill(-(pid as i32), libc::SIGKILL);
        }
    }
    child.start_kill().ok();
}

#[cfg(unix)]
async fn stock_codex_binary_observation() -> Observation {
    use std::io::Read;
    use std::os::fd::FromRawFd;
    use std::process::Stdio;

    let codex = std::env::var_os("SUPERCODE_STOCK_CODEX_BIN")
        .expect("SUPERCODE_STOCK_CODEX_BIN must name the pinned unmodified Codex binary");
    let version = std::process::Command::new(&codex)
        .arg("--version")
        .output()
        .unwrap();
    assert!(version.status.success());
    assert_eq!(
        String::from_utf8_lossy(&version.stdout).trim(),
        "codex-cli 0.144.4"
    );

    let harness = harness("stock-codex-binary");
    let client_home = harness.root.join("codex-home");
    std::fs::create_dir(&client_home).unwrap();
    std::fs::write(
        client_home.join("config.toml"),
        r#"model_provider = "supercode_frontend"

[model_providers.supercode_frontend]
name = "Supercode SDK runtime"
base_url = "http://127.0.0.1:1/unused"
wire_api = "responses"
requires_openai_auth = false
request_max_retries = 0
stream_max_retries = 0
"#,
    )
    .unwrap();

    let observer = harness.engine.attach(100).await.unwrap();
    let (endpoint, bootstrap) =
        CodexEndpoint::new(harness.engine.clone(), SESSION_ID, &harness.root).unwrap();
    let endpoint = endpoint.with_client_home(&client_home).unwrap();
    let mut credential = endpoint
        .issue_interactive(&bootstrap, "stock-codex-conformance")
        .unwrap();
    let handle = endpoint.bind_websocket(0).await.unwrap();
    let address = match handle.kind() {
        CodexEndpointKind::WebSocket(address) => *address,
        #[cfg(unix)]
        CodexEndpointKind::Unix(_) => unreachable!(),
    };

    let mut master = -1;
    let mut slave = -1;
    let dimensions = libc::winsize {
        ws_row: 40,
        ws_col: 120,
        ws_xpixel: 0,
        ws_ypixel: 0,
    };
    // BSD libc declarations expose this input-only argument as mutable even
    // though openpty reads the requested dimensions. Use one raw pointer so
    // the proof builds under both BSD and Linux signatures without an
    // unnecessary mutable reference on Linux.
    let dimensions_ptr = std::ptr::addr_of!(dimensions).cast_mut();
    // SAFETY: all pointers reference initialized writable storage for the
    // duration of the call; no termios override is requested.
    assert_eq!(
        unsafe {
            libc::openpty(
                &mut master,
                &mut slave,
                std::ptr::null_mut(),
                std::ptr::null_mut(),
                dimensions_ptr,
            )
        },
        0,
        "opening stock-client proof PTY failed: {}",
        std::io::Error::last_os_error()
    );
    // Duplicate the slave before transferring each descriptor into a unique
    // owning File. The master is drained concurrently so a verbose TUI can
    // never block the semantic proof on PTY backpressure.
    let stdout_fd = unsafe { libc::dup(slave) };
    let stderr_fd = unsafe { libc::dup(slave) };
    assert!(stdout_fd >= 0 && stderr_fd >= 0);
    let mut master_file = unsafe { std::fs::File::from_raw_fd(master) };
    let stdin_file = unsafe { std::fs::File::from_raw_fd(slave) };
    let stdout_file = unsafe { std::fs::File::from_raw_fd(stdout_fd) };
    let stderr_file = unsafe { std::fs::File::from_raw_fd(stderr_fd) };
    let capture = tokio::task::spawn_blocking(move || {
        let mut bytes = Vec::new();
        let _ = master_file.read_to_end(&mut bytes);
        bytes
    });

    let token_env = "SUPERCODE_STOCK_CODEX_TEST_TOKEN";
    let mut command = tokio::process::Command::new(&codex);
    command
        .current_dir(&harness.root)
        .arg("--remote")
        .arg(format!("ws://{address}"))
        .arg("--remote-auth-token-env")
        .arg(token_env)
        .arg("--disable")
        .arg("plugins")
        .arg("--no-alt-screen")
        .arg(PROMPT)
        .env("CODEX_HOME", &client_home)
        .env("TERM", "xterm-256color")
        .env_remove(token_env)
        .stdin(Stdio::from(stdin_file))
        .stdout(Stdio::from(stdout_file))
        .stderr(Stdio::from(stderr_file))
        .process_group(0);
    for variable in [
        "OPENAI_API_KEY",
        "OPENROUTER_API_KEY",
        "ANTHROPIC_API_KEY",
        "SUPERCODE_API_KEY",
        "SUPERCODE_SERVER_TOKEN",
    ] {
        command.env_remove(variable);
    }
    let mut child = credential
        .spawn_tokio_child(&mut command, token_env)
        .unwrap();
    // `Command` is intentionally reusable and may retain its configured PTY
    // handles after spawn. Drop it now so the master observes EOF when the
    // one real child exits.
    drop(command);

    let event_kinds = tokio::time::timeout(
        std::time::Duration::from_secs(30),
        observe_turn(observer, &harness.provider),
    )
    .await
    .expect("unmodified stock Codex did not complete the deterministic turn");
    tokio::time::sleep(std::time::Duration::from_millis(250)).await;
    kill_stock_process_group(&mut child);
    tokio::time::timeout(std::time::Duration::from_secs(5), child.wait())
        .await
        .expect("stock Codex did not exit after proof completion")
        .unwrap();

    let capture = capture.await.unwrap();
    let capture = String::from_utf8_lossy(&capture);
    assert!(
        capture.contains("\u{1b}[") || capture.contains("OpenAI Codex"),
        "the stock binary emitted no TUI output: {capture:?}"
    );

    // Re-launch the unmodified client against the same live endpoint and
    // still-valid scoped credential. This is a genuine frontend replacement,
    // not a second runtime: no model turn is submitted and canonical bytes
    // below must remain identical to the other surfaces' one-turn fixture.
    tokio::time::sleep(std::time::Duration::from_millis(250)).await;
    let mut second_master = -1;
    let mut second_slave = -1;
    let second_dimensions = libc::winsize {
        ws_row: 40,
        ws_col: 120,
        ws_xpixel: 0,
        ws_ypixel: 0,
    };
    let second_dimensions_ptr = std::ptr::addr_of!(second_dimensions).cast_mut();
    assert_eq!(
        unsafe {
            libc::openpty(
                &mut second_master,
                &mut second_slave,
                std::ptr::null_mut(),
                std::ptr::null_mut(),
                second_dimensions_ptr,
            )
        },
        0
    );
    let second_stdout_fd = unsafe { libc::dup(second_slave) };
    let second_stderr_fd = unsafe { libc::dup(second_slave) };
    assert!(second_stdout_fd >= 0 && second_stderr_fd >= 0);
    let mut second_master_file = unsafe { std::fs::File::from_raw_fd(second_master) };
    let second_stdin_file = unsafe { std::fs::File::from_raw_fd(second_slave) };
    let second_stdout_file = unsafe { std::fs::File::from_raw_fd(second_stdout_fd) };
    let second_stderr_file = unsafe { std::fs::File::from_raw_fd(second_stderr_fd) };
    let second_capture = tokio::task::spawn_blocking(move || {
        let mut bytes = Vec::new();
        let _ = second_master_file.read_to_end(&mut bytes);
        bytes
    });
    let mut second_command = tokio::process::Command::new(&codex);
    second_command
        .current_dir(&harness.root)
        .arg("--remote")
        .arg(format!("ws://{address}"))
        .arg("--remote-auth-token-env")
        .arg(token_env)
        .arg("--disable")
        .arg("plugins")
        .arg("--no-alt-screen")
        .env("CODEX_HOME", &client_home)
        .env("TERM", "xterm-256color")
        .env_remove(token_env)
        .stdin(Stdio::from(second_stdin_file))
        .stdout(Stdio::from(second_stdout_file))
        .stderr(Stdio::from(second_stderr_file))
        .process_group(0);
    for variable in [
        "OPENAI_API_KEY",
        "OPENROUTER_API_KEY",
        "ANTHROPIC_API_KEY",
        "SUPERCODE_API_KEY",
        "SUPERCODE_SERVER_TOKEN",
    ] {
        second_command.env_remove(variable);
    }
    let mut second_child = credential
        .spawn_tokio_child(&mut second_command, token_env)
        .unwrap();
    drop(second_command);
    tokio::time::sleep(std::time::Duration::from_millis(750)).await;
    assert!(
        second_child.try_wait().unwrap().is_none(),
        "the replacement stock client exited instead of reattaching"
    );
    kill_stock_process_group(&mut second_child);
    tokio::time::timeout(std::time::Duration::from_secs(5), second_child.wait())
        .await
        .expect("replacement stock Codex did not exit")
        .unwrap();
    let second_capture = String::from_utf8_lossy(&second_capture.await.unwrap()).into_owned();
    for forbidden in [
        "busy",
        "unauthorized",
        "transport failed",
        "protocol version mismatch",
    ] {
        assert!(
            !second_capture.to_ascii_lowercase().contains(forbidden),
            "replacement stock client failed with `{forbidden}`: {second_capture:?}"
        );
    }

    endpoint
        .revoke_frontend(&bootstrap, &mut credential)
        .await
        .unwrap();
    handle.shutdown().await;

    let observation = read_observation(&harness, event_kinds).await;
    harness.engine.shutdown().await;
    std::fs::remove_dir_all(harness.root).ok();
    observation
}

#[cfg(unix)]
async fn stock_opencode_binary_observation(interrupt: bool) -> Option<Observation> {
    use std::io::{Read, Write};
    use std::os::fd::FromRawFd;
    use std::os::unix::fs::PermissionsExt;
    use std::process::Stdio;

    let opencode = std::env::var_os("SUPERCODE_STOCK_OPENCODE_BIN")
        .expect("SUPERCODE_STOCK_OPENCODE_BIN must name the pinned unmodified OpenCode binary");
    let version = std::process::Command::new(&opencode)
        .arg("--version")
        .output()
        .unwrap();
    assert!(version.status.success());
    assert_eq!(String::from_utf8_lossy(&version.stdout).trim(), "1.2.15");

    let harness = harness("stock-opencode-binary");
    let client_root = harness.root.join("opencode-client");
    std::fs::create_dir(&client_root).unwrap();
    std::fs::set_permissions(&client_root, std::fs::Permissions::from_mode(0o700)).unwrap();
    for child in [
        "home",
        "tmp",
        "config",
        "data",
        "state",
        "cache",
        "opencode-config",
    ] {
        std::fs::create_dir(client_root.join(child)).unwrap();
    }

    let observer = harness.engine.attach(100).await.unwrap();
    let (endpoint, credential) = supercode_opencode_frontend::OpenCodeEndpoint::new(
        harness.engine.clone(),
        SESSION_ID,
        &harness.root,
    )
    .unwrap();
    let handle = endpoint.bind(0).await.unwrap();

    let mut trace_proxy = None;
    let client_url = if interrupt {
        let upstream = url::Url::parse(&handle.url()).unwrap();
        let upstream_port = upstream.port().unwrap();
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let proxy_port = listener.local_addr().unwrap().port();
        drop(listener);
        let raw_trace = harness.root.join("stock-opencode-interrupt-raw.jsonl");
        let script = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .join("../../scripts/client-protocol-corpus/capture-http-proxy.mjs");
        let mut proxy = tokio::process::Command::new("node")
            .arg(script)
            .arg(format!("--listen={proxy_port}"))
            .arg(format!("--upstream={upstream_port}"))
            .arg(format!("--output={}", raw_trace.display()))
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .spawn()
            .unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
        assert!(proxy.try_wait().unwrap().is_none(), "trace proxy exited");
        trace_proxy = Some((proxy, raw_trace));
        format!("http://127.0.0.1:{proxy_port}")
    } else {
        handle.url()
    };

    let mut master = -1;
    let mut slave = -1;
    let dimensions = libc::winsize {
        ws_row: 40,
        ws_col: 120,
        ws_xpixel: 0,
        ws_ypixel: 0,
    };
    let dimensions_ptr = std::ptr::addr_of!(dimensions).cast_mut();
    assert_eq!(
        unsafe {
            libc::openpty(
                &mut master,
                &mut slave,
                std::ptr::null_mut(),
                std::ptr::null_mut(),
                dimensions_ptr,
            )
        },
        0,
        "opening stock OpenCode proof PTY failed: {}",
        std::io::Error::last_os_error()
    );
    let writer_fd = unsafe { libc::dup(master) };
    let stdout_fd = unsafe { libc::dup(slave) };
    let stderr_fd = unsafe { libc::dup(slave) };
    assert!(writer_fd >= 0 && stdout_fd >= 0 && stderr_fd >= 0);
    let mut writer = unsafe { std::fs::File::from_raw_fd(writer_fd) };
    let mut master_file = unsafe { std::fs::File::from_raw_fd(master) };
    let stdin_file = unsafe { std::fs::File::from_raw_fd(slave) };
    let stdout_file = unsafe { std::fs::File::from_raw_fd(stdout_fd) };
    let stderr_file = unsafe { std::fs::File::from_raw_fd(stderr_fd) };
    let capture = Arc::new(std::sync::Mutex::new(Vec::new()));
    let capture_reader = capture.clone();
    let capture_task = tokio::task::spawn_blocking(move || {
        let mut chunk = [0_u8; 8192];
        loop {
            match master_file.read(&mut chunk) {
                Ok(0) | Err(_) => break,
                Ok(count) => capture_reader
                    .lock()
                    .unwrap()
                    .extend_from_slice(&chunk[..count]),
            }
        }
    });

    let path = std::env::var_os("PATH").unwrap_or_default();
    let mut command = tokio::process::Command::new(&opencode);
    command
        .env_clear()
        .current_dir(&harness.root)
        .arg("attach")
        .arg(&client_url)
        .arg("--dir")
        .arg(&harness.root)
        .arg("--session")
        .arg(endpoint.session_id())
        .env("PATH", path)
        .env("TERM", "xterm-256color")
        .env("HOME", client_root.join("home"))
        .env("TMPDIR", client_root.join("tmp"))
        .env("TMP", client_root.join("tmp"))
        .env("TEMP", client_root.join("tmp"))
        .env("XDG_CONFIG_HOME", client_root.join("config"))
        .env("XDG_DATA_HOME", client_root.join("data"))
        .env("XDG_STATE_HOME", client_root.join("state"))
        .env("XDG_CACHE_HOME", client_root.join("cache"))
        .env("OPENCODE_CONFIG_DIR", client_root.join("opencode-config"))
        .stdin(Stdio::from(stdin_file))
        .stdout(Stdio::from(stdout_file))
        .stderr(Stdio::from(stderr_file))
        .kill_on_drop(true)
        .process_group(0);
    for variable in [
        "OPENCODE_DISABLE_AUTOUPDATE",
        "OPENCODE_DISABLE_AUTOCOMPACT",
        "OPENCODE_DISABLE_CLAUDE_CODE",
        "OPENCODE_DISABLE_DEFAULT_PLUGINS",
        "OPENCODE_DISABLE_EXTERNAL_SKILLS",
        "OPENCODE_DISABLE_FILETIME_CHECK",
        "OPENCODE_DISABLE_LSP_DOWNLOAD",
        "OPENCODE_DISABLE_MODELS_FETCH",
        "OPENCODE_DISABLE_PROJECT_CONFIG",
        "OPENCODE_DISABLE_PRUNE",
        "OPENCODE_DISABLE_SHARE",
        "OPENCODE_DISABLE_TERMINAL_TITLE",
    ] {
        command.env(variable, "1");
    }
    let mut child = credential.spawn_tokio_child(&mut command).unwrap();
    drop(command);

    let ready_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(20);
    loop {
        let screen = capture.lock().unwrap().clone();
        if screen
            .windows(b"imported work complete".len())
            .any(|window| window == b"imported work complete")
        {
            break;
        }
        if screen.windows(4).any(|window| window == b"\x1b[6n") {
            writer.write_all(b"\x1b[1;1R").unwrap();
            writer.flush().unwrap();
        }
        assert!(
            tokio::time::Instant::now() < ready_deadline,
            "stock OpenCode did not render the imported session: {:?}",
            String::from_utf8_lossy(&screen)
        );
        tokio::time::sleep(std::time::Duration::from_millis(25)).await;
    }
    writer.write_all(b"\x1b[200~").unwrap();
    writer.write_all(PROMPT.as_bytes()).unwrap();
    writer.write_all(b"\x1b[201~").unwrap();
    writer.flush().unwrap();
    tokio::time::sleep(std::time::Duration::from_millis(30)).await;
    writer.write_all(b"\r").unwrap();
    writer.flush().unwrap();

    let entered = harness.provider.entered.notified();
    tokio::pin!(entered);
    let submit_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(20);
    loop {
        tokio::select! {
            _ = &mut entered => break,
            _ = tokio::time::sleep(std::time::Duration::from_secs(1)) => {
                assert!(tokio::time::Instant::now() < submit_deadline, "stock OpenCode never submitted the typed prompt");
                writer.write_all(b"\r").unwrap();
                writer.flush().unwrap();
            }
        }
    }
    if interrupt {
        // OpenCode's stock composer maps Escape during a busy turn to
        // POST /session/{id}/abort. The adapter must carry that through the
        // canonical SDK interrupt action rather than handling it locally.
        writer.write_all(b"\x1b").unwrap();
        writer.flush().unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        writer.write_all(b"\x1b").unwrap();
        writer.flush().unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        writer.write_all(b"\x1b").unwrap();
        writer.flush().unwrap();
    } else {
        harness.provider.release.notify_waiters();
    }
    let mut events = Vec::new();
    let mut observer = observer;
    loop {
        let event = tokio::time::timeout(std::time::Duration::from_secs(5), observer.next_event())
            .await
            .expect("stock OpenCode continuation timed out")
            .unwrap();
        let terminal = event.kind
            == if interrupt {
                "turn_interrupted"
            } else {
                "turn_succeeded"
            };
        events.push(event);
        if terminal {
            break;
        }
    }
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    kill_stock_process_group(&mut child);
    tokio::time::timeout(std::time::Duration::from_secs(5), child.wait())
        .await
        .expect("stock OpenCode did not exit after proof completion")
        .unwrap();
    drop(writer);
    capture_task.await.unwrap();
    let capture = String::from_utf8_lossy(&capture.lock().unwrap()).into_owned();
    assert!(
        capture.contains("imported work complete"),
        "the stock OpenCode binary emitted no imported session: {capture:?}"
    );

    if let Some((mut proxy, raw_trace)) = trace_proxy {
        proxy.kill().await.ok();
        proxy.wait().await.ok();
        if let Some(output) = std::env::var_os("SUPERCODE_STOCK_OPENCODE_TRACE_OUT") {
            let receipt = std::env::var_os("SUPERCODE_STOCK_OPENCODE_LINEAGE_OUT")
                .expect("trace output requires SUPERCODE_STOCK_OPENCODE_LINEAGE_OUT");
            let finalizer = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
                .join("../../scripts/client-protocol-corpus/finalize-opencode-interrupt.mjs");
            let captured_at = format!(
                "unix:{}",
                std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_secs()
            );
            let status = std::process::Command::new("node")
                .arg(finalizer)
                .arg("finalize")
                .arg("--raw")
                .arg(&raw_trace)
                .arg("--capture-root")
                .arg(&harness.root)
                .arg("--sanitized")
                .arg(output)
                .arg("--receipt")
                .arg(receipt)
                .arg("--captured-at")
                .arg(captured_at)
                .status()
                .unwrap();
            assert!(
                status.success(),
                "finalizing stock interrupt lineage failed"
            );
        }
    }

    handle.shutdown().await;
    let observation = if interrupt {
        assert_eq!(
            events.last().map(|event| event.kind.as_str()),
            Some("turn_interrupted")
        );
        assert!(!events.iter().any(|event| event.kind == "turn_succeeded"));
        None
    } else {
        Some(read_observation(&harness, events).await)
    };
    harness.engine.shutdown().await;
    std::fs::remove_dir_all(harness.root).ok();
    observation
}

fn assert_no_frontend_residue(observation: &Observation) {
    for bytes in [
        &observation.transcript,
        &observation.sidecar,
        &observation.reduction_log,
        &observation.claude_export,
    ] {
        let text = String::from_utf8_lossy(bytes);
        for forbidden in [
            "codex-tui",
            "remoteControl/status/changed",
            "codex_app_server/v0_144",
            "SUPERCODE_CODEX_REMOTE_TOKEN",
            "schema-extended",
            "opencode_http/v1_2_15",
            "msg_sup62_conformance",
            "prt_sup62_conformance",
            "OPENCODE_SERVER_PASSWORD",
        ] {
            assert!(
                !text.contains(forbidden),
                "frontend residue `{forbidden}` leaked"
            );
        }
    }
}

#[tokio::test]
async fn embedded_http_acp_codex_opencode_and_goose_are_byte_identical_after_a_reduced_turn() {
    let embedded = embedded_observation().await;
    let http = http_observation().await;
    let acp = acp_observation().await;
    let codex = codex_observation().await;
    let opencode = opencode_observation().await;
    let goose = goose_observation().await;

    for (surface, observation) in [
        ("embedded", &embedded),
        ("http", &http),
        ("acp", &acp),
        ("codex", &codex),
        ("opencode", &opencode),
        ("goose", &goose),
    ] {
        assert_eq!(observation.session_id, SESSION_ID, "{surface}");
        assert_eq!(observation.sdk_actions.len(), 1, "{surface} SDK actions");
        let action = &observation.sdk_actions[0];
        assert_eq!(action.operation, SdkOperation::Input, "{surface}");
        assert_eq!(action.prompt, PROMPT, "{surface}");
        assert_eq!(
            action
                .request_messages
                .last()
                .and_then(|message| message.content.as_deref()),
            Some(PROMPT),
            "{surface} canonical input payload"
        );
        assert_eq!(observation.history_cursor, 5, "{surface} history cursor");
        assert_eq!(
            observation
                .events
                .iter()
                .map(|event| (event.sequence, event.kind.as_str()))
                .collect::<Vec<_>>(),
            [
                (1, "user_message"),
                (2, "turn_started"),
                (3, "text_delta"),
                (4, "usage"),
                (5, "turn_completed"),
                (6, "turn_succeeded"),
            ],
            "{surface} canonical event order"
        );
        assert_no_frontend_residue(observation);
    }
    assert_eq!(embedded, http);
    assert_eq!(embedded, acp);
    assert_eq!(embedded, codex);
    assert_eq!(embedded, opencode);
    assert_eq!(embedded, goose);
}

#[tokio::test]
#[ignore = "requires npm ci and build in sdk/frontend-pi plus Node 22.19+"]
async fn pi_joins_every_frontend_in_the_byte_identical_reduced_turn_matrix() {
    let embedded = embedded_observation().await;
    let http = http_observation().await;
    let acp = acp_observation().await;
    let codex = codex_observation().await;
    let opencode = opencode_observation().await;
    let goose = goose_observation().await;
    let pi = pi_observation().await;

    for (surface, observation) in [
        ("embedded", &embedded),
        ("http", &http),
        ("acp", &acp),
        ("codex", &codex),
        ("opencode", &opencode),
        ("goose", &goose),
        ("pi", &pi),
    ] {
        assert_eq!(observation.session_id, SESSION_ID, "{surface}");
        assert_eq!(observation.sdk_actions.len(), 1, "{surface} SDK actions");
        assert_eq!(observation.history_cursor, 5, "{surface} history cursor");
        assert_no_frontend_residue(observation);
    }
    assert_eq!(embedded, http);
    assert_eq!(embedded, acp);
    assert_eq!(embedded, codex);
    assert_eq!(embedded, opencode);
    assert_eq!(embedded, goose);
    assert_eq!(embedded, pi);
}

#[tokio::test]
async fn multimodal_payload_is_exact_through_local_http_acp_and_controller_lease() {
    const PROMPT: &str = "inspect both runtime-owned images";
    let image_urls = vec![
        "data:image/png;base64,cG5n".to_string(),
        "https://example.test/runtime-image.webp".to_string(),
    ];

    let (root, runtime, capture) = image_runtime("local");
    assert_eq!(
        FrontendRuntime::submit_with_images(&*runtime, PROMPT.into(), image_urls.clone())
            .await
            .unwrap(),
        "multimodal-ok"
    );
    assert_exact_multimodal_request("local", &capture, PROMPT, &image_urls);
    runtime.shutdown().await;
    std::fs::remove_dir_all(root).ok();

    let (root, runtime, capture) = image_runtime("http");
    let token: Arc<str> = "sup62-multimodal-http-token".into();
    let address = supercode::server::run_http(runtime.clone(), "127.0.0.1:0", token.clone())
        .await
        .unwrap();
    let remote = HttpFrontendRuntime::connect(format!("http://{address}"), token.to_string())
        .await
        .unwrap();
    assert_eq!(
        FrontendRuntime::submit_with_images(&*remote, PROMPT.into(), image_urls.clone())
            .await
            .unwrap(),
        "multimodal-ok"
    );
    assert_exact_multimodal_request("HTTP", &capture, PROMPT, &image_urls);
    remote.detach().await.unwrap();
    runtime.shutdown().await;
    std::fs::remove_dir_all(root).ok();

    let (root, runtime, capture) = image_runtime("acp");
    let token = "sup62-multimodal-acp-token";
    let address =
        supercode::server::run_http(runtime.clone(), "127.0.0.1:0", Arc::<str>::from(token))
            .await
            .unwrap();
    let mut env = BTreeMap::new();
    env.insert("SUPERCODE_SERVER_TOKEN".into(), token.into());
    let remote = supercode::AcpFrontendRuntime::connect(supercode::AcpFrontendConnectOptions {
        launch: supercode::RuntimeLaunch {
            program: env!("CARGO_BIN_EXE_supercode").into(),
            arguments: vec![
                "acp".into(),
                "--connect".into(),
                format!("http://{address}"),
            ],
            env,
        },
        cwd: Some(root.clone()),
        session_id: None,
        after_sequence: None,
    })
    .await
    .unwrap();
    assert_eq!(
        FrontendRuntime::submit_with_images(&*remote, PROMPT.into(), image_urls.clone())
            .await
            .unwrap(),
        "multimodal-ok"
    );
    assert_exact_multimodal_request("ACP", &capture, PROMPT, &image_urls);
    remote.detach().await.unwrap();
    runtime.shutdown().await;
    std::fs::remove_dir_all(root).ok();

    let (root, runtime, capture) = image_runtime("lease");
    let coordinated = CoordinatedRuntime::new(runtime.clone());
    let observer = coordinated.client(
        RuntimeClientId::parse("sup62-multimodal-observer").unwrap(),
        RuntimeAuthorization::observer(),
    );
    let denied = FrontendRuntime::submit_with_images(&*observer, PROMPT.into(), image_urls.clone())
        .await
        .unwrap_err();
    assert_eq!(denied.code(), supercode::SdkErrorCode::Unauthorized);
    assert!(
        capture
            .0
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .is_empty(),
        "unauthorized lease submission reached the provider"
    );
    let controller = coordinated.client(
        RuntimeClientId::parse("sup62-multimodal-controller").unwrap(),
        RuntimeAuthorization::owner(),
    );
    controller.take_control().unwrap();
    assert_eq!(
        FrontendRuntime::submit_with_images(&*controller, PROMPT.into(), image_urls.clone())
            .await
            .unwrap(),
        "multimodal-ok"
    );
    assert_exact_multimodal_request("controller lease", &capture, PROMPT, &image_urls);
    runtime.shutdown().await;
    std::fs::remove_dir_all(root).ok();
}

/// Release proof that the adapter-level Codex row above is also exercised by
/// the exact unmodified pinned TUI binary. The dedicated compatibility CI job
/// installs 0.144.4, supplies its path, and runs this ignored live-process
/// gate; ordinary workspace tests remain hermetic and need no stock install.
#[cfg(unix)]
#[tokio::test]
#[ignore = "requires SUPERCODE_STOCK_CODEX_BIN pointing to unmodified codex-cli 0.144.4"]
async fn unmodified_stock_codex_matches_embedded_http_and_acp_byte_for_byte() {
    let embedded = embedded_observation().await;
    let http = http_observation().await;
    let acp = acp_observation().await;
    let opencode = opencode_observation().await;
    let stock = stock_codex_binary_observation().await;

    assert_eq!(embedded, http);
    assert_eq!(embedded, acp);
    assert_eq!(embedded, opencode);
    assert_eq!(embedded, stock);
    assert_no_frontend_residue(&stock);
}

/// Release proof for the second rendering/protocol stack. The unmodified
/// OpenCode TUI attaches over the production authenticated endpoint, renders
/// on a real PTY, submits through its stock composer, and must produce the
/// exact same durable observation as the embedded SDK surface.
#[cfg(unix)]
#[tokio::test]
#[ignore = "requires SUPERCODE_STOCK_OPENCODE_BIN pointing to unmodified OpenCode 1.2.15"]
async fn unmodified_stock_opencode_matches_embedded_byte_for_byte() {
    let embedded = embedded_observation().await;
    let stock = stock_opencode_binary_observation(false)
        .await
        .expect("completed stock turn observation");

    assert_eq!(embedded, stock);
    assert_no_frontend_residue(&stock);
}

/// Live release proof that the pinned, unmodified stock OpenCode composer
/// emits its abort mutation and reaches the canonical SDK interruption path.
#[cfg(unix)]
#[tokio::test]
#[ignore = "requires SUPERCODE_STOCK_OPENCODE_BIN pointing to unmodified OpenCode 1.2.15"]
async fn unmodified_stock_opencode_interrupts_the_sdk_owned_turn() {
    assert!(stock_opencode_binary_observation(true).await.is_none());
}