Skip to main content

aion_server/api/http/
router.rs

1//! Public HTTP router construction.
2
3use 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
52/// Build the public HTTP application: workflow-management routes first, then
53/// the ops-console static asset fallback. The ops console adds no data API.
54///
55/// # Errors
56///
57/// Returns [`ServerError::Config`] when ops-console assets are misconfigured, or
58/// when the MCP surface is enabled and its published tool catalog cannot be
59/// constructed.
60pub 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    // The MCP endpoint is another route on THIS listener: one port, one
67    // process. It is merged here rather than inside `workflow_router` because
68    // it is a top-level protocol surface alongside `/metrics` and `/health/*`,
69    // not a workflow-management route.
70    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    // CORS is applied last so it wraps every public route the browser ops console
89    // calls (the workflow API, /metrics, /health/*, the ops-console fallback).
90    // With no configured origins `cors_layer` returns None and the router is
91    // byte-identical to before — no cross-origin request is allowed (the secure
92    // default). With origins set the layer also answers OPTIONS preflight.
93    Ok(match cors {
94        Some(cors) => router.layer(cors),
95        None => router,
96    })
97}
98
99/// Build the CORS layer for the public HTTP router from the operator-configured
100/// allowed origins.
101///
102/// Returns `Ok(None)` when no origins are configured — the secure default:
103/// the layer is not installed and no cross-origin request is permitted. When
104/// origins are configured the layer is scoped to exactly those origins (never
105/// `Any`, so it is safe to pair with credentialed requests), permits the
106/// methods the ops console uses (GET, POST, PUT, DELETE, and OPTIONS preflight), and allows
107/// exactly the request headers the API consumes.
108///
109/// # Errors
110///
111/// Returns [`ServerError::Config`] when a configured origin is not a valid HTTP
112/// header value. Startup validation already rejects malformed origins, so this
113/// is defense in depth.
114fn 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
145/// Disabled deploy surface: a plain 404 with no body, indistinguishable
146/// from an unmounted route family.
147async fn deploy_disabled() -> StatusCode {
148    StatusCode::NOT_FOUND
149}
150
151/// Disabled authoring surface: a plain 404 with no body, indistinguishable
152/// from an unmounted route family. When `[authoring].gleam_path` is absent the
153/// server compiles no Gleam and deploys pre-built `.aion` files only (CN7).
154async fn authoring_disabled() -> StatusCode {
155    StatusCode::NOT_FOUND
156}
157
158/// Disabled dev surface: a plain 404 with no body, indistinguishable from an
159/// unmounted route family. When `[dev].enabled` is false the server mounts no
160/// dev endpoints and installs no activity-mock decorator (CN4).
161async fn dev_disabled() -> StatusCode {
162    StatusCode::NOT_FOUND
163}
164
165/// Disabled deployment surfaces: a plain 404 with no body for the deployed-AWL
166/// reader, worker-deployment management, and managed-worker lifecycle
167/// families. All follow the deploy surface's switch and stay dark when the
168/// operator closes it.
169async fn deploy_surface_disabled() -> StatusCode {
170    StatusCode::NOT_FOUND
171}
172
173/// The read-only deployed-AWL route family.
174///
175/// GET only: both routes carry a single `get`, so every other method is a 405
176/// the router answers — there is no mutation handler on this family to reach,
177/// guarded or otherwise. With the deploy surface off the family is a plain 404,
178/// covering the bare path and everything under it (a `{*rest}` wildcard does
179/// not match the bare path).
180fn 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
216/// The managed-worker route family: the always-mounted status join, and the
217/// three lifecycle commands mounted only when `[deploy].enabled` is set.
218///
219/// The mount condition matches the surfaces these commands belong to: their
220/// gRPC counterparts live on `DeployService`, which joins the listener only
221/// under `[deploy].enabled` (`crate::run`), and the sibling
222/// `/worker-deployments/*` family follows the same switch. With deploy off,
223/// every path under `/workers/managed/` is a plain 404 — a server that is not
224/// a deploy target exposes no lifecycle mutation on any transport. Within a
225/// mounted route, authorization is the handlers' deploy-grant check (decided
226/// per caller; with auth disabled the single-tenant operator holds the grant
227/// server-side). The read stays mounted regardless, exactly as it always has
228/// been: reporting the fleet is not a deploy-surface mutation.
229fn 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
243/// The MCP route family, mounted on the SERVER'S OWN listener.
244///
245/// One port, one process: the MCP endpoint is another route on the same HTTP
246/// surface the console and the workflow API are served from, not a second
247/// listener. It is dark unless `[mcp].enabled` is set, and a dark surface is a
248/// plain 404 — indistinguishable from a build that never had the route.
249///
250/// A construction failure (a published tool schema that will not compile) is
251/// NOT downgraded to a dark surface: it is a server defect, and an operator who
252/// asked for the endpoint would otherwise get a 404 and no idea why.
253///
254/// Each call builds one runtime, and a runtime owns one task store. That is
255/// correct because a task is reachable only through the identifier its creator
256/// was handed: two routers are two disjoint sets of handles, never a split view
257/// of one set. The serving path builds the router exactly once.
258fn 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
266/// Build the public workflow-management HTTP router.
267pub fn workflow_router(state: ServerState) -> Router {
268    // The deploy surface is dark by default: when `[deploy].enabled` is
269    // false the routes are not mounted and every `/deploy/*` path is a
270    // plain 404 (the explicit catch-all keeps the ops-console SPA fallback
271    // from answering for the deploy namespace). The archive upload route
272    // disables the default body limit because `read_archive_body` enforces
273    // the operator-configured `deploy.max_archive_bytes` ceiling while
274    // streaming.
275    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    // The authoring surface is dark by default, gated on
288    // `[authoring].gleam_path`: when it is unset the routes are not mounted and
289    // every `/authoring/*` path is a plain 404 (the explicit catch-all keeps
290    // the ops-console SPA fallback from answering for the authoring namespace).
291    // With it absent the server compiles no Gleam and deploys pre-built `.aion`
292    // files only (CN7).
293    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    // The full AWL studio is always mounted. Stock config supplies the
299    // `aion-authoring` workspace; the typed unconfigured refusal remains for
300    // manually constructed runtime configs that explicitly omit it.
301    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    // The dev surface is dark by default, gated on `[dev].enabled`: when off the
323    // routes are not mounted and every `/dev/*` path is a plain 404 (the
324    // explicit catch-all keeps the ops-console SPA fallback from answering for
325    // the dev namespace), and the engine runs the bare production dispatcher.
326    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        // The built-in assistant, always mounted: it ships in the binary like
344        // the ops console, so there is no configuration under which this server
345        // has one and does not say so.
346        .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        // Preflight: the browser sends OPTIONS with the requested method/header;
511        // the layer must answer with the matching allow-origin and allow-methods.
512        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        // Actual request from the allowed origin echoes the allow-origin header.
533        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        // runtime_config() leaves cors_allowed_origins empty (the secure default).
570        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        // This test rides the production `build_with_store` startup path, so
665        // under `feature = "auth"` the configured jwks_url must be a live
666        // endpoint for the initial JWKS fetch.
667        #[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}