1use axum::{
4 Router,
5 extract::DefaultBodyLimit,
6 http::{HeaderName, HeaderValue, Method, StatusCode, header},
7 routing::{any, get, post},
8};
9use tower_http::cors::CorsLayer;
10
11use super::assistant::{assistant_descriptor, assistant_document};
12use super::authoring::compile_source;
13use super::awl::{
14 bind_run, check, create_document, deploy_authoring, edit, emit, format, get_document,
15 get_layout, get_revision, get_run_status, list_documents, put_document, put_layout, scaffold,
16 worker_availability,
17};
18use super::awl_deployed::{get_deployed_document, list_deployed};
19use super::build::build_identity;
20use super::cluster_command::cluster_command;
21use super::deploy::{list_versions, route_version, unload_version, upload_package};
22use super::describe_live::describe_live;
23use super::dev_ui::{dev_register_mock, dev_replay_run, dev_trigger_run};
24use super::events::subscribe_events_socket;
25use super::history::{fetch_event, fetch_history};
26use super::intervene::{intervene, list_attempts};
27use super::managed_workers::{
28 list_managed_workers, restart_managed_worker, start_managed_worker, stop_managed_worker,
29};
30use super::outbox::list_dead_letters;
31use super::queues::list_unserved_queues;
32use super::schedules::{
33 create_schedule, delete_schedule, describe_schedule, list_schedules, pause_schedule,
34 resume_schedule, update_schedule,
35};
36use super::transcripts::{fetch_transcript, list_transcript_streams};
37use super::unrecoverable::list_unrecoverable_runs;
38use super::whoami::whoami;
39use super::worker_deployments::{
40 delete_worker_deployment, get_worker_deployment, list_worker_deployments,
41 put_worker_deployment, set_worker_deployment_desired_state,
42};
43use super::workers::{drain_worker, stop_worker};
44use super::workflows::{
45 cancel_workflow, count_workflows, describe_workflow, get_workflows, list_namespace_records,
46 list_namespaces, post_list_workflows, post_namespace, query_workflow, reopen_workflow,
47 set_namespace_placement, signal_workflow, start_workflow,
48};
49use crate::mcp::{McpRuntime, mcp_disabled_router, mcp_router};
50use crate::{ServerError, ServerState, observability, ops_console::assets};
51
52pub fn http_router(state: ServerState) -> Result<Router, ServerError> {
61 let ops_console = assets::ops_console_router(&state.runtime_config().ops_console)?;
62 let cors = cors_layer(&state.runtime_config().cors_allowed_origins)?;
63 let metrics = state.metrics().cloned();
64 let health = state.health().cloned();
65 let mut router = workflow_router(state.clone());
66 router = router.merge(mcp_family(&state)?.with_state(state));
71 if let Some(metrics) = metrics {
72 router = router.merge(Router::new().route(
73 "/metrics",
74 get(observability::metrics::metrics_handler).with_state(metrics),
75 ));
76 }
77 if let Some(health) = health {
78 router = router.merge(
79 Router::new()
80 .route("/health/live", get(observability::health::live))
81 .route(
82 "/health/ready",
83 get(observability::health::ready).with_state(health),
84 ),
85 );
86 }
87 let router = router.merge(ops_console);
88 Ok(match cors {
94 Some(cors) => router.layer(cors),
95 None => router,
96 })
97}
98
99fn cors_layer(allowed_origins: &[String]) -> Result<Option<CorsLayer>, ServerError> {
115 if allowed_origins.is_empty() {
116 return Ok(None);
117 }
118 let mut origins = Vec::with_capacity(allowed_origins.len());
119 for origin in allowed_origins {
120 let value = origin
121 .parse::<HeaderValue>()
122 .map_err(|source| ServerError::Config {
123 message: format!("invalid CORS origin `{origin}`: {source}"),
124 })?;
125 origins.push(value);
126 }
127 let layer = CorsLayer::new()
128 .allow_origin(origins)
129 .allow_methods([
130 Method::GET,
131 Method::POST,
132 Method::PUT,
133 Method::DELETE,
134 Method::OPTIONS,
135 ])
136 .allow_headers([
137 header::CONTENT_TYPE,
138 header::AUTHORIZATION,
139 HeaderName::from_static("x-aion-namespaces"),
140 HeaderName::from_static("x-aion-subject"),
141 ]);
142 Ok(Some(layer))
143}
144
145async fn deploy_disabled() -> StatusCode {
148 StatusCode::NOT_FOUND
149}
150
151async fn authoring_disabled() -> StatusCode {
155 StatusCode::NOT_FOUND
156}
157
158async fn dev_disabled() -> StatusCode {
162 StatusCode::NOT_FOUND
163}
164
165async fn deploy_surface_disabled() -> StatusCode {
170 StatusCode::NOT_FOUND
171}
172
173fn deployed_awl_router(deploy_enabled: bool) -> Router<ServerState> {
181 if deploy_enabled {
182 Router::new()
183 .route("/awl/deployed", get(list_deployed))
184 .route(
185 "/awl/deployed/{workflow_type}/{content_hash}",
186 get(get_deployed_document),
187 )
188 } else {
189 Router::new()
190 .route("/awl/deployed", any(deploy_surface_disabled))
191 .route("/awl/deployed/{*rest}", any(deploy_surface_disabled))
192 }
193}
194
195fn worker_deployment_router(deploy_enabled: bool) -> Router<ServerState> {
196 if deploy_enabled {
197 Router::new()
198 .route("/worker-deployments", get(list_worker_deployments))
199 .route(
200 "/worker-deployments/{name}",
201 get(get_worker_deployment)
202 .put(put_worker_deployment)
203 .delete(delete_worker_deployment),
204 )
205 .route(
206 "/worker-deployments/{name}/desired-state",
207 post(set_worker_deployment_desired_state),
208 )
209 } else {
210 Router::new()
211 .route("/worker-deployments", any(deploy_surface_disabled))
212 .route("/worker-deployments/{*rest}", any(deploy_surface_disabled))
213 }
214}
215
216fn managed_worker_router(deploy_enabled: bool) -> Router<ServerState> {
230 let read = Router::new().route("/workers/managed", get(list_managed_workers));
231 if deploy_enabled {
232 read.route("/workers/managed/{name}/start", post(start_managed_worker))
233 .route("/workers/managed/{name}/stop", post(stop_managed_worker))
234 .route(
235 "/workers/managed/{name}/restart",
236 post(restart_managed_worker),
237 )
238 } else {
239 read.route("/workers/managed/{*rest}", any(deploy_surface_disabled))
240 }
241}
242
243fn mcp_family(state: &ServerState) -> Result<Router<ServerState>, ServerError> {
259 if !state.runtime_config().mcp.enabled {
260 return Ok(mcp_disabled_router());
261 }
262 let runtime = McpRuntime::build(state.clone(), &state.runtime_config().mcp)?;
263 Ok(mcp_router(std::sync::Arc::new(runtime)))
264}
265
266pub fn workflow_router(state: ServerState) -> Router {
268 let deploy = if state.runtime_config().deploy.enabled {
276 Router::new()
277 .route(
278 "/deploy/packages",
279 post(upload_package).layer(DefaultBodyLimit::disable()),
280 )
281 .route("/deploy/versions", get(list_versions))
282 .route("/deploy/route", post(route_version))
283 .route("/deploy/unload", post(unload_version))
284 } else {
285 Router::new().route("/deploy/{*rest}", any(deploy_disabled))
286 };
287 let authoring = if state.runtime_config().authoring.gleam_path.is_some() {
294 Router::new().route("/authoring/compile", post(compile_source))
295 } else {
296 Router::new().route("/authoring/{*rest}", any(authoring_disabled))
297 };
298 let awl_documents = Router::new()
302 .route("/awl/documents", get(list_documents).post(create_document))
303 .route(
304 "/awl/documents/{*path}",
305 get(get_document).put(put_document),
306 )
307 .route("/awl/layout/{*path}", get(get_layout).put(put_layout));
308 let awl_deployed = deployed_awl_router(state.runtime_config().deploy.enabled);
309 let awl = Router::new()
310 .route("/awl/check", post(check))
311 .route("/awl/emit", post(emit))
312 .route("/awl/deploy", post(deploy_authoring))
313 .route("/awl/revisions/{hash}", get(get_revision))
314 .route("/awl/workers/availability", post(worker_availability))
315 .route("/awl/runs/{deployment_id}", get(get_run_status))
316 .route("/awl/runs/{deployment_id}/binding", post(bind_run))
317 .route("/awl/edit", post(edit))
318 .route("/awl/fmt", post(format))
319 .route("/awl/scaffold", post(scaffold))
320 .merge(awl_documents)
321 .merge(awl_deployed);
322 let dev = if state.runtime_config().dev.enabled {
327 Router::new()
328 .route("/dev/runs", post(dev_trigger_run))
329 .route("/dev/mocks", post(dev_register_mock))
330 .route("/dev/replay", post(dev_replay_run))
331 } else {
332 Router::new().route("/dev/{*rest}", any(dev_disabled))
333 };
334 deploy
335 .merge(authoring)
336 .merge(awl)
337 .merge(dev)
338 .merge(worker_deployment_router(
339 state.runtime_config().deploy.enabled,
340 ))
341 .route("/whoami", get(whoami))
342 .route("/build", get(build_identity))
343 .route("/assistant", get(assistant_descriptor))
347 .route("/assistant/document", get(assistant_document))
348 .route("/namespaces", get(list_namespaces).post(post_namespace))
349 .route("/namespaces/records", get(list_namespace_records))
350 .route(
351 "/namespaces/{name}/placement",
352 axum::routing::put(set_namespace_placement),
353 )
354 .route("/workflows", get(get_workflows))
355 .route("/workflows/count", get(count_workflows))
356 .route("/workflows/start", post(start_workflow))
357 .route("/workflows/signal", post(signal_workflow))
358 .route("/workflows/query", post(query_workflow))
359 .route("/workflows/cancel", post(cancel_workflow))
360 .route("/workflows/reopen", post(reopen_workflow))
361 .route("/workflows/list", post(post_list_workflows))
362 .route("/workflows/describe", post(describe_workflow))
363 .route("/workflows/describe-live", post(describe_live))
364 .route("/workflows/history", post(fetch_history))
365 .route("/workflows/event", post(fetch_event))
366 .route("/workflows/intervene", post(intervene))
367 .route("/workflows/attempts", post(list_attempts))
368 .route("/workflows/transcript", post(fetch_transcript))
369 .route("/workflows/transcripts", post(list_transcript_streams))
370 .route("/workflows/unrecoverable", get(list_unrecoverable_runs))
371 .route("/events/stream", get(subscribe_events_socket))
372 .route("/cluster/command", post(cluster_command))
373 .route("/outbox/dead-letters", post(list_dead_letters))
374 .route("/queues/unserved", get(list_unserved_queues))
375 .merge(managed_worker_router(state.runtime_config().deploy.enabled))
376 .route("/workers/{worker_id}/drain", post(drain_worker))
377 .route("/workers/{worker_id}/stop", post(stop_worker))
378 .route("/schedules", post(create_schedule).get(list_schedules))
379 .route(
380 "/schedules/{id}",
381 get(describe_schedule)
382 .put(update_schedule)
383 .delete(delete_schedule),
384 )
385 .route("/schedules/{id}/pause", post(pause_schedule))
386 .route("/schedules/{id}/resume", post(resume_schedule))
387 .with_state(state)
388}
389
390#[cfg(test)]
391mod tests {
392 use std::{fs, sync::Arc};
393
394 use aion::EngineBuilder;
395 use aion_store::{EventStore, InMemoryStore};
396 use axum::{body, http::Request, http::StatusCode};
397 use tower::ServiceExt;
398
399 use super::super::test_support::{
400 NAMESPACE, json_request, read_json, read_text, runtime_config, server_state,
401 };
402 use super::*;
403 use crate::{
404 NamespaceResolver, StaticScheduleNamespaces, StaticWorkflowNamespaces,
405 config::{NamespaceConfig, NamespaceMode, OpsConsoleAssetSource, OpsConsoleConfig},
406 };
407
408 #[tokio::test]
409 async fn ops_console_assets_serve_index_asset_and_do_not_shadow_public_api()
410 -> Result<(), Box<dyn std::error::Error>> {
411 let bundle = crate::test_support::private_tempdir()?;
412 fs::write(
413 bundle.path().join("index.html"),
414 "<!doctype html><title>Aion</title><script src=\"/app.js\"></script>",
415 )?;
416 fs::write(bundle.path().join("app.js"), "window.AION = true;")?;
417
418 let store: Arc<dyn EventStore> = Arc::new(InMemoryStore::default());
419 let engine = Arc::new(
420 EngineBuilder::new()
421 .store_arc(Arc::clone(&store))
422 .in_memory_visibility()
423 .scheduler_threads(1)
424 .build()
425 .await?,
426 );
427 let resolver = NamespaceResolver::from_parts(
428 NamespaceMode::SharedEngine,
429 Some(engine),
430 Arc::new(StaticWorkflowNamespaces::default()),
431 Arc::new(StaticScheduleNamespaces::default()),
432 );
433 let mut config = runtime_config();
434 config.ops_console = OpsConsoleConfig {
435 source: OpsConsoleAssetSource::FileSystem {
436 asset_path: bundle.path().to_path_buf(),
437 },
438 };
439 let router = http_router(server_state(resolver, config).await?)?;
440
441 let root = router
442 .clone()
443 .oneshot(Request::builder().uri("/").body(body::Body::empty())?)
444 .await?;
445 assert_eq!(root.status(), StatusCode::OK);
446 assert!(read_text(root).await?.contains("<title>Aion</title>"));
447
448 let asset = router
449 .clone()
450 .oneshot(
451 Request::builder()
452 .uri("/app.js")
453 .body(body::Body::empty())?,
454 )
455 .await?;
456 assert_eq!(asset.status(), StatusCode::OK);
457 assert_eq!(read_text(asset).await?, "window.AION = true;");
458
459 let spa = router
460 .clone()
461 .oneshot(
462 Request::builder()
463 .uri("/ops-console/workflows/demo")
464 .body(body::Body::empty())?,
465 )
466 .await?;
467 assert_eq!(spa.status(), StatusCode::OK);
468 assert!(read_text(spa).await?.contains("<title>Aion</title>"));
469
470 let list = serde_json::json!({
471 "namespace": NAMESPACE,
472 "filter": { "workflow_type": "nonexistent" },
473 });
474 let list_response = router
475 .oneshot(json_request("/workflows/list", &list)?)
476 .await?;
477 assert_eq!(list_response.status(), StatusCode::OK);
478 let list_body: serde_json::Value = read_json(list_response).await?;
479 assert!(
480 list_body["summaries"]
481 .as_array()
482 .ok_or("summaries missing")?
483 .is_empty()
484 );
485 Ok(())
486 }
487
488 #[tokio::test]
489 async fn cors_preflight_and_actual_request_carry_allow_headers_for_configured_origin()
490 -> Result<(), Box<dyn std::error::Error>> {
491 let store: Arc<dyn EventStore> = Arc::new(InMemoryStore::default());
492 let engine = Arc::new(
493 EngineBuilder::new()
494 .store_arc(Arc::clone(&store))
495 .in_memory_visibility()
496 .scheduler_threads(1)
497 .build()
498 .await?,
499 );
500 let resolver = NamespaceResolver::from_parts(
501 NamespaceMode::SharedEngine,
502 Some(engine),
503 Arc::new(StaticWorkflowNamespaces::default()),
504 Arc::new(StaticScheduleNamespaces::default()),
505 );
506 let mut config = runtime_config();
507 config.cors_allowed_origins = vec!["http://localhost:5173".to_owned()];
508 let router = http_router(server_state(resolver, config).await?)?;
509
510 let preflight = router
513 .clone()
514 .oneshot(
515 Request::builder()
516 .method("OPTIONS")
517 .uri("/workflows/list")
518 .header("origin", "http://localhost:5173")
519 .header("access-control-request-method", "POST")
520 .header("access-control-request-headers", "x-aion-namespaces")
521 .body(body::Body::empty())?,
522 )
523 .await?;
524 assert_eq!(
525 preflight
526 .headers()
527 .get("access-control-allow-origin")
528 .and_then(|value| value.to_str().ok()),
529 Some("http://localhost:5173")
530 );
531
532 let actual = router
534 .oneshot(
535 Request::builder()
536 .uri("/health/live")
537 .header("origin", "http://localhost:5173")
538 .body(body::Body::empty())?,
539 )
540 .await?;
541 assert_eq!(actual.status(), StatusCode::OK);
542 assert_eq!(
543 actual
544 .headers()
545 .get("access-control-allow-origin")
546 .and_then(|value| value.to_str().ok()),
547 Some("http://localhost:5173")
548 );
549 Ok(())
550 }
551
552 #[tokio::test]
553 async fn cors_absent_origins_install_no_layer() -> Result<(), Box<dyn std::error::Error>> {
554 let store: Arc<dyn EventStore> = Arc::new(InMemoryStore::default());
555 let engine = Arc::new(
556 EngineBuilder::new()
557 .store_arc(Arc::clone(&store))
558 .in_memory_visibility()
559 .scheduler_threads(1)
560 .build()
561 .await?,
562 );
563 let resolver = NamespaceResolver::from_parts(
564 NamespaceMode::SharedEngine,
565 Some(engine),
566 Arc::new(StaticWorkflowNamespaces::default()),
567 Arc::new(StaticScheduleNamespaces::default()),
568 );
569 let router = http_router(server_state(resolver, runtime_config()).await?)?;
571
572 let response = router
573 .oneshot(
574 Request::builder()
575 .uri("/health/live")
576 .header("origin", "http://localhost:5173")
577 .body(body::Body::empty())?,
578 )
579 .await?;
580 assert_eq!(response.status(), StatusCode::OK);
581 assert!(
582 response
583 .headers()
584 .get("access-control-allow-origin")
585 .is_none(),
586 "no CorsLayer must be installed when no origins are configured"
587 );
588 Ok(())
589 }
590
591 #[cfg(feature = "auth")]
592 #[tokio::test]
593 async fn worker_availability_allows_granted_jwt_namespace_and_denies_foreign_namespace()
594 -> Result<(), Box<dyn std::error::Error>> {
595 let (engine, _, _) = super::super::test_support::shared_engine().await?;
596 let resolver = NamespaceResolver::from_config(
597 NamespaceConfig {
598 mode: NamespaceMode::SharedEngine,
599 },
600 engine,
601 );
602 let router = http_router(server_state(resolver, runtime_config()).await?)?;
603
604 let allowed = router
605 .clone()
606 .oneshot(json_request(
607 "/awl/workers/availability",
608 &serde_json::json!({
609 "namespace": NAMESPACE,
610 "task_queue": "orders",
611 }),
612 )?)
613 .await?;
614 assert_eq!(allowed.status(), StatusCode::OK);
615
616 let foreign = router
617 .oneshot(json_request(
618 "/awl/workers/availability",
619 &serde_json::json!({
620 "namespace": "tenant-b",
621 "task_queue": "orders",
622 }),
623 )?)
624 .await?;
625 assert_eq!(foreign.status(), StatusCode::FORBIDDEN);
626 let body: serde_json::Value = read_json(foreign).await?;
627 assert_eq!(body["code"], "namespace_denied");
628 Ok(())
629 }
630
631 #[tokio::test]
632 async fn worker_availability_keeps_auth_off_operator_access()
633 -> Result<(), Box<dyn std::error::Error>> {
634 let (engine, _, _) = super::super::test_support::shared_engine().await?;
635 let resolver = NamespaceResolver::from_config(
636 NamespaceConfig {
637 mode: NamespaceMode::SharedEngine,
638 },
639 engine,
640 );
641 let mut config = runtime_config();
642 config.auth.enabled = false;
643 config.auth.jwks_url = None;
644 let router = http_router(crate::ServerState::from_parts(resolver, config))?;
645 let request = Request::builder()
646 .method("POST")
647 .uri("/awl/workers/availability")
648 .header("content-type", "application/json")
649 .body(body::Body::from(serde_json::to_vec(&serde_json::json!({
650 "namespace": "operator-selected-namespace",
651 "task_queue": "orders",
652 }))?))?;
653
654 let response = router.oneshot(request).await?;
655 assert_eq!(response.status(), StatusCode::OK);
656 let body: serde_json::Value = read_json(response).await?;
657 assert_eq!(body["connected_workers"], 0);
658 Ok(())
659 }
660
661 #[tokio::test]
662 async fn observability_routes_are_public_and_expose_expected_payloads()
663 -> Result<(), Box<dyn std::error::Error>> {
664 #[cfg(feature = "auth")]
668 let config = {
669 let mut config = runtime_config();
670 config.auth.jwks_url = Some(crate::auth::test_support::serve_jwks()?);
671 config
672 };
673 #[cfg(not(feature = "auth"))]
674 let config = runtime_config();
675 let router = http_router(
676 crate::ServerState::build_with_store(InMemoryStore::default(), config).await?,
677 )?;
678
679 let metrics_response = router
680 .clone()
681 .oneshot(
682 Request::builder()
683 .uri("/metrics")
684 .body(body::Body::empty())?,
685 )
686 .await?;
687 assert_eq!(metrics_response.status(), StatusCode::OK);
688 assert_eq!(
689 metrics_response
690 .headers()
691 .get(axum::http::header::CONTENT_TYPE)
692 .and_then(|value| value.to_str().ok()),
693 Some("text/plain; version=0.0.4; charset=utf-8")
694 );
695 let metrics_body = read_text(metrics_response).await?;
696 assert!(metrics_body.contains("# HELP aion_workflows_started_total"));
697 assert!(metrics_body.contains("# TYPE aion_workflows_started_total counter"));
698 assert!(metrics_body.contains("# HELP aion_activity_duration_seconds"));
699 assert!(metrics_body.contains("# TYPE aion_activity_duration_seconds histogram"));
700 assert!(metrics_body.contains("aion_activity_duration_seconds_bucket"));
701 assert!(metrics_body.contains("aion_store_operation_duration_seconds_bucket"));
702
703 let live_response = router
704 .clone()
705 .oneshot(
706 Request::builder()
707 .uri("/health/live")
708 .body(body::Body::empty())?,
709 )
710 .await?;
711 assert_eq!(live_response.status(), StatusCode::OK);
712
713 let ready_response = router
714 .oneshot(
715 Request::builder()
716 .uri("/health/ready")
717 .body(body::Body::empty())?,
718 )
719 .await?;
720 assert_eq!(ready_response.status(), StatusCode::OK);
721 Ok(())
722 }
723}