cellos-server 0.5.3

//! `/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())?;

    // FUZZ-HIGH-1: name-validation MUST run before structural admission.
    // The fuzz wave admitted `""`, `"   "`, and `"a\nb"` as formation
    // names — names then corrupted by-name lookup (URL routing can't
    // carry newlines), log lines, and cellctl rendering. Reject hostile
    // names with the generic `/problems/bad-request` discriminant before
    // any other check so the operator sees the precise rule violated.
    validate_formation_name(&doc.name)?;

    validate_formation(&doc)?;

    // FUZZ-HIGH-2: enforce name uniqueness at admission. Without this
    // check two formations can share a name; `GET /v1/formations/by-name/{name}`
    // then returns the first match by UUID order and silently hides the
    // rest. Both lookup and admission must agree that names are unique.
    //
    // The duplicate-name check and the insert happen under the SAME
    // write lock so two concurrent admissions with the same name cannot
    // both succeed (TOCTOU: check-then-insert under a read lock would
    // race). The sibling FIX-RT3-HIGH-3 hardens
    // `delete_formation_by_name` against legacy duplicates that may
    // already exist in long-lived projections.
    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,
    };

    {
        let mut map = state.formations.write().await;
        if let Some(existing) = map.values().find(|r| r.name == doc.name) {
            return Err(AppError::new(
                AppErrorKind::Conflict,
                format!(
                    "formation name '{}' already in use by {}",
                    doc.name, existing.id
                ),
            ));
        }
        map.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.
    //
    // RT3-HIGH-3 (CTL-002-A): admission today does not yet enforce
    // name-uniqueness across formations (sibling fix-agent stream). When
    // two formations share a name, picking the BTreeMap-first UUID and
    // deleting THAT one is a silent wrong-deletion — exactly the
    // operator-trust failure the red-team flagged for DELETE. Defense
    // in depth: enumerate all matches; if there is more than one,
    // refuse and force the operator to disambiguate by UUID. (GET by
    // name still picks first; that is read-only and low-stakes.)
    let id = {
        let map = state.formations.read().await;
        let matches: Vec<Uuid> = map
            .values()
            .filter(|r| r.name == name)
            .map(|r| r.id)
            .collect();
        match matches.len() {
            0 => return Err(AppError::not_found(format!("formation '{name}' not found"))),
            1 => matches[0],
            _ => {
                // Sort so the detail string is deterministic across
                // BTreeMap-iteration variants (the test relies on this).
                let mut ids = matches;
                ids.sort();
                let id_list = ids
                    .iter()
                    .map(Uuid::to_string)
                    .collect::<Vec<_>>()
                    .join(", ");
                return Err(AppError::new(
                    AppErrorKind::Conflict,
                    format!(
                        "multiple formations share name '{name}': [{id_list}]; \
                         delete by UUID via /v1/formations/{{id}} to disambiguate"
                    ),
                ));
            }
        }
    };
    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)?;

    // EVT-CONTENT-001-C: capture the CloudEvents `time` field BEFORE
    // we take the write lock so it reflects when the state transition
    // *happened*, not when this task happened to win the lock. Under
    // load the lock can be held by sibling /status writers for hundreds
    // of milliseconds; capturing `time` after release skewed CloudEvent
    // ordering in audit consumers (they could see two events with the
    // same `id` but a `time` ordering that disagreed with stream-sequence
    // ordering). The replacement guarantees `time` is monotone with
    // request arrival.
    //
    // The wave-4 source-position test `update_formation_status_captures
    // _time_before_lock` asserts this ordering textually so a future
    // refactor that moves the capture back below the lock fails loudly.
    let now_rfc3339 = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);

    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. EVT-CONTENT-001-C moved the capture above
    // the write lock — `now_rfc3339` is reused below so all phases on
    // a single state transition share one envelope time and that time
    // reflects request arrival, not lock-release.

    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"];

    // CTL-003-B: unknown top-level keys in a kubectl-style document are
    // operator typos or projection drift (e.g. someone splicing the
    // server-side `status` block from a GET response back into a POST
    // body). Silently dropping them is the worst failure mode — the
    // operator thinks they applied A but the server admitted B. We
    // reject early with `/problems/bad-request` and name the offending
    // field so cellctl can render a precise error. The check runs on
    // the kubectl-signal arm only; the flat arm is the server's
    // historical contract and tolerating its extras would silently
    // break existing operators on upgrade.
    const KUBECTL_ALLOWED: &[&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());
    }

    // CTL-003-B: kubectl-style. Reject unknown top-level keys before
    // anything else so the operator sees a precise error and the
    // dropped-field failure mode is impossible by construction.
    let unknown_top_level: Vec<&str> = obj
        .keys()
        .map(|s| s.as_str())
        .filter(|k| !KUBECTL_ALLOWED.contains(k))
        .collect();
    if !unknown_top_level.is_empty() {
        return Err(AppError::bad_request(format!(
            "kubectl-style formation: unknown top-level field(s) {unknown_top_level:?}; \
             allowed: {KUBECTL_ALLOWED:?}",
        )));
    }

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

        // RT3-HIGH-3 (CTL-003-A): kubectl convention says `metadata.name`
        // (and at member level, `name`) is the canonical identifier;
        // spec-level fields MUST NOT shadow it. Earlier code populated
        // `id := name` and then iterated the member's keys with
        // last-write-wins, so an operator who declared BOTH
        // `name: alice` AND `id: bob` ended up with `id: bob` — the
        // operator's mistake silently overrode the kubectl identifier.
        // Admission is strict: reject with bad-request and name the
        // conflict so the operator knows which field to drop.
        if m_obj.contains_key("id") {
            return Err(AppError::bad_request(format!(
                "kubectl-style formation: spec.members[{idx}] declares both 'name' \
                 and 'id'; kubectl manifests address members by 'name' only — \
                 remove the 'id' field"
            )));
        }

        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`
            }
            // `id` is rejected above; the loop now only carries through
            // `authorizedBy` and operator-supplied extras.
            rewritten.insert(k.clone(), v.clone());
        }
        members_flat.push(serde_json::Value::Object(rewritten));
    }

    // CTL-003-B: preserve `metadata.labels` and `metadata.annotations`
    // verbatim on the normalized record so GET round-trips them. The
    // kubectl ecosystem uses these for selectors, ownership, and
    // tooling annotations; silently dropping them on admission breaks
    // the principle of least surprise for kubectl-style operators.
    // We place them under a flat `metadata` object so the flat shape
    // remains self-describing: anything not `name`/`coordinator`/
    // `members` is metadata, mirroring the kubectl envelope.
    let mut flat_metadata = serde_json::Map::new();
    if let Some(labels) = metadata.get("labels") {
        // Reject non-object labels early — kubectl convention is
        // `metadata.labels` is `{string: string}`. We don't enforce the
        // value type here (operators may emit numeric values that
        // serde will round-trip), but the outer container must be an
        // object.
        if !labels.is_object() {
            return Err(AppError::bad_request(
                "kubectl-style formation: 'metadata.labels' must be an object",
            ));
        }
        flat_metadata.insert("labels".to_string(), labels.clone());
    }
    if let Some(annotations) = metadata.get("annotations") {
        if !annotations.is_object() {
            return Err(AppError::bad_request(
                "kubectl-style formation: 'metadata.annotations' must be an object",
            ));
        }
        flat_metadata.insert("annotations".to_string(), annotations.clone());
    }

    let mut flat = serde_json::Map::with_capacity(4);
    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),
    );
    if !flat_metadata.is_empty() {
        flat.insert(
            "metadata".to_string(),
            serde_json::Value::Object(flat_metadata),
        );
    }

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

/// Validate a formation `name` against the conservative character set
/// the by-name lookup, URL routing, and log rendering can all carry
/// safely. FUZZ-WAVE-1 (FUZZ-HIGH-1) found admission accepted `""`,
/// `"   "`, and `"a\nb"`; downstream routing then broke (URLs can't
/// carry newlines, empty names are ambiguous) and log lines were
/// corrupted by control characters.
///
/// Rules (deliberately conservative — relaxing is non-breaking, but
/// tightening after release would be):
///
/// - **Length**: `1 ≤ name.len() ≤ 253` (matches DNS label length).
/// - **Characters**: `[A-Za-z0-9._-]` only. Reject every control
///   character, whitespace, newline, slash, and every byte outside
///   ASCII.
/// - **Edges**: cannot start or end with `-` or `.`.
/// - **Reserved**: cannot be `.` or `..` (these collide with relative
///   filesystem paths cellctl renders into).
///
/// All violations surface as `/problems/bad-request` (RFC 9457 §3.1)
/// with a `detail` string naming the rule. Tighter formation-specific
/// discriminants would be a breaking expansion of the type-uri namespace
/// for what is, fundamentally, a malformed input.
fn validate_formation_name(name: &str) -> Result<(), AppError> {
    // Length.
    if name.is_empty() {
        return Err(AppError::bad_request(
            "formation name must not be empty".to_string(),
        ));
    }
    if name.len() > 253 {
        return Err(AppError::bad_request(format!(
            "formation name length {} exceeds maximum of 253 bytes",
            name.len()
        )));
    }

    // Reserved names.
    if name == "." || name == ".." {
        return Err(AppError::bad_request(format!(
            "formation name '{name}' is reserved"
        )));
    }

    // Character class — operate on bytes; the allowed set is pure ASCII,
    // so any non-ASCII byte (and thus any multi-byte UTF-8 sequence)
    // fails this check and the operator gets a precise reason.
    for (idx, b) in name.as_bytes().iter().enumerate() {
        let ok = matches!(b, b'a'..=b'z' | b'A'..=b'Z' | b'0'..=b'9' | b'-' | b'_' | b'.');
        if !ok {
            // Render the offending byte: printable ASCII as itself,
            // otherwise as a `\xNN` escape so newlines/control bytes
            // don't smuggle themselves into the response body.
            let rendered = if b.is_ascii_graphic() {
                format!("'{}'", *b as char)
            } else {
                format!("\\x{b:02x}")
            };
            return Err(AppError::bad_request(format!(
                "formation name contains disallowed character {rendered} at byte offset {idx} \
                 (allowed: A-Z a-z 0-9 . - _)"
            )));
        }
    }

    // Edges. Safe to index by byte: ASCII-only at this point.
    let first = name.as_bytes()[0];
    let last = name.as_bytes()[name.len() - 1];
    if first == b'-' || first == b'.' {
        return Err(AppError::bad_request(format!(
            "formation name '{name}' must not start with '-' or '.'"
        )));
    }
    if last == b'-' || last == b'.' {
        return Err(AppError::bad_request(format!(
            "formation name '{name}' must not end with '-' or '.'"
        )));
    }

    Ok(())
}

/// 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. duplicateMemberId — `members[*].id` uniqueness. FUZZ-MED-4:
    // historically this case re-used the `multiple-coordinators`
    // discriminant because ADR-0010 §Consequences narrated the same
    // scenario ("two members both named `coord`"). Operators reading
    // the problem-type couldn't tell "manifest declares two
    // coordinators" (structural design error) from "manifest declares
    // two members with the same id" (typo). We now emit a dedicated
    // discriminant so cellctl can render a typo-specific hint. The
    // legacy `multiple-coordinators` discriminant is preserved on the
    // AppErrorKind enum so callers who explicitly construct it still
    // see the old wire shape — only the admission path here switched.
    let mut seen: HashSet<&str> = HashSet::new();
    for m in &doc.members {
        if !seen.insert(m.id.as_str()) {
            return Err(AppError::new(
                AppErrorKind::FormationDuplicateMemberId,
                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);
    }

    /// FUZZ-MED-4: duplicate `members[*].id` now surfaces its own
    /// `/problems/formation/duplicate-member-id` discriminant rather
    /// than riding on `multiple-coordinators`. The JSON schema declares
    /// `uniqueItems`; we re-enforce because the server cannot assume
    /// schema validation ran on the client.
    ///
    /// Wire-compat note: the old discriminant
    /// `/problems/formation/multiple-coordinators` is preserved on the
    /// enum and continues to be emitted by other admission paths that
    /// legitimately mean "two coordinators". Clients pinning on the
    /// old type still see 400; they just won't trigger on this
    /// specific case anymore.
    #[tokio::test]
    async fn rejects_duplicate_member_ids_with_dedicated_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/duplicate-member-id",
            "duplicate member ids must surface the dedicated discriminant"
        );
        // Detail must name the offending id so operators can find the
        // typo without re-reading their whole manifest.
        let detail = parsed["detail"].as_str().unwrap_or("");
        assert!(
            detail.contains("coord"),
            "detail must name the duplicate id, got: {detail}",
        );
    }

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

    /// RT3-HIGH-3 (CTL-003-A): a kubectl-style member that declares
    /// BOTH `name` and `id` is a manifest mistake — kubectl manifests
    /// address members by `name` only, and the previous normalization
    /// loop silently let the operator-supplied `id` win over the
    /// canonical name (`Map::insert` is last-write-wins). Admission is
    /// strict: reject with `/problems/bad-request` and name the conflict.
    #[tokio::test]
    async fn kubectl_member_with_explicit_id_returns_400() {
        let app = router(test_state());
        let body = serde_json::json!({
            "apiVersion": "cellos.dev/v1",
            "kind": "Formation",
            "metadata": { "name": "rt3-ctl-003-a" },
            "spec": {
                "coordinator": "alice",
                "members": [
                    // The mistake: declaring BOTH name AND id. The old
                    // code would set id := "name", then overwrite with
                    // id := "bob" from the spec field. New behaviour:
                    // reject the manifest outright.
                    { "name": "alice", "id": "bob" },
                    { "name": "worker-a", "authorizedBy": "alice" }
                ]
            }
        })
        .to_string();

        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(
            resp.status(),
            StatusCode::BAD_REQUEST,
            "manifest with both name+id at member level must be rejected"
        );
        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/bad-request",
            "kubectl id-conflict is a generic bad-request, not an ADR-0010 discriminant"
        );
        let detail = parsed["detail"].as_str().unwrap_or_default();
        assert!(
            detail.contains("'name'") && detail.contains("'id'"),
            "detail must name both conflicting fields; got {detail:?}"
        );
    }

    /// RT3-HIGH-3 (CTL-002-A): when two formations share a name,
    /// `DELETE /v1/formations/by-name/{name}` MUST refuse with 409
    /// Conflict instead of silently deleting the BTreeMap-first match.
    /// Silent wrong-deletion is the operator-trust failure mode the
    /// red-team flagged; admission-time uniqueness is being added in a
    /// sibling stream, but defense in depth keeps this guard in place.
    #[tokio::test]
    async fn delete_by_name_with_duplicates_returns_409() {
        let state = test_state();

        // Inject two FormationRecords with the same name directly into
        // the projection. This bypasses admission (which is the whole
        // point of the test: admission may or may not catch this, but
        // DELETE must NOT silently pick one).
        let id_a = Uuid::new_v4();
        let id_b = Uuid::new_v4();
        {
            let mut map = state.formations.write().await;
            for id in [id_a, id_b] {
                map.insert(
                    id,
                    FormationRecord {
                        id,
                        name: "rt3-dup".to_string(),
                        status: FormationStatus::Pending,
                        document: serde_json::json!({"name": "rt3-dup"}),
                    },
                );
            }
        }

        let resp = router(state.clone())
            .oneshot(auth_req("DELETE", "/v1/formations/by-name/rt3-dup", None))
            .await
            .expect("router response");
        assert_eq!(
            resp.status(),
            StatusCode::CONFLICT,
            "duplicate-name DELETE must surface 409, not silently delete"
        );

        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/conflict");
        let detail = parsed["detail"].as_str().unwrap_or_default();
        assert!(
            detail.contains(&id_a.to_string()) && detail.contains(&id_b.to_string()),
            "detail must list BOTH conflicting UUIDs so the operator can disambiguate; \
             got {detail:?}"
        );
        assert!(
            detail.contains("rt3-dup"),
            "detail must name the conflicting formation name; got {detail:?}"
        );

        // Defense in depth: neither formation should have been mutated.
        let map = state.formations.read().await;
        assert!(map.contains_key(&id_a), "id_a must still exist after 409");
        assert!(map.contains_key(&id_b), "id_b must still exist after 409");
        assert_eq!(map.get(&id_a).unwrap().status, FormationStatus::Pending);
        assert_eq!(map.get(&id_b).unwrap().status, FormationStatus::Pending);
    }

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

    // ----------------------------------------------------------------
    // FUZZ-HIGH-1 — name validation
    //
    // Fuzz wave 1 admitted hostile names (`""`, `"   "`, `"a\nb"`).
    // Every negative case below MUST return 400 problem+json with the
    // generic `/problems/bad-request` discriminant; every positive case
    // must reach the ADR-0010 admission gate and succeed with 201.
    // ----------------------------------------------------------------

    /// Build a minimal valid POST body for a given name. Used by both
    /// name-validation and uniqueness tests below. Keeps the test cases
    /// focused on the field actually under test.
    fn minimal_body(name: &str) -> String {
        serde_json::json!({
            "name": name,
            "coordinator": "coord",
            "members": [
                { "id": "coord" },
                { "id": "worker-a", "authorizedBy": "coord" }
            ]
        })
        .to_string()
    }

    /// Assert that POSTing `name` yields 400 `/problems/bad-request`.
    /// The `expect_in_detail` substring lets each case prove its OWN
    /// rule fired, not a different one — empty-name and 254-byte-name
    /// both produce bad-request, but for different reasons.
    async fn assert_name_rejected_bad_request(name: &str, expect_in_detail: &str) {
        let app = router(test_state());
        let body = minimal_body(name);
        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(
            resp.status(),
            StatusCode::BAD_REQUEST,
            "name {name:?} must be rejected with 400; got {:?}",
            resp.status()
        );
        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"),
            "name {name:?} must surface 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/bad-request",
            "name {name:?} must surface generic bad-request, not an admission discriminant; got {parsed}"
        );
        let detail = parsed["detail"].as_str().unwrap_or_default();
        assert!(
            detail.contains(expect_in_detail),
            "detail for {name:?} must contain {expect_in_detail:?}; got {detail:?}"
        );
    }

    /// FUZZ-HIGH-1 / F14 — admission accepted `""`. Reject as bad-request.
    #[tokio::test]
    async fn rejects_empty_name() {
        assert_name_rejected_bad_request("", "empty").await;
    }

    /// FUZZ-HIGH-1 — whitespace-only name. Spaces are not in the
    /// allow-set, so the character-class check fires.
    #[tokio::test]
    async fn rejects_whitespace_only_name() {
        assert_name_rejected_bad_request("   ", "disallowed character").await;
    }

    /// FUZZ-HIGH-1 / F44 — newline in name. The fuzz report flagged this
    /// because it corrupts log lines and breaks URL routing.
    #[tokio::test]
    async fn rejects_newline_in_name() {
        assert_name_rejected_bad_request("a\nb", "disallowed character").await;
    }

    /// FUZZ-HIGH-1 — tab character. Same class as newline: control byte
    /// outside the allow-set.
    #[tokio::test]
    async fn rejects_tab_in_name() {
        assert_name_rejected_bad_request("a\tb", "disallowed character").await;
    }

    /// FUZZ-HIGH-1 — embedded NUL byte. Breaks C-string boundaries in
    /// any downstream consumer that ever hands the name to a syscall.
    #[tokio::test]
    async fn rejects_nul_byte_in_name() {
        assert_name_rejected_bad_request("a\0b", "disallowed character").await;
    }

    /// FUZZ-HIGH-1 — non-ASCII (UTF-8 multi-byte). The allow-set is
    /// pure ASCII; emoji and accented characters fail the byte-class
    /// check.
    #[tokio::test]
    async fn rejects_non_ascii_in_name() {
        assert_name_rejected_bad_request("café", "disallowed character").await;
    }

    /// FUZZ-HIGH-1 — length cap. A 254-byte name (one over the DNS
    /// label limit) is rejected.
    #[tokio::test]
    async fn rejects_overlong_name() {
        let long = "a".repeat(254);
        assert_name_rejected_bad_request(&long, "exceeds maximum").await;
    }

    /// FUZZ-HIGH-1 — leading `-`. Mirrors DNS label rules; many
    /// downstream tools special-case leading hyphens as flags.
    #[tokio::test]
    async fn rejects_leading_hyphen() {
        assert_name_rejected_bad_request("-leading", "start with").await;
    }

    /// FUZZ-HIGH-1 — trailing `.`. Trailing dots collide with relative
    /// filesystem paths and DNS-style fully-qualified-name conventions.
    #[tokio::test]
    async fn rejects_trailing_dot() {
        assert_name_rejected_bad_request("trailing.", "end with").await;
    }

    /// FUZZ-HIGH-1 — reserved name `.`. Would alias the current-directory
    /// path segment in cellctl rendering. The reserved-name rule runs
    /// before the edge rule, so the operator sees the precise reason.
    #[tokio::test]
    async fn rejects_single_dot_reserved_name() {
        assert_name_rejected_bad_request(".", "reserved").await;
    }

    /// FUZZ-HIGH-1 — reserved name `..`. Same class as `.`; also begins
    /// with `.` so the edge rule fires first. We assert bad-request
    /// without pinning which rule trips — both are correct rejections.
    #[tokio::test]
    async fn rejects_double_dot_reserved_name() {
        let app = router(test_state());
        let body = minimal_body("..");
        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");
    }

    /// FUZZ-HIGH-1 positive — simple lowercase name.
    #[tokio::test]
    async fn accepts_simple_lowercase_name() {
        let app = router(test_state());
        let body = minimal_body("demo");
        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::CREATED);
    }

    /// FUZZ-HIGH-1 positive — hyphenated name (the dominant convention
    /// in the existing test corpus: `demo`, `with-cursor`, `valid-dag`).
    #[tokio::test]
    async fn accepts_hyphenated_name() {
        let app = router(test_state());
        let body = minimal_body("my-formation-v2");
        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::CREATED);
    }

    /// FUZZ-HIGH-1 positive — dotted name (mid-name dots and underscores
    /// are both allowed; only edge dots/hyphens are rejected).
    #[tokio::test]
    async fn accepts_dotted_and_underscored_name() {
        let app = router(test_state());
        let body = minimal_body("team.alpha_one");
        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(resp.status(), StatusCode::CREATED);
    }

    // ----------------------------------------------------------------
    // FUZZ-HIGH-2 — name uniqueness
    //
    // Two formations sharing a name break `GET /v1/formations/by-name/{name}`
    // (it returns the first match and hides the rest). Admission must
    // enforce uniqueness so by-name lookup is total.
    // ----------------------------------------------------------------

    /// FUZZ-HIGH-2 / F-dup-name — POST `name=demo` twice. The first
    /// request succeeds with 201; the second MUST return 409 with the
    /// `/problems/conflict` discriminant. The first formation must
    /// remain queryable by-name (proves we didn't accidentally evict it).
    #[tokio::test]
    async fn duplicate_name_returns_409() {
        let state = test_state();
        let body = minimal_body("demo");

        // First POST — succeeds.
        let resp = router(state.clone())
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(
            resp.status(),
            StatusCode::CREATED,
            "first POST must succeed"
        );
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let first: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        let first_id = first["id"]
            .as_str()
            .expect("first POST returned uuid")
            .to_owned();

        // Second POST with same name — MUST conflict.
        let resp = router(state.clone())
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(
            resp.status(),
            StatusCode::CONFLICT,
            "duplicate name must surface 409; got {:?}",
            resp.status()
        );
        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"),
            "409 must use 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/conflict");
        let detail = parsed["detail"].as_str().unwrap_or_default();
        assert!(
            detail.contains(&first_id),
            "conflict detail must name the existing UUID {first_id}; got {detail:?}"
        );
        assert!(
            detail.contains("demo"),
            "conflict detail must name the conflicting name; got {detail:?}"
        );

        // First formation MUST still be queryable by-name (proves the
        // second request didn't clobber or shadow it).
        let resp = router(state)
            .oneshot(auth_req("GET", "/v1/formations/by-name/demo", 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_eq!(
            parsed["id"].as_str(),
            Some(first_id.as_str()),
            "first formation must remain addressable by name"
        );
    }

    /// CTL-003-B: a kubectl manifest with `metadata.labels` and
    /// `metadata.annotations` MUST round-trip through POST → GET. The
    /// kubectl ecosystem leans on these for selectors and tooling
    /// annotations; silently dropping them on admission would break
    /// the principle of least surprise for kubectl-style operators.
    #[tokio::test]
    async fn kubectl_metadata_labels_and_annotations_round_trip() {
        let state = test_state();
        let body = serde_json::json!({
            "apiVersion": "cellos.dev/v1",
            "kind": "Formation",
            "metadata": {
                "name": "labelled",
                "labels": {
                    "app": "celltest",
                    "tier": "frontend"
                },
                "annotations": {
                    "operator": "ryan@cellos.dev",
                    "audit/ticket": "OPS-1234"
                }
            },
            "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 resp = router(state)
            .oneshot(auth_req("GET", "/v1/formations/by-name/labelled", 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 labels = &parsed["document"]["metadata"]["labels"];
        assert_eq!(labels["app"], "celltest", "label 'app' lost in round-trip");
        assert_eq!(
            labels["tier"], "frontend",
            "label 'tier' lost in round-trip"
        );

        let annotations = &parsed["document"]["metadata"]["annotations"];
        assert_eq!(
            annotations["operator"], "ryan@cellos.dev",
            "annotation 'operator' lost in round-trip"
        );
        assert_eq!(
            annotations["audit/ticket"], "OPS-1234",
            "annotation 'audit/ticket' lost in round-trip"
        );
    }

    /// CTL-003-B: a kubectl manifest with a `status` block (or any
    /// other unknown top-level field) MUST be rejected with 400 + a
    /// detail naming the field, so an operator who copy-pasted a GET
    /// response back into a POST body sees a precise error rather than
    /// silently shipping a stripped manifest.
    #[tokio::test]
    async fn kubectl_unknown_top_level_field_returns_400() {
        let app = router(test_state());
        let body = serde_json::json!({
            "apiVersion": "cellos.dev/v1",
            "kind": "Formation",
            "metadata": { "name": "with-status" },
            "spec": {
                "coordinator": "coord",
                "members": [{ "name": "coord" }]
            },
            "status": {
                "phase": "Running",
                "observedGeneration": 42
            }
        })
        .to_string();
        let resp = app
            .oneshot(auth_req("POST", "/v1/formations", Some(&body)))
            .await
            .expect("router response");
        assert_eq!(
            resp.status(),
            StatusCode::BAD_REQUEST,
            "unknown top-level 'status' must surface 400; 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["type"], "/problems/bad-request");
        let detail = parsed["detail"].as_str().unwrap_or_default();
        assert!(
            detail.contains("status"),
            "detail must name the offending field 'status'; got: {detail}",
        );
        assert!(
            detail.contains("unknown top-level"),
            "detail must say 'unknown top-level' so the operator knows the failure class; got: {detail}",
        );
    }

    /// CTL-003-B: `metadata.labels` must be an object — non-object
    /// labels (e.g. a string or array) are rejected with a precise
    /// detail so the operator sees the shape mismatch immediately
    /// rather than discovering it via a downstream selector failure.
    #[tokio::test]
    async fn kubectl_non_object_labels_returns_400() {
        let app = router(test_state());
        let body = serde_json::json!({
            "apiVersion": "cellos.dev/v1",
            "kind": "Formation",
            "metadata": {
                "name": "bad-labels",
                "labels": "app=celltest"
            },
            "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();
        let detail = parsed["detail"].as_str().unwrap_or_default();
        assert!(
            detail.contains("metadata.labels"),
            "detail must name 'metadata.labels'; got: {detail}",
        );
    }

    /// EVT-CONTENT-001-C source-position test: `now_rfc3339` MUST be
    /// captured BEFORE `state.formations.write().await` is acquired so
    /// the CloudEvent `time` field reflects transition arrival, not
    /// post-commit time. We assert textually because the alternative —
    /// driving two concurrent /status updates and comparing wall-clock
    /// against event order — is non-deterministic and ties the unit
    /// test to a real broker. A textual ordering check is brittle to
    /// formatting changes but precise about the invariant: regressions
    /// that move the capture back below the lock fail loudly.
    #[test]
    fn update_formation_status_captures_time_before_lock() {
        let src = include_str!("formations.rs");
        // Locate the function body. The function signature is stable
        // (its arguments are part of the axum handler contract); we
        // anchor on the `pub async fn update_formation_status(` opener
        // and walk to the matching closing brace by depth-counting
        // outside string literals. For this assertion we only need a
        // window that is unambiguously inside the function — a window
        // bounded by the next sibling `fn`/`async fn` is sufficient
        // and avoids the full brace-matcher complexity.
        let fn_start = src
            .find("pub async fn update_formation_status(")
            .expect("update_formation_status not found");
        let after_fn = &src[fn_start..];
        // The next `async fn ` or `fn ` outside of this function bounds
        // the body. We look for one indented at 0 columns (top-level).
        let next_top_level_fn = after_fn[1..]
            .find("\nfn ")
            .map(|i| i + 1)
            .or_else(|| after_fn[1..].find("\nasync fn ").map(|i| i + 1))
            .or_else(|| after_fn[1..].find("\npub fn ").map(|i| i + 1))
            .or_else(|| after_fn[1..].find("\npub async fn ").map(|i| i + 1))
            .unwrap_or(after_fn.len() - 1);
        let body = &after_fn[..next_top_level_fn];

        let time_capture_pos = body
            .find("let now_rfc3339 = chrono::Utc::now().to_rfc3339_opts(")
            .expect(
                "EVT-CONTENT-001-C regression: `let now_rfc3339 = chrono::Utc::now()` not found \
                 in update_formation_status",
            );
        let lock_pos = body.find("state.formations.write().await").expect(
            "update_formation_status no longer acquires a write lock — test must be updated",
        );

        assert!(
            time_capture_pos < lock_pos,
            "EVT-CONTENT-001-C regression: `now_rfc3339` was captured at byte offset \
             {time_capture_pos} but the write lock was taken at offset {lock_pos}. \
             Move the `chrono::Utc::now().to_rfc3339_opts(...)` call ABOVE the \
             `state.formations.write().await` line so CloudEvent `time` reflects \
             transition arrival rather than post-commit wall-clock.",
        );
    }
}