did-hosting-control 0.10.1

//! DIDComm messaging for the control plane.
//!
//! **Inbound:** Uses the `affinidi-messaging-didcomm-service` framework for
//! mediator connection, message dispatch, and response handling. Handles
//! the full VTA provisioning protocol (did/request, did/publish, etc.)
//! as well as sync acknowledgements from servers.
//!
//! **Outbound:** Sync push messages are sent via `server_push.rs` using the
//! shared `DIDCommService` — no separate ATM connection needed.

use affinidi_messaging_didcomm::Message;
use affinidi_messaging_didcomm_service::{
    DIDCommResponse, DIDCommServiceError, Extension, HandlerContext, MESSAGE_PICKUP_STATUS_TYPE,
    MessagePolicy, MiddlewareResult, Next, Router, TRUST_PING_TYPE, handler_fn, ignore_handler,
    middleware_fn, trust_ping_handler,
};
use did_hosting_common::didcomm_types::*;
use did_hosting_common::server::problem_report::log_problem_report;
use did_hosting_common::server::trust_tasks::ProofRule;
use serde_json::{Value, json};
use tracing::{debug, info, warn};

use crate::acl::check_acl;
use crate::server::AppState;
use crate::server_push;

// ---------------------------------------------------------------------------
// Inbound router (framework-managed)
// ---------------------------------------------------------------------------

/// Build the DIDComm router for the control plane's inbound messages.
///
/// Handles the full VTA provisioning protocol (authenticate, did/request,
/// did/publish, etc.) as well as sync acknowledgements from servers.
pub fn build_control_router(state: AppState) -> Result<Router, DIDCommServiceError> {
    Ok(Router::new()
        .extension(state)
        // Standard DIDComm
        .route(TRUST_PING_TYPE, handler_fn(trust_ping_handler))?
        .route(MESSAGE_PICKUP_STATUS_TYPE, handler_fn(ignore_handler))?
        // Every DID-management, auth and infrastructure operation arrives as a
        // signed Trust Task document in the trust-task envelope below; the
        // bare-message routes that used to sit here authorised on the
        // transport's report of the sender alone and have been removed. A peer
        // still sending one gets `handle_fallback`'s problem report naming the
        // type, not a silent timeout.
        // Trust Tasks envelope (v0.7.0+) — routes the five `acl/*`
        // ops and `trust-task-discovery` through the same handlers
        // the HTTPS transport hits at `POST /api/trust-tasks`.
        .route(
            trust_tasks_didcomm::ENVELOPE_TYPE,
            handler_fn(handle_trust_tasks_envelope),
        )?
        .fallback(handler_fn(handle_fallback))
        .layer(
            MessagePolicy::new()
                .require_encrypted(true)
                .require_sender_did(true),
        )
        .layer(middleware_fn(filtered_request_logging)))
}

/// Request logging middleware that silences noisy health/stats messages.
async fn filtered_request_logging(
    ctx: HandlerContext,
    message: Message,
    meta: affinidi_messaging_didcomm::UnpackMetadata,
    next: Next,
) -> MiddlewareResult {
    const QUIET: &[&str] = &[
        MSG_HEALTH_PING,
        MSG_HEALTH_PONG,
        MSG_STATS_SYNC,
        MSG_STATS_ACK,
        MESSAGE_PICKUP_STATUS_TYPE,
    ];

    let msg_type = message.typ.clone();
    let result = next.run(ctx, message, meta).await;

    if !QUIET.iter().any(|t| msg_type == *t) {
        let status = match &result {
            Ok(Some(_)) => "ok(response)",
            Ok(None) => "ok(empty)",
            Err(_) => "error",
        };
        info!(message_type = %msg_type, status, "DIDComm request processed");
    }

    result
}
// ---------------------------------------------------------------------------
// DID operation dispatch
// ---------------------------------------------------------------------------

/// Project a stored [`DidRecord`] into the canonical `DidRecord` wire
/// shape used by the `did-management/did/*` Trust Task family
/// (camelCase keys, RFC3339 timestamps).
///
/// `did_url` is the resolvable location of the DID log document; it is
/// stable from the initial reservation (`versionCount: 0`) and lets the
/// owner know where to publish and where resolvers will fetch. `didId`
/// is only meaningful once a log entry exists, so it is emitted solely
/// when `versionCount > 0`.
pub(crate) fn spec_did_record_json(
    record: &did_hosting_common::did_ops::DidRecord,
    did_url: &str,
) -> Value {
    let rfc3339 = |secs: u64| {
        chrono::DateTime::<chrono::Utc>::from_timestamp(secs as i64, 0)
            .unwrap_or_default()
            .to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
    };
    let mut rec = serde_json::Map::new();
    rec.insert("mnemonic".into(), json!(record.mnemonic));
    rec.insert("owner".into(), json!(record.owner));
    rec.insert("createdAt".into(), json!(rfc3339(record.created_at)));
    rec.insert("updatedAt".into(), json!(rfc3339(record.updated_at)));
    rec.insert("versionCount".into(), json!(record.version_count));
    rec.insert("method".into(), json!(record.method));
    rec.insert("disabled".into(), json!(record.disabled));
    rec.insert("didUrl".into(), json!(did_url));
    if !record.domain.is_empty() {
        rec.insert("domain".into(), json!(record.domain));
    }
    if let Some(did_id) = record.did_id.as_ref().filter(|_| record.version_count > 0) {
        rec.insert("didId".into(), json!(did_id));
    }
    // The agent-name registry, so the four `agent-name/*` verbs are
    // self-describing: without it a client has to follow every `set` or
    // `disable` with an authenticated `GET /api/dids/{mnemonic}` just to learn
    // what its own call did. A *parked* name is invisible in the DID document
    // by design, so the response is the only place the caller can see it.
    //
    // Emitted only when non-empty, matching `domain`/`didId` above: a DID with
    // no names produces a byte-identical response to before, which keeps this
    // additive for the `did-management/did/*` family that shares this shape.
    //
    // The shared `DidRecord` schema is closed, so the names travel under this
    // host's extension namespace rather than as a member it does not define.
    if !record.agent_names.is_empty() {
        rec.insert(
            "ext".into(),
            json!({
                did_hosting_common::server::trust_tasks::ext::WEBVH_EXT_KEY: {
                    "agentNames": record.agent_names,
                }
            }),
        );
    }
    Value::Object(rec)
}

// ---------------------------------------------------------------------------
// Stats sync (server → control plane)
// ---------------------------------------------------------------------------

/// Tracks the last accepted sequence number per server DID.
static LAST_SEQ: std::sync::LazyLock<std::sync::RwLock<std::collections::HashMap<String, u64>>> =
    std::sync::LazyLock::new(|| std::sync::RwLock::new(std::collections::HashMap::new()));

/// Check and update the idempotency sequence for a server.
///
/// Returns `true` if the payload should be accepted (new seq), or `false`
/// if it's stale/replayed.
fn accept_seq(server_did: &str, seq: u64) -> bool {
    // seq=0 means server restart — always accept
    if seq > 0
        && let Ok(map) = LAST_SEQ.read()
        && let Some(&last) = map.get(server_did)
        && seq <= last
    {
        return false;
    }

    if let Ok(mut map) = LAST_SEQ.write() {
        map.insert(server_did.to_string(), seq);
    }
    true
}

/// Transport-agnostic core of stats sync, for the `.../server/stats-sync/0.1`
/// trust task over DIDComm **or** TSP.
///
/// `signer` is the document's proven issuer (see
/// [`dispatch_trust_task_doc`]), never a transport's report. Returns the ack
/// body — `accepted`, or `skipped` for a stale sequence — or a rejection when
/// the signer is not a Service.
pub(crate) async fn do_stats_sync(
    state: &AppState,
    signer: &str,
    body: &Value,
) -> Result<Value, InfraRejection> {
    // Require Service role — matches REST `/api/control/stats` which is gated
    // on ServiceAuth. An Owner-role DID must not be able to write per-DID
    // stats deltas: doing so would let any tenant tamper with another
    // server's resolved/update counters.
    if !matches!(
        check_acl(&state.acl_ks, signer).await,
        Ok(crate::acl::Role::Service)
    ) {
        warn!(did = signer, "stats sync rejected: Service role required");
        return Err(InfraRejection::new(
            "e.p.stats.unauthorized",
            "service role required",
        ));
    }

    // Bind the payload to the transport-proven signer, as the REST handler
    // binds it to the JWT. `server_did` keys the replay window, so accepting a
    // foreign one would let one server advance — and so suppress — another's.
    let server_did = body
        .get("server_did")
        .and_then(|v| v.as_str())
        .unwrap_or(signer);
    if server_did != signer {
        warn!(
            did = signer,
            server_did, "stats sync rejected: server_did does not match signer"
        );
        return Err(InfraRejection::new(
            "e.p.stats.sender_mismatch",
            "server_did does not match signer",
        ));
    }

    let seq = body.get("seq").and_then(|v| v.as_u64()).unwrap_or(0);

    // Idempotency check
    if !accept_seq(server_did, seq) {
        debug!(server_did, seq, "stats sync: stale sequence");
        return Ok(json!({ "status": "skipped", "reason": "stale_seq" }));
    }

    let deltas = body
        .get("did_deltas")
        .and_then(|v| v.as_array())
        .map(Vec::as_slice)
        .unwrap_or_default();
    const MAX_DELTAS_PER_SYNC: usize = 10_000;
    if deltas.len() > MAX_DELTAS_PER_SYNC {
        warn!(
            did = signer,
            count = deltas.len(),
            "stats sync rejected: too many deltas"
        );
        return Err(InfraRejection::new(
            "e.p.stats.too-large",
            "too many deltas in one sync",
        ));
    }
    for d in deltas {
        let mnemonic = d.get("mnemonic").and_then(|v| v.as_str()).unwrap_or("");
        if mnemonic.is_empty() {
            continue;
        }
        let resolve_delta = d.get("resolve_delta").and_then(|v| v.as_u64()).unwrap_or(0);
        let update_delta = d.get("update_delta").and_then(|v| v.as_u64()).unwrap_or(0);
        let last_resolved_at = d.get("last_resolved_at").and_then(|v| v.as_u64());
        let last_updated_at = d.get("last_updated_at").and_then(|v| v.as_u64());

        state.stats_collector.record_deltas(
            mnemonic,
            resolve_delta,
            update_delta,
            last_resolved_at,
            last_updated_at,
        );
    }

    debug!(
        server_did,
        seq,
        delta_count = deltas.len(),
        "stats sync accepted"
    );

    Ok(json!({ "status": "accepted" }))
}

// ---------------------------------------------------------------------------
// Health pong handler (server → control plane)
// ---------------------------------------------------------------------------

/// Transport-agnostic core of health-pong handling, for the
/// `.../server/health/0.1#response` trust task over DIDComm or TSP.
///
/// `signer` is the pong's proven issuer. Only a Service-role DID may mark an
/// instance Active: the registry's liveness view decides which servers the
/// dashboard reports healthy, so it must not be writable by any DID that can
/// reach the mediator.
///
/// A pong is terminal: it never produces a reply.
pub(crate) async fn do_health_pong(state: &AppState, signer: &str, body: &Value) {
    use crate::registry::{self, ServiceStatus};

    if !matches!(
        check_acl(&state.acl_ks, signer).await,
        Ok(crate::acl::Role::Service)
    ) {
        warn!(did = signer, "health pong ignored: Service role required");
        return;
    }

    let status = body
        .get("status")
        .and_then(|v| v.as_str())
        .unwrap_or("unknown");
    let version = body
        .get("version")
        .and_then(|v| v.as_str())
        .unwrap_or("unknown");

    debug!(signer, status, version, "health pong received from server");

    // Find the instance by signer DID and mark it active
    let instance_id = signer.replace(':', "_");
    let now = crate::auth::session::now_epoch();
    if let Err(e) = registry::update_instance_status(
        &state.registry_ks,
        &instance_id,
        ServiceStatus::Active,
        now,
    )
    .await
    {
        warn!(instance_id, error = %e, "failed to update instance status from health pong");
    }
}

// ---------------------------------------------------------------------------
// Sync / domain acknowledgements (server → control plane)
// ---------------------------------------------------------------------------

/// A server acknowledged a `sync/*` push. Informational: logged, nothing else.
pub(crate) fn do_sync_ack(signer: &str, type_uri: &str, body: &Value) {
    let status = body
        .get("status")
        .and_then(|v| v.as_str())
        .unwrap_or("unknown");
    let mnemonic = body
        .get("mnemonic")
        .and_then(|v| v.as_str())
        .unwrap_or("unknown");
    info!(
        signer,
        mnemonic, status, type_uri, "DID sync: server acknowledged"
    );
}

/// A server acknowledged a `domain/{assign,unassign,purge}` push.
///
/// Mirrors the server's view of `served_domains` into the registry so the UI
/// reflects the change without waiting for the next registration:
/// `applied` on an assign ack adds the domain, anything
/// else removes it. `signer` is the ack's proven issuer and must hold the
/// Service role — otherwise any DID could rewrite a server's `served_domains`
/// (dropping a domain hides it from the next purge fan-out).
pub(crate) async fn do_domain_ack(state: &AppState, signer: &str, type_uri: &str, body: &Value) {
    let status = body
        .get("status")
        .and_then(|v| v.as_str())
        .unwrap_or("unknown");
    let Some(domain) = body.get("domain").and_then(|v| v.as_str()) else {
        return;
    };
    let op = match type_uri {
        MSG_REPLICA_DOMAIN_ASSIGN_ACK => "assign",
        MSG_REPLICA_DOMAIN_UNASSIGN_ACK => "unassign",
        _ => "purge",
    };
    info!(
        signer,
        domain, status, op, "server acknowledged domain {op}"
    );

    match check_acl(&state.acl_ks, signer).await {
        Ok(crate::acl::Role::Service) => {}
        Ok(other) => {
            warn!(did = signer, role = %other, domain, op, "domain ack rejected: Service role required");
            return;
        }
        Err(_) => {
            warn!(
                did = signer,
                domain, op, "domain ack rejected: DID not in ACL"
            );
            return;
        }
    }
    // An acknowledged unassign or purge no longer needs re-sending on the next
    // registration (see `server_push::resend_domain_intents`).
    let settled = match op {
        "assign" => crate::server_push::DomainOp::Assign,
        "unassign" => crate::server_push::DomainOp::Unassign,
        _ => crate::server_push::DomainOp::Purge,
    };
    crate::server_push::settle_domain_intent(state, signer, domain, settled).await;

    let instance_id = signer.replace(':', "_");
    match crate::registry::get_instance(&state.registry_ks, &instance_id).await {
        Ok(Some(mut instance)) => {
            // `replica/domain/assign/0.1` answers only `applied`.
            let add = op == "assign" && status == "applied";
            let mutated = if add {
                if !instance.served_domains.iter().any(|d| d == domain) {
                    instance.served_domains.push(domain.to_string());
                    instance.served_domains.sort();
                    true
                } else {
                    false
                }
            } else {
                let before = instance.served_domains.len();
                instance.served_domains.retain(|d| d != domain);
                instance.served_domains.len() != before
            };
            if mutated
                && let Err(e) =
                    crate::registry::register_instance(&state.registry_ks, &instance).await
            {
                warn!(instance_id, error = %e, "failed to update served_domains after domain ack");
            }
        }
        Ok(None) => {
            // Ack arrived before registration, or the registry was wiped; the
            // next register cycle reconciles.
            warn!(
                instance_id,
                "domain ack received for unregistered server — skipping registry update"
            );
        }
        Err(e) => {
            warn!(instance_id, error = %e, "failed to load instance for ack reconciliation");
        }
    }
}

// ---------------------------------------------------------------------------
// Server registration handler
// ---------------------------------------------------------------------------

/// A rejection an infrastructure core (registration, stats sync) can report;
/// the trust-task route turns it into a framework `ErrorResponse`.
pub(crate) struct InfraRejection {
    pub code: &'static str,
    pub comment: String,
}

impl InfraRejection {
    fn new(code: &'static str, comment: impl Into<String>) -> Self {
        Self {
            code,
            comment: comment.into(),
        }
    }
}

/// Transport-agnostic core of server registration, for the
/// `.../server/register/0.1` trust task arriving over DIDComm **or** TSP.
///
/// `signer` is the registration document's proven issuer — the DID whose key
/// signed it — so a server is registered, and synced every tenant's DIDs, only
/// on its own signature. Returns the ack body on success.
pub(crate) async fn do_server_register(
    state: &AppState,
    signer: &str,
    body: &Value,
) -> Result<Value, InfraRejection> {
    use crate::acl::check_acl;
    use crate::registry::{self, ServiceInstance, ServiceStatus, ServiceType};

    // Require pre-approved ACL entry with the Service role — matches REST
    // `/api/control/register-service` which is gated on ServiceAuth. An
    // Owner-role DID must not be able to register as a server: registration
    // triggers `sync_all_dids_to_server`, which would push every tenant's
    // DID log + witness content to the caller's inbox and add them to the
    // active-server registry so future `notify_servers_did` updates also
    // reach them.
    let role = match check_acl(&state.acl_ks, signer).await {
        Ok(crate::acl::Role::Service) => crate::acl::Role::Service,
        Ok(other) => {
            warn!(
                did = signer,
                role = %other,
                "server registration rejected: Service role required"
            );
            return Err(InfraRejection::new(
                "e.p.registration.unauthorized",
                "service role required to register as a server",
            ));
        }
        Err(_) => {
            warn!(
                did = signer,
                "server registration rejected: DID not in ACL (requires pre-approval)"
            );
            return Err(InfraRejection::new(
                "e.p.registration.unauthorized",
                "server DID must be pre-approved in the ACL before registering",
            ));
        }
    };

    let public_url = body
        .get("public_url")
        .and_then(|v| v.as_str())
        .unwrap_or_default();

    // Apply the same URL allowlist that the REST `register_service`
    // route enforces. Without this gate, an ACL'd Service-role caller
    // could register an arbitrary URL (cloud-metadata IP, RFC1918,
    // attacker-controlled host) and then wait for an Admin to hit
    // `/api/proxy/server/{instance_id}/...` — `proxy_to_service` would
    // forward the Admin's `Authorization: Bearer ...` to that URL.
    if let Err(e) =
        registry::validate_registered_url(public_url, &state.config.registry.url_allowlist)
    {
        warn!(
            did = signer,
            requested = public_url,
            "server registration rejected: URL host not in registry.url_allowlist",
        );
        return Err(InfraRejection::new(
            "e.p.registration.unauthorized",
            e.user_message(),
        ));
    }

    let label = body.get("label").and_then(|v| v.as_str()).map(String::from);

    // T27: parse capability declaration. Backwards-compat: pre-T27
    // servers don't send these fields; default to webvh-only.
    let enabled_methods: Vec<String> = body
        .get("enabled_methods")
        .and_then(|v| v.as_array())
        .map(|arr| {
            arr.iter()
                .filter_map(|v| v.as_str().map(String::from))
                .collect()
        })
        .unwrap_or_else(|| vec!["webvh".to_string()]);
    let claimed_served_domains: Vec<String> = body
        .get("served_domains")
        .and_then(|v| v.as_array())
        .map(|arr| {
            arr.iter()
                .filter_map(|v| v.as_str().map(String::from))
                .collect()
        })
        .unwrap_or_default();
    // Defense against a misbehaving server fabricating `served_domains`
    // entries to mislead operators (and the purge-fanout in
    // `delete_domain_route`). We intersect the self-asserted list
    // against the control plane's authoritative DomainEntry registry,
    // dropping names the control plane has no record of. This filters
    // the "fabricate-a-domain-name" attack vector.
    //
    // KNOWN GAP (tracked for follow-up): we can't yet verify that the
    // server is actually *authorised* to host the domains it claims —
    // the control plane fires assign/unassign messages without
    // recording them. A peer Service-role server could still claim to
    // host a real but unrelated tenant's domain. Closing that gap
    // requires a new control-plane-side dispatched-assignments table
    // (one row per outbound replica/domain/{assign,unassign}) the register
    // handler can intersect against. Out of scope for this security
    // fix; HIGH-1 (control-plane pinning on the server-side handlers)
    // already removes the most damaging exploit (a peer issuing
    // forged domain.purge to wipe data).
    let served_domains: Vec<String> = if claimed_served_domains.is_empty() {
        Vec::new()
    } else {
        let mut keep = Vec::with_capacity(claimed_served_domains.len());
        for name in claimed_served_domains {
            match did_hosting_common::server::domain::get_domain(&state.store, &name).await {
                Ok(Some(_)) => keep.push(name),
                Ok(None) => warn!(
                    did = signer,
                    domain = %name,
                    "register: dropping served_domains entry — control plane has no DomainEntry for this name"
                ),
                Err(e) => warn!(
                    did = signer,
                    domain = %name,
                    error = %e,
                    "register: failed to look up DomainEntry during served_domains validation — dropping entry"
                ),
            }
        }
        keep
    };
    let protocol_version = body
        .get("protocol_version")
        .and_then(|v| v.as_str())
        .unwrap_or("1.0")
        .to_string();

    // Self-asserted: only a server that ships the trust-task dispatcher sends
    // this. Absent (older fleet) → false → the health loop keeps sending the
    // legacy `MSG_HEALTH_PING`, so upgrading the control plane first never
    // strands a server as Unreachable.
    let trust_task_capable = body
        .get("trust_task_capable")
        .and_then(|v| v.as_bool())
        .unwrap_or(false);
    let sync_batch_capable = body
        .get("sync_batch")
        .and_then(|v| v.as_bool())
        .unwrap_or(false);

    // Use the signer DID as a stable instance ID (one registration per DID)
    let instance_id = signer.replace(':', "_");

    // `register_instance` overwrites the whole record, so carry the previous
    // badge cache forward rather than blanking it. The refresh below
    // re-resolves and supersedes this; if that resolve fails, the operator
    // keeps seeing the last known services instead of an empty badge row.
    let previous = registry::get_instance(&state.registry_ks, &instance_id)
        .await
        .ok()
        .flatten();

    let instance = ServiceInstance {
        instance_id: instance_id.clone(),
        service_type: ServiceType::Server,
        label,
        url: public_url.to_string(),
        status: ServiceStatus::Active,
        last_health_check: None,
        registered_at: crate::auth::session::now_epoch(),
        metadata: json!({ "did": signer }),
        enabled_methods,
        served_domains,
        protocol_version,
        advertised_services: previous
            .as_ref()
            .and_then(|p| p.advertised_services.clone()),
        services_checked_at: previous.as_ref().and_then(|p| p.services_checked_at),
        trust_task_capable,
        sync_batch_capable,
        // `register_instance` overwrites the whole record, so the observed-link
        // history has to survive a re-register exactly as the badge cache does.
        // The caller records *this* registration's inbound transport right
        // after; the outbound side has no other source and would otherwise be
        // blanked every time a server restarts.
        last_inbound_transport: previous.as_ref().and_then(|p| p.last_inbound_transport),
        last_inbound_at: previous.as_ref().and_then(|p| p.last_inbound_at),
        last_outbound_transport: previous.as_ref().and_then(|p| p.last_outbound_transport),
        last_outbound_at: previous.as_ref().and_then(|p| p.last_outbound_at),
        last_ack_at: previous.as_ref().and_then(|p| p.last_ack_at),
        last_reconcile_at: previous.as_ref().and_then(|p| p.last_reconcile_at),
    };

    if let Err(e) = registry::register_instance(&state.registry_ks, &instance).await {
        warn!(did = signer, error = %e, "server registration failed");
        return Err(InfraRejection::new(
            "e.p.registration.internal-error",
            e.to_string(),
        ));
    }

    info!(
        did = signer,
        instance_id = %instance_id,
        public_url = public_url,
        role = %role,
        "server registered"
    );

    // Cache what the registering server's DID document advertises, so the
    // Servers list can badge it. Best-effort: a registration must not fail
    // because the DID momentarily won't resolve.
    if let Err(e) = registry::refresh_advertised_services(
        &state.registry_ks,
        &instance_id,
        state.did_resolver.as_ref(),
        crate::auth::session::now_epoch(),
    )
    .await
    {
        warn!(
            did = signer,
            error = %e,
            "failed to cache advertised services for registering server"
        );
    }

    // Sync DIDs to the newly registered server — only the ones it doesn't
    // already have. The server reports what it holds in `preloaded_dids`
    // (mnemonic → DID + version_count); anything absent, stale, or a different
    // DID at that slot is pushed. A server with an empty store sends none and
    // gets a full push. This is what stops a reboot from re-syncing every DID.
    let reported = server_push::parse_reported(body);
    server_push::sync_all_dids_to_server(state, signer.to_string(), reported);
    server_push::sync_all_domains_to_server(state, signer.to_string());

    Ok(json!({
        "instance_id": instance_id,
        "status": "registered",
    }))
}

/// Answer a message this router has no arm for with a problem-report, so the
/// caller fails immediately and *knows which task we refused*.
///
/// Dropping an unrouted type silently — which this did — turns every retirement
/// into a mystery timeout on the client. When #144 retired `did/publish/0.1` and
/// the `agent-name` set/enable/disable trio, a VTA still sending them got no
/// reply at all: it burned its full 30s `send_and_wait` and surfaced
/// "bad gateway: request timed out" with no mention of the task, on four
/// separate operations. The retirements were correct; the silence is what made
/// them expensive to diagnose. Naming the type turns a 30s dead end into an
/// immediate, greppable error, and does so for the *next* retirement too.
///
/// Threading to `message.id` is load-bearing: callers demux replies by thread
/// id, so an unthreaded problem-report is discarded and the caller waits out its
/// timeout anyway — the bug would look unfixed.
///
/// Loop-safe: an inbound problem-report returns above, before we would reply,
/// and `is_problem_report` matches the type we emit — so a peer that cannot
/// route our reply logs and drops it rather than answering back.
async fn handle_fallback(
    ctx: HandlerContext,
    message: Message,
) -> Result<Option<DIDCommResponse>, DIDCommServiceError> {
    let sender = ctx.sender_did.as_deref();
    if log_problem_report("control", sender, &message) {
        return Ok(None);
    }
    warn!(
        sender = sender.unwrap_or("unknown"),
        msg_type = %message.typ,
        "inbound DIDComm: unhandled message type — replying with a problem-report"
    );
    let (typ, body) = run_unsupported_task(&message.typ);
    Ok(Some(
        DIDCommResponse::new(typ, body).thid(message.id.clone()),
    ))
}

/// The problem-report [`handle_fallback`] answers an unrouted type with.
///
/// Returns `(typ, body)` rather than a `DIDCommResponse` for the same reason
/// the `run_*` helpers in this module do: the response type exposes no getters,
/// so a test can only assert what the builder was handed. The threading stays at
/// the call site, where `message.id` lives.
fn run_unsupported_task(msg_type: &str) -> (String, Value) {
    (
        MSG_PROBLEM_REPORT.to_string(),
        json!({
            "code": "e.p.msg.unsupported-task",
            // The type is the whole diagnostic: a retired task, a typo, and a
            // peer predating the task are indistinguishable without it, and the
            // caller cannot log what it was never told.
            "comment": format!(
                "this control plane has no handler for `{msg_type}` — it may have been retired, \
                 or this peer may be a version that predates it"
            ),
        }),
    )
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

fn require_sender(ctx: &HandlerContext) -> Result<&str, DIDCommServiceError> {
    ctx.sender_did
        .as_deref()
        .map(|did| did.split('#').next().unwrap_or(did))
        .ok_or_else(|| DIDCommServiceError::Internal("missing sender DID".into()))
}

// ---------------------------------------------------------------------------
// Trust Tasks envelope handler (v0.7.0+)
// ---------------------------------------------------------------------------

/// DIDComm handler for the Trust Tasks envelope
/// (`https://trusttasks.org/binding/didcomm/0.1/envelope`).
///
/// `message.body` is the JSON of the inner `TrustTask<Value>` document
/// (per the binding's wire shape — see `trust-tasks-didcomm/src/pack.rs`).
/// `ctx.sender_did` is the messaging layer's report of who sent it: a routing
/// hint that [`dispatch_trust_task_doc`] requires to agree with the document's
/// proof, never an authorisation on its own.
///
/// We construct a [`DidcommHandler`] reporting `local = server_did`
/// and `peer = sender`, then hand the document to the shared
/// [`dispatch_trust_task_doc`] core. The result is repacked as a new DIDComm
/// message of the same envelope type so the same routing rules
/// (mediator pickup, attachment, etc.) apply.
async fn handle_trust_tasks_envelope(
    ctx: HandlerContext,
    message: Message,
    Extension(state): Extension<AppState>,
) -> Result<Option<DIDCommResponse>, DIDCommServiceError> {
    let sender = require_sender(&ctx)?;
    info!(sender = sender, "inbound DIDComm: trust-tasks envelope");
    match run_trust_tasks_envelope(&state, sender, &message).await? {
        Some((response_type, response_body)) => Ok(Some(
            DIDCommResponse::new(response_type, response_body).thid(message.id.clone()),
        )),
        None => Ok(None),
    }
}

/// Compute the wire-level `(response_type, response_body)` tuple for
/// an inbound trust-tasks envelope. Extracted from
/// `handle_trust_tasks_envelope` so the dispatch + body-parse +
/// repack logic is testable without an `ATM`-backed
/// [`HandlerContext`]. Returns `None`
/// for the SPEC.md §8.1 routing exception (identity-mismatch with no
/// transport sender), which is unreachable on the dispatcher's
/// `require_sender_did(true)` gate.
pub(crate) async fn run_trust_tasks_envelope(
    state: &AppState,
    sender: &str,
    message: &Message,
) -> Result<Option<(String, Value)>, DIDCommServiceError> {
    // Per-type document size, the same gate the HTTPS and TSP transports
    // apply before their own parse (`size`). The DIDComm messaging layer
    // has already decoded `message.body` into a `Value` by the time it
    // reaches us — there is no raw pre-parse buffer left to peek — so the
    // check runs against its re-serialised bytes instead. That still
    // refuses an oversized document before anything below does the
    // heavier work of typed parsing, proof verification or dispatch, which
    // is the property the gate exists for.
    //
    // A document large enough to need its claimed (not yet verified)
    // issuer's ACL entry is charged against that sender's large-document
    // budget first (`vid:{sender}` — the transport's own authenticated
    // report of who sent it, not yet the document's proven issuer).
    let body_bytes = serde_json::to_vec(&message.body).unwrap_or_default();
    let large_doc_charge =
        match did_hosting_common::server::trust_tasks::size::check_for_known_issuer(
            &body_bytes,
            &crate::control_tasks::SERVED_TRUST_TASK_URIS,
            &state.acl_ks,
            &state.large_document_budget,
            &format!("vid:{sender}"),
            did_hosting_common::server::auth::session::now_epoch(),
        )
        .await
        {
            Ok(charge) => charge,
            Err(err) => {
                let body =
                    serde_json::to_value(&err).expect("trust-task-error document serialises");
                return Ok(Some((trust_tasks_didcomm::ENVELOPE_TYPE.to_string(), body)));
            }
        };

    let doc: trust_tasks_rs::TrustTask<Value> = match serde_json::from_value(message.body.clone()) {
        Ok(d) => d,
        Err(e) => {
            warn!(
                sender = sender,
                error = %e,
                "trust-tasks envelope: inner body did not parse as TrustTask<Value>"
            );
            // Mirror the routed-error pattern from the HTTPS
            // transport: build a `malformed_request` document and
            // pack it back inside the same envelope type so the
            // sender sees a consistent error shape across transports.
            let err_doc = body_parse_error(&e.to_string());
            let body =
                serde_json::to_value(&err_doc).expect("trust-task-error document serialises");
            return Ok(Some((trust_tasks_didcomm::ENVELOPE_TYPE.to_string(), body)));
        }
    };

    let my_vid = state
        .config
        .server_did
        .as_deref()
        .ok_or_else(|| DIDCommServiceError::Internal("server_did not configured".into()))?;

    // Replay protection is not here: a DIDComm message id is chosen per
    // message by whoever sends it, so a captured document re-wrapped in a new
    // message would sail past a `(sender, message.id)` check. The unified
    // dispatcher keys its replay cache on `(proven issuer, document id)`
    // instead — both covered by the document's own proof.
    let transport =
        trust_tasks_didcomm::DidcommHandler::new(my_vid.to_string(), sender.to_string());

    // Route through the unified trust-task dispatcher — the same entry TSP and
    // HTTPS use. Every family is reachable here: the typed
    // `did-hosting/*/1.0` protocol, auth, infra, ACL + discovery, and the
    // legacy `MSG_*` bridge.
    let verifier = require_verifier(state)?;
    match dispatch_trust_task_doc(
        state,
        sender,
        None,
        None,
        &transport,
        doc,
        verifier,
        large_doc_charge,
    )
    .await?
    .into_document()
    {
        Some(body) => Ok(Some((trust_tasks_didcomm::ENVELOPE_TYPE.to_string(), body))),
        None => {
            // SPEC §8.1 routing exception: identity-mismatch with no
            // transport sender. Unreachable under `require_sender_did(true)`;
            // if it fires, the invariant has broken.
            tracing::error!(
                should_not_happen = true,
                sender = sender,
                "trust-tasks envelope: dispatch suppressed (identity_mismatch w/ no transport sender)"
            );
            Ok(None)
        }
    }
}

/// Build the same body-parse `trust-task-error` document that the
/// HTTPS transport uses, kept here so the two transports emit
/// byte-identical error shapes for parse failures.
pub(crate) fn body_parse_error(reason: &str) -> trust_tasks_rs::ErrorResponse {
    use trust_tasks_rs::{ErrorPayload, RejectReason, TrustTask};
    let reject = RejectReason::MalformedRequest {
        reason: format!("body did not parse as a Trust Task document: {reason}"),
    };
    let payload: ErrorPayload = reject.into();
    TrustTask {
        id: format!("urn:uuid:{}", uuid::Uuid::new_v4()),
        thread_id: None,
        // The body never parsed, so there is no request to read an enclosing
        // exchange from (SPEC §4.9.2).
        parent_thread_id: None,
        // No ceremony, for the same reason as `parent_thread_id` above: SPEC
        // §7.1 carries the member forward from the request so a reply stays
        // inside its enactment, and here there is no request to carry it from.
        ceremony: None,
        type_uri: did_hosting_common::server::trust_tasks::framework_error_type_uri(),
        issuer: None,
        recipient: None,
        issued_at: Some(chrono::Utc::now()),
        expires_at: None,
        payload,
        context: None,
        proof: None,
        extra: Default::default(),
    }
}

// ---------------------------------------------------------------------------
// Unified trust-task dispatch (all transports)
// ---------------------------------------------------------------------------

/// The single dispatch entry for an inbound `TrustTask<Value>` document,
/// shared by **every** transport that carries trust-task documents — TSP,
/// the DIDComm trust-task envelope, and (behind its bearer-auth proof
/// pre-check) HTTPS `POST /api/trust-tasks`.
///
/// Routes by Type URI:
/// - Type URIs the framework dispatcher owns (ACL grant/revoke/change-role/
///   show/list, discovery) → the typed SPEC §7.2 pipeline via
///   [`dispatch_inbound`](did_hosting_common::server::trust_tasks::dispatch_inbound).
/// - Everything else (the legacy DID-management ops) → [`bridge_did_management`],
///   which reuses the transport-agnostic [`dispatch_did_op`] engine.
///
/// Returns the response document as a JSON value (`None` = the SPEC §8.1
/// identity-mismatch "suppressed" case). Each transport serialises the
/// value for its own wire (TSP → bytes, DIDComm → envelope body, HTTPS →
/// JSON response). `transport` is the caller's [`TransportHandler`] so the
/// framework path resolves identities with the right binding.
/// What a routed Trust Task produced, before any transport turns it into a
/// reply.
///
/// The variants exist because **HTTPS needs the reject code and the other two
/// transports do not**. SPEC's status table maps a framework rejection onto an
/// HTTP status; DIDComm and TSP have no status to carry, so they serialise the
/// document either way. Flattening to `Option<Value>` here — as this router used
/// to — meant the HTTPS route could not use it and had to re-implement the
/// routing itself, which is how it came to serve a strict subset: the typed
/// `did-hosting/*/1.0` family, the auth family and the infra family were all
/// unreachable over HTTPS, and an agent-name update sent there fell through to a
/// legacy table that has never heard of it and answered "unknown op".
pub(crate) enum RoutedReply {
    /// A framework outcome. Carries the reject code, so HTTPS can map a status.
    ///
    /// Boxed: `DispatchOutcome` is ~750 bytes against a `Value`'s few, and this
    /// enum is returned from every dispatch on every transport.
    Framework(Box<did_hosting_common::server::trust_tasks::DispatchOutcome>),
    /// A reply from a bridged or typed op. There is no framework reject code to
    /// map, and the problem-report shape carries any error, so HTTPS answers
    /// `200` exactly as the other two transports answer with the document.
    Document(Value),
    /// Nothing goes on the wire (SPEC §8.1: identity mismatch with no
    /// transport-authenticated sender).
    Suppressed,
    /// A reply a bound-key row (`ProofRule::BoundKey`) produced whole: a
    /// success its task signed under its own specification's rule (wallet
    /// sign-in's step 1 and step 2 for `assertionMethod`, and a step 2 whose
    /// signed bytes the member's grant is bound to), or an unsigned
    /// `trust-task-error` in its family's documented shape. [`seal_reply`]
    /// passes it through untouched — re-signing would break exactly that
    /// binding — and HTTPS answers with its status, never cached.
    Sealed(crate::control_tasks::SealedReply),
}

impl RoutedReply {
    /// The reply document, for a transport with no status codes to carry.
    pub(crate) fn into_document(self) -> Option<Value> {
        use did_hosting_common::server::trust_tasks::DispatchOutcome;
        match self {
            RoutedReply::Framework(outcome) => match *outcome {
                DispatchOutcome::Handled(doc) => {
                    Some(serde_json::to_value(&doc).expect("response document serialises"))
                }
                DispatchOutcome::Rejected(err) => {
                    Some(serde_json::to_value(&err).expect("error document serialises"))
                }
                DispatchOutcome::Suppressed => None,
            },
            RoutedReply::Document(value) => Some(value),
            RoutedReply::Suppressed => None,
            RoutedReply::Sealed(sealed) => sealed.into_document(),
        }
    }
}

/// What the HTTPS binding knows about the request a document arrived in,
/// which the document cannot carry: the client address, the headers (a
/// browser's `Origin`, its `User-Agent`) and the body's size. `None` on the
/// messaging transports.
///
/// Read only by a bound-key row's admission and handler (`ProofRule::
/// BoundKey`), which is served over HTTPS alone.
pub(crate) struct HttpsFacts<'a> {
    /// The client address, as `routes::trust_tasks` resolved it; `None` for a
    /// direct call with no socket (tests), which shares one bucket.
    pub ip: Option<std::net::IpAddr>,
    pub headers: &'a axum::http::HeaderMap,
    pub body_len: usize,
}

/// Route one Trust Task document, for **every** transport.
///
/// HTTPS, DIDComm and TSP all arrive here. Each opens its own binding and
/// establishes its own authenticated sender — that is what a binding is for —
/// and then the question "what does this document mean" is answered once, here.
///
/// The order of the checks below is load bearing and each one carries the
/// reason it sits where it does; they are not interchangeable.
///
/// `sender` is the transport's report of who sent the document (empty for an
/// anonymous HTTPS request that names no issuer); `bearer` is the session an
/// HTTPS request presented, when it presented one. Neither authorises anything
/// on its own.
///
/// `large_doc_charge` is what the caller's own `size::check_for_known_issuer`
/// returned, if this document was large enough to need one — settled below,
/// once proof verification concludes, against exactly the address it was
/// charged against.
// One argument per thing a binding knows; a struct here would only rename them.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn dispatch_trust_task_doc(
    state: &AppState,
    sender: &str,
    bearer: Option<&crate::auth::AuthClaims>,
    https: Option<&HttpsFacts<'_>>,
    transport: &(impl trust_tasks_rs::TransportHandler + Sync),
    doc: trust_tasks_rs::TrustTask<Value>,
    verifier: &did_hosting_common::server::trust_tasks::TransportBoundVerifier,
    large_doc_charge: Option<did_hosting_common::server::trust_tasks::size::LargeDocumentCharge>,
) -> Result<RoutedReply, DIDCommServiceError> {
    // Fail closed: a control plane that cannot sign refuses to start (see
    // `signing::require_signing_identity`), and this is the same rule at
    // request time — no trust task is acted on that could not then be
    // answered with a signed reply. There is no unsigned mode.
    let secret = match state
        .config
        .server_did
        .as_deref()
        .ok_or_else(|| "server_did not configured".to_string())
        .and_then(|did| {
            crate::signing::control_signing_secret(state, did).map_err(|e| e.to_string())
        }) {
        Ok(secret) => secret,
        Err(e) => {
            tracing::error!(error = %e, "refusing trust task: this control plane cannot sign its replies");
            return Ok(RoutedReply::Framework(Box::new(
                did_hosting_common::server::trust_tasks::DispatchOutcome::Rejected(
                    doc.reject_with(
                        format!("urn:uuid:{}", uuid::Uuid::new_v4()),
                        trust_tasks_rs::RejectReason::InternalError {
                            reason: "this service cannot sign its replies".into(),
                        },
                    ),
                ),
            )));
        }
    };
    let reply = route_trust_task_doc(
        state,
        sender,
        bearer,
        https,
        transport,
        doc,
        verifier,
        large_doc_charge,
    )
    .await?;
    Ok(seal_reply(state, &secret, sender, reply).await)
}

/// Sign every non-error reply the control plane emits, on every transport.
///
/// A response is a document from this control plane to the requester, and a
/// requester must be able to attribute it by signature, not by the channel it
/// arrived on: `auth/authenticate/0.2` declares a proof REQUIRED and SPEC §7.3
/// item 8 applies that to its response, and clients refuse unsigned non-error
/// replies. So each reply is re-stamped `issuer` = this control plane,
/// `recipient` = the requester (when the request named none), `issuedAt` = now,
/// and signed with the operational key (`proofPurpose: authentication`).
///
/// `trust-task-error` documents stay unsigned, as the framework allows. A
/// control plane that cannot sign does not start, and
/// [`dispatch_trust_task_doc`] refuses every request up front should the key
/// ever be missing at request time.
async fn seal_reply(
    state: &AppState,
    secret: &affinidi_tdk::secrets_resolver::secrets::Secret,
    requester: &str,
    reply: RoutedReply,
) -> RoutedReply {
    use did_hosting_common::server::trust_tasks::DispatchOutcome;
    match reply {
        RoutedReply::Framework(outcome) => match *outcome {
            DispatchOutcome::Handled(doc) => RoutedReply::Framework(Box::new(
                DispatchOutcome::Handled(sign_reply_doc(state, secret, requester, doc).await),
            )),
            other => RoutedReply::Framework(Box::new(other)),
        },
        RoutedReply::Document(value) => {
            let is_error = value
                .get("type")
                .and_then(Value::as_str)
                .is_some_and(|t| t.starts_with(vta_sdk::inbound::TRUST_TASK_ERROR_PREFIX));
            if is_error {
                return RoutedReply::Document(value);
            }
            match serde_json::from_value::<trust_tasks_rs::TrustTask<Value>>(value.clone()) {
                Ok(doc) => RoutedReply::Document(
                    serde_json::to_value(sign_reply_doc(state, secret, requester, doc).await)
                        .expect("signed reply serialises"),
                ),
                Err(_) => RoutedReply::Document(value),
            }
        }
        RoutedReply::Suppressed => RoutedReply::Suppressed,
        sealed @ RoutedReply::Sealed(_) => sealed,
    }
}

/// Stamp and sign one reply document. See [`seal_reply`].
///
/// A reply that cannot be signed is replaced by an (unsigned) internal-error
/// document rather than sent unsigned.
pub(crate) async fn sign_reply_doc(
    state: &AppState,
    secret: &affinidi_tdk::secrets_resolver::secrets::Secret,
    requester: &str,
    mut doc: trust_tasks_rs::TrustTask<Value>,
) -> trust_tasks_rs::TrustTask<Value> {
    let my_vid = state.config.server_did.clone().unwrap_or_default();
    doc.issuer = Some(my_vid);
    // An anonymous request (a public read that named no issuer) is answered
    // with no `recipient`.
    if doc.recipient.is_none() && !requester.is_empty() {
        doc.recipient = Some(requester.to_string());
    }
    doc.issued_at = Some(chrono::Utc::now());
    doc.proof = None;
    match did_hosting_common::server::trust_tasks::sign_document(&doc, secret).await {
        Ok(signed) => signed,
        Err(e) => {
            tracing::error!(error = %e, type_uri = %doc.type_uri, "cannot sign trust-task reply; refusing it");
            let mut err = doc.reject_with(
                format!("urn:uuid:{}", uuid::Uuid::new_v4()),
                trust_tasks_rs::RejectReason::InternalError {
                    reason: "this service could not sign its reply".into(),
                },
            );
            // Rejecting a *reply* swaps its parties; address it back to the
            // requester.
            err.issuer = doc.issuer.clone();
            err.recipient = Some(requester.to_string());
            serde_json::from_value(serde_json::to_value(&err).expect("error serialises"))
                .expect("error re-reads as a TrustTask")
        }
    }
}

/// The routing half of [`dispatch_trust_task_doc`]; replies are signed by the
/// caller.
// One argument per thing a binding knows; a struct here would only rename them.
#[allow(clippy::too_many_arguments)]
async fn route_trust_task_doc(
    state: &AppState,
    sender: &str,
    bearer: Option<&crate::auth::AuthClaims>,
    https: Option<&HttpsFacts<'_>>,
    transport: &(impl trust_tasks_rs::TransportHandler + Sync),
    doc: trust_tasks_rs::TrustTask<Value>,
    verifier: &did_hosting_common::server::trust_tasks::TransportBoundVerifier,
    large_doc_charge: Option<did_hosting_common::server::trust_tasks::size::LargeDocumentCharge>,
) -> Result<RoutedReply, DIDCommServiceError> {
    use did_hosting_common::server::trust_tasks::{
        DispatchOutcome, TransportBoundVerifier, TrustTaskContext, dispatch_inbound,
        verify_bound_key, verify_sender_bound, verify_sender_bound_approval,
    };

    let my_vid = state
        .config
        .server_did
        .as_deref()
        .ok_or_else(|| DIDCommServiceError::Internal("server_did not configured".into()))?;

    let type_uri = doc.type_uri.to_string();

    // An error is terminal: it is somebody's account of a failure, not a request
    // to act on. Checked before anything else — including the proof gate below,
    // which would otherwise answer an unsigned error with an error of its own,
    // and the peer would do the same with ours. One failure became a permanent
    // exchange between two services politely answering each other's errors
    // once already, and it stopped only when the mediator started
    // rate-limiting.
    //
    // The reason is logged, not just the fact: a peer's error is often the only
    // account of what went wrong anywhere in the exchange.
    //
    // The prefix comes from `vta_sdk::inbound`, not a literal here: which URIs
    // are terminal errors is a fact about the *document*, the same for whoever
    // reads one, and a local copy is a second place for it to drift from.
    if type_uri.starts_with(vta_sdk::inbound::TRUST_TASK_ERROR_PREFIX) {
        let code = doc
            .payload
            .get("code")
            .and_then(Value::as_str)
            .unwrap_or("?");
        let message = doc
            .payload
            .get("message")
            .and_then(Value::as_str)
            .unwrap_or("");
        warn!(
            sender,
            %code,
            %message,
            "inbound trust-task error from a peer — terminal, not answered"
        );
        // A server's *signed*, non-retryable refusal of an op this control
        // plane sent settles that outbox entry: the target has said it will not
        // apply it, so re-sending would only stall its queue. An unsigned error
        // settles nothing — anyone could send one.
        let retryable = doc
            .payload
            .get("retryable")
            .and_then(Value::as_bool)
            .unwrap_or(false);
        if doc.proof.is_some()
            && !retryable
            && (crate::server_push::is_configured_watcher(state, sender)
                || matches!(
                    did_hosting_common::server::acl::get_acl_entry(&state.acl_ks, sender).await,
                    Ok(Some(_))
                ))
            && let Ok(signer) =
                verify_sender_bound(&doc, None, Some(sender), my_vid, verifier).await
        {
            crate::trust_tasks_infra::acknowledge_outbox(state, &signer, &doc).await;
        }
        return Ok(RoutedReply::Suppressed);
    }

    // ── The proof gate. Every document that asks this control plane to *do*
    //    something — change state, disclose a record, mint a session — must be
    //    signed by its issuer, addressed here, and fresh; and the issuer must be
    //    the DID the transport reported (`verify_sender_bound`). From here on
    //    `principal` — the proven signer — is the only identity anything below
    //    authorises on. The transport's `sender` is a routing hint, and a
    //    document whose proof disagrees with it is refused rather than resolved
    //    in either's favour.
    //
    //    Each task's `ProofRule` says what its proof must be:
    //    - `Optional`: capability discovery, the public `server/info`,
    //      opening a passkey login, and the whole `auth/challenge` /
    //      `auth/authenticate` / `auth/refresh` family — not because a proof
    //      is optional there (several of these require one), but because the
    //      ACL pre-filter below is the wrong question for a family whose own
    //      handlers already decide who may act, on a signer the ACL may
    //      legitimately never list (a proxied login's delegate) or has not
    //      listed yet (a first login);
    //    - `Authentication`: an operational request, `proofPurpose:
    //      authentication` under the signer's `authentication` relationship;
    //    - `AssertionMethod`: a human approver's decision, an attestation under
    //      the approver's `assertionMethod` relationship;
    //    - `SessionKey`: a ceremony that mints a session for a key the caller
    //      has just generated, signed by that `did:key` — which has no ACL entry
    //      of its own; the ceremony names the subject;
    //    - `BoundKey(purposes)`: a step of an exchange whose parties are fresh
    //      Ed25519 `did:key`s the exchange binds (wallet sign-in). HTTPS only.
    //      The row's admission runs first — its size cap, rate limits and, for
    //      a browser's step, the origin — before any proof work. Then a proof
    //      by exactly the issuer's own key, for one of the row's purposes,
    //      verified locally (`verify_bound_key`); no ACL pre-filter (the key is
    //      never in the ACL) and no shared replay cache (the row keeps its own,
    //      which a stranger cannot fill; see `ProofRule::BoundKey`). The row
    //      then checks that the key is the one the exchange recorded.
    //
    //    Before any signature work — which can mean resolving a DID over the
    //    network — the reported sender must at least *have* an ACL entry: every
    //    privileged op requires one, so a stranger is turned away for the cost
    //    of a key lookup, and never reaches the resolver or the replay cache.
    //    This is a cheap pre-filter, not the authorisation; the proof still has
    //    to bind the document to that same DID below.
    //
    //    A document that carries a proof it was not required to carry is held
    //    to the same rule: a proof that is present is verified, never ignored.
    let rule = proof_rule(&type_uri);
    if let ProofRule::BoundKey(_) = rule {
        let Some(https) = https else {
            warn!(sender, %type_uri, "trust task refused: a bound-key exchange is served over HTTPS only");
            return Ok(RoutedReply::Framework(Box::new(DispatchOutcome::Rejected(
                doc.reject_with(
                    format!("urn:uuid:{}", uuid::Uuid::new_v4()),
                    trust_tasks_rs::RejectReason::PermissionDenied {
                        reason: "this task is served over HTTPS only".into(),
                    },
                ),
            ))));
        };
        if let Err(refused) = crate::control_tasks::admit_bound(state, https, &doc).await {
            return Ok(RoutedReply::Sealed(refused));
        }
    }
    let signed = rule != ProofRule::Optional || doc.proof.is_some();
    let needs_acl = matches!(rule, ProofRule::Authentication | ProofRule::AssertionMethod);
    // A session ceremony is signed by a `did:key` the caller holds: as the
    // issuer itself, or — for a session that already exists — as that
    // session's delegate key.
    if rule == ProofRule::SessionKey
        && doc
            .proof
            .as_ref()
            .is_some_and(|p| !p.verification_method.starts_with("did:key:"))
    {
        warn!(sender, %type_uri, "trust task refused: a session ceremony must be signed by a did:key");
        return Ok(RoutedReply::Framework(Box::new(DispatchOutcome::Rejected(
            doc.reject_with(
                format!("urn:uuid:{}", uuid::Uuid::new_v4()),
                trust_tasks_rs::RejectReason::PermissionDenied {
                    reason:
                        "this ceremony must be signed by the did:key its session will be bound to"
                            .into(),
                },
            ),
        ))));
    }
    // A configured watcher has no ACL entry: its only standing here is
    // acknowledging the syncs it was sent, so that — and nothing else — gets
    // past this pre-filter on its DID.
    let watcher_ack = matches!(
        type_uri.as_str(),
        MSG_SYNC_UPDATE_ACK | MSG_SYNC_DELETE_ACK | MSG_SYNC_BATCH_ACK
    ) && crate::server_push::is_configured_watcher(state, sender);
    if needs_acl
        && !watcher_ack
        && !matches!(
            did_hosting_common::server::acl::get_acl_entry(&state.acl_ks, sender).await,
            Ok(Some(_))
        )
    {
        warn!(sender, %type_uri, "trust task refused: sender has no ACL entry");
        return Ok(RoutedReply::Framework(Box::new(DispatchOutcome::Rejected(
            doc.reject_with(
                format!("urn:uuid:{}", uuid::Uuid::new_v4()),
                trust_tasks_rs::RejectReason::PermissionDenied {
                    reason: "caller is not present in the maintainer's ACL".into(),
                },
            ),
        ))));
    }
    let principal = if signed {
        let bound = match rule {
            ProofRule::AssertionMethod => {
                verify_sender_bound_approval(&doc, Some(sender), my_vid, verifier).await
            }
            ProofRule::BoundKey(purposes) => match verify_bound_key(&doc, purposes).await {
                Ok(issuer) => Ok(issuer),
                Err(e) => {
                    did_hosting_common::server::trust_tasks::size::settle_large_document_charge(
                        &state.large_document_budget,
                        large_doc_charge.as_ref(),
                        None,
                    );
                    // Not `sender`: here it is the document's own, unverified
                    // issuer, as long as the caller cares to make it.
                    warn!(%type_uri, error = %e, "trust task refused: not signed by its bound key");
                    return Ok(RoutedReply::Sealed(crate::control_tasks::refuse_bound(
                        state, &doc, &e,
                    )));
                }
            },
            _ => verify_sender_bound(&doc, None, Some(sender), my_vid, verifier).await,
        };
        // Settle any large-document charge against this verification's own
        // outcome — before the early return below, so a claimed issuer that
        // failed to verify is penalised exactly as one that verified to
        // someone else. See `size`'s module docs.
        did_hosting_common::server::trust_tasks::size::settle_large_document_charge(
            &state.large_document_budget,
            large_doc_charge.as_ref(),
            bound.as_ref().ok().map(String::as_str),
        );
        match bound {
            Ok(principal) => principal,
            Err(e) => {
                warn!(sender, %type_uri, error = %e, "trust task refused: not bound to its sender");
                let reason = e.reject_reason();
                return Ok(RoutedReply::Framework(Box::new(DispatchOutcome::Rejected(
                    doc.reject_with(format!("urn:uuid:{}", uuid::Uuid::new_v4()), reason),
                ))));
            }
        }
    } else {
        sender.to_string()
    };

    // Replay gate, keyed on what the proof covers: the proven issuer and the
    // document id. A captured document re-submitted inside the freshness window
    // — on any transport, in any framing — is refused as an id conflict.
    // At its bounds the cache refuses (retryable) rather than forgetting.
    if signed
        && rule.uses_shared_replay_cache()
        && let Err(e) = state.replay_cache.check(&principal, &doc.id)
    {
        warn!(did = %principal, doc_id = %doc.id, %type_uri, ?e, "trust task refused: replay cache");
        let reason = match e {
            did_hosting_common::server::replay::ReplayError::Duplicate => {
                trust_tasks_rs::RejectReason::IdConflict
            }
            did_hosting_common::server::replay::ReplayError::Full => {
                trust_tasks_rs::RejectReason::Unavailable { retry_after: None }
            }
        };
        return Ok(RoutedReply::Framework(Box::new(DispatchOutcome::Rejected(
            doc.reject_with(format!("urn:uuid:{}", uuid::Uuid::new_v4()), reason),
        ))));
    }

    // Proofs are verified on every path — there is no configuration under
    // which a proof-bearing document is accepted unverified.
    let policy: trust_tasks_rs::ProofPolicy<'_, TransportBoundVerifier> =
        trust_tasks_rs::ProofPolicy::Verify(verifier);

    // Auth (`auth/{challenge,authenticate,refresh}/0.1`). Before the framework
    // check below: the shared `build_dispatcher` does not carry this family —
    // its context is the ACL keyspace — so an auth document would fall through
    // to `bridge_did_management`, a table of DID operations that has never
    // heard of it, and come back a bogus "unknown op".
    if crate::trust_tasks_auth::owns(&type_uri) {
        return Ok(RoutedReply::Document(
            crate::trust_tasks_auth::dispatch(state, transport, policy, doc)
                .await
                .ok_or(DIDCommServiceError::Internal(
                    "auth dispatch produced no reply".into(),
                ))?,
        ));
    }

    // Control↔server infrastructure ops (server registration, health pong,
    // stats, sync and domain acks). Must be checked *before* the
    // `bridge_did_management` fallthrough below, which would otherwise hand
    // them to `dispatch_did_op` — a table of DID operations that has never
    // heard of them — and answer with a bogus "unknown op" problem report.
    if crate::trust_tasks_infra::owns(&type_uri) {
        // The dispatcher stays transport-agnostic; we only tell it which binding
        // the document came in on so the registry can record what actually moved.
        let via = did_hosting_common::server::didcomm_profile::ObservedTransport::from_binding_uri(
            transport.binding_uri(),
        );
        return Ok(
            match crate::trust_tasks_infra::dispatch(state, &principal, via, doc).await {
                Some(value) => RoutedReply::Document(value),
                None => RoutedReply::Suppressed,
            },
        );
    }

    // The control plane's own table: DID management, agent names, domains.
    // Every row is typed against its generated request and response.
    //
    // The gate above has verified every proof this document carries, on the
    // document as it arrived. Narrowing it into a generated type can
    // materialise a schema default the signer never wrote (`reserve: false`),
    // which a second verification over the re-serialised document would read
    // as tampering; so the rows do not verify again.
    let doc = match crate::control_tasks::route(
        state,
        bearer,
        https,
        transport,
        trust_tasks_rs::ProofPolicy::<TransportBoundVerifier>::AcceptUnverified,
        doc,
    )
    .await
    {
        Ok(reply) => return Ok(reply),
        Err(unclaimed) => *unclaimed,
    };

    // What is left is the shared framework family (ACL, discovery); a Type URI
    // nobody serves is refused there as `unsupportedType`.
    let ctx = TrustTaskContext {
        acl_ks: &state.acl_ks,
        acl_locks: &state.acl_locks,
        my_vid,
    };

    // The framework path, and the only one whose outcome carries a reject code
    // — so it is handed back whole rather than serialised here, and HTTPS maps
    // it to a status.
    Ok(RoutedReply::Framework(Box::new(
        dispatch_inbound::<TransportBoundVerifier>(&ctx, transport, policy, doc).await,
    )))
}

/// What [`dispatch_trust_task_doc`] requires of `type_uri`'s proof.
///
/// The control plane's table says for its own rows. Of the rest, capability
/// discovery and the whole auth family (`crate::trust_tasks_auth::owns`:
/// challenge, authenticate, refresh) get `Optional` too — not because their
/// proof is optional (`auth/authenticate`'s is REQUIRED by its own spec, and
/// `signed` below still verifies whichever proof is present), but because the
/// upfront **ACL pre-filter** `needs_acl` gates on is the wrong question for
/// this family. A proxied `auth/authenticate/0.3`'s issuer is a *delegate*
/// that may never itself hold an ACL entry — only the `principal` it
/// authenticates as does — and an ordinary login's own subject, and a
/// refresh's session key, are exactly the identities this family's own arms
/// already authorise (an ACL lookup on the acting party inside
/// `handle_authenticate`, a bound-key check inside `refresh_v2_arm`). Gating
/// on the *signer* holding one first would refuse every proxied login
/// outright, and gating an ordinary first login on an ACL entry that same
/// login is what establishes control of is circular. Every other task — ACL
/// reads included — is privileged, including a Type URI nobody serves, which
/// is then refused as unsupported only once it has been proven.
///
/// The one exception is an `auth/oob/*` version the table does not serve.
/// That family's callers are fresh `did:key`s that are never in the ACL, and a
/// wallet negotiates the version by trying its newest and falling back only on
/// `unsupportedType` (it sent `prove/0.2` before this service served it, and
/// got `permissionDenied` from the pre-filter, which ended the sign-in). So it
/// is `Optional`, and reaches the framework's `unsupportedType`. That says
/// only which versions of an already-public family are served.
pub(crate) fn proof_rule(type_uri: &str) -> ProofRule {
    use trust_tasks_rs::Payload;
    if let Some(rule) = crate::control_tasks::proof_rule(type_uri) {
        return rule;
    }
    if type_uri == trust_tasks_rs::specs::trust_task_discovery::v0_1::Payload::TYPE_URI
        || crate::trust_tasks_auth::owns(type_uri)
        || type_uri.starts_with(OOB_FAMILY_PREFIX)
    {
        return ProofRule::Optional;
    }
    ProofRule::Authentication
}

/// The `auth/oob/*` family's Type URIs (see [`proof_rule`]).
const OOB_FAMILY_PREFIX: &str = "https://trusttasks.org/spec/auth/oob/";

/// Whether a session key bound at login (`auth/authenticate/0.2`, or a passkey
/// login's browser key) may sign `type_uri` for its session's subject.
///
/// Only an ordinary operational request (`ProofRule::Authentication`) may be
/// signed this way, and not the auth family itself. That rules out:
///
/// - an approval (`AssertionMethod`: step-up approve-response). The spec
///   forbids a session key wherever an `assertionMethod` attestation is
///   required (Conformance item 11);
/// - `auth/authenticate` and `auth/refresh`, which would let a session key mint
///   or extend a session past what the subject's own login granted
///   (Conformance item 12). A refresh is authorised by the refresh token, and
///   any proof on it must be the subject's own;
/// - `Optional` and `SessionKey` ceremonies, which are never signed as a
///   session's subject.
pub(crate) fn session_key_may_sign(type_uri: &str) -> bool {
    proof_rule(type_uri) == ProofRule::Authentication && !crate::trust_tasks_auth::owns(type_uri)
}

/// The shared proof verifier, or an error when none is configured — without
/// one no privileged document can be accepted, so dispatch refuses outright.
pub(crate) fn require_verifier(
    state: &AppState,
) -> Result<&did_hosting_common::server::trust_tasks::TransportBoundVerifier, DIDCommServiceError> {
    state.trust_tasks_verifier.as_deref().ok_or_else(|| {
        DIDCommServiceError::Internal(
            "no trust-task proof verifier configured (a DID resolver is required)".into(),
        )
    })
}

#[cfg(test)]
mod tests {
    use did_hosting_common::server::store::{
        KS_ACL, KS_DIDS, KS_REGISTRY, KS_SESSIONS, KS_STATS, KS_TIMESERIES,
    };
    use std::path::PathBuf;
    use std::sync::{Arc, OnceLock};

    use affinidi_messaging_didcomm::Message;
    use did_hosting_common::did_ops::{DidRecord, did_key, owner_key};
    use did_hosting_common::server::acl::{AclEntry, Role, store_acl_entry};
    use did_hosting_common::server::config::{
        AuthConfig, FeaturesConfig, LogConfig, SecretsConfig, ServerConfig, StoreConfig, VtaConfig,
    };
    use did_hosting_common::server::stats_collector::StatsCollector;
    use did_hosting_common::server::store::Store;
    use serde_json::json;

    use crate::config::{AppConfig, RegistryConfig};
    use crate::server::AppState;

    use super::*;

    /// Build a minimal `AppState` backed by a tempdir-rooted fjall store. The
    /// returned `_dir` guard must outlive `state` — when it drops, fjall
    /// removes the partition files on disk.
    async fn test_state() -> (AppState, tempfile::TempDir) {
        let dir = tempfile::tempdir().expect("temp dir");
        let store_config = StoreConfig {
            data_dir: PathBuf::from(dir.path()),
            ..StoreConfig::default()
        };
        let store = Store::open(&store_config).await.expect("open store");
        let sessions_ks = store.keyspace(KS_SESSIONS).expect("sessions ks");
        let acl_ks = store.keyspace(KS_ACL).expect("acl ks");
        let registry_ks = store.keyspace(KS_REGISTRY).expect("registry ks");
        let dids_ks = store.keyspace(KS_DIDS).expect("dids ks");
        let stats_ks = store.keyspace(KS_STATS).expect("stats ks");

        let config = AppConfig {
            features: FeaturesConfig::default(),
            server_did: Some("did:webvh:test:control.example.com".into()),
            mediator_did: None,
            public_url: Some("http://control.test".into()),
            did_hosting_url: Some("http://control.test".into()),
            server: ServerConfig::default(),
            log: LogConfig::default(),
            store: store_config,
            fjall: Default::default(),
            auth: AuthConfig::default(),
            secrets: SecretsConfig::default(),
            vta: VtaConfig::default(),
            registry: RegistryConfig::default(),
            trust_tasks: Default::default(),
            hosting: Default::default(),
            identity: Default::default(),
            config_path: PathBuf::new(),
        };

        let state = AppState {
            store: store.clone(),
            sessions_ks,
            acl_ks,
            registry_ks,
            dids_ks,
            config: Arc::new(config),
            did_resolver: None,
            secrets_resolver: None,
            identity: Some(
                did_hosting_common::server::identity::ServiceIdentity::generated_for(
                    "did:webvh:test:control.example.com",
                )
                .await
                .unwrap(),
            ),
            trust_tasks_verifier: None,
            jwt_keys: None,
            webauthn: None,
            http_client: reqwest::Client::new(),
            didcomm_service: Arc::new(OnceLock::new()),
            stats_collector: Arc::new(StatsCollector::new()),
            stats_ks: stats_ks.clone(),
            timeseries_ks: store.keyspace(KS_TIMESERIES).expect("timeseries ks"),
            signing_key_bytes: None,
            replay_cache: Arc::new(crate::replay::ReplayCache::new()),
            path_locks: crate::path_locks::PathLocks::new(),
            acl_locks: did_hosting_common::server::path_locks::PathLocks::new(),
            pending_challenges: Arc::new(crate::pending_challenges::PendingChallengeTracker::new()),
            ip_rate_limiter: Arc::new(crate::rate_limit::IpRateLimiter::new()),
            redeem_rate_limiter: Arc::new(crate::rate_limit::SourceRateLimiter::new()),
            large_document_budget: Arc::new(
                did_hosting_common::server::trust_tasks::size::LargeDocumentBudget::new(),
            ),
            outbox_notify: Arc::new(tokio::sync::Notify::new()),
            cache_invalidate: None,
            oob: Default::default(),
        };

        (state, dir)
    }

    fn build_msg(typ: &str, body: serde_json::Value) -> Message {
        Message::build("msg-id".to_string(), typ.to_string(), body).finalize()
    }

    /// Seed a fully-formed `DidRecord` with both the `did:` and `owner:`
    /// index entries so list/info/delete dispatch arms have data to read.
    async fn seed_did(state: &AppState, owner_did: &str, mnemonic: &str) {
        let record = DidRecord {
            services: None,
            owner: owner_did.into(),
            mnemonic: mnemonic.into(),
            created_at: 1,
            updated_at: 1,
            version_count: 1,
            did_id: Some(format!("did:webvh:abc:{mnemonic}")),
            content_size: 42,
            disabled: false,
            deleted_at: None,

            // T12: legacy construction site; T13 migration fills `domain`.
            method: "webvh".to_string(),
            domain: String::new(),
            agent_names: Vec::new(),
        };
        state
            .dids_ks
            .insert(did_key(mnemonic), &record)
            .await
            .expect("seed did record");
        state
            .dids_ks
            .insert_raw(owner_key(owner_did, mnemonic), mnemonic.as_bytes().to_vec())
            .await
            .expect("seed owner index");
    }

    /// ACL gate covers DIDComm authentication and DID ops alike. This pins
    /// the integration: a DID added to the ACL with role `Owner` resolves
    /// through `check_acl` to that role — the input `handle_authenticate`
    /// uses to mint a JWT and `handle_webvh_message` uses for dispatch.
    #[tokio::test]
    async fn check_acl_returns_role_for_seeded_did() {
        use did_hosting_common::server::acl::check_acl;

        let (state, _dir) = test_state().await;
        let did = "did:example:owner-a";
        store_acl_entry(
            &state.acl_ks,
            &AclEntry {
                did: did.into(),
                role: Role::Owner,
                label: None,
                created_at: 0,
                max_total_size: None,
                max_did_count: None,

                domains: did_hosting_common::server::domain::DomainScope::All,
            },
        )
        .await
        .unwrap();

        let role = check_acl(&state.acl_ks, did).await.unwrap();
        assert_eq!(role, Role::Owner);

        // DID not in ACL → Forbidden.
        let err = check_acl(&state.acl_ks, "did:example:stranger")
            .await
            .unwrap_err();
        assert!(matches!(err, crate::error::AppError::Forbidden(_)));
    }

    /// An unrouted type is answered, not dropped. Dropping it is what made the
    /// #144 retirements expensive: the caller learned nothing for 30s and then
    /// reported a bare gateway timeout. The reply must name the type, or the
    /// caller still cannot say which task was refused.
    #[test]
    fn unsupported_task_reply_names_the_type() {
        let (typ, body) = run_unsupported_task("https://example.org/spec/made-up/0.1");
        assert_eq!(typ, MSG_PROBLEM_REPORT);
        assert_eq!(
            body.get("code").and_then(|v| v.as_str()),
            Some("e.p.msg.unsupported-task")
        );
        let comment = body.get("comment").and_then(|v| v.as_str()).unwrap();
        assert!(
            comment.contains("https://example.org/spec/made-up/0.1"),
            "the refused type is the whole diagnostic; got {comment}"
        );
    }

    /// The reply is itself a problem-report by `is_problem_report`'s reckoning,
    /// which is what makes this loop-safe: a peer that cannot route our reply
    /// logs and drops it in its own fallback instead of answering back. If the
    /// emitted type ever stopped matching, two fallbacks would ping-pong.
    #[test]
    fn unsupported_task_reply_cannot_start_a_problem_report_loop() {
        let (typ, _) = run_unsupported_task("https://example.org/spec/made-up/0.1");
        assert!(
            did_hosting_common::server::problem_report::is_problem_report(&typ),
            "{typ} must be recognised as a problem-report so peers drop rather than reply"
        );
    }

    /// The four URIs #144 retired each produce a reply naming them rather than
    /// silence — these are the exact tasks a VTA predating the cutover sends,
    /// and each one cost a 30s timeout.
    ///
    /// This asserts the *reply*, not that the URIs are unrouted: `Router`
    /// exposes no route introspection, so "is it still routed" isn't checkable
    /// from here. Re-adding a route would make these URIs bypass the fallback
    /// entirely and this test would keep passing while testing nothing about
    /// them — it guards the reply shape, which is what the client depends on.
    #[test]
    fn retired_tasks_are_refused_by_name() {
        for retired in [
            "https://trusttasks.org/spec/did-management/did/publish/0.1",
            "https://trusttasks.org/spec/did-management/agent-name/set/0.1",
            "https://trusttasks.org/spec/did-management/agent-name/enable/0.1",
            "https://trusttasks.org/spec/did-management/agent-name/disable/0.1",
            // Bare-message forms of ops that now travel only as signed Trust
            // Task documents inside the envelope.
            MSG_AUTHENTICATE,
            MSG_DELETE,
            MSG_SERVER_REGISTER,
        ] {
            let (typ, body) = run_unsupported_task(retired);
            assert_eq!(typ, MSG_PROBLEM_REPORT, "{retired}");
            assert!(
                body.get("comment")
                    .and_then(|v| v.as_str())
                    .is_some_and(|c| c.contains(retired)),
                "{retired} must be named in the reply"
            );
        }
    }

    // -----------------------------------------------------------------
    // run_trust_tasks_envelope tests (v0.7.0)
    //
    // The DIDComm-side dispatch was previously uncovered because the
    // outer `handle_trust_tasks_envelope` takes `HandlerContext`, which
    // an in-process test can't build. The `run_*` extraction lets us
    // exercise the unpack + dispatch + repack logic without an
    // ATM-backed context.
    // -----------------------------------------------------------------

    const CONTROL: &str = "did:webvh:test:control.example.com";

    /// `test_state` with the production verifier over the local `did:key`
    /// resolver, so signed envelopes verify without I/O.
    async fn signing_state() -> (AppState, tempfile::TempDir) {
        let (mut state, dir) = test_state().await;
        state.trust_tasks_verifier = Some(Arc::new(TransportBoundVerifier::with_resolver(
            Arc::new(DidKeyResolver),
        )));
        (state, dir)
    }

    async fn seed_role(state: &AppState, did: &str, role: Role) {
        store_acl_entry(
            &state.acl_ks,
            &AclEntry {
                did: did.into(),
                role,
                label: None,
                created_at: 1_700_000_000,
                max_total_size: None,
                max_did_count: None,
                domains: did_hosting_common::server::domain::DomainScope::All,
            },
        )
        .await
        .unwrap();
    }

    /// An unsigned request document from `issuer` to the control plane.
    fn request_doc(type_uri: &str, issuer: Option<&str>, payload: Value) -> Value {
        let mut doc = json!({
            "id": format!("urn:uuid:{}", uuid::Uuid::new_v4()),
            "type": type_uri,
            "recipient": CONTROL,
            // Whole seconds, `Z`: the form the typed document re-serialises
            // to, so the signature over this JSON still covers it.
            "issuedAt": chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
            "payload": payload,
        });
        if let Some(issuer) = issuer {
            doc["issuer"] = json!(issuer);
        }
        doc
    }

    async fn sign(doc: Value, signer: &affinidi_tdk::secrets_resolver::secrets::Secret) -> Value {
        crate::signing::test_util::sign_operational(doc, signer).await
    }

    fn envelope(body: Value) -> Message {
        // A fresh DIDComm message id each time: replay protection must not
        // depend on the transport's id.
        Message::build(
            uuid::Uuid::new_v4().to_string(),
            trust_tasks_didcomm::ENVELOPE_TYPE.to_string(),
            body,
        )
        .finalize()
    }

    async fn dispatch_envelope(state: &AppState, sender: &str, body: Value) -> Value {
        let (resp_type, resp_body) =
            super::run_trust_tasks_envelope(state, sender, &envelope(body))
                .await
                .expect("dispatch returns Ok")
                .expect("envelope produces a response");
        assert_eq!(resp_type, trust_tasks_didcomm::ENVELOPE_TYPE);
        resp_body
    }

    fn list_payload() -> Value {
        json!({})
    }

    /// Happy path: an ACL read (`acl/list/0.1`) signed by the Admin it names.
    #[tokio::test]
    async fn trust_tasks_envelope_happy_path_list_returns_handled_response() {
        let (state, _dir) = signing_state().await;
        let (admin, admin_key) = crate::signing::test_util::did_key_signer(&[21u8; 32]);
        seed_role(&state, &admin, Role::Admin).await;

        let doc = sign(
            request_doc(
                "https://trusttasks.org/spec/acl/list/0.1",
                Some(&admin),
                list_payload(),
            ),
            &admin_key,
        )
        .await;
        let resp = dispatch_envelope(&state, &admin, doc).await;

        let inner_type = resp["type"].as_str().unwrap();
        assert!(
            inner_type.ends_with("/acl/list/0.1#response"),
            "expected acl/list response type, got {inner_type}"
        );
        let entries = resp["payload"]["entries"].as_array().unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0]["subject"], admin.as_str());
    }

    /// An ACL read is privileged: without a proof it is refused even from an
    /// Admin sender.
    #[tokio::test]
    async fn trust_tasks_envelope_unsigned_acl_read_is_refused() {
        let (state, _dir) = signing_state().await;
        let (admin, _) = crate::signing::test_util::did_key_signer(&[21u8; 32]);
        seed_role(&state, &admin, Role::Admin).await;

        let doc = request_doc(
            "https://trusttasks.org/spec/acl/list/0.1",
            Some(&admin),
            list_payload(),
        );
        let resp = dispatch_envelope(&state, &admin, doc).await;
        assert_eq!(resp["payload"]["code"], "proofRequired", "{resp}");
    }

    #[tokio::test]
    async fn trust_tasks_envelope_malformed_inner_body_returns_routed_error() {
        let (state, _dir) = test_state().await;
        // Body is a valid JSON value but not a TrustTask document.
        let msg = build_msg(
            trust_tasks_didcomm::ENVELOPE_TYPE,
            json!({"not": "a-trust-task"}),
        );
        let (resp_type, resp_body) =
            super::run_trust_tasks_envelope(&state, "did:example:admin", &msg)
                .await
                .expect("dispatch returns Ok")
                .expect("malformed body emits an error doc");

        assert_eq!(resp_type, trust_tasks_didcomm::ENVELOPE_TYPE);
        assert_eq!(
            resp_body["type"].as_str().unwrap(),
            did_hosting_common::server::trust_tasks::framework_error_type_uri().to_string()
        );
        // trust-tasks-rs 0.2 serialises StandardCode in camelCase
        // (`malformedRequest`); it still reads the 0.1 snake_case form on
        // inbound, so peers on either version interoperate.
        assert_eq!(resp_body["payload"]["code"], "malformedRequest");
    }

    /// A signed DID-management op (`did/check-name`) in the envelope is bridged
    /// to `dispatch_did_op` and answered to its signer.
    #[tokio::test]
    async fn trust_tasks_envelope_bridges_did_management() {
        let (state, _dir) = signing_state().await;
        let (admin, admin_key) = crate::signing::test_util::did_key_signer(&[22u8; 32]);
        seed_role(&state, &admin, Role::Admin).await;

        let doc = sign(
            request_doc(
                MSG_DID_REQUEST,
                Some(&admin),
                json!({ "path": "bob", "reserve": false }),
            ),
            &admin_key,
        )
        .await;
        let resp = dispatch_envelope(&state, &admin, doc).await;

        assert_eq!(
            resp["type"],
            "https://trusttasks.org/spec/did-management/did/check-name/0.1#response"
        );
        assert_eq!(resp["payload"]["available"], true);
        assert_eq!(resp["issuer"], CONTROL);
        assert_eq!(resp["recipient"], admin.as_str());
    }

    /// The unsigned form of the same op — what every DIDComm client sent before
    /// — is refused, and refused *before* the ACL is consulted, so an Admin
    /// sender name on the envelope buys nothing.
    #[tokio::test]
    async fn trust_tasks_envelope_unsigned_did_management_is_refused() {
        let (state, _dir) = signing_state().await;
        let admin = "did:example:admin";
        seed_role(&state, admin, Role::Admin).await;
        seed_did(&state, admin, "victim").await;

        let doc = request_doc(MSG_DELETE, Some(admin), json!({ "mnemonic": "victim" }));
        let resp = dispatch_envelope(&state, admin, doc).await;
        assert_eq!(resp["payload"]["code"], "proofRequired", "{resp}");
        assert!(
            state
                .dids_ks
                .get::<did_hosting_common::did_ops::DidRecord>(did_key("victim"))
                .await
                .unwrap()
                .is_some(),
            "nothing was deleted"
        );
    }

    /// A validly-signed document from one DID, delivered under another DID's
    /// name (an Admin's), is refused: the proof's issuer and the transport's
    /// reported sender disagree, and the reported sender is never enough.
    #[tokio::test]
    async fn trust_tasks_envelope_proof_from_someone_else_is_refused() {
        let (state, _dir) = signing_state().await;
        let admin = "did:example:admin";
        seed_role(&state, admin, Role::Admin).await;
        seed_did(&state, admin, "victim").await;
        let (attacker, attacker_key) = crate::signing::test_util::did_key_signer(&[66u8; 32]);

        let doc = sign(
            request_doc(MSG_DELETE, Some(&attacker), json!({ "mnemonic": "victim" })),
            &attacker_key,
        )
        .await;
        let resp = dispatch_envelope(&state, admin, doc).await;
        assert_eq!(resp["payload"]["code"], "permissionDenied", "{resp}");
        assert!(
            state
                .dids_ks
                .get::<did_hosting_common::did_ops::DidRecord>(did_key("victim"))
                .await
                .unwrap()
                .is_some(),
            "nothing was deleted"
        );
    }

    /// The `acl/grant` shape that would otherwise mint a permanent Admin: a
    /// valid proof by the attacker, no in-band `issuer`, an Admin's name on the
    /// envelope. Refused — an issuer-less proof verifies as nobody.
    #[tokio::test]
    async fn trust_tasks_envelope_issuerless_grant_under_admin_name_is_refused() {
        let (state, _dir) = signing_state().await;
        let admin = "did:example:admin";
        seed_role(&state, admin, Role::Admin).await;
        let (attacker, attacker_key) = crate::signing::test_util::did_key_signer(&[66u8; 32]);

        // Sign with an issuer (the signer insists), then strip it: the proof
        // is still a valid signature by the attacker over the rest.
        let mut signed = sign(
            request_doc(
                "https://trusttasks.org/spec/acl/grant/0.1",
                Some(&attacker),
                json!({ "entry": { "subject": attacker, "role": "admin" } }),
            ),
            &attacker_key,
        )
        .await;
        signed.as_object_mut().unwrap().remove("issuer");
        let resp = dispatch_envelope(&state, admin, signed).await;
        assert_ne!(
            resp["type"].as_str().unwrap_or(""),
            "https://trusttasks.org/spec/acl/grant/0.1#response",
            "{resp}"
        );
        assert!(
            did_hosting_common::server::acl::get_acl_entry(&state.acl_ks, &attacker)
                .await
                .unwrap()
                .is_none(),
            "no ACL entry was created for the attacker"
        );
    }

    /// Replaying a captured signed document is refused, even wrapped in a fresh
    /// DIDComm message — the gate keys on the proven issuer and the document
    /// id, both covered by the proof. Uses a destructive op because that is
    /// what replay protection exists for.
    #[tokio::test]
    async fn trust_tasks_envelope_replay_is_refused() {
        let (state, _dir) = signing_state().await;
        let (admin, admin_key) = crate::signing::test_util::did_key_signer(&[23u8; 32]);
        seed_role(&state, &admin, Role::Admin).await;
        seed_did(&state, &admin, "replayed").await;

        let doc = sign(
            request_doc(MSG_DELETE, Some(&admin), json!({ "mnemonic": "replayed" })),
            &admin_key,
        )
        .await;

        let first = dispatch_envelope(&state, &admin, doc.clone()).await;
        assert_eq!(
            first["type"], MSG_DELETE_CONFIRM,
            "the first delivery must be handled, not rejected: {first}"
        );

        let replayed = dispatch_envelope(&state, &admin, doc).await;
        assert_eq!(replayed["payload"]["code"], "idConflict", "{replayed}");
    }

    /// Over the DIDComm envelope too, a non-error reply is signed by the
    /// control plane for the requester; error documents are not.
    #[tokio::test]
    async fn envelope_replies_are_signed_by_the_control_plane() {
        use did_hosting_common::server::identity::ServiceIdentity;
        use did_hosting_common::server::trust_tasks::verify_sender_bound;

        let (mut state, _dir) = signing_state().await;
        let (control, control_key) = crate::signing::test_util::did_key_signer(&[71u8; 32]);
        let mut cfg = (*state.config).clone();
        cfg.server_did = Some(control.clone());
        state.config = Arc::new(cfg);
        state.identity = Some(
            ServiceIdentity::from_signing_secret(&control, control_key)
                .await
                .unwrap(),
        );
        let (admin, admin_key) = crate::signing::test_util::did_key_signer(&[72u8; 32]);
        seed_role(&state, &admin, Role::Admin).await;

        let mut doc = request_doc(MSG_DID_REQUEST, Some(&admin), json!({ "path": "bob" }));
        doc["recipient"] = json!(control);
        let reply = dispatch_envelope(&state, &admin, sign(doc, &admin_key).await).await;
        let reply: trust_tasks_rs::TrustTask<Value> = serde_json::from_value(reply).unwrap();
        assert!(
            reply
                .type_uri
                .to_string()
                .ends_with("check-name/0.1#response")
        );
        verify_sender_bound(
            &reply,
            Some(&control),
            None,
            &admin,
            state.trust_tasks_verifier.as_deref().unwrap(),
        )
        .await
        .expect("reply is signed by the control plane for the requester");

        // An error reply (unsigned request) stays unsigned.
        let mut unsigned = request_doc(MSG_DID_REQUEST, Some(&admin), json!({ "path": "bob" }));
        unsigned["recipient"] = json!(control);
        let err = dispatch_envelope(&state, &admin, unsigned).await;
        assert!(err["type"].as_str().unwrap().contains("/trust-task-error/"));
        assert!(err.get("proof").is_none());
    }

    /// A queued op is settled only by its target's *signed* acknowledgement
    /// of the exact document sent; an ack from another Service-role server
    /// does not settle it.
    #[tokio::test]
    async fn an_outbox_entry_is_settled_by_its_targets_signed_ack() {
        let (state, _dir) = signing_state().await;
        let (edge, edge_key) = crate::signing::test_util::did_key_signer(&[81u8; 32]);
        let (other, other_key) = crate::signing::test_util::did_key_signer(&[82u8; 32]);
        seed_role(&state, &edge, Role::Service).await;
        seed_role(&state, &other, Role::Service).await;
        crate::outbox::enqueue(
            &state.store,
            &edge,
            MSG_SYNC_UPDATE,
            json!({"mnemonic": "alice"}),
        )
        .await
        .unwrap();
        let (key, entry) = crate::outbox::list_pending_for_target(&state.store, &edge)
            .await
            .unwrap()
            .remove(0);
        crate::outbox::record_sent(&state.store, key, &entry, "urn:uuid:sent-1")
            .await
            .unwrap();

        let ack = |issuer: &str| {
            let mut doc = request_doc(
                MSG_SYNC_UPDATE_ACK,
                Some(issuer),
                json!({"mnemonic": "alice", "status": "applied"}),
            );
            doc["threadId"] = json!("urn:uuid:sent-1");
            doc
        };
        // Unsigned, and signed by another server: nothing settled.
        super::run_trust_tasks_envelope(&state, &edge, &envelope(ack(&edge)))
            .await
            .unwrap();
        super::run_trust_tasks_envelope(
            &state,
            &other,
            &envelope(sign(ack(&other), &other_key).await),
        )
        .await
        .unwrap();
        assert_eq!(
            crate::outbox::list_pending_for_target(&state.store, &edge)
                .await
                .unwrap()
                .len(),
            1
        );
        // The target's signed ack settles it.
        super::run_trust_tasks_envelope(
            &state,
            &edge,
            &envelope(sign(ack(&edge), &edge_key).await),
        )
        .await
        .unwrap();
        assert!(
            crate::outbox::list_pending_for_target(&state.store, &edge)
                .await
                .unwrap()
                .is_empty()
        );
    }

    /// Regression: a signer with no ACL entry is turned away before any
    /// signature work, and leaves nothing in the replay cache.
    #[tokio::test]
    async fn an_unauthorised_signer_does_not_populate_the_replay_cache() {
        let (state, _dir) = signing_state().await;
        let (stranger, key) = crate::signing::test_util::did_key_signer(&[83u8; 32]);
        let doc = sign(
            request_doc(MSG_DID_REQUEST, Some(&stranger), json!({ "path": "alice" })),
            &key,
        )
        .await;
        for _ in 0..2 {
            let reply = dispatch_envelope(&state, &stranger, doc.clone()).await;
            assert_eq!(reply["payload"]["code"], "permissionDenied", "{reply}");
        }
        assert!(state.replay_cache.is_empty());
    }

    /// Fail closed: a control plane that cannot sign its replies serves no
    /// trust task — not even the proofless ones — and acts on nothing.
    #[tokio::test]
    async fn a_control_plane_without_a_signing_identity_refuses_every_trust_task() {
        let (mut state, _dir) = signing_state().await;
        state.identity = None;
        let (admin, admin_key) = crate::signing::test_util::did_key_signer(&[95u8; 32]);
        seed_role(&state, &admin, Role::Admin).await;

        let signed = sign(
            request_doc(
                MSG_DID_REQUEST,
                Some(&admin),
                json!({ "path": "bob", "reserve": true }),
            ),
            &admin_key,
        )
        .await;
        let challenge = request_doc(
            "https://trusttasks.org/spec/auth/challenge/0.1",
            Some(&admin),
            json!({ "purpose": "login" }),
        );
        for doc in [signed, challenge] {
            let reply = dispatch_envelope(&state, &admin, doc).await;
            assert_eq!(reply["payload"]["code"], "internalError", "{reply}");
            assert!(reply.get("proof").is_none());
        }
        assert!(
            state
                .dids_ks
                .get::<did_hosting_common::did_ops::DidRecord>(did_key("bob"))
                .await
                .unwrap()
                .is_none(),
            "nothing was reserved"
        );
    }

    fn register_payload() -> Value {
        json!({
            "public_url": "https://edge.example",
            "label": "edge",
            "trust_task_capable": true,
        })
    }

    /// A server registers only on its own signature. Unsigned, the document is
    /// refused even when the reported sender holds the Service role — the
    /// registration would otherwise enrol the sender as a sync target for every
    /// tenant's DIDs on a name alone.
    #[tokio::test]
    async fn unsigned_server_registration_is_refused() {
        let (state, _dir) = signing_state().await;
        let edge = "did:example:edge";
        seed_role(&state, edge, Role::Service).await;

        let doc = request_doc(MSG_SERVER_REGISTER, Some(edge), register_payload());
        let resp = dispatch_envelope(&state, edge, doc).await;
        assert_eq!(resp["payload"]["code"], "proofRequired", "{resp}");
        assert!(
            crate::registry::get_instance(&state.registry_ks, &edge.replace(':', "_"))
                .await
                .unwrap()
                .is_none(),
            "no instance was registered"
        );
    }

    /// The normal flow: a Service-role edge signs its registration and is
    /// registered and acknowledged.
    #[tokio::test]
    async fn signed_server_registration_is_accepted() {
        let (state, _dir) = signing_state().await;
        let (edge, edge_key) = crate::signing::test_util::did_key_signer(&[41u8; 32]);
        seed_role(&state, &edge, Role::Service).await;

        let doc = sign(
            request_doc(MSG_SERVER_REGISTER, Some(&edge), register_payload()),
            &edge_key,
        )
        .await;
        let resp = dispatch_envelope(&state, &edge, doc).await;
        assert_eq!(resp["type"], MSG_SERVER_REGISTER_ACK, "{resp}");
        assert!(
            crate::registry::get_instance(&state.registry_ks, &edge.replace(':', "_"))
                .await
                .unwrap()
                .is_some(),
            "the signing edge was registered"
        );
    }

    /// A signed pong from a DID without the Service role changes nothing: the
    /// registry's liveness view is not writable by any DID that can sign.
    #[tokio::test]
    async fn health_pong_from_a_non_service_signer_is_ignored() {
        use crate::registry::{ServiceInstance, ServiceStatus, ServiceType};

        let (state, _dir) = signing_state().await;
        let (edge, edge_key) = crate::signing::test_util::did_key_signer(&[42u8; 32]);
        seed_role(&state, &edge, Role::Owner).await;
        let instance_id = edge.replace(':', "_");
        crate::registry::register_instance(
            &state.registry_ks,
            &ServiceInstance {
                instance_id: instance_id.clone(),
                service_type: ServiceType::Server,
                label: None,
                url: "https://edge.example".into(),
                status: ServiceStatus::Unreachable,
                last_health_check: None,
                registered_at: 0,
                metadata: json!({ "did": edge }),
                enabled_methods: vec![],
                served_domains: vec![],
                protocol_version: "1.0".into(),
                advertised_services: None,
                services_checked_at: None,
                trust_task_capable: true,
                sync_batch_capable: false,
                last_inbound_transport: None,
                last_inbound_at: None,
                last_outbound_transport: None,
                last_outbound_at: None,
                last_ack_at: None,
                last_reconcile_at: None,
            },
        )
        .await
        .unwrap();

        let doc = sign(
            request_doc(MSG_HEALTH_PONG, Some(&edge), json!({ "status": "ok" })),
            &edge_key,
        )
        .await;
        let out = super::run_trust_tasks_envelope(&state, &edge, &envelope(doc))
            .await
            .expect("dispatch ok");
        assert!(out.is_none(), "a pong is terminal");
        let inst = crate::registry::get_instance(&state.registry_ks, &instance_id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(inst.status, ServiceStatus::Unreachable, "status unchanged");
    }

    #[tokio::test]
    async fn trust_tasks_envelope_returns_none_when_server_did_unconfigured() {
        // server_did = None — the dispatch can't run §7.2 recipient
        // enforcement and bubbles a DIDCommServiceError::Internal.
        let (mut state, _dir) = test_state().await;
        let cfg = AppConfig {
            features: state.config.features.clone(),
            server_did: None,
            mediator_did: None,
            public_url: state.config.public_url.clone(),
            did_hosting_url: state.config.did_hosting_url.clone(),
            server: state.config.server.clone(),
            log: state.config.log.clone(),
            store: state.config.store.clone(),
            fjall: state.config.fjall,
            auth: state.config.auth.clone(),
            secrets: state.config.secrets.clone(),
            vta: state.config.vta.clone(),
            registry: state.config.registry.clone(),
            trust_tasks: state.config.trust_tasks.clone(),
            hosting: state.config.hosting.clone(),
            identity: Default::default(),
            config_path: state.config.config_path.clone(),
        };
        state.config = Arc::new(cfg);

        let msg = envelope(request_doc(
            "https://trusttasks.org/spec/acl/list/0.1",
            Some("did:example:admin"),
            list_payload(),
        ));
        let err = super::run_trust_tasks_envelope(&state, "did:example:admin", &msg)
            .await
            .expect_err("missing server_did should fail");
        match err {
            DIDCommServiceError::Internal(msg) => {
                assert!(msg.contains("server_did"), "operator-actionable: {msg}");
            }
            other => panic!("expected Internal, got {other:?}"),
        }
    }

    use affinidi_data_integrity::DidKeyResolver;
    use did_hosting_common::server::trust_tasks::TransportBoundVerifier;

    use crate::signing::test_util::did_key_signer;

    // ── Signed REQUIRED-spec envelope dispatch (`acl/grant/0.1`) ──────

    /// Build an (unsigned) typed `acl/grant/0.1` document granting
    /// `subject` the `owner` role, issued by `admin_did` to this test
    /// control plane.
    fn unsigned_grant_doc(admin_did: &str, subject: &str) -> Value {
        use trust_tasks_rs::specs::acl::grant::v0_1 as grant;
        let entry: grant::AclEntry = grant::AclEntry::builder()
            .subject(subject)
            .role("owner")
            .scopes(vec![])
            .label(Some(
                "signed e2e grant"
                    .parse::<grant::AclEntryLabel>()
                    .expect("label"),
            ))
            .ext(
                serde_json::from_value::<Option<grant::Ext>>(json!({
                    "vnd.affinidi.webvh": { "domains": { "kind": "all" } }
                }))
                .expect("webvh ext parses"),
            )
            .try_into()
            .expect("e2e ACL entry is well formed");
        let payload: grant::Payload = grant::Payload::builder()
            .entry(entry)
            .reason(Some(
                "signed end-to-end grant"
                    .parse::<grant::PayloadReason>()
                    .expect("reason"),
            ))
            .try_into()
            .expect("e2e grant payload is well formed");
        let mut doc = trust_tasks_rs::TrustTask::for_payload(
            format!("urn:uuid:{}", uuid::Uuid::new_v4()),
            payload,
        );
        doc.issuer = Some(admin_did.into());
        doc.recipient = Some("did:webvh:test:control.example.com".into());
        doc.issued_at = Some(chrono::Utc::now());
        serde_json::to_value(&doc).expect("grant document serialises")
    }

    /// `test_state` + the production `did:key` verifier + `admin_did`
    /// seeded as Admin, for the REQUIRED-spec envelope tests.
    async fn grant_state(admin_did: &str) -> (AppState, tempfile::TempDir) {
        let (mut state, dir) = test_state().await;
        state.trust_tasks_verifier = Some(Arc::new(TransportBoundVerifier::with_resolver(
            Arc::new(DidKeyResolver),
        )));
        assert!(
            state.config.trust_tasks.enforce_proofs,
            "strict proofs are the default"
        );
        store_acl_entry(
            &state.acl_ks,
            &AclEntry {
                did: admin_did.into(),
                role: Role::Admin,
                label: None,
                created_at: 1_700_000_000,
                max_total_size: None,
                max_did_count: None,
                domains: did_hosting_common::server::domain::DomainScope::All,
            },
        )
        .await
        .expect("seed admin");
        (state, dir)
    }

    /// The signed REQUIRED-spec dispatch the retired deferral named:
    /// `acl/grant/0.1` (proof REQUIRED) carrying a *real* Data Integrity
    /// proof, dispatched through the DIDComm envelope path under
    /// `enforce_proofs = true` and the production verifier, landing the
    /// granted entry in the ACL store.
    #[tokio::test]
    async fn trust_tasks_envelope_signed_acl_grant_verifies_end_to_end() {
        let (admin_did, admin_signer) = did_key_signer(&[41u8; 32]);
        let (state, _dir) = grant_state(&admin_did).await;

        let subject = "did:example:grantee";
        let signed = crate::signing::test_util::sign_operational(
            unsigned_grant_doc(&admin_did, subject),
            &admin_signer,
        )
        .await;
        let msg = build_msg(trust_tasks_didcomm::ENVELOPE_TYPE, signed);

        let (resp_type, resp_body) = super::run_trust_tasks_envelope(&state, &admin_did, &msg)
            .await
            .expect("dispatch ok")
            .expect("a response is emitted");

        assert_eq!(resp_type, trust_tasks_didcomm::ENVELOPE_TYPE);
        let inner = resp_body["type"].as_str().unwrap();
        assert!(
            inner.ends_with("/acl/grant/0.1#response"),
            "expected grant response, got {inner}"
        );
        let stored = did_hosting_common::server::acl::get_acl_entry(&state.acl_ks, subject)
            .await
            .expect("acl read")
            .expect("granted entry stored");
        assert!(
            matches!(stored.role, Role::Owner),
            "granted role is owner, got {:?}",
            stored.role
        );
    }

    /// Tampering with a signed grant after signing is refused with
    /// `proofInvalid` — and nothing lands in the ACL store.
    #[tokio::test]
    async fn trust_tasks_envelope_signed_acl_grant_tampered_is_rejected() {
        let (admin_did, admin_signer) = did_key_signer(&[42u8; 32]);
        let (state, _dir) = grant_state(&admin_did).await;

        let mut signed = crate::signing::test_util::sign_operational(
            unsigned_grant_doc(&admin_did, "did:example:grantee"),
            &admin_signer,
        )
        .await;
        signed["payload"]["entry"]["subject"] = json!("did:example:mallory");
        let msg = build_msg(trust_tasks_didcomm::ENVELOPE_TYPE, signed);

        let (_, resp_body) = super::run_trust_tasks_envelope(&state, &admin_did, &msg)
            .await
            .expect("dispatch ok")
            .expect("an error document is emitted");

        assert_eq!(
            resp_body["type"],
            // Takes the version from the emitter rather than naming one, so a
            // framework bump moves the assertion with the code under test.
            // What this test is about is the `code`, not the document version.
            did_hosting_common::server::trust_tasks::framework_error_type_uri().to_string()
        );
        assert_eq!(resp_body["payload"]["code"], "proofInvalid");
        for did in ["did:example:mallory", "did:example:grantee"] {
            assert!(
                did_hosting_common::server::acl::get_acl_entry(&state.acl_ks, did)
                    .await
                    .expect("acl read")
                    .is_none(),
                "{did} must not be granted"
            );
        }
    }

    /// A proofless grant is refused (`proofRequired`) even from a
    /// seeded Admin sender — for REQUIRED specs the proof, not the
    /// transport session, authorizes.
    #[tokio::test]
    async fn trust_tasks_envelope_unsigned_acl_grant_is_rejected() {
        let (admin_did, _admin_signer) = did_key_signer(&[43u8; 32]);
        let (state, _dir) = grant_state(&admin_did).await;

        let msg = build_msg(
            trust_tasks_didcomm::ENVELOPE_TYPE,
            unsigned_grant_doc(&admin_did, "did:example:grantee"),
        );
        let (_, resp_body) = super::run_trust_tasks_envelope(&state, &admin_did, &msg)
            .await
            .expect("dispatch ok")
            .expect("an error document is emitted");

        assert_eq!(
            resp_body["type"],
            // Takes the version from the emitter rather than naming one, so a
            // framework bump moves the assertion with the code under test.
            // What this test is about is the `code`, not the document version.
            did_hosting_common::server::trust_tasks::framework_error_type_uri().to_string()
        );
        assert_eq!(resp_body["payload"]["code"], "proofRequired");
        assert!(
            did_hosting_common::server::acl::get_acl_entry(&state.acl_ks, "did:example:grantee")
                .await
                .expect("acl read")
                .is_none(),
            "nothing must be granted"
        );
    }

    // ── Delta re-sync compares identity, not just version ──────────────

    fn record_at(mnemonic: &str, did_id: &str, version_count: u64) -> DidRecord {
        DidRecord {
            services: None,
            owner: "did:example:owner".into(),
            mnemonic: mnemonic.into(),
            created_at: 1,
            updated_at: 1,
            version_count,
            did_id: Some(did_id.into()),
            content_size: 1,
            disabled: false,
            deleted_at: None,
            method: "webvh".into(),
            domain: String::new(),
            agent_names: Vec::new(),
        }
    }

    #[test]
    fn an_edge_is_current_only_for_the_same_did_at_the_same_version() {
        use crate::server_push::{ReportedDid, edge_is_current};
        let record = record_at("alice", "did:webvh:QmNew:host:alice", 2);
        let have = |did_id: Option<&str>, version_count| ReportedDid {
            did_id: did_id.map(String::from),
            version_count,
            disabled: None,
        };
        assert!(edge_is_current(
            &record,
            Some(&have(Some("did:webvh:QmNew:host:alice"), 2))
        ));
        assert!(!edge_is_current(&record, None), "absent → push");
        assert!(
            !edge_is_current(&record, Some(&have(Some("did:webvh:QmNew:host:alice"), 1))),
            "behind → push"
        );
        // Deleted and re-created at the same slot with the same version count:
        // the edge that missed the delete still holds the old DID.
        assert!(
            !edge_is_current(&record, Some(&have(Some("did:webvh:QmOld:host:alice"), 2))),
            "a different DID at the slot → push"
        );
        assert!(
            !edge_is_current(&record, Some(&have(None, 5))),
            "an edge that does not say which DID it holds → push"
        );
        // A disable the edge missed is re-sent with the content.
        let reported = |disabled| ReportedDid {
            disabled: Some(disabled),
            ..have(Some("did:webvh:QmNew:host:alice"), 2)
        };
        assert!(edge_is_current(&record, Some(&reported(false))));
        assert!(
            !edge_is_current(&record, Some(&reported(true))),
            "a different disabled state → push"
        );
    }

    #[test]
    fn preloaded_dids_carry_the_did_identity() {
        let reported = crate::server_push::parse_reported(&json!({
            "preloaded_dids": [
                { "mnemonic": "alice", "did_id": "did:webvh:QmA:h:alice", "version_count": 3 },
                { "mnemonic": "bob", "version_count": 1 },
            ]
        }));
        assert_eq!(
            reported["alice"].did_id.as_deref(),
            Some("did:webvh:QmA:h:alice")
        );
        assert_eq!(reported["alice"].version_count, 3);
        assert_eq!(reported["bob"].did_id, None);
    }

    /// End to end: an edge reporting the old DID at a re-created slot, at the
    /// same version count, is sent the new log.
    #[tokio::test]
    async fn resync_pushes_a_re_created_did_at_the_same_version() {
        let (state, _dir) = test_state().await;
        let record = record_at("alice", "did:webvh:QmNew:host:alice", 1);
        state
            .dids_ks
            .insert(did_key("alice"), &record)
            .await
            .unwrap();
        state
            .dids_ks
            .insert_raw(
                did_hosting_common::did_ops::content_log_key("alice"),
                b"{}".to_vec(),
            )
            .await
            .unwrap();
        let edge = "did:example:edge-resync";
        let reported = crate::server_push::parse_reported(&json!({
            "preloaded_dids": [
                { "mnemonic": "alice", "did_id": "did:webvh:QmOld:host:alice", "version_count": 1 },
            ]
        }));
        crate::server_push::sync_all_dids_to_server(&state, edge.to_string(), reported);
        let mut rows = Vec::new();
        for _ in 0..100 {
            rows = crate::outbox::list_pending_for_target(&state.store, edge)
                .await
                .unwrap();
            if !rows.is_empty() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        assert_eq!(rows.len(), 1, "the new DID is pushed");
        assert_eq!(rows[0].1.msg_type, MSG_SYNC_UPDATE);
        assert_eq!(rows[0].1.body["didId"], "did:webvh:QmNew:host:alice");
        assert_eq!(rows[0].1.body["disabled"], false);
    }

    // ── Domain ops are re-sent on registration until settled ───────────

    async fn seed_service(state: &AppState, did: &str) {
        store_acl_entry(
            &state.acl_ks,
            &AclEntry {
                did: did.into(),
                role: Role::Service,
                label: None,
                created_at: 1,
                max_total_size: None,
                max_did_count: None,
                domains: did_hosting_common::server::domain::DomainScope::All,
            },
        )
        .await
        .unwrap();
    }

    async fn pending_ops(state: &AppState, edge: &str) -> Vec<(String, String)> {
        crate::outbox::list_pending_for_target(&state.store, edge)
            .await
            .unwrap()
            .into_iter()
            .map(|(_, e)| {
                (
                    e.msg_type,
                    e.body["domain"].as_str().unwrap_or_default().to_string(),
                )
            })
            .collect()
    }

    async fn clear_outbox(state: &AppState, edge: &str) {
        for (k, _) in crate::outbox::list_pending_for_target(&state.store, edge)
            .await
            .unwrap()
        {
            crate::outbox::remove(&state.store, k).await.unwrap();
        }
    }

    #[tokio::test]
    async fn unsettled_domain_ops_are_re_sent_on_registration() {
        use crate::server_push::{
            DomainOp, domain_intents, resend_domain_intents, send_domain_assign, send_domain_purge,
            send_domain_unassign,
        };
        let (state, _dir) = test_state().await;
        let edge = "did:example:edge-domains";
        seed_service(&state, edge).await;

        send_domain_assign(&state, edge, "kept.example")
            .await
            .unwrap();
        send_domain_unassign(&state, edge, "gone.example")
            .await
            .unwrap();
        send_domain_purge(&state, edge, "wiped.example")
            .await
            .unwrap();
        // The edge was offline long enough for the outbox to give up.
        clear_outbox(&state, edge).await;

        resend_domain_intents(&state, edge).await;
        let mut ops = pending_ops(&state, edge).await;
        ops.sort();
        let mut want = vec![
            (
                MSG_REPLICA_DOMAIN_ASSIGN.to_string(),
                "kept.example".to_string(),
            ),
            (
                MSG_REPLICA_DOMAIN_UNASSIGN.to_string(),
                "gone.example".to_string(),
            ),
            (
                MSG_REPLICA_DOMAIN_UNASSIGN.to_string(),
                "wiped.example".to_string(),
            ),
            (
                MSG_REPLICA_DOMAIN_PURGE.to_string(),
                "wiped.example".to_string(),
            ),
        ];
        want.sort();
        assert_eq!(ops, want);

        // Acknowledged unassign / purge are settled; the assign stays (it is the
        // desired state).
        do_domain_ack(
            &state,
            edge,
            MSG_REPLICA_DOMAIN_UNASSIGN_ACK,
            &json!({ "domain": "gone.example", "status": "scheduled", "purgeAt": "2026-01-01T00:00:00Z" }),
        )
        .await;
        do_domain_ack(
            &state,
            edge,
            MSG_REPLICA_DOMAIN_PURGE_ACK,
            &json!({ "domain": "wiped.example", "status": "purged", "removed": 0 }),
        )
        .await;
        do_domain_ack(
            &state,
            edge,
            MSG_REPLICA_DOMAIN_ASSIGN_ACK,
            &json!({ "domain": "kept.example", "status": "applied" }),
        )
        .await;
        let intents = domain_intents(&state, edge).await.unwrap();
        assert_eq!(intents.len(), 1, "{intents:?}");
        assert_eq!(intents[0].domain, "kept.example");
        assert_eq!(intents[0].op, DomainOp::Assign);
    }

    /// The ack of an older unassign cannot erase a newer assign, and a stranger
    /// cannot settle anything.
    #[tokio::test]
    async fn a_stale_or_foreign_ack_does_not_settle_a_domain_op() {
        use crate::server_push::{
            DomainOp, domain_intents, send_domain_assign, send_domain_unassign,
        };
        let (state, _dir) = test_state().await;
        let edge = "did:example:edge-stale";
        seed_service(&state, edge).await;

        send_domain_unassign(&state, edge, "a.example")
            .await
            .unwrap();
        send_domain_assign(&state, edge, "a.example").await.unwrap();
        do_domain_ack(
            &state,
            edge,
            MSG_REPLICA_DOMAIN_UNASSIGN_ACK,
            &json!({ "domain": "a.example", "status": "scheduled", "purgeAt": "2026-01-01T00:00:00Z" }),
        )
        .await;
        let intents = domain_intents(&state, edge).await.unwrap();
        assert_eq!(intents[0].op, DomainOp::Assign, "the newer assign stands");

        send_domain_unassign(&state, edge, "a.example")
            .await
            .unwrap();
        do_domain_ack(
            &state,
            "did:example:not-a-service",
            MSG_REPLICA_DOMAIN_UNASSIGN_ACK,
            &json!({ "domain": "a.example", "status": "scheduled", "purgeAt": "2026-01-01T00:00:00Z" }),
        )
        .await;
        assert_eq!(
            domain_intents(&state, edge).await.unwrap()[0].op,
            DomainOp::Unassign,
            "another DID's ack settles nothing"
        );
    }
}