agent_runtime_http_server/

lib.rs

//! Axum transport for the Runtime Open API.

use std::{convert::Infallible, sync::Arc, time::Duration};

use async_trait::async_trait;
use axum::{
    Json, Router,
    extract::{
        DefaultBodyLimit, Path, Query, Request, State,
        rejection::JsonRejection,
        ws::{Message, WebSocket, WebSocketUpgrade},
    },
    http::{HeaderMap, StatusCode, header},
    middleware::{self, Next},
    response::{
        IntoResponse, Response, Sse,
        sse::{Event, KeepAlive},
    },
    routing::{get, post},
};
use futures_util::{SinkExt, Stream, StreamExt};
use runtime_api_contract::{
    AgentTask, AgentTaskEvent, AgentTaskEventPage, AgentTaskWebSocketClientMessage,
    AgentTaskWebSocketServerMessage, ApiErrorBody, ClaudeContentBlock, ClaudeError,
    ClaudeErrorEnvelope, ClaudeMessage, ClaudeMessageRequest, ClaudeUsage, CreateAgentTaskRequest,
    CreateExecutionRequest, ExecutionOptions, ExecutionState, ModelGenerationOptions,
    ModelToolChoice, OpenAiError, OpenAiErrorEnvelope, OpenAiInputTokensDetails,
    OpenAiOutputContent, OpenAiOutputItem, OpenAiOutputTokensDetails, OpenAiResponse,
    OpenAiResponseError, OpenAiResponseRequest, OpenAiUsage, RuntimeInput, RuntimeMessage,
    RuntimeMessageRole, SubmitInputRequest, openapi_document,
};
use runtime_link_api::{LinkError, RuntimeLink};
use runtime_types::{
    CallerScope, CredentialHandle, ExecutionId, RequestAuthority, RuntimeInstanceId,
};
use serde::Deserialize;
use serde_json::json;
use sha2::{Digest, Sha256};

#[derive(Debug)]
pub struct AuthError {
    pub message: String,
}

#[async_trait]
pub trait IngressAuthenticator: Send + Sync + 'static {
    async fn authenticate(&self, headers: &HeaderMap) -> Result<RequestAuthority, AuthError>;
}

/// Safe single-tenant authenticator for local/container deployments.
/// Production multi-tenant hosts should inject a JWT/introspection adapter.
pub struct StaticTokenAuthenticator {
    expected_token: String,
    caller: CallerScope,
}

impl StaticTokenAuthenticator {
    pub fn new(expected_token: impl Into<String>, caller: CallerScope) -> Result<Self, AuthError> {
        let expected_token = expected_token.into();
        if expected_token.is_empty() {
            return Err(AuthError {
                message: "static bearer token must not be empty".into(),
            });
        }
        caller.validate().map_err(|error| AuthError {
            message: error.to_string(),
        })?;
        Ok(Self {
            expected_token,
            caller,
        })
    }
}

#[async_trait]
impl IngressAuthenticator for StaticTokenAuthenticator {
    async fn authenticate(&self, headers: &HeaderMap) -> Result<RequestAuthority, AuthError> {
        let bearer = headers
            .get(header::AUTHORIZATION)
            .and_then(|value| value.to_str().ok())
            .and_then(|value| value.strip_prefix("Bearer "));
        let api_key = headers
            .get("x-api-key")
            .and_then(|value| value.to_str().ok());
        if bearer.is_some() && api_key.is_some() && bearer != api_key {
            return Err(AuthError {
                message: "Authorization and x-api-key credentials disagree".into(),
            });
        }
        let value = bearer
            .or(api_key)
            .filter(|value| !value.is_empty())
            .ok_or_else(|| AuthError {
                message: "missing Authorization bearer or x-api-key credential".into(),
            })?;
        if value != self.expected_token {
            return Err(AuthError {
                message: "invalid token credential".into(),
            });
        }
        Ok(RequestAuthority {
            caller: self.caller.clone(),
            credential: CredentialHandle::new(value).map_err(|error| AuthError {
                message: error.to_string(),
            })?,
        })
    }
}

/// Authenticator for a Runtime listener reachable only through a trusted
/// identity-aware gateway. The gateway owns credential verification and
/// injects the caller scope headers; Runtime forwards the bearer through an
/// opaque credential handle to its scoped Agent Infra client.
pub struct TrustedGatewayAuthenticator {
    shared_secret_digest: [u8; 32],
}

impl TrustedGatewayAuthenticator {
    pub fn new(shared_secret: &str) -> Result<Self, AuthError> {
        if shared_secret.len() < 32 || shared_secret.len() > 4096 {
            return Err(AuthError {
                message: "trusted gateway secret must contain 32..=4096 bytes".into(),
            });
        }
        Ok(Self {
            shared_secret_digest: Sha256::digest(shared_secret.as_bytes()).into(),
        })
    }
}

#[async_trait]
impl IngressAuthenticator for TrustedGatewayAuthenticator {
    async fn authenticate(&self, headers: &HeaderMap) -> Result<RequestAuthority, AuthError> {
        fn required(headers: &HeaderMap, name: &'static str) -> Result<String, AuthError> {
            headers
                .get(name)
                .and_then(|value| value.to_str().ok())
                .filter(|value| !value.is_empty())
                .map(ToOwned::to_owned)
                .ok_or_else(|| AuthError {
                    message: format!("trusted gateway header {name} is missing"),
                })
        }
        let supplied_secret = required(headers, "x-runtime-gateway-secret")?;
        let supplied_digest: [u8; 32] = Sha256::digest(supplied_secret.as_bytes()).into();
        let matches = self
            .shared_secret_digest
            .iter()
            .zip(supplied_digest)
            .fold(0_u8, |difference, (expected, actual)| {
                difference | (*expected ^ actual)
            })
            == 0;
        if !matches {
            return Err(AuthError {
                message: "trusted gateway authentication failed".into(),
            });
        }
        let token = headers
            .get(header::AUTHORIZATION)
            .and_then(|value| value.to_str().ok())
            .and_then(|value| value.strip_prefix("Bearer "))
            .ok_or_else(|| AuthError {
                message: "trusted gateway bearer credential is missing".into(),
            })?;
        let caller = CallerScope {
            subject: required(headers, "x-runtime-subject")?,
            tenant_id: required(headers, "x-runtime-tenant-id")?,
            project_id: required(headers, "x-runtime-project-id")?,
            capabilities: headers
                .get("x-runtime-capabilities")
                .and_then(|value| value.to_str().ok())
                .unwrap_or_default()
                .split(',')
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(ToOwned::to_owned)
                .collect(),
        };
        caller.validate().map_err(|error| AuthError {
            message: error.to_string(),
        })?;
        Ok(RequestAuthority {
            caller,
            credential: CredentialHandle::new(token).map_err(|error| AuthError {
                message: error.to_string(),
            })?,
        })
    }
}

#[async_trait]
pub trait ReadinessProbe: Send + Sync + 'static {
    async fn ready(&self) -> Result<(), String>;
}

pub struct AlwaysReady;
#[async_trait]
impl ReadinessProbe for AlwaysReady {
    async fn ready(&self) -> Result<(), String> {
        Ok(())
    }
}

/// Explicit route selection for the Runtime's single public listener.
///
/// WebSocket is an HTTP upgrade route on this listener, not an independently
/// spawned socket server. Health and OpenAPI remain registered even when an
/// execution transport is disabled so operators can inspect the process.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RuntimeTransportConfig {
    pub agent_tasks: bool,
    pub openai_responses: bool,
    pub claude_messages: bool,
    pub sse: bool,
    pub websocket: bool,
    pub websocket_max_message_bytes: usize,
}

#[derive(Clone)]
struct AppState {
    link: Arc<dyn RuntimeLink>,
    authenticator: Arc<dyn IngressAuthenticator>,
    readiness: Arc<dyn ReadinessProbe>,
    transport: RuntimeTransportConfig,
}

/// Construct the complete public Runtime transport surface.
///
/// REST commands, replayable SSE, and execution-scoped WebSocket upgrades are
/// registered independently. Both streaming protocols resume from a committed
/// Journal sequence; WebSocket client commands are restricted to submit-input
/// and cancel for the execution encoded in the route. Agent definitions cannot
/// add routes or listen on ports.
pub fn router(
    link: Arc<dyn RuntimeLink>,
    authenticator: Arc<dyn IngressAuthenticator>,
    readiness: Arc<dyn ReadinessProbe>,
    transport: RuntimeTransportConfig,
) -> Router {
    let state = AppState {
        link,
        authenticator,
        readiness,
        transport,
    };
    let mut router = Router::new()
        .route("/livez", get(live))
        .route("/readyz", get(ready))
        .route("/openapi.json", get(openapi));
    if transport.agent_tasks {
        router = router
            // Runtime-native Task API. This is the preferred surface for
            // background work, task interaction, and observability.
            .route("/v1/agent/tasks", post(create_agent_task))
            .route("/v1/agent/tasks/{task_id}", get(get_agent_task))
            .route(
                "/v1/agent/tasks/{task_id}/events",
                get(get_agent_task_events),
            )
            .route(
                "/v1/agent/tasks/{task_id}/inputs",
                post(submit_agent_task_input),
            )
            .route("/v1/agent/tasks/{task_id}/cancel", post(cancel_agent_task));
    }
    if transport.openai_responses {
        router = router
            .route("/v1/responses", post(create_openai_response))
            .route("/v1/responses/{response_id}", get(get_openai_response))
            .route(
                "/v1/responses/{response_id}/cancel",
                post(cancel_openai_response),
            );
    }
    if transport.claude_messages {
        router = router.route("/v1/messages", post(create_claude_message));
    }
    if transport.agent_tasks && transport.sse {
        router = router.route(
            "/v1/agent/tasks/{task_id}/stream",
            get(stream_agent_task_events),
        );
    }
    if transport.agent_tasks && transport.websocket {
        let max_message_bytes = transport.websocket_max_message_bytes;
        router = router.route(
            "/v1/agent/tasks/{task_id}/ws",
            get(
                move |upgrade: WebSocketUpgrade,
                      state: State<AppState>,
                      headers: HeaderMap,
                      path: Path<String>,
                      query: Query<EventQuery>| {
                    websocket_agent_task(
                        upgrade.max_message_size(max_message_bytes),
                        state,
                        headers,
                        path,
                        query,
                    )
                },
            ),
        );
    }
    router
        .layer(DefaultBodyLimit::max(1024 * 1024))
        .layer(middleware::from_fn(request_id_middleware))
        .with_state(state)
}

/// Add one correlation ID to every REST, SSE, WebSocket handshake, and error
/// response. Trusted short IDs are preserved; malformed IDs are replaced.
async fn request_id_middleware(mut request: Request, next: Next) -> Response {
    let request_id = request
        .headers()
        .get("x-request-id")
        .and_then(|value| value.to_str().ok())
        .filter(|value| {
            !value.is_empty() && value.len() <= 128 && !value.chars().any(char::is_control)
        })
        .map(ToOwned::to_owned)
        .unwrap_or_else(|| format!("req_{}", uuid::Uuid::new_v4()));
    if let Ok(value) = request_id.parse() {
        request.headers_mut().insert("x-request-id", value);
    }
    let mut response = next.run(request).await;
    if let Ok(value) = request_id.parse() {
        response.headers_mut().insert("x-request-id", value);
    }
    if let Ok(value) = request_id.parse() {
        response.headers_mut().insert("request-id", value);
    }
    response
}

async fn live() -> Json<serde_json::Value> {
    Json(json!({"status":"ok","service":"agent-runtime"}))
}

async fn openapi(State(state): State<AppState>) -> Json<serde_json::Value> {
    let mut document = openapi_document();
    let paths = document["paths"]
        .as_object_mut()
        .expect("embedded OpenAPI paths must be an object");
    if !state.transport.agent_tasks {
        for path in [
            "/v1/agent/tasks",
            "/v1/agent/tasks/{taskId}",
            "/v1/agent/tasks/{taskId}/events",
            "/v1/agent/tasks/{taskId}/inputs",
            "/v1/agent/tasks/{taskId}/cancel",
            "/v1/agent/tasks/{taskId}/stream",
            "/v1/agent/tasks/{taskId}/ws",
        ] {
            paths.remove(path);
        }
    }
    if !state.transport.openai_responses {
        for path in [
            "/v1/responses",
            "/v1/responses/{responseId}",
            "/v1/responses/{responseId}/cancel",
        ] {
            paths.remove(path);
        }
    }
    if !state.transport.claude_messages {
        paths.remove("/v1/messages");
    }
    if !state.transport.sse {
        paths.remove("/v1/agent/tasks/{taskId}/stream");
    }
    if !state.transport.websocket {
        paths.remove("/v1/agent/tasks/{taskId}/ws");
    }
    Json(document)
}

async fn ready(State(state): State<AppState>) -> Response {
    match state.readiness.ready().await {
        Ok(()) => Json(json!({"status":"ready"})).into_response(),
        Err(message) => (
            StatusCode::SERVICE_UNAVAILABLE,
            Json(json!({"status":"not_ready","message":message})),
        )
            .into_response(),
    }
}

async fn create_agent_task(
    State(state): State<AppState>,
    headers: HeaderMap,
    payload: Result<Json<CreateAgentTaskRequest>, JsonRejection>,
) -> Response {
    let authority = match authenticate(&state, &headers).await {
        Ok(value) => value,
        Err(response) => return response,
    };
    let Json(mut request) = match payload {
        Ok(value) => value,
        Err(error) => {
            return api_error(
                StatusCode::BAD_REQUEST,
                "TASK_REQUEST_INVALID",
                &error.body_text(),
            );
        }
    };
    attach_request_id_metadata(&headers, &mut request.metadata);
    let key = idempotency_key_or_random(&headers, "task");
    match state
        .link
        .create_execution(&authority, &key, request.into_execution())
        .await
    {
        Ok(value) => {
            let status = if value.replayed {
                StatusCode::OK
            } else {
                StatusCode::ACCEPTED
            };
            let task = AgentTask::from(value.execution);
            let mut response = (status, Json(task.clone())).into_response();
            if let Ok(location) = task.links.self_url.parse() {
                response.headers_mut().insert(header::LOCATION, location);
            }
            response
                .headers_mut()
                .insert(header::RETRY_AFTER, "1".parse().expect("static header"));
            response
        }
        Err(error) => link_error(error),
    }
}

async fn get_agent_task(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(id): Path<String>,
) -> Response {
    let authority = match authenticate(&state, &headers).await {
        Ok(value) => value,
        Err(response) => return response,
    };
    let id = match parse_id(id) {
        Ok(value) => value,
        Err(error) => return invalid_task_id(error),
    };
    match state.link.execution(&authority, &id).await {
        Ok(value) => Json(AgentTask::from(value)).into_response(),
        Err(error) => link_error(error),
    }
}

async fn get_agent_task_events(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(id): Path<String>,
    Query(query): Query<EventQuery>,
) -> Response {
    let authority = match authenticate(&state, &headers).await {
        Ok(value) => value,
        Err(response) => return response,
    };
    let id = match parse_id(id) {
        Ok(value) => value,
        Err(error) => return invalid_task_id(error),
    };
    let limit = query.limit.unwrap_or(100);
    if !(1..=1000).contains(&limit) {
        return api_error(
            StatusCode::BAD_REQUEST,
            "TASK_EVENT_LIMIT_INVALID",
            "limit must be within 1..=1000",
        );
    }
    match state.link.events(&authority, &id, query.after, limit).await {
        Ok(page) => Json(AgentTaskEventPage {
            items: page.items.into_iter().map(AgentTaskEvent::from).collect(),
            next_after: page.next_after,
        })
        .into_response(),
        Err(error) => link_error(error),
    }
}

async fn stream_agent_task_events(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(id): Path<String>,
    Query(query): Query<EventQuery>,
) -> Response {
    let authority = match authenticate(&state, &headers).await {
        Ok(value) => value,
        Err(response) => return response,
    };
    let id = match parse_id(id) {
        Ok(value) => value,
        Err(error) => return invalid_task_id(error),
    };
    let after = match sse_resume_after(&headers, query.after) {
        Ok(value) => value,
        Err(message) => {
            return api_error(StatusCode::BAD_REQUEST, "SSE_CURSOR_INVALID", &message);
        }
    };
    if let Err(error) = state.link.execution(&authority, &id).await {
        return link_error(error);
    }
    let stream = agent_task_event_stream(state.link, authority, id, after);
    Sse::new(stream)
        .keep_alive(
            KeepAlive::new()
                .interval(Duration::from_secs(15))
                .text("heartbeat"),
        )
        .into_response()
}

async fn submit_agent_task_input(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(id): Path<String>,
    Json(request): Json<SubmitInputRequest>,
) -> Response {
    let authority = match authenticate(&state, &headers).await {
        Ok(value) => value,
        Err(response) => return response,
    };
    let id = match parse_id(id) {
        Ok(value) => value,
        Err(error) => return invalid_task_id(error),
    };
    match state.link.submit_input(&authority, &id, request).await {
        Ok(value) => Json(AgentTask::from(value)).into_response(),
        Err(error) => link_error(error),
    }
}

async fn cancel_agent_task(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(id): Path<String>,
) -> Response {
    let authority = match authenticate(&state, &headers).await {
        Ok(value) => value,
        Err(response) => return response,
    };
    let id = match parse_id(id) {
        Ok(value) => value,
        Err(error) => return invalid_task_id(error),
    };
    match state.link.cancel(&authority, &id).await {
        Ok(value) => Json(AgentTask::from(value)).into_response(),
        Err(error) => link_error(error),
    }
}

async fn create_openai_response(
    State(state): State<AppState>,
    headers: HeaderMap,
    payload: Result<Json<OpenAiResponseRequest>, JsonRejection>,
) -> Response {
    let authority = match authenticate(&state, &headers).await {
        Ok(value) => value,
        Err(_) => {
            return openai_error(
                StatusCode::UNAUTHORIZED,
                "invalid_api_key",
                "authentication failed",
                None,
            );
        }
    };
    let Json(mut request) = match payload {
        Ok(value) => value,
        Err(error) => {
            return openai_error(
                StatusCode::BAD_REQUEST,
                "invalid_json",
                &error.body_text(),
                None,
            );
        }
    };
    attach_request_id_metadata(&headers, &mut request.metadata);
    if request.stream && !state.transport.sse {
        return openai_error(
            StatusCode::BAD_REQUEST,
            "streaming_disabled",
            "stream=true is disabled for this Runtime listener",
            Some("stream"),
        );
    }
    let execution_request = match openai_execution_request(&request) {
        Ok(value) => value,
        Err((param, message)) => {
            return openai_error(
                StatusCode::BAD_REQUEST,
                "invalid_request_error",
                &message,
                param.as_deref(),
            );
        }
    };
    let key = idempotency_key_or_random(&headers, "openai-response");
    let created = match state
        .link
        .create_execution(&authority, &key, execution_request)
        .await
    {
        Ok(value) => value.execution,
        Err(error) => return openai_link_error(error),
    };
    if request.stream {
        let stream = openai_response_stream(
            state.link,
            authority,
            created,
            request.model,
            request.instructions,
            request.max_output_tokens,
            request.metadata,
        );
        return Sse::new(stream)
            .keep_alive(
                KeepAlive::new()
                    .interval(Duration::from_secs(15))
                    .text("keepalive"),
            )
            .into_response();
    }
    if request.background {
        return Json(openai_response_from_view(
            created,
            request.model,
            request.instructions,
            request.max_output_tokens,
            request.metadata,
        ))
        .into_response();
    }
    match wait_for_provider_boundary(state.link, &authority, &created.id).await {
        Ok(view) => Json(openai_response_from_view(
            view,
            request.model,
            request.instructions,
            request.max_output_tokens,
            request.metadata,
        ))
        .into_response(),
        Err(error) => openai_link_error(error),
    }
}

async fn get_openai_response(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(id): Path<String>,
) -> Response {
    let authority = match authenticate(&state, &headers).await {
        Ok(value) => value,
        Err(_) => {
            return openai_error(
                StatusCode::UNAUTHORIZED,
                "invalid_api_key",
                "authentication failed",
                None,
            );
        }
    };
    let id = match parse_openai_response_id(id) {
        Ok(value) => value,
        Err(error) => {
            return openai_error(
                StatusCode::BAD_REQUEST,
                "invalid_request_error",
                &error.to_string(),
                Some("response_id"),
            );
        }
    };
    match state.link.execution(&authority, &id).await {
        Ok(view) => {
            let model = view.runtime_instance_id.to_string();
            Json(openai_response_from_view(
                view,
                model,
                None,
                None,
                Default::default(),
            ))
            .into_response()
        }
        Err(error) => openai_link_error(error),
    }
}

async fn cancel_openai_response(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(id): Path<String>,
) -> Response {
    let authority = match authenticate(&state, &headers).await {
        Ok(value) => value,
        Err(_) => {
            return openai_error(
                StatusCode::UNAUTHORIZED,
                "invalid_api_key",
                "authentication failed",
                None,
            );
        }
    };
    let id = match parse_openai_response_id(id) {
        Ok(value) => value,
        Err(error) => {
            return openai_error(
                StatusCode::BAD_REQUEST,
                "invalid_request_error",
                &error.to_string(),
                Some("response_id"),
            );
        }
    };
    match state.link.cancel(&authority, &id).await {
        Ok(view) => {
            let model = view.runtime_instance_id.to_string();
            Json(openai_response_from_view(
                view,
                model,
                None,
                None,
                Default::default(),
            ))
            .into_response()
        }
        Err(error) => openai_link_error(error),
    }
}

async fn create_claude_message(
    State(state): State<AppState>,
    headers: HeaderMap,
    payload: Result<Json<ClaudeMessageRequest>, JsonRejection>,
) -> Response {
    let authority = match authenticate(&state, &headers).await {
        Ok(value) => value,
        Err(_) => {
            return claude_error(
                StatusCode::UNAUTHORIZED,
                "authentication_error",
                "authentication failed",
            );
        }
    };
    let anthropic_version = headers
        .get("anthropic-version")
        .and_then(|value| value.to_str().ok())
        .filter(|value| !value.is_empty());
    if anthropic_version.is_none() {
        return claude_error(
            StatusCode::BAD_REQUEST,
            "invalid_request_error",
            "anthropic-version header is required",
        );
    }
    if anthropic_version != Some("2023-06-01") {
        return claude_error(
            StatusCode::BAD_REQUEST,
            "invalid_request_error",
            "unsupported anthropic-version; this listener supports 2023-06-01",
        );
    }
    let Json(mut request) = match payload {
        Ok(value) => value,
        Err(error) => {
            return claude_error(
                StatusCode::BAD_REQUEST,
                "invalid_request_error",
                &error.body_text(),
            );
        }
    };
    if let Some(request_id) = request_id(&headers) {
        request
            .metadata
            .entry("requestId".into())
            .or_insert_with(|| serde_json::Value::String(request_id.to_string()));
    }
    if request.stream && !state.transport.sse {
        return claude_error(
            StatusCode::BAD_REQUEST,
            "invalid_request_error",
            "stream=true is disabled for this Runtime listener",
        );
    }
    let execution_request = match claude_execution_request(&request) {
        Ok(value) => value,
        Err(message) => {
            return claude_error(StatusCode::BAD_REQUEST, "invalid_request_error", &message);
        }
    };
    let key = idempotency_key_or_random(&headers, "claude-message");
    let created = match state
        .link
        .create_execution(&authority, &key, execution_request)
        .await
    {
        Ok(value) => value.execution,
        Err(error) => return claude_link_error(error),
    };
    if request.stream {
        let stream = claude_message_stream(state.link, authority, created, request.model);
        return Sse::new(stream)
            .keep_alive(
                KeepAlive::new()
                    .interval(Duration::from_secs(15))
                    .text("ping"),
            )
            .into_response();
    }
    match wait_for_provider_boundary(state.link, &authority, &created.id).await {
        Ok(view) if view.state == ExecutionState::Completed => {
            Json(claude_message_from_view(view, request.model)).into_response()
        }
        Ok(view) => claude_terminal_error(view),
        Err(error) => claude_link_error(error),
    }
}

fn openai_execution_request(
    request: &OpenAiResponseRequest,
) -> Result<CreateExecutionRequest, (Option<String>, String)> {
    if let Some(field) = request.extra.keys().next() {
        return Err((
            Some(field.clone()),
            format!("{field} is not supported by this Agent endpoint"),
        ));
    }
    if !request.tools.is_empty() {
        return Err((Some("tools".into()), "client-defined tools are not supported by the Agent endpoint; configure tools on the Agent Definition".into()));
    }
    for (present, field) in [
        (
            request.previous_response_id.is_some(),
            "previous_response_id",
        ),
        (request.conversation.is_some(), "conversation"),
        (request.reasoning.is_some(), "reasoning"),
        (request.top_p.is_some(), "top_p"),
        (!request.include.is_empty(), "include"),
        (request.safety_identifier.is_some(), "safety_identifier"),
        (request.prompt_cache_key.is_some(), "prompt_cache_key"),
    ] {
        if present {
            return Err((
                Some(field.into()),
                format!("{field} cannot be preserved by this Agent endpoint"),
            ));
        }
    }
    if request.store == Some(true) {
        return Err((
            Some("store".into()),
            "store=true requires a durable provider response store and is not supported".into(),
        ));
    }
    if request
        .truncation
        .as_deref()
        .is_some_and(|value| value != "disabled")
    {
        return Err((
            Some("truncation".into()),
            "only truncation=disabled is supported".into(),
        ));
    }
    let messages =
        extract_openai_input(&request.input).map_err(|message| (Some("input".into()), message))?;
    let instructions = request
        .instructions
        .as_ref()
        .map(extract_text_value)
        .transpose()
        .map_err(|message| (Some("instructions".into()), message))?;
    let tool_choice = parse_openai_tool_choice(request.tool_choice.as_ref())
        .map_err(|message| (Some("tool_choice".into()), message))?;
    let response_format = request
        .text
        .as_ref()
        .map(openai_response_format)
        .transpose()
        .map_err(|message| (Some("text".into()), message))?;
    let agent = request.agent.as_ref();
    if agent
        .and_then(|value| value.conversation_id.as_ref())
        .is_some()
        && messages.len() > 1
    {
        return Err((
            Some("agent.conversationId".into()),
            "agent.conversationId cannot be combined with a multi-message stateless input history"
                .into(),
        ));
    }
    let runtime_instance_id = match agent.and_then(|value| value.runtime_id.clone()) {
        Some(value) => value,
        None => RuntimeInstanceId::new(request.model.clone())
            .map_err(|error| (Some("model".into()), error.to_string()))?,
    };
    Ok(CreateExecutionRequest {
        runtime_instance_id,
        conversation_id: agent.and_then(|value| value.conversation_id.clone()),
        input: RuntimeInput::Messages { messages },
        workspace_id: agent.and_then(|value| value.workspace_id.clone()),
        model: agent.and_then(|value| value.downstream_model.clone()),
        instructions,
        metadata: request.metadata.clone(),
        generation: ModelGenerationOptions {
            max_output_tokens: request.max_output_tokens,
            temperature: request.temperature.map(|value| value as f32),
            stop_sequences: Vec::new(),
            response_format,
            tool_choice,
        },
        options: ExecutionOptions {
            deadline_seconds: agent
                .and_then(|value| value.deadline_seconds)
                .unwrap_or(900),
            max_model_turns: agent.and_then(|value| value.max_model_turns).unwrap_or(64),
            max_tool_calls: agent.and_then(|value| value.max_tool_calls).unwrap_or(256),
        },
    })
}

fn claude_execution_request(
    request: &ClaudeMessageRequest,
) -> Result<CreateExecutionRequest, String> {
    if let Some(field) = request.extra.keys().next() {
        return Err(format!("{field} is not supported by this Agent endpoint"));
    }
    if request.max_tokens == 0 {
        return Err("max_tokens must be greater than zero".into());
    }
    if request
        .temperature
        .is_some_and(|value| !value.is_finite() || !(0.0..=1.0).contains(&value))
    {
        return Err("temperature must be within 0..=1 for Claude Messages".into());
    }
    if request.messages.is_empty() {
        return Err("messages must not be empty".into());
    }
    if !request.tools.is_empty() {
        return Err("client-defined tools are not supported by the Agent endpoint; configure tools on the Agent Definition".into());
    }
    if request.top_p.is_some() || request.top_k.is_some() {
        return Err("top_p and top_k cannot be preserved by this Agent endpoint".into());
    }
    if request.thinking.is_some()
        || request.output_config.is_some()
        || request.service_tier.is_some()
    {
        return Err(
            "thinking, output_config, and service_tier cannot be preserved by this Agent endpoint"
                .into(),
        );
    }
    let messages = extract_claude_messages(&request.messages)?;
    let instructions = request
        .system
        .as_ref()
        .map(extract_text_value)
        .transpose()?;
    let tool_choice = parse_claude_tool_choice(request.tool_choice.as_ref())?;
    let metadata = request
        .metadata
        .iter()
        .map(|(key, value)| (key.clone(), scalar_metadata(value)))
        .collect();
    let agent = request.agent.as_ref();
    if agent
        .and_then(|value| value.conversation_id.as_ref())
        .is_some()
        && messages.len() > 1
    {
        return Err(
            "agent.conversationId cannot be combined with a multi-message stateless history".into(),
        );
    }
    let runtime_instance_id = match agent.and_then(|value| value.runtime_id.clone()) {
        Some(value) => value,
        None => RuntimeInstanceId::new(request.model.clone()).map_err(|error| error.to_string())?,
    };
    Ok(CreateExecutionRequest {
        runtime_instance_id,
        conversation_id: agent.and_then(|value| value.conversation_id.clone()),
        input: RuntimeInput::Messages { messages },
        workspace_id: agent.and_then(|value| value.workspace_id.clone()),
        model: agent.and_then(|value| value.downstream_model.clone()),
        instructions,
        metadata,
        generation: ModelGenerationOptions {
            max_output_tokens: Some(request.max_tokens),
            temperature: request.temperature.map(|value| value as f32),
            stop_sequences: request.stop_sequences.clone(),
            response_format: None,
            tool_choice,
        },
        options: ExecutionOptions {
            deadline_seconds: agent
                .and_then(|value| value.deadline_seconds)
                .unwrap_or(900),
            max_model_turns: agent.and_then(|value| value.max_model_turns).unwrap_or(64),
            max_tool_calls: agent.and_then(|value| value.max_tool_calls).unwrap_or(256),
        },
    })
}

fn parse_openai_tool_choice(value: Option<&serde_json::Value>) -> Result<ModelToolChoice, String> {
    let Some(value) = value else {
        return Ok(ModelToolChoice::Auto);
    };
    match value.as_str() {
        Some("auto") => Ok(ModelToolChoice::Auto),
        Some("none") => Ok(ModelToolChoice::None),
        _ => Err(
            "only tool_choice=auto or none can be preserved across Agent Infra providers".into(),
        ),
    }
}

fn parse_claude_tool_choice(value: Option<&serde_json::Value>) -> Result<ModelToolChoice, String> {
    let Some(value) = value else {
        return Ok(ModelToolChoice::Auto);
    };
    let object = value
        .as_object()
        .ok_or_else(|| "tool_choice must be an object".to_string())?;
    if object.len() != 1 {
        return Err("tool_choice contains fields this Agent endpoint cannot preserve".into());
    }
    match value.get("type").and_then(serde_json::Value::as_str) {
        Some("auto") => Ok(ModelToolChoice::Auto),
        Some("none") => Ok(ModelToolChoice::None),
        _ => Err(
            "only tool_choice.type auto or none can be preserved across Agent Infra providers"
                .into(),
        ),
    }
}

fn openai_response_format(text: &serde_json::Value) -> Result<serde_json::Value, String> {
    let format = text
        .get("format")
        .ok_or_else(|| "text.format is required when text is supplied".to_string())?;
    if !format.is_object() {
        return Err("text.format must be an object".into());
    }
    match format.get("type").and_then(serde_json::Value::as_str) {
        Some("json_schema") => {
            let object = format.as_object().expect("format was checked as an object");
            let name = object
                .get("name")
                .and_then(serde_json::Value::as_str)
                .filter(|name| !name.trim().is_empty())
                .ok_or_else(|| "text.format json_schema requires name".to_string())?;
            let schema = object
                .get("schema")
                .filter(|schema| schema.is_object())
                .ok_or_else(|| "text.format json_schema requires an object schema".to_string())?;
            let mut json_schema = serde_json::Map::new();
            json_schema.insert("name".into(), json!(name));
            json_schema.insert("schema".into(), schema.clone());
            for field in ["description", "strict"] {
                if let Some(value) = object.get(field) {
                    json_schema.insert(field.into(), value.clone());
                }
            }
            Ok(json!({"type":"json_schema","json_schema":json_schema}))
        }
        Some("text" | "json_object") => Ok(format.clone()),
        _ => Err("text.format.type must be text, json_object, or json_schema".into()),
    }
}

fn extract_openai_input(input: &serde_json::Value) -> Result<Vec<RuntimeMessage>, String> {
    if let Some(text) = input.as_str() {
        return Ok(vec![RuntimeMessage {
            role: RuntimeMessageRole::User,
            text: non_empty_text(text)?,
        }]);
    }
    let items = input
        .as_array()
        .ok_or_else(|| "input must be a string or an array of message items".to_string())?;
    let mut messages = Vec::new();
    for item in items {
        let object = item
            .as_object()
            .ok_or_else(|| "input items must be objects".to_string())?;
        if let Some(field) = object.keys().find(|field| {
            !matches!(
                field.as_str(),
                "type" | "role" | "content" | "id" | "status"
            )
        }) {
            return Err(format!("input message field {field} is not supported"));
        }
        let item_type = object
            .get("type")
            .and_then(|value| value.as_str())
            .unwrap_or("message");
        if item_type != "message" {
            return Err(format!("input item type {item_type} is not supported"));
        }
        let role = object
            .get("role")
            .and_then(|value| value.as_str())
            .unwrap_or("user");
        let role = match role {
            "system" => RuntimeMessageRole::System,
            "developer" => RuntimeMessageRole::Developer,
            "user" => RuntimeMessageRole::User,
            "assistant" => RuntimeMessageRole::Assistant,
            role => return Err(format!("input message role {role} is not supported")),
        };
        let content = object
            .get("content")
            .ok_or_else(|| "message content is required".to_string())?;
        messages.push(RuntimeMessage {
            role,
            text: extract_text_value(content)?,
        });
    }
    if messages.is_empty() {
        Err("input message array must not be empty".into())
    } else {
        Ok(messages)
    }
}

fn extract_claude_messages(
    messages: &[runtime_api_contract::ClaudeInputMessage],
) -> Result<Vec<RuntimeMessage>, String> {
    let mut transcript = Vec::with_capacity(messages.len());
    for message in messages {
        if message.role != "user" && message.role != "assistant" {
            return Err("Claude message role must be user or assistant".into());
        }
        transcript.push(RuntimeMessage {
            role: if message.role == "user" {
                RuntimeMessageRole::User
            } else {
                RuntimeMessageRole::Assistant
            },
            text: extract_text_value(&message.content)?,
        });
    }
    Ok(transcript)
}

fn extract_text_value(value: &serde_json::Value) -> Result<String, String> {
    if let Some(text) = value.as_str() {
        return non_empty_text(text);
    }
    let blocks = value
        .as_array()
        .ok_or_else(|| "content must be text or an array of text blocks".to_string())?;
    let mut text = Vec::new();
    for block in blocks {
        let object = block
            .as_object()
            .ok_or_else(|| "content blocks must be objects".to_string())?;
        if let Some(field) = object
            .keys()
            .find(|field| !matches!(field.as_str(), "type" | "text"))
        {
            return Err(format!("content block field {field} is not supported"));
        }
        let kind = object
            .get("type")
            .and_then(|value| value.as_str())
            .unwrap_or("text");
        if kind != "text" && kind != "input_text" && kind != "output_text" {
            return Err(format!("content block type {kind} is not supported"));
        }
        let part = object
            .get("text")
            .and_then(|value| value.as_str())
            .ok_or_else(|| "text block requires text".to_string())?;
        text.push(part);
    }
    non_empty_text(&text.join("\n"))
}

fn non_empty_text(text: &str) -> Result<String, String> {
    if text.trim().is_empty() {
        Err("text content must not be empty".into())
    } else {
        Ok(text.to_string())
    }
}

fn scalar_metadata(value: &serde_json::Value) -> String {
    value
        .as_str()
        .map(ToOwned::to_owned)
        .unwrap_or_else(|| value.to_string())
}

/// Provider request/stream calls must not hang forever when an Agent pauses for
/// human input. WaitingForInput is therefore a response boundary even though
/// the native Task itself remains resumable and non-terminal.
async fn wait_for_provider_boundary(
    link: Arc<dyn RuntimeLink>,
    authority: &RequestAuthority,
    id: &ExecutionId,
) -> Result<runtime_api_contract::ExecutionView, LinkError> {
    loop {
        let view = link.execution(authority, id).await?;
        if view.state.is_terminal() || view.state == ExecutionState::WaitingForInput {
            return Ok(view);
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
}

fn openai_response_from_view(
    view: runtime_api_contract::ExecutionView,
    model: String,
    instructions: Option<serde_json::Value>,
    max_output_tokens: Option<u32>,
    metadata: std::collections::BTreeMap<String, String>,
) -> OpenAiResponse {
    let status = match view.state {
        ExecutionState::Queued | ExecutionState::Running | ExecutionState::Finalizing => {
            "in_progress"
        }
        ExecutionState::WaitingForInput => "incomplete",
        ExecutionState::Completed => "completed",
        ExecutionState::Failed => "failed",
        ExecutionState::Canceled => "cancelled",
    };
    let output = view
        .outcome
        .as_ref()
        .map(|outcome| {
            vec![OpenAiOutputItem {
                id: format!("msg_{}", view.id),
                item_type: "message".into(),
                status: "completed".into(),
                role: "assistant".into(),
                content: vec![OpenAiOutputContent {
                    content_type: "output_text".into(),
                    text: outcome.answer.clone(),
                    annotations: Vec::new(),
                }],
            }]
        })
        .unwrap_or_default();
    let usage = view.outcome.as_ref().map(|outcome| {
        let input = outcome.input_tokens.unwrap_or_default();
        let output = outcome.output_tokens.unwrap_or_default();
        OpenAiUsage {
            input_tokens: input,
            input_tokens_details: OpenAiInputTokensDetails { cached_tokens: 0 },
            output_tokens: output,
            output_tokens_details: OpenAiOutputTokensDetails {
                reasoning_tokens: 0,
            },
            total_tokens: input.saturating_add(output),
        }
    });
    let error = view.failure.as_ref().map(|failure| OpenAiResponseError {
        code: failure.code.clone(),
        message: failure.message.clone(),
    });
    OpenAiResponse {
        id: format!("resp_{}", view.id),
        object: "response".into(),
        created_at: view.created_at_ms / 1000,
        status: status.into(),
        completed_at: view
            .state
            .is_terminal()
            .then_some(view.updated_at_ms / 1000),
        error,
        incomplete_details: (view.state == ExecutionState::WaitingForInput)
            .then(|| json!({"reason":"requires_input"})),
        instructions,
        max_output_tokens,
        model,
        output,
        parallel_tool_calls: false,
        previous_response_id: None,
        tool_choice: json!("none"),
        tools: Vec::new(),
        metadata,
        usage,
    }
}

fn claude_message_from_view(
    view: runtime_api_contract::ExecutionView,
    model: String,
) -> ClaudeMessage {
    let outcome = view.outcome.expect("completed execution has an outcome");
    ClaudeMessage {
        id: format!("msg_{}", view.id),
        message_type: "message".into(),
        role: "assistant".into(),
        content: vec![ClaudeContentBlock::Text {
            text: outcome.answer,
        }],
        model,
        stop_reason: Some("end_turn".into()),
        stop_sequence: None,
        usage: ClaudeUsage {
            input_tokens: outcome.input_tokens.unwrap_or_default(),
            output_tokens: outcome.output_tokens.unwrap_or_default(),
        },
    }
}

fn openai_response_stream(
    link: Arc<dyn RuntimeLink>,
    authority: RequestAuthority,
    initial: runtime_api_contract::ExecutionView,
    model: String,
    instructions: Option<serde_json::Value>,
    max_output_tokens: Option<u32>,
    metadata: std::collections::BTreeMap<String, String>,
) -> impl Stream<Item = Result<Event, Infallible>> {
    async_stream::stream! {
        let created = openai_response_from_view(initial.clone(), model.clone(), instructions.clone(), max_output_tokens, metadata.clone());
        yield Ok(sse_json("response.created", json!({"type":"response.created","response":created,"sequence_number":0})));
        let final_view = wait_for_provider_boundary(link, &authority, &initial.id).await;
        match final_view {
            Ok(view) if view.state == ExecutionState::Completed => {
                let response = openai_response_from_view(view, model, instructions, max_output_tokens, metadata);
                let item = response.output.first().cloned().expect("completed response has output");
                let text = item.content.first().map(|content| content.text.clone()).unwrap_or_default();
                yield Ok(sse_json("response.output_item.added", json!({"type":"response.output_item.added","output_index":0,"item":{"id":item.id,"type":"message","status":"in_progress","role":"assistant","content":[]},"sequence_number":1})));
                yield Ok(sse_json("response.content_part.added", json!({"type":"response.content_part.added","item_id":item.id,"output_index":0,"content_index":0,"part":{"type":"output_text","text":"","annotations":[]},"sequence_number":2})));
                yield Ok(sse_json("response.output_text.delta", json!({"type":"response.output_text.delta","item_id":item.id,"output_index":0,"content_index":0,"delta":text,"logprobs":[],"sequence_number":3})));
                yield Ok(sse_json("response.output_text.done", json!({"type":"response.output_text.done","item_id":item.id,"output_index":0,"content_index":0,"text":text,"logprobs":[],"sequence_number":4})));
                yield Ok(sse_json("response.content_part.done", json!({"type":"response.content_part.done","item_id":item.id,"output_index":0,"content_index":0,"part":item.content[0],"sequence_number":5})));
                yield Ok(sse_json("response.output_item.done", json!({"type":"response.output_item.done","output_index":0,"item":item,"sequence_number":6})));
                yield Ok(sse_json("response.completed", json!({"type":"response.completed","response":response,"sequence_number":7})));
            }
            Ok(view) => {
                let event_type = match view.state {
                    ExecutionState::Canceled => "response.cancelled",
                    ExecutionState::WaitingForInput => "response.incomplete",
                    _ => "response.failed",
                };
                let response = openai_response_from_view(view, model, instructions, max_output_tokens, metadata);
                yield Ok(sse_json(event_type, json!({"type":event_type,"response":response,"sequence_number":1})));
            }
            Err(error) => {
                yield Ok(sse_json("error", json!({"type":"error","error":{"type":"api_error","code":"runtime_unavailable","message":error.to_string()}})));
            }
        }
    }
}

fn claude_message_stream(
    link: Arc<dyn RuntimeLink>,
    authority: RequestAuthority,
    initial: runtime_api_contract::ExecutionView,
    model: String,
) -> impl Stream<Item = Result<Event, Infallible>> {
    async_stream::stream! {
        yield Ok(sse_json("message_start", json!({
            "type":"message_start",
            "message":{"id":format!("msg_{}", initial.id),"type":"message","role":"assistant","content":[],"model":model,"stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":0,"output_tokens":0}}
        })));
        match wait_for_provider_boundary(link, &authority, &initial.id).await {
            Ok(view) if view.state == ExecutionState::Completed => {
                let message = claude_message_from_view(view, model);
                let text = match &message.content[0] { ClaudeContentBlock::Text { text } => text.clone() };
                yield Ok(sse_json("content_block_start", json!({"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}})));
                yield Ok(sse_json("content_block_delta", json!({"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":text}})));
                yield Ok(sse_json("content_block_stop", json!({"type":"content_block_stop","index":0})));
                yield Ok(sse_json("message_delta", json!({"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":message.usage.output_tokens}})));
                yield Ok(sse_json("message_stop", json!({"type":"message_stop"})));
            }
            Ok(view) => {
                let message = view.failure.map(|failure| failure.message).unwrap_or_else(|| "Agent Task was canceled".into());
                yield Ok(sse_json("error", json!({"type":"error","error":{"type":"api_error","message":message}})));
            }
            Err(error) => {
                yield Ok(sse_json("error", json!({"type":"error","error":{"type":"api_error","message":error.to_string()}})));
            }
        }
    }
}

fn sse_json(event_type: &str, value: serde_json::Value) -> Event {
    Event::default()
        .event(event_type)
        .json_data(value)
        .unwrap_or_else(|_| {
            Event::default()
                .event("error")
                .data("serialization failure")
        })
}

fn openai_error(status: StatusCode, code: &str, message: &str, param: Option<&str>) -> Response {
    (
        status,
        Json(OpenAiErrorEnvelope {
            error: OpenAiError {
                message: message.into(),
                error_type: if status == StatusCode::UNAUTHORIZED {
                    "authentication_error".into()
                } else {
                    "invalid_request_error".into()
                },
                param: param.map(ToOwned::to_owned),
                code: Some(code.into()),
            },
        }),
    )
        .into_response()
}

fn openai_link_error(error: LinkError) -> Response {
    match error {
        LinkError::Invalid(message) => {
            openai_error(StatusCode::BAD_REQUEST, "invalid_request", &message, None)
        }
        LinkError::NotFound => openai_error(
            StatusCode::NOT_FOUND,
            "not_found",
            "response not found",
            None,
        ),
        LinkError::Forbidden => openai_error(
            StatusCode::FORBIDDEN,
            "permission_denied",
            "caller cannot access this response",
            None,
        ),
        LinkError::Conflict(message) => {
            openai_error(StatusCode::CONFLICT, "conflict", &message, None)
        }
        LinkError::Overloaded => openai_error(
            StatusCode::TOO_MANY_REQUESTS,
            "rate_limit_exceeded",
            "runtime is at capacity",
            None,
        ),
        LinkError::Unavailable(message) | LinkError::Internal(message) => openai_error(
            StatusCode::SERVICE_UNAVAILABLE,
            "server_error",
            &message,
            None,
        ),
    }
}

fn claude_error(status: StatusCode, error_type: &str, message: &str) -> Response {
    (
        status,
        Json(ClaudeErrorEnvelope {
            envelope_type: "error".into(),
            error: ClaudeError {
                error_type: error_type.into(),
                message: message.into(),
            },
            request_id: None,
        }),
    )
        .into_response()
}

fn claude_link_error(error: LinkError) -> Response {
    match error {
        LinkError::Invalid(message) => {
            claude_error(StatusCode::BAD_REQUEST, "invalid_request_error", &message)
        }
        LinkError::NotFound => claude_error(
            StatusCode::NOT_FOUND,
            "not_found_error",
            "message task not found",
        ),
        LinkError::Forbidden => claude_error(
            StatusCode::FORBIDDEN,
            "permission_error",
            "caller cannot access this message task",
        ),
        LinkError::Overloaded => claude_error(
            StatusCode::TOO_MANY_REQUESTS,
            "rate_limit_error",
            "runtime is at capacity",
        ),
        LinkError::Conflict(message) => {
            claude_error(StatusCode::CONFLICT, "invalid_request_error", &message)
        }
        LinkError::Unavailable(message) | LinkError::Internal(message) => {
            claude_error(StatusCode::SERVICE_UNAVAILABLE, "api_error", &message)
        }
    }
}

fn claude_terminal_error(view: runtime_api_contract::ExecutionView) -> Response {
    let message = view
        .failure
        .map(|failure| failure.message)
        .unwrap_or_else(|| "Agent Task was canceled".into());
    claude_error(StatusCode::INTERNAL_SERVER_ERROR, "api_error", &message)
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct EventQuery {
    after: Option<u64>,
    limit: Option<usize>,
}

async fn websocket_agent_task(
    upgrade: WebSocketUpgrade,
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(id): Path<String>,
    Query(query): Query<EventQuery>,
) -> Response {
    let authority = match authenticate(&state, &headers).await {
        Ok(value) => value,
        Err(response) => return response,
    };
    let id = match parse_id(id) {
        Ok(value) => value,
        Err(error) => return invalid_task_id(error),
    };
    let task = match state.link.execution(&authority, &id).await {
        Ok(value) => AgentTask::from(value),
        Err(error) => return link_error(error),
    };
    upgrade
        .on_upgrade(move |socket| {
            websocket_agent_task_session(socket, state.link, authority, id, task, query.after)
        })
        .into_response()
}

async fn websocket_agent_task_session(
    mut socket: WebSocket,
    link: Arc<dyn RuntimeLink>,
    authority: RequestAuthority,
    id: ExecutionId,
    task: AgentTask,
    mut after: Option<u64>,
) {
    if send_agent_task_websocket(
        &mut socket,
        &AgentTaskWebSocketServerMessage::Connected {
            task,
            next_after: after,
            heartbeat_seconds: 15,
        },
    )
    .await
    .is_err()
    {
        return;
    }

    let mut poll = tokio::time::interval(Duration::from_millis(100));
    poll.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    let mut heartbeat = tokio::time::interval(Duration::from_secs(15));
    heartbeat.tick().await;
    loop {
        tokio::select! {
            incoming = socket.next() => {
                match incoming {
                    Some(Ok(Message::Text(text))) => {
                        let command = match serde_json::from_str::<AgentTaskWebSocketClientMessage>(&text) {
                            Ok(command) => command,
                            Err(_) => {
                                let _ = send_agent_task_websocket_error(&mut socket, "WEBSOCKET_MESSAGE_INVALID", "Agent Task WebSocket command is invalid").await;
                                continue;
                            }
                        };
                        let result = match command {
                            AgentTaskWebSocketClientMessage::SubmitInput { operation_id, input } => {
                                Some(link.submit_input(&authority, &id, SubmitInputRequest { operation_id, input }).await)
                            }
                            AgentTaskWebSocketClientMessage::Cancel {} => {
                                Some(link.cancel(&authority, &id).await)
                            }
                            AgentTaskWebSocketClientMessage::Ack { sequence } => {
                                // Acknowledgements are monotonic client resume hints. Never
                                // move the delivery cursor backwards or skip unsent events.
                                if after.is_none_or(|delivered| sequence > delivered) {
                                    let _ = send_agent_task_websocket_error(&mut socket, "ACK_SEQUENCE_INVALID", "cannot acknowledge an event that has not been delivered").await;
                                } else if send_agent_task_websocket(
                                    &mut socket,
                                    &AgentTaskWebSocketServerMessage::Acknowledged { sequence },
                                ).await.is_err() {
                                    return;
                                }
                                None
                            }
                        };
                        if let Some(result) = result {
                            match result {
                                Ok(view) => {
                                    if send_agent_task_websocket(&mut socket, &AgentTaskWebSocketServerMessage::Task { task: AgentTask::from(view) }).await.is_err() {
                                        return;
                                    }
                                }
                                Err(_) => {
                                    if send_agent_task_websocket_error(&mut socket, "WEBSOCKET_COMMAND_FAILED", "Agent Task command could not be committed").await.is_err() {
                                        return;
                                    }
                                }
                            }
                        }
                    }
                    Some(Ok(Message::Ping(value))) => {
                        if socket.send(Message::Pong(value)).await.is_err() { return; }
                    }
                    Some(Ok(Message::Close(_))) | None | Some(Err(_)) => return,
                    Some(Ok(Message::Binary(_))) | Some(Ok(Message::Pong(_))) => {}
                }
            }
            _ = poll.tick() => {
                let page = match link.events(&authority, &id, after, 100).await {
                    Ok(page) => page,
                    Err(_) => {
                        let _ = send_agent_task_websocket_error(&mut socket, "WEBSOCKET_EVENTS_UNAVAILABLE", "Agent Task events are unavailable").await;
                        return;
                    }
                };
                let empty = page.items.is_empty();
                for item in page.items {
                    after = Some(item.sequence);
                    if send_agent_task_websocket(
                        &mut socket,
                        &AgentTaskWebSocketServerMessage::Event { event: AgentTaskEvent::from(item) },
                    ).await.is_err() {
                        return;
                    }
                }
                if empty {
                    match link.execution(&authority, &id).await {
                        Ok(view) if view.state.is_terminal() => {
                            let _ = send_agent_task_websocket(&mut socket, &AgentTaskWebSocketServerMessage::Task { task: AgentTask::from(view) }).await;
                            let _ = socket.close().await;
                            return;
                        }
                        Err(_) => return,
                        _ => {}
                    }
                }
            }
            _ = heartbeat.tick() => {
                if socket.send(Message::Ping(Vec::new().into())).await.is_err() {
                    return;
                }
            }
        }
    }
}

async fn send_agent_task_websocket(
    socket: &mut WebSocket,
    message: &AgentTaskWebSocketServerMessage,
) -> Result<(), axum::Error> {
    let json = serde_json::to_string(message).map_err(axum::Error::new)?;
    socket.send(Message::Text(json.into())).await
}

async fn send_agent_task_websocket_error(
    socket: &mut WebSocket,
    code: &str,
    message: &str,
) -> Result<(), axum::Error> {
    send_agent_task_websocket(
        socket,
        &AgentTaskWebSocketServerMessage::Error {
            error: ApiErrorBody {
                code: code.into(),
                message: message.into(),
                request_id: None,
                details: None,
            },
        },
    )
    .await
}

fn agent_task_event_stream(
    link: Arc<dyn RuntimeLink>,
    authority: RequestAuthority,
    id: ExecutionId,
    initial_after: Option<u64>,
) -> impl Stream<Item = Result<Event, Infallible>> {
    async_stream::stream! {
        let mut after = initial_after;
        loop {
            match link.events(&authority, &id, after, 100).await {
                Ok(page) => {
                    let empty = page.items.is_empty();
                    for item in page.items {
                        after = Some(item.sequence);
                        let event = AgentTaskEvent::from(item);
                        let frame = Event::default()
                            .id(event.sequence.to_string())
                            .event(event.kind.clone())
                            .json_data(&event)
                            .unwrap_or_else(|_| Event::default().event("task.error").data("serialization failure"));
                        yield Ok(frame);
                    }
                    if empty {
                        match link.execution(&authority, &id).await {
                            Ok(view) if view.state.is_terminal() => break,
                            Err(_) => break,
                            _ => {}
                        }
                    }
                }
                Err(error) => {
                    yield Ok(Event::default().event("task.error").data(error.to_string()));
                    break;
                }
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
    }
}

fn idempotency_key_or_random(headers: &HeaderMap, prefix: &str) -> String {
    headers
        .get("idempotency-key")
        .and_then(|value| value.to_str().ok())
        .filter(|value| !value.is_empty())
        .map(ToOwned::to_owned)
        .unwrap_or_else(|| format!("{prefix}:{}", uuid::Uuid::new_v4()))
}

fn request_id(headers: &HeaderMap) -> Option<&str> {
    headers
        .get("x-request-id")
        .and_then(|value| value.to_str().ok())
}

fn sse_resume_after(headers: &HeaderMap, query_after: Option<u64>) -> Result<Option<u64>, String> {
    let header_after = headers
        .get("last-event-id")
        .map(|value| {
            value
                .to_str()
                .map_err(|_| "Last-Event-ID must be ASCII".to_string())?
                .parse::<u64>()
                .map_err(|_| "Last-Event-ID must be an unsigned integer".to_string())
        })
        .transpose()?;
    if query_after.is_some() && header_after.is_some() && query_after != header_after {
        return Err("after and Last-Event-ID must identify the same sequence".into());
    }
    Ok(query_after.or(header_after))
}

fn attach_request_id_metadata(
    headers: &HeaderMap,
    metadata: &mut std::collections::BTreeMap<String, String>,
) {
    if let Some(request_id) = request_id(headers) {
        metadata
            .entry("requestId".into())
            .or_insert_with(|| request_id.to_string());
    }
}

fn invalid_task_id(error: runtime_types::IdentifierError) -> Response {
    api_error(
        StatusCode::BAD_REQUEST,
        "TASK_ID_INVALID",
        &error.to_string(),
    )
}

async fn authenticate(state: &AppState, headers: &HeaderMap) -> Result<RequestAuthority, Response> {
    state
        .authenticator
        .authenticate(headers)
        .await
        .map_err(|_| {
            api_error(
                StatusCode::UNAUTHORIZED,
                "UNAUTHENTICATED",
                "request authentication failed",
            )
        })
}

fn parse_id(value: String) -> Result<ExecutionId, runtime_types::IdentifierError> {
    ExecutionId::new(value)
}

fn parse_openai_response_id(value: String) -> Result<ExecutionId, runtime_types::IdentifierError> {
    ExecutionId::new(value.strip_prefix("resp_").unwrap_or(&value).to_string())
}

fn link_error(error: LinkError) -> Response {
    match error {
        LinkError::Invalid(message) => {
            api_error(StatusCode::BAD_REQUEST, "INVALID_REQUEST", &message)
        }
        LinkError::NotFound => api_error(
            StatusCode::NOT_FOUND,
            "EXECUTION_NOT_FOUND",
            "execution not found",
        ),
        LinkError::Forbidden => api_error(
            StatusCode::FORBIDDEN,
            "FORBIDDEN",
            "caller cannot access this execution",
        ),
        LinkError::Conflict(message) => {
            api_error(StatusCode::CONFLICT, "EXECUTION_CONFLICT", &message)
        }
        LinkError::Overloaded => api_error(
            StatusCode::TOO_MANY_REQUESTS,
            "RUNTIME_OVERLOADED",
            "runtime is at capacity",
        ),
        LinkError::Unavailable(message) => api_error(
            StatusCode::SERVICE_UNAVAILABLE,
            "DEPENDENCY_UNAVAILABLE",
            &message,
        ),
        LinkError::Internal(message) => {
            tracing::error!(%message, "runtime internal error");
            api_error(
                StatusCode::INTERNAL_SERVER_ERROR,
                "INTERNAL",
                "runtime internal failure",
            )
        }
    }
}

fn api_error(status: StatusCode, code: &str, message: &str) -> Response {
    (
        status,
        Json(ApiErrorBody {
            code: code.into(),
            message: message.into(),
            request_id: None,
            details: None,
        }),
    )
        .into_response()
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use axum::{body::Body, http::Request};
    use http_body_util::BodyExt;
    use runtime_api_contract::{CreateExecutionResponse, EventPage, ExecutionState, ExecutionView};
    use runtime_types::RuntimeInstanceId;
    use tokio_tungstenite::{
        connect_async,
        tungstenite::{Message as ClientMessage, client::IntoClientRequest},
    };
    use tower::ServiceExt;

    use super::*;

    struct FakeLink {
        terminal: bool,
    }

    fn view(terminal: bool) -> ExecutionView {
        ExecutionView {
            id: ExecutionId::new("execution-1").unwrap(),
            runtime_instance_id: RuntimeInstanceId::new("runtime-1").unwrap(),
            conversation_id: None,
            workspace_id: None,
            model: None,
            metadata: Default::default(),
            state: if terminal {
                ExecutionState::Completed
            } else {
                ExecutionState::Queued
            },
            outcome: terminal.then_some(runtime_api_contract::ExecutionOutcome {
                answer: "complete".into(),
                model_turns: 1,
                tool_calls: 0,
                input_tokens: Some(10),
                output_tokens: Some(2),
            }),
            failure: None,
            created_at_ms: 1,
            updated_at_ms: 1,
        }
    }

    #[async_trait]
    impl RuntimeLink for FakeLink {
        async fn create_execution(
            &self,
            _: &RequestAuthority,
            _: &str,
            request: CreateExecutionRequest,
        ) -> Result<CreateExecutionResponse, LinkError> {
            let mut execution = view(self.terminal);
            execution.workspace_id = request.workspace_id;
            execution.model = request.model;
            execution.metadata = request.metadata;
            Ok(CreateExecutionResponse {
                execution,
                replayed: false,
            })
        }
        async fn execution(
            &self,
            _: &RequestAuthority,
            _: &ExecutionId,
        ) -> Result<ExecutionView, LinkError> {
            Ok(view(self.terminal))
        }
        async fn events(
            &self,
            _: &RequestAuthority,
            _: &ExecutionId,
            _: Option<u64>,
            _: usize,
        ) -> Result<EventPage, LinkError> {
            Ok(EventPage {
                items: Vec::new(),
                next_after: None,
            })
        }
        async fn submit_input(
            &self,
            _: &RequestAuthority,
            _: &ExecutionId,
            _: SubmitInputRequest,
        ) -> Result<ExecutionView, LinkError> {
            Ok(view(self.terminal))
        }
        async fn cancel(
            &self,
            _: &RequestAuthority,
            _: &ExecutionId,
        ) -> Result<ExecutionView, LinkError> {
            Ok(view(self.terminal))
        }
    }

    fn app() -> Router {
        app_with_terminal(false)
    }

    fn app_with_terminal(terminal: bool) -> Router {
        let auth = StaticTokenAuthenticator::new(
            "token",
            CallerScope {
                subject: "user".into(),
                tenant_id: "tenant".into(),
                project_id: "project".into(),
                capabilities: Vec::new(),
            },
        )
        .unwrap();
        router(
            Arc::new(FakeLink { terminal }),
            Arc::new(auth),
            Arc::new(AlwaysReady),
            RuntimeTransportConfig {
                agent_tasks: true,
                openai_responses: true,
                claude_messages: true,
                sse: true,
                websocket: true,
                websocket_max_message_bytes: 256 * 1024,
            },
        )
    }

    fn create_request(authenticated: bool) -> Request<Body> {
        let mut request =
            Request::post("/v1/agent/tasks").header("content-type", "application/json");
        if authenticated {
            request = request.header("authorization", "Bearer token");
        }
        request.body(Body::from(r#"{"runtimeId":"runtime-1","input":{"text":"hello"},"limits":{"deadlineSeconds":60,"maxModelTurns":4,"maxToolCalls":8}}"#)).unwrap()
    }

    #[tokio::test]
    async fn create_requires_authentication_and_returns_contract_body() {
        let unauthorized = app().oneshot(create_request(false)).await.unwrap();
        assert_eq!(unauthorized.status(), StatusCode::UNAUTHORIZED);

        let response = app().oneshot(create_request(true)).await.unwrap();
        assert_eq!(response.status(), StatusCode::ACCEPTED);
        assert!(response.headers().contains_key("x-request-id"));
        let body = response.into_body().collect().await.unwrap().to_bytes();
        let body: serde_json::Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(body["id"], "execution-1");
    }

    #[tokio::test]
    async fn native_task_creation_generates_idempotency_and_returns_task_links() {
        let response = app()
            .oneshot(
                Request::post("/v1/agent/tasks")
                    .header("authorization", "Bearer token")
                    .header("content-type", "application/json")
                    .body(Body::from(
                        r#"{"runtimeId":"runtime-1","input":{"text":"hello"},"workspace":{"id":"workspace-2"}}"#,
                    ))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::ACCEPTED);
        assert_eq!(
            response.headers().get(header::LOCATION).unwrap(),
            "/v1/agent/tasks/execution-1"
        );
        assert_eq!(response.headers().get(header::RETRY_AFTER).unwrap(), "1");
        let body = response.into_body().collect().await.unwrap().to_bytes();
        let body: serde_json::Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(body["object"], "agent.task");
        assert_eq!(body["workspace"]["id"], "workspace-2");
        assert!(body["metadata"]["requestId"].is_string());
        assert_eq!(
            body["links"]["stream"],
            "/v1/agent/tasks/execution-1/stream"
        );
    }

    #[tokio::test]
    async fn openai_background_response_uses_openai_envelope() {
        let response = app()
            .oneshot(
                Request::post("/v1/responses")
                    .header("authorization", "Bearer token")
                    .header("content-type", "application/json")
                    .body(Body::from(
                        r#"{"model":"runtime-1","input":"hello","background":true,"max_output_tokens":256}"#,
                    ))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        let body = response.into_body().collect().await.unwrap().to_bytes();
        let body: serde_json::Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(body["object"], "response");
        assert!(body["id"].as_str().unwrap().starts_with("resp_"));
        assert_eq!(body["status"], "in_progress");
        assert_eq!(body["max_output_tokens"], 256);
    }

    #[test]
    fn provider_parameters_fuse_into_one_generation_contract() {
        let openai: OpenAiResponseRequest = serde_json::from_value(json!({
            "model":"runtime-1",
            "input":"hello",
            "max_output_tokens":512,
            "temperature":0.25,
            "text":{"format":{"type":"json_object"}},
            "tool_choice":"none"
        }))
        .unwrap();
        let execution = openai_execution_request(&openai).unwrap();
        assert_eq!(execution.generation.max_output_tokens, Some(512));
        assert_eq!(execution.generation.temperature, Some(0.25));
        assert_eq!(
            execution.generation.response_format,
            Some(json!({"type":"json_object"}))
        );
        assert_eq!(execution.generation.tool_choice, ModelToolChoice::None);

        let claude: ClaudeMessageRequest = serde_json::from_value(json!({
            "model":"runtime-1",
            "max_tokens":256,
            "messages":[{"role":"user","content":"hello"}],
            "temperature":0.4,
            "stop_sequences":["</done>"],
            "tool_choice":{"type":"none"}
        }))
        .unwrap();
        let execution = claude_execution_request(&claude).unwrap();
        assert_eq!(execution.generation.max_output_tokens, Some(256));
        assert_eq!(execution.generation.temperature, Some(0.4));
        assert_eq!(execution.generation.stop_sequences, ["</done>"]);
        assert_eq!(execution.generation.tool_choice, ModelToolChoice::None);
        assert_eq!(execution.generation.response_format, None);
    }

    #[tokio::test]
    async fn provider_adapters_reject_fields_the_kernel_cannot_preserve() {
        let openai = app()
            .oneshot(
                Request::post("/v1/responses")
                    .header("authorization", "Bearer token")
                    .header("content-type", "application/json")
                    .body(Body::from(
                        r#"{"model":"runtime-1","input":"hello","tools":[{"type":"function","name":"remote"}]}"#,
                    ))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(openai.status(), StatusCode::BAD_REQUEST);
        let body = openai.into_body().collect().await.unwrap().to_bytes();
        let body: serde_json::Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(body["error"]["param"], "tools");

        let claude = app()
            .oneshot(
                Request::post("/v1/messages")
                    .header("authorization", "Bearer token")
                    .header("content-type", "application/json")
                    .body(Body::from(
                        r#"{"model":"runtime-1","max_tokens":128,"messages":[{"role":"user","content":"hello"}]}"#,
                    ))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(claude.status(), StatusCode::BAD_REQUEST);
        let body = claude.into_body().collect().await.unwrap().to_bytes();
        let body: serde_json::Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(body["type"], "error");
        assert_eq!(body["error"]["type"], "invalid_request_error");
    }

    #[tokio::test]
    async fn claude_sdk_style_api_key_authentication_is_accepted() {
        let response = app_with_terminal(true)
            .oneshot(
                Request::post("/v1/messages")
                    .header("x-api-key", "token")
                    .header("anthropic-version", "2023-06-01")
                    .header("content-type", "application/json")
                    .body(Body::from(
                        r#"{"model":"runtime-1","max_tokens":128,"messages":[{"role":"user","content":"hello"}]}"#,
                    ))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        let body = response.into_body().collect().await.unwrap().to_bytes();
        let body: serde_json::Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(body["type"], "message");
    }

    #[tokio::test]
    async fn provider_sse_uses_native_event_names_and_terminal_order() {
        let openai = app_with_terminal(true)
            .oneshot(
                Request::post("/v1/responses")
                    .header("authorization", "Bearer token")
                    .header("content-type", "application/json")
                    .body(Body::from(
                        r#"{"model":"runtime-1","input":"complete","stream":true}"#,
                    ))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(openai.status(), StatusCode::OK);
        let openai = String::from_utf8(
            openai
                .into_body()
                .collect()
                .await
                .unwrap()
                .to_bytes()
                .to_vec(),
        )
        .unwrap();
        let created = openai.find("event: response.created").unwrap();
        let delta = openai.find("event: response.output_text.delta").unwrap();
        let completed = openai.find("event: response.completed").unwrap();
        assert!(created < delta && delta < completed);

        let claude = app_with_terminal(true)
            .oneshot(
                Request::post("/v1/messages")
                    .header("authorization", "Bearer token")
                    .header("anthropic-version", "2023-06-01")
                    .header("content-type", "application/json")
                    .body(Body::from(
                        r#"{"model":"runtime-1","max_tokens":128,"messages":[{"role":"user","content":"complete"}],"stream":true}"#,
                    ))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(claude.status(), StatusCode::OK);
        let claude = String::from_utf8(
            claude
                .into_body()
                .collect()
                .await
                .unwrap()
                .to_bytes()
                .to_vec(),
        )
        .unwrap();
        let start = claude.find("event: message_start").unwrap();
        let delta = claude.find("event: content_block_delta").unwrap();
        let stop = claude.find("event: message_stop").unwrap();
        assert!(start < delta && delta < stop);
    }

    #[test]
    fn sse_resume_accepts_standard_header_and_rejects_ambiguous_cursors() {
        let mut headers = HeaderMap::new();
        headers.insert("last-event-id", "17".parse().unwrap());
        assert_eq!(sse_resume_after(&headers, None).unwrap(), Some(17));
        assert_eq!(sse_resume_after(&headers, Some(17)).unwrap(), Some(17));
        assert!(sse_resume_after(&headers, Some(18)).is_err());
    }

    #[test]
    fn provider_message_history_preserves_roles_for_the_kernel() {
        let openai: OpenAiResponseRequest = serde_json::from_value(json!({
            "model": "runtime-1",
            "input": [
                {"role":"user","content":"question"},
                {"role":"assistant","content":"draft"},
                {"role":"user","content":"continue"}
            ]
        }))
        .unwrap();
        let execution = openai_execution_request(&openai).unwrap();
        let RuntimeInput::Messages { messages } = execution.input else {
            panic!("provider history must remain structured");
        };
        assert_eq!(
            messages
                .iter()
                .map(|message| message.role)
                .collect::<Vec<_>>(),
            vec![
                RuntimeMessageRole::User,
                RuntimeMessageRole::Assistant,
                RuntimeMessageRole::User,
            ]
        );
    }

    #[tokio::test]
    async fn trusted_gateway_scope_requires_the_shared_ingress_secret() {
        let authenticator =
            TrustedGatewayAuthenticator::new("a-runtime-gateway-secret-with-at-least-32-bytes")
                .unwrap();
        let mut headers = HeaderMap::new();
        headers.insert("authorization", "Bearer delegated-token".parse().unwrap());
        headers.insert("x-runtime-subject", "user".parse().unwrap());
        headers.insert("x-runtime-tenant-id", "tenant".parse().unwrap());
        headers.insert("x-runtime-project-id", "project".parse().unwrap());
        assert!(authenticator.authenticate(&headers).await.is_err());
        headers.insert(
            "x-runtime-gateway-secret",
            "a-runtime-gateway-secret-with-at-least-32-bytes"
                .parse()
                .unwrap(),
        );
        let authority = authenticator.authenticate(&headers).await.unwrap();
        assert_eq!(authority.caller.subject, "user");
    }

    #[tokio::test]
    async fn transport_routes_are_registered_only_when_enabled() {
        let auth = StaticTokenAuthenticator::new(
            "token",
            CallerScope {
                subject: "user".into(),
                tenant_id: "tenant".into(),
                project_id: "project".into(),
                capabilities: Vec::new(),
            },
        )
        .unwrap();
        let app = router(
            Arc::new(FakeLink { terminal: false }),
            Arc::new(auth),
            Arc::new(AlwaysReady),
            RuntimeTransportConfig {
                agent_tasks: true,
                openai_responses: false,
                claude_messages: false,
                sse: false,
                websocket: false,
                websocket_max_message_bytes: 1024,
            },
        );
        let stream = app
            .clone()
            .oneshot(
                Request::get("/v1/agent/tasks/execution-1/stream")
                    .header("authorization", "Bearer token")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(stream.status(), StatusCode::NOT_FOUND);
        let openai = app
            .clone()
            .oneshot(Request::post("/v1/responses").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(openai.status(), StatusCode::NOT_FOUND);
        let openapi = app
            .oneshot(Request::get("/openapi.json").body(Body::empty()).unwrap())
            .await
            .unwrap();
        let body = openapi.into_body().collect().await.unwrap().to_bytes();
        let document: serde_json::Value = serde_json::from_slice(&body).unwrap();
        assert!(
            document["paths"]
                .get("/v1/agent/tasks/{taskId}/stream")
                .is_none()
        );
        assert!(
            document["paths"]
                .get("/v1/agent/tasks/{taskId}/ws")
                .is_none()
        );
        assert!(document["paths"].get("/v1/responses").is_none());
        assert!(document["paths"].get("/v1/messages").is_none());
    }

    #[tokio::test]
    async fn websocket_is_authenticated_and_accepts_scoped_commands() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let server = tokio::spawn(async move { axum::serve(listener, app()).await.unwrap() });

        let mut request = format!("ws://{address}/v1/agent/tasks/execution-1/ws")
            .into_client_request()
            .unwrap();
        request
            .headers_mut()
            .insert("authorization", "Bearer token".parse().unwrap());
        let (mut socket, response) = connect_async(request).await.unwrap();
        assert_eq!(response.status(), 101);

        let connected = socket.next().await.unwrap().unwrap().into_text().unwrap();
        let connected: AgentTaskWebSocketServerMessage = serde_json::from_str(&connected).unwrap();
        assert!(matches!(
            connected,
            AgentTaskWebSocketServerMessage::Connected { .. }
        ));

        socket
            .send(ClientMessage::Text(r#"{"type":"cancel"}"#.into()))
            .await
            .unwrap();
        let execution = socket.next().await.unwrap().unwrap().into_text().unwrap();
        let execution: AgentTaskWebSocketServerMessage = serde_json::from_str(&execution).unwrap();
        assert!(matches!(
            execution,
            AgentTaskWebSocketServerMessage::Task { .. }
        ));
        socket.close(None).await.unwrap();
        server.abort();
    }
}