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>;
}
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(),
})?,
})
}
}
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(())
}
}
#[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,
}
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
.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)
}
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())
}
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 } => {
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();
}
}