agent_runtime_http_server/
lib.rs

1//! Axum transport for the Runtime Open API.
2
3use std::{convert::Infallible, sync::Arc, time::Duration};
4
5use async_trait::async_trait;
6use axum::{
7    Json, Router,
8    extract::{DefaultBodyLimit, Path, Query, State},
9    http::{HeaderMap, StatusCode, header},
10    response::{
11        IntoResponse, Response, Sse,
12        sse::{Event, KeepAlive},
13    },
14    routing::{get, post},
15};
16use futures_util::Stream;
17use runtime_api_contract::{
18    ApiErrorBody, CreateExecutionRequest, SubmitInputRequest, openapi_document,
19};
20use runtime_link_api::{LinkError, RuntimeLink};
21use runtime_types::{CallerScope, CredentialHandle, ExecutionId, RequestAuthority};
22use serde::Deserialize;
23use serde_json::json;
24use sha2::{Digest, Sha256};
25
26#[derive(Debug)]
27pub struct AuthError {
28    pub message: String,
29}
30
31#[async_trait]
32pub trait IngressAuthenticator: Send + Sync + 'static {
33    async fn authenticate(&self, headers: &HeaderMap) -> Result<RequestAuthority, AuthError>;
34}
35
36/// Safe single-tenant authenticator for local/container deployments.
37/// Production multi-tenant hosts should inject a JWT/introspection adapter.
38pub struct StaticBearerAuthenticator {
39    expected_token: String,
40    caller: CallerScope,
41}
42
43impl StaticBearerAuthenticator {
44    pub fn new(expected_token: impl Into<String>, caller: CallerScope) -> Result<Self, AuthError> {
45        let expected_token = expected_token.into();
46        if expected_token.is_empty() {
47            return Err(AuthError {
48                message: "static bearer token must not be empty".into(),
49            });
50        }
51        caller.validate().map_err(|error| AuthError {
52            message: error.to_string(),
53        })?;
54        Ok(Self {
55            expected_token,
56            caller,
57        })
58    }
59}
60
61#[async_trait]
62impl IngressAuthenticator for StaticBearerAuthenticator {
63    async fn authenticate(&self, headers: &HeaderMap) -> Result<RequestAuthority, AuthError> {
64        let value = headers
65            .get(header::AUTHORIZATION)
66            .and_then(|value| value.to_str().ok())
67            .and_then(|value| value.strip_prefix("Bearer "))
68            .ok_or_else(|| AuthError {
69                message: "missing bearer credential".into(),
70            })?;
71        if value != self.expected_token {
72            return Err(AuthError {
73                message: "invalid bearer credential".into(),
74            });
75        }
76        Ok(RequestAuthority {
77            caller: self.caller.clone(),
78            credential: CredentialHandle::new(value).map_err(|error| AuthError {
79                message: error.to_string(),
80            })?,
81        })
82    }
83}
84
85/// Authenticator for a Runtime listener reachable only through a trusted
86/// identity-aware gateway. The gateway owns credential verification and
87/// injects the caller scope headers; Runtime forwards the bearer through an
88/// opaque credential handle to its scoped Agent Infra client.
89pub struct TrustedGatewayAuthenticator {
90    shared_secret_digest: [u8; 32],
91}
92
93impl TrustedGatewayAuthenticator {
94    pub fn new(shared_secret: &str) -> Result<Self, AuthError> {
95        if shared_secret.len() < 32 || shared_secret.len() > 4096 {
96            return Err(AuthError {
97                message: "trusted gateway secret must contain 32..=4096 bytes".into(),
98            });
99        }
100        Ok(Self {
101            shared_secret_digest: Sha256::digest(shared_secret.as_bytes()).into(),
102        })
103    }
104}
105
106#[async_trait]
107impl IngressAuthenticator for TrustedGatewayAuthenticator {
108    async fn authenticate(&self, headers: &HeaderMap) -> Result<RequestAuthority, AuthError> {
109        fn required(headers: &HeaderMap, name: &'static str) -> Result<String, AuthError> {
110            headers
111                .get(name)
112                .and_then(|value| value.to_str().ok())
113                .filter(|value| !value.is_empty())
114                .map(ToOwned::to_owned)
115                .ok_or_else(|| AuthError {
116                    message: format!("trusted gateway header {name} is missing"),
117                })
118        }
119        let supplied_secret = required(headers, "x-runtime-gateway-secret")?;
120        let supplied_digest: [u8; 32] = Sha256::digest(supplied_secret.as_bytes()).into();
121        let matches = self
122            .shared_secret_digest
123            .iter()
124            .zip(supplied_digest)
125            .fold(0_u8, |difference, (expected, actual)| {
126                difference | (*expected ^ actual)
127            })
128            == 0;
129        if !matches {
130            return Err(AuthError {
131                message: "trusted gateway authentication failed".into(),
132            });
133        }
134        let token = headers
135            .get(header::AUTHORIZATION)
136            .and_then(|value| value.to_str().ok())
137            .and_then(|value| value.strip_prefix("Bearer "))
138            .ok_or_else(|| AuthError {
139                message: "trusted gateway bearer credential is missing".into(),
140            })?;
141        let caller = CallerScope {
142            subject: required(headers, "x-runtime-subject")?,
143            tenant_id: required(headers, "x-runtime-tenant-id")?,
144            project_id: required(headers, "x-runtime-project-id")?,
145            capabilities: headers
146                .get("x-runtime-capabilities")
147                .and_then(|value| value.to_str().ok())
148                .unwrap_or_default()
149                .split(',')
150                .map(str::trim)
151                .filter(|value| !value.is_empty())
152                .map(ToOwned::to_owned)
153                .collect(),
154        };
155        caller.validate().map_err(|error| AuthError {
156            message: error.to_string(),
157        })?;
158        Ok(RequestAuthority {
159            caller,
160            credential: CredentialHandle::new(token).map_err(|error| AuthError {
161                message: error.to_string(),
162            })?,
163        })
164    }
165}
166
167#[async_trait]
168pub trait ReadinessProbe: Send + Sync + 'static {
169    async fn ready(&self) -> Result<(), String>;
170}
171
172pub struct AlwaysReady;
173#[async_trait]
174impl ReadinessProbe for AlwaysReady {
175    async fn ready(&self) -> Result<(), String> {
176        Ok(())
177    }
178}
179
180#[derive(Clone)]
181struct AppState {
182    link: Arc<dyn RuntimeLink>,
183    authenticator: Arc<dyn IngressAuthenticator>,
184    readiness: Arc<dyn ReadinessProbe>,
185}
186
187pub fn router(
188    link: Arc<dyn RuntimeLink>,
189    authenticator: Arc<dyn IngressAuthenticator>,
190    readiness: Arc<dyn ReadinessProbe>,
191) -> Router {
192    Router::new()
193        .route("/livez", get(live))
194        .route("/readyz", get(ready))
195        .route("/openapi.json", get(openapi))
196        .route("/v1/executions", post(create_execution))
197        .route("/v1/executions/{execution_id}", get(get_execution))
198        .route("/v1/executions/{execution_id}/events", get(get_events))
199        .route("/v1/executions/{execution_id}/stream", get(stream_events))
200        .route("/v1/executions/{execution_id}/inputs", post(submit_input))
201        .route(
202            "/v1/executions/{execution_id}/cancel",
203            post(cancel_execution),
204        )
205        .layer(DefaultBodyLimit::max(1024 * 1024))
206        .with_state(AppState {
207            link,
208            authenticator,
209            readiness,
210        })
211}
212
213async fn live() -> Json<serde_json::Value> {
214    Json(json!({"status":"ok","service":"runtime-host"}))
215}
216
217async fn openapi() -> Json<serde_json::Value> {
218    Json(openapi_document())
219}
220
221async fn ready(State(state): State<AppState>) -> Response {
222    match state.readiness.ready().await {
223        Ok(()) => Json(json!({"status":"ready"})).into_response(),
224        Err(message) => (
225            StatusCode::SERVICE_UNAVAILABLE,
226            Json(json!({"status":"not_ready","message":message})),
227        )
228            .into_response(),
229    }
230}
231
232async fn create_execution(
233    State(state): State<AppState>,
234    headers: HeaderMap,
235    Json(request): Json<CreateExecutionRequest>,
236) -> Response {
237    let authority = match authenticate(&state, &headers).await {
238        Ok(value) => value,
239        Err(response) => return response,
240    };
241    let key = match headers
242        .get("idempotency-key")
243        .and_then(|value| value.to_str().ok())
244    {
245        Some(value) => value,
246        None => {
247            return api_error(
248                StatusCode::BAD_REQUEST,
249                "IDEMPOTENCY_KEY_REQUIRED",
250                "Idempotency-Key header is required",
251            );
252        }
253    };
254    match state.link.create_execution(&authority, key, request).await {
255        Ok(value) => (StatusCode::ACCEPTED, Json(value)).into_response(),
256        Err(error) => link_error(error),
257    }
258}
259
260async fn get_execution(
261    State(state): State<AppState>,
262    headers: HeaderMap,
263    Path(id): Path<String>,
264) -> Response {
265    let authority = match authenticate(&state, &headers).await {
266        Ok(value) => value,
267        Err(response) => return response,
268    };
269    let id = match parse_id(id) {
270        Ok(value) => value,
271        Err(error) => return invalid_execution_id(error),
272    };
273    match state.link.execution(&authority, &id).await {
274        Ok(value) => Json(value).into_response(),
275        Err(error) => link_error(error),
276    }
277}
278
279#[derive(Debug, Deserialize)]
280#[serde(rename_all = "camelCase", deny_unknown_fields)]
281struct EventQuery {
282    after: Option<u64>,
283    limit: Option<usize>,
284}
285
286async fn get_events(
287    State(state): State<AppState>,
288    headers: HeaderMap,
289    Path(id): Path<String>,
290    Query(query): Query<EventQuery>,
291) -> Response {
292    let authority = match authenticate(&state, &headers).await {
293        Ok(value) => value,
294        Err(response) => return response,
295    };
296    let id = match parse_id(id) {
297        Ok(value) => value,
298        Err(error) => return invalid_execution_id(error),
299    };
300    match state
301        .link
302        .events(&authority, &id, query.after, query.limit.unwrap_or(100))
303        .await
304    {
305        Ok(value) => Json(value).into_response(),
306        Err(error) => link_error(error),
307    }
308}
309
310async fn stream_events(
311    State(state): State<AppState>,
312    headers: HeaderMap,
313    Path(id): Path<String>,
314    Query(query): Query<EventQuery>,
315) -> Response {
316    let authority = match authenticate(&state, &headers).await {
317        Ok(value) => value,
318        Err(response) => return response,
319    };
320    let id = match parse_id(id) {
321        Ok(value) => value,
322        Err(error) => return invalid_execution_id(error),
323    };
324    if let Err(error) = state.link.execution(&authority, &id).await {
325        return link_error(error);
326    }
327    let link = state.link.clone();
328    let stream = event_stream(link, authority, id, query.after);
329    Sse::new(stream)
330        .keep_alive(
331            KeepAlive::new()
332                .interval(Duration::from_secs(15))
333                .text("keepalive"),
334        )
335        .into_response()
336}
337
338fn event_stream(
339    link: Arc<dyn RuntimeLink>,
340    authority: RequestAuthority,
341    id: ExecutionId,
342    initial_after: Option<u64>,
343) -> impl Stream<Item = Result<Event, Infallible>> {
344    async_stream::stream! {
345        let mut after = initial_after;
346        loop {
347            match link.events(&authority, &id, after, 100).await {
348                Ok(page) => {
349                    let page_was_empty = page.items.is_empty();
350                    for item in page.items {
351                        after = Some(item.sequence);
352                        let event = Event::default().id(item.sequence.to_string()).event("runtime").json_data(&item).unwrap_or_else(|_| Event::default().event("error").data("serialization failure"));
353                        yield Ok(event);
354                    }
355                    if page_was_empty {
356                        match link.execution(&authority, &id).await {
357                            Ok(view) if view.state.is_terminal() => break,
358                            Err(_) => break,
359                            _ => {}
360                        }
361                    }
362                }
363                Err(error) => { yield Ok(Event::default().event("error").data(error.to_string())); break; }
364            }
365            tokio::time::sleep(Duration::from_millis(100)).await;
366        }
367    }
368}
369
370async fn submit_input(
371    State(state): State<AppState>,
372    headers: HeaderMap,
373    Path(id): Path<String>,
374    Json(request): Json<SubmitInputRequest>,
375) -> Response {
376    let authority = match authenticate(&state, &headers).await {
377        Ok(value) => value,
378        Err(response) => return response,
379    };
380    let id = match parse_id(id) {
381        Ok(value) => value,
382        Err(error) => return invalid_execution_id(error),
383    };
384    match state.link.submit_input(&authority, &id, request).await {
385        Ok(value) => Json(value).into_response(),
386        Err(error) => link_error(error),
387    }
388}
389
390async fn cancel_execution(
391    State(state): State<AppState>,
392    headers: HeaderMap,
393    Path(id): Path<String>,
394) -> Response {
395    let authority = match authenticate(&state, &headers).await {
396        Ok(value) => value,
397        Err(response) => return response,
398    };
399    let id = match parse_id(id) {
400        Ok(value) => value,
401        Err(error) => return invalid_execution_id(error),
402    };
403    match state.link.cancel(&authority, &id).await {
404        Ok(value) => Json(value).into_response(),
405        Err(error) => link_error(error),
406    }
407}
408
409async fn authenticate(state: &AppState, headers: &HeaderMap) -> Result<RequestAuthority, Response> {
410    state
411        .authenticator
412        .authenticate(headers)
413        .await
414        .map_err(|_| {
415            api_error(
416                StatusCode::UNAUTHORIZED,
417                "UNAUTHENTICATED",
418                "request authentication failed",
419            )
420        })
421}
422
423fn parse_id(value: String) -> Result<ExecutionId, runtime_types::IdentifierError> {
424    ExecutionId::new(value)
425}
426
427fn invalid_execution_id(error: runtime_types::IdentifierError) -> Response {
428    api_error(
429        StatusCode::BAD_REQUEST,
430        "EXECUTION_ID_INVALID",
431        &error.to_string(),
432    )
433}
434
435fn link_error(error: LinkError) -> Response {
436    match error {
437        LinkError::Invalid(message) => {
438            api_error(StatusCode::BAD_REQUEST, "INVALID_REQUEST", &message)
439        }
440        LinkError::NotFound => api_error(
441            StatusCode::NOT_FOUND,
442            "EXECUTION_NOT_FOUND",
443            "execution not found",
444        ),
445        LinkError::Forbidden => api_error(
446            StatusCode::FORBIDDEN,
447            "FORBIDDEN",
448            "caller cannot access this execution",
449        ),
450        LinkError::Conflict(message) => {
451            api_error(StatusCode::CONFLICT, "EXECUTION_CONFLICT", &message)
452        }
453        LinkError::Overloaded => api_error(
454            StatusCode::TOO_MANY_REQUESTS,
455            "RUNTIME_OVERLOADED",
456            "runtime is at capacity",
457        ),
458        LinkError::Unavailable(message) => api_error(
459            StatusCode::SERVICE_UNAVAILABLE,
460            "DEPENDENCY_UNAVAILABLE",
461            &message,
462        ),
463        LinkError::Internal(message) => {
464            tracing::error!(%message, "runtime internal error");
465            api_error(
466                StatusCode::INTERNAL_SERVER_ERROR,
467                "INTERNAL",
468                "runtime internal failure",
469            )
470        }
471    }
472}
473
474fn api_error(status: StatusCode, code: &str, message: &str) -> Response {
475    (
476        status,
477        Json(ApiErrorBody {
478            code: code.into(),
479            message: message.into(),
480            request_id: None,
481            details: None,
482        }),
483    )
484        .into_response()
485}
486
487#[cfg(test)]
488mod tests {
489    use std::sync::Arc;
490
491    use axum::{body::Body, http::Request};
492    use http_body_util::BodyExt;
493    use runtime_api_contract::{CreateExecutionResponse, EventPage, ExecutionState, ExecutionView};
494    use runtime_types::RuntimeInstanceId;
495    use tower::ServiceExt;
496
497    use super::*;
498
499    struct FakeLink;
500
501    fn view() -> ExecutionView {
502        ExecutionView {
503            id: ExecutionId::new("execution-1").unwrap(),
504            runtime_instance_id: RuntimeInstanceId::new("runtime-1").unwrap(),
505            conversation_id: None,
506            state: ExecutionState::Queued,
507            outcome: None,
508            failure: None,
509            created_at_ms: 1,
510            updated_at_ms: 1,
511        }
512    }
513
514    #[async_trait]
515    impl RuntimeLink for FakeLink {
516        async fn create_execution(
517            &self,
518            _: &RequestAuthority,
519            _: &str,
520            _: CreateExecutionRequest,
521        ) -> Result<CreateExecutionResponse, LinkError> {
522            Ok(CreateExecutionResponse {
523                execution: view(),
524                replayed: false,
525            })
526        }
527        async fn execution(
528            &self,
529            _: &RequestAuthority,
530            _: &ExecutionId,
531        ) -> Result<ExecutionView, LinkError> {
532            Ok(view())
533        }
534        async fn events(
535            &self,
536            _: &RequestAuthority,
537            _: &ExecutionId,
538            _: Option<u64>,
539            _: usize,
540        ) -> Result<EventPage, LinkError> {
541            Ok(EventPage {
542                items: Vec::new(),
543                next_after: None,
544            })
545        }
546        async fn submit_input(
547            &self,
548            _: &RequestAuthority,
549            _: &ExecutionId,
550            _: SubmitInputRequest,
551        ) -> Result<ExecutionView, LinkError> {
552            Ok(view())
553        }
554        async fn cancel(
555            &self,
556            _: &RequestAuthority,
557            _: &ExecutionId,
558        ) -> Result<ExecutionView, LinkError> {
559            Ok(view())
560        }
561    }
562
563    fn app() -> Router {
564        let auth = StaticBearerAuthenticator::new(
565            "token",
566            CallerScope {
567                subject: "user".into(),
568                tenant_id: "tenant".into(),
569                project_id: "project".into(),
570                capabilities: Vec::new(),
571            },
572        )
573        .unwrap();
574        router(Arc::new(FakeLink), Arc::new(auth), Arc::new(AlwaysReady))
575    }
576
577    fn create_request(authenticated: bool) -> Request<Body> {
578        let mut request = Request::post("/v1/executions")
579            .header("content-type", "application/json")
580            .header("idempotency-key", "key-1");
581        if authenticated {
582            request = request.header("authorization", "Bearer token");
583        }
584        request.body(Body::from(r#"{"runtimeInstanceId":"runtime-1","input":{"type":"user_message","text":"hello"},"options":{"deadlineSeconds":60,"maxModelTurns":4,"maxToolCalls":8}}"#)).unwrap()
585    }
586
587    #[tokio::test]
588    async fn create_requires_authentication_and_returns_contract_body() {
589        let unauthorized = app().oneshot(create_request(false)).await.unwrap();
590        assert_eq!(unauthorized.status(), StatusCode::UNAUTHORIZED);
591
592        let response = app().oneshot(create_request(true)).await.unwrap();
593        assert_eq!(response.status(), StatusCode::ACCEPTED);
594        let body = response.into_body().collect().await.unwrap().to_bytes();
595        let body: serde_json::Value = serde_json::from_slice(&body).unwrap();
596        assert_eq!(body["execution"]["id"], "execution-1");
597    }
598
599    #[tokio::test]
600    async fn trusted_gateway_scope_requires_the_shared_ingress_secret() {
601        let authenticator =
602            TrustedGatewayAuthenticator::new("a-runtime-gateway-secret-with-at-least-32-bytes")
603                .unwrap();
604        let mut headers = HeaderMap::new();
605        headers.insert("authorization", "Bearer delegated-token".parse().unwrap());
606        headers.insert("x-runtime-subject", "user".parse().unwrap());
607        headers.insert("x-runtime-tenant-id", "tenant".parse().unwrap());
608        headers.insert("x-runtime-project-id", "project".parse().unwrap());
609        assert!(authenticator.authenticate(&headers).await.is_err());
610        headers.insert(
611            "x-runtime-gateway-secret",
612            "a-runtime-gateway-secret-with-at-least-32-bytes"
613                .parse()
614                .unwrap(),
615        );
616        let authority = authenticator.authenticate(&headers).await.unwrap();
617        assert_eq!(authority.caller.subject, "user");
618    }
619}
agent_runtime_http_server/lib.rs

agent_runtime_http_server/
lib.rs
