vtc-service 0.12.0

//! Trust-Task handlers for the `rooms/*` family.
//!
//! These are deliberately thin. Every one does the same four things in the same order —
//! parse, load the room, authorize, act — and each of those lives somewhere else:
//! [`super::wire`] owns the shapes, [`super::storage`] owns the invariants, and
//! [`super::authz`] owns the decision. A handler that starts making its own storage or
//! authorization judgements is a handler that has drifted from the other four.
//!
//! # Authorization order
//!
//! The room is loaded first, then the presentation is authorized **against that room**.
//! That ordering matters: visibility is a property of the room, and the private-tier
//! subject-binding requirement cannot be checked without knowing the tier. Authorizing
//! before loading would mean either guessing the tier or checking it twice.
//!
//! # What a handler never does
//!
//! Reach for `state.acl_ks`, `state.members_ks`, or the caller's session. A room operation
//! is authorized by the chain the room issued and nothing else — invariant I5. There is no
//! `AuthClaims` parameter here, and its absence is the point rather than an omission.

use serde_json::Value;
use trust_tasks_rs::{AsyncDispatcher, TrustTask};

use super::policy as room_policy;
use crate::routing::trust_task_admission::{RoomTraffic, admit_room};
use crate::server::AppState;
use crate::trust_tasks::helpers::{TrustTaskOutcome, app_error_to_reject, success_response};
use vti_common::audit::{AuditEvent, RoomOperationData};
use vti_rooms::audit::{self as rooms_audit, RoomOperation};
use vti_rooms::authz::{self, Action};
use vti_rooms::storage;
use vti_rooms::wire::{
    ChainBody, ChainResponse, ClaimOwnerBody, CommitsBody, CommitsResponse, CreateRoomBody,
    CreateRoomResponse, CurateRecordBody, CurateRecordResponse, GetRecordBody, GetRecordResponse,
    ListRecordsBody, ListRecordsResponse, MintEpochBody, MintEpochResponse, OwnerResponse,
    PruneBody, PruneResponse, PutRecordBody, PutRecordResponse, TransferOwnerBody,
};
use vti_rooms::{Record, RecordStatus, Room};
use vti_rooms_dtg::{DataIntegrityKeys, DtgChainVerifier, nomination};

/// The DID that actually signed this request, and the verifier to judge its chain with.
///
/// Two things a room operation needs and a session does not supply. The presenter comes
/// from the document's own `eddsa-jcs-2022` proof — not from any field in the payload —
/// because a presentation names what may be done, not who is doing it: unbound, it is a
/// bearer token that anyone observing it inherits.
/// The rooms routing table, built once.
///
/// `registered_uris()` is derived from the registrations, so it cannot disagree
/// with what is actually served — which is the entire point of building this
/// rather than keeping a `const` array beside a `match`.
static ROOMS: std::sync::LazyLock<AsyncDispatcher<RoomCtx, TrustTaskOutcome>> =
    std::sync::LazyLock::new(dispatcher);

/// Does this service dispatch `type_uri` as a room task?
///
/// Asked by the spine to decide whether an inbound document belongs to this
/// family. Reading it off the dispatcher means adding a task is one
/// registration: the spine follows automatically, where before it needed an arm
/// of its own and a matching entry in two URI arrays.
#[must_use]
pub(crate) fn serves(type_uri: &str) -> bool {
    ROOMS.registered_uris().contains(&type_uri) || REVOCATION_URIS.contains(&type_uri)
}

/// `rooms/revocations/{put,get}/0.1`: served beside the dispatcher, over the payload as
/// received ([`super::files::handle_revocations`]), because a list must reach its
/// verifier exactly as the room signed it.
const REVOCATION_URIS: &[&str] = &[
    vti_rooms::revocation_tasks::PUT_TYPE,
    vti_rooms::revocation_tasks::GET_TYPE,
];

/// Every URI this family serves — the served list, not a copy of one.
#[must_use]
pub(crate) fn served_uris() -> Vec<&'static str> {
    let mut uris = ROOMS.registered_uris();
    uris.extend_from_slice(REVOCATION_URIS);
    uris
}

/// Dispatch a room task to its handler.
///
/// `dispatch_or_reject` builds the framework's own error document when the
/// payload does not downcast or fails its spec policy, so this never has to
/// reconstruct one — and never has to clone the document to keep a copy for the
/// error path.
pub(crate) async fn dispatch(
    state: &AppState,
    doc: TrustTask<Value>,
    presenter: &str,
) -> TrustTaskOutcome {
    let ctx = RoomCtx {
        state: state.clone(),
        presenter: presenter.to_string(),
    };
    let type_uri = doc.type_uri.to_string();
    if REVOCATION_URIS.contains(&type_uri.as_str()) {
        let put = type_uri == vti_rooms::revocation_tasks::PUT_TYPE;
        return super::files::handle_revocations(doc, ctx, put).await;
    }
    match ROOMS
        .dispatch_or_reject(doc, ctx, format!("urn:uuid:{}", uuid::Uuid::new_v4()))
        .await
    {
        Ok(outcome) => outcome,
        Err(err_doc) => crate::trust_tasks::helpers::error_response(err_doc),
    }
}

/// What a `rooms/*` handler needs that is not in the document.
///
/// Owned, not borrowed, because `AsyncDispatcher` requires each handler's future
/// to be `'static` — a registration cannot close over a borrow of `AppState`.
/// `AppState` is `Clone` and cheap to clone (it is handles, not data).
///
/// Note what is *not* here: no `JoinAuthCtx`, no ACL, no session. A room
/// operation is authorized by the authority chain the room itself issued —
/// invariant I5 of `docs/05-design-notes/data-rooms.md` — and `presenter` is a
/// cryptographic fact about the request, verified by the spine, not an authority
/// this service confers.
#[derive(Clone)]
pub(crate) struct RoomCtx {
    pub state: AppState,
    pub presenter: String,
}

/// Every `rooms/*` task, keyed on the **type** of its payload.
///
/// This is the whole routing table, and the reason it is worth building: each
/// registration names a Rust type, and the framework derives the rest — the type
/// URI from `Payload::TYPE_URI`, the downcast, version routing, and
/// `unsupportedType`-vs-`unsupportedVersion`. No URI literal appears here, so a
/// verb cannot be dispatched under a URI nobody serves, and `registered_uris()`
/// is the served list rather than a second one kept in step by hand.
///
/// That second list is not hypothetical: `rooms/records/curate` was dispatched
/// and named in neither of the two URI arrays this repo used to keep, so every
/// version hint the service emitted was wrong about it.
pub(crate) fn dispatcher() -> AsyncDispatcher<RoomCtx, TrustTaskOutcome> {
    AsyncDispatcher::new()
        .on_async(handle_create)
        .on_async(handle_put_record)
        .on_async(handle_get_record)
        .on_async(handle_list_records)
        .on_async(handle_curate_record)
        .on_async(handle_mint_epoch)
        .on_async(handle_epoch_chain)
        .on_async(handle_epoch_prune)
        .on_async(handle_epoch_commits)
        .on_async(handle_transfer_owner)
        .on_async(handle_claim_owner)
        .on_async(super::files::handle_put_record_v2)
        .on_async(super::files::handle_get_record_v2)
        .on_async(super::files::handle_list_records_v2)
        .on_async(super::files::handle_upload_begin)
        .on_async(super::files::handle_upload_chunk)
        .on_async(super::files::handle_upload_commit)
        .on_async(super::files::handle_upload_abort)
        .on_async(super::files::handle_blob_get)
        .on_async(super::files::handle_download_chunk)
        .on_async(super::files::handle_info)
        .on_async(super::files::handle_usage)
        .on_async(super::files::handle_upload_begin_v2)
        .on_async(super::files::handle_upload_chunk_v2)
        .on_async(super::files::handle_upload_commit_v2)
        .on_async(super::files::handle_upload_abort_v2)
        .on_async(super::files::handle_blob_get_v2)
}

/// Charge an operation the room has just authorized to the presenter's budget in that
/// room ([`crate::routing::trust_task_admission::admit_room`]), before any of its work
/// is done. A refusal is a `429` with no body, which the HTTPS door replaces with the
/// refusal it recorded; off that door nothing is charged.
pub(super) fn charged(
    room_id: &str,
    presenter: &str,
    traffic: RoomTraffic,
) -> Result<(), TrustTaskOutcome> {
    if admit_room(room_id, presenter, traffic) {
        Ok(())
    } else {
        Err(TrustTaskOutcome {
            status: axum::http::StatusCode::TOO_MANY_REQUESTS,
            body: Vec::new(),
        })
    }
}

pub(super) fn presenter_and_verifier(
    state: &AppState,
    presenter: &str,
) -> (String, DtgChainVerifier) {
    // `without_zk`: this service has no zero-knowledge profile for a private room's subject
    // binding, and the verifier refuses those rather than serving a pooling defence nobody
    // checked. Swap for `with_zk` when the working group settles the profile.
    (
        presenter.to_string(),
        DtgChainVerifier::without_zk(Box::new(DataIntegrityKeys(state.trust_task_vm_resolver()))),
    )
}

/// Record a room operation, per §8 of the design note.
///
/// **What may be recorded is decided by `vti_rooms::audit`, never here.** On a `private`
/// room the actor is the actorless marker rather than a DID, because an audit log naming
/// every actor *is* the membership this service was built never to learn — assembled one
/// entry at a time by the component least able to notice it is doing so.
///
/// A failed write is logged and swallowed. The alternative is refusing an operation that
/// already succeeded, which turns an audit outage into a room outage and tells the caller
/// their write failed when it did not. The audit chain's own integrity is protected by its
/// hash chain, which makes a gap visible to a verifier — that is the right place for this
/// to be noticed, not in a handler's return value.
pub(super) async fn audit_room(
    state: &AppState,
    room: &Room,
    authorized: &authz::AuthorizedAction,
    operation: RoomOperation,
    record_key: Option<&str>,
) {
    let Some(writer) = state.audit_writer.as_ref() else {
        return;
    };
    let entry = rooms_audit::for_operation(room.visibility, authorized, record_key);
    let action = operation.action_name();

    if let Err(e) = writer
        .write(
            entry.actor.as_str(),
            None,
            AuditEvent::RoomOperation(RoomOperationData {
                room_id: entry.room_id,
                action: action.to_string(),
                record_key: entry.record_key,
                visibility: format!("{:?}", room.visibility).to_lowercase(),
            }),
        )
        .await
    {
        tracing::error!(error = %e, action, "failed to record a room audit entry");
    }
}

/// Seconds since the Unix epoch.
pub(super) fn now() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// Default retention when a creator does not state one.
///
/// Ninety days after the epoch lapses. Long enough that a room is not lost to a quiet
/// month, short enough that an abandoned one does not accumulate forever.
const DEFAULT_RETENTION_DAYS: u32 = 90;

/// One epoch lifetime in seconds — how long a newly created room is live before it needs
/// renewing. Not a per-room parameter yet: `rooms/create/0.1` has no member for it, and
/// inventing one locally would put this host's rooms out of conformance with the schema.
const EPOCH_LIFETIME_DAYS_SECS: u64 =
    vti_rooms::lifecycle::DEFAULT_EPOCH_LIFETIME_DAYS as u64 * 24 * 60 * 60;

/// Ask the active `rooms` policy whether this community will host this room.
///
/// Reads the creator's member row — the only place in this file that touches the
/// roster, and legitimate because the question is "will we lend our disk", not
/// "who belongs to this room". A DID with no member row reaches the policy as
/// `member: false` rather than being refused here, so an operator whose policy
/// admits strangers gets what they wrote.
async fn govern_creation(
    state: &AppState,
    authorized: &authz::AuthorizedCreate,
    visibility: vti_rooms::Visibility,
    retention_days: Option<u32>,
) -> Result<(), vti_common::error::AppError> {
    // Membership is the member row; the role is the ACL entry beside it — the
    // same split `ceremony::assemble` uses, so a rooms policy branches on the
    // same two facts every other purpose does.
    let member = crate::members::storage::get_member(&state.members_ks, authorized.owner_did())
        .await
        .unwrap_or(None);
    let role = crate::ceremony::assemble::load_actor_role(state, authorized.owner_did())
        .await
        .unwrap_or(None);

    let verified = room_policy::VerifiedRoomCreation::assemble(room_policy::RoomCreationFacts {
        now: chrono::Utc::now(),
        actor: room_policy::Actor {
            did: authorized.owner_did().to_string(),
            role,
            member: member.is_some(),
        },
        room: room_policy::RoomRequest {
            room_id: authorized.room_id().to_string(),
            visibility,
            owner_did: authorized.owner_did().to_string(),
            retention_days,
            community_owned: false,
        },
    })?;

    // A VTC with no active rooms policy **refuses**, rather than answering 500
    // and inviting a retry. `install_defaults` fills the row at boot, so the
    // gap means either a first boot that has not reached it or an operator who
    // deleted the row — and in both cases "we are not currently deciding this"
    // is a closed door, not a server fault. The message says which, because a
    // bare 403 on a fresh upgrade would send an operator hunting their
    // credentials instead of their policy list.
    let policy = match crate::policy::load_active_compiled(
        &state.active_policies_ks,
        &state.policies_ks,
        crate::policy::model::PolicyPurpose::Rooms,
    )
    .await
    {
        Ok(p) => p,
        Err(e) => {
            tracing::warn!(
                error = %e,
                room = %authorized.room_id(),
                "no active rooms policy; refusing the registration"
            );
            return Err(vti_common::error::AppError::Forbidden(
                "room creation denied (no-policy): this community has no active `rooms` \
                 policy, so it is not currently deciding whether to host rooms. The shipped \
                 default installs at boot; check `GET /v1/policies`."
                    .into(),
            ));
        }
    };

    room_policy::decide_room_creation(&verified, &policy)
}

/// `rooms/create/0.1`.
///
/// The creator brings the room's identifier; this service does not assign one. A room
/// identified by something its host chose could not move to another host without changing
/// identity, and portability is the property the whole family rests on.
pub(crate) async fn handle_create(
    doc: TrustTask<CreateRoomBody>,
    ctx: RoomCtx,
) -> TrustTaskOutcome {
    let state = &ctx.state;
    let presenter = ctx.presenter.as_str();
    let req = &doc.payload;

    // The one room operation no chain can authorize: the room has issued nothing yet, so all
    // this service has is the proof on the request. That proof is what makes `ownerDid` a
    // fact rather than a field — and the decision itself lives in `vti-rooms`, so a room host
    // and this service cannot disagree about who may register a room.
    // `presenter` is the proof's signer, verified by the spine against the document
    // as received. This handler used to verify it here; it cannot, once it is handed
    // a typed payload, because re-deriving the signed bytes from a parsed payload is
    // only sound if that type round-trips losslessly — see `JoinAuthCtx::
    // verified_signer`.
    let authorized = match authz::authorize_create(&req.room_id, &req.owner_did, presenter) {
        Ok(a) => a,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    if let Err(o) = charged(authorized.room_id(), presenter, RoomTraffic::Operation) {
        return o;
    }

    // …and the one room operation this community governs. Registering a room is
    // asking a host for its disk, which is the host's to refuse; every operation
    // *within* the room stays authorized by the room's own credentials, and the
    // handlers below never consult the roster. See `super::policy`.
    if let Err(e) = govern_creation(state, &authorized, req.visibility, req.retention_days).await {
        return app_error_to_reject(&doc, &e);
    }

    let room = Room {
        room_id: authorized.room_id().to_string(),
        // Read from the authorization rather than the payload. The two are equal — that is
        // what was just established — and taking it from here is what keeps them equal if
        // this ever grows a second way to be authorized.
        owner_did: authorized.owner_did().to_string(),
        visibility: req.visibility,
        // Every room this host creates is chained: joining a room means being able to read
        // it, and the alternative silently loses the room's contents at the first membership
        // change. Not yet a per-room choice because `rooms/create/0.1` has no member for one
        // and its schema is `additionalProperties: false` — inventing a field locally would
        // put these rooms out of conformance. A room wanting `FromJoin` is a spec change
        // first, exactly as `retention_days` was.
        retention_policy: vti_rooms::RetentionPolicy::Chained,
        // What the caller asked for, or the default that draws no expectation.
        anchor_cadence: req.anchor_cadence.unwrap_or_default(),
        epoch: 1,
        next_version: 1,
        retention_days: req.retention_days.unwrap_or(DEFAULT_RETENTION_DAYS),
        // A room is live from creation for one epoch lifetime; minting the next epoch
        // renews it. Nothing else moves this clock — see `vti_rooms::lifecycle`.
        epoch_expires_at: Some(now() + EPOCH_LIFETIME_DAYS_SECS),
        created_at: now(),
        updated_at: now(),
        mirror_of: None,
    };

    if let Err(e) = storage::create_room(&state.rooms_ks, &room).await {
        return app_error_to_reject(&doc, &e);
    }
    // Pinned to today's default, so a later change of default does not move its new
    // uploads. A failure here is not the room's: the first upload assigns it instead.
    if let Err(e) = state.room_storage.assign_new_room(&room.room_id).await {
        tracing::warn!(room = %room.room_id, error = %e, "room not yet assigned a storage config");
    }

    success_response(
        &doc,
        CreateRoomResponse {
            room_id: req.room_id.clone(),
            epoch: 1,
            // What was recorded, never what was asked: the two differ on a
            // host that predates this member, and echoing the request would
            // tell a caller their choice took effect when it did not.
            anchor_cadence: Some(room.anchor_cadence),
        },
    )
}

/// `rooms/records/put/0.1`.
pub(crate) async fn handle_put_record(
    doc: TrustTask<PutRecordBody>,
    ctx: RoomCtx,
) -> TrustTaskOutcome {
    let req = doc.payload.clone();
    put_record(&doc, &ctx, req, Vec::new()).await
}

/// Write a record: `rooms/records/put` 0.1 (no `blobs`) or 0.2.
///
/// A 0.1 write names no blobs, so rewriting a record that carried a file under 0.1
/// releases it: the new version does not name it, which is what 0.2 says a release is.
pub(super) async fn put_record<P>(
    doc: &TrustTask<P>,
    ctx: &RoomCtx,
    req: PutRecordBody,
    blobs: Vec<String>,
) -> TrustTaskOutcome {
    let state = &ctx.state;

    let room = match storage::get_room(&state.rooms_ks, &req.room_id).await {
        Ok(r) => r,
        Err(e) => return app_error_to_reject(doc, &e),
    };
    let (presenter, verifier) = presenter_and_verifier(state, &ctx.presenter);
    let authorized = match authz::authorize(
        &room,
        &req.presentation,
        Action::Write,
        &presenter,
        now(),
        &verifier,
        &state.room_revocations_ks,
    )
    .await
    {
        Ok(a) => a,
        Err(e) => return app_error_to_reject(doc, &e),
    };
    if let Err(o) = charged(&room.room_id, &presenter, RoomTraffic::Operation) {
        return o;
    }
    // A record sealed under a superseded epoch is the declared `epochMismatch`, not a
    // generic validation failure: the writer's remedy (apply the commit it missed) follows
    // from the code alone.
    if let Some(refusal) =
        vti_rooms::blob_tasks::record_epoch_refused(&room, req.sealed.as_ref().map(|s| s.epoch))
    {
        return super::files::refused(doc, refusal);
    }

    // Held across the write: the blobs it names cannot be collected, and their limits
    // cannot change, between this check and the references it records.
    let references = match state
        .room_files
        .prepare_record(&room, authorized.member(), &req.key, &blobs)
        .await
    {
        Ok(r) => r,
        Err(e) => {
            return super::files::refused(doc, vti_rooms::blob_tasks::record_blobs_refused(e));
        }
    };

    // The author is recorded only where the tier discloses one. On a private room
    // authorship lives inside the sealed body, and the storage layer refuses it here.
    //
    // It is the *verified* subject — who the chain says is acting — not the room's owner.
    // Recording the owner would have credited every write to one person, which on the
    // attributed tier is the whole of what the tier is for.
    let author = room
        .visibility
        .discloses_actor()
        .then(|| authorized.subject().to_string());

    let record = Record {
        key: req.key.clone(),
        version: 0, // assigned by the store from the room's counter
        epoch: req.sealed.as_ref().map(|s| s.epoch),
        status: RecordStatus::Active,
        pinned: false,
        sealed: req.sealed.as_ref().map(|s| s.ciphertext.clone()),
        nonce: req.sealed.as_ref().map(|s| s.nonce.clone()),
        sealed_version: None,
        cleartext: req
            .cleartext
            .as_ref()
            .map(|c| serde_json::to_value(c).unwrap_or(Value::Null)),
        author,
        updated_at: 0,
    };

    match storage::put_record(
        &state.rooms_ks,
        &state.room_records_ks,
        &req.room_id,
        record,
        req.expected_version,
        now(),
    )
    .await
    {
        Ok(stored) => {
            if let Err(e) = references.apply().await {
                // The record is written; only its blob references are not. Said loudly,
                // because usage now disagrees with the room by one file.
                tracing::error!(error = %e, room = %req.room_id, record = %stored.key,
                    "record stored but its blob references were not updated");
            }
            audit_room(
                state,
                &room,
                &authorized,
                RoomOperation::PutRecord,
                Some(&stored.key),
            )
            .await;
            success_response(
                doc,
                PutRecordResponse {
                    key: stored.key,
                    version: stored.version,
                    epoch: stored.epoch,
                },
            )
        }
        Err(e) => app_error_to_reject(doc, &e),
    }
}

/// `rooms/records/get/0.1`.
///
/// A read presents exactly as a write does. Authorizing reads by session would record a
/// member identifier on every access, and a period of those records reconstructs the
/// membership a sealed room exists to withhold — without breaking any cryptography.
pub(crate) async fn handle_get_record(
    doc: TrustTask<GetRecordBody>,
    ctx: RoomCtx,
) -> TrustTaskOutcome {
    let req = doc.payload.clone();
    get_record(&doc, &ctx, req, false).await
}

/// Read a record: `rooms/records/get` 0.1, or 0.2 (`v2`), which also names its blobs.
pub(super) async fn get_record<P>(
    doc: &TrustTask<P>,
    ctx: &RoomCtx,
    req: GetRecordBody,
    v2: bool,
) -> TrustTaskOutcome {
    let state = &ctx.state;

    let room = match storage::get_room(&state.rooms_ks, &req.room_id).await {
        Ok(r) => r,
        Err(e) => return app_error_to_reject(doc, &e),
    };
    let (presenter, verifier) = presenter_and_verifier(state, &ctx.presenter);
    let authorized = match authz::authorize(
        &room,
        &req.presentation,
        Action::Read,
        &presenter,
        now(),
        &verifier,
        &state.room_revocations_ks,
    )
    .await
    {
        Ok(a) => a,
        Err(e) => return app_error_to_reject(doc, &e),
    };
    if let Err(o) = charged(&room.room_id, &presenter, RoomTraffic::Operation) {
        return o;
    }

    // A 0.2 read names the blobs of the version it returns: the record and its
    // references are read under the lock every record write holds.
    let snapshot = match v2 {
        true => Some(state.room_files.read_snapshot().await),
        false => None,
    };
    match storage::get_record(&state.room_records_ks, &req.room_id, &req.key).await {
        Ok(record) => {
            audit_room(
                state,
                &room,
                &authorized,
                RoomOperation::GetRecord,
                Some(&req.key),
            )
            .await;
            // The response is a type, not the stored record with a member bolted
            // on. Serialising `Record` was what put a bare `sealed` string and
            // six undeclared members on the wire — see `GetRecordResponse`.
            let (commitment, trace) = record_verification(state, &req.room_id, &req.key).await;
            let response = GetRecordResponse::of(&record, commitment.as_ref(), trace);
            let Some(snapshot) = snapshot else {
                return success_response(doc, response);
            };
            super::files::render(
                doc,
                vti_rooms::blob_tasks::get_v2_response(&snapshot, &req.room_id, &req.key, response)
                    .await,
            )
        }
        Err(e) => app_error_to_reject(doc, &e),
    }
}

/// `rooms/records/list/0.1`.
///
/// Returns metadata, never bodies — and returns tombstones to a watermark caller, because a
/// puller that never sees a retraction resurrects the record on its next full rebuild.
pub(crate) async fn handle_list_records(
    doc: TrustTask<ListRecordsBody>,
    ctx: RoomCtx,
) -> TrustTaskOutcome {
    let req = doc.payload.clone();
    list_records(&doc, &ctx, req, false).await
}

/// List records: `rooms/records/list` 0.1, or 0.2 (`v2`), whose rows also name their blobs.
pub(super) async fn list_records<P>(
    doc: &TrustTask<P>,
    ctx: &RoomCtx,
    req: ListRecordsBody,
    v2: bool,
) -> TrustTaskOutcome {
    let state = &ctx.state;

    let room = match storage::get_room(&state.rooms_ks, &req.room_id).await {
        Ok(r) => r,
        Err(e) => return app_error_to_reject(doc, &e),
    };
    let (presenter, verifier) = presenter_and_verifier(state, &ctx.presenter);
    let authorized = match authz::authorize(
        &room,
        &req.presentation,
        Action::Read,
        &presenter,
        now(),
        &verifier,
        &state.room_revocations_ks,
    )
    .await
    {
        Ok(a) => a,
        Err(e) => return app_error_to_reject(doc, &e),
    };
    if let Err(o) = charged(&room.room_id, &presenter, RoomTraffic::Operation) {
        return o;
    }

    let snapshot = match v2 {
        true => Some(state.room_files.read_snapshot().await),
        false => None,
    };
    // Paginated honestly: the page the caller asked for, and a cursor when more
    // remain. This used to `take(limit)` and drop the rest without a word, so a
    // short page and a complete room were indistinguishable.
    let page = match storage::list_records_page(
        &state.room_records_ks,
        &req.room_id,
        req.prefix.as_deref(),
        req.since_version,
        req.cursor.as_deref(),
        req.limit,
    )
    .await
    {
        Ok(p) => p,
        Err(e) => return app_error_to_reject(doc, &e),
    };

    // A listing names no single record, so the entry records the room and the fact of a
    // listing — which is the event: who has seen what this room holds.
    audit_room(state, &room, &authorized, RoomOperation::ListRecords, None).await;
    let head = room_head(state, &req.room_id).await;
    let response = ListRecordsResponse {
        records: page.records.iter().map(|r| r.metadata()).collect(),
        cursor: page.cursor,
        // One head, so the three values a reader compares cannot come
        // from three moments.
        data_commitment: head
            .as_ref()
            .map(|h| vti_rooms::merkle::to_multibase(&h.root)),
        record_count: head.as_ref().map(|h| h.record_count),
        head_version: head.as_ref().map(|h| h.head_version),
    };
    let Some(snapshot) = snapshot else {
        return success_response(doc, response);
    };
    super::files::render(
        doc,
        vti_rooms::blob_tasks::list_v2_response(&snapshot, &req.room_id, response).await,
    )
}

/// The room's commitment and this record's trace, from one snapshot.
///
/// Both or neither. A trace with no commitment is a path to a root the reader
/// was not given — the specification forbids it with `dependentRequired`, and
/// serving one would be a host offering evidence it withheld the subject of.
///
/// A failure is answered with neither rather than with an error: these are
/// additional guarantees the specification makes OPTIONAL precisely so a host
/// that cannot assert one says nothing instead of something false. A member who
/// asked for a record and got an error because the *tree* was unhappy has lost
/// a working operation to an advisory one.
async fn record_verification(
    state: &AppState,
    room_id: &str,
    key: &str,
) -> (
    Option<vti_rooms::merkle::TreeHead>,
    Option<vti_rooms::merkle::InclusionProof>,
) {
    match storage::tree_head_with_trace(&state.room_records_ks, room_id, key).await {
        Ok((head, trace)) => (Some(head), trace),
        Err(e) => {
            tracing::error!(
                room = %room_id,
                error = %e,
                "could not commit the room; answering the read without a commitment or a trace"
            );
            (None, None)
        }
    }
}

/// The room's data commitment, or `None` if it could not be computed.
///
/// **A failure here must not fail the read.** The commitment is an additional
/// guarantee, and the specification makes it OPTIONAL precisely so a host that
/// cannot assert one says nothing rather than something false — a member who
/// asked for records and got an error because the *tree* was unhappy has lost a
/// working operation to an advisory one.
///
/// Logged rather than swallowed silently: a host that has quietly stopped
/// committing looks, to a member, exactly like a host that never did.
async fn room_head(state: &AppState, room_id: &str) -> Option<vti_rooms::merkle::TreeHead> {
    match storage::tree_head(&state.room_records_ks, room_id).await {
        Ok(head) => Some(head),
        Err(e) => {
            tracing::error!(
                room = %room_id,
                error = %e,
                "could not compute the room's data commitment; answering without one"
            );
            None
        }
    }
}

/// `rooms/epoch/mint/0.1`.
///
/// Restricted to `admin`. If any key-holder could mint an epoch, any member could evict any
/// other by minting one and declining to seal them the new key — silently, and with no
/// check possible here on a room whose membership this service cannot see. Binding it to an
/// action the *room* confers is what makes the restriction enforceable by a service that
/// knows nothing about the membership.
pub(crate) async fn handle_mint_epoch(
    doc: TrustTask<MintEpochBody>,
    ctx: RoomCtx,
) -> TrustTaskOutcome {
    let state = &ctx.state;
    let presenter = ctx.presenter.as_str();
    let req = &doc.payload;

    let room = match storage::get_room(&state.rooms_ks, &req.room_id).await {
        Ok(r) => r,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    let (presenter, verifier) = presenter_and_verifier(state, presenter);
    let authorized = match authz::authorize(
        &room,
        &req.presentation,
        Action::Admin,
        &presenter,
        now(),
        &verifier,
        &state.room_revocations_ks,
    )
    .await
    {
        Ok(a) => a,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    if let Err(o) = charged(&room.room_id, &presenter, RoomTraffic::Operation) {
        return o;
    }

    // The room's policy governs whether it has a chain at all, so a rung for a room that
    // does not chain is refused rather than dropped. Storing it would give the room a chain
    // it declared it would not have; dropping it silently would let a client believe the
    // room's history was being retained when it was not.
    if req.link.is_some() && !room.retention_policy.links_epochs() {
        return app_error_to_reject(
            &doc,
            &vti_common::error::AppError::Validation(format!(
                "room `{}` does not keep an epoch key chain, so it accepts no epoch link",
                req.room_id
            )),
        );
    }

    // The rung, before the advance. A link that does not describe *this* advance is refused
    // outright rather than stored beside it: a host that accepted a mismatched one would be
    // laundering someone else's key material into this room's history, and the members who
    // walked the original rung would never see the difference.
    if let Some(link) = &req.link
        && link.epoch != req.epoch
    {
        return app_error_to_reject(
            &doc,
            &vti_common::error::AppError::Validation(format!(
                "the epoch link is for epoch {} but this mint advances to {}",
                link.epoch, req.epoch
            )),
        );
    }

    match storage::advance_epoch(&state.rooms_ks, &req.room_id, req.epoch, now()).await {
        Ok(updated) => {
            // After the advance, so a rejected advance leaves no orphan rung. `put_epoch_link`
            // refuses to replace one it already holds, which is what makes a retried mint
            // safe and a second, different rung impossible.
            if let Some(link) = &req.link
                && let Err(e) =
                    storage::put_epoch_link(&state.room_epoch_links_ks, &req.room_id, link).await
            {
                return app_error_to_reject(&doc, &e);
            }
            // The commit, on the same terms and for the same reason: this is the
            // moment the committer holds it, and a room that advances without
            // leaving it somewhere fetchable forks every member who was not
            // online. `put_commit` refuses to replace one, because the first
            // published is the one members may already have applied.
            if let Some(commit) = &req.commit
                && let Err(e) =
                    storage::put_commit(&state.room_epoch_links_ks, &req.room_id, req.epoch, commit)
                        .await
            {
                return app_error_to_reject(&doc, &e);
            }
            audit_room(state, &room, &authorized, RoomOperation::MintEpoch, None).await;
            success_response(
                &doc,
                MintEpochResponse {
                    room_id: updated.room_id,
                    epoch: updated.epoch,
                },
            )
        }
        Err(e) => app_error_to_reject(&doc, &e),
    }
}

/// `rooms/epoch/prune/0.1`.
///
/// Drop the epoch key chain below an epoch. The **rungs** go, not the records: a
/// pruned room still holds every record and this host still serves them, and
/// what is gone is the ability to derive the keys they were sealed under. Closer
/// to losing a key than to shredding a document, and irreversible either way.
pub(crate) async fn handle_epoch_prune(
    doc: TrustTask<PruneBody>,
    ctx: RoomCtx,
) -> TrustTaskOutcome {
    let state = &ctx.state;
    let presenter = ctx.presenter.as_str();
    let req = &doc.payload;

    let room = match storage::get_room(&state.rooms_ks, &req.room_id).await {
        Ok(r) => r,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    let (presenter, verifier) = presenter_and_verifier(state, presenter);
    // `admin`, not `curate`: pruning makes no statement about any record, and
    // every member who can write can curate — which would put "end the room's
    // readable history" within reach of every writer.
    let authorized = match authz::authorize(
        &room,
        &req.presentation,
        Action::Admin,
        &presenter,
        now(),
        &verifier,
        &state.room_revocations_ks,
    )
    .await
    {
        Ok(a) => a,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    if let Err(o) = charged(&room.room_id, &presenter, RoomTraffic::Operation) {
        return o;
    }

    // Refused rather than performed. `beforeEpoch: currentEpoch` reads like
    // "keep from here" and would drop every rung, leaving a member who restarts
    // unable to derive anything below the epoch they are handed next — a
    // plausible typo with a consequence nobody would choose.
    if req.before_epoch >= room.epoch {
        return app_error_to_reject(
            &doc,
            &vti_common::error::AppError::Validation(format!(
                "`beforeEpoch` {} is at or above the current epoch {} of `{}`; pruning there \
                 would drop the whole chain, not shorten it",
                req.before_epoch, room.epoch, req.room_id
            )),
        );
    }

    let (pruned, earliest_rung) = match storage::prune_chain(
        &state.room_epoch_links_ks,
        &req.room_id,
        req.before_epoch,
    )
    .await
    {
        Ok(r) => r,
        Err(e) => return app_error_to_reject(&doc, &e),
    };

    audit_room(state, &room, &authorized, RoomOperation::MintEpoch, None).await;
    success_response(
        &doc,
        PruneResponse {
            room_id: req.room_id.clone(),
            pruned: pruned as u32,
            earliest_rung,
        },
    )
}

/// `rooms/epoch/commits/0.1`.
///
/// Serve the commits a member missed. Relaying these is safe on every tier — a
/// commit is ciphertext plus a leaf index and **names nobody**, which is the
/// difference from a Welcome, and why the host is deliberately off that path and
/// on this one.
pub(crate) async fn handle_epoch_commits(
    doc: TrustTask<CommitsBody>,
    ctx: RoomCtx,
) -> TrustTaskOutcome {
    let state = &ctx.state;
    let presenter = ctx.presenter.as_str();
    let req = &doc.payload;

    let room = match storage::get_room(&state.rooms_ks, &req.room_id).await {
        Ok(r) => r,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    let (presenter, verifier) = presenter_and_verifier(state, presenter);
    let authorized = match authz::authorize(
        &room,
        &req.presentation,
        Action::Read,
        &presenter,
        now(),
        &verifier,
        &state.room_revocations_ks,
    )
    .await
    {
        Ok(a) => a,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    if let Err(o) = charged(&room.room_id, &presenter, RoomTraffic::Operation) {
        return o;
    }

    let commits = match storage::commits_since(
        &state.room_epoch_links_ks,
        &req.room_id,
        req.since_epoch,
        req.limit.unwrap_or(100).clamp(1, 100) as usize,
    )
    .await
    {
        Ok(c) => c,
        Err(e) => return app_error_to_reject(&doc, &e),
    };

    audit_room(state, &room, &authorized, RoomOperation::ListRecords, None).await;
    success_response(
        &doc,
        CommitsResponse {
            room_id: req.room_id.clone(),
            commits,
            // Read from the room, never computed from the answer. They differ
            // exactly when a commit is missing, and that difference is what
            // tells a member a delivery is missing rather than their own state
            // being broken.
            room_epoch: room.epoch,
        },
    )
}

/// `rooms/epoch/chain/0.1`.
///
/// Serves the room's epoch key chain so a member can read what was written before they
/// joined, or before the last membership change they were absent for.
///
/// # Why `read`, and why serving this is safe
///
/// Reading the room and reading the parts of it written earlier are the same act, so they
/// take the same grant. What leaves here is ciphertext: every rung is one epoch's storage
/// key sealed under the next, and the key that opens one is a storage key this service never
/// holds. A party who obtained the whole chain and no epoch key would learn how many epochs
/// the room has had — which `Room.epoch`, which they can already read, told them.
///
/// That is the property that lets a *host* answer this at all, rather than requiring the
/// room's owner to be online whenever somebody joins.
pub(crate) async fn handle_epoch_chain(
    doc: TrustTask<ChainBody>,
    ctx: RoomCtx,
) -> TrustTaskOutcome {
    let state = &ctx.state;
    let presenter = ctx.presenter.as_str();
    let req = &doc.payload;

    let room = match storage::get_room(&state.rooms_ks, &req.room_id).await {
        Ok(r) => r,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    let (presenter, verifier) = presenter_and_verifier(state, presenter);
    let authorized = match authz::authorize(
        &room,
        &req.presentation,
        Action::Read,
        &presenter,
        now(),
        &verifier,
        &state.room_revocations_ks,
    )
    .await
    {
        Ok(a) => a,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    if let Err(o) = charged(&room.room_id, &presenter, RoomTraffic::Operation) {
        return o;
    }

    let mut links = match storage::list_epoch_links(&state.room_epoch_links_ks, &req.room_id).await
    {
        Ok(l) => l,
        Err(e) => return app_error_to_reject(&doc, &e),
    };

    // Highest epoch first: the chain is walked downwards, so a member reads it in the order
    // they will use it.
    links.reverse();
    if let Some(from) = req.from_epoch {
        links.retain(|l| l.epoch <= from);
    }
    if let Some(limit) = req.limit {
        links.truncate(limit as usize);
    }

    audit_room(state, &room, &authorized, RoomOperation::ListRecords, None).await;
    success_response(
        &doc,
        ChainResponse {
            room_id: req.room_id.clone(),
            links,
        },
    )
}

/// `rooms/owner/transfer/0.1`.
///
/// An owner hands the room to another member, deliberately and while still present.
///
/// Gated on `admin` — the same grant that mints epochs, because transferring is the more
/// consequential of the two and it would be strange for the lesser act to ask for more.
///
/// # What this handler does not check, and why that is correct
///
/// That the incoming owner is a member. The spec requires them to be one, and **this host
/// cannot see the MLS group**: it holds no roster and no group state. The temptation is to
/// invent a check anyway — demand a credential, or refuse when unsure — and both are worse
/// than doing nothing. A host that refuses what it cannot verify fails *every* transfer,
/// including every correct one, and a host that treats its own ignorance as evidence has
/// converted "I don't know" into "no".
///
/// The spec says as much in the error code itself: `notAMember` is raised only when a host
/// "could independently establish" it, and this one cannot. What protects the incoming
/// owner is that the outgoing one holds `admin` and knows the roster.
pub(crate) async fn handle_transfer_owner(
    doc: TrustTask<TransferOwnerBody>,
    ctx: RoomCtx,
) -> TrustTaskOutcome {
    let state = &ctx.state;
    let presenter = ctx.presenter.as_str();
    let req = &doc.payload;

    let room = match storage::get_room(&state.rooms_ks, &req.room_id).await {
        Ok(r) => r,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    let (presenter, verifier) = presenter_and_verifier(state, presenter);
    let operation = RoomOperation::TransferOwner;
    let authorized = match authz::authorize(
        &room,
        &req.presentation,
        operation.required_action(),
        &presenter,
        now(),
        &verifier,
        &state.room_revocations_ks,
    )
    .await
    {
        Ok(a) => a,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    if let Err(o) = charged(&room.room_id, &presenter, RoomTraffic::Operation) {
        return o;
    }

    match storage::set_owner(&state.rooms_ks, &req.room_id, &req.new_owner_did, now()).await {
        Ok(updated) => {
            audit_room(state, &room, &authorized, operation, None).await;
            success_response(
                &doc,
                OwnerResponse {
                    room_id: updated.room_id,
                    owner_did: updated.owner_did,
                },
            )
        }
        Err(e) => app_error_to_reject(&doc, &e),
    }
}

/// `rooms/owner/claim/0.1`.
///
/// A nominated successor takes a room whose owner has stopped renewing it.
///
/// # Three conditions, checked in the order that gives the most useful answer
///
/// 1. **A nomination the room issued**, naming this claimant, in its window, rooted at the
///    room. [`vti_rooms_dtg::nomination`].
/// 2. **The room is dormant** — not merely lapsed. [`Lifecycle::admits_a_claim`].
/// 3. **The claimant holds membership**, which is the presentation every other room task
///    carries, gated at `read`.
///
/// The order is deliberate. A claimant whose nomination is bad should hear that first,
/// because it is the one they can do nothing about by waiting; telling them "the room is
/// still live" when their credential was never valid sends them back in thirty days to
/// receive the real answer.
///
/// # Renewing is the defence, and it is the same act as ordinary use
///
/// Condition 2 is why a hostile nomination is survivable. An owner who was merely away
/// defeats every pending claim by minting an epoch — which is what they would have done
/// anyway. Nothing has to be revoked and no dispute has to be raised, so an owner who is
/// present is structurally safe from succession without ever thinking about it.
///
/// # And a claim does not renew the room
///
/// [`storage::set_owner`] leaves a dormant room dormant. The new owner's first act should
/// be the one that proves they can perform it; see that function for why.
pub(crate) async fn handle_claim_owner(
    doc: TrustTask<ClaimOwnerBody>,
    ctx: RoomCtx,
) -> TrustTaskOutcome {
    let state = &ctx.state;
    let presenter = ctx.presenter.as_str();
    let req = &doc.payload;

    let room = match storage::get_room(&state.rooms_ks, &req.room_id).await {
        Ok(r) => r,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    let (presenter, verifier) = presenter_and_verifier(state, presenter);

    // 1. The nomination, against the party who actually signed this request — never against
    //    a DID named in the payload, which the claimant chooses.
    let keys = DataIntegrityKeys(state.trust_task_vm_resolver());
    let nominated =
        match nomination::verify(&req.nomination, &room.room_id, &presenter, &keys).await {
            Ok(n) => n,
            Err(e) => return app_error_to_reject(&doc, &e),
        };

    // 2. Dormant, not merely lapsed. An epoch expiring is frequently somebody on holiday.
    let lifecycle = room.lifecycle(now());
    if !lifecycle.admits_a_claim() {
        return app_error_to_reject(
            &doc,
            &vti_common::error::AppError::Forbidden(format!(
                "room `{}` is {} — a room becomes claimable only once its epoch has \
                 lapsed and the grace window after it has also passed without a renewal",
                room.room_id,
                lifecycle.as_str()
            )),
        );
    }

    // 3. Membership, which is also the pooling defence and the chain-reaches-the-room check.
    let operation = RoomOperation::ClaimOwner;
    let authorized = match authz::authorize(
        &room,
        &req.presentation,
        operation.required_action(),
        &presenter,
        now(),
        &verifier,
        &state.room_revocations_ks,
    )
    .await
    {
        Ok(a) => a,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    // A nomination is a credential the room issued, and a revocation list governs
    // succession like every other task (`rooms/revocations/put`, *Enforcement* item 2).
    if let Err(e) = authorized.also_unrevoked(nominated.facts()) {
        return app_error_to_reject(&doc, &e);
    }
    if let Err(o) = charged(&room.room_id, &presenter, RoomTraffic::Operation) {
        return o;
    }

    match storage::set_owner(&state.rooms_ks, &req.room_id, &presenter, now()).await {
        Ok(updated) => {
            audit_room(state, &room, &authorized, operation, None).await;
            success_response(
                &doc,
                OwnerResponse {
                    room_id: updated.room_id,
                    owner_did: updated.owner_did,
                },
            )
        }
        Err(e) => app_error_to_reject(&doc, &e),
    }
}

/// `rooms/records/curate/0.1`.
///
/// Gated on `Action::Curate`, which is deliberately not implied by `write`: deciding what a
/// room's shared knowledge is worth is a different grant from the ability to add to it. A
/// community can hand an agent `write` without handing it the standing to demote what a
/// person wrote.
pub(crate) async fn handle_curate_record(
    doc: TrustTask<CurateRecordBody>,
    ctx: RoomCtx,
) -> TrustTaskOutcome {
    let state = &ctx.state;
    let presenter = ctx.presenter.as_str();
    let req = &doc.payload;
    if req.status.is_none() && req.pinned.is_none() {
        return app_error_to_reject(
            &doc,
            &vti_common::error::AppError::Validation(
                "a curation must change something: supply `status`, `pinned`, or both".into(),
            ),
        );
    }

    let room = match storage::get_room(&state.rooms_ks, &req.room_id).await {
        Ok(r) => r,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    let (presenter, verifier) = presenter_and_verifier(state, presenter);
    let authorized = match authz::authorize(
        &room,
        &req.presentation,
        Action::Curate,
        &presenter,
        now(),
        &verifier,
        &state.room_revocations_ks,
    )
    .await
    {
        Ok(a) => a,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    if let Err(o) = charged(&room.room_id, &presenter, RoomTraffic::Operation) {
        return o;
    }

    // Held across the curation, so a 0.2 read never sees a retracted record still
    // naming the blobs the retraction released.
    let release = match state
        .room_files
        .prepare_release(&req.room_id, &req.key)
        .await
    {
        Ok(r) => r,
        Err(e) => return app_error_to_reject(&doc, &e),
    };
    match storage::curate_record(
        &state.rooms_ks,
        &state.room_records_ks,
        &req.room_id,
        &req.key,
        storage::Curation {
            status: req.status,
            pinned: req.pinned,
            expected_version: req.expected_version,
        },
        now(),
    )
    .await
    {
        Ok(curated) => {
            // A retraction keeps no body, so it keeps no file: every blob the record
            // named is released, and orphaned if nothing else names it.
            if curated.status == RecordStatus::Retracted
                && let Err(e) = release.apply().await
            {
                tracing::error!(error = %e, room = %req.room_id, record = %curated.key,
                    "record retracted but its blobs were not released");
            }
            audit_room(
                state,
                &room,
                &authorized,
                RoomOperation::CurateRecord,
                Some(&curated.key),
            )
            .await;
            success_response(
                &doc,
                CurateRecordResponse {
                    key: curated.key,
                    version: curated.version,
                    status: curated.status,
                    pinned: curated.pinned,
                },
            )
        }
        Err(e) => app_error_to_reject(&doc, &e),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_support::build_test_vtc;
    use serde_json::json;
    use vti_rooms::Visibility;
    use vti_rooms_dtg::test_support::RoomFixture;

    /// Verify the document's proof the way the **spine** does, and return the
    /// signer.
    ///
    /// Every `rooms/*` specification declares `proof` REQUIRED, so
    /// `dispatch_trust_task_core` verifies one and hands the signer to these
    /// handlers. A test that called a handler with a presenter of its own
    /// choosing would assert nothing about the signature — it would be checking
    /// that the handler uses the argument it was given. Deriving it here keeps
    /// what these tests have always proved: that the identity a room operation
    /// is authorized against is the one that *signed the request*.
    async fn spine_presenter(state: &AppState, doc: &TrustTask<Value>) -> String {
        crate::trust_tasks::helpers::verify_trust_task_proof(state, doc)
            .await
            .expect("a test document is signed by its fixture owner")
    }

    // Thin shims, deliberately named after the handlers they wrap, so the call
    // sites below read exactly as they did when each handler verified its own
    // proof. They shadow the glob-imported originals; `super::` reaches past
    // them.
    //
    // They now go through `super::dispatch`, which is the production path: the
    // document is routed by its own `type` through the registered dispatcher and
    // downcast to the handler's payload type. So these tests cover the
    // registration too — a handler registered under the wrong type, or not
    // registered at all, fails here rather than in a deployment.
    async fn handle_create(state: &AppState, doc: TrustTask<Value>) -> TrustTaskOutcome {
        let presenter = spine_presenter(state, &doc).await;
        super::dispatch(state, doc, &presenter).await
    }

    async fn handle_put_record(state: &AppState, doc: TrustTask<Value>) -> TrustTaskOutcome {
        let presenter = spine_presenter(state, &doc).await;
        super::dispatch(state, doc, &presenter).await
    }

    async fn handle_get_record(state: &AppState, doc: TrustTask<Value>) -> TrustTaskOutcome {
        let presenter = spine_presenter(state, &doc).await;
        super::dispatch(state, doc, &presenter).await
    }

    async fn handle_list_records(state: &AppState, doc: TrustTask<Value>) -> TrustTaskOutcome {
        let presenter = spine_presenter(state, &doc).await;
        super::dispatch(state, doc, &presenter).await
    }

    async fn handle_mint_epoch(state: &AppState, doc: TrustTask<Value>) -> TrustTaskOutcome {
        let presenter = spine_presenter(state, &doc).await;
        super::dispatch(state, doc, &presenter).await
    }

    async fn handle_curate_record(state: &AppState, doc: TrustTask<Value>) -> TrustTaskOutcome {
        let presenter = spine_presenter(state, &doc).await;
        super::dispatch(state, doc, &presenter).await
    }

    async fn handle_transfer_owner(state: &AppState, doc: TrustTask<Value>) -> TrustTaskOutcome {
        let presenter = spine_presenter(state, &doc).await;
        super::dispatch(state, doc, &presenter).await
    }

    async fn handle_claim_owner(state: &AppState, doc: TrustTask<Value>) -> TrustTaskOutcome {
        let presenter = spine_presenter(state, &doc).await;
        super::dispatch(state, doc, &presenter).await
    }

    /// A **signed** room document.
    ///
    /// Signing is not ceremony here: the presenter comes from this proof, and every handler
    /// below refuses an unsigned request before it looks at any chain.
    async fn doc(
        state: &AppState,
        uri: &str,
        payload: Value,
        signer_did: &str,
        signer_key: &str,
    ) -> TrustTask<Value> {
        let recipient = state
            .config
            .read()
            .await
            .vtc_did
            .clone()
            .unwrap_or_else(|| "did:key:zVtc".to_string());
        let signed = vta_sdk::trust_task_sign::build_signed(
            uri, payload, signer_did, signer_key, &recipient,
        )
        .await
        .expect("sign the request");
        serde_json::from_str(&signed).expect("a signed document is a document")
    }

    fn payload_of(out: &TrustTaskOutcome) -> Value {
        let d: Value = serde_json::from_slice(&out.body).expect("response is JSON");
        d.get("payload").cloned().unwrap_or(Value::Null)
    }

    /// Make the fixture's room owner a member of this community.
    ///
    /// Needed because registering a room is now governed (`super::policy`): the
    /// shipped default hosts rooms for its own members, and every fixture owner
    /// is a fresh `did:key` this VTC has never met. Seeding the row is what the
    /// tests below were implicitly assuming before creation was governed at all.
    async fn seed_creator(state: &AppState, f: &RoomFixture) {
        // The shipped defaults, which a real VTC installs at boot. Without them
        // there is no active rooms policy and every registration is refused —
        // correctly, and unhelpfully for a test about something else.
        crate::policy::default::install_defaults(&state.policies_ks, &state.active_policies_ks)
            .await
            .expect("install the shipped default policies");

        crate::members::storage::store_member(
            &state.members_ks,
            &crate::members::Member::fresh(f.room.owner_did.clone()),
        )
        .await
        .expect("seed the creator's member row");
    }

    /// Enable `private` rooms, the way an operator would: by activating a rooms
    /// policy that permits the tier.
    ///
    /// The shipped default denies it (§7.4), so a test about *serving* a private
    /// room has to say the community decided to host them — which is also the
    /// working example of how an operator turns the tier on.
    async fn permit_private(state: &AppState) {
        let source = "package vtc.rooms\n\n\
             import rego.v1\n\n\
             default decision := {\"effect\": \"deny\", \"with\": {\"code\": \"not-a-member\"}}\n\n\
             decision := {\"effect\": \"allow\"} if input.actor.member == true\n";
        let id = uuid::Uuid::new_v4();
        let compiled =
            crate::policy::engine::compile(source, id).expect("the permissive policy compiles");
        let mut policy = crate::policy::storage::new_policy(
            crate::policy::model::PolicyPurpose::Rooms,
            source.to_string(),
            *compiled.source_sha256(),
            "did:key:zOperator".to_string(),
            2,
        );
        policy.id = id;
        crate::policy::storage::store_policy(&state.policies_ks, &policy)
            .await
            .expect("store");
        crate::policy::storage::set_active_policy_id(
            &state.active_policies_ks,
            crate::policy::model::PolicyPurpose::Rooms,
            id,
        )
        .await
        .expect("activate");
    }

    /// Register the fixture's room with this VTC.
    async fn create(state: &AppState, f: &RoomFixture) -> TrustTaskOutcome {
        seed_creator(state, f).await;
        if matches!(f.room.visibility, Visibility::Private) {
            permit_private(state).await;
        }
        handle_create(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_CREATE_TYPE,
                json!({
                    "roomId": f.room.room_id,
                    "visibility": f.room.visibility,
                    "ownerDid": f.room.owner_did,
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await
    }

    /// SPEC §7.2 item 11, through the **spine** rather than the rooms
    /// sub-dispatcher — the layer every transport shares.
    ///
    /// Live delivery is at-least-once: the mediator re-pushes a recipient's whole
    /// undelivered inbox when a socket enables live delivery and when a duplicate
    /// socket displaces an existing session, so the same signed document arrives
    /// twice as routine traffic. The proof does not catch it — the redelivered
    /// copy carries the same valid proof, because it is the same document.
    ///
    /// Registration is the clearest witness: without a record the second copy is
    /// executed and *fails* ("that identifier is taken"), so a redelivery turns a
    /// success the caller may never have seen into an error. With one it is
    /// answered with the original result.
    #[tokio::test]
    async fn the_spine_executes_a_redelivered_document_once() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        seed_creator(state, &f).await;

        let signed = doc(
            state,
            vti_rooms::wire::ROOMS_CREATE_TYPE,
            json!({
                "roomId": f.room.room_id,
                "visibility": f.room.visibility,
                "ownerDid": f.room.owner_did,
            }),
            &f.owner.did,
            &f.owner.secret_multibase,
        )
        .await;
        // The bytes as they would arrive twice off the wire.
        let body = serde_json::to_vec(&signed).expect("a document serialises");

        let first = crate::trust_tasks::dispatch_trust_task_core(
            state,
            &crate::trust_tasks::JoinAuthCtx::rest(),
            &body,
        )
        .await;
        assert!(
            first.status.is_success(),
            "the first registration must succeed: {}",
            payload_of(&first)
        );

        let second = crate::trust_tasks::dispatch_trust_task_core(
            state,
            &crate::trust_tasks::JoinAuthCtx::rest(),
            &body,
        )
        .await;
        assert!(
            second.status.is_success(),
            "a duplicate is never a failure — the task did not fail, it already \
             happened: {}",
            payload_of(&second)
        );
        assert_eq!(
            payload_of(&second),
            payload_of(&first),
            "the duplicate must be answered with the original result"
        );
    }

    /// A *different* document under an already-spent `id` is a conflict, not a
    /// retry. The record is digest-keyed precisely so the two are
    /// distinguishable; absorbing the second as a duplicate is the one outcome
    /// §7.2 rules out.
    #[tokio::test]
    async fn the_spine_refuses_a_different_document_reusing_an_id() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        let g = RoomFixture::new(Visibility::Open).await;
        seed_creator(state, &f).await;
        crate::members::storage::store_member(
            &state.members_ks,
            &crate::members::Member::fresh(g.room.owner_did.clone()),
        )
        .await
        .expect("seed the second creator");

        let first = doc(
            state,
            vti_rooms::wire::ROOMS_CREATE_TYPE,
            json!({
                "roomId": f.room.room_id,
                "visibility": f.room.visibility,
                "ownerDid": f.room.owner_did,
            }),
            &f.owner.did,
            &f.owner.secret_multibase,
        )
        .await;
        let out = crate::trust_tasks::dispatch_trust_task_core(
            state,
            &crate::trust_tasks::JoinAuthCtx::rest(),
            &serde_json::to_vec(&first).unwrap(),
        )
        .await;
        assert!(out.status.is_success(), "{}", payload_of(&out));

        // A genuinely different registration, wearing the first document's id.
        let mut second = doc(
            state,
            vti_rooms::wire::ROOMS_CREATE_TYPE,
            json!({
                "roomId": g.room.room_id,
                "visibility": g.room.visibility,
                "ownerDid": g.room.owner_did,
            }),
            &g.owner.did,
            &g.owner.secret_multibase,
        )
        .await;
        second.id = first.id.clone();

        let out = crate::trust_tasks::dispatch_trust_task_core(
            state,
            &crate::trust_tasks::JoinAuthCtx::rest(),
            &serde_json::to_vec(&second).unwrap(),
        )
        .await;

        assert!(
            !out.status.is_success(),
            "a different document under a spent id must not be executed: {}",
            payload_of(&out)
        );
        assert!(
            vti_rooms::storage::get_room(&state.rooms_ks, &g.room.room_id)
                .await
                .is_err(),
            "and it must leave no room behind"
        );
    }

    /// The gap this closes: before the policy gate, anyone who could reach the
    /// endpoint could register a room here in their own name. A stranger signs a
    /// perfectly valid registration and is refused by the community, not by the
    /// proof — and no room row is left behind.
    #[tokio::test]
    async fn a_stranger_cannot_register_a_room_on_this_community() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;

        // The defaults, but no member row for the creator.
        crate::policy::default::install_defaults(&state.policies_ks, &state.active_policies_ks)
            .await
            .expect("install defaults");

        let out = handle_create(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_CREATE_TYPE,
                json!({
                    "roomId": f.room.room_id,
                    "visibility": f.room.visibility,
                    "ownerDid": f.room.owner_did,
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;

        assert!(
            !out.status.is_success(),
            "a non-member must not register a room here: {}",
            payload_of(&out)
        );
        assert!(
            String::from_utf8_lossy(&out.body).contains("not-a-member"),
            "the refusal carries the policy's own code: {}",
            payload_of(&out)
        );
        assert!(
            vti_rooms::storage::get_room(&state.rooms_ks, &f.room.room_id)
                .await
                .is_err(),
            "a refused registration must not leave the identifier taken"
        );
    }

    /// The community's own posture, over the wire: a member may create the tiers
    /// this community serves, and `private` is refused until an operator says
    /// otherwise (§7.4).
    #[tokio::test]
    async fn a_member_is_refused_a_private_room_until_the_community_enables_it() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Private).await;

        crate::policy::default::install_defaults(&state.policies_ks, &state.active_policies_ks)
            .await
            .expect("install defaults");
        crate::members::storage::store_member(
            &state.members_ks,
            &crate::members::Member::fresh(f.room.owner_did.clone()),
        )
        .await
        .expect("seed the member row");

        let out = handle_create(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_CREATE_TYPE,
                json!({
                    "roomId": f.room.room_id,
                    "visibility": f.room.visibility,
                    "ownerDid": f.room.owner_did,
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(
            String::from_utf8_lossy(&out.body).contains("private-tier-not-enabled"),
            "a member is still refused the tier the community has not enabled: {}",
            payload_of(&out)
        );

        // …and enabling it is a policy upload, not a code change.
        permit_private(state).await;
        assert!(
            create(state, &f).await.status.is_success(),
            "once the community permits private rooms, the same request succeeds"
        );
    }

    /// The community half of the same gate. Registration is the one verb no chain can
    /// authorize — the room has issued nothing yet — so the request's own proof is what
    /// makes `ownerDid` a fact rather than a field anyone can fill with anyone.
    #[tokio::test]
    async fn a_room_cannot_be_registered_in_somebody_elses_name() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        let mallory = vti_rooms_dtg::test_support::Party::new();

        let out = handle_create(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_CREATE_TYPE,
                json!({
                    "roomId": f.room.room_id,
                    "visibility": f.room.visibility,
                    "ownerDid": f.room.owner_did,
                }),
                &mallory.did,
                &mallory.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(
            !out.status.is_success(),
            "a signer who is not the named owner must be refused: {}",
            payload_of(&out)
        );
        assert!(
            vti_rooms::storage::get_room(&state.rooms_ks, &f.room.room_id)
                .await
                .is_err(),
            "a refused registration must not leave the identifier taken"
        );

        // The owner's own registration still succeeds — a gate, not a wall.
        assert!(create(state, &f).await.status.is_success());
    }

    /// `rooms/revocations/put` over this host's production path: the room-signed list is
    /// accepted with its head, the member it withdraws is refused from then on — including
    /// under `rooms/records/put` 0.1, a version published before revocation existed
    /// (*Enforcement* item 2) — and `get` serves the list back as the room signed it.
    #[tokio::test]
    async fn a_delivered_list_is_enforced_on_every_version_and_served_back() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        assert!(create(state, &f).await.status.is_success());

        let list = f
            .revocation_list(1, vec![RoomFixture::withdraw_subject(&f.successor.did)])
            .await;
        let call =
            |uri: &'static str, payload: Value, signer: &vti_rooms_dtg::test_support::Party| {
                let did = signer.did.clone();
                let key = signer.secret_multibase.clone();
                async move {
                    let d = doc(state, uri, payload, &did, &key).await;
                    let presenter = spine_presenter(state, &d).await;
                    super::dispatch(state, d, &presenter).await
                }
            };
        let out = call(
            vti_rooms::revocation_tasks::PUT_TYPE,
            json!({ "roomId": f.room.room_id, "presentation": f.as_owner(), "list": list }),
            &f.owner,
        )
        .await;
        assert!(out.status.is_success(), "put: {}", payload_of(&out));
        assert_eq!(payload_of(&out)["head"]["sequence"], 1);
        assert!(super::serves(vti_rooms::revocation_tasks::PUT_TYPE));

        let out = call(
            vti_rooms::wire::ROOMS_RECORDS_GET_TYPE,
            json!({ "roomId": f.room.room_id, "key": "k1", "presentation": f.as_successor() }),
            &f.successor,
        )
        .await;
        assert_eq!(out.status.as_u16(), 403, "{}", payload_of(&out));
        assert_eq!(payload_of(&out)["code"], "permissionDenied");

        let out = call(
            vti_rooms::revocation_tasks::GET_TYPE,
            json!({ "roomId": f.room.room_id, "presentation": f.as_owner() }),
            &f.owner,
        )
        .await;
        assert!(out.status.is_success(), "get: {}", payload_of(&out));
        assert_eq!(payload_of(&out)["list"], list, "served back unaltered");
    }

    #[tokio::test]
    async fn a_room_is_created_and_a_record_round_trips() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        let out = create(state, &f).await;
        assert!(out.status.is_success(), "create: {}", payload_of(&out));

        let out = handle_put_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE,
                json!({
                    "roomId": f.room.room_id, "key": "k1", "presentation": f.as_owner(),
                    "cleartext": { "body": "a decision" }
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(out.status.is_success(), "put: {}", payload_of(&out));
        assert_eq!(payload_of(&out)["version"], 1);

        let out = handle_get_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_GET_TYPE,
                json!({ "roomId": f.room.room_id, "key": "k1", "presentation": f.as_owner() }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(out.status.is_success());
        let got = payload_of(&out);
        assert_eq!(got["cleartext"]["body"], "a decision");
        assert_eq!(
            got["author"], f.owner.did,
            "the author is the subject the chain established, not the room's owner field"
        );
    }

    /// A record sealed under an epoch the room has moved past is the declared
    /// `epochMismatch`, the code a writer's client turns into "apply the commit you
    /// missed" — never a bare validation failure only its message explains.
    #[tokio::test]
    async fn a_record_sealed_under_a_superseded_epoch_is_epoch_mismatch() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Attributed).await;
        let out = create(state, &f).await;
        assert!(out.status.is_success(), "create: {}", payload_of(&out));
        let current = storage::get_room(&state.rooms_ks, &f.room.room_id)
            .await
            .unwrap()
            .epoch;

        let out = handle_put_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE,
                json!({
                    "roomId": f.room.room_id, "key": "k1", "presentation": f.as_owner(),
                    "sealed": {
                        "ciphertext": "9jK2_QhV1sVvR0m5xAqZ7A",
                        "nonce": "b0Zt8Qm2Yq1sVvR0",
                        "epoch": current + 1
                    }
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(!out.status.is_success(), "{}", payload_of(&out));
        assert_eq!(
            payload_of(&out)["code"],
            "rooms/records/put:epochMismatch",
            "{}",
            payload_of(&out)
        );
    }

    /// The invariant the whole family rests on: authorization is the chain, and a request
    /// without one is refused regardless of who is asking.
    #[tokio::test]
    async fn an_operation_with_no_authority_chain_is_refused() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        create(state, &f).await;

        let out = handle_get_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_GET_TYPE,
                json!({
                    "roomId": f.room.room_id, "key": "k1",
                    "presentation": { "membership": f.membership, "authority": [] }
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(!out.status.is_success());
    }

    /// The agent case, through the VTC rather than a standalone host — the two must reach
    /// the same conclusion, because both go through the same `vti-rooms-dtg`.
    #[tokio::test]
    async fn an_agent_reads_under_a_narrower_chain_and_cannot_write() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        create(state, &f).await;

        handle_put_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE,
                json!({
                    "roomId": f.room.room_id, "key": "k", "presentation": f.as_owner(),
                    "cleartext": { "body": "for the agent" }
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;

        let out = handle_get_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_GET_TYPE,
                json!({ "roomId": f.room.room_id, "key": "k", "presentation": f.as_agent() }),
                &f.agent.did,
                &f.agent.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(
            out.status.is_success(),
            "the agent reads: {}",
            payload_of(&out)
        );

        let out = handle_put_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE,
                json!({
                    "roomId": f.room.room_id, "key": "k2", "presentation": f.as_agent(),
                    "cleartext": { "body": "but must not write" }
                }),
                &f.agent.did,
                &f.agent.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(!out.status.is_success(), "a read-only chain must not write");
    }

    /// A private room is refused for want of a zero-knowledge profile — and the refusal is
    /// the same one a standalone room host gives, because it comes from the shared crate.
    #[tokio::test]
    async fn a_private_room_is_refused_for_want_of_a_zk_profile() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Private).await;
        create(state, &f).await;

        let mut p = f.as_owner();
        p.subject_binding = Some("a-binding-nobody-can-check".into());

        let out = handle_get_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_GET_TYPE,
                json!({ "roomId": f.room.room_id, "key": "k", "presentation": p }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(!out.status.is_success());
        assert!(
            payload_of(&out)["message"]
                .as_str()
                .unwrap_or_default()
                .contains("zero-knowledge profile"),
            "{}",
            payload_of(&out)
        );
    }

    /// A private room refuses a presentation with no binding at all before any credential
    /// is parsed — the shape check, which is `vti-rooms`' half.
    #[tokio::test]
    async fn a_private_room_refuses_a_presentation_without_a_subject_binding() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Private).await;
        create(state, &f).await;

        let out = handle_get_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_GET_TYPE,
                json!({ "roomId": f.room.room_id, "key": "k", "presentation": f.as_owner() }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(!out.status.is_success());
        assert!(
            payload_of(&out)["message"]
                .as_str()
                .unwrap_or_default()
                .contains("subject binding"),
            "{}",
            payload_of(&out)
        );
    }

    /// Every room operation leaves a trail. §8: on shared material, reads are the
    /// interesting event — a write log says what a room holds, a read log says who has
    /// seen it.
    #[tokio::test]
    async fn room_operations_are_audited() {
        let tv = crate::test_support::TestVtc::builder()
            .with_audit(true)
            .build()
            .await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        create(state, &f).await;

        handle_put_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE,
                json!({
                    "roomId": f.room.room_id, "key": "k", "presentation": f.as_owner(),
                    "cleartext": { "body": "x" }
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;

        handle_get_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_GET_TYPE,
                json!({ "roomId": f.room.room_id, "key": "k", "presentation": f.as_owner() }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;

        let rows = audit_rows(state).await;
        assert!(
            rows.contains("room.records.put") && rows.contains("room.records.get"),
            "both the write and the read must be recorded: {rows}"
        );
        assert!(
            rows.contains(&f.owner.did),
            "an open room discloses the actor: {rows}"
        );
    }

    /// The property this whole slice exists to protect. An audit log naming every actor on
    /// a private room *is* the membership the service was built never to learn — assembled
    /// one entry at a time, with nothing breaking and no test going red.
    #[tokio::test]
    async fn a_private_rooms_audit_trail_never_names_the_actor() {
        let tv = crate::test_support::TestVtc::builder()
            .with_audit(true)
            .build()
            .await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Private).await;
        create(state, &f).await;

        // A private room is refused for want of a ZK profile, so drive the audit path
        // directly with an authorization the verifier produced — the point under test is
        // what gets *recorded*, not what gets served.
        let room = storage::get_room(&state.rooms_ks, &f.room.room_id)
            .await
            .expect("room");
        let mut presentation = f.as_owner();
        presentation.subject_binding = Some("a-binding".into());
        let authorized = authz::authorize(
            &room,
            &presentation,
            Action::Read,
            &f.owner.did,
            now(),
            &AlwaysVouches,
            &state.room_revocations_ks,
        )
        .await
        .expect("authorized");

        audit_room(
            state,
            &room,
            &authorized,
            RoomOperation::GetRecord,
            Some("opaque-key"),
        )
        .await;

        let rows = audit_rows(state).await;
        assert!(
            rows.contains("room.records.get"),
            "the operation is recorded: {rows}"
        );
        assert!(
            rows.contains("opaque-key"),
            "the record is recorded — an opaque key identifies without describing: {rows}"
        );
        assert!(
            !rows.contains(&f.owner.did),
            "but the actor must appear nowhere in the audit trail: {rows}"
        );
    }

    /// Stands in for the chain verifier so the audit path can be reached on a tier the
    /// service otherwise refuses to serve.
    struct AlwaysVouches;

    #[async_trait::async_trait]
    impl vti_rooms::authz::ChainVerifier for AlwaysVouches {
        async fn verify(
            &self,
            _: &vti_rooms::Room,
            _: &vti_rooms::wire::AuthorityPresentation,
            _: Action,
            presenter: &str,
        ) -> Result<vti_rooms::authz::VerifiedChain, vti_common::error::AppError> {
            Ok(vti_rooms::authz::VerifiedChain {
                subject: presenter.to_string(),
                actions: vec!["read".into()],
                member: Some(presenter.to_string()),
                presented: Default::default(),
            })
        }
    }

    /// Every audit row, as one string to assert over.
    async fn audit_rows(state: &AppState) -> String {
        let raw = state
            .audit_ks
            .prefix_iter_raw(Vec::<u8>::new())
            .await
            .expect("read the audit keyspace");
        raw.into_iter()
            .map(|(_, v)| String::from_utf8_lossy(&v).into_owned())
            .collect::<Vec<_>>()
            .join("\n")
    }

    /// The grant separation, end to end. The fixture's agent chain confers `read` alone,
    /// so an agent that can read a room cannot demote what a person wrote in it.
    #[tokio::test]
    async fn curating_needs_its_own_grant_and_is_not_implied_by_write() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        create(state, &f).await;

        handle_put_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE,
                json!({
                    "roomId": f.room.room_id, "key": "k", "presentation": f.as_owner(),
                    "cleartext": { "body": "a decision" }
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;

        let out = handle_curate_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_CURATE_TYPE,
                json!({
                    "roomId": f.room.room_id, "key": "k",
                    "presentation": f.as_agent(), "status": "deprecated"
                }),
                &f.agent.did,
                &f.agent.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(
            !out.status.is_success(),
            "a read-only agent must not curate: {}",
            payload_of(&out)
        );

        // The owner's chain confers curate.
        let out = handle_curate_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_CURATE_TYPE,
                json!({
                    "roomId": f.room.room_id, "key": "k",
                    "presentation": f.as_owner(), "status": "deprecated", "pinned": true
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(out.status.is_success(), "{}", payload_of(&out));
        let body = payload_of(&out);
        assert_eq!(body["status"], "deprecated");
        assert_eq!(body["pinned"], true, "pinned and deprecated at once");
        assert_eq!(body["version"], 2, "curation assigns a new version");
    }

    /// A retraction drops the body and keeps the tombstone, and the tombstone is what a
    /// caller synchronising on `sinceVersion` needs in order to converge.
    #[tokio::test]
    async fn a_retraction_leaves_a_tombstone_a_listing_still_reports() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        create(state, &f).await;

        handle_put_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE,
                json!({
                    "roomId": f.room.room_id, "key": "k", "presentation": f.as_owner(),
                    "cleartext": { "body": "secret-body-text" }
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;

        handle_curate_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_CURATE_TYPE,
                json!({
                    "roomId": f.room.room_id, "key": "k",
                    "presentation": f.as_owner(), "status": "retracted"
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;

        let out = handle_list_records(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_LIST_TYPE,
                json!({ "roomId": f.room.room_id, "presentation": f.as_owner() }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        let text = payload_of(&out).to_string();
        assert!(
            text.contains("retracted"),
            "the tombstone must still be listed, or sync cannot converge: {text}"
        );

        let out = handle_get_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_GET_TYPE,
                json!({ "roomId": f.room.room_id, "key": "k", "presentation": f.as_owner() }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(
            !payload_of(&out).to_string().contains("secret-body-text"),
            "but the body is gone: {}",
            payload_of(&out)
        );
    }

    #[tokio::test]
    async fn listing_returns_metadata_and_never_bodies() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        create(state, &f).await;

        handle_put_record(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE,
                json!({
                    "roomId": f.room.room_id, "key": "a", "presentation": f.as_owner(),
                    "cleartext": { "body": "secret-body-text" }
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;

        let out = handle_list_records(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_RECORDS_LIST_TYPE,
                json!({ "roomId": f.room.room_id, "presentation": f.as_owner() }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(out.status.is_success());
        let text = payload_of(&out).to_string();
        assert!(text.contains("\"key\""));
        assert!(
            !text.contains("secret-body-text"),
            "a listing must never carry bodies: {text}"
        );
    }

    /// `admin`, not `write`: if any key-holder could mint an epoch, any member could evict
    /// any other, and the service — which cannot see the membership — would never know.
    #[tokio::test]
    async fn minting_an_epoch_requires_admin_and_advances_by_one() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        create(state, &f).await;

        // The agent's chain confers `read` alone.
        let out = handle_mint_epoch(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_EPOCH_MINT_TYPE,
                json!({ "roomId": f.room.room_id, "epoch": 2, "presentation": f.as_agent() }),
                &f.agent.did,
                &f.agent.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(!out.status.is_success(), "read may not mint an epoch");

        // The owner's confers `admin`.
        let out = handle_mint_epoch(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_EPOCH_MINT_TYPE,
                json!({ "roomId": f.room.room_id, "epoch": 2, "presentation": f.as_owner() }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(out.status.is_success(), "{}", payload_of(&out));
        assert_eq!(payload_of(&out)["epoch"], 2);

        // And it advances by exactly one — skipping would seal records under an epoch no
        // member was ever given a key for.
        let out = handle_mint_epoch(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_EPOCH_MINT_TYPE,
                json!({ "roomId": f.room.room_id, "epoch": 5, "presentation": f.as_owner() }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(!out.status.is_success(), "an epoch may not skip");
    }

    /// Register the fixture's room with a chosen epoch expiry, bypassing the handler.
    ///
    /// Succession is entirely about the passage of time; writing the room in the state a
    /// year would produce is the only way to test it without waiting one.
    async fn create_expiring(state: &AppState, f: &RoomFixture, epoch_expires_at: Option<u64>) {
        storage::create_room(
            &state.rooms_ks,
            &Room {
                epoch_expires_at,
                ..f.room.clone()
            },
        )
        .await
        .expect("register the room");
    }

    fn days_ago(n: u64) -> Option<u64> {
        Some(now() - n * 24 * 60 * 60)
    }

    async fn claim(state: &AppState, f: &RoomFixture, payload: Value) -> TrustTaskOutcome {
        handle_claim_owner(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_OWNER_CLAIM_TYPE,
                payload,
                &f.successor.did,
                &f.successor.secret_multibase,
            )
            .await,
        )
        .await
    }

    /// `admin`, and the audit entry names the operation rather than the authority.
    #[tokio::test]
    async fn transferring_requires_admin_and_names_the_new_owner() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        create(state, &f).await;

        // `read` is not enough; the agent's chain is the narrower one a member delegates.
        let out = handle_transfer_owner(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_OWNER_TRANSFER_TYPE,
                json!({
                    "roomId": f.room.room_id,
                    "newOwnerDid": f.agent.did,
                    "presentation": f.as_agent(),
                }),
                &f.agent.did,
                &f.agent.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(
            !out.status.is_success(),
            "an agent may not hand away the room"
        );

        let out = handle_transfer_owner(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_OWNER_TRANSFER_TYPE,
                json!({
                    "roomId": f.room.room_id,
                    "newOwnerDid": f.successor.did,
                    "presentation": f.as_owner(),
                    "reason": "stepping back",
                }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(out.status.is_success(), "{}", payload_of(&out));
        assert_eq!(payload_of(&out)["ownerDid"], f.successor.did);

        let stored = storage::get_room(&state.rooms_ks, &f.room.room_id)
            .await
            .unwrap();
        assert_eq!(stored.owner_did, f.successor.did, "and it persisted");
    }

    /// The whole point of the pair: a room outlives one person's availability.
    #[tokio::test]
    async fn a_nominated_successor_claims_a_dormant_room() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        create_expiring(state, &f, days_ago(60)).await;

        let out = claim(
            state,
            &f,
            json!({
                "roomId": f.room.room_id,
                "nomination": f.nominate(&f.successor.did, 24).await,
                "presentation": f.as_successor(),
                "reason": "unreachable since March",
            }),
        )
        .await;
        assert!(out.status.is_success(), "{}", payload_of(&out));
        assert_eq!(payload_of(&out)["ownerDid"], f.successor.did);

        // A claim hands over a dormant room; it does not revive one. The new owner's first
        // act should be the one that proves they can perform it.
        let stored = storage::get_room(&state.rooms_ks, &f.room.room_id)
            .await
            .unwrap();
        assert!(!stored.lifecycle(now()).accepts_writes());
    }

    /// The defence, and it is the same act as ordinary use: an owner who was merely away
    /// renews, and every pending claim stops working. Nothing has to be revoked.
    #[tokio::test]
    async fn renewing_defeats_a_pending_claim() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        create_expiring(state, &f, days_ago(60)).await;

        let nomination = f.nominate(&f.successor.did, 24).await;

        // The owner returns and mints an epoch — ordinary use, nothing succession-specific.
        let out = handle_mint_epoch(
            state,
            doc(
                state,
                vti_rooms::wire::ROOMS_EPOCH_MINT_TYPE,
                json!({ "roomId": f.room.room_id, "epoch": 2, "presentation": f.as_owner() }),
                &f.owner.did,
                &f.owner.secret_multibase,
            )
            .await,
        )
        .await;
        assert!(out.status.is_success(), "{}", payload_of(&out));

        let out = claim(
            state,
            &f,
            json!({
                "roomId": f.room.room_id,
                "nomination": nomination,
                "presentation": f.as_successor(),
            }),
        )
        .await;
        assert!(
            !out.status.is_success(),
            "the same nomination that would have worked a moment ago: {}",
            payload_of(&out)
        );

        let stored = storage::get_room(&state.rooms_ks, &f.room.room_id)
            .await
            .unwrap();
        assert_eq!(stored.owner_did, f.owner.did, "and the owner kept the room");
    }

    /// Dormancy alone confers nothing, or any member of any quiet room could take it.
    #[tokio::test]
    async fn a_member_of_a_dormant_room_cannot_claim_without_being_named() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        create_expiring(state, &f, days_ago(60)).await;

        // A genuine nomination the room issued — naming somebody else.
        let out = claim(
            state,
            &f,
            json!({
                "roomId": f.room.room_id,
                "nomination": f.nominate(&f.agent.did, 24).await,
                "presentation": f.as_successor(),
            }),
        )
        .await;
        assert!(
            !out.status.is_success(),
            "a nomination is bound to the party it names: {}",
            payload_of(&out)
        );
    }

    /// The order the handler checks in, which is the difference between an answer a
    /// claimant can act on and one that sends them back in a month for the real one.
    #[tokio::test]
    async fn a_bad_nomination_is_reported_before_the_room_is_still_live() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        create_expiring(state, &f, Some(now() + 30 * 24 * 60 * 60)).await;

        let out = claim(
            state,
            &f,
            json!({
                "roomId": f.room.room_id,
                "nomination": f.nominate(&f.agent.did, 24).await,
                "presentation": f.as_successor(),
            }),
        )
        .await;
        assert!(!out.status.is_success());
        let body = payload_of(&out).to_string();
        assert!(
            !body.contains("dormant"),
            "the nomination was never valid; saying `still live` hides that: {body}"
        );
    }

    // ── Room files (data-rooms-files.md §5–§6) ────────────────────────────────

    /// Dispatch any room document the way the spine does: verified signer, then the
    /// registered dispatcher.
    async fn send(
        state: &AppState,
        uri: &str,
        payload: Value,
        f: &RoomFixture,
    ) -> TrustTaskOutcome {
        let d = doc(state, uri, payload, &f.owner.did, &f.owner.secret_multibase).await;
        let presenter = spine_presenter(state, &d).await;
        super::dispatch(state, d, &presenter).await
    }

    /// A blob's ciphertext as an uploader holds it: its chunks and its manifest.
    fn blob(len: usize, seed: u8) -> (Vec<Vec<u8>>, vti_rooms::files::BlobManifest) {
        const CHUNK: usize = 16_384;
        let bytes: Vec<u8> = (0..len)
            .map(|i| (i as u8).wrapping_mul(31).wrapping_add(seed))
            .collect();
        let mut builder = vti_rooms::files::ManifestBuilder::new(CHUNK as u64);
        let chunks: Vec<Vec<u8>> = bytes.chunks(CHUNK).map(<[u8]>::to_vec).collect();
        for c in &chunks {
            builder.push(c);
        }
        (chunks, builder.finish().expect("a manifest"))
    }

    /// Upload and commit a blob into the fixture's room; its BlobRef.
    async fn upload(state: &AppState, f: &RoomFixture, seed: u8) -> String {
        use base64::Engine as _;
        let (chunks, manifest) = blob(40_000, seed);
        let out = send(
            state,
            vti_rooms::wire::ROOMS_BLOBS_UPLOAD_BEGIN_TYPE,
            json!({ "roomId": f.room.room_id, "presentation": f.as_owner(), "manifest": manifest }),
            f,
        )
        .await;
        assert!(out.status.is_success(), "begin: {}", payload_of(&out));
        let begun = payload_of(&out);
        let upload_id = begun["uploadId"].as_str().unwrap().to_string();
        assert_eq!(begun["missing"].as_array().unwrap().len(), chunks.len());
        for (i, c) in chunks.iter().enumerate() {
            let out = send(
                state,
                vti_rooms::wire::ROOMS_BLOBS_UPLOAD_CHUNK_TYPE,
                json!({
                    "uploadId": upload_id, "index": i,
                    "digestMultibase": vti_rooms::files::digest_multibase(c),
                    "data": base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(c),
                }),
                f,
            )
            .await;
            assert!(out.status.is_success(), "chunk {i}: {}", payload_of(&out));
        }
        let out = send(
            state,
            vti_rooms::wire::ROOMS_BLOBS_UPLOAD_COMMIT_TYPE,
            json!({ "uploadId": upload_id }),
            f,
        )
        .await;
        assert!(out.status.is_success(), "commit: {}", payload_of(&out));
        let blob_ref = payload_of(&out)["blobRef"].as_str().unwrap().to_string();
        assert_eq!(blob_ref, manifest.blob_ref().unwrap());
        blob_ref
    }

    async fn room_files(state: &AppState, f: &RoomFixture) -> u64 {
        let out = send(
            state,
            vti_rooms::wire::ROOMS_INFO_TYPE,
            json!({ "roomId": f.room.room_id, "presentation": f.as_owner() }),
            f,
        )
        .await;
        assert!(out.status.is_success(), "info: {}", payload_of(&out));
        payload_of(&out)["usage"]["room"]["files"].as_u64().unwrap()
    }

    async fn put_v2(
        state: &AppState,
        f: &RoomFixture,
        key: &str,
        blobs: &[&str],
    ) -> TrustTaskOutcome {
        send(
            state,
            vti_rooms::wire::ROOMS_RECORDS_PUT_V2_TYPE,
            json!({
                "roomId": f.room.room_id, "key": key, "presentation": f.as_owner(),
                "cleartext": { "body": "a file" }, "blobs": blobs,
            }),
            f,
        )
        .await
    }

    /// The whole path a member's client takes against the VTC: upload, a record naming the
    /// blob, a 0.2 read naming it back, the download, and the usage it is charged.
    #[tokio::test]
    async fn a_file_is_uploaded_named_read_back_and_counted() {
        use base64::Engine as _;
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        assert!(create(state, &f).await.status.is_success());

        let blob_ref = upload(state, &f, 1).await;
        let out = put_v2(state, &f, "doc", &[&blob_ref]).await;
        assert!(out.status.is_success(), "put 0.2: {}", payload_of(&out));

        let out = send(
            state,
            vti_rooms::wire::ROOMS_RECORDS_GET_V2_TYPE,
            json!({ "roomId": f.room.room_id, "key": "doc", "presentation": f.as_owner() }),
            &f,
        )
        .await;
        assert!(out.status.is_success(), "get 0.2: {}", payload_of(&out));
        assert_eq!(payload_of(&out)["blobs"], json!([blob_ref]));

        let out = send(
            state,
            vti_rooms::wire::ROOMS_RECORDS_LIST_V2_TYPE,
            json!({ "roomId": f.room.room_id, "presentation": f.as_owner() }),
            &f,
        )
        .await;
        assert!(out.status.is_success(), "list 0.2: {}", payload_of(&out));
        assert_eq!(payload_of(&out)["records"][0]["blobs"], json!([blob_ref]));

        let out = send(
            state,
            vti_rooms::wire::ROOMS_BLOBS_GET_TYPE,
            json!({ "roomId": f.room.room_id, "blobRef": blob_ref, "presentation": f.as_owner() }),
            &f,
        )
        .await;
        assert!(out.status.is_success(), "blob get: {}", payload_of(&out));
        let download_id = payload_of(&out)["downloadId"].as_str().unwrap().to_string();
        let (chunks, _) = blob(40_000, 1);
        for (i, expected) in chunks.iter().enumerate() {
            let out = send(
                state,
                vti_rooms::wire::ROOMS_BLOBS_CHUNK_TYPE,
                json!({ "downloadId": download_id, "index": i }),
                &f,
            )
            .await;
            assert!(
                out.status.is_success(),
                "download {i}: {}",
                payload_of(&out)
            );
            let data = base64::engine::general_purpose::URL_SAFE_NO_PAD
                .decode(payload_of(&out)["data"].as_str().unwrap())
                .unwrap();
            assert_eq!(&data, expected);
        }

        assert_eq!(room_files(state, &f).await, 1);
    }

    /// A record can only name a blob committed in its own room.
    #[tokio::test]
    async fn a_record_naming_an_uncommitted_blob_is_refused_with_the_declared_code() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        assert!(create(state, &f).await.status.is_success());

        let (_, never_uploaded) = blob(20_000, 9);
        let out = put_v2(state, &f, "doc", &[&never_uploaded.blob_ref().unwrap()]).await;
        assert!(!out.status.is_success());
        assert_eq!(
            payload_of(&out)["code"],
            "rooms/records/put:blobNotFound",
            "{}",
            payload_of(&out)
        );
    }

    /// A retraction keeps no body, so it keeps no file; a 0.1 rewrite names no blobs, so
    /// it releases them too. Either way the room stops being charged at once.
    #[tokio::test]
    async fn retracting_or_rewriting_a_record_releases_its_file() {
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        assert!(create(state, &f).await.status.is_success());

        let first = upload(state, &f, 1).await;
        assert!(put_v2(state, &f, "a", &[&first]).await.status.is_success());
        let second = upload(state, &f, 2).await;
        assert!(put_v2(state, &f, "b", &[&second]).await.status.is_success());
        assert_eq!(room_files(state, &f).await, 2);

        let out = send(
            state,
            vti_rooms::wire::ROOMS_RECORDS_CURATE_TYPE,
            json!({
                "roomId": f.room.room_id, "key": "a", "presentation": f.as_owner(),
                "status": "retracted",
            }),
            &f,
        )
        .await;
        assert!(out.status.is_success(), "retract: {}", payload_of(&out));
        assert_eq!(
            room_files(state, &f).await,
            1,
            "the retraction released its file"
        );

        let out = send(
            state,
            vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE,
            json!({
                "roomId": f.room.room_id, "key": "b", "presentation": f.as_owner(),
                "cleartext": { "body": "no file any more" },
            }),
            &f,
        )
        .await;
        assert!(out.status.is_success(), "0.1 rewrite: {}", payload_of(&out));
        assert_eq!(
            room_files(state, &f).await,
            0,
            "a 0.1 rewrite released its file"
        );
    }

    /// An upload belongs to the party that opened it. Anyone else sending to it is told
    /// there is no such upload, which says nothing about whether one exists.
    #[tokio::test]
    async fn only_the_uploader_can_send_chunks_to_an_upload() {
        use base64::Engine as _;
        let tv = build_test_vtc().await;
        let state = &tv.state;
        let f = RoomFixture::new(Visibility::Open).await;
        assert!(create(state, &f).await.status.is_success());

        let (chunks, manifest) = blob(20_000, 3);
        let out = send(
            state,
            vti_rooms::wire::ROOMS_BLOBS_UPLOAD_BEGIN_TYPE,
            json!({ "roomId": f.room.room_id, "presentation": f.as_owner(), "manifest": manifest }),
            &f,
        )
        .await;
        let upload_id = payload_of(&out)["uploadId"].as_str().unwrap().to_string();

        let d = doc(
            state,
            vti_rooms::wire::ROOMS_BLOBS_UPLOAD_CHUNK_TYPE,
            json!({
                "uploadId": upload_id, "index": 0,
                "digestMultibase": vti_rooms::files::digest_multibase(&chunks[0]),
                "data": base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(&chunks[0]),
            }),
            &f.agent.did,
            &f.agent.secret_multibase,
        )
        .await;
        let presenter = spine_presenter(state, &d).await;
        let out = super::dispatch(state, d, &presenter).await;
        assert_eq!(
            payload_of(&out)["code"],
            "rooms/blobs/upload/chunk:notFound",
            "{}",
            payload_of(&out)
        );
    }
}