cellos-server 0.5.0

//! `/v1/formations` — CRUD handlers.
//!
//! POST validates the submitted `FormationDocument` per ADR-0010
//! (single coordinator + every non-coordinator member carries
//! `authorizedBy`). The full DAG/cycle/scope-narrowing admission gate
//! lives in `cellos-supervisor`; we surface the same RFC 9457
//! discriminants here so cellctl can render either source uniformly.

use axum::extract::{Path, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::IntoResponse;
use axum::Json;
use cellos_core::events::{
    cloud_event_v1_formation_completed, cloud_event_v1_formation_created,
    cloud_event_v1_formation_degraded, cloud_event_v1_formation_failed,
    cloud_event_v1_formation_launching, cloud_event_v1_formation_running,
};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::auth::require_bearer;
use crate::error::{AppError, AppErrorKind};
use crate::state::{AppState, FormationRecord, FormationStatus};

/// Subset of the `formation-v1` document the admission gate cares about.
/// Additional fields are tolerated and preserved verbatim in
/// `FormationRecord::document` (via the captured `serde_json::Value`).
#[derive(Debug, Deserialize)]
pub struct FormationDocument {
    pub name: String,
    pub coordinator: String,
    pub members: Vec<FormationMember>,
}

#[derive(Debug, Deserialize)]
pub struct FormationMember {
    pub id: String,
    /// Required on every non-coordinator member; forbidden on the
    /// coordinator (ADR-0010 §Enforcement).
    #[serde(rename = "authorizedBy")]
    pub authorized_by: Option<String>,
}

#[derive(Debug, Serialize)]
pub struct FormationCreated {
    pub id: Uuid,
    pub name: String,
    pub status: FormationStatus,
}

/// POST /v1/formations — admit a new formation. Returns 201 with the
/// generated id on success; RFC 9457 problem+json on validation failure.
pub async fn create_formation(
    State(state): State<AppState>,
    headers: HeaderMap,
    body: axum::body::Bytes,
) -> Result<impl IntoResponse, AppError> {
    require_bearer(&headers, &state.api_token)?;

    // Parse twice: once into our validated struct, once into a generic
    // Value so we can echo the original document back on GET. This keeps
    // the public schema forward-compatible with fields we don't yet
    // recognise.
    let raw: serde_json::Value = serde_json::from_slice(&body)?;
    let doc: FormationDocument = serde_json::from_value(raw.clone())?;

    validate_formation(&doc)?;

    let id = Uuid::new_v4();
    let record = FormationRecord {
        id,
        name: doc.name.clone(),
        status: FormationStatus::Pending,
        document: raw,
    };

    state.formations.write().await.insert(id, record);

    // Emit formation.v1.created so the WebSocket stream and audit log see it.
    let cell_count = doc.members.len() as u32;
    let no_failed: &[String] = &[];
    let event = cloud_event_v1_formation_created(
        &id.to_string(),
        id.to_string().as_str(),
        &id.to_string(),
        &doc.name,
        cell_count,
        no_failed,
        None,
    );
    let subject = format!("cellos.events.formations.{id}.created");
    publish_event(&state, &subject, event).await;

    let body = FormationCreated {
        id,
        name: doc.name,
        status: FormationStatus::Pending,
    };
    Ok((StatusCode::CREATED, Json(body)))
}

/// Response shape for `GET /v1/formations` per ADR-0015 §D2. The
/// `cursor` is the highest JetStream stream-sequence the server's
/// projection has applied; clients hand it back as
/// `/ws/events?since=<cursor>` so they can resume the live stream
/// without missing any event between the snapshot and the WS open.
#[derive(Debug, Serialize)]
pub struct FormationsSnapshot {
    pub formations: Vec<FormationRecord>,
    pub cursor: u64,
}

/// GET /v1/formations — list all known formations plus the current
/// projection cursor (ADR-0015).
pub async fn list_formations(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Json<FormationsSnapshot>, AppError> {
    require_bearer(&headers, &state.api_token)?;
    let map = state.formations.read().await;
    Ok(Json(FormationsSnapshot {
        formations: map.values().cloned().collect(),
        cursor: state.cursor(),
    }))
}

/// GET /v1/formations/{id} — fetch one formation by uuid.
pub async fn get_formation(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(id): Path<Uuid>,
) -> Result<Json<FormationRecord>, AppError> {
    require_bearer(&headers, &state.api_token)?;
    let map = state.formations.read().await;
    map.get(&id)
        .cloned()
        .map(Json)
        .ok_or_else(|| AppError::not_found(format!("formation {id} not found")))
}

/// DELETE /v1/formations/{id} — best-effort cancellation. The actual
/// teardown is performed asynchronously by the supervisor once the
/// `formation.cancelled` event lands on JetStream; we only mark the
/// in-memory projection.
pub async fn delete_formation(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(id): Path<Uuid>,
) -> Result<StatusCode, AppError> {
    require_bearer(&headers, &state.api_token)?;
    let mut map = state.formations.write().await;
    let (name, cell_count) = {
        let entry = map
            .get_mut(&id)
            .ok_or_else(|| AppError::not_found(format!("formation {id} not found")))?;
        entry.status = FormationStatus::Cancelled;
        let members = entry
            .document
            .get("members")
            .and_then(|m| m.as_array())
            .map(|a| a.len() as u32)
            .unwrap_or(0);
        (entry.name.clone(), members)
    };
    drop(map);

    let no_failed: &[String] = &[];
    let event = cloud_event_v1_formation_failed(
        &id.to_string(),
        id.to_string().as_str(),
        &id.to_string(),
        &name,
        cell_count,
        no_failed,
        Some("deleted by operator"),
    );
    let subject = format!("cellos.events.formations.{id}.failed");
    publish_event(&state, &subject, event).await;

    Ok(StatusCode::NO_CONTENT)
}

/// POST /v1/formations/{id}/status — receive a state-transition notification
/// from the supervisor or an operator tool. Updates the in-memory projection
/// and emits the matching `formation.v1.*` CloudEvent to NATS so the
/// WebSocket stream carries it to connected web-view clients.
#[derive(Debug, Deserialize)]
pub struct StatusTransition {
    pub state: String,
    pub reason: Option<String>,
    pub failed_cells: Option<Vec<String>>,
}

pub async fn update_formation_status(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(id): Path<Uuid>,
    Json(body): Json<StatusTransition>,
) -> Result<StatusCode, AppError> {
    require_bearer(&headers, &state.api_token)?;

    let (new_status, name, cell_count, failed) = {
        let mut map = state.formations.write().await;
        let entry = map
            .get_mut(&id)
            .ok_or_else(|| AppError::not_found(format!("formation {id} not found")))?;

        let new_status = match body.state.to_uppercase().as_str() {
            "RUNNING" | "LAUNCHING" => FormationStatus::Running,
            "DEGRADED" => FormationStatus::Running, // DEGRADED keeps running
            "COMPLETED" => FormationStatus::Succeeded,
            "FAILED" => FormationStatus::Failed,
            other => {
                // RFC-9457 §3.1: this is a generic bad-request, not an
                // ADR-0010 admission-gate rejection. Returning the
                // `FormationNoCoordinator` discriminant here would
                // hijack a load-bearing identifier that clients switch
                // on per ADR-0010 §Enforcement.
                return Err(AppError::new(
                    AppErrorKind::BadRequest,
                    format!("unknown state: {other}"),
                ));
            }
        };
        entry.status = new_status;

        let members = entry
            .document
            .get("members")
            .and_then(|m| m.as_array())
            .map(|a| a.len() as u32)
            .unwrap_or(0);
        let failed = body.failed_cells.unwrap_or_default();
        (new_status, entry.name.clone(), members, failed)
    };

    let sid = id.to_string();
    let reason = body.reason.as_deref();
    let empty: &[String] = &[];

    let (event, phase) = match body.state.to_uppercase().as_str() {
        "LAUNCHING" => (
            cloud_event_v1_formation_launching(&sid, &sid, &sid, &name, cell_count, empty, reason),
            "launching",
        ),
        "RUNNING" => (
            cloud_event_v1_formation_running(&sid, &sid, &sid, &name, cell_count, empty, reason),
            "running",
        ),
        "DEGRADED" => (
            cloud_event_v1_formation_degraded(&sid, &sid, &sid, &name, cell_count, &failed, reason),
            "degraded",
        ),
        "COMPLETED" => (
            cloud_event_v1_formation_completed(&sid, &sid, &sid, &name, cell_count, empty, reason),
            "completed",
        ),
        _ => (
            cloud_event_v1_formation_failed(&sid, &sid, &sid, &name, cell_count, &failed, reason),
            "failed",
        ),
    };

    let subject = format!("cellos.events.formations.{id}.{phase}");
    publish_event(&state, &subject, event).await;

    let _ = new_status; // used above
    Ok(StatusCode::NO_CONTENT)
}

/// Publish a CloudEvent JSON payload to NATS if a client is connected.
/// Failures are logged and swallowed — event loss is surfaced via the DLQ
/// (P3-03) once that crate lands; the HTTP response is never blocked by NATS.
async fn publish_event(state: &AppState, subject: &str, event: impl serde::Serialize) {
    let Some(nats) = &state.nats else { return };
    let payload = match serde_json::to_vec(&event) {
        Ok(b) => b,
        Err(e) => {
            tracing::warn!(subject, error = %e, "failed to serialise formation CloudEvent");
            return;
        }
    };
    if let Err(e) = nats.publish(subject.to_owned(), payload.into()).await {
        tracing::warn!(subject, error = %e, "failed to publish formation CloudEvent to NATS");
    }
}

/// Apply the structural admission-gate checks ADR-0010 §Enforcement
/// requires the server to re-run regardless of client behaviour:
///
/// 1. **noCoordinator** — the coordinator named in `coordinator` MUST
///    appear in `members`.
/// 2. **multipleCoordinators** — every `members[*].id` MUST be unique.
///    The JSON schema declares `uniqueItems`; we re-enforce because
///    the server cannot assume schema validation ran on the client.
/// 3. **authorityNotNarrowing** — the coordinator MUST NOT carry
///    `authorizedBy`; every non-coordinator MUST carry it AND the
///    referenced parent MUST exist in `members` (an orphan parent is
///    an unbounded A₀ — exactly the failure mode ADR-0010 §Proof
///    forbids).
/// 4. **cycle** — the `authorizedBy` edges MUST form a DAG. A cycle
///    (including the self-edge `authorizedBy: self`) breaks the
///    induction that proves every member's authority chains back to
///    the coordinator.
///
/// The per-edge authority-subset check (`A_c ⊆ A_p`) lives in the
/// supervisor today because the `formation-v1` document parsed here
/// does not yet carry per-member declared authority sets; that is the
/// only ADR-0010 check the server still defers.
fn validate_formation(doc: &FormationDocument) -> Result<(), AppError> {
    use std::collections::{HashMap, HashSet};

    // 1. noCoordinator.
    let coord_present = doc.members.iter().any(|m| m.id == doc.coordinator);
    if !coord_present {
        return Err(AppError::new(
            AppErrorKind::FormationNoCoordinator,
            format!(
                "coordinator '{}' must appear in members list",
                doc.coordinator
            ),
        ));
    }

    // 2. multipleCoordinators — duplicate member ids. We treat the
    // duplicate-id failure under this discriminant because the ADR
    // §Consequences canonical case is "two members both named
    // `coord`": admission cannot pick which one is the coordinator.
    let mut seen: HashSet<&str> = HashSet::new();
    for m in &doc.members {
        if !seen.insert(m.id.as_str()) {
            return Err(AppError::new(
                AppErrorKind::FormationMultipleCoordinators,
                format!("duplicate member id '{}'", m.id),
            ));
        }
    }

    // 3. authorityNotNarrowing — coord-forbid, non-coord require, plus
    //    orphan-parent rejection. An `authorizedBy` reference that has
    //    no member entry has no parent edge → no narrowing → admission
    //    fails.
    for m in &doc.members {
        let is_coord = m.id == doc.coordinator;
        match (is_coord, &m.authorized_by) {
            (true, Some(_)) => {
                return Err(AppError::new(
                    AppErrorKind::FormationAuthorityNotNarrowing,
                    format!("coordinator '{}' must not declare authorizedBy", m.id),
                ));
            }
            (false, None) => {
                return Err(AppError::new(
                    AppErrorKind::FormationAuthorityNotNarrowing,
                    format!("non-coordinator member '{}' missing authorizedBy", m.id),
                ));
            }
            (false, Some(parent)) => {
                if !seen.contains(parent.as_str()) {
                    return Err(AppError::new(
                        AppErrorKind::FormationAuthorityNotNarrowing,
                        format!("member '{}' references unknown parent '{}'", m.id, parent),
                    ));
                }
            }
            _ => {}
        }
    }

    // 4. cycle — walk each non-coordinator's authorizedBy chain. In a
    //    valid DAG with exactly one out-edge per non-root, the walk
    //    terminates at the coordinator within strictly fewer hops than
    //    members.len(). Self-loops are caught on the first hop.
    let parent: HashMap<&str, &str> = doc
        .members
        .iter()
        .filter_map(|m| m.authorized_by.as_deref().map(|p| (m.id.as_str(), p)))
        .collect();

    for m in &doc.members {
        if m.id == doc.coordinator {
            continue;
        }
        let mut cursor = m.id.as_str();
        for _ in 0..doc.members.len() {
            let Some(&p) = parent.get(cursor) else {
                // No outgoing edge from cursor → cursor is the
                // coordinator (proven in check 1 to be present). Done.
                break;
            };
            if p == m.id {
                return Err(AppError::new(
                    AppErrorKind::FormationCycle,
                    format!("authorizedBy cycle detected involving member '{}'", m.id),
                ));
            }
            cursor = p;
        }
        if parent.contains_key(cursor) {
            // Exhausted hop budget without reaching the coordinator —
            // a cycle exists on the chain (not necessarily through
            // `m.id` itself).
            return Err(AppError::new(
                AppErrorKind::FormationCycle,
                format!(
                    "authorizedBy cycle detected on chain starting at '{}'",
                    m.id
                ),
            ));
        }
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::router;
    use axum::body::Body;
    use axum::http::{header, Request};
    use http_body_util::BodyExt;
    use tower::ServiceExt;

    const TOKEN: &str = "test-token";

    fn test_state() -> AppState {
        AppState::new(None, TOKEN)
    }

    fn auth_req(method: &str, uri: &str, body: Option<&str>) -> Request<Body> {
        let mut b = Request::builder()
            .method(method)
            .uri(uri)
            .header(header::AUTHORIZATION, format!("Bearer {TOKEN}"));
        if body.is_some() {
            b = b.header(header::CONTENT_TYPE, "application/json");
        }
        b.body(
            body.map(|s| Body::from(s.to_owned()))
                .unwrap_or_else(Body::empty),
        )
        .expect("build request")
    }

    #[tokio::test]
    async fn post_valid_formation_returns_201() {
        let app = router(test_state());
        let body = serde_json::json!({
            "name": "demo",
            "coordinator": "coord",
            "members": [
                { "id": "coord" },
                { "id": "worker-a", "authorizedBy": "coord" },
                { "id": "worker-b", "authorizedBy": "coord" }
            ]
        })
        .to_string();

        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::CREATED);

        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(parsed["status"], "PENDING");
        assert_eq!(parsed["name"], "demo");
        assert!(parsed["id"].as_str().is_some());
    }

    #[tokio::test]
    async fn post_formation_missing_coordinator_returns_400() {
        let app = router(test_state());
        // coordinator names "coord" but no such member exists.
        let body = serde_json::json!({
            "name": "demo",
            "coordinator": "coord",
            "members": [
                { "id": "worker-a", "authorizedBy": "coord" }
            ]
        })
        .to_string();

        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let ct = resp
            .headers()
            .get(header::CONTENT_TYPE)
            .and_then(|v| v.to_str().ok())
            .unwrap_or_default()
            .to_owned();
        assert!(
            ct.starts_with("application/problem+json"),
            "expected RFC 9457 media type, got {ct:?}"
        );
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(parsed["type"], "/problems/formation/no-coordinator");
    }

    #[tokio::test]
    async fn post_formation_member_missing_authorized_by_returns_400() {
        let app = router(test_state());
        let body = serde_json::json!({
            "name": "demo",
            "coordinator": "coord",
            "members": [
                { "id": "coord" },
                { "id": "worker-a" } // missing authorizedBy
            ]
        })
        .to_string();

        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(
            parsed["type"], "/problems/formation/authority-not-narrowing",
            "expected authority-not-narrowing discriminant, got {parsed}"
        );
    }

    #[tokio::test]
    async fn get_formations_returns_snapshot_with_cursor() {
        // ADR-0015 §D2: GET /v1/formations is `{ formations: [...], cursor: u64 }`.
        let app = router(test_state());
        let resp = app
            .oneshot(auth_req("GET", "/v1/formations", None))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert!(parsed.is_object(), "expected snapshot object, got {parsed}");
        let arr = parsed["formations"].as_array().expect("formations array");
        assert_eq!(arr.len(), 0);
        assert!(
            parsed["cursor"].is_u64(),
            "cursor field must be an unsigned integer, got {}",
            parsed["cursor"]
        );
        assert_eq!(parsed["cursor"].as_u64(), Some(0));
    }

    #[tokio::test]
    async fn snapshot_returns_cursor() {
        // ADR-0015 §D2 + §E: after POSTing a formation, the snapshot
        // response MUST carry a `cursor` field of integer type so the
        // client can hand it to `/ws/events?since=<cursor>`.
        let app = router(test_state());
        let body = serde_json::json!({
            "name": "with-cursor",
            "coordinator": "coord",
            "members": [
                { "id": "coord" },
                { "id": "worker-a", "authorizedBy": "coord" }
            ]
        })
        .to_string();

        let resp = app
            .clone()
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::CREATED);

        let resp = app
            .oneshot(auth_req("GET", "/v1/formations", None))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert!(
            parsed["cursor"].is_u64(),
            "cursor must be unsigned integer; got {}",
            parsed["cursor"]
        );
        let formations = parsed["formations"].as_array().expect("formations array");
        assert_eq!(formations.len(), 1, "expected 1 formation after POST");
        assert_eq!(formations[0]["name"], "with-cursor");
    }

    #[tokio::test]
    async fn missing_bearer_returns_401() {
        let app = router(test_state());
        let resp = app
            .oneshot(
                Request::builder()
                    .method("GET")
                    .uri("/v1/formations")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    /// ADR-0010 §Enforcement `multipleCoordinators` discriminant.
    /// Duplicate `members[*].id` MUST be rejected with this type even
    /// when the duplicates carry valid `authorizedBy`. The JSON schema
    /// has `uniqueItems` but the server cannot assume schema validation
    /// ran on the client.
    #[tokio::test]
    async fn rejects_duplicate_member_ids_with_multiple_coordinators_type() {
        let app = router(test_state());
        let body = serde_json::json!({
            "name": "dup-ids",
            "coordinator": "coord",
            "members": [
                { "id": "coord" },
                { "id": "coord", "authorizedBy": "coord" }
            ]
        })
        .to_string();
        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(
            parsed["type"], "/problems/formation/multiple-coordinators",
            "duplicate member ids must surface multipleCoordinators"
        );
    }

    /// ADR-0010 §Enforcement `cycle` discriminant. `authorizedBy: self`
    /// is the minimal cycle.
    #[tokio::test]
    async fn rejects_self_authorized_cycle() {
        let app = router(test_state());
        let body = serde_json::json!({
            "name": "self-cycle",
            "coordinator": "coord",
            "members": [
                { "id": "coord" },
                { "id": "worker-a", "authorizedBy": "worker-a" }
            ]
        })
        .to_string();
        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(parsed["type"], "/problems/formation/cycle");
    }

    /// Two-node cycle a→b→a; neither chains back to coordinator.
    #[tokio::test]
    async fn rejects_two_node_cycle() {
        let app = router(test_state());
        let body = serde_json::json!({
            "name": "two-cycle",
            "coordinator": "coord",
            "members": [
                { "id": "coord" },
                { "id": "a", "authorizedBy": "b" },
                { "id": "b", "authorizedBy": "a" }
            ]
        })
        .to_string();
        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(parsed["type"], "/problems/formation/cycle");
    }

    /// Orphan parent: `authorizedBy: ghost` where `ghost` is not in
    /// `members`. Without a parent edge the member has no narrowing
    /// path → `authorityNotNarrowing`.
    #[tokio::test]
    async fn rejects_orphan_parent_reference() {
        let app = router(test_state());
        let body = serde_json::json!({
            "name": "orphan-parent",
            "coordinator": "coord",
            "members": [
                { "id": "coord" },
                { "id": "worker-a", "authorizedBy": "ghost" }
            ]
        })
        .to_string();
        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(
            parsed["type"],
            "/problems/formation/authority-not-narrowing"
        );
    }

    /// Red-team finding: POST /v1/formations was using axum's default
    /// 2 MiB body limit. We cap at 64 KiB so a >64 KiB payload returns
    /// 413 Payload Too Large rather than burning CPU on serde parsing.
    #[tokio::test]
    async fn post_formation_oversized_body_returns_413() {
        let app = router(test_state());
        // Build a JSON document larger than 64 KiB by stuffing a long
        // `name` field. The body-limit layer rejects before serde
        // parses, so the document does not need to be semantically
        // valid past the limit.
        let big = "x".repeat(70 * 1024);
        let body =
            format!(r#"{{"name":"{big}","coordinator":"coord","members":[{{"id":"coord"}}]}}"#,);
        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(
            resp.status(),
            StatusCode::PAYLOAD_TOO_LARGE,
            "oversized body must surface 413; got {:?}",
            resp.status(),
        );
    }

    /// Sanity-probe: the parameterized route `/v1/formations/{id}`
    /// actually captures the path segment. Existing tests only hit the
    /// non-parameterized `/v1/formations`; if the route registration
    /// ever regresses to literal-brace matching (e.g. on a future axum
    /// upgrade that changes path syntax), this test fails loudly.
    ///
    /// We POST a real formation then GET it by id and check the
    /// returned record matches. A 404 with empty/non-problem+json
    /// content-type indicates router-level miss (literal route); a
    /// 404 with problem+json indicates handler-level not-found
    /// (route matched, formation absent). We expect 200.
    #[tokio::test]
    async fn get_formation_by_id_captures_path() {
        let state = test_state();
        let body = serde_json::json!({
            "name": "probe",
            "coordinator": "coord",
            "members": [
                { "id": "coord" },
                { "id": "worker-a", "authorizedBy": "coord" }
            ]
        })
        .to_string();
        let resp = router(state.clone())
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::CREATED);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        let id = parsed["id"].as_str().expect("uuid string");

        let resp = router(state)
            .oneshot(auth_req("GET", &format!("/v1/formations/{id}"), None))
            .await
            .expect("router response");
        assert_eq!(
            resp.status(),
            StatusCode::OK,
            "GET /v1/formations/<id> must capture the path segment; got {:?}",
            resp.status(),
        );
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(parsed["name"], "probe");
    }

    /// Red-team finding: `update_formation_status` previously returned
    /// the ADR-0010 `no-coordinator` discriminant for unknown state
    /// strings, hijacking a load-bearing admission-gate identifier.
    /// Unknown state is a generic bad-request.
    #[tokio::test]
    async fn unknown_state_returns_bad_request_problem_type() {
        let state = test_state();
        let body = serde_json::json!({
            "name": "demo",
            "coordinator": "coord",
            "members": [
                { "id": "coord" },
                { "id": "worker-a", "authorizedBy": "coord" }
            ]
        })
        .to_string();
        let resp = router(state.clone())
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::CREATED);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        let id = parsed["id"].as_str().expect("uuid string").to_owned();

        let bad = serde_json::json!({ "state": "TELEPORTING" }).to_string();
        let resp = router(state)
            .oneshot(auth_req(
                "POST",
                &format!("/v1/formations/{id}/status"),
                Some(&bad),
            ))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(
            parsed["type"], "/problems/bad-request",
            "unknown state must surface generic bad-request, not an ADR-0010 discriminant"
        );
    }
}