cellos-server 0.5.1

//! `/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`).
///
/// **Wire shapes accepted.** Operators may submit either:
///
/// 1. **Flat** (server-internal canonical form):
///    `{ "name": "...", "coordinator": "...", "members": [ { "id": "...", "authorizedBy": "..." } ] }`
/// 2. **Kubectl-style** (matches `contracts/schemas/formation-v1.schema.json`):
///    `{ "apiVersion": "cellos.dev/v1", "kind": "Formation",
///       "metadata": { "name": "..." },
///       "spec": { "coordinator": "...", "members": [ { "name": "...", "authorizedBy": "..." } ] } }`
///
/// `normalize_formation_document` runs first; everything below operates on
/// the canonical flat shape. See ADR-0010 §Enforcement for why admission
/// re-runs server-side regardless of client behaviour.
#[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 the wire payload, then normalize kubectl-style → flat. The
    // canonical internal form is the flat `{name, coordinator, members:
    // [{id, authorizedBy}]}` shape; the public schema documents the
    // kubectl form. `normalize_formation_document` collapses both into
    // the flat form before any admission validation runs, so existing
    // ADR-0010 checks below operate on a single shape. We then parse
    // twice: once into our validated struct, once kept as a generic
    // Value (already-normalized) so GET echoes a stable internal shape.
    let raw: serde_json::Value = serde_json::from_slice(&body)?;
    let normalized = normalize_formation_document(&raw)?;
    let doc: FormationDocument = serde_json::from_value(normalized.clone())?;

    validate_formation(&doc)?;

    let id = Uuid::new_v4();
    let record = FormationRecord {
        id,
        name: doc.name.clone(),
        status: FormationStatus::Pending,
        // Store the normalized (flat) form so GET, replay projection,
        // and downstream consumers see one stable shape regardless of
        // whether the operator submitted kubectl-style or flat-style.
        document: normalized,
    };

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

    // Emit formation.v1.created so the WebSocket stream and audit log see it.
    //
    // EVT-CONTENT-001: the second positional argument is the CloudEvents 1.0
    // `time` field (RFC3339 timestamp); published 0.5.0 incorrectly passed
    // the formation UUID here, producing spec-non-compliant envelopes that
    // failed schema-validating consumers (gateways, audit log, etc.).
    let cell_count = doc.members.len() as u32;
    let no_failed: &[String] = &[];
    let now_rfc3339 = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
    let event = cloud_event_v1_formation_created(
        &id.to_string(),
        &now_rfc3339,
        &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")))
}

/// GET /v1/formations/by-name/{name} — fetch one formation by name.
///
/// CTL-002 (E2E report): `cellctl describe formation <name>` and
/// `cellctl delete formation <name>` previously sent the name verbatim
/// to `/v1/formations/{id}` which rejected with `Invalid URL: UUID
/// parsing failed`. This parallel route lets cellctl address formations
/// by name without changing the existing UUID extractor on
/// `/v1/formations/{id}` (no parser ambiguity, one round-trip).
///
/// Name uniqueness is NOT currently enforced at admission (see CTL-002
/// follow-up); when multiple formations share a name this route returns
/// the first match by UUID order (BTreeMap iteration order). That is a
/// known limitation tracked separately from CTL-002.
pub async fn get_formation_by_name(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(name): Path<String>,
) -> Result<Json<FormationRecord>, AppError> {
    require_bearer(&headers, &state.api_token)?;
    let map = state.formations.read().await;
    map.values()
        .find(|r| r.name == name)
        .cloned()
        .map(Json)
        .ok_or_else(|| AppError::not_found(format!("formation '{name}' not found")))
}

/// DELETE /v1/formations/by-name/{name} — name-addressed counterpart of
/// [`delete_formation`]. Looks up the formation by name and delegates to
/// the same cancellation path so both routes emit the same
/// `formation.v1.failed` event and surface identical projection state.
pub async fn delete_formation_by_name(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(name): Path<String>,
) -> Result<StatusCode, AppError> {
    require_bearer(&headers, &state.api_token)?;
    // Resolve name → id under a read lock; release before re-acquiring
    // the write lock in `delete_formation`. The two-step resolve is
    // race-tolerant: if the formation is deleted between resolve and
    // delegate, the second step surfaces the same 404 the UUID-addressed
    // route would.
    let id = {
        let map = state.formations.read().await;
        map.values()
            .find(|r| r.name == name)
            .map(|r| r.id)
            .ok_or_else(|| AppError::not_found(format!("formation '{name}' not found")))?
    };
    delete_formation(State(state), headers, Path(id)).await
}

/// 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] = &[];
    // EVT-CONTENT-001: second arg is the CloudEvents 1.0 `time` field
    // (RFC3339); published 0.5.0 passed the UUID here. See create_formation.
    let now_rfc3339 = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
    let event = cloud_event_v1_formation_failed(
        &id.to_string(),
        &now_rfc3339,
        &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] = &[];
    // EVT-CONTENT-001: the second positional arg to every
    // `cloud_event_v1_formation_*` constructor is the CloudEvents 1.0 `time`
    // field (RFC3339); published 0.5.0 incorrectly passed the formation
    // UUID here. Capture one timestamp per HTTP request so all phases on a
    // single state transition share an envelope time.
    let now_rfc3339 = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);

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

/// Detect the wire shape of an incoming formation document and
/// normalize to the server's canonical flat form.
///
/// Two shapes are accepted (CTL-003 / SCHEMA-001 fix):
///
/// - **Flat** (canonical, what the server consumed historically):
///   `{ "name", "coordinator", "members": [ { "id", "authorizedBy"? } ] }`
/// - **Kubectl-style** (matches `contracts/schemas/formation-v1.schema.json`):
///   `{ "apiVersion": "cellos.dev/v1", "kind": "Formation",
///      "metadata": { "name", ... }, "spec": { "coordinator", "members": [...] } }`
///
/// Mapping (kubectl → flat):
///
/// | kubectl path                       | flat path                |
/// |------------------------------------|--------------------------|
/// | `metadata.name`                    | `name`                   |
/// | `spec.coordinator`                 | `coordinator`            |
/// | `spec.members[].name`              | `members[].id`           |
/// | `spec.members[].authorizedBy`      | `members[].authorizedBy` |
///
/// **Hybrid documents are rejected.** A document carrying BOTH a
/// top-level `name`/`coordinator`/`members` AND any of `apiVersion`,
/// `kind`, `metadata`, `spec` is ambiguous: the operator likely meant
/// one form but accidentally typed both. We surface a generic
/// `/problems/bad-request` (RFC 9457 §3.1) listing the conflicting
/// fields so cellctl can render a precise error.
///
/// `apiVersion` and `kind` MUST match the kubectl envelope literals
/// (`cellos.dev/v1` and `Formation`); other values are rejected.
///
/// Non-object roots (arrays, strings, etc.) are passed through
/// unchanged — the subsequent `serde_json::from_value::<FormationDocument>`
/// will fail with the same descriptive error operators already see.
fn normalize_formation_document(raw: &serde_json::Value) -> Result<serde_json::Value, AppError> {
    // Detection rules.
    //
    // Flat signal:    top-level `name` or `members` (the two fields a
    //                 flat document is required to carry).
    // Kubectl signal: top-level `apiVersion`, `kind`, `metadata`, or
    //                 `spec` (any of the four envelope fields).
    //
    // We look at the union so we can detect hybrids precisely.
    let Some(obj) = raw.as_object() else {
        // Non-object: let the downstream typed parse produce the
        // canonical error message.
        return Ok(raw.clone());
    };

    const FLAT_KEYS: &[&str] = &["name", "coordinator", "members"];
    const KUBECTL_KEYS: &[&str] = &["apiVersion", "kind", "metadata", "spec"];

    let flat_keys_present: Vec<&str> = FLAT_KEYS
        .iter()
        .copied()
        .filter(|k| obj.contains_key(*k))
        .collect();
    let kubectl_keys_present: Vec<&str> = KUBECTL_KEYS
        .iter()
        .copied()
        .filter(|k| obj.contains_key(*k))
        .collect();

    let has_flat = !flat_keys_present.is_empty();
    let has_kubectl = !kubectl_keys_present.is_empty();

    if has_flat && has_kubectl {
        return Err(AppError::bad_request(format!(
            "hybrid formation document: top-level flat field(s) {flat:?} \
             conflict with kubectl-style envelope field(s) {kubectl:?}; \
             pick exactly one shape (see contracts/schemas/formation-v1.schema.json)",
            flat = flat_keys_present,
            kubectl = kubectl_keys_present,
        )));
    }

    if !has_kubectl {
        // No envelope fields → flat (or so malformed the typed parse
        // will reject it). Pass through.
        return Ok(raw.clone());
    }

    // Kubectl-style. Validate envelope literals.
    let api_version = obj
        .get("apiVersion")
        .and_then(|v| v.as_str())
        .ok_or_else(|| {
            AppError::bad_request(
                "kubectl-style formation: missing or non-string 'apiVersion' (expected \"cellos.dev/v1\")"
                    .to_string(),
            )
        })?;
    if api_version != "cellos.dev/v1" {
        return Err(AppError::bad_request(format!(
            "kubectl-style formation: unsupported apiVersion '{api_version}' (expected \"cellos.dev/v1\")"
        )));
    }

    let kind = obj.get("kind").and_then(|v| v.as_str()).ok_or_else(|| {
        AppError::bad_request(
            "kubectl-style formation: missing or non-string 'kind' (expected \"Formation\")"
                .to_string(),
        )
    })?;
    if kind != "Formation" {
        return Err(AppError::bad_request(format!(
            "kubectl-style formation: unsupported kind '{kind}' (expected \"Formation\")"
        )));
    }

    let metadata = obj
        .get("metadata")
        .and_then(|v| v.as_object())
        .ok_or_else(|| {
            AppError::bad_request("kubectl-style formation: missing 'metadata' object".to_string())
        })?;
    let name = metadata
        .get("name")
        .and_then(|v| v.as_str())
        .ok_or_else(|| {
            AppError::bad_request("kubectl-style formation: missing 'metadata.name'".to_string())
        })?;

    let spec = obj.get("spec").and_then(|v| v.as_object()).ok_or_else(|| {
        AppError::bad_request("kubectl-style formation: missing 'spec' object".to_string())
    })?;

    let coordinator = spec
        .get("coordinator")
        .and_then(|v| v.as_str())
        .ok_or_else(|| {
            AppError::bad_request("kubectl-style formation: missing 'spec.coordinator'".to_string())
        })?;

    let members_raw = spec
        .get("members")
        .and_then(|v| v.as_array())
        .ok_or_else(|| {
            AppError::bad_request(
                "kubectl-style formation: missing or non-array 'spec.members'".to_string(),
            )
        })?;

    // Rewrite each member: `name` → `id`. `authorizedBy` carries
    // through. Any extra fields (`critical`, `spec`, future fields)
    // are preserved verbatim — the admission gate only inspects
    // `id`/`authorizedBy`, but we keep the rest so downstream
    // consumers (supervisor, projection) see what the operator wrote.
    let mut members_flat = Vec::with_capacity(members_raw.len());
    for (idx, m) in members_raw.iter().enumerate() {
        let m_obj = m.as_object().ok_or_else(|| {
            AppError::bad_request(format!(
                "kubectl-style formation: spec.members[{idx}] is not an object"
            ))
        })?;
        let member_name = m_obj.get("name").and_then(|v| v.as_str()).ok_or_else(|| {
            AppError::bad_request(format!(
                "kubectl-style formation: spec.members[{idx}] missing 'name'"
            ))
        })?;

        let mut rewritten = serde_json::Map::with_capacity(m_obj.len());
        rewritten.insert(
            "id".to_string(),
            serde_json::Value::String(member_name.to_string()),
        );
        for (k, v) in m_obj.iter() {
            if k == "name" {
                continue; // already rewritten to `id`
            }
            rewritten.insert(k.clone(), v.clone());
        }
        members_flat.push(serde_json::Value::Object(rewritten));
    }

    let mut flat = serde_json::Map::with_capacity(3);
    flat.insert(
        "name".to_string(),
        serde_json::Value::String(name.to_string()),
    );
    flat.insert(
        "coordinator".to_string(),
        serde_json::Value::String(coordinator.to_string()),
    );
    flat.insert(
        "members".to_string(),
        serde_json::Value::Array(members_flat),
    );

    Ok(serde_json::Value::Object(flat))
}

/// 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");
    }

    /// CTL-003 / SCHEMA-001 — kubectl-style happy path.
    /// `contracts/schemas/formation-v1.schema.json` documents the
    /// kubectl envelope; the server now accepts it via an admission
    /// adapter and normalizes to flat internally.
    #[tokio::test]
    async fn post_kubectl_style_formation_returns_201() {
        let app = router(test_state());
        let body = serde_json::json!({
            "apiVersion": "cellos.dev/v1",
            "kind": "Formation",
            "metadata": { "name": "kubectl-demo" },
            "spec": {
                "coordinator": "coord",
                "members": [
                    { "name": "coord" },
                    { "name": "worker-a", "authorizedBy": "coord" },
                    { "name": "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"], "kubectl-demo");
        assert!(parsed["id"].as_str().is_some());
    }

    /// Kubectl-style → ADR-0010 admission still fires on the
    /// normalized flat form. Missing-coordinator on a kubectl-shaped
    /// payload must surface the same `/problems/formation/no-coordinator`
    /// discriminant the flat path produces.
    #[tokio::test]
    async fn post_kubectl_style_missing_coordinator_returns_no_coordinator() {
        let app = router(test_state());
        let body = serde_json::json!({
            "apiVersion": "cellos.dev/v1",
            "kind": "Formation",
            "metadata": { "name": "missing-coord" },
            "spec": {
                "coordinator": "coord",
                "members": [
                    { "name": "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 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");
    }

    /// Hybrid: top-level `name` + top-level `apiVersion`. Operator
    /// ambiguity — reject with 400 `/problems/bad-request` listing the
    /// conflicting fields.
    #[tokio::test]
    async fn post_hybrid_formation_returns_400_bad_request() {
        let app = router(test_state());
        let body = serde_json::json!({
            "apiVersion": "cellos.dev/v1",
            "kind": "Formation",
            "metadata": { "name": "hybrid" },
            "spec": {
                "coordinator": "coord",
                "members": [ { "name": "coord" } ]
            },
            // Stray flat-style field — operator confused two shapes.
            "name": "hybrid",
            "members": [ { "id": "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/bad-request",
            "hybrid shape must surface a generic bad-request, not an admission discriminant"
        );
        let detail = parsed["detail"].as_str().unwrap_or_default();
        assert!(
            detail.contains("hybrid"),
            "detail must mention 'hybrid'; got {detail:?}"
        );
    }

    /// Kubectl-style with wrong `apiVersion` is rejected as bad-request
    /// before admission runs.
    #[tokio::test]
    async fn post_kubectl_style_wrong_api_version_returns_400() {
        let app = router(test_state());
        let body = serde_json::json!({
            "apiVersion": "cellos.dev/v2",
            "kind": "Formation",
            "metadata": { "name": "wrong-api" },
            "spec": {
                "coordinator": "coord",
                "members": [ { "name": "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/bad-request");
        let detail = parsed["detail"].as_str().unwrap_or_default();
        assert!(
            detail.contains("apiVersion") && detail.contains("cellos.dev/v2"),
            "detail must name the bad apiVersion; got {detail:?}"
        );
    }

    /// Kubectl-style with wrong `kind` is rejected as bad-request.
    #[tokio::test]
    async fn post_kubectl_style_wrong_kind_returns_400() {
        let app = router(test_state());
        let body = serde_json::json!({
            "apiVersion": "cellos.dev/v1",
            "kind": "Cell",
            "metadata": { "name": "wrong-kind" },
            "spec": {
                "coordinator": "coord",
                "members": [ { "name": "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/bad-request");
        let detail = parsed["detail"].as_str().unwrap_or_default();
        assert!(
            detail.contains("kind") && detail.contains("Cell"),
            "detail must name the bad kind; got {detail:?}"
        );
    }

    /// After a kubectl-style POST, the GET round-trip MUST echo the
    /// normalized (flat) shape so downstream consumers see one stable
    /// document layout.
    #[tokio::test]
    async fn kubectl_style_post_then_get_returns_normalized_flat_document() {
        let state = test_state();
        let body = serde_json::json!({
            "apiVersion": "cellos.dev/v1",
            "kind": "Formation",
            "metadata": { "name": "roundtrip" },
            "spec": {
                "coordinator": "coord",
                "members": [
                    { "name": "coord" },
                    { "name": "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);
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let parsed: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        let doc = &parsed["document"];
        assert_eq!(doc["name"], "roundtrip", "flat 'name' present");
        assert_eq!(doc["coordinator"], "coord", "flat 'coordinator' present");
        let members = doc["members"]
            .as_array()
            .expect("members array on normalized doc");
        assert_eq!(members.len(), 2);
        assert_eq!(members[0]["id"], "coord");
        assert_eq!(members[1]["id"], "worker-a");
        assert_eq!(members[1]["authorizedBy"], "coord");
        // Envelope fields stripped on normalization.
        assert!(
            doc.get("apiVersion").is_none(),
            "kubectl envelope must not leak into normalized doc"
        );
        assert!(doc.get("kind").is_none());
        assert!(doc.get("metadata").is_none());
        assert!(doc.get("spec").is_none());
    }

    /// 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"
        );
    }
}