supercode-core 0.2.1

//! P5-8 (§2 module 31 `server`, D8 "remote attach") — the HTTP transport:
//! auth-gated `POST /rpc`, `GET /events` SSE, and graceful shutdown (no
//! orphaned listener). Uses `reqwest` (already a real dependency of this
//! crate, for `crate::mcp`'s HTTP/SSE CLIENT) as a genuine network client
//! against the real `run_http` listener bound to `127.0.0.1:0` — no mocked
//! transport, only the model provider is a mock (no network/billed calls).

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex as StdMutex};
use std::time::Duration;

use async_trait::async_trait;
use futures::{SinkExt, StreamExt};
use supercode::server::{
    run_frontend_websocket, run_http, run_http_authorized, run_http_authorized_with_lease_ttl,
    FrontendRequestBridge, RpcEngine, RuntimeHttpCredential,
};
use supercode::{
    configfile::{resolve, ResolveOptions},
    Agent, ApprovalPolicy, ChatMessage, ChatRequest, ClaudeRuntimeManifest, Config,
    FrontendApprovalDecision, FrontendElicitationAction, FrontendFacadeMethod,
    FrontendOperationInvocation, FrontendOperationKind, FrontendOperationResult, FrontendRequest,
    FrontendResponse, FrontendRuntime, FrontendRuntimeError, FrontendRuntimeMetadata,
    FrontendTurnState, FunctionCall, HttpFrontendRuntime, Provider, Role, RuntimeClientId, Session,
    ToolCall, Usage,
};

struct SaysProvider(String);
struct StreamingSaysProvider(String);

#[async_trait]
impl Provider for SaysProvider {
    async fn complete(
        &self,
        _req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode::Result<(ChatMessage, Usage)> {
        Ok((ChatMessage::assistant(self.0.clone()), Usage::default()))
    }
}

#[async_trait]
impl Provider for StreamingSaysProvider {
    async fn complete(
        &self,
        _req: &ChatRequest,
        on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode::Result<(ChatMessage, Usage)> {
        on_delta(&self.0);
        Ok((ChatMessage::assistant(self.0.clone()), Usage::default()))
    }
}

struct RecordingOperationProvider {
    prompts: Arc<StdMutex<Vec<String>>>,
}

#[async_trait]
impl Provider for RecordingOperationProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode::Result<(ChatMessage, Usage)> {
        let prompt = req
            .messages
            .iter()
            .rev()
            .find(|message| message.role == Role::User)
            .and_then(|message| message.content.clone())
            .expect("operation turn must contain a user prompt");
        self.prompts
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .push(prompt);
        Ok((
            ChatMessage::assistant("operation complete"),
            Usage::default(),
        ))
    }
}

fn temp_dir(tag: &str) -> std::path::PathBuf {
    let dir = std::env::temp_dir().join(format!(
        "supercode-server-http-{tag}-{}",
        std::process::id()
    ));
    std::fs::create_dir_all(&dir).unwrap();
    dir
}

async fn spawn_server(reply: &str) -> (Arc<RpcEngine>, String, String) {
    let dir = temp_dir("spawn");
    let config = Config::builder().cwd(dir).build();
    let agent = Agent::with_provider(config, Box::new(SaysProvider(reply.to_string())));
    let engine = RpcEngine::new(agent, None);
    let token: Arc<str> = "s3cr3t-test-token".into();
    let addr = run_http(engine.clone(), "127.0.0.1:0", token.clone())
        .await
        .expect("bind loopback ephemeral port");
    (engine, format!("http://{addr}"), token.to_string())
}

#[tokio::test]
async fn reference_browser_observer_is_static_public_but_runtime_data_stays_authenticated() {
    let (engine, base_url, _token) = spawn_server("unused").await;
    let client = reqwest::Client::new();
    let page = client
        .get(format!("{base_url}/observer/"))
        .send()
        .await
        .unwrap();
    assert!(page.status().is_success());
    assert_eq!(
        page.headers()
            .get("x-content-type-options")
            .and_then(|value| value.to_str().ok()),
        Some("nosniff")
    );
    assert!(page
        .headers()
        .get("content-security-policy")
        .and_then(|value| value.to_str().ok())
        .is_some_and(|value| value.contains("connect-src 'self'")));
    let html = page.text().await.unwrap();
    assert!(html.contains("Session observer"));
    assert!(html.contains("not a coding IDE or agent engine"));
    assert!(!html.contains("unused"));

    let source = client
        .get(format!("{base_url}/observer/client.mjs"))
        .send()
        .await
        .unwrap();
    assert!(source.status().is_success());
    assert_eq!(
        source
            .headers()
            .get("content-type")
            .and_then(|value| value.to_str().ok()),
        Some("text/javascript; charset=utf-8")
    );
    assert!(source
        .text()
        .await
        .unwrap()
        .contains("@volter-ai-dev/supercode-frontend"));

    let unauthorized = client
        .post(format!("{base_url}/rpc"))
        .json(&serde_json::json!({
            "jsonrpc":"2.0", "id":1,
            "method":FrontendFacadeMethod::Describe.wire_name(), "params":{}
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(unauthorized.status(), reqwest::StatusCode::UNAUTHORIZED);
    engine.shutdown().await;
}

#[tokio::test]
async fn v2_frontend_methods_and_legacy_aliases_project_the_same_descriptor() {
    let (_engine, base_url, token) = spawn_server("unused").await;
    let client = reqwest::Client::new();
    let request = |method: &str| {
        client
            .post(format!("{base_url}/rpc"))
            .bearer_auth(&token)
            .header("x-supercode-client-id", "schema-alias-test")
            .header("x-supercode-permissions", "observe")
            .json(&serde_json::json!({"jsonrpc":"2.0","id":1,"method":method,"params":{}}))
    };
    let canonical: serde_json::Value = request(FrontendFacadeMethod::Describe.wire_name())
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let legacy: serde_json::Value = request("frontend.describe")
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(canonical["result"], legacy["result"]);
    assert_eq!(canonical["result"]["schema_version"], 2);
}

#[tokio::test]
async fn http_sse_is_armed_before_snapshot_and_loses_no_boundary_event() {
    let fixture: serde_json::Value = serde_json::from_str(include_str!(
        "../../../sdk/frontend/test/fixtures/conformance.json"
    ))
    .unwrap();
    let engine = RpcEngine::new_named(
        Agent::with_provider(
            Config::builder()
                .cwd(temp_dir("http-atomic-attach"))
                .system_prompt(fixture["system"].as_str().unwrap())
                .build(),
            Box::new(StreamingSaysProvider(
                fixture["reply"].as_str().unwrap().into(),
            )),
        ),
        fixture["session_id"].as_str().unwrap(),
        None,
    );
    let token: Arc<str> = "atomic-attach-token".into();
    let address = run_http(engine.clone(), "127.0.0.1:0", token.clone())
        .await
        .unwrap();
    let base_url = format!("http://{address}");
    let client = reqwest::Client::new();
    let headers = |request: reqwest::RequestBuilder| {
        request
            .bearer_auth(token.as_ref())
            .header("x-supercode-client-id", "atomic-http-client")
            .header("x-supercode-permissions", "observe,interact,approve")
    };

    // The live subscriber is fully established before the snapshot request,
    // exactly as the generated TypeScript HTTP client now does it.
    let response = headers(client.get(format!("{base_url}/frontend/events")))
        .send()
        .await
        .unwrap();
    assert!(response.status().is_success());
    let snapshot: serde_json::Value = headers(client.post(format!("{base_url}/rpc")))
        .json(&serde_json::json!({
            "jsonrpc":"2.0", "id":1,
            "method":FrontendFacadeMethod::Attach.wire_name(),
            "params":{"limit":1000,"after_sequence":0}
        }))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let history_cursor = snapshot["result"]["history_cursor"].as_u64().unwrap();
    let accepted: serde_json::Value = headers(client.post(format!("{base_url}/rpc")))
        .json(&serde_json::json!({
            "jsonrpc":"2.0", "id":2,
            "method":FrontendFacadeMethod::SendInput.wire_name(),
            "params":{"prompt":fixture["prompt"]}
        }))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(accepted["result"]["accepted"], true);

    let mut stream = response.bytes_stream();
    let mut pending = String::new();
    let mut observed = Vec::new();
    while observed
        .last()
        .is_none_or(|event: &supercode::FrontendEvent| event.kind != "turn_succeeded")
    {
        let chunk = tokio::time::timeout(Duration::from_secs(3), stream.next())
            .await
            .expect("frontend SSE event timed out")
            .expect("frontend SSE closed before terminal event")
            .unwrap();
        pending.push_str(&String::from_utf8_lossy(&chunk));
        while let Some(boundary) = pending.find('\n') {
            let line = pending[..boundary].trim_end_matches('\r').to_string();
            pending.drain(..=boundary);
            let Some(data) = line.strip_prefix("data: ") else {
                continue;
            };
            let event: supercode::FrontendEvent = serde_json::from_str(data).unwrap();
            assert!(event.sequence > history_cursor);
            observed.push(event);
        }
    }
    assert_eq!(
        observed
            .iter()
            .map(|event| event.kind.as_str())
            .collect::<Vec<_>>(),
        fixture["event_kinds"]
            .as_array()
            .unwrap()
            .iter()
            .map(|kind| kind.as_str().unwrap())
            .collect::<Vec<_>>()
    );
    engine.shutdown().await;
}

#[tokio::test]
async fn authenticated_websocket_uses_the_same_v2_descriptor_and_event_sequence() {
    use tokio_tungstenite::tungstenite::client::IntoClientRequest;

    async fn next_json<S>(socket: &mut S, stage: &str) -> serde_json::Value
    where
        S: futures::Stream<
                Item = Result<
                    tokio_tungstenite::tungstenite::Message,
                    tokio_tungstenite::tungstenite::Error,
                >,
            > + Unpin,
    {
        let message = tokio::time::timeout(Duration::from_secs(3), socket.next())
            .await
            .unwrap_or_else(|_| panic!("WebSocket response timed out during {stage}"))
            .expect("WebSocket closed before response")
            .expect("WebSocket response failed");
        serde_json::from_str(message.to_text().unwrap()).unwrap()
    }

    let fixture: serde_json::Value = serde_json::from_str(include_str!(
        "../../../sdk/frontend/test/fixtures/conformance.json"
    ))
    .unwrap();
    let session_id = fixture["session_id"].as_str().unwrap();
    let prompt = fixture["prompt"].as_str().unwrap();
    let reply = fixture["reply"].as_str().unwrap();
    let expected_kinds = fixture["event_kinds"].as_array().unwrap();
    let config = Config::builder()
        .cwd(temp_dir("websocket-v2"))
        .system_prompt(fixture["system"].as_str().unwrap())
        .build();
    let engine = RpcEngine::new_named(
        Agent::with_provider(config, Box::new(StreamingSaysProvider(reply.into()))),
        session_id,
        None,
    );
    let server = run_frontend_websocket(
        engine.clone(),
        "127.0.0.1:0",
        vec![RuntimeHttpCredential::owner("ws-secret")],
    )
    .await
    .unwrap();
    let uri = format!("ws://{}/frontend/v2", server.address());

    let unauthorized = tokio::time::timeout(
        Duration::from_secs(3),
        tokio_tungstenite::connect_async(uri.clone()),
    )
    .await
    .expect("unauthenticated handshake timed out")
    .unwrap_err();
    assert!(matches!(
        unauthorized,
        tokio_tungstenite::tungstenite::Error::Http(ref response)
            if response.status() == 401
    ));

    let mut request = uri.into_client_request().unwrap();
    request
        .headers_mut()
        .insert("authorization", "Bearer ws-secret".parse().unwrap());
    request
        .headers_mut()
        .insert("x-supercode-client-id", "ws-controller".parse().unwrap());
    request.headers_mut().insert(
        "x-supercode-permissions",
        "observe,interact,approve,terminate".parse().unwrap(),
    );
    let (mut socket, _) = tokio_tungstenite::connect_async(request).await.unwrap();
    socket
        .send(tokio_tungstenite::tungstenite::Message::Text(
            serde_json::json!({
                "jsonrpc":"2.0",
                "id":1,
                "method":FrontendFacadeMethod::Describe.wire_name(),
                "params":{},
            })
            .to_string()
            .into(),
        ))
        .await
        .unwrap();
    let described = next_json(&mut socket, "describe").await;
    assert_eq!(described["result"]["session_id"], session_id);
    assert_eq!(described["result"]["schema_version"], 2);

    socket
        .send(tokio_tungstenite::tungstenite::Message::Text(
            serde_json::json!({
                "jsonrpc":"2.0",
                "id":2,
                "method":FrontendFacadeMethod::SendInput.wire_name(),
                "params":{"prompt":prompt},
            })
            .to_string()
            .into(),
        ))
        .await
        .unwrap();
    let mut accepted = false;
    let mut sequences = Vec::new();
    let mut kinds = Vec::new();
    while !accepted || sequences.len() != expected_kinds.len() {
        let message = next_json(&mut socket, "input/events").await;
        if message["id"] == 2 {
            assert_eq!(message["result"]["accepted"], true);
            accepted = true;
        } else if message["method"] == "frontend.v2.event" {
            sequences.push(message["params"]["event"]["sequence"].as_u64().unwrap());
            kinds.push(
                message["params"]["event"]["kind"]
                    .as_str()
                    .unwrap()
                    .to_string(),
            );
        }
    }
    assert_eq!(sequences, vec![1, 2, 3, 4, 5, 6]);
    assert_eq!(
        kinds,
        expected_kinds
            .iter()
            .map(|value| value.as_str().unwrap())
            .collect::<Vec<_>>()
    );
    socket.close(None).await.unwrap();
    assert_eq!(engine.frontend_attach(20).unwrap().history.len(), 3);
    engine.shutdown().await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn scoped_http_clients_share_one_controller_and_many_observers() {
    let config = Config::builder().cwd(temp_dir("scoped-leases")).build();
    let agent = Agent::with_provider(
        config,
        Box::new(SaysProvider("coordinated reply".to_string())),
    );
    let engine = RpcEngine::new_named(agent, "scoped-leases", None);
    let owner_token = "scoped-owner-token";
    let observer_token = "scoped-observer-token";
    let address = run_http_authorized(
        engine.clone(),
        "127.0.0.1:0",
        vec![
            RuntimeHttpCredential::owner(owner_token),
            RuntimeHttpCredential::observer(observer_token),
        ],
    )
    .await
    .unwrap();
    let base_url = format!("http://{address}");
    let observer = HttpFrontendRuntime::connect_with_client_id(
        base_url.clone(),
        observer_token,
        RuntimeClientId::parse("observer-a").unwrap(),
    )
    .await
    .unwrap();
    let first = HttpFrontendRuntime::connect_with_client_id(
        base_url.clone(),
        owner_token,
        RuntimeClientId::parse("controller-a").unwrap(),
    )
    .await
    .unwrap();
    let second = HttpFrontendRuntime::connect_with_client_id(
        base_url,
        owner_token,
        RuntimeClientId::parse("controller-b").unwrap(),
    )
    .await
    .unwrap();

    let observer_descriptor = observer.describe().await.unwrap();
    assert!(!observer_descriptor.actions.submit);
    assert!(!observer_descriptor.actions.interrupt);
    assert!(!observer_descriptor.actions.respond);
    assert!(!observer_descriptor.actions.close);
    assert!(observer_descriptor.actions.detach);
    assert!(matches!(
        observer.submit("denied".into()).await,
        Err(FrontendRuntimeError::Unauthorized { ref permission }) if permission == "interact"
    ));
    assert!(matches!(
        observer.close().await,
        Err(FrontendRuntimeError::Unauthorized { ref permission }) if permission == "terminate"
    ));

    assert_eq!(
        first.submit("first controller".into()).await.unwrap(),
        "coordinated reply"
    );
    assert!(matches!(
        second.submit("implicit takeover denied".into()).await,
        Err(FrontendRuntimeError::ControllerRequired {
            holder: Some(ref holder),
            ..
        }) if holder == "controller-a"
    ));
    let takeover = second.take_control().await.unwrap();
    assert_eq!(
        takeover.controller.unwrap().client_id.as_str(),
        "controller-b"
    );
    assert_eq!(
        second.submit("after takeover".into()).await.unwrap(),
        "coordinated reply"
    );
    assert!(matches!(
        first.submit("stale controller".into()).await,
        Err(FrontendRuntimeError::ControllerRequired {
            holder: Some(ref holder),
            ..
        }) if holder == "controller-b"
    ));

    let observed = observer.lease_snapshot().await.unwrap();
    assert_eq!(observed.observers.len(), 3);
    second.detach().await.unwrap();
    assert_eq!(
        first.submit("reclaimed".into()).await.unwrap(),
        "coordinated reply"
    );
    first.close().await.unwrap();
    engine.wait_for_shutdown().await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn bootstrap_mints_bound_frontend_credentials_and_revocation_closes_the_channel() {
    let config = Config::builder()
        .cwd(temp_dir("minted-scoped-credentials"))
        .build();
    let agent = Agent::with_provider(
        config,
        Box::new(SaysProvider("scoped credential reply".to_string())),
    );
    let engine = RpcEngine::new_named(agent, "credential-generation-runtime", None);
    let bootstrap = "bootstrap-only-owner-token";
    let address = run_http_authorized(
        engine.clone(),
        "127.0.0.1:0",
        vec![RuntimeHttpCredential::owner(bootstrap)],
    )
    .await
    .unwrap();
    let base_url = format!("http://{address}");
    let client = reqwest::Client::new();

    let query_auth = client
        .post(format!(
            "{base_url}/_supercode/frontend-credentials/mint?token={bootstrap}"
        ))
        .json(&serde_json::json!({
            "clientId":"remote-controller",
            "grant":"interactive"
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(query_auth.status().as_u16(), 401);
    let query_auth = query_auth.json::<serde_json::Value>().await.unwrap();
    assert_eq!(query_auth["error"]["name"], "unauthenticated");
    assert_eq!(query_auth["error"]["code"], -32030);

    let mint = |client_id: &'static str, grant: &'static str| {
        let client = client.clone();
        let base_url = base_url.clone();
        async move {
            let response = client
                .post(format!("{base_url}/_supercode/frontend-credentials/mint"))
                .bearer_auth(bootstrap)
                .json(&serde_json::json!({"clientId":client_id,"grant":grant}))
                .send()
                .await
                .unwrap();
            assert_eq!(response.status().as_u16(), 200);
            response.text().await.unwrap()
        }
    };
    let interactive_token = mint("remote-controller", "interactive").await;
    let observer_token = mint("remote-observer", "observer").await;
    assert_eq!(interactive_token.len(), 64);
    assert_eq!(observer_token.len(), 64);
    assert_ne!(interactive_token, observer_token);
    assert_ne!(interactive_token, bootstrap);

    let frontend_control = client
        .post(format!("{base_url}/_supercode/frontend-credentials/mint"))
        .bearer_auth(&interactive_token)
        .header("x-supercode-client-id", "remote-controller")
        .json(&serde_json::json!({
            "clientId":"forbidden-delegation",
            "grant":"observer"
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(frontend_control.status().as_u16(), 403);
    let frontend_control = frontend_control.json::<serde_json::Value>().await.unwrap();
    assert_eq!(frontend_control["error"]["name"], "unauthorized");
    assert_eq!(frontend_control["error"]["code"], -32031);
    assert_eq!(frontend_control["error"]["permission"], "bootstrap");

    let other_engine = RpcEngine::new_named(
        Agent::with_provider(
            Config::builder()
                .cwd(temp_dir("other-credential-generation"))
                .build(),
            Box::new(SaysProvider("other generation".to_string())),
        ),
        "other-credential-generation",
        None,
    );
    let other_address = run_http_authorized(
        other_engine.clone(),
        "127.0.0.1:0",
        vec![RuntimeHttpCredential::owner(bootstrap)],
    )
    .await
    .unwrap();
    let cross_generation = client
        .post(format!("http://{other_address}/rpc"))
        .bearer_auth(&interactive_token)
        .header("x-supercode-client-id", "remote-controller")
        .json(&serde_json::json!({
            "jsonrpc":"2.0","id":8,
            "method":FrontendFacadeMethod::Describe.wire_name(),"params":{}
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(cross_generation.status().as_u16(), 401);
    other_engine.shutdown().await;

    let query_scoped = client
        .get(format!(
            "{base_url}/frontend/events?token={interactive_token}"
        ))
        .header("x-supercode-client-id", "remote-controller")
        .send()
        .await
        .unwrap();
    assert_eq!(query_scoped.status().as_u16(), 401);

    let wrong_id = client
        .get(format!("{base_url}/frontend/events"))
        .bearer_auth(&interactive_token)
        .header("x-supercode-client-id", "wrong-client")
        .send()
        .await
        .unwrap();
    assert_eq!(wrong_id.status().as_u16(), 403);
    let wrong_id = wrong_id.json::<serde_json::Value>().await.unwrap();
    assert_eq!(wrong_id["error"]["name"], "unauthorized");
    assert_eq!(wrong_id["error"]["code"], -32031);
    assert_eq!(wrong_id["error"]["permission"], "client_id");

    let interactive = HttpFrontendRuntime::connect_with_authorization(
        base_url.clone(),
        interactive_token.clone(),
        RuntimeClientId::parse("remote-controller").unwrap(),
        supercode::RuntimeAuthorization::interactive(),
    )
    .await
    .unwrap();
    let descriptor = interactive.describe().await.unwrap();
    assert!(descriptor.actions.submit);
    assert!(!descriptor.actions.close);
    assert_eq!(
        interactive.submit("scoped turn".into()).await.unwrap(),
        "scoped credential reply"
    );
    assert!(matches!(
        interactive.close().await,
        Err(FrontendRuntimeError::Unauthorized { ref permission }) if permission == "terminate"
    ));

    let observer = HttpFrontendRuntime::connect_with_authorization(
        base_url.clone(),
        observer_token,
        RuntimeClientId::parse("remote-observer").unwrap(),
        supercode::RuntimeAuthorization::observer(),
    )
    .await
    .unwrap();
    assert!(matches!(
        observer.submit("forbidden".into()).await,
        Err(FrontendRuntimeError::Unauthorized { ref permission }) if permission == "interact"
    ));

    let revoked = client
        .post(format!("{base_url}/_supercode/frontend-credentials/revoke"))
        .bearer_auth(bootstrap)
        .json(&serde_json::json!({"clientId":"remote-controller"}))
        .send()
        .await
        .unwrap();
    assert_eq!(revoked.status().as_u16(), 200);
    assert_eq!(
        revoked.json::<serde_json::Value>().await.unwrap()["revoked"],
        true
    );
    tokio::time::timeout(Duration::from_secs(2), async {
        while !interactive.is_disconnected() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("the client must observe the server-acknowledged event-channel closure");

    let after_revoke = client
        .post(format!("{base_url}/rpc"))
        .bearer_auth(&interactive_token)
        .header("x-supercode-client-id", "remote-controller")
        .header("x-supercode-permissions", "observe")
        .json(&serde_json::json!({
            "jsonrpc":"2.0","id":9,
            "method":FrontendFacadeMethod::Describe.wire_name(),"params":{}
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(after_revoke.status().as_u16(), 401);
    assert!(!engine.is_shutting_down());
    engine.shutdown().await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn expired_controller_fails_by_name_before_another_client_claims() {
    let config = Config::builder().cwd(temp_dir("lease-expiry")).build();
    let agent = Agent::with_provider(config, Box::new(SaysProvider("expiry reply".to_string())));
    let engine = RpcEngine::new_named(agent, "lease-expiry", None);
    let token = "lease-expiry-owner";
    let address = run_http_authorized_with_lease_ttl(
        engine.clone(),
        "127.0.0.1:0",
        vec![RuntimeHttpCredential::owner(token)],
        25,
    )
    .await
    .unwrap();
    let first = HttpFrontendRuntime::connect_with_client_id(
        format!("http://{address}"),
        token,
        RuntimeClientId::parse("expiring-controller").unwrap(),
    )
    .await
    .unwrap();
    let second = HttpFrontendRuntime::connect_with_client_id(
        format!("http://{address}"),
        token,
        RuntimeClientId::parse("replacement-controller").unwrap(),
    )
    .await
    .unwrap();
    assert_eq!(first.submit("claim".into()).await.unwrap(), "expiry reply");
    tokio::time::sleep(Duration::from_millis(40)).await;
    assert_eq!(
        first.submit("expired".into()).await.unwrap_err(),
        FrontendRuntimeError::LeaseExpired
    );
    assert_eq!(
        second.submit("replacement".into()).await.unwrap(),
        "expiry reply"
    );
    second.close().await.unwrap();
    engine.wait_for_shutdown().await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn dropped_controller_transport_releases_ownership_without_closing_runtime() {
    let config = Config::builder().cwd(temp_dir("controller-drop")).build();
    let agent = Agent::with_provider(config, Box::new(SaysProvider("drop reply".to_string())));
    let engine = RpcEngine::new_named(agent, "controller-drop", None);
    let token = "controller-drop-owner";
    let address = run_http_authorized(
        engine.clone(),
        "127.0.0.1:0",
        vec![RuntimeHttpCredential::owner(token)],
    )
    .await
    .unwrap();
    let first = HttpFrontendRuntime::connect_with_client_id(
        format!("http://{address}"),
        token,
        RuntimeClientId::parse("crashed-controller").unwrap(),
    )
    .await
    .unwrap();
    assert_eq!(first.submit("claim".into()).await.unwrap(), "drop reply");
    drop(first);

    let replacement = HttpFrontendRuntime::connect_with_client_id(
        format!("http://{address}"),
        token,
        RuntimeClientId::parse("replacement-after-crash").unwrap(),
    )
    .await
    .unwrap();
    let reply = tokio::time::timeout(Duration::from_secs(2), async {
        loop {
            match replacement.submit("after crash".into()).await {
                Ok(reply) => break reply,
                Err(FrontendRuntimeError::ControllerRequired { .. }) => {
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
                Err(error) => panic!("unexpected replacement failure: {error}"),
            }
        }
    })
    .await
    .expect("transport loss must release the controller before its lease TTL");
    assert_eq!(reply, "drop reply");
    assert!(!engine.is_shutting_down());
    replacement.close().await.unwrap();
    engine.wait_for_shutdown().await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn close_race_preserves_named_interruption_and_observer_cursors() {
    let (engine, remote, started, release) = spawn_gated_server().await;
    let mut first = remote.attach(20).await.unwrap();
    let mut second = remote.attach(20).await.unwrap();
    let submit_runtime = remote.clone();
    let submit = tokio::spawn(async move { submit_runtime.submit("close race".into()).await });
    tokio::time::timeout(Duration::from_secs(2), started.notified())
        .await
        .expect("active turn should start before close");
    remote.close().await.unwrap();
    release.notify_waiters();
    assert_eq!(
        submit.await.unwrap().unwrap_err(),
        FrontendRuntimeError::Submit(supercode::RuntimeSubmitError::Interrupted)
    );

    let first_collect = async {
        let mut seen = Vec::new();
        loop {
            let event = first.next_event().await.unwrap();
            let terminal = event.kind == "turn_interrupted";
            seen.push((event.sequence, event.kind));
            if terminal {
                break seen;
            }
        }
    };
    let second_collect = async {
        let mut seen = Vec::new();
        loop {
            let event = second.next_event().await.unwrap();
            let terminal = event.kind == "turn_interrupted";
            seen.push((event.sequence, event.kind));
            if terminal {
                break seen;
            }
        }
    };
    let (first_events, second_events) = tokio::time::timeout(Duration::from_secs(2), async {
        tokio::join!(first_collect, second_collect)
    })
    .await
    .expect("both observers must receive the close-race terminal event");
    assert_eq!(first_events, second_events);
    assert!(first_events.windows(2).all(|pair| pair[0].0 < pair[1].0));
    engine.wait_for_shutdown().await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn local_and_http_frontends_receive_the_same_scheduled_lifecycle() {
    let config = Config::builder().cwd(temp_dir("scheduler-parity")).build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(SaysProvider("scheduled over both transports".into())),
    );
    let source = [
        serde_json::json!({
            "type": "user", "sessionId": "scheduler-parity", "cwd": "/tmp",
            "timestamp": "2020-01-01T00:00:00Z",
            "message": {"role": "user", "content": "initial"}
        }),
        serde_json::json!({
            "type": "assistant", "timestamp": "2020-01-01T00:00:01Z",
            "message": {"role": "assistant", "content": [{
                "type": "tool_use", "id": "wake-http", "name": "ScheduleWakeup",
                "input": {"delaySeconds": 1, "prompt": "SCHEDULED_OVER_HTTP"}
            }]}
        }),
        serde_json::json!({
            "type": "user", "timestamp": "2020-01-01T00:00:02Z",
            "message": {"role": "user", "content": [{
                "type": "tool_result", "tool_use_id": "wake-http",
                "content": "Next wakeup scheduled for 00:00:02 (in 1s)."
            }]}
        }),
    ]
    .into_iter()
    .map(|value| value.to_string())
    .collect::<Vec<_>>()
    .join("\n");
    let session = Session::from_claude_code_str(&source).unwrap();
    let mut manifest = ClaudeRuntimeManifest::from_session(&session).unwrap();
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_secs() as i64;
    manifest.activate_scheduler(now).unwrap();
    agent.set_claude_runtime_manifest(manifest);

    let engine = RpcEngine::new_named(agent, "scheduler-parity", None);
    let token: Arc<str> = "scheduler-parity-token".into();
    let addr = run_http(engine.clone(), "127.0.0.1:0", token.clone())
        .await
        .unwrap();
    let remote = HttpFrontendRuntime::connect(format!("http://{addr}"), token.to_string())
        .await
        .unwrap();
    let local = FrontendRuntime::attach(engine.as_ref(), 20).await.unwrap();
    let over_http = FrontendRuntime::attach(remote.as_ref(), 20).await.unwrap();

    let collect = |mut attachment: supercode::FrontendAttachment| async move {
        let mut events = Vec::new();
        loop {
            let event = attachment.next_event().await.unwrap();
            events.push((event.kind.clone(), event.payload));
            if event.kind == "scheduled_prompt_completed" {
                break events;
            }
        }
    };
    assert!(engine.start_claude_scheduler());
    let (local_events, http_events) = tokio::time::timeout(Duration::from_secs(3), async {
        tokio::join!(collect(local), collect(over_http))
    })
    .await
    .expect("both frontend transports should observe the scheduled lifecycle");
    assert_eq!(http_events, local_events);
    assert_eq!(local_events.first().unwrap().0, "scheduled_prompt_started");
    assert_eq!(local_events.last().unwrap().0, "scheduled_prompt_completed");
    engine.shutdown().await;
}

struct GatedStreamingProvider {
    started: Arc<tokio::sync::Notify>,
    release: Arc<tokio::sync::Notify>,
}

struct HttpSteeringProvider {
    calls: AtomicUsize,
    started: Arc<tokio::sync::Notify>,
    release: Arc<tokio::sync::Notify>,
}

struct HttpApprovalProvider {
    calls: AtomicUsize,
    command: String,
}

#[async_trait]
impl Provider for HttpApprovalProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode::Result<(ChatMessage, Usage)> {
        if self.calls.fetch_add(1, Ordering::SeqCst) == 0 {
            Ok((
                ChatMessage {
                    role: Role::Assistant,
                    content: None,
                    content_parts: None,
                    tool_calls: Some(vec![ToolCall {
                        id: "http-approval".into(),
                        kind: "function".into(),
                        function: FunctionCall {
                            name: "bash".into(),
                            arguments: serde_json::json!({"command": self.command}).to_string(),
                        },
                    }]),
                    tool_call_id: None,
                    name: None,
                    metadata: Default::default(),
                },
                Usage::default(),
            ))
        } else {
            let result = req
                .messages
                .last()
                .and_then(|message| message.content.as_deref());
            assert!(!result.unwrap_or_default().contains("was not approved"));
            Ok((ChatMessage::assistant("HTTP approved"), Usage::default()))
        }
    }
}

#[async_trait]
impl Provider for HttpSteeringProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode::Result<(ChatMessage, Usage)> {
        if self.calls.fetch_add(1, Ordering::SeqCst) == 0 {
            self.started.notify_one();
            self.release.notified().await;
            Ok((
                ChatMessage {
                    role: Role::Assistant,
                    content: None,
                    content_parts: None,
                    tool_calls: Some(vec![ToolCall {
                        id: "http-steer-tool".into(),
                        kind: "function".into(),
                        function: FunctionCall {
                            name: "list_dir".into(),
                            arguments: "{}".into(),
                        },
                    }]),
                    tool_call_id: None,
                    name: None,
                    metadata: Default::default(),
                },
                Usage::default(),
            ))
        } else {
            assert!(req.messages.iter().any(|message| {
                message.role == Role::User && message.content.as_deref() == Some("steer over HTTP")
            }));
            Ok((ChatMessage::assistant("remotely steered"), Usage::default()))
        }
    }
}

#[async_trait]
impl Provider for GatedStreamingProvider {
    async fn complete(
        &self,
        _req: &ChatRequest,
        on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode::Result<(ChatMessage, Usage)> {
        on_delta("remote partial");
        self.started.notify_one();
        self.release.notified().await;
        Ok((ChatMessage::assistant("remote final"), Usage::default()))
    }
}

async fn spawn_gated_server() -> (
    Arc<RpcEngine>,
    Arc<HttpFrontendRuntime>,
    Arc<tokio::sync::Notify>,
    Arc<tokio::sync::Notify>,
) {
    let started = Arc::new(tokio::sync::Notify::new());
    let release = Arc::new(tokio::sync::Notify::new());
    let config = Config::builder().cwd(temp_dir("frontend-gated")).build();
    let agent = Agent::with_provider(
        config,
        Box::new(GatedStreamingProvider {
            started: started.clone(),
            release: release.clone(),
        }),
    );
    let engine = RpcEngine::new_named_with_frontend_metadata(
        agent,
        "http-frontend-session",
        FrontendRuntimeMetadata {
            source_harness: Some("claude-code".into()),
            emulation_profile: Some("cc-parity".into()),
        },
        None,
    );
    let token: Arc<str> = "frontend-http-token".into();
    let addr = run_http(engine.clone(), "127.0.0.1:0", token.clone())
        .await
        .unwrap();
    let remote = HttpFrontendRuntime::connect(format!("http://{addr}"), token.to_string())
        .await
        .unwrap();
    (engine, remote, started, release)
}

#[tokio::test]
async fn rpc_without_a_token_is_refused_401() {
    let (_engine, base, _token) = spawn_server("hi").await;
    let client = reqwest::Client::new();
    let resp = client
        .post(format!("{base}/rpc"))
        .json(&serde_json::json!({"id": 1, "method": "status"}))
        .send()
        .await
        .expect("request should reach the server");
    assert_eq!(resp.status().as_u16(), 401);
}

#[tokio::test]
async fn rpc_with_a_wrong_token_is_refused_401() {
    let (_engine, base, _token) = spawn_server("hi").await;
    let client = reqwest::Client::new();
    let resp = client
        .post(format!("{base}/rpc"))
        .bearer_auth("not-the-real-token")
        .json(&serde_json::json!({"id": 1, "method": "status"}))
        .send()
        .await
        .expect("request should reach the server");
    assert_eq!(resp.status().as_u16(), 401);
}

#[tokio::test]
async fn rpc_with_the_correct_bearer_token_dispatches_status() {
    let (_engine, base, token) = spawn_server("hi").await;
    let client = reqwest::Client::new();
    let resp = client
        .post(format!("{base}/rpc"))
        .bearer_auth(&token)
        .json(&serde_json::json!({"id": 42, "method": "status"}))
        .send()
        .await
        .expect("request should reach the server");
    assert_eq!(resp.status().as_u16(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["id"], 42);
    assert_eq!(body["result"]["busy"], false);
}

#[tokio::test]
async fn rpc_submit_round_trip_over_http_returns_the_reply() {
    let (engine, base, token) = spawn_server("hello from http").await;
    let client = reqwest::Client::new();
    let resp = client
        .post(format!("{base}/rpc"))
        .bearer_auth(&token)
        .json(&serde_json::json!({
            "id": "HTTP_ENVELOPE_SENTINEL",
            "method": "submit",
            "params": {
                "prompt": "hi",
                "_surface": "HTTP_PARAM_SENTINEL",
            }
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["result"]["reply"], "hello from http");
    let canonical = serde_json::to_string(&engine.history(50).await).unwrap();
    assert!(!canonical.contains("HTTP_ENVELOPE_SENTINEL"));
    assert!(!canonical.contains("HTTP_PARAM_SENTINEL"));
}

#[tokio::test]
async fn events_endpoint_requires_a_token_via_query_param_and_streams_sse() {
    let (_engine, base, token) = spawn_server("streamed").await;
    let client = reqwest::Client::new();

    // No token at all -> 401, connection never upgrades to a stream.
    let unauthed = client.get(format!("{base}/events")).send().await.unwrap();
    assert_eq!(unauthed.status().as_u16(), 401);

    // Correct token via query param (the browser-EventSource-compatible
    // fallback) -> 200, text/event-stream.
    let resp = client
        .get(format!("{base}/events?token={token}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    assert_eq!(
        resp.headers()
            .get("content-type")
            .and_then(|v| v.to_str().ok()),
        Some("text/event-stream")
    );

    let mut stream = resp.bytes_stream();

    // Trigger a turn on a SEPARATE connection while the SSE stream is open,
    // so an event actually flows.
    let base2 = base.clone();
    let token2 = token.clone();
    tokio::spawn(async move {
        let c = reqwest::Client::new();
        let _ = c
            .post(format!("{base2}/rpc"))
            .bearer_auth(&token2)
            .json(&serde_json::json!({"id": 1, "method": "submit", "params": {"prompt": "go"}}))
            .send()
            .await;
    });

    // The SSE endpoint's `data:` framing IS the "this is an event" signal
    // (unlike the stdio transport, which multiplexes RPC responses AND
    // events on the same line-oriented channel and so wraps events in
    // `{"event": ...}` to tell them apart — see `run_stdio`) — each SSE
    // `data:` line is `AgentEvent::to_json`'s bare `{"type": ..., ...}`
    // projection, so look for a `turn_completed` line, the event this
    // triggered turn always ends with.
    let mut saw_turn_completed = false;
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    while tokio::time::Instant::now() < deadline {
        match tokio::time::timeout(Duration::from_millis(500), stream.next()).await {
            Ok(Some(Ok(bytes))) => {
                if String::from_utf8_lossy(&bytes).contains("\"turn_completed\"") {
                    saw_turn_completed = true;
                    break;
                }
            }
            Ok(Some(Err(_))) => break,
            Ok(None) => break,
            Err(_) => continue,
        }
    }
    assert!(
        saw_turn_completed,
        "expected an SSE `data: {{\"type\":\"turn_completed\"}}` frame"
    );
}

#[tokio::test]
async fn local_and_http_frontend_descriptors_are_semantically_equal() {
    let (engine, remote, _started, _release) = spawn_gated_server().await;
    let local = FrontendRuntime::describe(engine.as_ref()).await.unwrap();
    let over_http = FrontendRuntime::describe(remote.as_ref()).await.unwrap();
    assert_eq!(over_http, local);
    assert_eq!(over_http.session_id, "http-frontend-session");
    assert_eq!(over_http.source_harness.as_deref(), Some("claude-code"));
    assert_eq!(over_http.emulation_profile.as_deref(), Some("cc-parity"));
    assert!(over_http.actions.steer);
    assert!(!over_http.actions.respond);
    let unsupported = FrontendRuntime::respond(
        remote.as_ref(),
        FrontendResponse::Approval {
            request_id: 999,
            decision: FrontendApprovalDecision::Deny,
        },
    )
    .await;
    assert!(matches!(
        unsupported,
        Err(supercode::FrontendRuntimeError::UnsupportedAction(
            "respond"
        ))
    ));
}

fn resolve_operation_profile(profile: &str) -> supercode::configfile::Resolved {
    let toml = if profile == "minimal-custom" {
        "schema_version = 1\n[core.prompts]\nminimal-check = \"MINIMAL CUSTOM OPERATION {args}\"\n"
            .to_string()
    } else {
        format!("extends = \"{profile}\"\n")
    };
    resolve(&toml, None, &ResolveOptions::default())
        .unwrap_or_else(|error| panic!("profile {profile} did not resolve: {error}"))
}

#[tokio::test]
async fn composable_profiles_continue_identically_through_local_and_http_frontends() {
    for profile in [
        "cc-parity",
        "cx-parity",
        "oc-parity",
        "pi-core",
        "token-saver",
    ] {
        let workspace = temp_dir(&format!("continuation-{profile}"));
        let make_runtime = || {
            let mut resolved = resolve_operation_profile(profile);
            resolved.config.cwd = workspace.clone();
            let expected_modules = resolved
                .config
                .module_activation
                .iter()
                .map(ToString::to_string)
                .collect::<Vec<_>>();
            let runtime = RpcEngine::new_named_with_frontend_metadata(
                Agent::with_provider(
                    resolved.config,
                    Box::new(SaysProvider("profile continuation complete".into())),
                ),
                format!("sup46-{profile}"),
                FrontendRuntimeMetadata {
                    source_harness: Some("claude-code".into()),
                    emulation_profile: Some(profile.into()),
                },
                None,
            );
            (runtime, expected_modules)
        };

        let (local, expected_modules) = make_runtime();
        let (hosted, hosted_modules) = make_runtime();
        assert_eq!(hosted_modules, expected_modules, "profile {profile}");
        let token: Arc<str> = format!("sup46-profile-token-{profile}").into();
        let address = run_http(hosted.clone(), "127.0.0.1:0", token.clone())
            .await
            .unwrap();
        let remote = HttpFrontendRuntime::connect(format!("http://{address}"), token.to_string())
            .await
            .unwrap();

        let mut local_attachment = FrontendRuntime::attach(local.as_ref(), 50).await.unwrap();
        let mut remote_attachment = FrontendRuntime::attach(remote.as_ref(), 50).await.unwrap();
        assert_eq!(
            remote_attachment.descriptor, local_attachment.descriptor,
            "profile {profile}: transport changed the visible capability descriptor"
        );
        assert_eq!(
            local_attachment.descriptor.active_modules, expected_modules,
            "profile {profile}: descriptor diverged from resolved ModuleActivation"
        );

        let local_reply = FrontendRuntime::submit(local.as_ref(), "continue identically".into())
            .await
            .unwrap();
        let remote_reply = FrontendRuntime::submit(remote.as_ref(), "continue identically".into())
            .await
            .unwrap();
        assert_eq!(remote_reply, local_reply, "profile {profile}");

        let mut local_events = Vec::new();
        let mut remote_events = Vec::new();
        for (attachment, kinds) in [
            (&mut local_attachment, &mut local_events),
            (&mut remote_attachment, &mut remote_events),
        ] {
            loop {
                let event = tokio::time::timeout(Duration::from_secs(2), attachment.next_event())
                    .await
                    .unwrap()
                    .unwrap();
                let terminal = event.kind == "turn_succeeded";
                kinds.push(event.kind);
                if terminal {
                    break;
                }
            }
        }
        assert_eq!(remote_events, local_events, "profile {profile}");

        let local_after = FrontendRuntime::attach(local.as_ref(), 50).await.unwrap();
        let remote_after = FrontendRuntime::attach(remote.as_ref(), 50).await.unwrap();
        assert_eq!(
            serde_json::to_value(&remote_after.history).unwrap(),
            serde_json::to_value(&local_after.history).unwrap(),
            "profile {profile}: public continuation transcript diverged"
        );
        assert!(local_after
            .history
            .iter()
            .any(|message| message.content.as_deref() == Some("continue identically")));
        assert!(local_after.history.iter().any(|message| {
            message.content.as_deref() == Some("profile continuation complete")
        }));

        remote.detach().await.unwrap();
        local.shutdown().await;
        hosted.shutdown().await;
    }
}

#[tokio::test]
async fn profile_operation_catalogs_are_transport_equal_and_exhaustively_invocable() {
    for profile in [
        "cc-parity",
        "cx-parity",
        "oc-parity",
        "pi-core",
        "token-saver",
        "minimal-custom",
    ] {
        let mut resolved = resolve_operation_profile(profile);
        if profile == "minimal-custom" {
            assert_eq!(
                resolved
                    .harness
                    .core
                    .prompts
                    .get("minimal-check")
                    .map(String::as_str),
                Some("MINIMAL CUSTOM OPERATION {args}"),
                "the custom profile must pass through the trusted TOML resolver"
            );
            assert!(resolved.preset_chain.is_empty());
        }
        resolved.config.cwd = temp_dir(&format!("operation-{profile}"));
        let config = resolved.config;
        let templates = config.prompts.clone();
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let agent = Agent::with_provider(
            config,
            Box::new(RecordingOperationProvider {
                prompts: seen.clone(),
            }),
        );
        let engine = RpcEngine::new_named_with_frontend_metadata(
            agent,
            format!("operation-{profile}"),
            FrontendRuntimeMetadata {
                source_harness: Some("claude-code".into()),
                emulation_profile: Some(profile.into()),
            },
            None,
        );
        let token: Arc<str> = format!("operation-token-{profile}").into();
        let address = run_http(engine.clone(), "127.0.0.1:0", token.clone())
            .await
            .unwrap();
        let remote = HttpFrontendRuntime::connect(format!("http://{address}"), token.to_string())
            .await
            .unwrap();

        let local_descriptor = FrontendRuntime::describe(engine.as_ref()).await.unwrap();
        let http_descriptor = FrontendRuntime::describe(remote.as_ref()).await.unwrap();
        assert_eq!(http_descriptor, local_descriptor, "profile {profile}");
        assert_eq!(local_descriptor.schema_version, 2, "profile {profile}");

        let advertised_ids = local_descriptor
            .operations
            .iter()
            .map(|operation| operation.id.clone())
            .collect::<Vec<_>>();
        let mut expected_ids = templates
            .keys()
            .map(|name| format!("prompt:{name}"))
            .collect::<Vec<_>>();
        expected_ids.sort();
        assert_eq!(advertised_ids, expected_ids, "profile {profile}");
        if profile == "minimal-custom" {
            assert_eq!(
                advertised_ids,
                ["prompt:code-review", "prompt:minimal-check"],
                "trusted resolver defaults and custom prompts must both be advertised"
            );
        } else {
            assert_eq!(advertised_ids, ["prompt:code-review"], "profile {profile}");
        }
        assert!(local_descriptor.operations.iter().all(|operation| {
            operation.kind == FrontendOperationKind::Prompt && operation.command.is_some()
        }));
        for forbidden in [
            FrontendOperationKind::File,
            FrontendOperationKind::Model,
            FrontendOperationKind::Session,
            FrontendOperationKind::Subagent,
            FrontendOperationKind::Image,
            FrontendOperationKind::Reduction,
        ] {
            assert!(
                local_descriptor
                    .operations
                    .iter()
                    .all(|operation| operation.kind != forbidden),
                "profile {profile} inferred phantom {forbidden:?} operation from modules"
            );
        }

        for operation in &local_descriptor.operations {
            let command = operation.command.as_ref().unwrap();
            let arguments = format!("arguments-for-{profile}");
            let expected_prompt = templates[&command.name].replace("{args}", &arguments);
            let invocation = FrontendOperationInvocation::Prompt {
                operation_id: operation.id.clone(),
                arguments,
            };
            let local_result = FrontendRuntime::invoke(engine.as_ref(), invocation.clone())
                .await
                .unwrap();
            let http_result = FrontendRuntime::invoke(remote.as_ref(), invocation)
                .await
                .unwrap();
            assert_eq!(
                local_result,
                FrontendOperationResult::Prompt {
                    reply: "operation complete".into()
                }
            );
            assert_eq!(http_result, local_result);
            let observed = seen
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            assert_eq!(
                &observed[observed.len() - 2..],
                [expected_prompt.clone(), expected_prompt],
                "profile {profile} operation did not reach the configured prompt effect"
            );
        }

        let missing_id = format!("prompt:not-advertised-{profile}");
        let missing = FrontendOperationInvocation::Prompt {
            operation_id: missing_id.clone(),
            // An argument that looks like a slash command must never select
            // a different catalog operation.
            arguments: "/code-review attacker-selected".into(),
        };
        for error in [
            FrontendRuntime::invoke(engine.as_ref(), missing.clone())
                .await
                .unwrap_err(),
            FrontendRuntime::invoke(remote.as_ref(), missing)
                .await
                .unwrap_err(),
        ] {
            assert!(
                matches!(error, FrontendRuntimeError::UnsupportedOperation(ref id) if id == &missing_id),
                "profile {profile} lost unsupported operation identity: {error}"
            );
        }
        engine.shutdown().await;
    }
}

#[tokio::test]
async fn http_frontend_steer_reaches_an_active_agent_without_locking_it() {
    let started = Arc::new(tokio::sync::Notify::new());
    let release = Arc::new(tokio::sync::Notify::new());
    let agent = Agent::with_provider(
        Config::builder().cwd(temp_dir("http-steer")).build(),
        Box::new(HttpSteeringProvider {
            calls: AtomicUsize::new(0),
            started: started.clone(),
            release: release.clone(),
        }),
    );
    let engine = RpcEngine::new(agent, None);
    let token: Arc<str> = "http-steer-token".into();
    let addr = run_http(engine, "127.0.0.1:0", token.clone())
        .await
        .unwrap();
    let remote = HttpFrontendRuntime::connect(format!("http://{addr}"), token.to_string())
        .await
        .unwrap();
    let submit_runtime = remote.clone();
    let submit = tokio::spawn(async move {
        FrontendRuntime::submit(submit_runtime.as_ref(), "start".into()).await
    });
    tokio::time::timeout(Duration::from_secs(2), started.notified())
        .await
        .expect("provider should hold the remote runtime's active agent lock");

    tokio::time::timeout(
        Duration::from_secs(2),
        FrontendRuntime::steer(remote.as_ref(), "steer over HTTP".into()),
    )
    .await
    .expect("HTTP steering must remain responsive during the turn")
    .unwrap();
    release.notify_one();
    assert_eq!(submit.await.unwrap().unwrap(), "remotely steered");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn http_frontend_can_answer_a_typed_approval_request() {
    let dir = temp_dir("http-approval");
    let marker = dir.join("approved.marker");
    let mut config = Config::builder()
        .cwd(dir.clone())
        .approval(ApprovalPolicy::OnRequest)
        .build();
    config.permissions_enabled = true;
    let agent = Agent::with_provider(
        config,
        Box::new(HttpApprovalProvider {
            calls: AtomicUsize::new(0),
            command: format!("touch {}", marker.display()),
        }),
    );
    let engine = RpcEngine::new_named_with_frontend_requests(
        agent,
        "http-approval",
        FrontendRuntimeMetadata::default(),
        None,
    );
    let token: Arc<str> = "http-approval-token".into();
    let addr = run_http(engine, "127.0.0.1:0", token.clone())
        .await
        .unwrap();
    let remote = HttpFrontendRuntime::connect(format!("http://{addr}"), token.to_string())
        .await
        .unwrap();
    let mut attachment = FrontendRuntime::attach(remote.as_ref(), 20).await.unwrap();
    let mut observer = FrontendRuntime::attach(remote.as_ref(), 20).await.unwrap();
    assert!(attachment.descriptor.actions.respond);
    let submit_runtime = remote.clone();
    let submit = tokio::spawn(async move {
        FrontendRuntime::submit(submit_runtime.as_ref(), "run it".into()).await
    });
    let request: FrontendRequest = tokio::time::timeout(Duration::from_secs(2), async {
        loop {
            let event = attachment.next_event().await.unwrap();
            if event.kind == "request" {
                break serde_json::from_value(event.payload["request"].clone()).unwrap();
            }
        }
    })
    .await
    .expect("approval request should cross authenticated SSE");
    tokio::time::timeout(
        Duration::from_secs(2),
        FrontendRuntime::respond(
            remote.as_ref(),
            FrontendResponse::Approval {
                request_id: request.id,
                decision: FrontendApprovalDecision::Allow,
            },
        ),
    )
    .await
    .expect("approval response RPC should not wait for the active turn")
    .unwrap();
    for stream in [&mut attachment, &mut observer] {
        let resolved = tokio::time::timeout(Duration::from_secs(2), async {
            loop {
                let event = stream.next_event().await.unwrap();
                if event.kind == "request_resolved" {
                    break event;
                }
            }
        })
        .await
        .expect("every HTTP frontend should observe the approval resolution");
        assert_eq!(resolved.payload["request_id"], request.id);
        assert_eq!(resolved.payload["response"]["kind"], "approval");
        assert_eq!(resolved.payload["response"]["decision"], "allow");
    }
    assert_eq!(
        tokio::time::timeout(Duration::from_secs(2), submit)
            .await
            .expect("approved turn should complete")
            .unwrap()
            .unwrap(),
        "HTTP approved"
    );
    assert!(marker.exists());
    std::fs::remove_dir_all(&dir).ok();
}

#[tokio::test]
async fn http_frontend_can_answer_a_typed_mcp_elicitation() {
    let bridge = FrontendRequestBridge::new();
    let elicitation = bridge.elicitation_handler();
    let agent = Agent::with_provider(
        Config::builder().cwd(temp_dir("http-elicitation")).build(),
        Box::new(SaysProvider("unused".into())),
    );
    let engine = RpcEngine::new_named_with_frontend_bridge(
        agent,
        "http-elicitation",
        FrontendRuntimeMetadata::default(),
        bridge,
        None,
    );
    let token: Arc<str> = "http-elicitation-token".into();
    let addr = run_http(engine, "127.0.0.1:0", token.clone())
        .await
        .unwrap();
    let remote = HttpFrontendRuntime::connect(format!("http://{addr}"), token.to_string())
        .await
        .unwrap();
    let mut attachment = FrontendRuntime::attach(remote.as_ref(), 20).await.unwrap();
    let pending = tokio::spawn(async move {
        elicitation
            .handle(&supercode::mcp::ElicitationRequest {
                message: "Pick a region".into(),
                requested_schema: serde_json::json!({"type": "string"}),
            })
            .await
    });
    let request: FrontendRequest = loop {
        let event = attachment.next_event().await.unwrap();
        if event.kind == "request" {
            break serde_json::from_value(event.payload["request"].clone()).unwrap();
        }
    };
    FrontendRuntime::respond(
        remote.as_ref(),
        FrontendResponse::Elicitation {
            request_id: request.id,
            action: FrontendElicitationAction::Accept,
            content: Some(serde_json::json!({"region": "us-east-1"})),
        },
    )
    .await
    .unwrap();
    let response = pending.await.unwrap();
    assert_eq!(response.action, supercode::mcp::ElicitationAction::Accept);
    assert_eq!(
        response.content,
        Some(serde_json::json!({"region": "us-east-1"}))
    );
}

#[tokio::test]
async fn dropping_http_frontend_closes_its_transport_without_stopping_the_turn() {
    let (engine, remote, started, release) = spawn_gated_server().await;
    let submit_engine = engine.clone();
    let submit = tokio::spawn(async move { submit_engine.submit("keep running").await });
    tokio::time::timeout(Duration::from_secs(2), started.notified())
        .await
        .expect("runtime should enter the active turn");

    let remote_weak = Arc::downgrade(&remote);
    drop(remote);
    tokio::time::timeout(Duration::from_secs(1), async {
        while remote_weak.upgrade().is_some() {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await
    .expect("the SSE task must not retain a disconnected frontend");

    release.notify_one();
    assert_eq!(submit.await.unwrap().unwrap(), "remote final");
    assert!(!engine.status().busy);
    assert_eq!(
        engine.frontend_descriptor().connection_state,
        supercode::FrontendConnectionState::Connected
    );
}

#[tokio::test]
async fn http_frontend_attach_matches_local_replay_and_live_completion() {
    let (engine, remote, started, release) = spawn_gated_server().await;
    let submit_runtime = remote.clone();
    let submit = tokio::spawn(async move {
        FrontendRuntime::submit(submit_runtime.as_ref(), "hello over HTTP".into()).await
    });
    tokio::time::timeout(Duration::from_secs(2), started.notified())
        .await
        .expect("remote provider should enter its mid-turn gate");

    let mut local = engine.frontend_attach(20).unwrap();
    let mut over_http = FrontendRuntime::attach(remote.as_ref(), 20).await.unwrap();
    assert_eq!(local.descriptor.turn_state, FrontendTurnState::Busy);
    assert_eq!(over_http.descriptor, local.descriptor);

    for attachment in [&mut local, &mut over_http] {
        let user = attachment.next_event().await.unwrap();
        let turn_started = attachment.next_event().await.unwrap();
        let delta = attachment.next_event().await.unwrap();
        assert_eq!(user.kind, "user_message");
        assert_eq!(user.payload["text"], "hello over HTTP");
        assert_eq!(turn_started.kind, "turn_started");
        assert_eq!(delta.kind, "text_delta");
        assert_eq!(delta.payload["text"], "remote partial");
    }

    release.notify_one();
    assert_eq!(submit.await.unwrap().unwrap(), "remote final");
    for attachment in [&mut local, &mut over_http] {
        tokio::time::timeout(Duration::from_secs(2), async {
            loop {
                if attachment.next_event().await.unwrap().kind == "turn_completed" {
                    break;
                }
            }
        })
        .await
        .expect("local and HTTP frontends should observe completion");
    }

    let local_late = engine.frontend_attach(2).unwrap();
    let http_late = FrontendRuntime::attach(remote.as_ref(), 2).await.unwrap();
    assert_eq!(
        serde_json::to_value(&http_late.history).unwrap(),
        serde_json::to_value(&local_late.history).unwrap(),
        "HTTP history must equal the local public wire projection"
    );
    assert_eq!(http_late.history[0].role, Role::User);
    assert_eq!(
        http_late.history[0].content.as_deref(),
        Some("hello over HTTP")
    );
    assert_eq!(http_late.history[1].role, Role::Assistant);
    assert_eq!(
        http_late.history[1].content.as_deref(),
        Some("remote final")
    );
}

#[tokio::test]
async fn unknown_route_is_404() {
    let (_engine, base, token) = spawn_server("hi").await;
    let client = reqwest::Client::new();
    let resp = client
        .get(format!("{base}/nope"))
        .bearer_auth(&token)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 404);
}

#[tokio::test]
async fn shutdown_stops_the_accept_loop_no_orphaned_listener() {
    let (engine, base, token) = spawn_server("hi").await;
    let client = reqwest::Client::new();

    // Works before shutdown.
    let ok = client
        .post(format!("{base}/rpc"))
        .bearer_auth(&token)
        .json(&serde_json::json!({"id": 1, "method": "status"}))
        .send()
        .await
        .unwrap();
    assert_eq!(ok.status().as_u16(), 200);

    let shutdown = client
        .post(format!("{base}/rpc"))
        .bearer_auth(&token)
        .json(&serde_json::json!({"id": 2, "method": "shutdown"}))
        .send()
        .await
        .unwrap();
    assert_eq!(shutdown.status().as_u16(), 200);
    assert!(engine.is_shutting_down());

    // Give the accept loop's `select!` a moment to observe the shutdown
    // signal and break out.
    tokio::time::sleep(Duration::from_millis(200)).await;

    // A NEW connection attempt must now fail to connect at all (the accept
    // loop task has exited, though the OS socket itself — bound by the
    // `TcpListener` `run_http` created — is dropped along with that task,
    // so the port is released too).
    let client2 = reqwest::Client::builder()
        .timeout(Duration::from_millis(800))
        .build()
        .unwrap();
    let base = base.clone();
    let result = client2
        .post(format!("{base}/rpc"))
        .bearer_auth(&token)
        .json(&serde_json::json!({"id": 3, "method": "status"}))
        .send()
        .await;
    assert!(
        result.is_err(),
        "expected the listener to be gone after shutdown, got: {result:?}"
    );
}