vta-service 0.25.0

use std::sync::Arc;
use std::time::Duration;

use affinidi_did_resolver_cache_sdk::{DIDCacheClient, config::DIDCacheConfigBuilder};
use affinidi_tdk::common::TDKSharedState;
use affinidi_tdk::common::config::TDKConfig;
use affinidi_tdk::messaging::ATM;
use affinidi_tdk::messaging::config::ATMConfig;
use affinidi_tdk::secrets_resolver::{SecretsResolver, ThreadedSecretsResolver};
use vti_common::slip10::ExtendedSigningKey;

use base64::Engine;
use base64::engine::general_purpose::URL_SAFE_NO_PAD as BASE64;

use crate::auth::AuthState;
use crate::auth::jwt::JwtKeys;
use crate::auth::session::cleanup_expired_sessions;
use crate::config::{AppConfig, AuthConfig};
#[cfg(any(feature = "didcomm", feature = "tsp"))]
use crate::didcomm_bridge::DIDCommBridge;
use crate::error::AppError;
use crate::keys::KeyRecord;
use crate::keys::derivation::Bip32Extension;
use crate::keys::seed_store::SeedStore;
use crate::keys::seeds::load_seed_bytes;
#[cfg(feature = "rest")]
use crate::routes;
use crate::store::{KeyspaceHandle, Store};
use tokio::sync::{RwLock, watch};
#[cfg(feature = "rest")]
use tower_http::trace::{DefaultMakeSpan, DefaultOnRequest, DefaultOnResponse, TraceLayer};
use tracing::Level;
use tracing::{debug, error, info, warn};

// D2 P2a: the reliable-messaging delivery layer replaces the
// `affinidi-messaging-didcomm-service` framework. `MessagingService` (built in
// `messaging::service`) drives inbound + outbound over a `DidCommTransport`.
#[cfg(any(feature = "didcomm", feature = "tsp"))]
use tokio_util::sync::CancellationToken;
// The mediator ACL is keyed on the VTA DID, so it authorises whichever
// protocol rides the socket — either transport wires it on connect.
#[cfg(any(feature = "didcomm", feature = "tsp"))]
use vta_sdk::acl_setup;

/// TEE context passed by the caller (main.rs or vta-enclave).
/// None when running outside a TEE.
///
/// When the `tee` feature is not compiled in, this is a unit struct
/// that is never constructed — callers pass `None::<TeeContext>`.
#[derive(Clone)]
#[cfg(feature = "tee")]
pub struct TeeContext {
    pub state: crate::tee::TeeState,
    pub mnemonic_guard: Option<Arc<crate::tee::mnemonic_guard::MnemonicExportGuard>>,
}

/// Stub type when TEE is not compiled in. Never constructed.
#[derive(Clone)]
#[cfg(not(feature = "tee"))]
pub struct TeeContext(());

/// Trigger a soft restart after a short delay, allowing the current
/// response to be sent before threads shut down.
pub fn trigger_restart(restart_tx: &watch::Sender<bool>) {
    let tx = restart_tx.clone();
    tokio::spawn(async move {
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        let _ = tx.send(true);
    });
}

/// Shared handle graph for every route, task handler and sweeper.
///
/// **`#[non_exhaustive]` is load-bearing, not decoration.** This struct is 46
/// public fields with no private field and no constructor guard, so before
/// this attribute any crate could write an `AppState { .. }` literal — which
/// made *adding a field* a source-breaking change under
/// `constructible_struct_adds_field`. Twenty-nine commits have added at least
/// one field to it; one added five.
///
/// That break was never reported, because `cargo semver-checks` cannot build
/// this crate's baseline (its published dependency ranges resolve two
/// `trust-tasks-rs` versions into one graph) and a crate whose baseline fails
/// to build is silently not compared. Version numbers stayed correct anyway
/// — at 0.x a break needs a minor bump and conventional commits already force
/// one for `feat:`, which every field addition since release-plz happened to
/// be. Both of those are coincidences: `refactor:`/`fix:`/`perf:`/`chore:`
/// yield a patch and can add fields just as easily (c93c7b57 added five under
/// `refactor:`), and the alignment disappears entirely at 1.0, where a break
/// needs a major and `feat:` gives only a minor.
///
/// Marking it non-exhaustive closes the class permanently rather than relying
/// on any of that holding. Construction inside this crate is unaffected —
/// both sites (`build_state` below and `test_support`) are in-crate, and
/// `MockVta` is the supported entry point for consumers.
#[derive(Clone)]
#[non_exhaustive]
pub struct AppState {
    pub keys_ks: KeyspaceHandle,
    pub sessions_ks: KeyspaceHandle,
    pub acl_ks: KeyspaceHandle,
    pub contexts_ks: KeyspaceHandle,
    pub did_templates_ks: KeyspaceHandle,
    /// The audit keyspace, for **reads and retention only** — the audit-list
    /// query and `cleanup_expired_logs`. Writes go through [`Self::audit_sink`].
    ///
    /// The two are separate because an alternative sink may be write-only and
    /// remote, and "delete rows older than N days" is not an operation an
    /// append-only or anchored backend can offer — the immutability is the
    /// point. Retention therefore stays a property of the local keyspace, which
    /// is what the retention API is documented to govern.
    pub audit_ks: KeyspaceHandle,
    /// Pluggable audit sink — every audit *write* goes through this.
    /// Default impl is the `audit` keyspace ([`KeyspaceAuditSink`]);
    /// an append-only log, transparency log, or hash chain plugs in via the
    /// `AuditSink` trait without any call site changing (#1031).
    pub audit_sink: vta_audit::SharedAuditSink,
    pub imported_ks: KeyspaceHandle,
    /// Non-extractable internal signing keys. Separate from `imported_ks`
    /// because that keyspace wraps under a seed-derived KEK; these must have
    /// no path back to the mnemonic.
    pub internal_ks: KeyspaceHandle,
    pub cache_ks: KeyspaceHandle,
    /// Vault — third-party credentials the holder has stored on this VTA.
    /// M1 reads only; upsert/delete/sync/release land in M2+. Encrypted at
    /// rest like every other secret-bearing keyspace.
    pub vault_ks: KeyspaceHandle,
    /// Persistent runtime state for service enable/disable
    /// (`operations::protocol::runtime_state`). Replaces the legacy
    /// `[services]` block in `config.toml` as the source of truth for whether
    /// REST / DIDComm are currently active.
    pub service_state_ks: KeyspaceHandle,
    /// Anti-replay log for sealed-bootstrap `bundle_id`s. One row per seal;
    /// `PersistentNonceStore` refuses duplicates.
    pub sealed_nonces_ks: KeyspaceHandle,
    /// Idempotency records for keyed Trust Tasks (`trust_tasks::idempotency`).
    /// One row per `(actor, idempotency-key)`, so a client's retry of a lost
    /// reply converges on the first execution instead of producing a second
    /// durable effect.
    pub idempotency_ks: KeyspaceHandle,
    /// In-flight backup-bundle records for the descriptor-pattern
    /// export/import slice (see
    /// `docs/05-design-notes/backup-descriptor-pattern.md`). Holds
    /// only the control-plane state — `.vtabak` bytes live on disk
    /// under [`Self::backup_blob_dir`]. Encrypted at rest (records
    /// include hashed bearer tokens; nothing useful leaks if the
    /// keyspace is read, but encrypting it keeps storage-layer
    /// invariants uniform across slices).
    pub backup_bundles_ks: KeyspaceHandle,
    /// Filesystem directory under which `.vtabak` byte blobs are
    /// staged for in-flight backup bundles. Created lazily at first
    /// `initiate-*` call. Permissions: 0700 (owner-only). Each
    /// blob is at `{backup_blob_dir}/{bundle_id}.vtabak` with mode
    /// 0600. The sweeper deletes both the file and the record
    /// when a bundle ages out.
    pub backup_blob_dir: std::path::PathBuf,
    #[cfg(feature = "webvh")]
    pub webvh_ks: KeyspaceHandle,
    /// IACA roots this VTA accepts as ISO 18013-5 mdoc issuers, parsed once at
    /// boot from `[vault] mdoc_iaca_trust_anchors`.
    ///
    /// Empty unless configured, and the resolver fails closed on empty — mdoc
    /// is the one credential format whose issuer is not a resolvable DID, so
    /// there is no safe default to fall back to.
    pub mdoc_trust: Arc<vta_vault::mdoc_trust::IacaTrustAnchors>,
    /// In-flight WebAuthn registration state for the
    /// passkey-as-verificationMethod enrolment ceremony. Holds
    /// `PasskeyRegistration` keyed by ceremony id; consumed (taken)
    /// at finish.
    #[cfg(feature = "webvh")]
    pub passkey_vms_ks: KeyspaceHandle,
    /// Inbound-messaging consent store (grants + pending requests).
    pub consent_ks: KeyspaceHandle,
    /// Per-(platform, context) approver bindings for consent routing.
    pub consent_approvers_ks: KeyspaceHandle,
    /// VTA-issued credentials minted via `vta/credentials/issue/0.1` and
    /// revoked via `vta/credentials/revoke/0.1`. Keyed `cred:<id>`; revoke is a
    /// tombstone (`revokedAt`), not a delete.
    pub issued_credentials_ks: KeyspaceHandle,
    /// Per-context key/value store for AI-agent memory (`vta/memory/{put,list,
    /// delete}/0.1`). Entries keyed `mem:<contextId>:<key>`; gated on context
    /// access. Durable user data.
    pub memory_ks: KeyspaceHandle,
    /// A member's MLS group for each room they belong to. Group secrets — see
    /// [`crate::keyspaces::ROOM_GROUPS`].
    pub room_groups_ks: KeyspaceHandle,
    /// Invitations already spent joining a room.
    pub room_invitations_ks: KeyspaceHandle,
    /// Versioned, namespaced application state (`vta/app-state/*`) — the third
    /// store, beside the vault and agent memory, for JSON an application owns
    /// and the VTA does not interpret. Records keyed
    /// `app:<contextId>:<namespace>:<key>`, with a `appv:` version index and an
    /// `appc:` per-namespace write counter alongside; gated on context access.
    /// Durable user data an account's recoverability depends on.
    pub app_state_ks: KeyspaceHandle,
    /// The holder's persona store. Encrypted at rest like the vault: these are
    /// the person's own identity attributes, not application data.
    pub persona_ks: KeyspaceHandle,
    /// Per-agent key for the persona correlation index's keyed hash.
    ///
    /// Derived beside the at-rest key and never leaving the agent, which is
    /// what makes the blinded index blinded: the store can answer "does this
    /// value appear elsewhere" without holding a plaintext index of the
    /// holder's personal data.
    pub persona_correlation_key: [u8; 32],
    /// One lock per `(contextId, namespace)`, serialising the read-modify-write
    /// sequences the application-state store needs and that the store layer
    /// cannot make atomic on its own (it serialises individual operations, not
    /// sequences of them). See `operations::app_state` for why this is sound
    /// for a single-writer VTA and what it does not cover.
    pub app_state_locks: crate::operations::app_state::NamespaceLocks,
    /// Rego policy modules for the Policy Decision Point (`policy/*`). One
    /// [`crate::policy::PolicyModule`] per id; the active set is every enabled
    /// row, priority-ordered. A migration-safe baseline is boot-installed if
    /// empty. Durable operator security config.
    pub policy_ks: KeyspaceHandle,
    /// Task-execution consent: pending approvals + granted consents the PDP's
    /// `requireConsent` disposition uses. Distinct from `consent_ks` (messaging).
    pub task_consent_ks: KeyspaceHandle,
    /// Persisted drain set for the protocol-management feature
    /// (`docs/05-design-notes/didcomm-protocol-management.md`).
    /// Keyed by mediator DID; replayed at boot.
    #[cfg(feature = "webvh")]
    pub drains_ks: KeyspaceHandle,
    /// Per-kind previous-config snapshot store for fail-forward
    /// rollback (spec §3.5a). Populated alongside `drains_ks`.
    #[cfg(feature = "webvh")]
    pub snapshot_ks: KeyspaceHandle,
    /// In-process registry of active + draining mediator listeners.
    /// Owns the per-listener bounded outbound buffer and the
    /// active/drain state machine.
    #[cfg(all(feature = "webvh", feature = "didcomm"))]
    pub mediator_registry: Arc<crate::messaging::registry::MediatorListenerRegistry>,
    /// Per-webvh-server async mutex registry for serializing
    /// daemon-REST auth-cache read-modify-writes. Two concurrent
    /// operations against the same server can't both refresh and
    /// last-writer-wins; locks are keyed by server id so unrelated
    /// servers don't contend.
    #[cfg(feature = "webvh")]
    pub webvh_auth_locks: crate::operations::did_webvh::WebvhAuthLocks,
    /// Per-mediator TTL sweeper. Arms a `tokio::time::sleep_until`
    /// task per drain entry; on expiry, calls
    /// `record_expiries_persisted` and signals upstream listener
    /// teardown via the teardown channel.
    #[cfg(all(feature = "webvh", feature = "didcomm"))]
    pub drain_sweeper: Arc<crate::messaging::drain_sweeper::DrainSweeper>,
    /// Pluggable telemetry sink for mediator-attribution events.
    /// Default impl is the in-memory ring buffer; alternative
    /// backends plug in via the `TelemetrySink` trait.
    pub telemetry: vti_common::telemetry::SharedTelemetrySink,
    pub wrapping_cache: crate::keys::wrapping::WrappingKeyCache,
    pub config: Arc<RwLock<AppConfig>>,
    pub seed_store: Arc<dyn SeedStore>,
    pub did_resolver: Option<DIDCacheClient>,
    /// Live status-list resolver for the present path: when set, the holder
    /// **re-resolves** a credential's revocation status at present time rather
    /// than trusting the stored tag (§14.5). `None` falls back to the stored
    /// status (the pre-live behaviour).
    pub status_list_resolver: Option<Arc<dyn crate::vault::status::StatusListResolver>>,
    pub secrets_resolver: Option<Arc<ThreadedSecretsResolver>>,
    /// Verification-method id for the VTA's signing key (e.g.
    /// `{did}#key-0`). Populated by `init_auth`. Needed by the
    /// live mediator-handshake prover to fetch the corresponding
    /// secret out of [`Self::secrets_resolver`].
    ///
    /// **Not feature-gated**, unlike its key-agreement sibling below, because
    /// the Trust-Task spine signs every success response with it and
    /// conformance must not depend on which transports were compiled in. A
    /// `--no-default-features --features rest` build answers the same
    /// specifications as a full one, and those specifications require a proof
    /// on the response (SPEC §7.3 item 7).
    pub signing_vm_id: Option<String>,
    /// Verification-method id for the VTA's key-agreement key
    /// (e.g. `{did}#key-1`). Populated by `init_auth`.
    #[cfg(any(feature = "didcomm", feature = "tsp"))]
    pub ka_vm_id: Option<String>,
    /// Outbound seam over the mediator socket. Present for either transport —
    /// the service it holds multiplexes DIDComm and TSP.
    #[cfg(any(feature = "didcomm", feature = "tsp"))]
    pub didcomm_bridge: Arc<DIDCommBridge>,

    /// Learn-from-inbound TSP reachability: which device DIDs were last seen
    /// sending over TSP, so device-push can prefer TSP over DIDComm for them.
    /// Populated by the inbound TSP dispatcher from the proven `sender_vid`.
    #[cfg(feature = "tsp")]
    pub tsp_reach: Arc<crate::messaging::tsp_reach::TspReachability>,
    pub jwt_keys: Option<Arc<JwtKeys>>,
    pub atm: Option<ATM>,
    /// VTA's registered TSP profile, used to unpack `tsp-message` sealed
    /// envelopes (e.g. on `vault/upsert`). Built in `init_auth` when the `tsp`
    /// feature is on and an ATM is available; `None` if the VTA keys are
    /// missing or profile registration fails (TSP unseal then stays
    /// unavailable, never panics). Unpack reads the decryption key from the
    /// ATM's own secrets resolver, so the profile carries no secrets itself.
    #[cfg(feature = "tsp")]
    pub tsp_profile: Option<std::sync::Arc<affinidi_tdk::messaging::profiles::ATMProfile>>,
    pub tee: Option<TeeContext>,
    /// Send `true` to trigger a soft restart (threads shut down and re-initialize).
    pub restart_tx: watch::Sender<bool>,
    /// Prometheus metrics handle for rendering `/metrics` endpoint.
    #[cfg(feature = "rest")]
    pub metrics_handle: Option<crate::metrics::PrometheusHandle>,
}

impl AppState {
    /// The resolver Trust Task Data-Integrity proofs are verified with.
    ///
    /// Carries the configured DID cache, so a proof by any DID that names a key
    /// — `did:webvh:<scid>:example.com:glenn#key-0` as much as a `did:key` —
    /// resolves. With no cache configured this degrades to `did:key` only,
    /// which is what a deployment with no outbound DID resolution gets.
    pub fn trust_task_vm_resolver(&self) -> vti_common::auth::TrustTaskVmResolver {
        vti_common::auth::TrustTaskVmResolver::from_optional(self.did_resolver.clone())
    }
}

impl AuthState for AppState {
    fn jwt_keys(&self) -> Option<&Arc<JwtKeys>> {
        self.jwt_keys.as_ref()
    }
    fn sessions_ks(&self) -> &KeyspaceHandle {
        &self.sessions_ks
    }
}

/// Run-path-specific shared components injected into [`build_app_state`].
///
/// The full server path (`run()`) builds these *before* the `AppState` because
/// it needs them for drain replay and the teardown-channel consumer, and they
/// must be the live wiring rather than the inert defaults. Non-axum front-ends
/// (Lambda, offline CLI) pass [`AppStateParts::default`] and `build_app_state`
/// fills in self-contained defaults: a fresh telemetry ring buffer + registry,
/// a drain sweeper whose teardown signals go nowhere, a placeholder DIDComm
/// bridge, and no metrics handle.
///
/// Injecting these (rather than letting `run()` assemble its own `AppState`
/// literal) keeps `build_app_state` the single `AppState` constructor — one
/// `WebvhAuthLocks::new()`, one config `RwLock`, one `init_auth` — so the REST
/// and DIDComm transports can't diverge (P1.1).
/// **`#[non_exhaustive]` for the same reason [`AppState`] carries it** (#1024),
/// and the gap that motivated adding it here was found the same way: a field
/// addition landing as a source-breaking change nobody had labelled.
///
/// Every field is public with no private field and no constructor guard, so
/// before this attribute any crate could write an `AppStateParts { .. }`
/// literal — including the functional-update form — which made *adding a
/// field* break them under `constructible_struct_adds_field`. That is not
/// hypothetical: #1049 added `audit_sink` and #1051 nearly added
/// `app_state_locks` before the break was noticed and routed around.
///
/// Construction inside this crate is unaffected. Outside it, `Default` plus
/// field assignment replaces the literal:
///
/// ```ignore
/// let mut parts = AppStateParts::default();
/// parts.audit_sink = Some(sink);
/// ```
///
/// which is what this crate's own integration tests now do — they are separate
/// crates, so they were the first thing the attribute broke, and they are a
/// fair proxy for what a consumer has to change.
#[derive(Default)]
#[non_exhaustive]
pub struct AppStateParts {
    /// Telemetry sink shared with the mediator registry. `None` → fresh ring buffer.
    pub telemetry: Option<vti_common::telemetry::SharedTelemetrySink>,
    /// Audit sink. `None` → the `audit` keyspace, i.e. what the VTA has always
    /// done. An operator wiring tamper-evidence supplies one here — typically a
    /// `FanOutAuditSink` keeping the keyspace so the query API goes on working.
    pub audit_sink: Option<vta_audit::SharedAuditSink>,
    /// Live mediator listener registry. `None` → fresh registry over `telemetry`.
    #[cfg(all(feature = "webvh", feature = "didcomm"))]
    pub mediator_registry: Option<Arc<crate::messaging::registry::MediatorListenerRegistry>>,
    /// Drain sweeper wired to a real teardown channel. `None` → dead-channel no-op sweeper.
    #[cfg(all(feature = "webvh", feature = "didcomm"))]
    pub drain_sweeper: Option<Arc<crate::messaging::drain_sweeper::DrainSweeper>>,
    /// Outbound mediator bridge. `None` → placeholder (no live service).
    #[cfg(any(feature = "didcomm", feature = "tsp"))]
    pub didcomm_bridge: Option<Arc<DIDCommBridge>>,
    /// Prometheus handle for `/metrics`. `None` → no metrics rendering.
    #[cfg(feature = "rest")]
    pub metrics_handle: Option<crate::metrics::PrometheusHandle>,
}

/// Build the shared application state from config, store, and TEE context.
///
/// This is the **single** `AppState` constructor. The server path (`run()`)
/// injects its live shared components via `parts` and then derives the
/// DIDComm-transport `VtaState` from the returned `AppState` (so both
/// transports share the same Arcs); non-axum front-ends (e.g., Lambda handlers)
/// pass `AppStateParts::default()` and manage their own request loop.
pub async fn build_app_state(
    config: AppConfig,
    store: &Store,
    seed_store: Arc<dyn SeedStore>,
    storage_encryption_key: Option<[u8; 32]>,
    tee_context: Option<TeeContext>,
    restart_tx: watch::Sender<bool>,
    parts: AppStateParts,
) -> Result<AppState, AppError> {
    // Parse the configured IACA roots once, here, so a malformed certificate
    // fails the boot rather than surfacing as a puzzling rejection on the first
    // mdoc that arrives. Empty is legal and means "this VTA accepts no mdoc
    // issuers" — the resolver fails closed on it.
    let mdoc_trust = Arc::new(vta_vault::mdoc_trust::IacaTrustAnchors::from_pem(
        &config.vault.mdoc_iaca_trust_anchors,
    )?);

    let apply_encryption = |ks: KeyspaceHandle| -> KeyspaceHandle {
        if let Some(key) = storage_encryption_key {
            ks.with_encryption(key)
        } else {
            ks
        }
    };

    let keys_ks = apply_encryption(store.keyspace(crate::keyspaces::KEYS)?);
    let sessions_ks = apply_encryption(store.keyspace(crate::keyspaces::SESSIONS)?);
    let acl_ks = apply_encryption(store.keyspace(crate::keyspaces::ACL)?);
    let contexts_ks = apply_encryption(store.keyspace(crate::keyspaces::CONTEXTS)?);
    let did_templates_ks = apply_encryption(store.keyspace(crate::keyspaces::DID_TEMPLATES)?);
    let audit_ks = apply_encryption(store.keyspace(crate::keyspaces::AUDIT)?);
    let imported_ks = apply_encryption(store.keyspace(crate::keyspaces::IMPORTED_SECRETS)?);
    let internal_ks = apply_encryption(store.keyspace(crate::keyspaces::INTERNAL_KEYS)?);
    let cache_ks = apply_encryption(store.keyspace(crate::keyspaces::CACHE)?);
    let vault_ks = apply_encryption(store.keyspace(crate::keyspaces::VAULT)?);
    // Persistent runtime state for service enable/disable. Encrypted because
    // a couple of bool records are cheap and the keyspace may grow.
    let service_state_ks = apply_encryption(store.keyspace(crate::keyspaces::SERVICE_STATE)?);
    // Sealed-transfer anti-replay store. Bundle_ids are not secret and the
    // row is a one-byte sentinel, so the keyspace is intentionally
    // unencrypted — saves a decrypt hop on every request.
    let sealed_nonces_ks = apply_encryption(store.keyspace(crate::keyspaces::SEALED_NONCES)?);
    // Trust-Task idempotency records. Encrypted like every other keyspace that
    // may hold response bodies — a `Keyed` record carries the original response
    // verbatim, which is caller data even when it is not secret material.
    let idempotency_ks = apply_encryption(store.keyspace(crate::keyspaces::IDEMPOTENCY)?);
    let backup_bundles_ks = apply_encryption(store.keyspace(crate::keyspaces::BACKUP_BUNDLES)?);
    // Stage `.vtabak` blobs under `{data_dir}/backups`. Created lazily
    // by the op layer at first `initiate-*` call (so a VTA that never
    // does backups doesn't get an empty directory). See
    // `docs/05-design-notes/backup-descriptor-pattern.md` §"State
    // machine" for the file-system layout.
    let backup_blob_dir = config.store.data_dir.join("backups");
    #[cfg(feature = "webvh")]
    let webvh_ks = apply_encryption(store.keyspace(crate::keyspaces::WEBVH)?);
    #[cfg(feature = "webvh")]
    let passkey_vms_ks = apply_encryption(store.keyspace(crate::keyspaces::PASSKEY_VMS)?);
    let consent_ks = apply_encryption(store.keyspace(crate::keyspaces::CONSENT)?);
    let consent_approvers_ks =
        apply_encryption(store.keyspace(crate::keyspaces::CONSENT_APPROVERS)?);
    let issued_credentials_ks =
        apply_encryption(store.keyspace(crate::keyspaces::ISSUED_CREDENTIALS)?);
    let memory_ks = apply_encryption(store.keyspace(crate::keyspaces::MEMORY)?);
    // Encrypted alongside the key store where a TEE deployment provides a storage key: a
    // group snapshot decrypts every record its room holds, which is the same class of
    // material as a derived key and deserves the same treatment.
    let room_groups_ks = apply_encryption(store.keyspace(crate::keyspaces::ROOM_GROUPS)?);
    let room_invitations_ks = apply_encryption(store.keyspace(crate::keyspaces::ROOM_INVITATIONS)?);
    let app_state_ks = apply_encryption(store.keyspace(crate::keyspaces::APP_STATE)?);
    let persona_ks = apply_encryption(store.keyspace(crate::keyspaces::PERSONA)?);
    // Domain-separated from the at-rest key so that compromise of one does not
    // hand over the other.
    let persona_correlation_key: [u8; 32] = {
        use sha2::{Digest, Sha256};
        let mut h = Sha256::new();
        h.update(b"vta-persona/correlation-index/v1");
        h.update(storage_encryption_key.unwrap_or([0u8; 32]));
        h.finalize().into()
    };
    let policy_ks = apply_encryption(store.keyspace(crate::keyspaces::POLICY)?);
    let task_consent_ks = apply_encryption(store.keyspace(crate::keyspaces::TASK_CONSENT)?);
    #[cfg(feature = "webvh")]
    let drains_ks = apply_encryption(store.keyspace(crate::keyspaces::DRAINS)?);
    #[cfg(feature = "webvh")]
    let snapshot_ks =
        apply_encryption(store.keyspace(crate::operations::protocol::snapshot::KEYSPACE_NAME)?);

    let auth = init_auth(
        &config,
        &*seed_store,
        &keys_ks,
        #[cfg(feature = "webvh")]
        Some(&webvh_ks),
        #[cfg(not(feature = "webvh"))]
        None,
    )
    .await;

    // Telemetry sink: reuse the run-path's live sink when injected, else a
    // fresh ring buffer for non-axum front-ends.
    let telemetry: vti_common::telemetry::SharedTelemetrySink = parts
        .telemetry
        .unwrap_or_else(|| Arc::new(vti_common::telemetry::RingBufferTelemetry::new()));
    #[cfg(all(feature = "webvh", feature = "didcomm"))]
    let mediator_registry = parts.mediator_registry.unwrap_or_else(|| {
        Arc::new(crate::messaging::registry::MediatorListenerRegistry::new(
            Arc::clone(&telemetry),
        ))
    });
    // The full `run()` path injects a sweeper whose teardown receiver is
    // consumed by a real task that calls `DIDCommService::remove_listener`.
    // Non-axum front-ends (e.g. Lambda) that don't run a teardown consumer get
    // a sweeper whose channel sender goes nowhere, so signals become no-ops.
    #[cfg(all(feature = "webvh", feature = "didcomm"))]
    let drain_sweeper = parts.drain_sweeper.unwrap_or_else(|| {
        let (tx, _rx) = crate::messaging::drain_sweeper::teardown_channel(
            crate::messaging::drain_sweeper::DEFAULT_TEARDOWN_CHANNEL_CAPACITY,
        );
        Arc::new(crate::messaging::drain_sweeper::DrainSweeper::new(
            Arc::clone(&mediator_registry),
            drains_ks.clone(),
            tx,
        ))
    });

    // Default sink = the audit keyspace, i.e. unchanged behaviour. A caller
    // that supplied its own keeps the keyspace handle too (`audit_ks` above),
    // because reads and retention deliberately do not route through the sink.
    let audit_sink: vta_audit::SharedAuditSink = parts
        .audit_sink
        .unwrap_or_else(|| Arc::new(vta_audit::KeyspaceAuditSink::new(audit_ks.clone())));

    Ok(AppState {
        keys_ks,
        sessions_ks,
        acl_ks,
        contexts_ks,
        did_templates_ks,
        audit_ks,
        audit_sink,
        imported_ks,
        internal_ks,
        cache_ks,
        vault_ks,
        service_state_ks,
        sealed_nonces_ks,
        idempotency_ks,
        backup_bundles_ks,
        backup_blob_dir,
        #[cfg(feature = "webvh")]
        webvh_ks,
        #[cfg(feature = "webvh")]
        passkey_vms_ks,
        consent_ks,
        consent_approvers_ks,
        issued_credentials_ks,
        memory_ks,
        room_groups_ks,
        room_invitations_ks,
        app_state_ks,
        persona_ks,
        persona_correlation_key,
        app_state_locks: crate::operations::app_state::NamespaceLocks::default(),
        policy_ks,
        task_consent_ks,
        #[cfg(feature = "webvh")]
        drains_ks,
        #[cfg(feature = "webvh")]
        snapshot_ks,
        #[cfg(all(feature = "webvh", feature = "didcomm"))]
        mediator_registry,
        #[cfg(all(feature = "webvh", feature = "didcomm"))]
        drain_sweeper,
        #[cfg(feature = "webvh")]
        webvh_auth_locks: crate::operations::did_webvh::WebvhAuthLocks::new(),
        telemetry,
        wrapping_cache: crate::keys::wrapping::WrappingKeyCache::new(),
        mdoc_trust,
        config: Arc::new(RwLock::new(config)),
        seed_store,
        did_resolver: auth.did_resolver.clone(),
        status_list_resolver: crate::vault::status::default_status_resolver(auth.did_resolver),
        secrets_resolver: auth.secrets_resolver,
        signing_vm_id: auth.signing_vm_id,
        #[cfg(any(feature = "didcomm", feature = "tsp"))]
        ka_vm_id: auth.ka_vm_id,
        #[cfg(any(feature = "didcomm", feature = "tsp"))]
        didcomm_bridge: parts
            .didcomm_bridge
            .unwrap_or_else(|| Arc::new(DIDCommBridge::placeholder())),
        #[cfg(feature = "tsp")]
        tsp_reach: Arc::new(crate::messaging::tsp_reach::TspReachability::new()),
        jwt_keys: auth.jwt_keys,
        atm: auth.atm,
        #[cfg(feature = "tsp")]
        tsp_profile: auth.tsp_profile,
        tee: tee_context,
        restart_tx,
        #[cfg(feature = "rest")]
        metrics_handle: parts.metrics_handle,
    })
}

// `config` is only mutated when the `webvh` feature is on (mirror of
// runtime-state into the in-memory `config.services`). Allow the lint
// in other feature combos so `cargo check -D warnings` stays clean.
#[cfg_attr(not(feature = "webvh"), allow(unused_mut))]
pub async fn run(
    mut config: AppConfig,
    store: Store,
    seed_store: Arc<dyn SeedStore>,
    storage_encryption_key: Option<[u8; 32]>,
    tee_context: Option<TeeContext>,
    allow_degraded: bool,
    // Recovery: when set (via `--flush-queues`), purge BOTH this DID's mediator
    // inbox and its outbound (sender) queue before going live. The outbound
    // purge is the piece the config-driven `drain_inbox_on_start` can't do — a
    // message loop can fill the sender queue at the mediator (`limits.queue.
    // sender`) so no new sends get through until it's cleared.
    // Consumed by the DIDComm outbox/inbox drain below; a REST-only build has
    // no queues to flush but keeps the parameter so callers are unconditional.
    #[cfg_attr(not(feature = "didcomm"), allow(unused_variables))] flush_queues: bool,
) -> Result<(), AppError> {
    // Fail fast on a broken config rather than booting a half-started
    // service that passes a port-liveness check but can't function (P0.9).
    config.validate()?;

    // Refuse to boot on a store left half-imported by an interrupted backup
    // restore (P0.5). The sentinel is written + fsynced before apply_import
    // clears the store and removed only after the import completes; if it is
    // still present, the rewrite didn't finish and the state is hybrid.
    {
        let keys_ks_boot = {
            let ks = store.keyspace(crate::keyspaces::KEYS)?;
            match storage_encryption_key {
                Some(key) => ks.with_encryption(key),
                None => ks,
            }
        };
        if keys_ks_boot
            .get_raw(crate::operations::backup::IMPORT_IN_PROGRESS_KEY)
            .await?
            .is_some()
        {
            return Err(AppError::Internal(
                "a previous backup import did not complete — the store is in a \
                 half-imported, inconsistent state. Re-run the import to restore a \
                 consistent snapshot before starting the VTA."
                    .into(),
            ));
        }
    }

    // Open the runtime-state keyspace once up front so the boot decisions
    // below can read it (and the migration can seed it from the legacy
    // `[services]` block on first boot post-upgrade). Same encryption policy
    // as the rest of the keyspaces.
    let boot_service_state_ks = {
        let ks = store.keyspace(crate::keyspaces::SERVICE_STATE)?;
        match storage_encryption_key {
            Some(key) => ks.with_encryption(key),
            None => ks,
        }
    };
    // Runtime-state-to-fjall migration + read-back lives in
    // `operations::protocol`, which is `#[cfg(feature = "webvh")]`-
    // gated (the protocol-management surface that owns service
    // toggles only exists in webvh builds). Without webvh the
    // boot path falls back to reading `config.services.*` directly
    // from `config.toml` — the legacy behaviour, still useful for
    // headless / secrets-only builds in the CI feature-combos
    // matrix.
    #[cfg(feature = "webvh")]
    {
        crate::operations::protocol::runtime_state::migrate_from_config(
            &boot_service_state_ks,
            &config,
        )
        .await?;

        // Runtime state in fjall is authoritative; mirror it into the in-memory
        // `config.services` so the existing readers across the codebase keep
        // working unchanged. The on-disk `config.toml` [services] block is now
        // legacy (consumed only by the first-boot migration above).
        config.services.rest =
            crate::operations::protocol::runtime_state::is_rest_enabled(&boot_service_state_ks)
                .await?;
        config.services.didcomm =
            crate::operations::protocol::runtime_state::is_didcomm_enabled(&boot_service_state_ks)
                .await?;
    }
    #[cfg(not(feature = "webvh"))]
    {
        let _ = &boot_service_state_ks;
    }

    // Reconcile the retired-seed archive (P0.7b): migrate any legacy plaintext
    // archive to ciphertext under the active seed, and repair any archive left
    // under a predecessor's KEK by an interrupted rotation. Idempotent and a
    // no-op for a never-rotated VTA (the overwhelmingly common case). Runs once
    // per process, before the restart loop. Failure is non-fatal — log and
    // continue (a stale/legacy archive is still readable via the load path).
    {
        let keys_ks_boot = {
            let ks = store.keyspace(crate::keyspaces::KEYS)?;
            match storage_encryption_key {
                Some(key) => ks.with_encryption(key),
                None => ks,
            }
        };
        match crate::keys::seeds::reconcile_archive(&keys_ks_boot, &*seed_store).await {
            Ok(0) => {}
            Ok(n) => info!(rewritten = n, "seed archive reconciled at boot"),
            Err(e) => warn!(error = %e, "seed archive reconcile failed — continuing"),
        }
    }

    // TEE anti-rollback anchor (P0.2). Verify the MAC'd integrity manifest
    // (Layer 0, P0.2a) plus the external monotonic counter (P0.2b) against the
    // live store, install the runtime sealer, and fail closed on a covered
    // singleton deleted / replayed / rolled back. Gated on a KMS storage key
    // (i.e. a real TEE); a `None` key is the non-TEE path with no
    // untrusted-parent threat. Security-critical, so a failure aborts the boot.
    #[cfg(feature = "tee")]
    if let Some(storage_key) = storage_encryption_key
        && let Some(kms) = config.tee.kms.as_ref()
    {
        let enc = |name: &str| -> Result<KeyspaceHandle, AppError> {
            Ok(store.keyspace(name)?.with_encryption(storage_key))
        };
        // Build the external counter when configured. It is keyed by the VTA
        // DID; without an identity there is nothing to key on, so fall back to
        // manifest-only (P0.2a) with a warning.
        let anchor: Option<Arc<dyn vti_common::integrity::AnchorCounter>> =
            match (kms.anchor.as_ref(), config.vta_did.as_ref()) {
                (Some(anchor_cfg), Some(vta_did)) => {
                    // P0.2c: if a sealed writer credential is configured, unseal
                    // it through the attestation-gated KMS Decrypt so the counter
                    // is written with the `vta-anchor-writer` principal (which the
                    // instance role is IAM-denied) rather than the instance role
                    // a root-on-parent attacker shares. A configured-but-
                    // unsealable credential is fatal — falling back to the
                    // instance role would silently downgrade to P0.2b.
                    let writer = match anchor_cfg.writer_credential_ciphertext.as_ref() {
                        Some(b64) => {
                            let ct = base64::engine::general_purpose::STANDARD
                                .decode(b64)
                                .map_err(|e| {
                                    AppError::Config(format!(
                                        "tee.kms.anchor.writer_credential_ciphertext is not \
                                         valid base64: {e}"
                                    ))
                                })?;
                            let pt = crate::tee::kms_bootstrap::attested_decrypt(kms, &ct).await?;
                            let creds: crate::tee::anchor::WriterCredentials =
                                serde_json::from_slice(&pt).map_err(|e| {
                                    AppError::Config(format!(
                                        "anchor writer credential did not decrypt to \
                                         {{access_key_id, secret_access_key}}: {e}"
                                    ))
                                })?;
                            info!("anchor writer credential unsealed (attestation-gated, P0.2c)");
                            Some(creds)
                        }
                        None => None,
                    };
                    Some(Arc::new(
                        crate::tee::anchor::DynamoAnchorCounter::new(
                            &kms.region,
                            anchor_cfg.table_name.clone(),
                            vta_did.clone(),
                            writer,
                        )
                        .await,
                    ))
                }
                (Some(_), None) => {
                    warn!(
                        "tee.kms.anchor is configured but vta_did is unset — booting \
                         manifest-only (P0.2a); the external rollback counter is disabled"
                    );
                    None
                }
                (None, _) => None,
            };
        let outcome = vti_common::integrity::boot_verify_and_install(
            vti_common::integrity::derive_mac_key(&storage_key),
            enc("keys")?,
            store.keyspace(crate::keyspaces::BOOTSTRAP)?, // unencrypted, KMS-protected
            enc("acl")?,
            enc("contexts")?,
            anchor,
            kms.allow_anchor_init,
            kms.allow_unanchored,
        )
        .await?;
        info!(?outcome, "TEE anti-rollback anchor checked");
    }

    // Determine which services will actually start (feature flag AND
    // persisted runtime state, the latter set by `pnm services {kind}
    // {enable,disable}`).
    let rest_enabled = cfg!(feature = "rest") && config.services.rest;
    let didcomm_enabled = cfg!(feature = "didcomm") && config.services.didcomm;

    if !rest_enabled && !didcomm_enabled {
        return Err(AppError::Config(
            "no services enabled — enable at least one of REST or DIDComm \
             (compile-time feature flags + `pnm services {kind} enable`)"
                .into(),
        ));
    }

    // Bind TCP listener once (persists across soft restarts)
    #[cfg(feature = "rest")]
    let std_listener = if rest_enabled {
        let addr = format!("{}:{}", config.server.host, config.server.port);
        let listener = std::net::TcpListener::bind(&addr).map_err(AppError::Io)?;
        listener.set_nonblocking(true).map_err(AppError::Io)?;
        info!("server listening addr={addr}");
        Some(listener)
    } else {
        None
    };

    // Install the Prometheus recorder once per process (persists across
    // soft restarts, same as the TCP listener above). The global recorder
    // can only be set once — installing it inside the restart loop panics
    // the REST thread on the second iteration with FailedToSetGlobalRecorder.
    // The handle is cloned into each iteration's AppState below.
    #[cfg(feature = "rest")]
    let metrics_handle = if rest_enabled {
        Some(crate::metrics::install())
    } else {
        None
    };

    // ── Restart loop ──────────────────────────────────────────────
    // Each iteration starts all service threads, waits for shutdown
    // or restart signal, tears everything down, then either exits
    // or loops back to re-initialize with updated state.
    loop {
        // Keyspace handles `run()` needs directly: the storage-thread inputs
        // (un-gated, so the storage thread compiles in every feature combo) and
        // the drain set (webvh — needed for boot replay + the sweeper, which
        // are built before `AppState` exists). Every other keyspace is opened
        // by `build_app_state`, the single `AppState` constructor (P1.1).
        let apply_encryption = |ks: KeyspaceHandle| -> KeyspaceHandle {
            match storage_encryption_key {
                Some(key) => ks.with_encryption(key),
                None => ks,
            }
        };
        let sessions_ks = apply_encryption(store.keyspace(crate::keyspaces::SESSIONS)?);
        let acl_ks = apply_encryption(store.keyspace(crate::keyspaces::ACL)?);
        let audit_ks = apply_encryption(store.keyspace(crate::keyspaces::AUDIT)?);
        let consent_ks = apply_encryption(store.keyspace(crate::keyspaces::CONSENT)?);
        let idempotency_ks = apply_encryption(store.keyspace(crate::keyspaces::IDEMPOTENCY)?);
        let task_consent_ks = apply_encryption(store.keyspace(crate::keyspaces::TASK_CONSENT)?);
        let vault_ks = apply_encryption(store.keyspace(crate::keyspaces::VAULT)?);
        let backup_bundles_ks = apply_encryption(store.keyspace(crate::keyspaces::BACKUP_BUNDLES)?);
        let backup_blob_dir = config.store.data_dir.join("backups");
        // Read by the DIDComm drain machinery only; a TSP-only build has no
        // drain window, so it never opens the keyspace.
        #[cfg(all(feature = "webvh", feature = "didcomm"))]
        let drains_ks = apply_encryption(store.keyspace(crate::keyspaces::DRAINS)?);

        // Pluggable telemetry sink + multi-mediator listener registry.
        // The registry holds active/drain state and the per-mediator
        // bounded outbound buffer; spec
        // `docs/05-design-notes/didcomm-protocol-management.md`.
        let telemetry: vti_common::telemetry::SharedTelemetrySink =
            Arc::new(vti_common::telemetry::RingBufferTelemetry::new());
        #[cfg(all(feature = "webvh", feature = "didcomm"))]
        let mediator_registry = Arc::new(
            crate::messaging::registry::MediatorListenerRegistry::new(Arc::clone(&telemetry)),
        );
        // Drain sweeper: TTL-keyed `tokio::time::sleep_until` per
        // drain entry. On expiry, the sweeper signals the
        // teardown channel; the consumer task spawned below
        // translates each signal into a
        // `DIDCommService::remove_listener` call.
        #[cfg(all(feature = "webvh", feature = "didcomm"))]
        let (teardown_tx, teardown_rx) = crate::messaging::drain_sweeper::teardown_channel(
            crate::messaging::drain_sweeper::DEFAULT_TEARDOWN_CHANNEL_CAPACITY,
        );
        #[cfg(all(feature = "webvh", feature = "didcomm"))]
        let drain_sweeper = Arc::new(crate::messaging::drain_sweeper::DrainSweeper::new(
            Arc::clone(&mediator_registry),
            drains_ks.clone(),
            teardown_tx,
        ));
        // Boot replay: load any drains persisted from a previous
        // run, drop already-expired entries, register the live
        // ones with the registry, and arm the sweeper for each.
        #[cfg(all(feature = "webvh", feature = "didcomm"))]
        match mediator_registry.replay_drains(&drains_ks).await {
            Ok(live) => {
                if !live.is_empty() {
                    info!(count = live.len(), "drain set replayed from keyspace");
                }
                drain_sweeper.arm_all(&live).await;
            }
            Err(e) => {
                warn!(error = %e, "drain replay failed — starting with empty drain set");
            }
        }

        // Shutdown + restart coordination
        let (shutdown_tx, shutdown_rx) = watch::channel(false);
        let (restart_tx, mut restart_rx) = watch::channel(false);

        #[cfg(any(feature = "didcomm", feature = "tsp"))]
        let didcomm_shutdown = CancellationToken::new();

        // Spawn signal handler. First signal triggers cooperative shutdown;
        // a second signal forces an immediate exit so an operator can always
        // bail out if cleanup hangs (e.g., a mediator handshake that won't
        // complete before its timeout fires).
        tokio::spawn({
            let shutdown_tx = shutdown_tx.clone();
            #[cfg(any(feature = "didcomm", feature = "tsp"))]
            let didcomm_shutdown = didcomm_shutdown.clone();
            async move {
                shutdown_signal().await;
                info!("shutting down — press Ctrl-C again to force exit");
                let _ = shutdown_tx.send(true);
                #[cfg(any(feature = "didcomm", feature = "tsp"))]
                didcomm_shutdown.cancel();

                shutdown_signal().await;
                eprintln!("\nForcing exit.");
                std::process::exit(130);
            }
        });

        // Gather storage thread inputs
        let storage_store = store.clone();
        let storage_sessions_ks = sessions_ks.clone();
        let storage_audit_ks = audit_ks.clone();
        // Built once here and handed to *both* the storage thread and
        // `AppStateParts` below, so the sweepers and the request path share one
        // sink object rather than two that merely happen to agree today.
        let audit_sink: vta_audit::SharedAuditSink =
            Arc::new(vta_audit::KeyspaceAuditSink::new(audit_ks.clone()));
        let storage_audit_sink = Arc::clone(&audit_sink);
        let storage_acl_ks = acl_ks.clone();
        let storage_consent_ks = consent_ks.clone();
        let storage_idempotency_ks = idempotency_ks.clone();
        let storage_task_consent_ks = task_consent_ks.clone();
        let storage_vault_ks = vault_ks.clone();
        let storage_app_state_ks = apply_encryption(store.keyspace(crate::keyspaces::APP_STATE)?);
        let storage_app_state_retention_days = config.app_state.tombstone_retention_days;
        let storage_backup_bundles_ks = backup_bundles_ks.clone();
        let storage_backup_blob_dir = backup_blob_dir.clone();
        let storage_audit_config = config.audit.clone();
        let storage_auth_config = config.auth.clone();

        // Shared DIDComm bridge for outbound request-response messaging.
        // The service reference is set after DIDCommService::start().
        #[cfg(any(feature = "didcomm", feature = "tsp"))]
        let didcomm_bridge: Arc<DIDCommBridge> = Arc::new(DIDCommBridge::new("vta-main"));

        // Build the shared `AppState` once, via the single constructor
        // (`build_app_state`), injecting the live shared components `run()`
        // built above (telemetry sink, mediator registry, drain sweeper, the
        // real DIDComm bridge, the metrics handle). Both the REST front-end and
        // the DIDComm → trust-task dispatch bridge clone from this one owned
        // copy. `AppState` is `Clone`. (P1.1)
        #[cfg(any(feature = "rest", feature = "didcomm"))]
        let app_state = {
            let parts = AppStateParts {
                telemetry: Some(Arc::clone(&telemetry)),
                // The same sink the storage thread's sweepers hold, so an
                // operator-installed backend covers unattended removals too.
                audit_sink: Some(Arc::clone(&audit_sink)),
                #[cfg(all(feature = "webvh", feature = "didcomm"))]
                mediator_registry: Some(Arc::clone(&mediator_registry)),
                #[cfg(all(feature = "webvh", feature = "didcomm"))]
                drain_sweeper: Some(Arc::clone(&drain_sweeper)),
                #[cfg(any(feature = "didcomm", feature = "tsp"))]
                didcomm_bridge: Some(didcomm_bridge.clone()),
                #[cfg(feature = "rest")]
                metrics_handle: metrics_handle.clone(), // installed once, before the loop
            };
            build_app_state(
                config.clone(),
                &store,
                seed_store.clone(),
                storage_encryption_key,
                tee_context.clone(),
                restart_tx.clone(),
                parts,
            )
            .await?
        };

        // The tombstone sweeper takes a namespace's lock to reap it, so it must
        // hold the *same* map the request path writes through. Cloned from the
        // built `AppState` rather than injected through `AppStateParts`: adding
        // a public field to that struct is a semver break (it is not
        // `#[non_exhaustive]`, unlike `AppState`), and the storage thread is
        // spawned after `build_app_state` anyway, so there is nothing to inject.
        let storage_app_state_locks = app_state.app_state_locks.clone();
        // The wrapping-key cache reaper is a run()-path concern (build_app_state
        // just constructs the cache); arm it on the live state.
        #[cfg(any(feature = "rest", feature = "didcomm"))]
        app_state.wrapping_cache.clone().spawn_reaper();

        // Policy bootstrap reads `app_state.policy_ks`, and `app_state` itself
        // only exists when a serving surface does — `build_app_state` is what
        // constructs the policy keyspace. Carry the same gate as the binding
        // above, or a `--no-default-features` build names a value that was
        // configured out.
        #[cfg(any(feature = "rest", feature = "didcomm"))]
        {
            // Boot-install the migration-safe default PDP baseline if the operator
            // has no policy yet. Idempotent; never clobbers an operator upload.
            crate::policy::install_default_policy(
                &app_state.policy_ks,
                &chrono::Utc::now().to_rfc3339(),
            )
            .await?;

            // Drop any `config:require-consent` row synthesized by a previous
            // release. `[[policy.require_consent]]` is retired and the config
            // loader now refuses it — but a VTA upgraded from a release that had
            // the block would otherwise keep enforcing the row it last wrote,
            // with nothing left that can explain or remove it. One-way and
            // idempotent; it never writes.
            crate::policy::remove_stale_config_consent_policy(&app_state.policy_ks).await?;

            // Seed the declarative approvals row from config the first time we
            // boot without one. Unlike the reconcile above, this runs ONCE: the
            // rules are editable at runtime (`pnm approvals`), so re-reading the
            // file every boot would silently revert an operator's change on the
            // next restart.
            {
                let cfg = app_state.config.read().await;
                crate::policy::seed_declarative_approvals(
                    &app_state.policy_ks,
                    &cfg.policy.approvals,
                    &cfg.policy.approver_sets,
                    &chrono::Utc::now().to_rfc3339(),
                )
                .await?;
            }
        }

        // Fail-closed on missing identity (P0.9b). `init_auth` (inside
        // build_app_state) yields `jwt_keys: Some` only when the VTA has a
        // complete, usable signing identity: a configured `vta_did`, its key
        // records + seed present, and a decodable JWT signing key. With any of
        // those missing the VTA still *boots* but every authenticated endpoint
        // returns 401 — a service that looks "up" to a liveness probe while
        // being inert. Refuse to start unless the operator explicitly opted
        // into a degraded boot (e.g. to inspect or finish provisioning a
        // half-set-up instance). TEE front-ends pass `allow_degraded = true`:
        // their identity is established by KMS autogen / admin-bootstrap earlier
        // in enclave boot, and a degraded first boot there is an existing,
        // documented state.
        #[cfg(any(feature = "rest", feature = "didcomm"))]
        if app_state.jwt_keys.is_none() && !allow_degraded {
            return Err(AppError::Config(missing_identity_message(&config)));
        }

        // Whether a usable signing identity is present — drives session cleanup
        // in the storage thread (and the TEE warn below). Defined in every
        // feature combo so the un-gated storage thread compiles; a build with
        // neither transport returns at the "no services enabled" guard above
        // before this is read.
        #[cfg(any(feature = "rest", feature = "didcomm"))]
        let has_auth = app_state.jwt_keys.is_some();
        #[cfg(not(any(feature = "rest", feature = "didcomm")))]
        let has_auth = false;

        // In TEE required mode, warn if auth isn't initialized.
        #[cfg(feature = "tee")]
        if config.tee.mode == crate::config::TeeMode::Required && !has_auth {
            warn!(
                "TEE mode is 'required' but authentication is not initialized \
                 (vta_did not configured). The VTA will start but authenticated \
                 endpoints will return 401."
            );
        }

        // Spawn REST thread (conditional)
        #[cfg(feature = "rest")]
        let rest_handle = if let Some(ref listener_ref) = std_listener {
            let listener = listener_ref.try_clone().map_err(AppError::Io)?;
            let state = app_state.clone();
            let mut rest_shutdown_rx = shutdown_rx.clone();
            Some(
                std::thread::Builder::new()
                    .name("vta-rest".into())
                    .spawn(move || run_rest_thread(listener, state, &mut rest_shutdown_rx))
                    .map_err(|e| AppError::Internal(format!("failed to spawn REST thread: {e}")))?,
            )
        } else {
            None
        };
        #[cfg(not(feature = "rest"))]
        let rest_handle: Option<std::thread::JoinHandle<()>> = None;

        // Start the delivery-layer messaging service (conditional).
        //
        // D2 P2a: the `MessagingService` over a `DidCommTransport` (built in
        // `messaging::service::build_messaging`) replaces the framework
        // `DIDCommService`. On success it is published into the outbound
        // `DIDCommBridge` (so REST + DIDComm handlers can send) and drives the
        // protocol-routed inbound loop; the `Arc<MessagingService>` handle is
        // retained for drain teardown. `DidCommTransport::inbound()` multiplexes
        // BOTH DIDComm and TSP frames off the one mediator socket (one socket
        // per DID — a second would be evicted as `duplicate-channel`), so TSP
        // receive is always on when compiled with `tsp`; `config.services.tsp`
        // governs advertisement only.
        //
        // Self-readiness gate + persistent reconnect, both owned by the spawned
        // `MessagingConnect` supervisor.
        //
        // The supervisor first runs the self-readiness gate: it waits until the
        // VTA's own DID resolves over the network — the same operation the
        // mediator performs to fetch our sender key — so the mediator can
        // authenticate us instead of 403-storming on a cold start. It then
        // connects, and keeps reconnecting: the classic initial failure is the
        // mediator's *own* resolver still negative-caching the VTA host, which
        // clears itself on its own timer, so the VTA self-heals with no operator
        // restart. It also re-connects if an established session's inbound loop
        // ends, rather than going silently deaf for the rest of the process.
        //
        // Everything — gate included — runs in the spawned task, never on the
        // startup path. `run()` must reach the shutdown/restart select below for
        // a SIGTERM to be honoured, so awaiting a gate here (up to
        // `max_wait_secs`, default 300s) would hold the process open with no
        // listener after the REST thread had already exited.
        //
        // The live `MessagingService` handle is published into
        // `app_state.didcomm_bridge` on each successful connect; the
        // drain-teardown consumer + status read it there.
        //
        // `on_timeout = "fail"` has to fail the *process*, and the gate no longer
        // runs on this path, so it can't do that by returning `Err` from here.
        // The supervisor sets this flag and signals shutdown; `run` converts it
        // back into an `Err` after the threads are joined, so the exit status
        // still says "failed" for a systemd `Restart=on-failure` or anything else
        // that reads it.
        #[cfg(any(feature = "didcomm", feature = "tsp"))]
        let readiness_fatal = Arc::new(std::sync::atomic::AtomicBool::new(false));
        // Either advertised transport needs the mediator socket: TSP receive
        // arrives on it (ADR 0005 — one websocket per DID) and TSP send is an
        // HTTP post through the same ATM. Gating this on DIDComm alone is what
        // made a TSP-only VTA impossible — it would have advertised `#tsp` and
        // never connected to anything.
        #[cfg(any(feature = "didcomm", feature = "tsp"))]
        if config.services.didcomm || config.services.tsp {
            match (
                &app_state.secrets_resolver,
                &config.vta_did,
                &config.messaging,
            ) {
                (Some(_), Some(vta_did), Some(messaging_config)) => {
                    // Compute the outbox keyspace here (it needs `?`), then hand
                    // the whole gate + connect + reconnect loop to the background
                    // supervisor.
                    let outbox_ks = apply_encryption(store.keyspace(crate::keyspaces::OUTBOX)?);
                    let supervisor = MessagingConnect {
                        app_state: app_state.clone(),
                        vta_did: vta_did.clone(),
                        messaging_config: messaging_config.clone(),
                        readiness: config.mediator_readiness.clone(),
                        resolver_url: config.resolver_url.clone(),
                        outbox_ks,
                        flush_queues,
                        shutdown: didcomm_shutdown.clone(),
                        fatal_shutdown: shutdown_tx.clone(),
                        fatal_flag: readiness_fatal.clone(),
                    };
                    tokio::spawn(supervisor.run());
                }
                _ => {
                    info!("DIDComm not configured — service not started");
                }
            }
        }

        // TSP inbound is no longer a standalone websocket. The delivery-layer
        // `DidCommTransport` built above multiplexes TSP frames off its single
        // mediator websocket (`Inbound.message.protocol` tags DIDComm vs TSP);
        // the inbound loop routes each TSP frame to `messaging::tsp_inbound`.
        // Opening a second socket here — as the earlier `run_tsp_inbound` loop
        // did — made the mediator evict a connection as `duplicate-channel`,
        // flapping the VTA. See `messaging::tsp_inbound`.

        // Spawn the teardown channel consumer. The drain sweeper
        // sends mediator DIDs over `teardown_rx` whenever a TTL
        // fires; this task translates each signal into a
        // `DIDCommService::remove_listener` call. If DIDComm
        // isn't running, the loop still runs but every recv is a
        // no-op — drains still get cleaned up at the registry +
        // keyspace level by the sweeper.
        #[cfg(all(feature = "webvh", feature = "didcomm"))]
        let _teardown_handle = {
            // Source the live `MessagingService` from the bridge at recv time
            // rather than a captured handle: the connect now happens in the
            // background `MessagingConnect` supervisor, so the service may not be
            // published yet when this consumer is spawned (and it self-heals /
            // (re)connects later). `messaging_handle()` returns the current
            // handle once the supervisor's first successful connect sets it.
            let teardown_app_state = app_state.clone();
            let mut teardown_rx = teardown_rx;
            let mut shutdown_rx_for_teardown = shutdown_rx.clone();
            tokio::spawn(async move {
                loop {
                    tokio::select! {
                        biased;
                        _ = shutdown_rx_for_teardown.changed() => {
                            if *shutdown_rx_for_teardown.borrow() {
                                break;
                            }
                        }
                        msg = teardown_rx.recv() => {
                            match msg {
                                None => break,
                                Some(mediator_did) => {
                                    // The drain window kept the OLD mediator's
                                    // transport installed (still receiving inbound
                                    // via the merged dispatcher) after promote; its
                                    // fjall drain-entry TTL has now expired, so drop
                                    // it from the delivery-layer service.
                                    if let Some(svc) =
                                        teardown_app_state.didcomm_bridge.messaging_handle()
                                    {
                                        svc.remove_transport(&mediator_did);
                                        info!(
                                            mediator = %mediator_did,
                                            "drain teardown: transport removed"
                                        );
                                    } else {
                                        debug!(
                                            mediator = %mediator_did,
                                            "drain teardown: DIDComm not connected, skipping remove_transport"
                                        );
                                    }
                                }
                            }
                        }
                    }
                }
                debug!("teardown consumer task exiting");
            })
        };
        #[cfg(not(all(feature = "webvh", feature = "didcomm")))]
        let _teardown_handle: Option<tokio::task::JoinHandle<()>> = None;

        // Storage thread always runs
        let mut storage_shutdown_rx = shutdown_rx.clone();
        let storage_handle = std::thread::Builder::new()
            .name("vta-storage".into())
            .spawn(move || {
                run_storage_thread(
                    storage_store,
                    storage_sessions_ks,
                    storage_audit_ks,
                    storage_audit_sink,
                    storage_acl_ks,
                    storage_consent_ks,
                    storage_idempotency_ks,
                    storage_task_consent_ks,
                    storage_vault_ks,
                    storage_app_state_ks,
                    storage_app_state_locks,
                    storage_app_state_retention_days,
                    storage_backup_bundles_ks,
                    storage_backup_blob_dir,
                    storage_audit_config,
                    storage_auth_config,
                    has_auth,
                    &mut storage_shutdown_rx,
                )
            })
            .map_err(|e| AppError::Internal(format!("failed to spawn storage thread: {e}")))?;

        // ── Wait for shutdown or restart ──────────────────────────
        let mut any_panic = false;
        let is_restart;

        if let Some(handle) = rest_handle {
            // REST thread blocks — wait for it, or for restart signal
            tokio::select! {
                result = tokio::task::spawn_blocking(move || handle.join()) => {
                    match result {
                        Ok(Ok(())) => info!("REST thread stopped"),
                        Ok(Err(_panic)) => { error!("REST thread panicked"); any_panic = true; }
                        Err(e) => { error!("failed to join REST thread: {e}"); any_panic = true; }
                    }
                    is_restart = false;
                }
                _ = restart_rx.changed() => {
                    info!("soft restart requested — shutting down services");
                    let _ = shutdown_tx.send(true);
                    is_restart = true;
                }
            }
        } else {
            // No REST thread — wait for shutdown or restart
            tokio::select! {
                _ = async {
                    let mut wait_rx = shutdown_rx.clone();
                    let _ = wait_rx.changed().await;
                } => {
                    is_restart = false;
                }
                _ = restart_rx.changed() => {
                    info!("soft restart requested — shutting down services");
                    let _ = shutdown_tx.send(true);
                    is_restart = true;
                }
            }
        }

        // Stop mediator messaging: cancel the inbound loop's shutdown token
        // (also stops the `MessagingConnect` reconnect supervisor between
        // attempts). The delivery-layer background tasks (dispatcher, transport
        // forwarder, outbox drain loops) are detached — as in the VTC pilot —
        // and wind down as their `Arc<MessagingService>`/transport handles drop.
        #[cfg(any(feature = "didcomm", feature = "tsp"))]
        {
            didcomm_shutdown.cancel();
            info!("mediator messaging stopped");
        }

        if any_panic {
            let _ = shutdown_tx.send(true);
        }

        // Join storage last — guarantees all writes flushed before database closes
        match storage_handle.join() {
            Ok(()) => info!("storage thread stopped"),
            Err(_panic) => {
                error!("storage thread panicked");
                any_panic = true;
            }
        }

        if any_panic {
            return Err(AppError::Internal("one or more threads panicked".into()));
        }

        // Surface a `fail`-policy readiness timeout as a non-zero exit, now that
        // the threads are joined and the store is flushed.
        #[cfg(any(feature = "didcomm", feature = "tsp"))]
        if readiness_fatal.load(std::sync::atomic::Ordering::Relaxed) {
            return Err(AppError::Internal(
                "mediator self-readiness gate timed out with on_timeout = \"fail\"".into(),
            ));
        }

        if !is_restart {
            info!("server shut down");
            return Ok(());
        }

        // ── Soft restart: reload config and re-derive keys ───────
        info!("soft restart: re-initializing services");

        // Config and seed are updated in-memory by the import handler.
        // The restart loop re-initializes auth and keyspace handles from
        // the current config and seed_store on the next iteration.
    }
}

/// Storage thread: runs session cleanup loop and persists the store on shutdown.
#[allow(clippy::too_many_arguments)]
fn run_storage_thread(
    store: Store,
    sessions_ks: KeyspaceHandle,
    // `audit_ks` for `cleanup_expired_logs` — retention prunes the keyspace.
    // `audit_sink` for the sweepers' own audit rows: the *same* sink the
    // request path uses, threaded in rather than rebuilt from `audit_ks`. A
    // sweeper's `acl.expire` is exactly the kind of unattended removal an
    // operator installs a tamper-evident backend to capture, and rebuilding one
    // here would route it back to the keyspace whatever they configured.
    audit_ks: KeyspaceHandle,
    audit_sink: vta_audit::SharedAuditSink,
    acl_ks: KeyspaceHandle,
    consent_ks: KeyspaceHandle,
    idempotency_ks: KeyspaceHandle,
    task_consent_ks: KeyspaceHandle,
    vault_ks: KeyspaceHandle,
    app_state_ks: KeyspaceHandle,
    app_state_locks: crate::operations::app_state::NamespaceLocks,
    app_state_retention_days: u32,
    backup_bundles_ks_storage: KeyspaceHandle,
    backup_blob_dir_storage: std::path::PathBuf,
    audit_config: crate::config::AuditConfig,
    auth_config: AuthConfig,
    has_auth: bool,
    shutdown_rx: &mut watch::Receiver<bool>,
) {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("failed to build storage runtime");

    rt.block_on(async {
        info!("storage thread started");

        if has_auth {
            let interval = Duration::from_secs(auth_config.session_cleanup_interval);
            let mut timer = tokio::time::interval(interval);
            // First tick completes immediately; skip it so cleanup doesn't run at startup
            timer.tick().await;

            loop {
                tokio::select! {
                    _ = timer.tick() => {
                        if let Err(e) = cleanup_expired_sessions(&sessions_ks, auth_config.challenge_ttl).await {
                            warn!("session cleanup error: {e}");
                        }
                        // Also clean up expired audit logs
                        let audit_retention = audit_config.retention_days;
                        if let Err(e) = crate::audit::cleanup_expired_logs(&audit_ks, audit_retention).await {
                            warn!("audit cleanup error: {e}");
                        }
                        // Prune expired AclEntry rows and PendingBootstrap rows.
                        // `audit_ks` threaded in so each deletion produces
                        // an `acl.expire` audit entry — without it, the
                        // sweeper's removals leave no trail and operators
                        // can't distinguish "entry was never created" from
                        // "entry was created then expired and pruned".
                        if let Err(e) =
                            crate::acl_sweeper::sweep_expired(&acl_ks, &audit_sink).await
                        {
                            warn!("acl sweeper error: {e}");
                        }
                        // Prune expired pending consents (never answered) and
                        // lapsed grants, so the consent keyspace can't grow
                        // unbounded at inbound-message rate.
                        if let Err(e) =
                            crate::consent_sweeper::sweep_expired(&consent_ks, &audit_sink).await
                        {
                            warn!("consent sweeper error: {e}");
                        }
                        // Idempotency records are read-through-expiry, so a
                        // missed pass costs space and nothing else — that is why
                        // this one logs rather than propagating, and why it is
                        // not audited (retry bookkeeping, not a security event).
                        crate::idempotency_sweeper::sweep_expired_logged(&idempotency_ks).await;
                        // Same for task-execution consent: an unanswered pending
                        // would otherwise sit in the keyspace forever, since the
                        // gate only expires one lazily when its digest is re-read.
                        let now = std::time::SystemTime::now()
                            .duration_since(std::time::UNIX_EPOCH)
                            .map(|d| d.as_secs())
                            .unwrap_or(0);
                        match crate::policy::consent::sweep_expired(&task_consent_ks, now).await {
                            Ok(n) if n > 0 => debug!("task-consent sweeper pruned {n} rows"),
                            Ok(_) => {}
                            Err(e) => warn!("task-consent sweeper error: {e}"),
                        }
                        // Expire & retention-prune in-flight backup
                        // bundles (descriptor-pattern slice). TTL
                        // pass transitions stale non-terminal records
                        // to Expired; retention pass deletes terminal
                        // records older than the 24h audit window.
                        if let Err(e) = crate::backup_bundle_sweeper::sweep_bundles(
                            &backup_bundles_ks_storage,
                            &backup_blob_dir_storage,
                        )
                        .await
                        {
                            warn!("backup bundle sweeper error: {e}");
                        }
                        // Reclaim deferred-presentation records that are
                        // terminal or stale (P0.12). Without this the
                        // `pending-present:` namespace grows unbounded at
                        // DIDComm message rate — every untrusted-verifier
                        // query writes one.
                        match crate::operations::credential_exchange::pending::sweep(
                            &vault_ks,
                            chrono::Utc::now(),
                        )
                        .await
                        {
                            Ok(n) if n > 0 => {
                                info!(reclaimed = n, "pending-present sweeper")
                            }
                            Ok(_) => {}
                            Err(e) => warn!("pending-present sweeper error: {e}"),
                        }
                        // Hard-purge grace-expired vault + credential tombstones
                        // (soft-deleted entries past their recovery window), so
                        // the trash can't linger forever. Audits each purge.
                        if let Err(e) =
                            crate::vault_sweeper::sweep_expired(&vault_ks, &audit_sink).await
                        {
                            warn!("vault sweeper error: {e}");
                        }
                        // Reap application-state tombstones past their retention
                        // window. This is what makes the window real rather than
                        // advertised — and what makes `watermarkTooOld`
                        // reachable, so a consumer resuming from a watermark
                        // whose deletions have been discarded is told to rebuild
                        // instead of being served a feed that silently omits
                        // them.
                        // `0` days disables reaping outright: tombstones are
                        // kept forever, no watermark ever expires, and the
                        // keyspace grows unbounded. That is a legitimate choice
                        // for a deployment that would rather spend disk than
                        // ever force a consumer to rebuild — so it is a skip
                        // here rather than a zero cutoff, which would mean the
                        // opposite and reap everything.
                        if app_state_retention_days > 0 {
                            match crate::operations::app_state::sweep_expired_tombstones(
                                &app_state_ks,
                                &app_state_locks,
                                &audit_sink,
                                u64::from(app_state_retention_days) * 24 * 60 * 60,
                            )
                            .await
                            {
                                Ok(n) if n > 0 => {
                                    info!(reaped = n, "app-state tombstone sweeper")
                                }
                                Ok(_) => {}
                                Err(e) => warn!("app-state tombstone sweeper error: {e}"),
                            }
                        }
                    }
                    _ = shutdown_rx.changed() => {
                        info!("storage thread shutting down");
                        break;
                    }
                }
            }
        } else {
            // No auth — just wait for shutdown
            let _ = shutdown_rx.changed().await;
            info!("storage thread shutting down");
        }

        // Persist store before closing
        if let Err(e) = store.persist().await {
            error!("failed to persist store on shutdown: {e}");
        } else {
            info!("store persisted");
        }
    });
}

/// REST thread: serves the Axum HTTP server.
#[cfg(feature = "rest")]
fn run_rest_thread(
    std_listener: std::net::TcpListener,
    state: AppState,
    shutdown_rx: &mut watch::Receiver<bool>,
) {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("failed to build REST runtime");

    rt.block_on(async {
        info!("REST thread started");

        // The Prometheus recorder is installed once per process (before the
        // restart loop in `run`); `state.metrics_handle` already carries the
        // handle. Installing here would panic on every soft restart.

        let listener = tokio::net::TcpListener::from_std(std_listener)
            .expect("failed to convert std TcpListener to tokio TcpListener");

        // Snapshot the CORS origins for the router build. The config
        // is reloadable, but a router rebuild requires a full
        // service restart (which the operator triggers via
        // /vta/restart after editing the file), so reading the
        // current values here is correct.
        let (cors_origins, trust_xff, interval_secs, burst) = {
            let cfg = state.config.read().await;
            (
                cfg.server.cors_origins.clone(),
                cfg.server.trust_xff,
                cfg.server.rate_limit_interval_secs,
                cfg.server.rate_limit_burst,
            )
        };
        let traced_routes =
            routes::router_with_cors(&cors_origins, trust_xff, interval_secs, burst)
                .with_state(state.clone())
                .layer(axum::middleware::from_fn(crate::metrics::track_metrics))
                .layer(
                    TraceLayer::new_for_http()
                        .make_span_with(DefaultMakeSpan::new().level(Level::INFO))
                        .on_request(DefaultOnRequest::new().level(Level::INFO))
                        .on_response(DefaultOnResponse::new().level(Level::INFO)),
                );

        // `/health` stays out of the trace + metrics layers (it's a
        // high-frequency probe), but still needs the API's CORS policy
        // so browser tools can run their cross-origin connectivity
        // check against it.
        let app =
            traced_routes.merge(routes::health_router_with_cors(&cors_origins).with_state(state));

        let shutdown_rx = shutdown_rx.clone();
        axum::serve(
            listener,
            app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
        )
        .with_graceful_shutdown(async move {
            let mut rx = shutdown_rx;
            let _ = rx.changed().await;
        })
        .await
        .expect("axum serve failed");

        info!("REST thread shutting down");
    });
}

/// Initialize DID resolver, secrets resolver, and JWT keys for authentication.
///
/// Returns `None` values if the VTA DID is not configured (server still starts
/// so the setup wizard can be run first).
/// Result of auth initialization, bundling all outputs including the
/// verification-method IDs that were inserted into the secrets resolver.
struct AuthInit {
    did_resolver: Option<DIDCacheClient>,
    secrets_resolver: Option<Arc<ThreadedSecretsResolver>>,
    jwt_keys: Option<Arc<JwtKeys>>,
    atm: Option<ATM>,
    /// VTA's registered TSP profile (see [`AppState::tsp_profile`]). Built
    /// alongside `atm` when the `tsp` feature is on; `None` on any failure.
    #[cfg(feature = "tsp")]
    tsp_profile: Option<std::sync::Arc<affinidi_tdk::messaging::profiles::ATMProfile>>,
    /// Signing verification method ID (e.g. `{did}#key-0` or `{did}#{ed_pub_mb}`).
    /// Consumed only by the DIDComm secret-collection path; cfg-gated to
    /// keep non-didcomm builds warning-free.
    #[cfg_attr(not(feature = "didcomm"), allow(dead_code))]
    signing_vm_id: Option<String>,
    /// Key-agreement verification method ID (e.g. `{did}#key-1` or `{did}#{x_pub_mb}`).
    #[cfg_attr(not(feature = "didcomm"), allow(dead_code))]
    ka_vm_id: Option<String>,
}

impl AuthInit {
    fn empty() -> Self {
        Self {
            did_resolver: None,
            secrets_resolver: None,
            jwt_keys: None,
            atm: None,
            #[cfg(feature = "tsp")]
            tsp_profile: None,
            signing_vm_id: None,
            ka_vm_id: None,
        }
    }
}

/// Build the operator-facing boot-refusal message for the missing-identity
/// gate (P0.9b). Names the specific gap so the fix is obvious, and always
/// points at the `--allow-degraded` escape hatch (mirrors CLAUDE.md's
/// "operator errors should suggest the fix").
fn missing_identity_message(config: &AppConfig) -> String {
    let cause = if config.vta_did.is_none() {
        "vta_did is not configured — this VTA has no identity. Run `vta setup` \
         to provision one"
            .to_string()
    } else if config.auth.jwt_signing_key.is_none() {
        "auth.jwt_signing_key is not configured — the VTA can't issue access \
         tokens. Run `vta setup`, or restore the key to config.toml"
            .to_string()
    } else {
        format!(
            "vta_did is set ({}) but its signing identity could not be loaded — \
             the VTA key records may be missing from the store or the seed \
             backend may be unreachable. Check the store data_dir and the \
             secrets backend",
            config.vta_did.as_deref().unwrap_or_default()
        )
    };
    format!(
        "refusing to start: {cause}. A VTA without a usable signing identity \
         boots but answers every authenticated request with 401. To start \
         anyway (e.g. to inspect or finish provisioning a half-set-up \
         instance), pass `--allow-degraded`."
    )
}

async fn init_auth(
    config: &AppConfig,
    seed_store: &dyn SeedStore,
    keys_ks: &KeyspaceHandle,
    // Local `did.jsonl` source for the self-DID resolver preload (webvh only).
    // Always present in the signature; the caller passes `None` when the
    // `webvh` feature is off. Keeps the signature identical across feature sets.
    webvh_ks: Option<&KeyspaceHandle>,
) -> AuthInit {
    let vta_did = match &config.vta_did {
        Some(did) => did.clone(),
        None => {
            warn!("vta_did not configured — auth endpoints will not work (run setup first)");
            return AuthInit::empty();
        }
    };

    // Look up VTA key paths from stored key records
    let (signing_path, ka_path, vta_seed_id) = match find_vta_key_paths(&vta_did, keys_ks).await {
        Ok(paths) => paths,
        Err(e) => {
            warn!(
                "failed to find VTA key records: {e} — auth endpoints will not work (run setup first)"
            );
            return AuthInit::empty();
        }
    };

    // Load seed for VTA keys (uses the seed generation from the key record)
    let seed = match load_seed_bytes(keys_ks, seed_store, vta_seed_id).await {
        Ok(s) => s,
        Err(e) => {
            warn!("failed to load seed: {e} — auth endpoints will not work");
            return AuthInit::empty();
        }
    };

    let root = match ExtendedSigningKey::from_seed(&seed) {
        Ok(r) => r,
        Err(e) => {
            warn!("failed to create BIP-32 root key: {e} — auth endpoints will not work");
            return AuthInit::empty();
        }
    };

    // 1. DID resolver (network mode if resolver_url is set, local mode otherwise)
    let resolver_config = {
        let mut builder = DIDCacheConfigBuilder::default();
        if let Some(ref url) = config.resolver_url {
            info!(url = %url, "DID resolver using network mode (remote resolver)");
            builder = builder.with_network_mode(url);
        } else {
            info!("DID resolver using local mode");
        }
        builder.build()
    };
    let mut did_resolver = match DIDCacheClient::new(resolver_config).await {
        Ok(r) => r,
        Err(e) => {
            warn!("failed to create DID resolver: {e} — auth endpoints will not work");
            return AuthInit::empty();
        }
    };

    // 1.2. Preload the VTA's DID document into the resolver so that DIDComm consumers
    // can resolve it without a network round-trip.
    preload_self_did_document(&mut did_resolver, &vta_did, webvh_ks).await;

    // 2. Secrets resolver with VTA's Ed25519 + X25519 secrets
    let (secrets_resolver, _handle) = ThreadedSecretsResolver::new(None).await;

    // Track verification-method IDs so DIDComm consumers use the right fragment.
    let mut signing_vm_id: Option<String> = None;
    let mut ka_vm_id: Option<String> = None;

    if vta_did.starts_with("did:key:") {
        // did:key uses fragment IDs like {did}#{ed_pub_mb} and {did}#{x_pub_mb},
        // and the X25519 key is derived FROM the Ed25519 key (not independently).
        // Use the SDK helper which handles both correctly.
        let dp: vti_common::slip10::DerivationPath = match signing_path.parse() {
            Ok(p) => p,
            Err(e) => {
                warn!("invalid signing derivation path: {e}");
                return AuthInit {
                    did_resolver: Some(did_resolver),
                    ..AuthInit::empty()
                };
            }
        };
        match root.derive(&dp) {
            Ok(derived) => {
                let seed_bytes: &[u8; 32] = derived.signing_key.as_bytes();
                match vta_sdk::did_key::secrets_from_did_key(&vta_did, seed_bytes) {
                    Ok(secrets) => {
                        signing_vm_id = Some(secrets.signing.id.clone());
                        ka_vm_id = Some(secrets.key_agreement.id.clone());
                        info!(signing_id = %secrets.signing.id, ka_id = %secrets.key_agreement.id, "did:key secrets loaded");
                        secrets_resolver.insert(secrets.signing).await;
                        secrets_resolver.insert(secrets.key_agreement).await;
                    }
                    Err(e) => {
                        warn!("failed to build did:key secrets: {e} — auth will not work");
                        return AuthInit {
                            did_resolver: Some(did_resolver),
                            ..AuthInit::empty()
                        };
                    }
                }
            }
            Err(e) => warn!("failed to derive VTA signing key: {e}"),
        }
    } else {
        // did:webvh / other methods: use #key-0 / #key-1 fragment convention
        // with independently derived Ed25519 + X25519 keys.
        let ka_path = match ka_path {
            Some(p) => p,
            None => {
                warn!(
                    "VTA key-agreement record missing — auth endpoints will not work (run setup first)"
                );
                return AuthInit {
                    did_resolver: Some(did_resolver),
                    ..AuthInit::empty()
                };
            }
        };

        signing_vm_id = Some(format!("{vta_did}#key-0"));
        ka_vm_id = Some(format!("{vta_did}#key-1"));

        // Load stored key records for validation
        let stored_signing: Option<KeyRecord> = keys_ks
            .get(crate::keys::store_key(&format!("{vta_did}#key-0")))
            .await
            .ok()
            .flatten();
        let stored_ka: Option<KeyRecord> = keys_ks
            .get(crate::keys::store_key(&format!("{vta_did}#key-1")))
            .await
            .ok()
            .flatten();

        // Derive and insert VTA signing secret (Ed25519)
        match root.derive_ed25519(&signing_path) {
            Ok(mut signing_secret) => {
                if let Some(ref record) = stored_signing {
                    match signing_secret.get_public_keymultibase() {
                        Ok(runtime_pub) if runtime_pub != record.public_key => {
                            error!(
                                key_id = %format!("{vta_did}#key-0"),
                                stored = %record.public_key,
                                runtime = %runtime_pub,
                                "SIGNING KEY MISMATCH: runtime-derived Ed25519 public key does not match \
                                 the key stored in the key record (and published in the DID document). \
                                 DIDComm message signing/verification will fail. \
                                 This likely means the DID was created with different code or seed."
                            );
                        }
                        Ok(runtime_pub) => {
                            info!(key_id = %format!("{vta_did}#key-0"), pub_key = %runtime_pub, "signing key validated");
                        }
                        Err(e) => warn!("could not extract signing public key for validation: {e}"),
                    }
                }
                signing_secret.id = format!("{vta_did}#key-0");
                secrets_resolver.insert(signing_secret).await;
            }
            Err(e) => warn!("failed to derive VTA signing key: {e}"),
        }

        // Derive and insert VTA key-agreement secret (X25519)
        match root.derive_x25519(&ka_path) {
            Ok(mut ka_secret) => {
                if let Some(ref record) = stored_ka {
                    match ka_secret.get_public_keymultibase() {
                        Ok(runtime_pub) if runtime_pub != record.public_key => {
                            error!(
                                key_id = %format!("{vta_did}#key-1"),
                                stored = %record.public_key,
                                runtime = %runtime_pub,
                                "KEY-AGREEMENT KEY MISMATCH: runtime-derived X25519 public key does not match \
                                 the key stored in the key record (and published in the DID document). \
                                 DIDComm encryption/decryption will fail. Others will encrypt to the DID \
                                 document key but this VTA holds a different private key. \
                                 The DID document must be updated or the VTA identity must be regenerated."
                            );
                        }
                        Ok(runtime_pub) => {
                            info!(key_id = %format!("{vta_did}#key-1"), pub_key = %runtime_pub, "key-agreement key validated");
                        }
                        Err(e) => warn!("could not extract KA public key for validation: {e}"),
                    }
                }
                ka_secret.id = format!("{vta_did}#key-1");
                secrets_resolver.insert(ka_secret).await;
            }
            Err(e) => warn!("failed to derive VTA key-agreement key: {e}"),
        }
    }

    // 3. JWT signing key from config (random key, not BIP-32 derived)
    let jwt_keys = match &config.auth.jwt_signing_key {
        Some(b64) => match decode_jwt_key(b64) {
            Ok(k) => k,
            Err(e) => {
                warn!("failed to load JWT signing key: {e} — auth endpoints will not work");
                return AuthInit {
                    did_resolver: Some(did_resolver),
                    secrets_resolver: Some(Arc::new(secrets_resolver)),
                    signing_vm_id,
                    ka_vm_id,
                    ..AuthInit::empty()
                };
            }
        },
        None => {
            warn!(
                "auth.jwt_signing_key not configured — auth endpoints will not work (run setup first)"
            );
            return AuthInit {
                did_resolver: Some(did_resolver),
                secrets_resolver: Some(Arc::new(secrets_resolver)),
                signing_vm_id,
                ka_vm_id,
                ..AuthInit::empty()
            };
        }
    };

    // 4. Build ATM for DIDComm message unpacking (used by auth endpoints)
    let secrets_resolver = Arc::new(secrets_resolver);
    let atm = {
        let tdk_config = TDKConfig::builder()
            .with_did_resolver(did_resolver.clone())
            .with_secrets_resolver((*secrets_resolver).clone())
            .with_load_environment(false)
            .build();
        match tdk_config {
            Ok(cfg) => match TDKSharedState::new(cfg).await {
                Ok(tdk) => {
                    match ATM::new(ATMConfig::builder().build().unwrap(), Arc::new(tdk)).await {
                        Ok(a) => Some(a),
                        Err(e) => {
                            warn!("failed to create ATM for auth unpack: {e}");
                            None
                        }
                    }
                }
                Err(e) => {
                    warn!("failed to create TDK shared state: {e}");
                    None
                }
            },
            Err(e) => {
                warn!("failed to build TDK config: {e}");
                None
            }
        }
    };

    // 5. Build + register the VTA's TSP profile (used to unpack `tsp-message`
    // sealed envelopes). The unpack path reads the VTA's decryption key from
    // the ATM's own secrets resolver — which already holds the signing + KA
    // secrets inserted above (mirroring the DIDComm listener's secret
    // collection in `run()`) — so the profile itself carries no secrets and
    // needs no mediator. On any failure we warn! and leave it None; TSP unseal
    // just stays unavailable, never a panic.
    #[cfg(feature = "tsp")]
    let tsp_profile = if let Some(ref atm) = atm {
        match affinidi_tdk::messaging::profiles::ATMProfile::new(
            atm,
            Some("VTA".to_string()),
            vta_did.clone(),
            None,
        )
        .await
        {
            Ok(profile) => match atm.profile_add(&profile, false).await {
                Ok(arc) => {
                    info!("TSP profile registered for DID {vta_did}");
                    Some(arc)
                }
                Err(e) => {
                    warn!("failed to register TSP profile (TSP unseal disabled): {e}");
                    None
                }
            },
            Err(e) => {
                warn!("failed to build TSP profile (TSP unseal disabled): {e}");
                None
            }
        }
    } else {
        warn!("ATM unavailable — TSP profile not built (TSP unseal disabled)");
        None
    };

    info!("auth initialized for DID {vta_did}");

    AuthInit {
        did_resolver: Some(did_resolver),
        secrets_resolver: Some(secrets_resolver),
        jwt_keys: Some(Arc::new(jwt_keys)),
        atm,
        #[cfg(feature = "tsp")]
        tsp_profile,
        signing_vm_id,
        ka_vm_id,
    }
}

/// Seed the resolver with this VTA's own `did:webvh` document so it can
/// pack/unpack DIDComm messages without a network round-trip. `did:webvh`
/// self-resolution normally needs HTTPS to the VTA's own public domain, which
/// may be unreachable from inside the VTA's network (e.g. VPC-internal); the
/// locally stored `did.jsonl` (WEBVH keyspace) is the authoritative source, so
/// we seed the cache from it instead.
///
/// Scope: **`did:webvh` only** — the preload reads the local webvh log.
/// `did:web` and other network-resolved methods have no local log to seed from,
/// so they are left to normal resolver behaviour.
///
/// Staleness: this runs once at init, and the seeded resolver is now reused by
/// the DIDComm listener path (so auth + listener share one cache view). Every
/// runtime DID-log mutation — the did-webvh lifecycle (`create` / `update`) and
/// all protocol `services {…}` ops, which funnel through `update_did_webvh` —
/// reseeds this shared entry via `refresh_resolver_doc_from_log` right after the
/// new log is persisted, so service-advertisement changes stay in sync in the
/// VTA's in-process self-view.
///
/// Best-effort and fail-safe: if local state is missing or malformed we warn and
/// keep the last-known-good cache entry (never evict, never poison), falling back
/// to normal resolver behaviour where none exists.
pub(crate) async fn preload_self_did_document(
    did_resolver: &mut DIDCacheClient,
    vta_did: &str,
    webvh_ks: Option<&KeyspaceHandle>,
) {
    #[cfg(not(feature = "webvh"))]
    {
        // Without the `webvh` feature there is no local webvh log to seed from;
        // nothing to preload. Consume the params so the non-webvh build is
        // warning-clean under `-D warnings`.
        let _ = (&did_resolver, vta_did, webvh_ks);
    }

    #[cfg(feature = "webvh")]
    {
        if !vta_did.starts_with("did:webvh:") {
            return;
        }

        let Some(webvh_ks) = webvh_ks else {
            warn!(
                did = %vta_did,
                "webvh keyspace not available; self DID preload skipped"
            );
            return;
        };

        let Some(did_log) = (match crate::webvh_store::get_did_log(webvh_ks, vta_did).await {
            Ok(log) => log,
            Err(e) => {
                warn!(did = %vta_did, error = %e, "failed to read local did.jsonl for resolver preload");
                return;
            }
        }) else {
            warn!(did = %vta_did, "no local did.jsonl found for resolver preload");
            return;
        };

        let doc_value = match crate::operations::protocol::document::current_document_from_log(
            &did_log,
        ) {
            Ok(doc) => doc,
            Err(e) => {
                warn!(did = %vta_did, error = %e, "failed to parse local did.jsonl for resolver preload");
                return;
            }
        };

        let doc = match serde_json::from_value(doc_value) {
            Ok(doc) => doc,
            Err(e) => {
                warn!(did = %vta_did, error = %e, "failed to decode DID document for resolver preload");
                return;
            }
        };

        did_resolver.add_did_document(vta_did, doc).await;
        info!(did = %vta_did, "preloaded VTA DID into resolver cache from local did.jsonl");
    }
}

/// Look up VTA signing and key-agreement derivation paths from stored key records.
///
/// `did:webvh` (and other methods with independently-derived X25519) stores
/// records at both `#key-0` and `#key-1`. `did:key` stores only `#key-0`
/// because its X25519 key is curve-converted from Ed25519 at runtime, not
/// independently derived — there is no separate path to record.
///
/// Returns `(signing_path, ka_path, seed_id)` where `ka_path` is `None` for
/// `did:key` and `seed_id` comes from the signing key record.
async fn find_vta_key_paths(
    vta_did: &str,
    keys_ks: &KeyspaceHandle,
) -> Result<(String, Option<String>, Option<u32>), AppError> {
    let signing_key_id = format!("{vta_did}#key-0");

    let signing: KeyRecord = keys_ks
        .get(crate::keys::store_key(&signing_key_id))
        .await?
        .ok_or_else(|| AppError::NotFound("VTA signing key not found".into()))?;

    let ka_path = if vta_did.starts_with("did:key:") {
        None
    } else {
        let ka_key_id = format!("{vta_did}#key-1");
        let ka: KeyRecord = keys_ks
            .get(crate::keys::store_key(&ka_key_id))
            .await?
            .ok_or_else(|| AppError::NotFound("VTA key-agreement key not found".into()))?;
        Some(ka.derivation_path)
    };

    debug!(signing_path = %signing.derivation_path, ka_path = ?ka_path, "VTA key paths resolved");
    Ok((signing.derivation_path, ka_path, signing.seed_id))
}

/// Decode a base64url-no-pad JWT signing key and construct `JwtKeys`.
fn decode_jwt_key(b64: &str) -> Result<JwtKeys, AppError> {
    let bytes = BASE64
        .decode(b64)
        .map_err(|e| AppError::Config(format!("invalid jwt_signing_key base64: {e}")))?;
    let key_bytes: [u8; 32] = bytes
        .try_into()
        .map_err(|_| AppError::Config("jwt_signing_key must be exactly 32 bytes".into()))?;
    let keys = JwtKeys::from_ed25519_bytes(&key_bytes, "VTA")?;
    debug!("JWT signing key decoded successfully");
    Ok(keys)
}

/// Mediator-connect supervisor: readiness gate, connect, and reconnect.
///
/// Owns the whole mediator lifecycle that used to run inline once during
/// startup:
///
/// 1. **Self-readiness gate** — wait until the VTA's own DID resolves over the
///    network, so the mediator (which authenticates us by resolving that DID)
///    can actually fetch our sender key instead of 403-ing the handshake.
/// 2. **Connect**, with capped exponential backoff + full jitter. The classic
///    initial failure is the mediator's *own* resolver still negative-caching
///    the VTA host: it clears on its own timer, so retrying self-heals with no
///    operator restart where the previous one-shot connect needed one.
/// 3. **Supervise the session** — the inbound loop is awaited here rather than
///    detached, so when it ends the supervisor tears the session down and
///    reconnects instead of leaving the VTA silently deaf for the rest of the
///    process.
///
/// All of it runs in a spawned task: nothing here may block `server::run`, which
/// has to reach its shutdown/restart select for a signal to be honoured.
#[cfg(any(feature = "didcomm", feature = "tsp"))]
struct MessagingConnect {
    app_state: AppState,
    vta_did: String,
    messaging_config: crate::config::MessagingConfig,
    readiness: crate::config::MediatorReadinessConfig,
    resolver_url: Option<String>,
    outbox_ks: KeyspaceHandle,
    flush_queues: bool,
    shutdown: CancellationToken,
    /// Brings the whole process down when the readiness gate's `fail` policy
    /// trips. The gate runs off the startup path (so a SIGTERM mid-gate is
    /// honoured), which means it can no longer surface a fatal by returning
    /// `Err` from `run` — it signals here instead.
    fatal_shutdown: watch::Sender<bool>,
    /// Set alongside `fatal_shutdown` so `run` can turn the same condition into
    /// a non-zero exit status once the threads are joined.
    fatal_flag: Arc<std::sync::atomic::AtomicBool>,
}

#[cfg(any(feature = "didcomm", feature = "tsp"))]
impl MessagingConnect {
    /// Gate, then run the connect/supervise/reconnect loop until shutdown, the
    /// configured horizon, or a `reconnect = false` single-shot failure.
    async fn run(self) {
        match crate::messaging::readiness::run_gate(
            &self.vta_did,
            &self.readiness,
            self.resolver_url.as_deref(),
            &self.shutdown,
        )
        .await
        {
            Ok(crate::messaging::readiness::GateDecision::Proceed) => {}
            Ok(crate::messaging::readiness::GateDecision::Skip) => {
                info!(
                    "DIDComm messaging not started this boot \
                     (mediator self-readiness gate: skip)"
                );
                return;
            }
            Err(e) => {
                error!("mediator self-readiness gate failed, shutting down: {e}");
                self.fatal_flag
                    .store(true, std::sync::atomic::Ordering::Relaxed);
                let _ = self.fatal_shutdown.send(true);
                return;
            }
        }

        // One-shot, *before* the first connect attempt and never repeated. Both
        // of these delete queued messages at the mediator, so re-running them on
        // every retry would keep destroying traffic that arrived between
        // attempts.
        self.run_startup_recovery().await;

        self.supervise().await;
    }

    /// The connect → supervise-session → reconnect loop.
    async fn supervise(&self) {
        // The give-up / how-long-to-wait arithmetic lives in `ReconnectPolicy`
        // so it is unit-testable without a mediator; this loop owns the effects.
        let policy = crate::messaging::readiness::ReconnectPolicy::from_config(&self.readiness);
        // `failing_for` is measured from the start of the current run of
        // failures, not from process start — a VTA that ran healthily for a week
        // and then dropped gets the full horizon to reconnect, rather than
        // blowing a 1-hour budget it exhausted six days ago.
        let mut window_started = tokio::time::Instant::now();
        let mut attempt: u32 = 0;

        loop {
            if self.shutdown.is_cancelled() {
                return;
            }

            // Never touch the mediator unless the VTA can resolve its own DID
            // over the network (the same path the mediator takes). A cold VTA
            // that can't resolve itself would only storm the mediator with
            // unresolvable-sender auth attempts.
            if crate::messaging::readiness::self_did_network_resolvable(
                &self.vta_did,
                self.resolver_url.as_deref(),
            )
            .await
            {
                match self.connect_once().await {
                    Ok(messaging) => {
                        info!("DIDComm messaging started");
                        let session_started = tokio::time::Instant::now();

                        // Await the inbound loop rather than detaching it: its
                        // return is the only signal that this session is over.
                        crate::messaging::service::run_inbound_loop(
                            messaging.clone(),
                            self.app_state.clone(),
                            self.vta_did.clone(),
                            self.messaging_config.mediator_did.clone(),
                            self.shutdown.clone(),
                        )
                        .await;

                        // Session over. Unpublish before tearing down so no
                        // caller packs onto a socket we're about to stop, then
                        // stop it — the mediator allows one socket per DID, so a
                        // reconnect while the old socket still auto-reconnects
                        // would have the two duel over the slot.
                        self.app_state.didcomm_bridge.clear_messaging();
                        messaging.atm.graceful_shutdown().await;

                        if self.shutdown.is_cancelled() {
                            return;
                        }

                        let lasted = session_started.elapsed();
                        warn!(
                            session_secs = lasted.as_secs(),
                            "mediator session ended unexpectedly; reconnecting"
                        );
                        if policy.session_was_healthy(lasted) {
                            attempt = 0;
                            window_started = tokio::time::Instant::now();
                        }
                    }
                    Err(e) => warn!("failed to start DIDComm messaging: {e}"),
                }
            } else {
                warn!(
                    vta_did = %self.vta_did,
                    "VTA not yet self-resolvable over the network; deferring mediator connect"
                );
            }

            // `None` = stop: either reconnect is disabled (legacy single-shot:
            // give up until the next restart) or the horizon is exhausted.
            let Some(sleep_for) = policy.next_backoff(attempt, window_started.elapsed()) else {
                if self.readiness.reconnect {
                    warn!(
                        waited_secs = window_started.elapsed().as_secs(),
                        "mediator reconnect gave up after the configured horizon; \
                         a later restart will retry"
                    );
                }
                return;
            };
            debug!(
                attempt = attempt + 1,
                ceiling_secs = policy.ceiling_for(attempt).as_secs_f64(),
                sleep_secs = sleep_for.as_secs_f64(),
                "mediator connection not established; backing off before retry \
                 (exponential backoff + full jitter)"
            );
            tokio::select! {
                _ = self.shutdown.cancelled() => {
                    info!("shutdown during mediator reconnect backoff; stopping supervisor");
                    return;
                }
                _ = tokio::time::sleep(sleep_for) => {}
            }
            attempt = attempt.saturating_add(1);
        }
    }

    /// Best-effort, destructive queue recovery that runs **once** per process,
    /// before the first connect attempt.
    ///
    /// Clearing this DID's mediator inbox stops a poison / undeliverable backlog
    /// stalling the pickup handshake and wedging the (shared DIDComm+TSP)
    /// socket. `--flush-queues` additionally clears the OUTBOUND (sender) queue:
    /// a message loop can fill it, after which the mediator rejects every new
    /// send until it drains, and the inbox drain cannot touch it.
    async fn run_startup_recovery(&self) {
        let messaging_config = &self.messaging_config;
        let vta_did = self.vta_did.as_str();

        if messaging_config.drain_inbox_on_start || self.flush_queues {
            match self.app_state.atm.as_ref() {
                Some(atm) => {
                    let cleared =
                        drain_mediator_inbox(atm, &messaging_config.mediator_did, vta_did).await;
                    info!(
                        count = cleared,
                        mediator = %messaging_config.mediator_did,
                        "cleared queued mediator inbox before going live"
                    );
                }
                None => warn!("inbox drain requested but no ATM is available; skipping"),
            }
        }

        if self.flush_queues {
            match self.app_state.atm.as_ref() {
                Some(atm) => {
                    let cleared =
                        flush_mediator_outbox(atm, &messaging_config.mediator_did, vta_did).await;
                    info!(
                        count = cleared,
                        mediator = %messaging_config.mediator_did,
                        "flush_queues: cleared outbound sender queue before going live"
                    );
                }
                None => {
                    warn!("--flush-queues set but no ATM is available; skipping outbox flush")
                }
            }
        }
    }

    /// One connect attempt: build the messaging service and, on success, wire
    /// the bridge/registry/ACL and hand the live session back to the caller,
    /// which owns its inbound loop. Returns `Err` (to be retried) if the connect
    /// fails; `build_messaging` tears its own socket down on every error path.
    async fn connect_once(&self) -> Result<Arc<crate::messaging::service::VtaMessaging>, String> {
        let app_state = &self.app_state;
        let messaging_config = &self.messaging_config;
        let vta_did = self.vta_did.as_str();

        let sr = app_state
            .secrets_resolver
            .as_ref()
            .ok_or_else(|| "no secrets resolver available".to_string())?;

        // Collect secrets using the VM IDs from init_auth (correct for both
        // did:key and did:webvh — avoids hardcoding #key-0/#key-1 fragments).
        let mut secrets = Vec::new();
        if let Some(ref signing_id) = app_state.signing_vm_id
            && let Some(s) = sr.get_secret(signing_id).await
        {
            secrets.push(s);
        }
        if let Some(ref ka_id) = app_state.ka_vm_id
            && let Some(s) = sr.get_secret(ka_id).await
        {
            secrets.push(s);
        }

        let messaging = Arc::new(
            crate::messaging::service::build_messaging(
                secrets,
                vta_did,
                &messaging_config.mediator_did,
                self.outbox_ks.clone(),
                app_state.did_resolver.as_ref(),
                self.resolver_url.as_deref(),
            )
            .await?,
        );

        // Publish the outbound wiring so every REST/DIDComm component can send to
        // a peer over this one connection (`DIDCommBridge` → `MessagingService`).
        // Replaces any previous session's wiring on a reconnect.
        app_state.didcomm_bridge.set_messaging(
            messaging.service.clone(),
            (*messaging.atm).clone(),
            vta_did.to_string(),
        );

        // Register the config-loaded mediator in the listener registry so the
        // delegated step-up push (buffered through the registry) can reach
        // approvers on this mediator.
        #[cfg(all(feature = "webvh", feature = "didcomm"))]
        app_state
            .mediator_registry
            .record_activate(crate::messaging::registry::MediatorBinding {
                mediator_did: messaging_config.mediator_did.clone(),
                endpoint: messaging_config.mediator_url.clone(),
            })
            .await;

        // Set the VTA's own allow-all ACL on the mediator (keyed on the VTA DID,
        // so it authorises both DIDComm and TSP on the shared socket). Only when
        // `setup_acl` is set (mediators in ExplicitAllow mode).
        if messaging_config.setup_acl {
            if let Some(atm) = app_state.atm.as_ref() {
                acl_setup::set_client_acl_on_connection(
                    atm,
                    vta_did,
                    messaging_config.mediator_did.as_str(),
                    "vta-main",
                    "vta",
                )
                .await;
            } else {
                warn!("setup_acl = true but no ATM available; skipping mediator ACL provisioning");
            }
        }

        Ok(messaging)
    }
}

/// Best-effort REST drain of this DID's mediator inbox, run at startup when
/// `messaging.drain_inbox_on_start` is set.
///
/// The mediator allows one live-delivery websocket per DID, so an
/// undeliverable/poison message queued for this DID can stall the pickup
/// handshake and wedge the (shared DIDComm + TSP) listener indefinitely. REST
/// auth + message-pickup keep working even when the websocket stalls, so this
/// fetches the queued messages over REST and deletes them, clearing the wedge
/// before the live listener starts. Each cleared message is logged; a batch that
/// can't be fetched is logged loudly and stops the drain (so it can't spin).
/// Returns how many were cleared. Never panics — a failure just skips the drain.
#[cfg(any(feature = "didcomm", feature = "tsp"))]
async fn drain_mediator_inbox(atm: &ATM, mediator_did: &str, vta_did: &str) -> usize {
    // A mediator-connected profile for REST pickup — added without live delivery
    // (`false`), so this never opens the websocket that's the thing wedging.
    let profile = match affinidi_tdk::messaging::profiles::ATMProfile::new(
        atm,
        Some("VTA-drain".to_string()),
        vta_did.to_string(),
        Some(mediator_did.to_string()),
    )
    .await
    {
        Ok(p) => match atm.profile_add(&p, false).await {
            Ok(arc) => arc,
            Err(e) => {
                warn!(error = %e, "drain: could not register mediator profile; skipping drain");
                return 0;
            }
        },
        Err(e) => {
            warn!(error = %e, "drain: could not build mediator profile; skipping drain");
            return 0;
        }
    };

    let mut cleared = 0usize;
    // Bounded so a mediator that keeps re-queuing can never spin forever.
    for _round in 0..200 {
        let batch = match atm
            .message_pickup()
            .send_delivery_request(&profile, Some(20), true)
            .await
        {
            Ok(b) => b,
            Err(e) => {
                warn!(error = %e, cleared, "drain: delivery-request failed; stopping — a queued message may be un-fetchable, inspect the mediator");
                break;
            }
        };
        if batch.is_empty() {
            break;
        }
        let ids: Vec<String> = batch.iter().map(|(msg, _meta)| msg.id.clone()).collect();
        for (msg, _meta) in &batch {
            warn!(id = %msg.id, msg_type = %msg.typ, "drain: clearing queued mediator message");
        }
        match atm
            .message_pickup()
            .send_messages_received(&profile, &ids, true)
            .await
        {
            Ok(_) => cleared += ids.len(),
            Err(e) => {
                warn!(error = %e, cleared, "drain: delete failed; stopping to avoid a loop");
                break;
            }
        }
    }
    cleared
}

/// Clear this DID's OUTBOUND (sender) queue at the mediator — messages it sent
/// that are still queued for delivery to recipients.
///
/// A message loop (e.g. a wedged consent/update retry) can pile these up until
/// the mediator's per-sender cap (`limits.queue.sender`) trips and rejects every
/// new send, wedging the VTA's outbound path. [`drain_mediator_inbox`] can't
/// touch these — they live in the sender's outbox, not this DID's inbox — so
/// this lists the outbox and deletes each message. The VTA is authorised to
/// delete them because the mediator's owner check admits the message's `FROM`
/// (sender), not only its `TO` (recipient). Best-effort; returns how many were
/// cleared and never panics.
#[cfg(any(feature = "didcomm", feature = "tsp"))]
async fn flush_mediator_outbox(atm: &ATM, mediator_did: &str, vta_did: &str) -> usize {
    use affinidi_tdk::messaging::messages::{DeleteMessageRequest, Folder};

    let profile = match affinidi_tdk::messaging::profiles::ATMProfile::new(
        atm,
        Some("VTA-flush-outbox".to_string()),
        vta_did.to_string(),
        Some(mediator_did.to_string()),
    )
    .await
    {
        Ok(p) => match atm.profile_add(&p, false).await {
            Ok(arc) => arc,
            Err(e) => {
                warn!(error = %e, "flush outbox: could not register mediator profile; skipping");
                return 0;
            }
        },
        Err(e) => {
            warn!(error = %e, "flush outbox: could not build mediator profile; skipping");
            return 0;
        }
    };

    let list = match atm.list_messages(&profile, Folder::Outbox).await {
        Ok(l) => l,
        Err(e) => {
            warn!(error = %e, "flush outbox: could not list outbound queue; skipping");
            return 0;
        }
    };
    let ids: Vec<String> = list.iter().map(|m| m.msg_id.clone()).collect();
    if ids.is_empty() {
        return 0;
    }

    // The mediator caps a single delete request at 100 ids, so batch.
    let mut cleared = 0usize;
    for chunk in ids.chunks(100) {
        match atm
            .delete_messages_direct(
                &profile,
                &DeleteMessageRequest {
                    message_ids: chunk.to_vec(),
                },
            )
            .await
        {
            Ok(r) => cleared += r.success.len(),
            Err(e) => {
                warn!(error = %e, cleared, "flush outbox: delete batch failed; stopping");
                break;
            }
        }
    }
    cleared
}

// The framework's `ListenerEvent`-based `spawn_event_logger` is gone with the
// `DIDCommService`. Messaging connectivity is now read live (non-latched, R6.2)
// off `MessagingService::status()` via `DIDCommBridge::messaging_status_str`.

async fn shutdown_signal() {
    let ctrl_c = async {
        tokio::signal::ctrl_c()
            .await
            .expect("failed to install Ctrl+C handler");
    };

    #[cfg(unix)]
    let terminate = async {
        tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
            .expect("failed to install SIGTERM handler")
            .recv()
            .await;
    };

    #[cfg(not(unix))]
    let terminate = std::future::pending::<()>();

    tokio::select! {
        () = ctrl_c => info!("received SIGINT"),
        () = terminate => info!("received SIGTERM"),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::keys::{KeyType, save_key_record};
    use crate::store::Store;
    #[cfg(feature = "webvh")]
    use crate::webvh_store;
    use affinidi_did_resolver_cache_sdk::config::DIDCacheConfigBuilder;
    use vti_common::config::StoreConfig;

    fn temp_keys_ks() -> (Store, KeyspaceHandle, tempfile::TempDir) {
        let dir = tempfile::tempdir().expect("temp dir");
        let store = Store::open(&StoreConfig {
            data_dir: dir.path().to_path_buf(),
        })
        .expect("store open");
        let keys_ks = store
            .keyspace(crate::keyspaces::KEYS)
            .expect("keys keyspace");
        (store, keys_ks, dir)
    }

    #[cfg(feature = "webvh")]
    fn temp_webvh_ks() -> (Store, KeyspaceHandle, tempfile::TempDir) {
        let dir = tempfile::tempdir().expect("temp dir");
        let store = Store::open(&StoreConfig {
            data_dir: dir.path().to_path_buf(),
        })
        .expect("store open");
        let webvh_ks = store
            .keyspace(crate::keyspaces::WEBVH)
            .expect("webvh keyspace");
        (store, webvh_ks, dir)
    }

    /// `did:key` VTAs only store the Ed25519 signing record at `#key-0`;
    /// the X25519 key-agreement secret is curve-converted from Ed25519 at
    /// runtime, so a `#key-1` record is intentionally absent.
    /// `find_vta_key_paths` must succeed without it.
    #[tokio::test]
    async fn find_vta_key_paths_returns_none_ka_for_did_key() {
        let (_store, keys_ks, _dir) = temp_keys_ks();
        let did = "did:key:z6MkTestKey";

        save_key_record(
            &keys_ks,
            &format!("{did}#key-0"),
            "m/44'/0'/0'",
            KeyType::Ed25519,
            "z6MkSigningPub",
            "VTA signing key",
            Some("vta"),
            Some(0),
        )
        .await
        .unwrap();

        let (signing_path, ka_path, seed_id) =
            find_vta_key_paths(did, &keys_ks).await.expect("paths");

        assert_eq!(signing_path, "m/44'/0'/0'");
        assert!(ka_path.is_none(), "did:key must not require #key-1 lookup");
        assert_eq!(seed_id, Some(0));
    }

    /// `did:webvh` (and any non-`did:key` method) keeps the
    /// independently-derived X25519 record at `#key-1`, and
    /// `find_vta_key_paths` must surface it.
    #[tokio::test]
    async fn find_vta_key_paths_loads_ka_for_did_webvh() {
        let (_store, keys_ks, _dir) = temp_keys_ks();
        let did = "did:webvh:abc:example.com:vta";

        save_key_record(
            &keys_ks,
            &format!("{did}#key-0"),
            "m/44'/0'/0'",
            KeyType::Ed25519,
            "z6MkSigningPub",
            "VTA signing key",
            Some("vta"),
            Some(0),
        )
        .await
        .unwrap();
        save_key_record(
            &keys_ks,
            &format!("{did}#key-1"),
            "m/44'/0'/1'",
            KeyType::X25519,
            "z6LSKaPub",
            "VTA key-agreement key",
            Some("vta"),
            Some(0),
        )
        .await
        .unwrap();

        let (signing_path, ka_path, _seed_id) =
            find_vta_key_paths(did, &keys_ks).await.expect("paths");

        assert_eq!(signing_path, "m/44'/0'/0'");
        assert_eq!(ka_path.as_deref(), Some("m/44'/0'/1'"));
    }

    /// A `did:webvh` setup that is missing its `#key-1` record is broken —
    /// `find_vta_key_paths` must return `NotFound` rather than silently
    /// degrading to a `None` ka_path. (The `did:key` short-circuit is
    /// keyed off the DID prefix, not off record presence, so a missing
    /// record for non-`did:key` is genuinely an error.)
    #[tokio::test]
    async fn find_vta_key_paths_errors_when_did_webvh_missing_ka() {
        let (_store, keys_ks, _dir) = temp_keys_ks();
        let did = "did:webvh:abc:example.com:vta";

        save_key_record(
            &keys_ks,
            &format!("{did}#key-0"),
            "m/44'/0'/0'",
            KeyType::Ed25519,
            "z6MkSigningPub",
            "VTA signing key",
            Some("vta"),
            Some(0),
        )
        .await
        .unwrap();

        let result = find_vta_key_paths(did, &keys_ks).await;
        assert!(
            matches!(result, Err(AppError::NotFound(_))),
            "expected NotFound for did:webvh missing #key-1, got {result:?}"
        );
    }

    /// Build an `AppConfig` from a TOML snippet for the message tests.
    fn cfg(toml_str: &str) -> AppConfig {
        toml::from_str::<AppConfig>(toml_str).expect("parse test config")
    }

    /// P0.9b: the missing-identity refusal names the *specific* gap and always
    /// points at the `--allow-degraded` escape hatch.
    #[test]
    fn missing_identity_message_names_absent_vta_did() {
        let msg = missing_identity_message(&cfg(""));
        assert!(msg.contains("vta_did is not configured"), "{msg}");
        assert!(msg.contains("vta setup"), "{msg}");
        assert!(msg.contains("--allow-degraded"), "{msg}");
    }

    #[test]
    fn missing_identity_message_names_absent_jwt_key() {
        // vta_did present, JWT signing key absent → the message must point at
        // the JWT key, not the DID.
        let msg = missing_identity_message(&cfg("vta_did = \"did:key:z6MkTest\"\n"));
        assert!(msg.contains("auth.jwt_signing_key"), "{msg}");
        assert!(!msg.contains("vta_did is not configured"), "{msg}");
        assert!(msg.contains("--allow-degraded"), "{msg}");
    }

    #[test]
    fn missing_identity_message_falls_back_to_key_material() {
        // Both identity fields present but key material unloadable (the real
        // gate reaches this arm when `init_auth` can't derive/find the keys).
        let msg = missing_identity_message(&cfg(
            "vta_did = \"did:key:z6MkTest\"\n[auth]\njwt_signing_key = \"AAAA\"\n",
        ));
        assert!(msg.contains("could not be loaded"), "{msg}");
        assert!(msg.contains("did:key:z6MkTest"), "{msg}");
        assert!(msg.contains("--allow-degraded"), "{msg}");
    }

    /// once we seed the resolver from local did.jsonl,
    /// resolving the VTA DID is cache-only and does not
    /// require network reachability to its own public domain.
    #[cfg(feature = "webvh")]
    #[tokio::test]
    async fn preload_self_did_document_makes_vta_did_resolvable_from_cache() {
        let (_store, webvh_ks, _dir) = temp_webvh_ks();
        let did = "did:webvh:QmScid:vta.example.com:vta";

        let log_line = serde_json::json!({
            "versionId": "1-test",
            "versionTime": "2026-05-06T00:00:00Z",
            "parameters": {},
            "state": {
                "@context": ["https://www.w3.org/ns/did/v1"],
                "id": did,
            },
        });
        let log = serde_json::to_string(&log_line).expect("serialize log line");
        webvh_store::store_did_log(&webvh_ks, did, &log)
            .await
            .expect("store did log");

        let mut resolver = DIDCacheClient::new(DIDCacheConfigBuilder::default().build())
            .await
            .expect("resolver init");

        preload_self_did_document(&mut resolver, did, Some(&webvh_ks)).await;

        let resolved = resolver.resolve(did).await.expect("resolve preloaded did");
        assert!(
            resolved.cache_hit,
            "preloaded DID was not served from cache"
        );

        let expected_value = crate::operations::protocol::document::current_document_from_log(&log)
            .expect("extract current DID document from did.jsonl");
        let expected_doc =
            serde_json::from_value(expected_value).expect("deserialize expected DID document");

        assert_eq!(
            resolved.doc, expected_doc,
            "resolved DID document should match local did.jsonl current state"
        );
    }

    /// Malformed local did.jsonl must be a safe no-op: preload logs a warning
    /// and leaves resolver behavior unchanged (no poisoned cache entry).
    #[cfg(feature = "webvh")]
    #[tokio::test]
    async fn preload_self_did_document_ignores_malformed_local_log() {
        let (_store, webvh_ks, _dir) = temp_webvh_ks();
        let did = "did:webvh:QmBadScid:vta.example.com:vta";

        webvh_store::store_did_log(&webvh_ks, did, "not-json")
            .await
            .expect("store malformed did log");

        let mut resolver = DIDCacheClient::new(DIDCacheConfigBuilder::default().build())
            .await
            .expect("resolver init");

        preload_self_did_document(&mut resolver, did, Some(&webvh_ks)).await;

        let result = resolver.resolve(did).await;
        assert!(
            result.is_err(),
            "malformed preload input should not seed resolver cache"
        );
    }
}