vta-service 0.23.2

//! The schema-conformance sweep (#857): every task this dispatcher binds to a
//! **published** canonical URI must carry that URI's wire shape.
//!
//! ## Why this exists
//!
//! Two parity checks already run — every SDK constant is dispatched
//! (routing), and every dispatched `spec/` URI exists in the registry
//! (coverage, `vtc-service/tests/trust_task_manifest.rs`). Nothing checked
//! **correctness**: that a task bound to a published URI actually speaks that
//! URI's schema. Folding a task onto a *published* URI silently turns
//! `validate_payload` on at the dispatch spine, and a non-canonical body then
//! fails on live traffic with no compile-time signal — which is exactly how
//! `acl/update` shipped broken (#856, via #842).
//!
//! ## How it works
//!
//! The URI census is **derived, not hand-maintained**: `dispatched_uris()`
//! (generated by `dispatch_table!` from the same declarations as the match
//! arms) filtered to what `trust_tasks_rs::schema_index::schema_for`
//! resolves. Every URI in that set MUST have an entry in [`table`] — a
//! [`Witness`] carrying a representative request and response, or an
//! annotated [`Conformance::KnownDrift`] debt entry. The witness's request
//! must parse as the generated `specs::…::Payload` and the response as
//! `specs::…::Response`; the generated types carry `deny_unknown_fields`
//! plus the spec's required set.
//!
//! The request is then validated against its embedded
//! `payload.schema.json` — the same check `validate_payload` runs on the
//! dispatch spine. Parsing alone is strictly weaker and let a whole class of
//! defect through: serde accepts `null` into an `Option<T>` field, JSON
//! Schema types that member `"string"` and rejects it. `keys/create/0.1`
//! shipped a body that serialized every unset member as `null`, its witness
//! parsed green, and live key creation over DIDComm/TSP failed on arrival.
//! See [`validates`].
//!
//! **Non-vacuity is enforced mechanically.** The `acl/update` defect
//! survived three PRs because an assertion of the form "adding a forbidden
//! member makes this fail" passed while the body was *already* failing for
//! an unrelated reason. Here every witness is drifted (an unknown member is
//! injected) after the good case passes, and the drifted form MUST be
//! rejected — a schema that accepts anything cannot produce a green row.
//!
//! ## Typed bodies, or raw JSON
//!
//! Witnesses are written **typed when the spec preceded the type, raw JSON
//! when it did not — and the second case is a debt**, not a style choice.
//!
//! A typed witness is the stronger form: it proves *our types* conform,
//! rather than that someone can hand-write a body the schema accepts. But it
//! only proves that when the type did not derive its shape from the same
//! misreading that produced the body. If it did, the witness asserts the code
//! agrees with itself and passes green over the defect.
//!
//! That is not hypothetical. When the `vta/contexts/*` and `vta/webvh/*`
//! schemas arrived (#1015), two of the producer types were already wrong —
//! `create_did_webvh` sent `context_id` where the schema names `contextId`,
//! and `update_context` dropped `contextPolicy` on the floor. Witnesses built
//! from those types would have encoded both defects and reported conformance.
//! So the 22 went in as raw JSON, against the schema as the authority.
//!
//! It is the same principle `scripts/check-bindings-conformance.mjs` applies
//! one layer down in `dtgwg-trust-tasks-tf`, where the binding rules are
//! deliberately re-implemented rather than imported from the generator: a
//! check that shares a source with the thing it checks has stopped being a
//! check. Its header makes the trade explicit — a re-implementation can be
//! wrong, but a wrong one disagrees loudly, where a shared one agrees
//! silently.
//!
//! Which makes the retro-fit a real step rather than tidying. Converting those
//! 22 to typed bodies turns "the schema accepts a shape someone wrote" into
//! "our types conform" — and it is sound only now that #1015 has corrected the
//! types. Done earlier it would have laundered the bugs into a green row.
//!
//! ## When this fails after a rebase
//!
//! Sibling consolidation streams are rebinding URIs. A newly bound published
//! URI fails the coverage assertion here **by design** — add a witness (or,
//! with a stated reason, a debt entry) for it. That is this check doing its
//! job, not an obstacle to route around.
//!
//! It has already happened once, and the shape is worth keeping: #864 folded
//! the `did-templates` family onto new URIs, the census dropped the old ones,
//! and the *stale-witness* direction of the coverage assertion fired. Both
//! directions matter — a witness for a URI nothing dispatches any more is as
//! much drift as a dispatched URI with no witness, and only the first tells
//! you a fold landed.
//!
//! And then the other direction fired for the same family: the
//! `trust-tasks-rs` 0.2.43 → 0.2.51 bump indexed the six 2.0 specs, the
//! derived census picked them straight back up, and the missing-witness
//! assertion asked for all six. Neither half was hand-listed — a fold
//! upstream and a dependency bump each moved the census on their own.

use serde::de::DeserializeOwned;
use serde_json::{Value, json};
use std::collections::{BTreeSet, HashMap};

use trust_tasks_rs::specs;
use trust_tasks_rs::validate::ValidatedPayload;
use vta_sdk::trust_tasks as uris;

use vta_sdk::acl::ContextDirection;
use vta_sdk::keys::{KeyOrigin, KeyRecord, KeyStatus, KeyType};
use vta_sdk::protocols::acl_management::change_role::ChangeRoleBody;
use vta_sdk::protocols::acl_management::create::{CreateAclBody, CreateAclResponseBody};
use vta_sdk::protocols::acl_management::delete::{DeleteAclBody, DeleteAclResultBody};
use vta_sdk::protocols::acl_management::entry::{AclEntry, Approve, StepUp};
use vta_sdk::protocols::acl_management::get::{GetAclBody, GetAclResultBody};
use vta_sdk::protocols::acl_management::list::{ListAclBody, ListAclResultBody};
use vta_sdk::protocols::acl_management::swap::{SwapKeyBody, SwapKeyResultBody};
use vta_sdk::protocols::acl_management::update::UpdateAclBody;
use vta_sdk::protocols::app_state::{
    AppStateDeleteBody, AppStateDeleteResponse, AppStateGetBody, AppStateGetManyBody,
    AppStateGetManyResponse, AppStateGetResponse, AppStateListBody, AppStateListResponse,
    AppStatePutBody, AppStatePutManyBody, AppStatePutManyResponse, AppStatePutResponse,
    AppStateRecord, AppStateWrite, PutManyMode, WriteOutcome, WriteResult,
};
use vta_sdk::protocols::audit_management::list::{
    AuditEnvelope, ListAuditLogsBody, ListAuditLogsResultBody,
};
use vta_sdk::protocols::auth::{RevokeSessionRequest, RevokeSessionResponse};
use vta_sdk::protocols::consent_management::{
    ConsentApproverListBody, ConsentApproverSetBody, ConsentDecisionBody, ConsentListBody,
    ConsentRequestBody, ConsentRevokeBody,
};
use vta_sdk::protocols::credential_exchange::{
    self as credx, PendingApproveBody, PendingDenyBody, PendingDenyResponse, PendingListResponse,
    PendingPresentationSummary, PresentBody, RequestedCredentialSummary,
};
use vta_sdk::protocols::credentials_issuance::{
    IssueCredentialBody, IssueCredentialResponse, RevokeCredentialBody, RevokeCredentialResponse,
};
use vta_sdk::protocols::device_management::{
    DeviceDisableBody, DeviceHeartbeatBody, DeviceRegisterBody, DeviceSetWakeBody, DeviceWipeBody,
    WakeHandle,
};
use vta_sdk::protocols::did_management::servers::{
    AgentOnlyDid, HostOnlyDid, ListWebvhServerDomainsBody, ListWebvhServerDomainsResultBody,
    ReconcileWebvhServerDidsBody, ReconcileWebvhServerDidsResultBody, RetireOrphanSlotBody,
    RetireOrphanSlotResultBody, WebvhServerDomainEntry,
};
use vta_sdk::protocols::key_management::create::{
    CreateKeyBody, CreateKeyResponseBody, CreateKeyResultBody,
};
use vta_sdk::protocols::key_management::derive_and_sign::{
    DeriveAndSignBody, DeriveAndSignResultBody,
};
use vta_sdk::protocols::key_management::derive_and_sign_document::{
    DeriveAndSignDocumentBody, DeriveAndSignDocumentResultBody,
};
use vta_sdk::protocols::key_management::get::{GetKeyBody, GetKeyResponseBody};
use vta_sdk::protocols::key_management::import::ImportKeyBody;
use vta_sdk::protocols::key_management::list::{ListKeysBody, ListKeysResultBody};
use vta_sdk::protocols::key_management::rename::{RenameKeyBody, RenameKeyResultBody};
use vta_sdk::protocols::key_management::revoke::{RevokeKeyBody, RevokeKeyResultBody};
use vta_sdk::protocols::key_management::sign::{SignAlgorithm, SignRequestBody, SignResultBody};
use vta_sdk::protocols::memory::{
    MemoryDeleteBody, MemoryDeleteResponse, MemoryItem, MemoryListBody, MemoryListResponse,
    MemoryPutBody, MemoryPutResponse,
};
use vta_sdk::protocols::policy_management::{
    DeletePolicyBody, DeletePolicyResultBody, GetPolicyBody, GetPolicyResultBody, ListPoliciesBody,
    ListPoliciesResultBody, PolicyModuleView, UpsertPolicyBody, UpsertPolicyResultBody,
};
use vta_sdk::protocols::vault_management::VaultUpsertBody;
use vta_sdk::protocols::vta_management::get_config::{
    ConfigField, GetConfigBody, GetConfigResultBody,
};
use vta_sdk::protocols::vta_management::update_config::{
    RejectedKey, UpdateConfigBody, UpdateConfigResultBody,
};

use super::dispatched_uris;

/// The sweep's scope: dispatched ∩ published. Derived, never hand-listed.
fn resolved_uris() -> BTreeSet<&'static str> {
    dispatched_uris()
        .into_iter()
        .filter(|u| trust_tasks_rs::schema_index::schema_for(u).is_some())
        .collect()
}

type ParseFn = fn(Value) -> Result<(), String>;
type ValidateFn = fn(&Value) -> Result<(), String>;

fn parses<T: DeserializeOwned>(v: Value) -> Result<(), String> {
    serde_json::from_value::<T>(v)
        .map(|_| ())
        .map_err(|e| e.to_string())
}

/// Validate against the embedded `payload.schema.json`, which is what the
/// dispatch spine actually runs (`validate_payload` → `against_schema`).
///
/// This is **not** what [`parses`] checks, and the gap is where a whole class
/// of defect lived. serde reads `null` into an `Option<T>` field happily; JSON
/// Schema types the same member `"string"` and rejects null outright. So a body
/// that serialized every unset member as `null` round-tripped through the
/// generated type — green sweep — and was refused by the first maintainer it
/// reached. `keys/create/0.1` shipped exactly that way: the witness below
/// serializes a real `CreateKeyBody` with `mnemonic: None`, it parsed, and
/// every live key creation over DIDComm/TSP failed with `null is not of type
/// "string"`. Both checks now run on every witness.
fn validates<T: ValidatedPayload>(v: &Value) -> Result<(), String> {
    T::validate_value(v).map_err(|e| e.to_string())
}

/// A representative request/response pair for one published, dispatched URI.
///
/// The request should be the serialization of the type the producer actually
/// sends, so the assertion is about **our wire form** rather than a hand-typed
/// fixture. A transcription only proves someone can type valid JSON: it stops
/// tracking the producer the moment the producer changes, and stays green while
/// live traffic fails. `dids/update` is the worked example — its witness was
/// transcribed, and #895 shipped anyway.
///
/// ## Where a transcription remains, and why
///
/// **14** of these witnesses still build their request with `json!`, and the
/// count is not the useful framing — what remains is three distinct situations,
/// only one of which is unfinished work. (The previous note said 21; it was
/// arithmetic, not measurement. The number above is counted from the table.)
///
/// - **`auth/{whoami,sessions-list,step-up/approve-response}`,
///   `task-consent/decision`** (5) — the VTA is the *consumer*; no `vta-sdk`
///   producer sends these, so there is no emission of ours to assert. A fixture
///   is the honest witness, and **nothing is owed here**.
/// - **`vault/*` (7), `device/list`** (8) — the caller supplies the whole
///   payload as a `Value` and the SDK is a pass-through contributing at most one
///   member. There is no SDK type, and giving them one is an **API decision**,
///   not a test change.
/// - **`messaging/ping`** (1) — a producer exists, but its payload is a single
///   required `nonce`. A body struct would guard nothing today: this defect
///   class lives in *unset optional* members, and ping has none. Worth folding
///   the day it gains a second member, not before.
///
/// The families that did have unset optionals — `keys/*` (#888), `webvh` (#924),
/// `device/*` (#925), `consent/*` (#928) — are folded, and their witnesses are
/// built from the producer's own body rather than transcribed.
struct Witness {
    request: Value,
    parse_request: ParseFn,
    validate_request: ValidateFn,
    response: Value,
    parse_response: ParseFn,
}

enum Conformance {
    Checked(Witness),
    /// Known non-conformance, kept visible and counted instead of silently
    /// tolerated. Every entry is a follow-up issue; the reason must name the
    /// drift precisely enough that the fix needs no re-diagnosis.
    #[allow(dead_code)] // empty today — the variant is the mechanism for future debt
    KnownDrift(&'static str),
}

macro_rules! checked {
    ($p:ty, $r:ty, $req:expr, $resp:expr) => {
        Conformance::Checked(Witness {
            request: $req,
            parse_request: parses::<$p>,
            validate_request: validates::<$p>,
            response: $resp,
            parse_response: parses::<$r>,
        })
    };
}

// ─── Shared fixture builders ─────────────────────────────────────────────

const SUBJECT: &str = "did:key:z6MkSubject";

/// A real `digestMultibase` — base58btc over the sha2-256 multihash
/// (`0x12 0x20` || digest), the form `vta_policy::consent` now emits.
///
/// The witness used to carry the placeholder `"b64u-digest"`, which the 0.4
/// registry rejects (`minLength` 16) — a placeholder that satisfied the old
/// schema by accident and tested nothing about the encoding.
const DIGEST_MULTIBASE: &str = "zQmSK9pGKFnmc77pqyNAPJyPKt8rMqctngfg3vwuMArwGYZ";
const TS: &str = "2026-07-29T00:00:00Z";

fn dt() -> chrono::DateTime<chrono::Utc> {
    chrono::DateTime::parse_from_rfc3339(TS)
        .expect("valid ts")
        .with_timezone(&chrono::Utc)
}

/// A live application-state record, with a version deliberately not 1: versions
/// are values of the *namespace* counter, so a fixture using 1 would quietly
/// model the per-record counter the design rejects.
fn app_state_record() -> AppStateRecord {
    AppStateRecord {
        context_id: "ctx-a".into(),
        namespace: "openvtc".into(),
        key: "community/acme".into(),
        version: 52,
        deleted: false,
        value: Some(serde_json::json!({ "label": "Acme", "role": "admin" })),
        value_bytes: Some(95),
        created_at: Some(TS.into()),
        updated_at: TS.into(),
        deleted_at: None,
    }
}

/// A tombstone: `deleted` true, no value, and a `deletedAt`. Carried in the
/// list witness because a change feed that cannot express one cannot converge.
fn app_state_tombstone() -> AppStateRecord {
    AppStateRecord {
        context_id: "ctx-a".into(),
        namespace: "openvtc".into(),
        key: "community/defunct".into(),
        version: 44,
        deleted: true,
        value: None,
        value_bytes: None,
        created_at: Some(TS.into()),
        updated_at: TS.into(),
        deleted_at: Some(TS.into()),
    }
}

/// The Rego the declarative approvals fixture synthesizes to. Derived, not
/// transcribed: a hand-copied literal would drift from the synthesizer and the
/// witness would stop testing the shape the VTA actually stores.
fn declarative_module() -> String {
    vta_sdk::approvals::synthesize_rego(&[vta_sdk::approvals::ApprovalRule::reauth(
        uris::TASK_ACL_GRANT_0_1,
    )])
}

/// `ext` for the reserved declarative row, matching [`declarative_module`].
fn declarative_ext() -> Value {
    json!({
        vta_sdk::approvals::EXT_KEY_RULES: [
            { "taskType": uris::TASK_ACL_GRANT_0_1, "requires": "reauth" }
        ],
        vta_sdk::approvals::EXT_KEY_APPROVER_SETS: {},
    })
}

/// A fully-populated canonical `PolicyModule` projection.
fn policy_module_view() -> PolicyModuleView {
    PolicyModuleView {
        id: "approvals".into(),
        name: "Declarative approvals".into(),
        description: Some("Declarative approval rules".into()),
        module: declarative_module(),
        applies_to: vec![],
        priority: vta_sdk::approvals::DECLARATIVE_POLICY_PRIORITY,
        enabled: true,
        version: 2,
        created_at: TS.into(),
        updated_at: TS.into(),
        ext: declarative_ext(),
    }
}

/// A fully-populated canonical `KeyRecord` (the keys family's shared component).
fn key_record() -> KeyRecord {
    KeyRecord {
        key_id: "app-signing-key".into(),
        derivation_path: "m/26'/2'/0'/1'".into(),
        key_type: KeyType::Ed25519,
        status: KeyStatus::Active,
        public_key: "z6MkpTHR8VNsBxYAAWHut2Geadd9jSwuBV8xRoAnwWsdvktH".into(),
        label: Some("app signing key".into()),
        context_id: Some("app".into()),
        seed_id: Some(1),
        origin: KeyOrigin::Derived,
        created_at: dt(),
        updated_at: dt(),
    }
}

/// The create/import response's record shape — a strict subset of `KeyRecord`.
fn key_result() -> CreateKeyResultBody {
    CreateKeyResultBody {
        key_id: "app-signing-key".into(),
        key_type: KeyType::Ed25519,
        derivation_path: "m/26'/2'/0'/1'".into(),
        public_key: "z6MkpTHR8VNsBxYAAWHut2Geadd9jSwuBV8xRoAnwWsdvktH".into(),
        status: KeyStatus::Active,
        label: Some("app signing key".into()),
        origin: KeyOrigin::Derived,
        created_at: dt(),
    }
}

/// A fully-populated canonical `AclEntry` (the family's shared component).
fn acl_entry() -> AclEntry {
    AclEntry {
        subject: SUBJECT.into(),
        role: "reader".into(),
        scopes: vec!["ctx-a".into()],
        label: Some("build agent".into()),
        created_at: Some(dt()),
        created_by: Some("did:key:z6MkAdmin".into()),
        updated_at: Some(dt()),
        updated_by: Some("did:key:z6MkAdmin".into()),
        expires_at: Some(dt()),
        step_up: Some(StepUp {
            approver: Some("did:key:z6MkApprover".into()),
            require: Some("delegated".into()),
        }),
        approve: Some(Approve {
            all: false,
            scopes: vec!["ctx-a".into()],
        }),
        // `allowedKeys` (#818) reached the published
        // `acl/_shared/0.1/acl-entry` component in `trust-tasks-rs` 0.2.51,
        // so the entry every ACL response embeds now carries it. Populated
        // rather than `None` deliberately: `None` and `Some(∅)` are opposite
        // grants and only a present value proves the member survives the
        // wire under its canonical `allowedKeys` spelling.
        allowed_keys: Some(vec!["tenant-key-a".into()]),
    }
}

fn to_v<T: serde::Serialize>(t: T) -> Value {
    serde_json::to_value(t).expect("serialize fixture")
}

/// The `Session` object the auth slice emits (whoami / sessions/list /
/// step-up elevation) — mirrors the `json!` shapes in `auth.rs` / `step_up.rs`.
fn session_json() -> Value {
    json!({
        "id": "sess-1",
        "subject": SUBJECT,
        "issuedAt": TS,
        "expiresAt": TS,
        "amr": ["swk"],
        "acr": "aal2",
    })
}

/// A wire `DeviceBinding` as `operations::device::to_wire_binding` emits it.
fn device_binding_json() -> Value {
    json!({
        "deviceId": "dev-1",
        "consumerKind": { "kind": "companion", "formFactor": "browser" },
        "displayName": "Laptop",
        "registeredAt": TS,
        "consumerDid": SUBJECT,
        "platform": "macOS",
        "pushCapable": false,
    })
}

/// A canonical `ConsentSubject` — mirrors `consent.rs`'s `WireSubject`.
fn consent_subject_json() -> Value {
    json!({
        "platform": "slack",
        "conversationRef": "conv-9f3",
        "kind": "dm",
        "agent": "agent-1",
    })
}

/// A fully-populated wire `VaultEntry` (`vti_common::vault::VaultEntry`,
/// authored against the canonical `vault/_shared` component).
fn vault_entry() -> vti_common::vault::VaultEntry {
    use vti_common::vault::{SecretKind, SiteTarget, VaultEntry};
    VaultEntry {
        id: "01HVAULTENTRY".into(),
        context_id: "ctx-a".into(),
        targets: vec![SiteTarget::WebOrigin {
            origin: "https://example.com".into(),
        }],
        label: "example.com login".into(),
        secret_kind: SecretKind::Password,
        tags: vec!["work".into()],
        notes: Some("shared team login".into()),
        favicon: None,
        selectors: vec![],
        custom_field_names: vec![],
        attachments: vec![],
        expires_at: Some(TS.into()),
        breached_at: None,
        password_changed_at: Some(TS.into()),
        created_at: TS.into(),
        created_by: Some(SUBJECT.into()),
        updated_at: TS.into(),
        updated_by: Some(SUBJECT.into()),
        last_used_at: Some(TS.into()),
        version: 3,
        principal_did: None,
        // Active entries omit the whole lifecycle block on the wire
        // (`skip_serializing_if`), which is what keeps them canonical — the
        // published `VaultEntry` component predates the archival lifecycle.
        status: vti_common::vault::VaultStatus::Active,
        archived_at: None,
        deleted_at: None,
        grace_until: None,
    }
}

// ─── The witness table ───────────────────────────────────────────────────

/// One entry per URI in [`resolved_uris`] — the coverage assertion below
/// holds the two in lockstep, in both directions.
///
/// `#[allow(deprecated)]` for the same reason the dispatch table carries it:
/// the 0.1 constants are deprecated in favour of their 0.2 siblings but are
/// still dispatched during the migration window, so they are still in the
/// census this table must cover.
#[allow(deprecated)]
fn table() -> Vec<(&'static str, Conformance)> {
    let mut t: Vec<(&'static str, Conformance)> = vec![
        // ─── discovery ───────────────────────────────────────────
        //
        // The one canonical family this agent serves that is *about* the agent
        // rather than its data. The request pins a real slug-glob rather than
        // `*`, so the witness exercises the pattern grammar the handler
        // implements; the response pins the shorthand (bare Type URI) form.
        (
            uris::TASK_TRUST_TASK_DISCOVERY_0_1,
            checked!(
                specs::trust_task_discovery::v0_1::Payload,
                specs::trust_task_discovery::v0_1::Response,
                json!({ "patterns": ["acl/*"] }),
                json!({
                    "frameworkVersion": "0.2",
                    "supportedTypes": [
                        "https://trusttasks.org/spec/acl/grant/0.1",
                        "https://trusttasks.org/spec/acl/revoke/0.1"
                    ]
                })
            ),
        ),
        // ─── auth ────────────────────────────────────────────────
        (
            uris::TASK_AUTH_REVOKE_SESSION_0_1,
            checked!(
                specs::auth::revoke_session::v0_1::Payload,
                specs::auth::revoke_session::v0_1::Response,
                to_v(RevokeSessionRequest {
                    all: None,
                    session_id: Some("sess-1".into()),
                }),
                to_v(RevokeSessionResponse { revoked_count: 1 })
            ),
        ),
        (
            uris::TASK_AUTH_WHOAMI_0_1,
            checked!(
                specs::auth::whoami::v0_1::Payload,
                specs::auth::whoami::v0_1::Response,
                json!({}), // whoami has no request members; the handler ignores the payload
                json!({ "session": session_json(), "roles": ["admin"], "scopes": ["ctx:root"] })
            ),
        ),
        (
            uris::TASK_AUTH_SESSIONS_LIST_0_1,
            checked!(
                specs::auth::sessions::list::v0_1::Payload,
                specs::auth::sessions::list::v0_1::Response,
                json!({}),
                json!({ "sessions": [session_json()] })
            ),
        ),
        (
            uris::TASK_AUTH_STEP_UP_APPROVE_RESPONSE_0_1,
            checked!(
                specs::auth::step_up::approve_response::v0_1::Payload,
                specs::auth::step_up::approve_response::v0_1::Response,
                // 0.1 wire: kebab-case evidence discriminator.
                json!({
                    "subject": SUBJECT,
                    "sessionId": "sess-1",
                    "challenge": "chal-0123456789abcdef",
                    "decision": "approved",
                    "grantedAcr": "aal2",
                    "evidence": { "kind": "did-signed" },
                }),
                json!({ "status": "elevated", "session": session_json() })
            ),
        ),
        (
            uris::TASK_AUTH_STEP_UP_APPROVE_RESPONSE_0_2,
            checked!(
                specs::auth::step_up::approve_response::v0_2::Payload,
                specs::auth::step_up::approve_response::v0_2::Response,
                // 0.2 wire: camelCase evidence discriminator — the member the
                // handler kebabizes on a copy (`wire_v0_2::kebabize_paths`).
                json!({
                    "subject": SUBJECT,
                    "sessionId": "sess-1",
                    "challenge": "chal-0123456789abcdef",
                    "decision": "approved",
                    "grantedAcr": "aal2",
                    "evidence": { "kind": "didSigned" },
                }),
                json!({ "status": "elevated", "session": session_json() })
            ),
        ),
        // `auth/step-up/policy/0.2` had a witness here until the VTA stopped
        // serving it. The task set `[auth.step_up]` — the config floors — and
        // there are none to set: the rules are the whole approvals model now,
        // managed through `policy/*` and `pnm approvals`. The coverage
        // assertion below checks both directions, so leaving the witness
        // behind would have failed as stale-witness drift.
        // ─── consent (types are consent.rs-private; fixtures mirror them) ──
        (
            uris::TASK_CONSENT_REQUEST_1_0,
            checked!(
                specs::consent::request::v1_0::Payload,
                specs::consent::request::v1_0::Response,
                // Every optional member set. The witness used to leave both
                // hints `None`, which serialized them away — so the fixture
                // proved the required trio and nothing about the optional
                // members' encoding. `firstMessageDigest` in particular is a
                // `DigestMultibase`, and an omitted member cannot fail its
                // pattern.
                to_v(ConsentRequestBody {
                    subject: consent_subject_json(),
                    scope: "converse".into(),
                    challenge: "chal-0123456789abcdef".into(),
                    display_hint: Some("Slack DM".into()),
                    first_message_digest: Some(DIGEST_MULTIBASE.into()),
                    context_hint: Some("ctx-a".into()),
                }),
                json!({ "status": "accepted", "requestId": "chal-1" })
            ),
        ),
        (
            uris::TASK_CONSENT_DECISION_1_0,
            checked!(
                specs::consent::decision::v1_0::Payload,
                specs::consent::decision::v1_0::Response,
                to_v(ConsentDecisionBody {
                    subject: consent_subject_json(),
                    effect: "allow".into(),
                    scope: Some("converse".into()),
                    challenge: Some("chal-0123456789abcdef".into()),
                    expires_at: None,
                }),
                json!({ "status": "recorded", "grantId": "urn:uuid:6d9c" })
            ),
        ),
        (
            uris::TASK_CONSENT_REVOKE_1_0,
            checked!(
                specs::consent::revoke::v1_0::Payload,
                specs::consent::revoke::v1_0::Response,
                to_v(ConsentRevokeBody {
                    subject: consent_subject_json(),
                    reason: None,
                }),
                json!({ "status": "revoked" })
            ),
        ),
        (
            uris::TASK_CONSENT_LIST_1_0,
            checked!(
                specs::consent::list::v1_0::Payload,
                specs::consent::list::v1_0::Response,
                // An unfiltered list — every member unset, which must
                // serialize to `{}` rather than a map of nulls.
                to_v(ConsentListBody::default()),
                json!({ "grants": [{
                    "subject": consent_subject_json(),
                    "effect": "allow",
                    "scope": "converse",
                    "grantedBy": "did:key:z6MkKeeper",
                    "grantedAt": TS,
                    "expiresAt": TS,
                    "evidence": "did-signed",
                }] })
            ),
        ),
        (
            uris::TASK_CONSENT_APPROVER_SET_1_0,
            checked!(
                specs::consent::approver_set::v1_0::Payload,
                specs::consent::approver_set::v1_0::Response,
                to_v(ConsentApproverSetBody {
                    platform: "slack".into(),
                    context: "ctx-a".into(),
                    approver: "did:key:z6MkKeeper".into(),
                    route: Some("wake".into()),
                    route_hint: None,
                }),
                json!({ "status": "set" })
            ),
        ),
        (
            uris::TASK_CONSENT_APPROVER_LIST_1_0,
            checked!(
                specs::consent::approver_list::v1_0::Payload,
                specs::consent::approver_list::v1_0::Response,
                to_v(ConsentApproverListBody::default()),
                json!({ "approvers": [{
                    "platform": "slack",
                    "context": "ctx-a",
                    "approver": "did:key:z6MkKeeper",
                    "route": "wake",
                }] })
            ),
        ),
        // ─── task-consent (types are task_consent.rs-private) ─────
        (
            uris::TASK_TASK_CONSENT_DECISION_0_1,
            checked!(
                specs::task_consent::decision::v0_1::Payload,
                specs::task_consent::decision::v0_1::Response,
                json!({
                    "challenge": "chal-0123456789abcdef",
                    "payloadDigest": DIGEST_MULTIBASE,
                    "decision": "approve",
                    "reason": "looks right",
                }),
                json!({
                    "status": "pending",
                    "payloadDigest": DIGEST_MULTIBASE,
                    "approvals": 1,
                    "needed": 2,
                })
            ),
        ),
        // ─── acl ─────────────────────────────────────────────────
        (
            uris::TASK_ACL_LIST_0_1,
            checked!(
                specs::acl::list::v0_1::Payload,
                specs::acl::list::v0_1::Response,
                to_v(ListAclBody {
                    role: Some("reader".into()),
                    scope: Some("ctx-a".into()),
                    direction: Some(ContextDirection::Subtree),
                    subject_prefix: Some("did:key:z6Mk".into()),
                    page_size: Some(50),
                    cursor: Some("cur-1".into()),
                }),
                to_v(ListAclResultBody {
                    entries: vec![acl_entry()],
                    truncated: false,
                    cursor: None,
                    redacted_fields: Vec::new(),
                })
            ),
        ),
        (
            uris::TASK_ACL_GRANT_0_1,
            checked!(
                specs::acl::grant::v0_1::Payload,
                specs::acl::grant::v0_1::Response,
                to_v(CreateAclBody {
                    entry: acl_entry(),
                    reason: Some("onboarding".into()),
                }),
                to_v(CreateAclResponseBody { entry: acl_entry() })
            ),
        ),
        (
            uris::TASK_ACL_SHOW_0_1,
            checked!(
                specs::acl::show::v0_1::Payload,
                specs::acl::show::v0_1::Response,
                to_v(GetAclBody {
                    subject: SUBJECT.into(),
                }),
                to_v(GetAclResultBody {
                    entry: acl_entry(),
                    redacted_fields: Vec::new(),
                })
            ),
        ),
        (
            uris::TASK_ACL_UPDATE_0_1,
            checked!(
                specs::acl::update::v0_1::Payload,
                specs::acl::update::v0_1::Response,
                to_v(UpdateAclBody {
                    did: SUBJECT.into(),
                    label: Some("renamed".into()),
                    allowed_contexts: Some(vec!["ctx-a".into(), "ctx-b".into()]),
                    expires_at: Some(dt()),
                    reason: Some("access review".into()),
                    step_up: Some(StepUp {
                        approver: Some("did:key:z6MkApprover".into()),
                        require: Some("delegated".into()),
                    }),
                    approve: Some(Approve {
                        all: false,
                        scopes: vec!["ctx-a".into()],
                    }),
                    // `allowedKeys` (#818) is in the published
                    // `acl/update/0.1` schema as of `trust-tasks-rs` 0.2.51.
                    // `Some(Some(..))` is the narrowing replacement — the
                    // arm that actually emits an array; `Some(None)` emits
                    // an explicit `null` (clear) and `None` emits nothing,
                    // and `update.rs`'s own tests pin all three.
                    allowed_keys: Some(Some(vec!["tenant-key-a".into()])),
                }),
                to_v(CreateAclResponseBody { entry: acl_entry() })
            ),
        ),
        (
            uris::TASK_ACL_CHANGE_ROLE_0_1,
            checked!(
                specs::acl::change_role::v0_1::Payload,
                specs::acl::change_role::v0_1::Response,
                to_v(ChangeRoleBody {
                    subject: SUBJECT.into(),
                    from_role: "reader".into(),
                    to_role: "application".into(),
                    reason: Some("promoted".into()),
                }),
                to_v(CreateAclResponseBody { entry: acl_entry() })
            ),
        ),
        (
            uris::TASK_ACL_REVOKE_0_1,
            checked!(
                specs::acl::revoke::v0_1::Payload,
                specs::acl::revoke::v0_1::Response,
                to_v(DeleteAclBody {
                    subject: SUBJECT.into(),
                    scopes: Some(vec!["ctx-a".into()]),
                    reason: Some("offboarding".into()),
                }),
                to_v(DeleteAclResultBody { entry: acl_entry() })
            ),
        ),
        (
            uris::TASK_ACL_SWAP_KEY_0_1,
            checked!(
                specs::acl::swap_key::v0_1::Payload,
                specs::acl::swap_key::v0_1::Response,
                to_v(SwapKeyBody {
                    current_subject: SUBJECT.into(),
                    new_subject: "did:key:z6MkNewSubject".into(),
                    link_proof: Some("eyJhbGciOiJFZERTQSJ9.e30.c2ln".into()),
                    reason: Some("wallet rotation".into()),
                }),
                to_v(SwapKeyResultBody {
                    entry: acl_entry(),
                    previous_subject: SUBJECT.into(),
                })
            ),
        ),
        // ─── webvh servers/domains (the relay, #171) ─────────────
        (
            uris::TASK_WEBVH_SERVERS_DOMAINS_0_1,
            checked!(
                specs::vta::webvh::servers::domains::v0_1::Payload,
                specs::vta::webvh::servers::domains::v0_1::Response,
                to_v(ListWebvhServerDomainsBody {
                    server_id: "primary-host".into(),
                }),
                to_v(ListWebvhServerDomainsResultBody {
                    domains: vec![WebvhServerDomainEntry {
                        name: "did.example.com".into(),
                        default_domain: true,
                        status: "active".into(),
                        label: Some("Production".into()),
                        // Required by the canonical DomainEntry the response
                        // items reference — a witness without it would pass
                        // while the real relay emitted a non-conformant entry.
                        created_at: Some("2026-03-01T00:00:00Z".into()),
                    }],
                    default: Some("did.example.com".into()),
                })
            ),
        ),
        // ─── webvh servers/reconcile (the host/VTA diff, #210) ───
        //
        // The witness earns its keep here more than most: this task's schema
        // was authored in the same change as the handler, so nothing but this
        // proves the two agree. Both divergence arrays are populated, and the
        // `hostOnly` entry without a `did` is the reserved-but-never-published
        // slot — the case the spec calls out and the one a DID-keyed
        // implementation would drop.
        (
            uris::TASK_WEBVH_SERVERS_RECONCILE_0_1,
            checked!(
                specs::vta::webvh::servers::reconcile::v0_1::Payload,
                specs::vta::webvh::servers::reconcile::v0_1::Response,
                to_v(ReconcileWebvhServerDidsBody {
                    server_id: "primary-host".into(),
                }),
                to_v(ReconcileWebvhServerDidsResultBody {
                    server_id: "primary-host".into(),
                    host_only: vec![
                        HostOnlyDid {
                            slot_id: "attract-case".into(),
                            did: Some(
                                "did:webvh:QmZ4rT9xK2mN8vB5cD1sA7wE3fH6jL0pQ:did.example.com:attract-case"
                                    .into(),
                            ),
                            domain: Some("did.example.com".into()),
                            disabled: false,
                        },
                        HostOnlyDid {
                            slot_id: "quiet-harbour".into(),
                            did: None,
                            domain: None,
                            disabled: false,
                        },
                    ],
                    agent_only: vec![AgentOnlyDid {
                        did: "did:webvh:QmY8nP3bV6xC1kM4hS9dF2gJ5tR7wL0zQ:did.example.com:never-landed"
                            .into(),
                        slot_id: "never-landed".into(),
                        context_id: "production".into(),
                    }],
                    in_both: 14,
                })
            ),
        ),
        // ─── keys (the signing-oracle + custody family, #167) ────
        (
            uris::TASK_KEYS_LIST_0_1,
            checked!(
                specs::keys::list::v0_1::Payload,
                specs::keys::list::v0_1::Response,
                to_v(ListKeysBody {
                    offset: Some(0),
                    limit: Some(50),
                    status: Some(KeyStatus::Active),
                    context_id: Some("app".into()),
                }),
                to_v(ListKeysResultBody {
                    keys: vec![key_record()],
                    total: 1,
                    offset: 0,
                    limit: 50,
                })
            ),
        ),
        (
            uris::TASK_KEYS_CREATE_0_1,
            checked!(
                specs::keys::create::v0_1::Payload,
                specs::keys::create::v0_1::Response,
                to_v(CreateKeyBody {
                    key_id: None,
                    internal: None,
                    key_type: KeyType::Ed25519,
                    derivation_path: Some("m/26'/2'/0'/1'".into()),
                    mnemonic: None,
                    label: Some("app signing key".into()),
                    context_id: Some("app".into()),
                }),
                to_v(CreateKeyResponseBody { key: key_result() })
            ),
        ),
        (
            uris::TASK_KEYS_IMPORT_0_1,
            checked!(
                specs::keys::import::v0_1::Payload,
                specs::keys::import::v0_1::Response,
                to_v(ImportKeyBody {
                    key_type: KeyType::Ed25519,
                    private_key_sealed: Some("-----BEGIN SEALED TRANSFER-----".into()),
                    private_key_jwe: None,
                    private_key_multibase: None,
                    label: Some("migrated signer".into()),
                    context_id: Some("app".into()),
                }),
                to_v(CreateKeyResponseBody { key: key_result() })
            ),
        ),
        (
            uris::TASK_KEYS_SHOW_0_1,
            checked!(
                specs::keys::show::v0_1::Payload,
                specs::keys::show::v0_1::Response,
                to_v(GetKeyBody {
                    key_id: "app-signing-key".into(),
                }),
                to_v(GetKeyResponseBody {
                    key: Some(key_record()),
                })
            ),
        ),
        (
            uris::TASK_KEYS_RENAME_0_1,
            checked!(
                specs::keys::rename::v0_1::Payload,
                specs::keys::rename::v0_1::Response,
                to_v(RenameKeyBody {
                    key_id: "app-signing-key".into(),
                    new_key_id: "app-signing-key-2026".into(),
                }),
                to_v(RenameKeyResultBody {
                    key_id: "app-signing-key-2026".into(),
                    updated_at: dt(),
                })
            ),
        ),
        (
            uris::TASK_KEYS_REVOKE_0_1,
            checked!(
                specs::keys::revoke::v0_1::Payload,
                specs::keys::revoke::v0_1::Response,
                to_v(RevokeKeyBody {
                    key_id: "app-signing-key".into(),
                    reason: Some("superseded".into()),
                }),
                to_v(RevokeKeyResultBody {
                    key_id: "app-signing-key".into(),
                    status: KeyStatus::Revoked,
                    updated_at: dt(),
                })
            ),
        ),
        (
            uris::TASK_KEYS_SIGN_0_1,
            checked!(
                specs::keys::sign::v0_1::Payload,
                specs::keys::sign::v0_1::Response,
                to_v(SignRequestBody {
                    key_id: "app-signing-key".into(),
                    payload: "aGVsbG8".into(),
                    algorithm: SignAlgorithm::EdDSA,
                }),
                to_v(SignResultBody {
                    key_id: "app-signing-key".into(),
                    signature: "3q2-7w".into(),
                    algorithm: SignAlgorithm::EdDSA,
                })
            ),
        ),
        (
            uris::TASK_KEYS_DERIVE_AND_SIGN_0_1,
            checked!(
                specs::keys::derive_and_sign::v0_1::Payload,
                specs::keys::derive_and_sign::v0_1::Response,
                to_v(DeriveAndSignBody {
                    key_type: KeyType::Ed25519,
                    derivation_path: "m/26'/9'/0'".into(),
                    payload: "aGVsbG8".into(),
                    algorithm: SignAlgorithm::EdDSA,
                }),
                to_v(DeriveAndSignResultBody {
                    public_key: "z6MkpTHR8VNsBxYAAWHut2Geadd9jSwuBV8xRoAnwWsdvktH".into(),
                    signature: "3q2-7w".into(),
                    algorithm: SignAlgorithm::EdDSA,
                })
            ),
        ),
        (
            uris::TASK_KEYS_DERIVE_AND_SIGN_DOCUMENT_0_1,
            checked!(
                specs::keys::derive_and_sign_document::v0_1::Payload,
                specs::keys::derive_and_sign_document::v0_1::Response,
                to_v(DeriveAndSignDocumentBody {
                    key_type: KeyType::Ed25519,
                    derivation_path: "m/26'/9'/0'".into(),
                    document: json!({ "id": "urn:uuid:1", "payload": { "a": 1 } }),
                    proof_purpose: Some("assertionMethod".into()),
                }),
                to_v(DeriveAndSignDocumentResultBody {
                    signer_did: "did:key:z6MkpTHR8VNsBxYAAWHut2Geadd9jSwuBV8xRoAnwWsdvktH".into(),
                    document: json!({ "id": "urn:uuid:1", "proof": { "type": "DataIntegrityProof" } }),
                })
            ),
        ),
        // ─── device (handlers parse the generated Payloads directly;
        //     responses mirror operations::device's json! emissions) ──
        (
            uris::TASK_DEVICE_REGISTER_0_1,
            checked!(
                specs::device::register::v0_1::Payload,
                specs::device::register::v0_1::Response,
                // The producer's own body, with its optionals unset — the
                // shape `device_register(kind, name, None, None)` emits, and
                // the one a caller reaches for first. The fixture used to set
                // `keyCustody` and `attestation`, which no producer here sends:
                // members we never emit prove nothing about our wire form.
                to_v(DeviceRegisterBody {
                    consumer_kind: json!({ "kind": "companion", "formFactor": "browser" }),
                    display_name: "Laptop".into(),
                    platform: None,
                    hpke_public_key: None,
                }),
                json!({ "binding": device_binding_json() })
            ),
        ),
        (
            uris::TASK_DEVICE_HEARTBEAT_0_1,
            checked!(
                specs::device::heartbeat::v0_1::Payload,
                specs::device::heartbeat::v0_1::Response,
                // A bare "still here" — every member unset, which must
                // serialize to `{}` rather than a map of nulls.
                to_v(DeviceHeartbeatBody::default()),
                json!({ "serverTime": TS, "queuedOperations": [], "syncHint": "up-to-date" })
            ),
        ),
        (
            uris::TASK_DEVICE_LIST_0_1,
            checked!(
                specs::device::list::v0_1::Payload,
                specs::device::list::v0_1::Response,
                json!({ "consumerKindFilter": "companion", "includeDisabled": true }),
                json!({ "devices": [device_binding_json()], "truncated": false })
            ),
        ),
        (
            uris::TASK_DEVICE_DISABLE_0_1,
            checked!(
                specs::device::disable::v0_1::Payload,
                specs::device::disable::v0_1::Response,
                to_v(DeviceDisableBody {
                    device_id: "dev-1".into(),
                    reason: None,
                }),
                json!({ "deviceId": "dev-1", "disabledAt": TS })
            ),
        ),
        (
            uris::TASK_DEVICE_WIPE_0_1,
            checked!(
                specs::device::wipe::v0_1::Payload,
                specs::device::wipe::v0_1::Response,
                to_v(DeviceWipeBody {
                    device_id: "dev-1".into(),
                    scope: "full".into(),
                    reason: "stolen".into(),
                }),
                json!({ "deviceId": "dev-1", "scope": "full", "completedAt": TS })
            ),
        ),
        (
            uris::TASK_DEVICE_SET_WAKE_0_1,
            checked!(
                specs::device::set_wake::v0_1::Payload,
                specs::device::set_wake::v0_1::Response,
                to_v(DeviceSetWakeBody {
                    wake_handle: Some(WakeHandle {
                        gateway: "did:web:gateway.example".into(),
                        handle: "h-1".into(),
                    }),
                    suggested_triggers: Some(vec!["did:web:mediator.example".into()]),
                }),
                json!({
                    "pushCapable": true,
                    "triggerPolicy": { "allowedTriggers": ["did:web:mediator.example"] },
                })
            ),
        ),
        // ─── messaging ───────────────────────────────────────────
        (
            uris::TASK_MESSAGING_PING_0_1,
            checked!(
                specs::messaging::ping::v0_1::Payload,
                specs::messaging::ping::v0_1::Response,
                json!({ "nonce": "n-1" }),
                json!({ "serverTime": TS, "status": "ok", "protocols": ["didcomm/v2"], "nonce": "n-1" })
            ),
        ),
        // ─── audit ───────────────────────────────────────────────
        (
            uris::TASK_AUDIT_LIST_0_1,
            checked!(
                specs::audit::list::v0_1::Payload,
                specs::audit::list::v0_1::Response,
                to_v(ListAuditLogsBody {
                    from: Some(dt()),
                    to: Some(dt()),
                    action: Some("acl.update".into()),
                    actor: Some(SUBJECT.into()),
                    outcome: Some("success".into()),
                    context_id: Some("ctx-a".into()),
                    page_size: Some(100),
                    cursor: Some("cur-1".into()),
                }),
                to_v(ListAuditLogsResultBody {
                    entries: vec![AuditEnvelope {
                        event_id: "evt-1".into(),
                        recorded_at: TS.into(),
                        action: "acl.update".into(),
                        outcome: Some("success".into()),
                        actor: Some(SUBJECT.into()),
                        target: Some("did:key:z6MkTarget".into()),
                        context_id: Some("ctx-a".into()),
                        detail: serde_json::Map::new(),
                    }],
                    truncated: true,
                    cursor: Some("cur-2".into()),
                })
            ),
        ),
        // ─── credential-exchange: deferred presentations ─────────
        (
            credx::PENDING_LIST,
            checked!(
                specs::credential_exchange::pending::list::v0_1::Payload,
                specs::credential_exchange::pending::list::v0_1::Response,
                json!({}),
                to_v(PendingListResponse {
                    pending: vec![PendingPresentationSummary {
                        id: "thr-1".into(),
                        verifier_did: "did:web:verifier.example".into(),
                        requested: vec![RequestedCredentialSummary {
                            credential_query_id: "q-1".into(),
                            credential_id: "cred-1".into(),
                            claims: vec!["givenName".into()],
                        }],
                        purpose: "age verification".into(),
                        created_at: dt(),
                        expires_at: dt(),
                    }],
                })
            ),
        ),
        (
            credx::PENDING_APPROVE,
            checked!(
                specs::credential_exchange::pending::approve::v0_1::Payload,
                specs::credential_exchange::pending::approve::v0_1::Response,
                to_v(PendingApproveBody { id: "thr-1".into() }),
                to_v(PresentBody {
                    vp_token: json!({ "vp": {} }),
                })
            ),
        ),
        (
            credx::PENDING_DENY,
            checked!(
                specs::credential_exchange::pending::deny::v0_1::Payload,
                specs::credential_exchange::pending::deny::v0_1::Response,
                to_v(PendingDenyBody { id: "thr-1".into() }),
                to_v(PendingDenyResponse {
                    id: "thr-1".into(),
                    status: "denied".into(),
                })
            ),
        ),
        // ─── vault (types are vault.rs-private; fixtures mirror them,
        //     entries via the shared vti_common::vault::VaultEntry) ──
        (
            uris::TASK_VAULT_LIST_0_1,
            checked!(
                specs::vault::list::v0_1::Payload,
                specs::vault::list::v0_1::Response,
                json!({
                    "contextId": "ctx-a",
                    "targetOriginPrefix": "https://example.com",
                    "secretKind": "password",
                    "tag": "work",
                    "status": "active",
                    "pageSize": 20,
                }),
                json!({ "entries": [to_v(vault_entry())], "truncated": false })
            ),
        ),
        (
            uris::TASK_VAULT_GET_0_1,
            checked!(
                specs::vault::get::v0_1::Payload,
                specs::vault::get::v0_1::Response,
                json!({ "id": "01HVAULTENTRY" }),
                json!({ "entry": to_v(vault_entry()) })
            ),
        ),
        (
            uris::TASK_VAULT_UPSERT_0_1,
            checked!(
                specs::vault::upsert::v0_1::Payload,
                specs::vault::upsert::v0_1::Response,
                {
                    // Built from the producer's body — the minimal create,
                    // with every optional unset. `sealedSecret` is not on the
                    // body: `vault_upsert_typed` inserts it, because sealing
                    // needs the client's HPKE context rather than the caller's.
                    let mut v = to_v(VaultUpsertBody {
                        context_id: "ctx-a".into(),
                        targets: vec![
                            json!({ "kind": "web-origin", "origin": "https://example.com" }),
                        ],
                        label: "example.com login".into(),
                        secret_kind: "password".into(),
                        id: None,
                        expected_version: None,
                        tags: None,
                        notes: None,
                        expires_at: None,
                        extra: serde_json::Map::new(),
                    });
                    v.as_object_mut().expect("body is an object").insert(
                        "sealedSecret".into(),
                        json!({ "envelope": "didcomm-authcrypt", "jwe": "eyJ0eXAiOiJKV0UifQ" }),
                    );
                    v
                },
                json!({ "entry": to_v(vault_entry()), "created": false })
            ),
        ),
        (
            uris::TASK_VAULT_DELETE_0_1,
            checked!(
                specs::vault::delete::v0_1::Payload,
                specs::vault::delete::v0_1::Response,
                // Built from the producer's body, with the precondition and
                // reason unset — the shape `vault_delete(id, None, false, None)`
                // emits, and the one that leaves both optionals to the skip.
                to_v(vta_sdk::protocols::vault_management::VaultDeleteBody {
                    id: "01HVAULTENTRY".into(),
                    force: false,
                    expected_version: None,
                    reason: None,
                }),
                json!({ "id": "01HVAULTENTRY", "deletedAt": TS, "graceUntil": TS })
            ),
        ),
        (
            uris::TASK_VAULT_RELEASE_0_1,
            checked!(
                specs::vault::release::v0_1::Payload,
                specs::vault::release::v0_1::Response,
                json!({
                    "entryId": "01HVAULTENTRY",
                    "target": { "kind": "web-origin", "origin": "https://example.com" },
                    "consumerContext": { "deviceId": "dev-1", "networkClass": "home" },
                    "ttlSecondsHint": 60,
                }),
                json!({
                    "sealedSecret": { "envelope": "didcomm-authcrypt", "jwe": "eyJwcm90ZWN0ZWQiOiJ..." },
                    "secretKind": "password",
                    "ttlSeconds": 60,
                })
            ),
        ),
        (
            uris::TASK_VAULT_PROXY_LOGIN_0_1,
            checked!(
                specs::vault::proxy_login::v0_1::Payload,
                specs::vault::proxy_login::v0_1::Response,
                json!({
                    "entryId": "01HVAULTENTRY",
                    "target": { "kind": "web-origin", "origin": "https://example.com" },
                    "nonce": "n-1",
                    "ttlSecondsHint": 60,
                }),
                // Mirrors `VaultProxyLoginResponseBody`: the session hints ride
                // in the vendor-namespaced `ext` (the canonical response object
                // is closed; #857).
                json!({
                    "sealedSessionBlob": { "envelope": "didcomm-authcrypt", "jwe": "eyJwcm90ZWN0ZWQiOiJ..." },
                    "ext": { "org.openvtc.vault-session": { "sessionId": "vs-1", "expiresAt": TS } },
                })
            ),
        ),
        (
            uris::TASK_VAULT_SIGN_TRUST_TASK_0_1,
            checked!(
                specs::vault::sign_trust_task::v0_1::Payload,
                specs::vault::sign_trust_task::v0_1::Response,
                json!({
                    "entryId": "01HVAULTENTRY",
                    "unsignedEnvelope": {
                        "id": "urn:uuid:7c11",
                        "type": "https://trusttasks.org/spec/messaging/ping/0.1",
                        "issuer": SUBJECT,
                        "recipient": "did:web:vta.example",
                        "issuedAt": TS,
                        "payload": { "nonce": "n-1" },
                    },
                }),
                json!({ "signedEnvelope": {
                    "id": "urn:uuid:7c11",
                    "type": "https://trusttasks.org/spec/messaging/ping/0.1",
                    "issuer": SUBJECT,
                    "recipient": "did:web:vta.example",
                    "issuedAt": TS,
                    "payload": { "nonce": "n-1" },
                    "proof": { "type": "DataIntegrityProof" },
                } })
            ),
        ),
        // ─── vta/credentials ─────────────────────────────────────
        (
            uris::TASK_VTA_CREDENTIALS_ISSUE_0_2,
            checked!(
                specs::vta::credentials::issue::v0_2::Payload,
                specs::vta::credentials::issue::v0_2::Response,
                to_v(IssueCredentialBody {
                    holder: SUBJECT.into(),
                    claims: json!({ "role": "agent" }),
                    credential_type: Some("AgentAuthorization".into()),
                    validity_seconds: 3600,
                    purpose: Some("agent onboarding".into()),
                    // Deliberately absent: `authorizationContext` is a member the
                    // published schema does not define, so a producer cannot
                    // send it through the validated transport at all.
                    authorization_context: None,
                }),
                to_v(IssueCredentialResponse {
                    credential_id: "cred-1".into(),
                    credential: json!({ "@context": [], "type": ["VerifiableCredential"] }),
                    expires_at: TS.into(),
                    issued_at: Some(TS.into()),
                })
            ),
        ),
        (
            uris::TASK_VTA_CREDENTIALS_REVOKE_0_1,
            checked!(
                specs::vta::credentials::revoke::v0_1::Payload,
                specs::vta::credentials::revoke::v0_1::Response,
                to_v(RevokeCredentialBody {
                    credential_id: "cred-1".into(),
                    reason: Some("holder offboarded".into()),
                }),
                to_v(RevokeCredentialResponse {
                    credential_id: "cred-1".into(),
                    revoked_at: TS.into(),
                })
            ),
        ),
        // ─── vta/memory ──────────────────────────────────────────
        (
            uris::TASK_VTA_MEMORY_PUT_0_1,
            checked!(
                specs::vta::memory::put::v0_1::Payload,
                specs::vta::memory::put::v0_1::Response,
                to_v(MemoryPutBody {
                    context_id: "ctx-a".into(),
                    key: "greeting".into(),
                    value: "hello".into(),
                }),
                to_v(MemoryPutResponse {
                    key: "greeting".into(),
                })
            ),
        ),
        (
            uris::TASK_VTA_MEMORY_LIST_0_1,
            checked!(
                specs::vta::memory::list::v0_1::Payload,
                specs::vta::memory::list::v0_1::Response,
                to_v(MemoryListBody {
                    context_id: "ctx-a".into(),
                }),
                to_v(MemoryListResponse {
                    items: vec![MemoryItem {
                        key: "greeting".into(),
                        value: "hello".into(),
                    }],
                })
            ),
        ),
        (
            uris::TASK_VTA_MEMORY_DELETE_0_1,
            checked!(
                specs::vta::memory::delete::v0_1::Payload,
                specs::vta::memory::delete::v0_1::Response,
                to_v(MemoryDeleteBody {
                    context_id: "ctx-a".into(),
                    key: "greeting".into(),
                }),
                to_v(MemoryDeleteResponse {
                    key: "greeting".into(),
                })
            ),
        ),
        // ─── vta/app-state ───────────────────────────────────────
        //
        // The fixtures deliberately exercise the members the design turns on
        // rather than a minimal payload: a version on every record, a conflict
        // carrying `currentValue`, a tombstone with `deleted: true` and no
        // value, and a change-feed response with its `highWatermark`. A witness
        // built from the smallest legal document would validate while testing
        // none of it.
        (
            uris::TASK_VTA_APP_STATE_GET_1_0,
            checked!(
                specs::vta::app_state::get::v1_0::Payload,
                specs::vta::app_state::get::v1_0::Response,
                to_v(AppStateGetBody {
                    context_id: "ctx-a".into(),
                    namespace: "openvtc".into(),
                    key: "community/acme".into(),
                    include_deleted: None,
                    ext: None,
                }),
                to_v(AppStateGetResponse {
                    record: app_state_record(),
                })
            ),
        ),
        (
            uris::TASK_VTA_APP_STATE_PUT_1_0,
            checked!(
                specs::vta::app_state::put::v1_0::Payload,
                specs::vta::app_state::put::v1_0::Response,
                to_v(AppStatePutBody {
                    context_id: "ctx-a".into(),
                    namespace: "openvtc".into(),
                    key: "community/acme".into(),
                    value: Some(serde_json::json!({ "label": "Acme" })),
                    merge_patch: None,
                    expected_version: Some(47),
                    ext: None,
                }),
                to_v(AppStatePutResponse {
                    context_id: "ctx-a".into(),
                    namespace: "openvtc".into(),
                    key: "community/acme".into(),
                    version: 52,
                    created: false,
                    updated_at: TS.into(),
                    value_bytes: Some(95),
                })
            ),
        ),
        (
            uris::TASK_VTA_APP_STATE_LIST_1_0,
            checked!(
                specs::vta::app_state::list::v1_0::Payload,
                specs::vta::app_state::list::v1_0::Response,
                to_v(AppStateListBody {
                    context_id: "ctx-a".into(),
                    namespace: Some("openvtc".into()),
                    prefix: Some("community/".into()),
                    since_version: Some(40),
                    include_values: Some(true),
                    include_deleted: None,
                    page_size: Some(50),
                    cursor: None,
                    ext: None,
                }),
                to_v(AppStateListResponse {
                    records: vec![app_state_record(), app_state_tombstone()],
                    truncated: false,
                    cursor: None,
                    high_watermark: Some(52),
                    tombstone_retention_seconds: Some(2_592_000),
                })
            ),
        ),
        (
            uris::TASK_VTA_APP_STATE_DELETE_1_0,
            checked!(
                specs::vta::app_state::delete::v1_0::Payload,
                specs::vta::app_state::delete::v1_0::Response,
                to_v(AppStateDeleteBody {
                    context_id: "ctx-a".into(),
                    namespace: "openvtc".into(),
                    key: "community/defunct".into(),
                    expected_version: Some(44),
                    ext: None,
                }),
                to_v(AppStateDeleteResponse {
                    context_id: "ctx-a".into(),
                    namespace: "openvtc".into(),
                    key: "community/defunct".into(),
                    existed: true,
                    version: Some(54),
                    deleted_at: Some(TS.into()),
                })
            ),
        ),
        (
            uris::TASK_VTA_APP_STATE_GET_MANY_1_0,
            checked!(
                specs::vta::app_state::get_many::v1_0::Payload,
                specs::vta::app_state::get_many::v1_0::Response,
                to_v(AppStateGetManyBody {
                    context_id: "ctx-a".into(),
                    namespace: "openvtc".into(),
                    keys: vec!["community/acme".into(), "profile/labels".into()],
                    include_deleted: None,
                    ext: None,
                }),
                to_v(AppStateGetManyResponse {
                    records: vec![app_state_record()],
                    missing: vec!["profile/labels".into()],
                    deferred: None,
                })
            ),
        ),
        (
            uris::TASK_VTA_APP_STATE_PUT_MANY_1_0,
            checked!(
                specs::vta::app_state::put_many::v1_0::Payload,
                specs::vta::app_state::put_many::v1_0::Response,
                to_v(AppStatePutManyBody {
                    context_id: "ctx-a".into(),
                    namespace: "openvtc".into(),
                    mode: Some(PutManyMode::Independent),
                    writes: vec![
                        AppStateWrite {
                            key: "community/acme".into(),
                            value: None,
                            merge_patch: Some(serde_json::json!({ "role": "owner" })),
                            expected_version: Some(52),
                        },
                        AppStateWrite {
                            key: "profile/labels".into(),
                            value: Some(serde_json::json!({ "colours": { "acme": "blue" } })),
                            merge_patch: None,
                            expected_version: Some(0),
                        },
                    ],
                    ext: None,
                }),
                to_v(AppStatePutManyResponse {
                    mode: PutManyMode::Independent,
                    results: vec![
                        WriteResult {
                            key: "community/acme".into(),
                            outcome: WriteOutcome::Written,
                            version: Some(59),
                            created: Some(false),
                            current_version: None,
                            current_value: None,
                            current_deleted: None,
                            limit_bytes: None,
                            actual_bytes: None,
                        },
                        // A conflict carrying the winner's view — the member
                        // that removes the re-read race, and the one most
                        // likely to be dropped by a well-meaning refactor.
                        WriteResult {
                            key: "profile/labels".into(),
                            outcome: WriteOutcome::Conflict,
                            version: None,
                            created: None,
                            current_version: Some(57),
                            current_value: Some(serde_json::json!({ "colours": {} })),
                            current_deleted: None,
                            limit_bytes: None,
                            actual_bytes: None,
                        },
                    ],
                    high_watermark: Some(60),
                })
            ),
        ),
        // ─── policy (runtime PDP management) ──────────────────────
        //
        // The reserved declarative row is the fixture throughout: it is the
        // shape that carries `ext`, and `ext` is the part most likely to drift
        // from the canonical `ExtKey` pattern (lowercase reverse-DNS, which is
        // why the key is `openvtc.approver-sets` and not `approverSets`).
        (
            uris::TASK_POLICY_LIST_0_2,
            checked!(
                specs::policy::list::v0_2::Payload,
                specs::policy::list::v0_2::Response,
                to_v(ListPoliciesBody {
                    context_id: Some("ctx-a".into()),
                    enabled_only: true,
                    cursor: None,
                    page_size: Some(50),
                }),
                to_v(ListPoliciesResultBody {
                    policies: vec![policy_module_view()],
                    truncated: false,
                    cursor: None,
                })
            ),
        ),
        (
            uris::TASK_POLICY_GET_0_1,
            checked!(
                specs::policy::get::v0_1::Payload,
                specs::policy::get::v0_1::Response,
                to_v(GetPolicyBody {
                    id: "approvals".into(),
                }),
                to_v(GetPolicyResultBody {
                    policy: policy_module_view(),
                })
            ),
        ),
        (
            uris::TASK_POLICY_UPSERT_0_2,
            checked!(
                specs::policy::upsert::v0_2::Payload,
                specs::policy::upsert::v0_2::Response,
                to_v(UpsertPolicyBody {
                    id: Some("approvals".into()),
                    name: "Declarative approvals".into(),
                    description: None,
                    module: declarative_module(),
                    applies_to: vec![],
                    priority: Some(vta_sdk::approvals::DECLARATIVE_POLICY_PRIORITY),
                    enabled: true,
                    expected_version: Some(1),
                    ext: declarative_ext(),
                }),
                to_v(UpsertPolicyResultBody {
                    policy: policy_module_view(),
                    created: false,
                })
            ),
        ),
        (
            uris::TASK_POLICY_DELETE_0_1,
            checked!(
                specs::policy::delete::v0_1::Payload,
                specs::policy::delete::v0_1::Response,
                to_v(DeletePolicyBody {
                    id: "legacy-rego".into(),
                    expected_version: Some(2),
                    reason: Some("superseded by the declarative rules".into()),
                }),
                to_v(DeletePolicyResultBody {
                    id: "legacy-rego".into(),
                    deleted_at: TS.into(),
                })
            ),
        ),
        // ─── config ──────────────────────────────────────────────
        (
            uris::TASK_CONFIG_SHOW_0_1,
            checked!(
                specs::config::show::v0_1::Payload,
                specs::config::show::v0_1::Response,
                to_v(GetConfigBody {
                    keys: Some(vec!["public_url".into()]),
                }),
                to_v(GetConfigResultBody {
                    fields: vec![ConfigField {
                        key: "public_url".into(),
                        value: json!("https://vta.example"),
                        source: "db".into(),
                        requires_restart: false,
                    }],
                })
            ),
        ),
        (
            uris::TASK_CONFIG_PATCH_0_1,
            checked!(
                specs::config::patch::v0_1::Payload,
                specs::config::patch::v0_1::Response,
                to_v(UpdateConfigBody {
                    overrides: [("public_url".to_string(), json!("https://vta.example"))]
                        .into_iter()
                        .collect(),
                }),
                to_v(UpdateConfigResultBody {
                    applied: vec!["public_url".into()],
                    pending_restart: vec![],
                    rejected: vec![RejectedKey {
                        key: "vta_did".into(),
                        reason: "immutable".into(),
                    }],
                })
            ),
        ),
    ];

    // ─── vta/did-templates ────────────────────────────────────────
    //
    // The merged 2.0 family (#851, bound by #864, published into the pinned
    // `trust-tasks-rs` by this bump). Its whole point is that scope moved out
    // of the slug hierarchy and into ONE optional `contextId` member: absent
    // = the global scope, present = that context's. A witness can only carry
    // one of the two, so each witness carries the context form (the richer)
    // and the global form is asserted alongside — otherwise "optional" would
    // be the one thing about this family the sweep never checks.
    {
        use vta_sdk::did_templates::{DidTemplate, DidTemplateRecord, Scope};
        use vta_sdk::protocols::did_template_management as tpl;

        const TPL_NAME: &str = "didcomm-mediator";
        const CTX: &str = "ctx-a";
        // UTC unix-epoch seconds, the record's own time unit (not `TS`).
        const EPOCH: u64 = 1_785_369_600;

        /// The authored template body `create`/`update` carry and every
        /// record embeds.
        fn template() -> DidTemplate {
            DidTemplate {
                schema_version: 1,
                name: TPL_NAME.into(),
                kind: "mediator".into(),
                description: Some("DIDComm mediator".into()),
                methods: vec!["webvh".into()],
                required_vars: vec!["SERVICE_ENDPOINT".into()],
                optional_vars: [("LABEL".to_string(), json!("mediator"))]
                    .into_iter()
                    .collect(),
                defaults: [("preRotationCount".to_string(), json!(2))]
                    .into_iter()
                    .collect(),
                document: json!({
                    "id": "{DID}",
                    "service": [{
                        "id": "{DID}#didcomm",
                        "type": "DIDCommMessaging",
                        "serviceEndpoint": "{SERVICE_ENDPOINT}",
                    }],
                }),
            }
        }

        /// The persisted record `create`/`get`/`update` return — the
        /// authored template flattened under the resolved scope and
        /// provenance metadata.
        fn record() -> DidTemplateRecord {
            DidTemplateRecord {
                template: template(),
                scope: Scope::Context {
                    context_id: CTX.into(),
                },
                created_at: EPOCH,
                updated_at: EPOCH,
                created_by: "did:key:z6MkAdmin".into(),
            }
        }

        fn rendered() -> Value {
            json!({
                "id": "did:webvh:scid:vta.example",
                "service": [{
                    "id": "did:webvh:scid:vta.example#didcomm",
                    "type": "DIDCommMessaging",
                    "serviceEndpoint": "https://mediator.example",
                }],
            })
        }

        // The global form of each request — same bodies with `contextId`
        // dropped — must parse too. `contextId` is the family's only scope
        // selector; if the published schema ever made it required, the
        // global half of all six ops would break with no other signal.
        for (what, parse, global) in [
            (
                "list",
                parses::<specs::vta::did_templates::list::v2_0::Payload> as ParseFn,
                to_v(tpl::list::ListDidTemplatesBody { context_id: None }),
            ),
            (
                "create",
                parses::<specs::vta::did_templates::create::v2_0::Payload> as ParseFn,
                to_v(tpl::create::CreateDidTemplateBody {
                    context_id: None,
                    template: template(),
                }),
            ),
            (
                "get",
                parses::<specs::vta::did_templates::get::v2_0::Payload> as ParseFn,
                to_v(tpl::get::GetDidTemplateBody {
                    context_id: None,
                    name: TPL_NAME.into(),
                }),
            ),
            (
                "update",
                parses::<specs::vta::did_templates::update::v2_0::Payload> as ParseFn,
                to_v(tpl::update::UpdateDidTemplateBody {
                    context_id: None,
                    name: TPL_NAME.into(),
                    template: template(),
                }),
            ),
            (
                "delete",
                parses::<specs::vta::did_templates::delete::v2_0::Payload> as ParseFn,
                to_v(tpl::delete::DeleteDidTemplateBody {
                    context_id: None,
                    name: TPL_NAME.into(),
                }),
            ),
            (
                "render",
                parses::<specs::vta::did_templates::render::v2_0::Payload> as ParseFn,
                to_v(tpl::render::RenderDidTemplateBody {
                    context_id: None,
                    name: TPL_NAME.into(),
                    vars: HashMap::new(),
                }),
            ),
        ] {
            assert!(
                global.get("contextId").is_none(),
                "did-templates/{what}: the global fixture must omit contextId"
            );
            parse(global.clone()).unwrap_or_else(|e| {
                panic!(
                    "did-templates/{what}: global-scope request is not canonical: {e}\n{global:#}"
                )
            });
        }

        t.extend([
            (
                uris::TASK_DID_TEMPLATES_LIST_2_0,
                checked!(
                    specs::vta::did_templates::list::v2_0::Payload,
                    specs::vta::did_templates::list::v2_0::Response,
                    to_v(tpl::list::ListDidTemplatesBody {
                        context_id: Some(CTX.into()),
                    }),
                    to_v(tpl::list::ListDidTemplatesResultBody {
                        templates: vec![record()],
                    })
                ),
            ),
            (
                uris::TASK_DID_TEMPLATES_CREATE_2_0,
                checked!(
                    specs::vta::did_templates::create::v2_0::Payload,
                    specs::vta::did_templates::create::v2_0::Response,
                    to_v(tpl::create::CreateDidTemplateBody {
                        context_id: Some(CTX.into()),
                        template: template(),
                    }),
                    // `handle_create` returns the persisted record itself.
                    to_v(record())
                ),
            ),
            (
                uris::TASK_DID_TEMPLATES_GET_2_0,
                checked!(
                    specs::vta::did_templates::get::v2_0::Payload,
                    specs::vta::did_templates::get::v2_0::Response,
                    to_v(tpl::get::GetDidTemplateBody {
                        context_id: Some(CTX.into()),
                        name: TPL_NAME.into(),
                    }),
                    to_v(record())
                ),
            ),
            (
                uris::TASK_DID_TEMPLATES_UPDATE_2_0,
                checked!(
                    specs::vta::did_templates::update::v2_0::Payload,
                    specs::vta::did_templates::update::v2_0::Response,
                    to_v(tpl::update::UpdateDidTemplateBody {
                        context_id: Some(CTX.into()),
                        name: TPL_NAME.into(),
                        template: template(),
                    }),
                    to_v(record())
                ),
            ),
            (
                uris::TASK_DID_TEMPLATES_DELETE_2_0,
                checked!(
                    specs::vta::did_templates::delete::v2_0::Payload,
                    specs::vta::did_templates::delete::v2_0::Response,
                    to_v(tpl::delete::DeleteDidTemplateBody {
                        context_id: Some(CTX.into()),
                        name: TPL_NAME.into(),
                    }),
                    to_v(tpl::delete::DeleteDidTemplateResultBody {
                        name: TPL_NAME.into(),
                        deleted: true,
                    })
                ),
            ),
            (
                uris::TASK_DID_TEMPLATES_RENDER_2_0,
                checked!(
                    specs::vta::did_templates::render::v2_0::Payload,
                    specs::vta::did_templates::render::v2_0::Response,
                    to_v(tpl::render::RenderDidTemplateBody {
                        context_id: Some(CTX.into()),
                        name: TPL_NAME.into(),
                        vars: [(
                            "SERVICE_ENDPOINT".to_string(),
                            json!("https://mediator.example")
                        )]
                        .into_iter()
                        .collect(),
                    }),
                    to_v(tpl::render::RenderDidTemplateResultBody {
                        document: rendered(),
                    })
                ),
            ),
        ]);
    }

    // ─── passkey-vms (feature-gated like their dispatch arms) ─────
    #[cfg(all(feature = "webvh", feature = "didcomm"))]
    {
        use vta_sdk::protocols::did_management::passkey_vms as pk;
        fn vm() -> pk::PasskeyVerificationMethod {
            pk::PasskeyVerificationMethod {
                id: "did:webvh:scid:vta.example#passkey-1".into(),
                vm_type: "Multikey".into(),
                controller: "did:webvh:scid:vta.example".into(),
                public_key_multibase: "z6MkPasskey".into(),
                webauthn_credential_id: "b64u-cred-id".into(),
                webauthn_transports: vec!["internal".into()],
                label: Some("YubiKey".into()),
            }
        }
        t.extend([
            (
                uris::TASK_PASSKEY_VMS_ENROLL_CHALLENGE_0_1,
                checked!(
                    specs::vta::passkey_vms::enroll_challenge::v0_1::Payload,
                    specs::vta::passkey_vms::enroll_challenge::v0_1::Response,
                    to_v(pk::EnrollPasskeyChallengeBody {
                        did: "did:webvh:scid:vta.example".into(),
                        label: Some("YubiKey".into()),
                    }),
                    to_v(pk::EnrollPasskeyChallengeResponse {
                        ceremony_id: "cer-1".into(),
                        challenge: "b64u-challenge".into(),
                        rp_id: "vta.example".into(),
                        rp_name: "VTA".into(),
                        user_handle: "b64u-handle".into(),
                        user_name: "operator".into(),
                        user_display_name: "Operator".into(),
                        timeout_ms: Some(60_000),
                    })
                ),
            ),
            (
                uris::TASK_PASSKEY_VMS_ENROLL_SUBMIT_0_1,
                checked!(
                    specs::vta::passkey_vms::enroll_submit::v0_1::Payload,
                    specs::vta::passkey_vms::enroll_submit::v0_1::Response,
                    to_v(pk::EnrollPasskeySubmitBody {
                        did: "did:webvh:scid:vta.example".into(),
                        ceremony_id: "cer-1".into(),
                        credential_id: "b64u-cred-id".into(),
                        public_key_multibase: "z6MkPasskey".into(),
                        cose_algorithm: -8,
                        attestation_object: "b64u-atto".into(),
                        client_data_json: "b64u-cdj".into(),
                        authenticator_data: "b64u-authdata".into(),
                        transports: vec!["internal".into()],
                        label: Some("YubiKey".into()),
                    }),
                    to_v(pk::EnrollPasskeySubmitResponse {
                        verification_method: vm(),
                        webvh_version: "2-zVer".into(),
                    })
                ),
            ),
            (
                uris::TASK_PASSKEY_VMS_LIST_0_1,
                checked!(
                    specs::vta::passkey_vms::list::v0_1::Payload,
                    specs::vta::passkey_vms::list::v0_1::Response,
                    to_v(pk::ListPasskeyVmsBody {
                        did: "did:webvh:scid:vta.example".into(),
                    }),
                    to_v(pk::ListPasskeyVmsResponse {
                        verification_methods: vec![vm()],
                    })
                ),
            ),
            (
                uris::TASK_PASSKEY_VMS_REVOKE_0_1,
                checked!(
                    specs::vta::passkey_vms::revoke::v0_1::Payload,
                    specs::vta::passkey_vms::revoke::v0_1::Response,
                    to_v(pk::RevokePasskeyVmBody {
                        did: "did:webvh:scid:vta.example".into(),
                        fragment: "passkey-1".into(),
                    }),
                    to_v(pk::RevokePasskeyVmResponse::default())
                ),
            ),
        ]);
    }

    // ─── provision/integration (webvh-gated like its dispatch arm) ─
    #[cfg(feature = "webvh")]
    {
        use vta_sdk::provision_integration::http as prov;
        // The request is consumed, not produced, by this service, and the
        // VP-framed `request` member is signed material — the fixture is the
        // canonical document shape (0.2 camelCase, `templateBootstrap` ask
        // tag) that `ProvisionIntegrationRequest` dual-accepts.
        let request = json!({
            "request": {
                "@context": ["https://www.w3.org/ns/credentials/v2",
                             "https://openvtc.org/contexts/bootstrap-v1"],
                "type": ["VerifiablePresentation", "BootstrapRequest"],
                "id": "urn:uuid:9be7",
                "holder": "did:key:z6MkAgentSetup",
                "nonce": "AAAAAAAAAAAAAAAAAAAAAA",
                "validUntil": TS,
                "ask": { "type": "templateBootstrap", "template": { "name": "didcomm-mediator" } },
                "proof": {
                    "type": "DataIntegrityProof",
                    "cryptosuite": "eddsa-jcs-2022",
                    "created": TS,
                    "verificationMethod": "did:key:z6MkAgentSetup#z6MkAgentSetup",
                    "proofPurpose": "authentication",
                    "proofValue": "z5TvSig",
                },
            },
            "context": "agents",
            "assertion": "didSigned",
            "vcValiditySeconds": 3600,
            "createContext": true,
        });
        let response = to_v(prov::ProvisionIntegrationResponse {
            bundle: "-----BEGIN VTA SEALED BUNDLE-----".into(),
            // A real `DigestMultibase`, not a placeholder: 0.3 constrains the
            // encoding, and a fixture that cannot satisfy the pattern proves
            // nothing about what this producer emits.
            digest_multibase: Some(DIGEST_MULTIBASE.into()),
            summary: prov::ProvisionSummary {
                client_did: "did:key:z6MkAgentSetup".into(),
                admin_did: "did:key:z6MkAdmin".into(),
                admin_rolled_over: true,
                integration_did: Some("did:webvh:scid:agents.example".into()),
                template_name: Some("didcomm-mediator".into()),
                template_kind: Some("mediator".into()),
                admin_template_name: Some("vta-admin".into()),
                bundle_id_hex: "0123456789abcdef0123456789abcdef".into(),
                secret_count: 2,
                output_count: 1,
                webvh_server_id: Some("srv-1".into()),
                context_created: true,
            },
        });
        // Prove the canonical request also parses into our consumer type —
        // dual-accept means the fixture must satisfy both sides.
        serde_json::from_value::<prov::ProvisionIntegrationRequest>(request.clone())
            .expect("canonical provision request parses into ProvisionIntegrationRequest");
        t.push((
            uris::TASK_PROVISION_INTEGRATION_0_3,
            checked!(
                specs::provision::integration::v0_3::Payload,
                specs::provision::integration::v0_3::Response,
                request,
                response
            ),
        ));
    }

    // ─── vta/webvh/dids/update (webvh-gated like its dispatch arm) ─
    #[cfg(feature = "webvh")]
    {
        use vta_sdk::protocols::did_management::update::{
            UpdateDidWebvhBody, UpdateDidWebvhResultBody,
        };
        t.push((
            uris::TASK_WEBVH_DIDS_UPDATE_1_0,
            checked!(
                specs::vta::webvh::dids::update::v1_0::Payload,
                specs::vta::webvh::dids::update::v1_0::Response,
                // Built by the producer's own shaping, not transcribed. The
                // comment here used to name a `UpdateDidWithDid` type that does
                // not exist — the shape lives in `flatten_with_did`, which is
                // now public for exactly this reason.
                //
                // **Deliberately a one-member update.** `--label x` and nothing
                // else is the exact invocation that could not run in #895: six
                // unset members each went out as `null` and the maintainer
                // refused all six. A witness that sets most members would leave
                // almost nothing unset and so would not exercise the defect
                // this task has already shipped. Setting one member and leaving
                // the rest is what keeps the other six honest.
                vta_sdk::client::flatten_with_did(
                    "did:webvh:scid:vta.example",
                    &UpdateDidWebvhBody {
                        label: Some("rotate after audit".into()),
                        ..Default::default()
                    },
                )
                .expect("the producer's own shaping succeeds"),
                to_v(UpdateDidWebvhResultBody {
                    did: "did:webvh:scid:vta.example".into(),
                    new_version_id: "2-zVer".into(),
                    new_scid: "scid".into(),
                    new_log_entry: "{}".into(),
                    update_keys_count: 1,
                    pre_rotation_key_count: 2,
                    serverless: false,
                })
            ),
        ));
    }

    // ─── vta/contexts + vta/webvh ────────────────────────────────────
    //
    // These 22 were dispatched against unpublished URIs until trust-tasks #240
    // specified them, so `schema_index` could not resolve a schema and the
    // sweep skipped them entirely — they sat in `UNSPECCED_DISPATCHED_URIS`
    // instead. Taking trust-tasks-rs 0.11 makes them resolvable, so they move
    // out of that debt list and into here, which is the whole point of the
    // list: the debt shrinks monotonically as specs land.
    //
    // The witnesses are written as raw JSON rather than through the producer
    // types on purpose. Two of those producer types were WRONG when the schemas
    // arrived — `CreateDidWebvhRequest` emitted snake_case and `update_context`
    // emitted `null` for unset members — so building a witness from them would
    // have encoded the defect and passed. The schema is the authority here.
    for (uri, req, resp) in webvh_and_context_witnesses() {
        t.push((
            uri,
            Conformance::Checked(Witness {
                request: req.0,
                parse_request: req.1,
                validate_request: req.2,
                response: resp.0,
                parse_response: resp.1,
            }),
        ));
    }

    t
}

type ReqParts = (Value, ParseFn, ValidateFn);
type RespParts = (Value, ParseFn);

/// Witnesses for the families trust-tasks #240 published.
///
/// Each request carries exactly its schema's required members and nothing else:
/// an over-specified witness leaves little unset and so would not exercise the
/// `null`-for-absent class that has already shipped twice (#895, #919).
fn webvh_and_context_witnesses() -> Vec<(&'static str, ReqParts, RespParts)> {
    use specs::vta::{contexts as ctx, services as svc, webvh as wv};

    let context_record = || {
        json!({
            "id": "personal", "name": "Personal", "basePath": "personal",
            "createdAt": "2026-08-19T09:00:00Z", "updatedAt": "2026-08-19T09:00:00Z"
        })
    };
    let did_record = || {
        json!({
            "did": "did:webvh:QmScid:host.example:alice", "serverId": "prod",
            "mnemonic": "alice", "scid": "QmScid", "contextId": "personal",
            "portable": true, "logEntryCount": 4,
            "createdAt": "2026-08-19T09:00:00Z", "updatedAt": "2026-08-19T09:00:00Z"
        })
    };
    let server_record = || {
        json!({
            "id": "prod", "did": "did:web:host.example",
            "createdAt": "2026-08-19T09:00:00Z", "updatedAt": "2026-08-19T09:00:00Z"
        })
    };
    let did = "did:webvh:QmScid:host.example:alice";
    let mutation_result = || {
        json!({
            "logEntryVersionId": "4-zQmLogEntry",
            "effectiveAt": "2026-08-20T09:00:00Z",
            "vtaDid": "did:webvh:QmAgent:vta.example",
            "serverless": false
        })
    };

    let mut v: Vec<(&'static str, ReqParts, RespParts)> = vec![
        (
            uris::TASK_CONTEXTS_LIST_1_0,
            (
                json!({}),
                parses::<ctx::list::v1_0::Payload>,
                validates::<ctx::list::v1_0::Payload>,
            ),
            (
                json!({ "contexts": [context_record()] }),
                parses::<ctx::list::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_CONTEXTS_GET_1_0,
            (
                json!({ "id": "personal" }),
                parses::<ctx::get::v1_0::Payload>,
                validates::<ctx::get::v1_0::Payload>,
            ),
            (context_record(), parses::<ctx::get::v1_0::Response>),
        ),
        (
            uris::TASK_CONTEXTS_CREATE_1_0,
            (
                json!({ "id": "personal", "name": "Personal" }),
                parses::<ctx::create::v1_0::Payload>,
                validates::<ctx::create::v1_0::Payload>,
            ),
            (context_record(), parses::<ctx::create::v1_0::Response>),
        ),
        (
            uris::TASK_CONTEXTS_UPDATE_1_0,
            (
                json!({ "id": "personal" }),
                parses::<ctx::update::v1_0::Payload>,
                validates::<ctx::update::v1_0::Payload>,
            ),
            (context_record(), parses::<ctx::update::v1_0::Response>),
        ),
        (
            uris::TASK_CONTEXTS_UPDATE_DID_1_0,
            (
                json!({ "id": "personal", "did": did }),
                parses::<ctx::update_did::v1_0::Payload>,
                validates::<ctx::update_did::v1_0::Payload>,
            ),
            (context_record(), parses::<ctx::update_did::v1_0::Response>),
        ),
        (
            uris::TASK_CONTEXTS_PREVIEW_DELETE_1_0,
            (
                json!({ "id": "personal" }),
                parses::<ctx::preview_delete::v1_0::Payload>,
                validates::<ctx::preview_delete::v1_0::Payload>,
            ),
            (
                // These four are arrays of ids, not counts. The names read like
                // tallies and do not behave like them — precisely what a witness
                // written by hand gets wrong and the round-trip check exists to catch.
                json!({ "id": "personal", "keys": ["k-1"], "webvhDids": [],
                 "aclEntriesRemoved": ["did:key:zRemoved"], "aclEntriesUpdated": [] }),
                parses::<ctx::preview_delete::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_CONTEXTS_DELETE_1_0,
            (
                json!({ "id": "personal" }),
                parses::<ctx::delete::v1_0::Payload>,
                validates::<ctx::delete::v1_0::Payload>,
            ),
            (
                json!({ "id": "personal", "deleted": true }),
                parses::<ctx::delete::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_DIDS_LIST_1_0,
            (
                json!({}),
                parses::<wv::dids::list::v1_0::Payload>,
                validates::<wv::dids::list::v1_0::Payload>,
            ),
            (
                json!({ "dids": [did_record()] }),
                parses::<wv::dids::list::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_DIDS_GET_1_0,
            (
                json!({ "did": did }),
                parses::<wv::dids::get::v1_0::Payload>,
                validates::<wv::dids::get::v1_0::Payload>,
            ),
            (
                json!({ "record": did_record() }),
                parses::<wv::dids::get::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_DIDS_CREATE_1_0,
            (
                json!({ "contextId": "personal" }),
                parses::<wv::dids::create::v1_0::Payload>,
                validates::<wv::dids::create::v1_0::Payload>,
            ),
            (
                json!({ "did": did, "contextId": "personal", "scid": "QmScid", "portable": true,
                  "signingKeyId": "k-sign", "kaKeyId": "k-ka", "preRotationKeyCount": 2,
                  "createdAt": "2026-08-19T09:00:00Z" }),
                parses::<wv::dids::create::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_DIDS_DELETE_1_0,
            (
                json!({ "did": did }),
                parses::<wv::dids::delete::v1_0::Payload>,
                validates::<wv::dids::delete::v1_0::Payload>,
            ),
            (
                json!({ "did": did, "deleted": true }),
                parses::<wv::dids::delete::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_DIDS_ROTATE_KEYS_1_0,
            (
                json!({ "did": did }),
                parses::<wv::dids::rotate_keys::v1_0::Payload>,
                validates::<wv::dids::rotate_keys::v1_0::Payload>,
            ),
            (
                json!({ "did": did, "newVersionId": "2-zVer", "newScid": "QmScid", "newLogEntry": "{}",
                  "updateKeysCount": 1, "preRotationKeyCount": 2, "serverless": false }),
                parses::<wv::dids::rotate_keys::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_DIDS_REGISTER_WITH_SERVER_1_0,
            (
                json!({ "did": did, "serverId": "prod" }),
                parses::<wv::dids::register_with_server::v1_0::Payload>,
                validates::<wv::dids::register_with_server::v1_0::Payload>,
            ),
            (
                json!({ "did": did, "serverId": "prod", "logEntryCount": 4 }),
                parses::<wv::dids::register_with_server::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_SERVERS_LIST_1_0,
            (
                json!({}),
                parses::<wv::servers::list::v1_0::Payload>,
                validates::<wv::servers::list::v1_0::Payload>,
            ),
            (
                json!({ "servers": [server_record()] }),
                parses::<wv::servers::list::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_SERVERS_REGISTER_1_0,
            (
                json!({ "id": "prod" }),
                parses::<wv::servers::register::v1_0::Payload>,
                validates::<wv::servers::register::v1_0::Payload>,
            ),
            (
                server_record(),
                parses::<wv::servers::register::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_SERVERS_RETIRE_ORPHAN_0_1,
            (
                // From the typed body rather than hand-written JSON, so this
                // asserts that *our* wire type conforms — the property that
                // catches drift, rather than merely that the schema accepts a
                // shape written by hand to match it.
                to_v(RetireOrphanSlotBody {
                    server_id: "primary-host".into(),
                    slot_id: "attract-case".into(),
                    expected_did: Some(
                        "did:webvh:QmZ4rT9xK2mN8vB5cD1sA7wE3fH6jL0pQ:did.example.com:attract-case"
                            .into(),
                    ),
                    reason: Some("orphaned by a create whose reply was lost".into()),
                }),
                parses::<wv::servers::retire_orphan::v0_1::Payload>,
                validates::<wv::servers::retire_orphan::v0_1::Payload>,
            ),
            (
                to_v(RetireOrphanSlotResultBody {
                    server_id: "primary-host".into(),
                    slot_id: "attract-case".into(),
                    retired: true,
                    did: Some(
                        "did:webvh:QmZ4rT9xK2mN8vB5cD1sA7wE3fH6jL0pQ:did.example.com:attract-case"
                            .into(),
                    ),
                }),
                parses::<wv::servers::retire_orphan::v0_1::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_SERVERS_REMOVE_1_0,
            (
                json!({ "id": "prod" }),
                parses::<wv::servers::remove::v1_0::Payload>,
                validates::<wv::servers::remove::v1_0::Payload>,
            ),
            (
                json!({ "id": "prod", "removed": true }),
                parses::<wv::servers::remove::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_AGENT_NAME_SET_1_0,
            (
                json!({ "did": did, "name": "alice" }),
                parses::<wv::agent_name::set::v1_0::Payload>,
                validates::<wv::agent_name::set::v1_0::Payload>,
            ),
            (
                json!({ "did": did, "name": "alice", "enabled": true }),
                parses::<wv::agent_name::set::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_AGENT_NAME_REMOVE_1_0,
            (
                json!({ "did": did, "name": "alice" }),
                parses::<wv::agent_name::remove::v1_0::Payload>,
                validates::<wv::agent_name::remove::v1_0::Payload>,
            ),
            (
                json!({ "did": did, "name": "alice", "enabled": false }),
                parses::<wv::agent_name::remove::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_AGENT_NAME_LIST_1_0,
            (
                json!({ "did": did }),
                parses::<wv::agent_name::list::v1_0::Payload>,
                validates::<wv::agent_name::list::v1_0::Payload>,
            ),
            (
                json!({ "did": did, "names": [{ "name": "alice", "enabled": true }] }),
                parses::<wv::agent_name::list::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_AGENT_NAME_CHECK_1_0,
            (
                json!({ "did": did, "name": "alice" }),
                parses::<wv::agent_name::check::v1_0::Payload>,
                validates::<wv::agent_name::check::v1_0::Payload>,
            ),
            (
                json!({ "name": "alice", "domain": "host.example", "available": true, "reserved": false }),
                parses::<wv::agent_name::check::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_AGENT_NAME_ENABLE_1_0,
            (
                json!({ "did": did, "name": "alice" }),
                parses::<wv::agent_name::enable::v1_0::Payload>,
                validates::<wv::agent_name::enable::v1_0::Payload>,
            ),
            (
                json!({ "did": did, "name": "alice", "enabled": true }),
                parses::<wv::agent_name::enable::v1_0::Response>,
            ),
        ),
        (
            uris::TASK_WEBVH_AGENT_NAME_DISABLE_1_0,
            (
                json!({ "did": did, "name": "alice" }),
                parses::<wv::agent_name::disable::v1_0::Payload>,
                validates::<wv::agent_name::disable::v1_0::Payload>,
            ),
            (
                json!({ "did": did, "name": "alice", "enabled": false }),
                parses::<wv::agent_name::disable::v1_0::Response>,
            ),
        ),
    ];

    // ─── vta/services ────────────────────────────────────────────────
    //
    // Extended rather than inlined because these are gated with their dispatch
    // entries. The sweep derives its census from what is actually dispatched and
    // requires the two to agree in BOTH directions, so a witness for a task that
    // is configured out fails it exactly as a missing one does.
    #[cfg(feature = "webvh")]
    {
        // Typed explicitly: without it the array literal takes its element type
        // from the first entry, and each `parses::<T>` is a distinct fn item
        // rather than the `ParseFn` pointer the alias expects.
        let services: [(&'static str, ReqParts, RespParts); 8] = [
            (
                uris::TASK_SERVICES_LIST_1_0,
                (
                    json!({}),
                    parses::<svc::list::v1_0::Payload>,
                    validates::<svc::list::v1_0::Payload>,
                ),
                (
                    json!({ "services": [{ "kind": "rest", "enabled": true, "url": "https://vta.example/api" }] }),
                    parses::<svc::list::v1_0::Response>,
                ),
            ),
            (
                uris::TASK_SERVICES_GET_1_0,
                (
                    json!({ "service": "didcomm" }),
                    parses::<svc::get::v1_0::Payload>,
                    validates::<svc::get::v1_0::Payload>,
                ),
                (
                    json!({ "state": { "kind": "didcomm", "enabled": true, "mediatorDid": "did:web:mediator.example" } }),
                    parses::<svc::get::v1_0::Response>,
                ),
            ),
            (
                uris::TASK_SERVICES_ENABLE_1_0,
                (
                    json!({ "service": "rest", "config": { "url": "https://vta.example/api" } }),
                    parses::<svc::enable::v1_0::Payload>,
                    validates::<svc::enable::v1_0::Payload>,
                ),
                (
                    json!({ "result": mutation_result() }),
                    parses::<svc::enable::v1_0::Response>,
                ),
            ),
            (
                uris::TASK_SERVICES_UPDATE_1_0,
                (
                    json!({ "service": "tsp", "config": { "mediatorDid": "did:web:mediator.example" } }),
                    parses::<svc::update::v1_0::Payload>,
                    validates::<svc::update::v1_0::Payload>,
                ),
                (
                    json!({ "result": mutation_result() }),
                    parses::<svc::update::v1_0::Response>,
                ),
            ),
            (
                uris::TASK_SERVICES_DISABLE_1_0,
                (
                    json!({ "service": "didcomm" }),
                    parses::<svc::disable::v1_0::Payload>,
                    validates::<svc::disable::v1_0::Payload>,
                ),
                (
                    json!({ "result": mutation_result() }),
                    parses::<svc::disable::v1_0::Response>,
                ),
            ),
            (
                uris::TASK_SERVICES_ROLLBACK_1_0,
                (
                    json!({ "service": "rest" }),
                    parses::<svc::rollback::v1_0::Payload>,
                    validates::<svc::rollback::v1_0::Payload>,
                ),
                (
                    json!({ "result": { "kind": "noOp", "serverless": false } }),
                    parses::<svc::rollback::v1_0::Response>,
                ),
            ),
            (
                uris::TASK_SERVICES_DRAIN_LIST_1_0,
                (
                    json!({}),
                    parses::<svc::drain::list::v1_0::Payload>,
                    validates::<svc::drain::list::v1_0::Payload>,
                ),
                (
                    json!({ "entries": [{ "mediatorDid": "did:web:old-mediator.example",
                                    "endpoint": "https://old-mediator.example/didcomm",
                                    "drainsUntil": "2026-08-20T21:00:00Z" }] }),
                    parses::<svc::drain::list::v1_0::Response>,
                ),
            ),
            (
                uris::TASK_SERVICES_DRAIN_CANCEL_1_0,
                (
                    json!({ "mediatorDid": "did:web:old-mediator.example" }),
                    parses::<svc::drain::cancel::v1_0::Payload>,
                    validates::<svc::drain::cancel::v1_0::Payload>,
                ),
                (
                    json!({ "mediatorDid": "did:web:old-mediator.example" }),
                    parses::<svc::drain::cancel::v1_0::Response>,
                ),
            ),
        ];
        v.extend(services);
    }

    v
}

// ─── The sweep ───────────────────────────────────────────────────────────

/// Coverage: the witness table and the derived census agree exactly, in both
/// directions. A sibling stream rebinding a URI onto a published spec lands
/// here first — that is the point.
#[test]
fn every_published_dispatched_uri_has_a_witness() {
    let expected = resolved_uris();
    let mut covered = BTreeSet::new();
    for (uri, _) in table() {
        assert!(
            covered.insert(uri),
            "duplicate witness entry for {uri} — one entry per URI"
        );
    }

    let missing: Vec<_> = expected.difference(&covered).collect();
    assert!(
        missing.is_empty(),
        "these dispatched URIs are published in the registry but have no \
         conformance witness:\n  {}\n\nAdd a `checked!` entry (request + \
         response built from the slice's wire types), or — only for a real, \
         understood non-conformance — a `KnownDrift` entry with a reason. If \
         this fired after a rebase, a sibling consolidation stream rebound a \
         task onto a published URI; the sweep is asking for its witness.",
        missing
            .iter()
            .map(|s| s.to_string())
            .collect::<Vec<_>>()
            .join("\n  ")
    );

    let stale: Vec<_> = covered.difference(&expected).collect();
    assert!(
        stale.is_empty(),
        "these witnesses cover URIs that are no longer both dispatched and \
         published:\n  {}\n\nRemove them (or fix the constant they name).",
        stale
            .iter()
            .map(|s| s.to_string())
            .collect::<Vec<_>>()
            .join("\n  ")
    );
}

/// Correctness: every witness's request parses as the generated `Payload`
/// and its response as the generated `Response` — and the check has teeth:
/// a drifted (unknown-member) form of each MUST be rejected, so a witness
/// cannot pass vacuously against a schema that accepts anything.
#[test]
fn every_witnessed_task_round_trips_through_its_generated_types() {
    let mut drift: Vec<String> = Vec::new();

    for (uri, conformance) in table() {
        let w = match conformance {
            Conformance::Checked(w) => w,
            Conformance::KnownDrift(reason) => {
                assert!(
                    !reason.trim().is_empty(),
                    "{uri}: KnownDrift entries must state the drift"
                );
                drift.push(format!("{uri}: {reason}"));
                continue;
            }
        };

        (w.parse_request)(w.request.clone())
            .unwrap_or_else(|e| panic!("{uri}: request is not canonical: {e}\n{:#}", w.request));
        (w.parse_response)(w.response.clone())
            .unwrap_or_else(|e| panic!("{uri}: response is not canonical: {e}\n{:#}", w.response));

        // And the request against the schema itself — the check the dispatch
        // spine runs on arrival. Parsing is strictly weaker: serde reads
        // `null` into an `Option<T>` where the schema types the member
        // `"string"` and refuses it. See [`validates`].
        //
        // Request-only: `trust-tasks-rs` codegen emits `ValidatedPayload` for
        // `Payload` but not for `Response` (0.4 and 0.5 alike), so there is no
        // embedded response schema to check against. The response side keeps
        // its parse check; closing that half needs the codegen to emit the
        // response schema, which is an upstream change.
        (w.validate_request)(&w.request).unwrap_or_else(|e| {
            panic!(
                "{uri}: request fails its own payload schema: {e}\n{:#}",
                w.request
            )
        });

        // Teeth: an unknown member must be rejected on both sides. The
        // generated types carry `deny_unknown_fields` wherever the spec closes
        // the object — `additionalProperties: false`, or `unevaluatedProperties`
        // on a composed one, which trust-tasks-rs 0.17.2 also honours. A URI for
        // which this stops holding needs a KnownDrift-style review, not a silent
        // pass: the check below injects an unknown member, so a type that
        // accepts it makes the whole witness pass vacuously.
        for (side, value, parse) in [
            ("request", &w.request, w.parse_request),
            ("response", &w.response, w.parse_response),
        ] {
            let mut drifted = value.clone();
            drifted
                .as_object_mut()
                .unwrap_or_else(|| panic!("{uri}: {side} witness must be a JSON object"))
                .insert("__conformance_sweep_drift".into(), json!(true));
            assert!(
                parse(drifted).is_err(),
                "{uri}: the generated {side} type accepted an unknown member — \
                 this witness can pass vacuously and proves nothing"
            );
        }
    }

    // Debt is visible in the test output every run, not buried in a table.
    for line in &drift {
        eprintln!("KNOWN DRIFT (follow-up issue required): {line}");
    }
}