car_server_core/

wire_schema.rs

//! Deterministic release wire-schema generation and the `server.schema` payload.
//!
//! The committed `docs/wire-schema.json` is generated from the very Rust types
//! the daemon serializes at each covered boundary. The daemon embeds that exact
//! release artifact instead of regenerating at runtime, so a release binary and
//! its published file cannot describe different contracts.
//!
//! ## Every covered surface is emitted from the type it is generated from
//!
//! The point of the artifact is that a consumer pinned to a digest can trust it
//! after an upgrade. A schema transcribed from an inline `serde_json::json!`
//! literal cannot deliver that: adding a field to the literal leaves the
//! document and the digest unchanged, which is exactly the 0.53.0
//! `server.handshake` break this file exists to prevent. So there are no
//! schema-only mirrors here. Each type below is the value a handler actually
//! returns:
//!
//! | Schema key | Emitter |
//! |---|---|
//! | `rpc.server.handshake.result` | [`ServerHandshakeResult`], returned by `handler::handle_server_handshake` |
//! | `rpc.tools.*.result` | [`ToolsListResult`] and the other typed tool results below, returned by the matching `handler::handle_tools_*` function |
//! | `rpc.state.*.result` | the typed state results below, returned by the matching `handler::handle_state_*` function |
//! | `rpc.capabilities.list.result` | [`CapabilitiesListResult`], returned by `handler::handle_capabilities_list` |
//! | `rpc.server.schema.result` | [`ServerSchemaResult`], returned by [`committed_payload`] |
//! | `cli.car_inspect.result` | [`ManagedAgentListRow`] from `handler::handle_agents_list`, and [`DeclarativeAgentRow`] from `coder::rpc::declarative_row` |
//! | `journal.event` | [`car_eventlog::Event`] |
//! | `rpc.infer.result` | [`car_inference::InferenceResult`] |
//! | `rpc.models.catalog_snapshot.result` | [`car_inference::catalog_identity::CatalogSnapshot`] |
//! | `type.action_result` | [`car_ir::ActionResult`] |
//!
//! Adding a field to any of those is therefore a change to the generated
//! document and to the digest, and is a compile error to do by any other route.
//!
//! ## Why `required` is still written out by hand, and how it is checked
//!
//! schemars derives `required` from serde's *input* rules: a field with
//! `#[serde(default)]` is optional, and an `Option<T>` is optional. CAR's output
//! rules are different — plenty of fields are always emitted (as `null` when
//! absent) precisely so a consumer can tell "not measured" from "your protocol
//! version has no such field". `#[schemars(required)]` does not bridge the gap:
//! it is inert on a `serde(default)` field, and on an `Option` it *removes* the
//! `null` from the type, which would be a second lie. So [`document`] states the
//! always-emitted set explicitly — and
//! `tests::required_lists_exactly_the_fields_a_minimal_value_emits` asserts
//! each list against a value whose every `Option` is `None`, where the emitted
//! key set *is* the always-emitted set. The list cannot drift in either
//! direction without a red test.
//!
//! ## The positive control is a source change, not a document edit
//!
//! Mutating the already-generated document only proves SHA-256 is sensitive to
//! bytes. To prove the *pipeline* is sensitive to a wire change, add a field to
//! one of the emitter types above in a scratch checkout, run
//! `bash scripts/build-wire-schema.sh`, confirm `docs/wire-schema.sha256`
//! changed, and revert. `tests::every_covered_schema_validates_its_real_emitted_value`
//! is the standing guard between those runs: it compiles each generated schema
//! and validates a real serialized value against it, so `additionalProperties:
//! false` turns any emitter/schema divergence into a failing test.
//!
//! ## The release version is deliberately outside the digested bytes
//!
//! `docs/wire-schema.json` carries no version string. `scripts/release.sh`
//! rewrites the workspace version in its bump commit and does not regenerate
//! this pair; if the version were digested, that commit would fail the required
//! `test` check and, if forced past, publish an artifact naming the previous
//! release. The version a caller needs is reported at serve time instead, in
//! [`ServerSchemaResult::car_version`], read from the running binary.

use std::collections::{BTreeMap, BTreeSet};

use schemars::{schema::RootSchema, schema_for, JsonSchema};
use serde::Serialize;
use serde_json::{json, Value};
use sha2::{Digest, Sha256};

pub const FORMAT: &str = "car.wire-schema.v1";
pub const DIGEST_ALGORITHM: &str = "sha256";
pub const FOLLOW_UP_BEAD: &str = "car-86cq.1";

// Read from the CRATE, not from `docs/`. `include_str!` reaching outside the
// package root compiles fine in the workspace and then fails
// `cargo package --verify`, whose tarball can only contain files under this
// directory:
//
//   error: couldn't read `src/../../../../docs/wire-schema.json`
//   error: failed to verify package tarball
//
// That made car-server-core unpublishable from the moment #1598 introduced
// these constants, and because crates.io is the last channel a release
// touches, it surfaced only when v0.55.0 reached the post-CI tail.
// `docs/wire-schema.json` remains the published copy; both are written by
// scripts/build-wire-schema.sh and compared by
// scripts/check-wire-schema-freshness.sh, so they cannot drift.
const COMMITTED_SCHEMA: &str = include_str!("../wire-schema.json");
const COMMITTED_DIGEST: &str = include_str!("../wire-schema.sha256");

/// The `server.handshake` result. Built and serialized by
/// `handler::handle_server_handshake`; nothing else may write that reply.
#[derive(Serialize, JsonSchema)]
pub(crate) struct ServerHandshakeResult {
    pub protocol_version: u32,
    pub server_version: String,
    pub client_protocol_version: u64,
    pub client_version: String,
    pub negotiated_capabilities: Vec<String>,
    pub assistant_name: String,
    pub assistant_aliases: Vec<String>,
    pub assistant_brand: String,
}

/// The `tools.list` result. Built and serialized by `handler::handle_tools_list`.
#[derive(Serialize, JsonSchema)]
pub(crate) struct ToolsListResult {
    pub tools: Vec<car_ir::ToolSchema>,
    pub count: usize,
}

/// The numeric `tools.register` result.
#[derive(Serialize, JsonSchema)]
#[serde(transparent)]
pub(crate) struct ToolsRegisterResult(pub usize);

/// The `tools.unregister` result.
#[derive(Serialize, JsonSchema)]
pub(crate) struct ToolsUnregisterResult {
    pub unregistered: String,
    pub removed: u32,
}

/// The `tools.cancel` result.
#[derive(Serialize, JsonSchema)]
pub(crate) struct ToolsCancelResult {
    pub cancelled: bool,
}

/// The `tools.stream.subscribe` result.
#[derive(Serialize, JsonSchema)]
pub(crate) struct ToolsStreamSubscribeResult {
    pub subscribed: bool,
}

/// The exact string returned by `state.set`.
#[derive(Serialize, JsonSchema)]
#[serde(rename_all = "lowercase")]
pub(crate) enum StateSetResult {
    Ok,
}

/// The boolean returned by `state.exists`.
#[derive(Serialize, JsonSchema)]
#[serde(transparent)]
pub(crate) struct StateExistsResult(pub bool);

/// The string array returned by `state.keys`.
#[derive(Serialize, JsonSchema)]
#[serde(transparent)]
pub(crate) struct StateKeysResult(pub Vec<String>);

/// The arbitrary-value map returned by `state.snapshot`.
#[derive(Serialize, JsonSchema)]
#[serde(transparent)]
pub(crate) struct StateSnapshotResult(
    #[schemars(with = "BTreeMap<String, Value>")] pub serde_json::Map<String, Value>,
);

/// The closed caller roles emitted by `capabilities.list`.
#[derive(Clone, Copy, Serialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub(crate) enum CapabilityRole {
    Agent,
    Owner,
    Operator,
    Host,
}

impl CapabilityRole {
    pub(crate) fn from_manifest(role: &str) -> Self {
        match role {
            "agent" => Self::Agent,
            "owner" => Self::Owner,
            "operator" => Self::Operator,
            "host" => Self::Host,
            other => panic!("generated RPC capability has unknown role `{other}`"),
        }
    }
}

/// One source-derived method row in `capabilities.list`.
#[derive(Serialize, JsonSchema)]
pub(crate) struct CapabilityMethodRow {
    pub method: String,
    pub role: CapabilityRole,
}

/// The `capabilities.list` result.
#[derive(Serialize, JsonSchema)]
pub(crate) struct CapabilitiesListResult {
    pub caller_role: CapabilityRole,
    pub count: usize,
    pub methods: Vec<CapabilityMethodRow>,
}

/// The `server.schema` result. Built and serialized by [`committed_payload`].
#[derive(Serialize, JsonSchema)]
pub(crate) struct ServerSchemaResult {
    pub schema: Value,
    pub digest: String,
    pub digest_algorithm: DigestAlgorithm,
    /// The release this daemon binary is, read at serve time. Deliberately not
    /// part of the digested document — see the module header.
    pub car_version: String,
}

#[derive(Serialize, JsonSchema)]
#[serde(rename_all = "lowercase")]
pub(crate) enum DigestAlgorithm {
    Sha256,
}

/// A lifecycle-managed agent as the daemon publishes it: the supervisor's
/// `ManagedAgent` minus its per-agent token.
///
/// The redaction lives in this projection rather than on `AgentSpec::token`
/// itself because `AgentSpec` is also the on-disk `agents.json` format, and
/// `skip_serializing` there would blank every token on the next manifest write.
#[derive(Serialize, JsonSchema)]
pub(crate) struct ManagedAgentWire {
    pub id: String,
    pub name: String,
    pub command: String,
    pub args: Vec<String>,
    pub cwd: Option<String>,
    pub env: BTreeMap<String, String>,
    pub restart: car_registry::supervisor::RestartPolicy,
    pub max_restarts: u32,
    pub backoff_secs: u64,
    pub auto_start: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub method_allowlist: Option<Vec<String>>,
    pub capabilities: Vec<String>,
    pub status: car_registry::supervisor::AgentStatus,
    pub pid: Option<u32>,
    pub last_exit_code: Option<i32>,
    pub restart_count: u32,
    pub started_at: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub blocked_by_pid: Option<i32>,
}

impl ManagedAgentWire {
    pub(crate) fn from_managed(agent: &car_registry::supervisor::ManagedAgent) -> Self {
        let spec = &agent.spec;
        Self {
            id: spec.id.clone(),
            name: spec.name.clone(),
            command: spec.command.clone(),
            args: spec.args.clone(),
            cwd: spec
                .cwd
                .as_ref()
                .map(|path| path.to_string_lossy().into_owned()),
            env: spec.env.clone(),
            restart: spec.restart,
            max_restarts: spec.max_restarts,
            backoff_secs: spec.backoff_secs,
            auto_start: spec.auto_start,
            method_allowlist: spec.method_allowlist.clone(),
            capabilities: spec.capabilities.clone(),
            status: agent.status,
            pid: agent.pid,
            last_exit_code: agent.last_exit_code,
            restart_count: agent.restart_count,
            started_at: agent.started_at,
            blocked_by_pid: agent.blocked_by_pid,
        }
    }
}

/// One `agents.list` row: the redacted agent plus the decorations only the
/// daemon holding the connection can supply.
#[derive(Serialize, JsonSchema)]
pub(crate) struct ManagedAgentListRow {
    #[serde(flatten)]
    pub agent: ManagedAgentWire,
    /// Whether the supervised process has called `session.auth { agent_id }`
    /// and bound a WebSocket connection to this daemon.
    pub attached: bool,
    /// The attached agent's current model-visible tool names, when it supports
    /// the bounded `agent.chat.tools` reverse query. Omitted for detached or
    /// older supervised agents rather than guessing from broad capabilities.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub tools: Option<Vec<String>>,
    pub manifest_path: String,
    pub log_path: String,
    pub stderr_log_path: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
}

/// A declarative (in-daemon) agent rendered as an `agents.list` row. Built and
/// serialized by `coder::rpc::declarative_row`.
#[derive(Serialize, JsonSchema)]
pub(crate) struct DeclarativeAgentRow {
    pub id: String,
    pub name: String,
    pub kind: DeclarativeAgentKind,
    pub enabled: bool,
    pub capabilities: Vec<String>,
    pub description: String,
    pub tools: Vec<String>,
    pub goal: Option<car_registry::declarative::DeclarativeGoal>,
    pub scenarios: usize,
}

impl DeclarativeAgentRow {
    pub(crate) fn from_spec(spec: &car_registry::declarative::DeclarativeAgentSpec) -> Self {
        let description = if spec.standing_goal.trim().is_empty() {
            spec.identity.trim()
        } else {
            spec.standing_goal.trim()
        };
        Self {
            id: spec.id.clone(),
            name: spec.name.clone(),
            kind: DeclarativeAgentKind::Declarative,
            enabled: spec.enabled,
            capabilities: vec!["chat".to_string()],
            description: description.to_string(),
            tools: spec.tools.clone(),
            goal: spec.goal.clone(),
            scenarios: spec.scenarios.len(),
        }
    }
}

#[derive(Serialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub(crate) enum DeclarativeAgentKind {
    Declarative,
}

/// `car inspect` and `agents.list` return either kind of row, untagged.
// The declarative arm is constructed through `declarative_row`, which returns a
// `Value`; this enum exists to generate the union both arms are validated
// against.
#[allow(dead_code)]
#[derive(Serialize, JsonSchema)]
#[serde(untagged)]
pub(crate) enum CarInspectResult {
    Managed(ManagedAgentListRow),
    Declarative(DeclarativeAgentRow),
}

fn closed_schema<T: JsonSchema>() -> Value {
    let root: RootSchema = schema_for!(T);
    let mut value = serde_json::to_value(root).expect("RootSchema serialization is infallible");
    close_declared_objects(&mut value);
    value
}

/// Every client-to-daemon result surface, derived from the generated dispatch
/// inventory plus the pre-dispatch handshake.
fn rpc_method_inventory() -> BTreeSet<&'static str> {
    let mut methods = BTreeSet::from(["server.handshake"]);
    for (method, _) in crate::generated_rpc_capabilities::RPC_CAPABILITIES {
        assert!(
            methods.insert(method),
            "duplicate daemon RPC method in source inventory: {method}"
        );
    }
    methods
}

fn insert_rpc_schema(
    schemas: &mut BTreeMap<String, Value>,
    covered_rpc_methods: &mut BTreeSet<&'static str>,
    method: &'static str,
    schema: Value,
) {
    assert!(
        rpc_method_inventory().contains(method),
        "wire schema covers unknown daemon RPC method: {method}"
    );
    assert!(
        covered_rpc_methods.insert(method),
        "wire schema covers daemon RPC method twice: {method}"
    );
    let key = format!("rpc.{method}.result");
    assert!(
        schemas.insert(key.clone(), schema).is_none(),
        "duplicate wire schema key: {key}"
    );
}

/// Read the closed strings from the generated `EventKind` schema. This keeps
/// the payload coverage inventory source-derived: adding an enum variant makes
/// the generated list grow even before that variant gains a typed payload.
fn event_kind_inventory(event_schema: &Value) -> BTreeSet<String> {
    fn collect(value: &Value, out: &mut BTreeSet<String>) {
        match value {
            Value::Array(values) => {
                for value in values {
                    collect(value, out);
                }
            }
            Value::Object(map) => {
                if let Some(values) = map.get("enum").and_then(Value::as_array) {
                    for value in values {
                        if let Some(value) = value.as_str() {
                            out.insert(value.to_string());
                        }
                    }
                }
                for value in map.values() {
                    collect(value, out);
                }
            }
            _ => {}
        }
    }

    let mut kinds = BTreeSet::new();
    collect(&event_schema["definitions"]["EventKind"], &mut kinds);
    assert!(!kinds.is_empty(), "generated EventKind inventory is empty");
    kinds
}

fn require_fields(schema: &mut Value, fields: &[&str]) {
    schema["required"] = Value::Array(
        fields
            .iter()
            .map(|field| Value::String((*field).to_string()))
            .collect(),
    );
}

/// Mark every declared property of every object in `value` as required.
///
/// **Precondition: no type in the subtree carries `skip_serializing_if`.** This
/// is used only for the model-catalog subtree, where it holds today and where
/// annotating ~40 fields across `ModelSchema`, `ModelSource`, `CostModel` and
/// friends would be the alternative. It becomes silently wrong the moment
/// someone adds a conditional field, so the precondition is asserted, not
/// assumed: `tests::catalog_snapshot_declares_no_conditionally_emitted_field`
/// serializes a catalog row whose every `Option` is `None` and validates it
/// against this schema, which fails the moment a field stops being emitted.
fn require_all_declared_fields(value: &mut Value) {
    match value {
        Value::Array(values) => {
            for value in values {
                require_all_declared_fields(value);
            }
        }
        Value::Object(map) => {
            for value in map.values_mut() {
                require_all_declared_fields(value);
            }
            if let Some(Value::Object(properties)) = map.get("properties") {
                let fields = properties.keys().cloned().map(Value::String).collect();
                map.insert("required".into(), Value::Array(fields));
            }
        }
        _ => {}
    }
}

/// Apply a required list to every occurrence of a derived type in `schema`.
///
/// Located by NAME rather than by a fixed `definitions["X"]` path, because
/// whether schemars hoists a type into `definitions` or inlines it at the
/// property is an implementation detail that moves with the field's attributes
/// — `Option<DeclarativeGoal>` hoists, the same field with a `schemars`
/// attribute inlines. A hard-coded path panics the day that flips. Hoisted
/// definitions are keyed by the type name and carry no `title`; a root or
/// inlined copy carries `title` and no key. Both are matched, and every copy is
/// updated when a type appears more than once.
fn require_fields_for(schema: &mut Value, title: &str, fields: &[&str]) {
    fn walk(
        value: &mut Value,
        title: &str,
        fields: &[&str],
        applied: &mut bool,
        own_name: Option<&str>,
        entries_are_definitions: bool,
    ) {
        match value {
            Value::Array(values) => {
                for value in values {
                    walk(value, title, fields, applied, None, false);
                }
            }
            Value::Object(map) => {
                let named = own_name == Some(title)
                    || map.get("title").and_then(Value::as_str) == Some(title);
                if named && map.contains_key("properties") {
                    let required = fields
                        .iter()
                        .map(|field| Value::String((*field).to_string()))
                        .collect();
                    map.insert("required".into(), Value::Array(required));
                    *applied = true;
                }
                for (key, child) in map.iter_mut() {
                    let child_name = entries_are_definitions.then(|| key.clone());
                    let child_defines = !entries_are_definitions && key == "definitions";
                    walk(
                        child,
                        title,
                        fields,
                        applied,
                        child_name.as_deref(),
                        child_defines,
                    );
                }
            }
            _ => {}
        }
    }

    let mut applied = false;
    walk(schema, title, fields, &mut applied, None, false);
    assert!(
        applied,
        "generated schema has no object named {title} to require fields on"
    );
}

/// Output schemas describe exact emitted objects. Any schema object that
/// declares named properties and is not already a map receives
/// `additionalProperties: false`; HashMap/JSON Value fields retain their own
/// explicitly permissive shape.
fn close_declared_objects(value: &mut Value) {
    match value {
        Value::Array(values) => {
            for value in values {
                close_declared_objects(value);
            }
        }
        Value::Object(map) => {
            for value in map.values_mut() {
                close_declared_objects(value);
            }
            // Schemars evaluates serde default functions. `Utc::now` would put
            // generation time into an OUTPUT contract and change the digest on
            // every run; defaults are input semantics, so remove them all.
            map.remove("default");
            if map.contains_key("properties") && !map.contains_key("additionalProperties") {
                map.insert("additionalProperties".into(), Value::Bool(false));
            }
        }
        _ => {}
    }
}

/// Build the deterministic schema document from the canonical Rust wire types.
///
/// Property sets, types and enum values come from the emitter types themselves.
/// `required` is stated here because serde's input rules and CAR's output rules
/// disagree — see the module header, and the test that checks every list below
/// against a real minimal value.
pub fn document() -> Value {
    let mut schemas = BTreeMap::new();
    let mut covered_rpc_methods = BTreeSet::new();

    let mut inspect = closed_schema::<CarInspectResult>();
    require_fields_for(
        &mut inspect,
        "ManagedAgentListRow",
        MANAGED_AGENT_LIST_ROW_REQUIRED,
    );
    require_fields_for(
        &mut inspect,
        "DeclarativeAgentRow",
        DECLARATIVE_AGENT_ROW_REQUIRED,
    );
    require_fields_for(&mut inspect, "DeclarativeGoal", DECLARATIVE_GOAL_REQUIRED);
    schemas.insert("cli.car_inspect.result".to_string(), inspect);

    let mut event = closed_schema::<car_eventlog::Event>();
    require_fields(&mut event, EVENT_REQUIRED);
    let all_event_kinds = event_kind_inventory(&event);
    schemas.insert("journal.event".to_string(), event);

    insert_rpc_schema(
        &mut schemas,
        &mut covered_rpc_methods,
        "capabilities.list",
        closed_schema::<CapabilitiesListResult>(),
    );

    let mut inference = closed_schema::<car_inference::InferenceResult>();
    require_fields(&mut inference, INFERENCE_RESULT_REQUIRED);
    require_fields_for(&mut inference, "TokenUsage", TOKEN_USAGE_REQUIRED);
    require_fields_for(&mut inference, "ToolCall", TOOL_CALL_REQUIRED);
    require_fields_for(&mut inference, "ThinkingBlock", THINKING_BLOCK_REQUIRED);
    require_fields_for(&mut inference, "BoundingBox", BOUNDING_BOX_REQUIRED);
    require_fields_for(&mut inference, "FallbackFrom", FALLBACK_FROM_REQUIRED);
    insert_rpc_schema(&mut schemas, &mut covered_rpc_methods, "infer", inference);

    let mut catalog = closed_schema::<car_inference::catalog_identity::CatalogSnapshot>();
    require_all_declared_fields(&mut catalog);
    // `Quantization`'s hand-written `Serialize` omits `bits` and `group_size`
    // when the label already carries them, so it is the one member of the
    // catalog subtree the blanket pass above must not close over.
    require_fields_for(
        &mut catalog,
        "QuantizationObjectWireSchema",
        QUANTIZATION_OBJECT_REQUIRED,
    );
    insert_rpc_schema(
        &mut schemas,
        &mut covered_rpc_methods,
        "models.catalog_snapshot",
        catalog,
    );

    // The handshake and schema replies carry no `Option` and no serde default,
    // so schemars already requires every field; nothing to state here.
    insert_rpc_schema(
        &mut schemas,
        &mut covered_rpc_methods,
        "server.handshake",
        closed_schema::<ServerHandshakeResult>(),
    );
    insert_rpc_schema(
        &mut schemas,
        &mut covered_rpc_methods,
        "server.schema",
        closed_schema::<ServerSchemaResult>(),
    );

    insert_rpc_schema(
        &mut schemas,
        &mut covered_rpc_methods,
        "state.get",
        closed_schema::<Value>(),
    );
    insert_rpc_schema(
        &mut schemas,
        &mut covered_rpc_methods,
        "state.set",
        closed_schema::<StateSetResult>(),
    );
    insert_rpc_schema(
        &mut schemas,
        &mut covered_rpc_methods,
        "state.exists",
        closed_schema::<StateExistsResult>(),
    );
    insert_rpc_schema(
        &mut schemas,
        &mut covered_rpc_methods,
        "state.keys",
        closed_schema::<StateKeysResult>(),
    );
    insert_rpc_schema(
        &mut schemas,
        &mut covered_rpc_methods,
        "state.snapshot",
        closed_schema::<StateSnapshotResult>(),
    );

    insert_rpc_schema(
        &mut schemas,
        &mut covered_rpc_methods,
        "tools.register",
        closed_schema::<ToolsRegisterResult>(),
    );
    let mut tools = closed_schema::<ToolsListResult>();
    require_fields_for(&mut tools, "ToolSchema", TOOL_SCHEMA_REQUIRED);
    insert_rpc_schema(&mut schemas, &mut covered_rpc_methods, "tools.list", tools);
    insert_rpc_schema(
        &mut schemas,
        &mut covered_rpc_methods,
        "tools.unregister",
        closed_schema::<ToolsUnregisterResult>(),
    );
    let mut poll = closed_schema::<Option<car_engine::tool_handles::ToolPollResult>>();
    require_fields_for(&mut poll, "ToolPollResult", TOOL_POLL_RESULT_REQUIRED);
    insert_rpc_schema(&mut schemas, &mut covered_rpc_methods, "tools.poll", poll);
    insert_rpc_schema(
        &mut schemas,
        &mut covered_rpc_methods,
        "tools.cancel",
        closed_schema::<ToolsCancelResult>(),
    );
    insert_rpc_schema(
        &mut schemas,
        &mut covered_rpc_methods,
        "tools.stream.subscribe",
        closed_schema::<ToolsStreamSubscribeResult>(),
    );

    let mut action_result = closed_schema::<car_ir::ActionResult>();
    require_fields(&mut action_result, ACTION_RESULT_REQUIRED);
    schemas.insert("type.action_result".to_string(), action_result);

    let all_rpc_methods = rpc_method_inventory();
    let uncovered_rpc_methods: Vec<&str> = all_rpc_methods
        .difference(&covered_rpc_methods)
        .copied()
        .collect();
    let covered_event_kinds: BTreeSet<String> = BTreeSet::new();
    let uncovered_event_kinds: Vec<&str> = all_event_kinds
        .difference(&covered_event_kinds)
        .map(String::as_str)
        .collect();
    let complete = uncovered_rpc_methods.is_empty() && uncovered_event_kinds.is_empty();

    json!({
        "format": FORMAT,
        "json_schema_draft": "http://json-schema.org/draft-07/schema#",
        "digest": {
            "algorithm": DIGEST_ALGORITHM,
            "scope": "exact UTF-8 bytes of docs/wire-schema.json"
        },
        "coverage": {
            "complete": complete,
            "covered": schemas.keys().collect::<Vec<_>>(),
            "rpc_results": {
                "total": all_rpc_methods.len(),
                "covered": covered_rpc_methods,
                "uncovered": uncovered_rpc_methods
            },
            "journal_event_payloads": {
                "total": all_event_kinds.len(),
                "covered": covered_event_kinds,
                "uncovered": uncovered_event_kinds
            },
            "limitations": [
                "journal.event covers the exact envelope and closed EventKind enum; coverage.journal_event_payloads.uncovered names every kind whose Event.data remains an open JSON object",
                "coverage.rpc_results.uncovered is derived from the daemon dispatch inventory and names every result that still emits inline or otherwise lacks a schema from its real Rust type",
                "the release version is reported at serve time in the server.schema result and is deliberately absent from these digested bytes, so the digest tracks wire shape alone"
            ],
            "follow_up": FOLLOW_UP_BEAD
        },
        "schemas": schemas
    })
}

/// The always-emitted field sets. Each is asserted against a real minimal value
/// by `tests::required_lists_exactly_the_fields_a_minimal_value_emits`.
const MANAGED_AGENT_LIST_ROW_REQUIRED: &[&str] = &[
    "id",
    "name",
    "command",
    "args",
    "cwd",
    "env",
    "restart",
    "max_restarts",
    "backoff_secs",
    "auto_start",
    "capabilities",
    "status",
    "pid",
    "last_exit_code",
    "restart_count",
    "started_at",
    "attached",
    "manifest_path",
    "log_path",
    "stderr_log_path",
];
const DECLARATIVE_AGENT_ROW_REQUIRED: &[&str] = &[
    "id",
    "name",
    "kind",
    "enabled",
    "capabilities",
    "description",
    "tools",
    "goal",
    "scenarios",
];
const DECLARATIVE_GOAL_REQUIRED: &[&str] = &["check", "max_iterations"];
const EVENT_REQUIRED: &[&str] = &["kind", "data", "timestamp"];
const INFERENCE_RESULT_REQUIRED: &[&str] = &[
    "text",
    "tool_calls",
    "trace_id",
    "model_used",
    "requested_model_id",
    "resolved_model_id",
    "row_digest",
    "catalog_revision",
    "latency_ms",
    "time_to_first_token_ms",
    "usage",
    "stop_reason",
];
const TOKEN_USAGE_REQUIRED: &[&str] = &[
    "prompt_tokens",
    "completion_tokens",
    "total_tokens",
    "context_window",
    "cache_read_input_tokens",
    "cache_creation_input_tokens",
];
const TOOL_CALL_REQUIRED: &[&str] = &["name", "arguments"];
const THINKING_BLOCK_REQUIRED: &[&str] = &["text"];
const BOUNDING_BOX_REQUIRED: &[&str] = &["x1", "y1", "x2", "y2"];
const FALLBACK_FROM_REQUIRED: &[&str] = &["candidate", "reason"];
const TOOL_SCHEMA_REQUIRED: &[&str] =
    &["name", "source", "description", "parameters", "idempotent"];
const TOOL_POLL_RESULT_REQUIRED: &[&str] = &["handle", "tool", "action_id", "status", "chunks"];
const ACTION_RESULT_REQUIRED: &[&str] = &["action_id", "status", "state_changes", "timestamp"];
const QUANTIZATION_OBJECT_REQUIRED: &[&str] = &["scheme", "label"];

/// Pretty, LF-terminated bytes committed and published with each release.
pub fn rendered_document() -> Vec<u8> {
    let mut bytes = serde_json::to_vec_pretty(&document()).expect("wire schema serializes");
    bytes.push(b'\n');
    bytes
}

pub fn sha256_hex(bytes: &[u8]) -> String {
    format!("{:x}", Sha256::digest(bytes))
}

pub fn rendered_digest_file(schema_bytes: &[u8]) -> Vec<u8> {
    format!("{}  wire-schema.json\n", sha256_hex(schema_bytes)).into_bytes()
}

fn committed_digest() -> Result<&'static str, String> {
    let mut fields = COMMITTED_DIGEST.split_whitespace();
    let digest = fields
        .next()
        .ok_or("committed wire schema digest is empty")?;
    let filename = fields
        .next()
        .ok_or("committed wire schema digest omits its filename")?;
    if fields.next().is_some() || filename != "wire-schema.json" {
        return Err("committed wire schema digest must be '<sha256>  wire-schema.json'".into());
    }
    if digest.len() != 64
        || !digest
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    {
        return Err("committed wire schema digest is not lowercase SHA-256".into());
    }
    Ok(digest)
}

/// Return the exact schema document embedded in this daemon release, the
/// SHA-256 digest published beside it, and the release this binary is.
pub fn committed_payload() -> Result<Value, String> {
    let digest = committed_digest()?;
    let actual = sha256_hex(COMMITTED_SCHEMA.as_bytes());
    if actual != digest {
        return Err(format!(
            "embedded wire schema digest mismatch: expected {digest}, got {actual}"
        ));
    }
    let schema: Value = serde_json::from_str(COMMITTED_SCHEMA)
        .map_err(|error| format!("embedded wire schema is invalid JSON: {error}"))?;
    serde_json::to_value(ServerSchemaResult {
        schema,
        digest: digest.to_string(),
        digest_algorithm: DigestAlgorithm::Sha256,
        car_version: env!("CARGO_PKG_VERSION").to_string(),
    })
    .map_err(|error| format!("server.schema payload does not serialize: {error}"))
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;

    #[test]
    fn repeated_generation_is_byte_identical() {
        assert_eq!(rendered_document(), rendered_document());
        assert_eq!(
            rendered_digest_file(&rendered_document()),
            rendered_digest_file(&rendered_document())
        );
    }

    /// A version bump must not touch the digested bytes.
    ///
    /// `scripts/release.sh` rewrites the workspace version and commits without
    /// regenerating this pair. If the version reached these bytes, that commit
    /// would turn the required `test` check red and, forced past, would publish
    /// a document naming the previous release. The only compile-time source of
    /// the version is `CARGO_PKG_VERSION`, so proving the rendered bytes do not
    /// contain it proves a bump cannot change the digest.
    #[test]
    fn a_version_bump_alone_cannot_change_the_digest() {
        let document = document();
        assert!(
            document.get("car_version").is_none(),
            "the digested document must carry no release version"
        );
        let rendered = String::from_utf8(rendered_document()).expect("schema is UTF-8");
        assert!(
            !rendered.contains(env!("CARGO_PKG_VERSION")),
            "the crate version leaked into the digested bytes, so every release bump \
             would change the digest and fail freshness on the bump commit"
        );
    }

    /// The release version is still reachable — at serve time, from the binary.
    #[test]
    fn server_schema_reports_the_running_release() {
        let payload = committed_payload().expect("committed payload");
        assert_eq!(payload["car_version"], env!("CARGO_PKG_VERSION"));
    }

    #[test]
    fn committed_artifacts_match_generation_and_rpc_payload() {
        let generated = rendered_document();
        assert_eq!(COMMITTED_SCHEMA.as_bytes(), generated);
        assert_eq!(
            COMMITTED_DIGEST.as_bytes(),
            rendered_digest_file(&generated)
        );

        let payload = committed_payload().expect("committed payload");
        assert_eq!(
            crate::handler::handle_server_schema().expect("server.schema RPC payload"),
            payload
        );
        assert_eq!(payload["schema"], document());
        assert_eq!(payload["digest"], sha256_hex(&generated));
        assert_eq!(payload["digest_algorithm"], DIGEST_ALGORITHM);
    }

    fn managed_agent() -> car_registry::supervisor::ManagedAgent {
        car_registry::supervisor::ManagedAgent {
            spec: car_registry::supervisor::AgentSpec {
                id: "trader".into(),
                name: "Trader".into(),
                command: "/usr/local/bin/node".into(),
                args: vec!["index.js".into()],
                cwd: None,
                env: BTreeMap::new(),
                restart: car_registry::supervisor::RestartPolicy::OnFailure,
                max_restarts: 10,
                backoff_secs: 5,
                auto_start: false,
                token: "secret".into(),
                method_allowlist: None,
                capabilities: Vec::new(),
            },
            status: car_registry::supervisor::AgentStatus::Stopped,
            pid: None,
            last_exit_code: None,
            restart_count: 0,
            started_at: None,
            blocked_by_pid: None,
        }
    }

    fn declarative_spec() -> car_registry::declarative::DeclarativeAgentSpec {
        car_registry::declarative::DeclarativeAgentSpec {
            id: "newsroom".into(),
            name: "Newsroom".into(),
            identity: "You watch the journal.".into(),
            tools: vec!["fs.read".into()],
            denied_tools: Vec::new(),
            standing_goal: String::new(),
            goal: None,
            cadence: None,
            scenarios: Vec::new(),
            builder_draft: None,
            previous: None,
            enabled: true,
            context: Default::default(),
        }
    }

    fn model_schema() -> car_inference::schema::ModelSchema {
        // Every `Option` is `None` on purpose: that is what makes this value a
        // detector for a newly added `skip_serializing_if` anywhere in the
        // catalog subtree.
        car_inference::schema::ModelSchema {
            id: "qwen/qwen3-4b".into(),
            name: "Qwen3 4B".into(),
            provider: "qwen".into(),
            family: "qwen3".into(),
            version: String::new(),
            capabilities: vec![car_inference::schema::ModelCapability::Generate],
            context_length: 32_768,
            max_output_tokens: None,
            param_count: String::new(),
            quantization: None,
            performance: Default::default(),
            cost: Default::default(),
            source: car_inference::schema::ModelSource::Ollama {
                model_tag: "qwen3:4b".into(),
                host: "http://localhost:11434".into(),
            },
            tags: Vec::new(),
            supported_params: Vec::new(),
            public_benchmarks: Vec::new(),
            trust_tier: Default::default(),
            deprecated: false,
            available: false,
            weights_ready: false,
        }
    }

    fn inference_result() -> car_inference::InferenceResult {
        car_inference::InferenceResult {
            text: "hello".into(),
            tool_calls: Vec::new(),
            bounding_boxes: Vec::new(),
            trace_id: "trace-1".into(),
            model_used: "qwen/qwen3-4b".into(),
            model_identity: Default::default(),
            latency_ms: 12,
            time_to_first_token_ms: None,
            usage: None,
            provider_output_items: Vec::new(),
            thinking: Vec::new(),
            stop_reason: None,
            auth_fallback_from: None,
            local_last_resort: false,
            fallback_from: Vec::new(),
        }
    }

    fn validate(document: &Value, key: &str, emitted: &Value) {
        let schema = &document["schemas"][key];
        let validator =
            jsonschema::validator_for(schema).unwrap_or_else(|e| panic!("{key} compiles: {e}"));
        if let Err(error) = validator.validate(emitted) {
            panic!(
                "the value {key} actually emits does not satisfy its generated schema: {error}\n\
                 emitted: {}",
                serde_json::to_string_pretty(emitted).unwrap()
            );
        }
    }

    /// Every covered schema is checked against a value built the way the daemon
    /// builds it, through the same type.
    ///
    /// This is the guard that a hand-transcribed schema cannot pass: with
    /// `additionalProperties: false` everywhere, a field the emitter gains and
    /// the schema does not fails here, and a field the schema requires and the
    /// emitter drops fails here too.
    #[test]
    fn every_covered_schema_validates_its_real_emitted_value() {
        let document = document();
        let mut checked: Vec<&str> = Vec::new();

        let mut check = |key: &'static str, emitted: Value| {
            validate(&document, key, &emitted);
            checked.push(key);
        };

        check(
            "cli.car_inspect.result",
            serde_json::to_value(CarInspectResult::Managed(managed_list_row())).unwrap(),
        );
        check(
            "journal.event",
            serde_json::to_value(minimal_event()).unwrap(),
        );
        check(
            "rpc.capabilities.list.result",
            serde_json::to_value(capabilities_list_result()).unwrap(),
        );
        check(
            "rpc.infer.result",
            serde_json::to_value(inference_result()).unwrap(),
        );
        check(
            "rpc.models.catalog_snapshot.result",
            serde_json::to_value(
                car_inference::catalog_identity::CatalogSnapshot::new([model_schema()])
                    .expect("catalog snapshot"),
            )
            .unwrap(),
        );
        check(
            "rpc.server.handshake.result",
            serde_json::to_value(handshake_result()).unwrap(),
        );
        check(
            "rpc.server.schema.result",
            committed_payload().expect("committed payload"),
        );
        check(
            "rpc.state.get.result",
            json!({"arbitrary": [true, null, 3]}),
        );
        check(
            "rpc.state.set.result",
            serde_json::to_value(StateSetResult::Ok).unwrap(),
        );
        check(
            "rpc.state.exists.result",
            serde_json::to_value(StateExistsResult(false)).unwrap(),
        );
        check(
            "rpc.state.keys.result",
            serde_json::to_value(StateKeysResult(vec!["ready".into()])).unwrap(),
        );
        check(
            "rpc.state.snapshot.result",
            serde_json::to_value(StateSnapshotResult(serde_json::Map::from_iter([(
                "ready".into(),
                Value::Bool(true),
            )])))
            .unwrap(),
        );
        check(
            "rpc.tools.register.result",
            serde_json::to_value(ToolsRegisterResult(1)).unwrap(),
        );
        check(
            "rpc.tools.list.result",
            serde_json::to_value(ToolsListResult {
                tools: vec![minimal_tool_schema()],
                count: 1,
            })
            .unwrap(),
        );
        check(
            "rpc.tools.unregister.result",
            serde_json::to_value(ToolsUnregisterResult {
                unregistered: "fs.read".into(),
                removed: 1,
            })
            .unwrap(),
        );
        check(
            "rpc.tools.poll.result",
            serde_json::to_value(Some(minimal_tool_poll_result())).unwrap(),
        );
        check(
            "rpc.tools.cancel.result",
            serde_json::to_value(ToolsCancelResult { cancelled: true }).unwrap(),
        );
        check(
            "rpc.tools.stream.subscribe.result",
            serde_json::to_value(ToolsStreamSubscribeResult { subscribed: true }).unwrap(),
        );
        check(
            "type.action_result",
            serde_json::to_value(minimal_action_result()).unwrap(),
        );

        // A new entry in `coverage.covered` without a case above would leave
        // that surface unproven, so the two sets must match exactly.
        let covered: Vec<String> = document["coverage"]["covered"]
            .as_array()
            .expect("covered list")
            .iter()
            .map(|value| value.as_str().expect("covered key").to_string())
            .collect();
        let mut checked: Vec<String> = checked.into_iter().map(str::to_string).collect();
        checked.sort();
        assert_eq!(
            covered, checked,
            "every covered schema needs a real emitted value checked against it"
        );
    }

    /// The declarative row is the second arm of `cli.car_inspect.result`.
    #[test]
    fn declarative_rows_validate_against_the_inspect_schema() {
        let document = document();
        let spec = declarative_spec();
        validate(
            &document,
            "cli.car_inspect.result",
            &crate::coder::rpc::declarative_row(&spec),
        );

        let mut with_goal = spec;
        with_goal.goal = Some(car_registry::declarative::DeclarativeGoal {
            check: "cargo test".into(),
            max_iterations: 8,
        });
        validate(
            &document,
            "cli.car_inspect.result",
            &crate::coder::rpc::declarative_row(&with_goal),
        );
    }

    /// Asserts the precondition [`require_all_declared_fields`] depends on.
    ///
    /// The blanket pass marks every declared catalog property required, which
    /// is a lie the moment a catalog type gains a `skip_serializing_if`. The
    /// snapshot below has every `Option` set to `None`, so the first such field
    /// disappears from the serialized value and fails this validation.
    #[test]
    fn catalog_snapshot_declares_no_conditionally_emitted_field() {
        let document = document();
        let snapshot = car_inference::catalog_identity::CatalogSnapshot::new([model_schema()])
            .expect("catalog snapshot");
        validate(
            &document,
            "rpc.models.catalog_snapshot.result",
            &serde_json::to_value(&snapshot).unwrap(),
        );
    }

    #[test]
    fn coverage_inventory_accounts_for_every_daemon_result_and_event_kind() {
        let document = document();
        let coverage = &document["coverage"];

        let strings = |value: &Value| -> BTreeSet<String> {
            value
                .as_array()
                .expect("coverage list")
                .iter()
                .map(|value| value.as_str().expect("coverage name").to_string())
                .collect()
        };

        let all_rpc: BTreeSet<String> = rpc_method_inventory()
            .into_iter()
            .map(str::to_string)
            .collect();
        let covered_rpc = strings(&coverage["rpc_results"]["covered"]);
        let uncovered_rpc = strings(&coverage["rpc_results"]["uncovered"]);
        assert!(covered_rpc.is_disjoint(&uncovered_rpc));
        assert_eq!(
            all_rpc,
            covered_rpc.union(&uncovered_rpc).cloned().collect(),
            "the source-derived daemon method inventory must be partitioned exactly"
        );
        assert_eq!(coverage["rpc_results"]["total"], Value::from(all_rpc.len()));
        for method in &covered_rpc {
            assert!(
                document["schemas"]
                    .get(format!("rpc.{method}.result"))
                    .is_some(),
                "covered RPC method has no result schema: {method}"
            );
        }

        let event_schema = &document["schemas"]["journal.event"];
        let all_events = event_kind_inventory(event_schema);
        let covered_events = strings(&coverage["journal_event_payloads"]["covered"]);
        let uncovered_events = strings(&coverage["journal_event_payloads"]["uncovered"]);
        assert!(covered_events.is_disjoint(&uncovered_events));
        assert_eq!(
            all_events,
            covered_events.union(&uncovered_events).cloned().collect(),
            "the source-derived EventKind inventory must be partitioned exactly"
        );
        assert_eq!(
            coverage["journal_event_payloads"]["total"],
            Value::from(all_events.len())
        );

        assert_eq!(
            coverage["complete"],
            Value::Bool(uncovered_rpc.is_empty() && uncovered_events.is_empty()),
            "coverage.complete may be true only when both source inventories are exhausted"
        );
    }

    #[test]
    fn priority_enums_are_closed_and_objects_reject_unknown_fields() {
        let document = document();
        let event_kind = &document["schemas"]["journal.event"]["definitions"]["EventKind"];
        let variants = event_kind["oneOf"]
            .as_array()
            .expect("documented EventKind variants");
        assert!(!variants.is_empty());
        assert!(variants.iter().all(|variant| variant["enum"].is_array()));
        assert_eq!(
            document["schemas"]["type.action_result"]["additionalProperties"],
            false
        );
        assert_eq!(
            document["schemas"]["rpc.tools.list.result"]["additionalProperties"],
            false
        );

        let validator = jsonschema::validator_for(&document["schemas"]["type.action_result"])
            .expect("ActionResult schema compiles");
        let valid = json!({
            "action_id": "action-1",
            "status": "succeeded",
            "state_changes": {},
            "timestamp": "2026-09-15T00:00:00Z"
        });
        assert!(validator.is_valid(&valid));
        let mut unknown = valid;
        unknown["unexpected"] = Value::Bool(true);
        assert!(!validator.is_valid(&unknown));
    }

    fn managed_list_row() -> ManagedAgentListRow {
        ManagedAgentListRow {
            agent: ManagedAgentWire::from_managed(&managed_agent()),
            attached: false,
            tools: None,
            manifest_path: "/tmp/agents/trader/manifest.toml".into(),
            log_path: "/tmp/logs/trader.stdout.log".into(),
            stderr_log_path: "/tmp/logs/trader.stderr.log".into(),
            session_id: None,
        }
    }

    fn capabilities_list_result() -> CapabilitiesListResult {
        CapabilitiesListResult {
            caller_role: CapabilityRole::Operator,
            count: 1,
            methods: vec![CapabilityMethodRow {
                method: "capabilities.list".into(),
                role: CapabilityRole::Operator,
            }],
        }
    }

    fn minimal_tool_schema() -> car_ir::ToolSchema {
        car_ir::ToolSchema {
            name: "fs.read".into(),
            source: car_ir::ToolSourceKind::Builtin,
            description: "Read a file".into(),
            parameters: json!({"type": "object"}),
            returns: None,
            idempotent: true,
            cache_ttl_secs: None,
            rate_limit: None,
        }
    }

    fn minimal_tool_poll_result() -> car_engine::tool_handles::ToolPollResult {
        car_engine::tool_handles::ToolPollResult {
            handle: "tool-1".into(),
            tool: "fs.read".into(),
            action_id: "action-1".into(),
            status: car_ir::ToolStatus::Running,
            chunks: Vec::new(),
            dropped_chunks: 0,
            result: None,
            error: None,
        }
    }

    fn minimal_event() -> car_eventlog::Event {
        car_eventlog::Event {
            kind: car_eventlog::EventKind::ActionSucceeded,
            run_id: None,
            client_id: None,
            policy_session_id: None,
            action_id: None,
            proposal_id: None,
            data: HashMap::new(),
            timestamp: chrono::Utc::now(),
            prev_hash: None,
            hash: None,
        }
    }

    fn minimal_action_result() -> car_ir::ActionResult {
        car_ir::ActionResult {
            action_id: "action-1".into(),
            status: car_ir::ActionStatus::Succeeded,
            output: None,
            error: None,
            terminal: false,
            rolled_back: false,
            state_changes: HashMap::new(),
            duration_ms: None,
            timestamp: chrono::Utc::now(),
        }
    }

    fn handshake_result() -> ServerHandshakeResult {
        ServerHandshakeResult {
            protocol_version: car_proto::PROTOCOL_VERSION,
            server_version: env!("CARGO_PKG_VERSION").to_string(),
            client_protocol_version: u64::from(car_proto::PROTOCOL_VERSION),
            client_version: "unknown".into(),
            negotiated_capabilities: Vec::new(),
            assistant_name: "Parslee".into(),
            assistant_aliases: vec!["parslee".into()],
            assistant_brand: car_identity::BRAND_NAME.to_string(),
        }
    }

    /// Find the generated schema object for a derived type by name.
    ///
    /// Mirrors `require_fields_for`'s lookup, and for the same reason: a
    /// hoisted definition is keyed by the type name, an inlined or root copy
    /// carries a `title`, and which one you get moves with the field's
    /// attributes.
    fn by_title<'a>(schema: &'a Value, title: &str) -> &'a Value {
        fn walk<'a>(
            value: &'a Value,
            title: &str,
            own_name: Option<&str>,
            entries_are_definitions: bool,
        ) -> Option<&'a Value> {
            match value {
                Value::Array(values) => values
                    .iter()
                    .find_map(|value| walk(value, title, None, false)),
                Value::Object(map) => {
                    let named = own_name == Some(title)
                        || map.get("title").and_then(Value::as_str) == Some(title);
                    if named && map.contains_key("properties") {
                        return Some(value);
                    }
                    map.iter().find_map(|(key, child)| {
                        let child_name = entries_are_definitions.then_some(key.as_str());
                        let child_defines = !entries_are_definitions && key == "definitions";
                        walk(child, title, child_name, child_defines)
                    })
                }
                _ => None,
            }
        }
        walk(schema, title, None, false)
            .unwrap_or_else(|| panic!("no generated object named {title}"))
    }

    /// The contract every hand-written `required` list must satisfy.
    ///
    /// `sample` must be MINIMAL: every `Option` `None`, every collection empty,
    /// every `skip_serializing_if` predicate true. Its key set is then exactly
    /// the set of fields CAR always emits, which is exactly what `required`
    /// must name. Comparing the two catches both a list that over-claims (a
    /// field the emitter can omit) and one that under-claims (a guarantee the
    /// consumer is not given).
    fn assert_required_matches_emitted(node: &Value, sample: &Value, label: &str) {
        let required: std::collections::BTreeSet<&str> = node["required"]
            .as_array()
            .unwrap_or_else(|| panic!("{label} declares no required list"))
            .iter()
            .map(|value| value.as_str().expect("required entries are strings"))
            .collect();
        let emitted: std::collections::BTreeSet<&str> = sample
            .as_object()
            .unwrap_or_else(|| panic!("{label} sample is not an object"))
            .keys()
            .map(String::as_str)
            .collect();
        assert_eq!(
            required, emitted,
            "{label}: `required` must name exactly the fields a minimal value emits"
        );
    }

    #[test]
    fn required_lists_exactly_the_fields_a_minimal_value_emits() {
        let document = document();
        fn value<T: Serialize>(item: T) -> Value {
            serde_json::to_value(item).expect("wire value serializes")
        }

        let inspect = &document["schemas"]["cli.car_inspect.result"];
        assert_required_matches_emitted(
            by_title(inspect, "ManagedAgentListRow"),
            &value(managed_list_row()),
            "ManagedAgentListRow",
        );
        assert_required_matches_emitted(
            by_title(inspect, "DeclarativeAgentRow"),
            &crate::coder::rpc::declarative_row(&declarative_spec()),
            "DeclarativeAgentRow",
        );
        assert_required_matches_emitted(
            by_title(inspect, "DeclarativeGoal"),
            &value(car_registry::declarative::DeclarativeGoal {
                check: "cargo test".into(),
                max_iterations: 8,
            }),
            "DeclarativeGoal",
        );

        let event = &document["schemas"]["journal.event"];
        assert_required_matches_emitted(event, &value(minimal_event()), "Event");

        let capabilities = &document["schemas"]["rpc.capabilities.list.result"];
        assert_required_matches_emitted(
            capabilities,
            &value(capabilities_list_result()),
            "CapabilitiesListResult",
        );
        assert_required_matches_emitted(
            by_title(capabilities, "CapabilityMethodRow"),
            &value(CapabilityMethodRow {
                method: "capabilities.list".into(),
                role: CapabilityRole::Operator,
            }),
            "CapabilityMethodRow",
        );

        let inference = &document["schemas"]["rpc.infer.result"];
        assert_required_matches_emitted(inference, &value(inference_result()), "InferenceResult");
        assert_required_matches_emitted(
            by_title(inference, "TokenUsage"),
            &value(car_inference::TokenUsage::default()),
            "TokenUsage",
        );
        assert_required_matches_emitted(
            by_title(inference, "ToolCall"),
            &value(car_inference::tasks::generate::ToolCall {
                id: None,
                name: "fs.read".into(),
                arguments: HashMap::new(),
            }),
            "ToolCall",
        );
        assert_required_matches_emitted(
            by_title(inference, "ThinkingBlock"),
            &value(car_inference::tasks::generate::ThinkingBlock::default()),
            "ThinkingBlock",
        );
        assert_required_matches_emitted(
            by_title(inference, "BoundingBox"),
            &value(car_inference::tasks::grounding::BoundingBox {
                x1: 0,
                y1: 0,
                x2: 1,
                y2: 1,
                label: String::new(),
                confidence: None,
            }),
            "BoundingBox",
        );
        assert_required_matches_emitted(
            by_title(inference, "FallbackFrom"),
            &value(car_inference::FallbackFrom {
                candidate: "gpt-5".into(),
                reason: car_inference::FallbackReason::Failed,
            }),
            "FallbackFrom",
        );

        let tools = &document["schemas"]["rpc.tools.list.result"];
        assert_required_matches_emitted(
            by_title(tools, "ToolSchema"),
            &value(minimal_tool_schema()),
            "ToolSchema",
        );
        let tool_poll = &document["schemas"]["rpc.tools.poll.result"];
        assert_required_matches_emitted(
            by_title(tool_poll, "ToolPollResult"),
            &value(minimal_tool_poll_result()),
            "ToolPollResult",
        );

        let action_result = &document["schemas"]["type.action_result"];
        assert_required_matches_emitted(
            action_result,
            &value(minimal_action_result()),
            "ActionResult",
        );

        let handshake = &document["schemas"]["rpc.server.handshake.result"];
        assert_required_matches_emitted(
            handshake,
            &value(handshake_result()),
            "ServerHandshakeResult",
        );

        let schema_result = &document["schemas"]["rpc.server.schema.result"];
        assert_required_matches_emitted(
            schema_result,
            &committed_payload().expect("committed payload"),
            "ServerSchemaResult",
        );
    }

    /// The one catalog member the blanket required pass must not close over.
    #[test]
    fn the_structured_quantization_form_omits_the_fields_its_label_carries() {
        let document = document();
        let catalog = &document["schemas"]["rpc.models.catalog_snapshot.result"];
        let quantization = car_inference::schema::Quantization {
            bits: None,
            scheme: car_inference::schema::QuantScheme::AffineGroupInt,
            group_size: None,
            label: "an-unparseable-label".into(),
        };
        let emitted = serde_json::to_value(&quantization).expect("quantization serializes");
        assert!(
            emitted.is_object(),
            "this label must take the structured form for the assertion to mean anything"
        );
        assert_required_matches_emitted(
            by_title(catalog, "QuantizationObjectWireSchema"),
            &emitted,
            "QuantizationObjectWireSchema",
        );
    }

    /// The token that authenticates an agent must never reach the wire.
    #[test]
    fn the_managed_agent_projection_drops_the_agent_token() {
        let wire = serde_json::to_value(ManagedAgentWire::from_managed(&managed_agent())).unwrap();
        assert!(wire.get("token").is_none());
        assert!(!document()["schemas"]["cli.car_inspect.result"]
            .to_string()
            .contains("\"token\""));
    }
}