vtc-service 0.12.0

//! The `git-ns/*` operations, with no transport in them.
//!
//! Each function is one task's *Request* section, in the order the
//! specification lists its steps: resolve what the request names, check the
//! entitlement, apply the fixed rules ([`super::rules`]), the consent-class
//! gate ([`consent_gate`]), then the community's policy ([`super::policy`]),
//! and only then write. A write is durable before the function returns — the
//! grant and revoke specifications forbid answering first — and each accepted
//! change leaves a `GitNsOperation` audit row, which is what wakes the
//! projector.
//!
//! The actor is always a DID the caller has already established: the signer
//! of the document's proof, verified by the dispatch spine against the
//! document as received. Nothing here reads an identity from a payload.

use chrono::{DateTime, Timelike, Utc};
use serde_json::json;
use trust_tasks_rs::specs::git_ns::account::{
    link::v0_1 as link, link_status::v0_1 as link_status, unlink::v0_1 as unlink,
};
use trust_tasks_rs::specs::git_ns::bridge::job::v0_4 as job_wire;
use trust_tasks_rs::specs::git_ns::namespace::{
    bind::v0_1 as bind, reseat::v0_3 as reseat, unbind::v0_1 as unbind,
};
use trust_tasks_rs::specs::git_ns::repo::{
    adopt::v0_1 as adopt, archive::v0_1 as archive, create::v0_3 as create,
    transfer::v0_1 as transfer,
};
use trust_tasks_rs::specs::git_ns::right::{
    break_glass::v0_1 as break_glass, grant::v0_3 as grant, ratify::v0_1 as ratify,
    revoke::v0_3 as revoke,
};
use vti_common::audit::{AuditEvent, GitNsOperationData};
use vti_common::error::AppError;

use crate::server::AppState;

use super::bridge::{self, JobKind, NewJob};
use super::model::{
    Bootstrap, LinkAttempt, LinkState, Mode, Namespace, NamespaceState, OwnerKind, Repo, RepoState,
    Resource, Right, RightRow, Scope, SyncState, SyncStatus, Visibility, new_id,
};
use super::policy::{self, Capabilities, GitNsFacts, Party, VerifiedGitNsFacts};
use super::rules::{self, Refusal};
use super::store::{self, Snapshot};
use super::wire;

// ── the codes this family declares ──────────────────────────────────────────

pub const UNKNOWN_NAMESPACE: &str = grant::error_codes::UNKNOWN_NAMESPACE.code;
pub const NAMESPACE_NOT_BOUND: &str = grant::error_codes::NAMESPACE_NOT_BOUND.code;
pub const UNKNOWN_REPO: &str = grant::error_codes::UNKNOWN_REPO.code;
pub const REPO_NOT_ACTIVE: &str = grant::error_codes::REPO_NOT_ACTIVE.code;
pub const SCOPE_VIOLATION: &str = grant::error_codes::SCOPE_VIOLATION.code;
pub const ESCALATION: &str = grant::error_codes::ESCALATION.code;
pub const MEMBERS_ONLY: &str = grant::error_codes::MEMBERS_ONLY.code;
pub const POLICY_DENIED: &str = grant::error_codes::POLICY_DENIED.code;
pub const EXPIRY_IN_PAST: &str = grant::error_codes::EXPIRY_IN_PAST.code;
pub const LAST_OWNER: &str = revoke::error_codes::LAST_OWNER.code;
pub const LAST_ADMIN: &str = revoke::error_codes::LAST_ADMIN.code;
pub const NOT_GRANTED: &str = revoke::error_codes::NOT_GRANTED.code;
pub const ALREADY_BOUND: &str = bind::error_codes::ALREADY_BOUND.code;
pub const NO_BRIDGE: &str = bind::error_codes::NO_BRIDGE.code;
pub const NAME_TAKEN: &str = create::error_codes::NAME_TAKEN.code;
pub const ALREADY_MANAGED: &str = adopt::error_codes::ALREADY_MANAGED.code;
pub const NOT_OWNER: &str = transfer::error_codes::NOT_OWNER.code;
pub const SELF_TRANSFER: &str = transfer::error_codes::SELF_TRANSFER.code;
pub const UNSUPPORTED_FORGE: &str = link::error_codes::UNSUPPORTED_FORGE.code;
pub const UNKNOWN_LINK: &str = link_status::error_codes::UNKNOWN_LINK.code;
pub const NOT_LINKED: &str = unlink::error_codes::NOT_LINKED.code;
pub const NOT_HEADLESS: &str = reseat::error_codes::NOT_HEADLESS.code;
pub const SELF_GRANT_NOT_ALLOWED: &str = grant::error_codes::SELF_GRANT_NOT_ALLOWED.code;
pub const BREAK_GLASS_DISABLED: &str = break_glass::error_codes::DISABLED.code;
pub const BREAK_GLASS_NOT_HEADLESS: &str = break_glass::error_codes::NOT_HEADLESS.code;
pub const NOT_BREAK_GLASS: &str = ratify::error_codes::NOT_BREAK_GLASS.code;
pub const RECORD_CHANGED: &str = ratify::error_codes::RECORD_CHANGED.code;
pub const SELF_RATIFICATION: &str = ratify::error_codes::SELF_RATIFICATION.code;

/// Why an operation refused, as the wire will carry it.
#[derive(Debug)]
pub enum OpError {
    /// A code the task's specification declares.
    Declared {
        code: &'static str,
        message: String,
    },
    /// The framework's `permissionDenied`.
    PermissionDenied(String),
    /// The framework's `malformedRequest`.
    Malformed(String),
    /// The framework's `unavailable`: a bridge that must answer in-line did
    /// not. Retryable, and honest about it.
    Unavailable(String),
    /// The framework's `permissionDenied`, with an operation-bound step-up
    /// inline as `details.stepUpRequest` (`crate::acl::bound_step_up`): the
    /// same document, re-sent once the gesture is recorded, succeeds.
    StepUpRequired {
        message: String,
        request: Box<crate::acl::bound_step_up::ApproveRequest>,
    },
    /// The framework's `unsupportedVersion`: this VTC serves the task, but
    /// not this operation at this version (a drift/resolve 0.1 `adopt`).
    UnsupportedVersion(String),
    Internal(AppError),
}

impl From<AppError> for OpError {
    fn from(e: AppError) -> Self {
        OpError::Internal(e)
    }
}

impl std::fmt::Display for OpError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            OpError::Declared { code, message } => write!(f, "{code}: {message}"),
            OpError::PermissionDenied(m) => write!(f, "permissionDenied: {m}"),
            OpError::Malformed(m) => write!(f, "malformedRequest: {m}"),
            OpError::Unavailable(m) => write!(f, "unavailable: {m}"),
            OpError::StepUpRequired { message, .. } => {
                write!(f, "permissionDenied (step-up required): {message}")
            }
            OpError::UnsupportedVersion(m) => write!(f, "unsupportedVersion: {m}"),
            OpError::Internal(e) => write!(f, "internal: {e}"),
        }
    }
}

pub(super) fn declared(code: &'static str, message: impl Into<String>) -> OpError {
    OpError::Declared {
        code,
        message: message.into(),
    }
}

impl From<Refusal> for OpError {
    fn from(r: Refusal) -> Self {
        match r {
            Refusal::PermissionDenied(m) => OpError::PermissionDenied(m),
            Refusal::ScopeViolation(m) => declared(SCOPE_VIOLATION, m),
            Refusal::Escalation(m) => declared(ESCALATION, m),
            Refusal::MembersOnly(m) => declared(MEMBERS_ONLY, m),
            Refusal::SelfGrant(m) => declared(SELF_GRANT_NOT_ALLOWED, m),
        }
    }
}

pub type OpResult<T> = Result<T, OpError>;

/// Now, to the second — the resolution every record and every response
/// carries, so a record read back compares equal to the one written.
pub fn now() -> DateTime<Utc> {
    let n = Utc::now();
    n.with_nanosecond(0).unwrap_or(n)
}

// ── who is asking ───────────────────────────────────────────────────────────

/// What this community knows about a DID.
#[derive(Debug, Clone, Default)]
pub struct Standing {
    pub did: String,
    /// A current member: an unexpired ACL entry that is not an `application`
    /// entry, and no departure recorded on the member row. The ACL entry is what membership removal deletes, so it
    /// is the authoritative half; the row's `removed_at` covers a departure
    /// that kept the row.
    pub member: bool,
    pub role: Option<String>,
    /// Holds `git.ns.admin` **unqualified** — every namespace this community
    /// governs (`vtc-admin-roles.md` §4). What binds, reseats a headless
    /// namespace and ratifies; never a namespace administrator's grant
    /// authority, which is a qualified `git.ns.admin` resource grant on the
    /// entry (phase C3, `crate::acl::resource_grant`).
    pub community_admin: bool,
}

pub async fn standing(state: &AppState, did: &str) -> Result<Standing, AppError> {
    let entry = crate::acl::get_acl_entry(&state.acl_ks, did).await?;
    let entry = entry.filter(|e| !e.is_expired(crate::auth::session::now_epoch()));
    let row = crate::members::get_member(&state.members_ks, did).await?;
    let departed = row.as_ref().is_some_and(|m| m.removed_at.is_some());
    // An `application` entry holds git rights for a non-member (the bridge, an
    // external signer); it is never a membership.
    let member = entry.as_ref().is_some_and(|e| !e.is_application()) && !departed;
    let community_admin = member
        && entry
            .as_ref()
            .is_some_and(|e| e.can(crate::acl::Capability::GitNsAdmin, None));
    Ok(Standing {
        did: did.to_string(),
        member,
        role: entry.map(|e| e.role.to_string()),
        community_admin,
    })
}

fn party(s: &Standing, rights: impl IntoIterator<Item = Right>) -> Party {
    Party {
        did: s.did.clone(),
        member: s.member,
        role: s.role.clone(),
        rights: rights.into_iter().map(|r| r.as_str().to_string()).collect(),
    }
}

// ── consent classes (design §6) ─────────────────────────────────────────────

/// How much confirmation an action warrants.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum ConsentClass {
    Normal,
    Elevated,
    Destructive,
}

/// Design §6: `normal` — grant/revoke `commit.sign` and `maintain`, create,
/// link; `elevated` — grant/revoke `own`, transfer, archive, adopt, grant
/// `repo.create`; `destructive` — bind, unbind, grant `ns.admin`, and reseat
/// (a grant of `ns.admin`; its spec declares `sideEffects: destructive`).
/// Reseat does not go through `consent_gate`: it already requires the
/// community-administrator capability, which is all the gate would add.
///
/// The design names the grant of `repo.create` and `ns.admin`; revoking one
/// is classed the same, because taking a namespace-level right away is the
/// same weight of decision as conferring it.
pub fn consent_class(action: &str, right: Option<Right>) -> ConsentClass {
    match (action, right) {
        ("namespace.bind" | "namespace.unbind" | "namespace.reseat", _) => {
            ConsentClass::Destructive
        }
        ("right.grant" | "right.revoke", Some(Right::NsAdmin)) => ConsentClass::Destructive,
        ("right.grant" | "right.revoke", Some(Right::RepoOwn | Right::RepoCreate)) => {
            ConsentClass::Elevated
        }
        ("repo.transfer" | "repo.archive" | "repo.adopt", _) => ConsentClass::Elevated,
        _ => ConsentClass::Normal,
    }
}

/// The stand-in for the step-up this VTC cannot yet ask a member for — see
/// [`super::GitNsConfig::elevated_requires_admin`]. It narrows, never widens:
/// the actor must already be entitled by the rights model to reach here.
pub(super) async fn consent_gate(
    state: &AppState,
    actor: &Standing,
    action: &str,
    right: Option<Right>,
) -> OpResult<()> {
    let class = consent_class(action, right);
    if class == ConsentClass::Normal {
        return Ok(());
    }
    let gated = state.config.read().await.git_ns.elevated_requires_admin;
    if gated && !actor.community_admin {
        let what = match class {
            ConsentClass::Destructive => "destructive",
            _ => "elevated",
        };
        return Err(OpError::PermissionDenied(format!(
            "{action} is a {what} action, which needs a step-up this community cannot yet ask a \
             member for; until it can, only a community administrator may perform it \
             (`[git_ns] elevated_requires_admin`)"
        )));
    }
    Ok(())
}

// ── policy ──────────────────────────────────────────────────────────────────

pub struct PolicyInput<'a> {
    pub action: &'a str,
    pub actor: &'a Standing,
    pub actor_rights: Vec<Right>,
    pub resource: &'a Resource,
    pub right: Option<Right>,
    pub subject: Option<(&'a Standing, Vec<Right>)>,
    pub visibility: Option<Visibility>,
    pub expires_at: Option<DateTime<Utc>>,
    pub namespace: Option<&'a Namespace>,
    /// The fixed rules' admission of this request. Without one there is no
    /// policy evaluation, so there is no way to reach the policy first.
    pub passed: rules::RulesPassed,
}

/// Fixed rule 6: evaluated after the fixed rules, and able only to refuse.
/// Returns the policy's version, for the audit row (VTI-VTC-031).
pub(crate) async fn check_policy(
    state: &AppState,
    input: PolicyInput<'_>,
) -> OpResult<Option<u32>> {
    check_policy_via(state, input, None).await
}

/// [`check_policy`], saying how the request arose (`input.via`).
pub(crate) async fn check_policy_via(
    state: &AppState,
    input: PolicyInput<'_>,
    via: Option<&str>,
) -> OpResult<Option<u32>> {
    let active = policy::load(state).await.map_err(|e| {
        declared(
            POLICY_DENIED,
            e.to_string().trim_start_matches("forbidden: "),
        )
    })?;
    let caps = match input.namespace {
        Some(ns) => Capabilities {
            bridge: ns.mode == Mode::Bridge,
            bot_can_create_repos: can_bot_create(ns),
            kind: ns.kind.map(|k| k.as_str().to_string()),
            bridge_did: ns.bridge_did.clone(),
        },
        None => Capabilities::default(),
    };
    let facts = GitNsFacts {
        now: now(),
        action: input.action.to_string(),
        actor: party(input.actor, input.actor_rights),
        resource: input.resource.to_string(),
        forge: input.resource.forge.clone(),
        right: input.right.map(|r| r.as_str().to_string()),
        subject: input.subject.map(|(s, rights)| party(s, rights)),
        visibility: input.visibility.map(|v| v.as_str().to_string()),
        expires_at: input.expires_at,
        capabilities: caps,
        via: via.map(str::to_string),
    };
    let verified = VerifiedGitNsFacts::after_fixed_rules(facts, input.passed)?;
    policy::decide(&verified, &active.compiled).map_err(|d| declared(POLICY_DENIED, d.message))?;
    Ok(active.version)
}

/// Whether a bot can create repositories in `ns` — a bridge serves it and
/// the owner is an organisation (`git-ns/bridge/job`: `createRepo` is refused
/// `notCapable` on a personal account).
pub fn can_bot_create(ns: &Namespace) -> bool {
    ns.mode == Mode::Bridge && ns.kind == Some(OwnerKind::Organization)
}

// ── audit ───────────────────────────────────────────────────────────────────

pub struct Audit<'a> {
    pub action: &'a str,
    pub namespace: Option<&'a str>,
    pub resource: Option<String>,
    pub right: Option<Right>,
    pub policy_version: Option<u32>,
    pub detail: Option<String>,
}

/// Record one git-namespace change.
///
/// A failed write is logged and swallowed, as `rooms` does: the change is
/// already durable, and refusing it now would tell the caller a write failed
/// that did not. The projector does not depend on this row either — it also
/// reconciles on a timer — so a lost row delays a projection, never loses it.
pub async fn audit(state: &AppState, actor: &str, target: Option<&str>, a: Audit<'_>) {
    let Some(writer) = state.audit_writer.as_ref() else {
        return;
    };
    let data = GitNsOperationData {
        action: a.action.to_string(),
        namespace: a.namespace.map(str::to_string),
        resource: a.resource,
        right: a.right.map(|r| r.as_str().to_string()),
        policy_version: a.policy_version,
        detail: a.detail,
    };
    if let Err(e) = writer
        .write(actor, target, AuditEvent::GitNsOperation(data))
        .await
    {
        tracing::error!(error = %e, action = a.action, "failed to record a git-ns audit entry");
    }
}

// ── lookups ─────────────────────────────────────────────────────────────────

/// Refuse, as the framework's `malformedRequest`, anything that is not a
/// DID-core DID ([`super::model::validate_did_core`]).
pub(super) fn did_core(label: &str, value: &str) -> OpResult<()> {
    super::model::validate_did_core(label, value).map_err(OpError::Malformed)
}

pub(super) fn parse_resource(raw: &str) -> OpResult<Resource> {
    Resource::parse(raw).map_err(OpError::Malformed)
}

/// Step "refuses a resource inside no bound namespace with
/// `git-ns:unknownNamespace`, and one inside a pending namespace with
/// `git-ns:namespaceNotBound`".
pub(super) fn bound_namespace_for<'a>(
    snap: &'a Snapshot,
    resource: &Resource,
) -> OpResult<&'a Namespace> {
    let ns = snap.namespace_containing(resource).ok_or_else(|| {
        declared(
            UNKNOWN_NAMESPACE,
            format!("no namespace bound to this VTC contains {resource}"),
        )
    })?;
    if ns.state != NamespaceState::Bound {
        return Err(declared(
            NAMESPACE_NOT_BOUND,
            format!(
                "{} is still pending: its binding has not completed",
                ns.resource()
            ),
        ));
    }
    Ok(ns)
}

/// Refuse to put a repository at `resource` while rights published for a
/// previous repository there are still being withdrawn (design §9, *Rename
/// attacks*: "the old resource's tuples are deleted before any new repo can
/// be adopted at that name"). `except` is the repository the call is about,
/// whose own tuples do not count against it.
async fn refuse_while_withdrawing(
    state: &AppState,
    snap: &Snapshot,
    resource: &Resource,
    except: Option<&str>,
) -> OpResult<()> {
    if super::projection::withdrawal_pending(state, snap, &resource.to_string(), except).await? {
        return Err(OpError::Unavailable(format!(
            "rights published for an earlier repository at {resource} are still being withdrawn \
             from the Trust Registry; retry once they are gone"
        )));
    }
    Ok(())
}

pub(super) fn namespace_by_id<'a>(snap: &'a Snapshot, id: &str) -> OpResult<&'a Namespace> {
    snap.namespace(id).ok_or_else(|| {
        declared(
            UNKNOWN_NAMESPACE,
            format!("no namespace `{id}` is bound to this VTC"),
        )
    })
}

pub(super) fn repo_at<'a>(snap: &'a Snapshot, resource: &Resource) -> OpResult<&'a Repo> {
    lookup(snap, resource)?.ok_or_else(|| {
        declared(
            UNKNOWN_REPO,
            format!("this VTC records no repository at {resource}"),
        )
    })
}

/// The repository recorded at `resource`; an ambiguous name is refused
/// rather than guessed at.
pub(super) fn lookup<'a>(snap: &'a Snapshot, resource: &Resource) -> OpResult<Option<&'a Repo>> {
    snap.lookup_repo(&resource.to_string())
        .map_err(OpError::Unavailable)
}

/// The scope a resource's rights hang on, if it is recorded.
pub(super) fn scope_for(snap: &Snapshot, resource: &Resource) -> OpResult<Option<Scope>> {
    if resource.is_namespace() {
        Ok(snap
            .namespaces
            .iter()
            .find(|n| n.resource() == *resource)
            .map(|n| Scope::Namespace(n.id.clone())))
    } else {
        Ok(lookup(snap, resource)?.map(|r| Scope::Repo(r.id.clone())))
    }
}

async fn settings(state: &AppState) -> policy::Settings {
    policy::active_settings(state).await
}

fn right_from_wire(s: &str) -> OpResult<Right> {
    Right::parse(s).ok_or_else(|| OpError::Malformed(format!("`{s}` is not a git right")))
}

fn visibility_from_wire(s: &str) -> OpResult<Visibility> {
    match s {
        "public" => Ok(Visibility::Public),
        "private" => Ok(Visibility::Private),
        other => Err(OpError::Malformed(format!("`{other}` is not a visibility"))),
    }
}

fn to_string_json<T: serde::Serialize>(v: &T) -> String {
    serde_json::to_value(v)
        .ok()
        .and_then(|v| v.as_str().map(str::to_string))
        .unwrap_or_default()
}

pub(super) fn new_row(
    subject: &str,
    right: Right,
    granted_by: &str,
    subject_member: bool,
) -> RightRow {
    RightRow {
        subject: subject.to_string(),
        right,
        granted_by: granted_by.to_string(),
        granted_at: now(),
        expires_at: None,
        reason: None,
        subject_was_member: subject_member,
        granter_was_member: true,
        break_glass: None,
        review: None,
        single_admin: None,
    }
}

// ── git-ns/namespace/bind/0.1 ───────────────────────────────────────────────

pub async fn bind(state: &AppState, actor_did: &str, p: bind::Payload) -> OpResult<bind::Response> {
    let actor = standing(state, actor_did).await?;
    // Item 1 — the community-administrator capability. No git right suffices:
    // before binding there are none.
    let passed = rules::bind_admitted(actor.community_admin)?;
    let forge = p.forge.to_string();
    let owner = p.owner.to_string();
    let mode = match to_string_json(&p.mode).as_str() {
        "bridge" => Mode::Bridge,
        "manual" => Mode::Manual,
        other => return Err(OpError::Malformed(format!("mode `{other}` is not served"))),
    };
    let resource = Resource::namespace(&forge, &owner);

    let bridge_did = {
        let _guard = store::write_lock().await;
        let snap = Snapshot::load(&state.git_ns).await?;
        // Item 2.
        if snap
            .namespaces
            .iter()
            .any(|n| n.forge == forge && n.owner == owner)
        {
            return Err(declared(
                ALREADY_BOUND,
                format!("{resource} is already bound, or being bound, to this VTC"),
            ));
        }
        consent_gate(state, &actor, "namespace.bind", None).await?;
        // Item 3.
        let version = check_policy(
            state,
            PolicyInput {
                action: "namespace.bind",
                actor: &actor,
                actor_rights: vec![],
                resource: &resource,
                right: Some(Right::NsAdmin),
                subject: None,
                visibility: None,
                expires_at: None,
                namespace: None,
                passed,
            },
        )
        .await?;

        if mode == Mode::Manual {
            // Item 5 — bound at once, and the binder is its first admin.
            let ns = Namespace {
                id: new_id("ns"),
                forge: forge.clone(),
                owner: owner.clone(),
                mode,
                state: NamespaceState::Bound,
                owner_id: None,
                kind: None,
                bridge_did: None,
                bind_job_id: None,
                bound_by: actor.did.clone(),
                requested_at: now(),
                bound_at: Some(now()),
                roles_digest: None,
                installation_removed: false,
                forge_status: None,
                role_map: None,
            };
            store::put_namespace(&state.git_ns.ks, &ns).await?;
            let scope = Scope::Namespace(ns.id.clone());
            let mut set = store::get_rights(&state.git_ns, &scope).await?;
            set.rows.push(new_row(
                &actor.did,
                Right::NsAdmin,
                &actor.did,
                actor.member,
            ));
            store::put_rights(&state.git_ns, &scope, &set).await?;
            audit(
                state,
                &actor.did,
                None,
                Audit {
                    action: "gitNs.namespace.bound",
                    namespace: Some(&ns.id),
                    resource: Some(resource.to_string()),
                    right: None,
                    policy_version: version,
                    detail: Some("manual".into()),
                },
            )
            .await;
            audit(
                state,
                &actor.did,
                Some(&actor.did),
                Audit {
                    action: "gitNs.right.granted",
                    namespace: Some(&ns.id),
                    resource: Some(resource.to_string()),
                    right: Some(Right::NsAdmin),
                    policy_version: version,
                    detail: Some("binding".into()),
                },
            )
            .await;
            super::projection::warn_hook_overlaps_for(state, &ns).await;
            return Ok(wire::into(json!({ "namespace": wire::namespace(&ns) }))?);
        }

        // Item 4 — bridge mode needs a bridge for this forge.
        let bridges = state.config.read().await.git_ns.bridges.clone();
        bridges.get(&forge).cloned().ok_or_else(|| {
            declared(
                NO_BRIDGE,
                format!(
                    "no bridge serving {forge} is configured for this VTC; add \
                     `[git_ns.bridges] \"{forge}\" = \"<bridge DID>\"`, or bind in manual mode"
                ),
            )
        })?
    };

    // The bridge must answer in-line: `next.url` is what this response
    // carries. The lock is not held across the round trip — a slow bridge
    // must not stall every other git-namespace write — so the namespace is
    // recorded only once the bridge has accepted the job (R2.1: nothing local
    // before the remote effect), and the duplicate check is repeated then.
    let ns_id = new_id("ns");
    let job_id = new_id("job");
    let payload = json!({
        "jobId": job_id,
        "namespace": ns_id,
        "kind": "beginBind",
        "target": { "forge": forge, "owner": owner },
    });
    let ack = bridge::send_inline(state, &bridge_did, &payload).await?;
    let next = ack.next.ok_or_else(|| {
        OpError::Unavailable(
            "the bridge accepted the binding but returned nowhere to send you".into(),
        )
    })?;

    let _guard = store::write_lock().await;
    let snap = Snapshot::load(&state.git_ns).await?;
    if snap
        .namespaces
        .iter()
        .any(|n| n.forge == forge && n.owner == owner)
    {
        return Err(declared(
            ALREADY_BOUND,
            format!("{resource} is already bound, or being bound, to this VTC"),
        ));
    }
    let ns = Namespace {
        id: ns_id.clone(),
        forge: forge.clone(),
        owner: owner.clone(),
        mode,
        state: NamespaceState::Pending,
        owner_id: None,
        kind: None,
        bridge_did: Some(bridge_did.clone()),
        bind_job_id: Some(job_id.clone()),
        bound_by: actor.did.clone(),
        requested_at: now(),
        bound_at: None,
        roles_digest: None,
        installation_removed: false,
        forge_status: None,
        role_map: None,
    };
    store::put_namespace(&state.git_ns.ks, &ns).await?;
    bridge::record_inline_job(
        state,
        &bridge_did,
        &ns_id,
        JobKind::BeginBind,
        payload,
        None,
        None,
    )
    .await?;
    audit(
        state,
        &actor.did,
        None,
        Audit {
            action: "gitNs.namespace.bindRequested",
            namespace: Some(&ns.id),
            resource: Some(resource.to_string()),
            right: None,
            policy_version: None,
            detail: Some("bridge".into()),
        },
    )
    .await;
    Ok(wire::into(json!({
        "namespace": wire::namespace(&ns),
        "next": { "url": next.url },
    }))?)
}

// ── git-ns/namespace/unbind/0.1 ─────────────────────────────────────────────

pub async fn unbind(
    state: &AppState,
    actor_did: &str,
    p: unbind::Payload,
) -> OpResult<unbind::Response> {
    let actor = standing(state, actor_did).await?;
    let _guard = store::write_lock().await;
    let snap = Snapshot::load(&state.git_ns).await?;
    let t = now();
    // Item 1.
    let ns = namespace_by_id(&snap, &p.namespace)?.clone();
    let resource = ns.resource();
    // Item 2 — `git.ns.admin` on the namespace, or the community-administrator
    // capability, which could bind it again anyway.
    let passed = rules::unbind_admitted(&snap, &actor.did, &resource, actor.community_admin, t)?;
    consent_gate(state, &actor, "namespace.unbind", None).await?;
    let version = check_policy(
        state,
        PolicyInput {
            action: "namespace.unbind",
            actor: &actor,
            actor_rights: rules::effective_on(&snap, &actor.did, &resource, t)
                .into_iter()
                .collect(),
            resource: &resource,
            right: None,
            subject: None,
            visibility: None,
            expires_at: None,
            namespace: Some(&ns),
            passed,
        },
    )
    .await?;

    // Item 3 — every record in the namespace, admins included: the invariants
    // do not apply to a namespace that is going away.
    let mut revoked = 0u64;
    let ns_scope = Scope::Namespace(ns.id.clone());
    let mut scopes = vec![(ns_scope.clone(), resource.clone())];
    let repos: Vec<Repo> = snap
        .repos
        .iter()
        .filter(|r| r.namespace_id == ns.id)
        .cloned()
        .collect();
    for r in &repos {
        if let Some(res) = r.resource() {
            scopes.push((Scope::Repo(r.id.clone()), res));
        }
    }
    for (scope, res) in &scopes {
        for row in snap.rows(scope) {
            revoked += 1;
            audit(
                state,
                &actor.did,
                Some(&row.subject),
                Audit {
                    action: "gitNs.right.revoked",
                    namespace: Some(&ns.id),
                    resource: Some(res.to_string()),
                    right: Some(row.right),
                    policy_version: version,
                    detail: Some("unbound".into()),
                },
            )
            .await;
        }
        store::put_rights(&state.git_ns, scope, &Default::default()).await?;
    }
    // Item 4 — detached, and no job to the bridge: removing the app or bot
    // is for a forge owner to do.
    for mut r in repos.clone() {
        r.state = RepoState::Detached;
        r.roles_digest = None;
        store::put_repo(&state.git_ns.ks, &r).await?;
    }
    bridge::cancel_namespace_jobs(state, &ns.id).await?;
    // Item 5.
    store::delete_namespace(&state.git_ns.ks, &ns.id).await?;
    // The v0.1 role-derived grants this namespace's projection published go
    // back to the hook relay, which would otherwise not grant them again
    // until each member's next membership event. Queued now, so the relay
    // re-asserts them on its next tick; the projector hands the same keys
    // back rather than withdrawing them.
    let after = Snapshot::load(&state.git_ns).await?;
    let handed: Vec<(String, String)> = super::projection::role_mapped(state)
        .await?
        .into_iter()
        .filter(|(_, r)| {
            Resource::parse(r).is_ok_and(|r| resource.contains(&r))
                && !super::projection::in_bound_namespace(&after, &r.to_string())
        })
        .collect();
    if !handed.is_empty() {
        crate::hooks::enqueue_grants(
            &state.hooks_queue_ks,
            &format!("git-ns-unbind:{}:{}", ns.id, t.to_rfc3339()),
            &handed,
            t,
        )
        .await?;
    }
    audit(
        state,
        &actor.did,
        None,
        Audit {
            action: "gitNs.namespace.unbound",
            namespace: Some(&ns.id),
            resource: Some(resource.to_string()),
            right: None,
            policy_version: version,
            detail: None,
        },
    )
    .await;
    Ok(wire::into(json!({
        "namespace": ns.id,
        "rightsRevoked": revoked,
        "reposDetached": repos.len(),
    }))?)
}

// ── git-ns/namespace/reseat/0.3 ───────────────────────────────────────────

/// Recovery for a headless namespace: a community administrator grants
/// `git.ns.admin` on it to a current member. The capability is worth nothing
/// here unless the namespace is headless — no live `git.ns.admin` record whose
/// subject is a current member — which is checked, and the right recorded,
/// under one hold of the store lock (step 6: of two reseats, or a reseat and
/// a lapse sweep, exactly one outcome is recorded).
pub async fn namespace_reseat(
    state: &AppState,
    actor_did: &str,
    p: reseat::Payload,
) -> OpResult<reseat::Response> {
    did_core("subject", &p.subject.to_string())?;
    let actor = standing(state, actor_did).await?;
    let payload_json = serde_json::to_value(&p).map_err(AppError::from)?;
    // Step 1.
    if !actor.community_admin {
        return Err(OpError::PermissionDenied(
            "reseating a namespace needs the community-administrator capability".into(),
        ));
    }
    let _guard = store::write_lock().await;
    let snap = Snapshot::load(&state.git_ns).await?;
    let t = now();
    // Step 2.
    let ns = namespace_by_id(&snap, &p.namespace)?.clone();
    if ns.state != NamespaceState::Bound {
        return Err(declared(
            NAMESPACE_NOT_BOUND,
            format!(
                "{} is still pending: the binding in progress records its admin",
                ns.resource()
            ),
        ));
    }
    let resource = ns.resource();
    let scope = Scope::Namespace(ns.id.clone());
    // Step 3 — headless: no live record, of a current member. A record with
    // an expiry counts while it is live (the invariant, not this, discounts
    // it). The refusal names nobody.
    let mut live_admin = false;
    for row in snap
        .rows(&scope)
        .iter()
        .filter(|r| r.right == Right::NsAdmin && r.is_live(t))
    {
        if standing(state, &row.subject).await?.member {
            live_admin = true;
            break;
        }
    }
    if live_admin {
        return Err(declared(
            NOT_HEADLESS,
            format!(
                "{resource} has a live git.ns.admin, so it is not headless; its admins grant \
                 git.ns.admin with git-ns/right/grant"
            ),
        ));
    }
    // Fixed rule 7 of `git-ns/right/grant/0.3` binds reseat: a community
    // administrator reseating a headless namespace to themselves is a
    // self-grant of `git.ns.admin`, which is `git-ns/right/break-glass`'s —
    // unless single-administrator mode finds nobody else who could reseat it
    // (VTI-APV-022, `single_admin`).
    let subject = p.subject.to_string();
    did_core("subject", &subject)?;
    let waiver =
        match rules::separation_of_duties(&actor.did, &subject, Right::NsAdmin, true, &resource) {
            Ok(()) => None,
            Err(Refusal::SelfGrant(m)) => {
                let settings = settings(state).await;
                match super::single_admin::waivable(
                    state,
                    &snap,
                    &ns,
                    &actor,
                    Right::NsAdmin,
                    &resource,
                    settings.rules,
                )
                .await?
                {
                    Some(w) => Some(w),
                    None => return Err(Refusal::SelfGrant(m).into()),
                }
            }
            Err(e) => return Err(e.into()),
        };
    // Step 4 — fixed rule 5.
    let subject_standing = standing(state, &subject).await?;
    if !subject_standing.member {
        return Err(declared(
            MEMBERS_ONLY,
            "git.ns.admin goes only to a current member of the community",
        ));
    }
    let passed =
        rules::reseat_admitted(actor.community_admin, !live_admin, subject_standing.member)
            .ok_or_else(|| OpError::PermissionDenied("the reseat is not admitted".into()))?;
    // Step 5.
    let version = check_policy(
        state,
        PolicyInput {
            action: "namespace.reseat",
            actor: &actor,
            actor_rights: rules::effective_on(&snap, &actor.did, &resource, t)
                .into_iter()
                .collect(),
            resource: &resource,
            right: Some(Right::NsAdmin),
            subject: Some((
                &subject_standing,
                rules::effective_on(&snap, &subject, &resource, t)
                    .into_iter()
                    .collect(),
            )),
            visibility: None,
            expires_at: None,
            namespace: Some(&ns),
            passed,
        },
    )
    .await?;

    // Step 7's evidence: how each admin record the namespace held ended —
    // from the audit log for those already gone (revoked, and by whom and
    // why; lapsed, and when; removed on departure), and from the rights set
    // for those not yet swept. Records are named by when they ended or were
    // granted, not by whom they were held: a departed member's DID does not
    // go into a new audit row in plaintext.
    let statement = p.statement.to_string();
    let mut evidence = admin_record_history(state, &ns.id).await?;
    for row in snap
        .rows(&scope)
        .iter()
        .filter(|r| r.right == Right::NsAdmin)
    {
        let ended = if row.is_lapsed(t) {
            json!({ "ended": "lapsed", "at": row.expires_at.map(wire::timestamp) })
        } else {
            json!({ "ended": "departed" })
        };
        let mut e = json!({ "grantedAt": wire::timestamp(row.granted_at) });
        if let (Some(o), Some(extra)) = (e.as_object_mut(), ended.as_object()) {
            o.extend(extra.clone());
        }
        evidence.push(e);
    }

    // Single-administrator mode: the step-up and the `Critical` row, both
    // before the write; either missing refuses it.
    let mark = match &waiver {
        Some(w) => Some(
            super::single_admin::authorize(
                state,
                w,
                super::single_admin::WaivedOp {
                    type_uri: <reseat::Payload as trust_tasks_rs::Payload>::TYPE_URI,
                    payload: &payload_json,
                    kind: "namespace.reseat",
                },
            )
            .await?,
        ),
        None => None,
    };

    // Step 6 — permanent, so the recovered namespace meets the last-admin
    // invariant from the moment it has an admin again.
    let mut set = store::get_rights(&state.git_ns, &scope).await?;
    // The subject's own lapsed admin record goes, as a grant replaces one
    // (`git-ns/right/grant`, item 7): one record per subject and right.
    set.rows
        .retain(|r| !(r.subject == subject && r.right == Right::NsAdmin && !r.is_live(t)));
    let mut row = new_row(&subject, Right::NsAdmin, &actor.did, true);
    row.reason = Some(statement.clone());
    row.granter_was_member = actor.member;
    row.single_admin = mark;
    set.rows.push(row.clone());
    store::put_rights(&state.git_ns, &scope, &set).await?;
    // Step 8 — no forge projection to queue: `git.ns.admin` projects to no
    // forge role.

    // Step 7.
    audit(
        state,
        &actor.did,
        Some(&subject),
        Audit {
            action: "gitNs.namespace.reseated",
            namespace: Some(&ns.id),
            resource: Some(resource.to_string()),
            right: Some(Right::NsAdmin),
            policy_version: version,
            detail: Some(
                json!({ "statement": statement, "headlessEvidence": evidence }).to_string(),
            ),
        },
    )
    .await;
    audit(
        state,
        &actor.did,
        Some(&subject),
        Audit {
            action: "gitNs.right.granted",
            namespace: Some(&ns.id),
            resource: Some(resource.to_string()),
            right: Some(Right::NsAdmin),
            policy_version: version,
            detail: Some("reseat".into()),
        },
    )
    .await;
    let mut body = json!({ "right": wire::right_record(&row, &resource, true) });
    if let Some(w) = &waiver {
        super::single_admin::note(state, w, &ns, "namespace.reseat").await;
        body["ext"] = super::single_admin::response_ext(w);
    }
    Ok(wire::into(body)?)
}

/// How each `git.ns.admin` record of a namespace ended, from the audit log:
/// revoked (by whom, why), lapsed (when), or removed on its holder's
/// departure (when). The holders are not named.
async fn admin_record_history(
    state: &AppState,
    ns_id: &str,
) -> Result<Vec<serde_json::Value>, AppError> {
    let mut out = Vec::new();
    for (_, v) in state.audit_ks.prefix_iter_raw(Vec::new()).await? {
        let Ok(env) = serde_json::from_slice::<vti_common::audit::AuditEnvelope>(&v) else {
            continue;
        };
        let AuditEvent::GitNsOperation(d) = env.event else {
            continue;
        };
        if d.namespace.as_deref() != Some(ns_id) || d.right.as_deref() != Some("git.ns.admin") {
            continue;
        }
        let at = wire::timestamp(env.timestamp);
        let entry = match (d.action.as_str(), d.detail.as_deref()) {
            ("gitNs.right.lapsed", _) => json!({ "ended": "lapsed", "at": at }),
            ("gitNs.right.revoked", Some("departed" | "granterDeparted")) => {
                json!({ "ended": "departed", "at": at })
            }
            ("gitNs.right.revoked" | "gitNs.right.breakGlassRevoked", why) => {
                let mut e = json!({ "ended": "revoked", "at": at, "by": env.actor_did_plain });
                if let Some(w) = why {
                    e["why"] = json!(w);
                }
                e
            }
            _ => continue,
        };
        out.push(entry);
    }
    Ok(out)
}

// ── git-ns/repo/create/0.3 ──────────────────────────────────────────────────

pub async fn repo_create(
    state: &AppState,
    actor_did: &str,
    p: create::Payload,
) -> OpResult<create::Response> {
    let actor = standing(state, actor_did).await?;
    let payload_json = serde_json::to_value(&p).map_err(AppError::from)?;
    let _guard = store::write_lock().await;
    let snap = Snapshot::load(&state.git_ns).await?;
    let t = now();
    // Item 1.
    let ns = namespace_by_id(&snap, &p.namespace)?.clone();
    if ns.state != NamespaceState::Bound {
        return Err(declared(
            NAMESPACE_NOT_BOUND,
            format!(
                "{} is still pending: its binding has not completed",
                ns.resource()
            ),
        ));
    }
    let ns_res = ns.resource();
    // Item 2 — `git.repo.create` on the namespace, explicit or implied.
    rules::holds_admitted(&snap, &actor.did, Right::RepoCreate, &ns_res, t)?;
    let actor_rights = rules::effective_on(&snap, &actor.did, &ns_res, t);
    let visibility = visibility_from_wire(&to_string_json(&p.visibility))?;
    let name = p.name.to_string();
    let resource = ns_res.child(&name);
    // Who owns it (`git-ns/repo/create/0.3`, *Authorization*): the requester
    // by default, and only on an explicit `git.repo.create` — an implied one
    // (from `git.ns.admin`) makes the requester's own ownership a self-grant.
    let owner_dids: Vec<String> = match &p.owners {
        Some(o) => o.iter().map(|d| d.to_string()).collect(),
        None => vec![actor.did.clone()],
    };
    let mut owners = Vec::with_capacity(owner_dids.len());
    for o in owner_dids {
        did_core("owners", &o)?;
        let member = standing(state, &o).await?.member;
        owners.push((o, member));
    }
    let st = settings(state).await;
    let admit = |waiver: Option<&super::single_admin::Waivable>| {
        rules::create_owners_admitted(
            &snap,
            &actor.did,
            actor.member,
            &Scope::Namespace(ns.id.clone()),
            &ns_res,
            &resource,
            &owners,
            st.rules,
            t,
            waiver,
        )
    };
    // An implied `repo.create` naming its holder owner is a self-grant of
    // `own` — waived only where single-administrator mode finds nobody else
    // who could make it (VTI-APV-022, `single_admin`).
    let (passed, waiver) = match admit(None) {
        Ok(passed) => (passed, None),
        Err(Refusal::SelfGrant(m)) => {
            let Some(w) = super::single_admin::waivable(
                state,
                &snap,
                &ns,
                &actor,
                Right::RepoOwn,
                &resource,
                st.rules,
            )
            .await?
            else {
                return Err(Refusal::SelfGrant(m).into());
            };
            (admit(Some(&w))?, Some(w))
        }
        Err(e) => return Err(e.into()),
    };
    consent_gate(state, &actor, "repo.create", None).await?;
    let version = check_policy(
        state,
        PolicyInput {
            action: "repo.create",
            actor: &actor,
            actor_rights: actor_rights.into_iter().collect(),
            resource: &resource,
            right: Some(Right::RepoOwn),
            subject: None,
            visibility: Some(visibility),
            expires_at: None,
            namespace: Some(&ns),
            passed,
        },
    )
    .await?;
    // Item 3.
    if lookup(&snap, &resource)?.is_some() {
        return Err(declared(
            NAME_TAKEN,
            format!("this VTC already records a repository at {resource}"),
        ));
    }
    refuse_while_withdrawing(state, &snap, &resource, None).await?;
    let mark = match &waiver {
        Some(w) => Some(
            super::single_admin::authorize(
                state,
                w,
                super::single_admin::WaivedOp {
                    type_uri: <create::Payload as trust_tasks_rs::Payload>::TYPE_URI,
                    payload: &payload_json,
                    kind: "repo.create",
                },
            )
            .await?,
        ),
        None => None,
    };
    // Item 4 — reserved, and the requester owns the reservation. Published
    // only once active.
    let bot = can_bot_create(&ns);
    let repo = Repo {
        id: new_id("repo"),
        namespace_id: ns.id.clone(),
        resource: resource.to_string(),
        forge_id: None,
        visibility,
        description: p.description.as_ref().map(|d| d.to_string()),
        state: RepoState::PendingCreate,
        created_by: Some(actor.did.clone()),
        created_at: t,
        bootstrap: Bootstrap::default(),
        sync: SyncStatus::new(if bot {
            SyncState::Pending
        } else {
            SyncState::Unchecked
        }),
        failed_step: None,
        last_error: None,
        roles_digest: None,
        forge_report: Default::default(),
    };
    store::put_repo(&state.git_ns.ks, &repo).await?;
    let scope = Scope::Repo(repo.id.clone());
    let mut set = store::get_rights(&state.git_ns, &scope).await?;
    for (owner, member) in &owners {
        let mut row = new_row(owner, Right::RepoOwn, &actor.did, *member);
        if *owner == actor.did {
            row.single_admin = mark.clone();
        }
        set.rows.push(row);
    }
    store::put_rights(&state.git_ns, &scope, &set).await?;
    audit(
        state,
        &actor.did,
        None,
        Audit {
            action: "gitNs.repo.reserved",
            namespace: Some(&ns.id),
            resource: Some(resource.to_string()),
            right: None,
            policy_version: version,
            detail: None,
        },
    )
    .await;
    for (owner, _) in &owners {
        audit(
            state,
            &actor.did,
            Some(owner),
            Audit {
                action: "gitNs.right.granted",
                namespace: Some(&ns.id),
                resource: Some(resource.to_string()),
                right: Some(Right::RepoOwn),
                policy_version: version,
                detail: Some(
                    if *owner == actor.did {
                        "creator"
                    } else {
                        "create"
                    }
                    .into(),
                ),
            },
        )
        .await;
    }

    if let Some(w) = &waiver {
        super::single_admin::note(state, w, &ns, "repo.create").await;
    }

    let owners: Vec<String> = owners.into_iter().map(|(o, _)| o).collect();
    let mut response = json!({ "repo": wire::repo_summary(&repo, &owners) });
    if let Some(w) = &waiver {
        response["ext"] = super::single_admin::response_ext(w);
    }
    if bot {
        // Item 5 — the bridge creates it and turns commit trust on.
        let mut spec = json!({ "visibility": visibility.as_str() });
        if let Some(d) = &repo.description {
            spec["description"] = json!(d);
        }
        // The roles it is created with are a projection: read through the
        // view (a suspended namespace committer gets none), never written.
        let view = super::projection::ProjectionView::over(state, snap.clone()).await?;
        let roles = bridge::desired_roles_for_repo(state, &view, &ns, &repo, &owners).await?;
        let mut payload = json!({
            "namespace": ns.id,
            "kind": "createRepo",
            "repo": resource.to_string(),
            "spec": spec,
        });
        if !roles.is_empty() {
            payload["desiredRoles"] = json!(roles);
        }
        bridge::enqueue(
            state,
            NewJob {
                namespace_id: ns.id.clone(),
                kind: JobKind::CreateRepo,
                payload,
                repo_id: Some(repo.id.clone()),
                link_id: None,
            },
        )
        .await?;
    } else {
        // Item 6 — the steps a person must take. The last is always to adopt.
        let vtc_did = state
            .config
            .read()
            .await
            .vtc_did
            .clone()
            .unwrap_or_default();
        response["manualSteps"] = json!([
            format!(
                "Create the repository `{}/{}` on {}, {}, with no initial commit.",
                ns.owner,
                name,
                ns.forge,
                visibility.as_str()
            ),
            format!(
                "In a clone of it, run `vgi repo init --vtc {vtc_did} --resource {resource}` to \
                 commit the verify-trust workflow and set its variables and branch protection."
            ),
            format!(
                "Run `cnm git adopt {resource} --owner {}` to tell the VTC the repository exists.",
                actor.did
            ),
        ]);
    }
    Ok(wire::into(response)?)
}

// ── git-ns/repo/adopt/0.1 ───────────────────────────────────────────────────

pub async fn repo_adopt(
    state: &AppState,
    actor_did: &str,
    p: adopt::Payload,
) -> OpResult<adopt::Response> {
    let actor = standing(state, actor_did).await?;
    let payload_json = serde_json::to_value(&p).map_err(AppError::from)?;
    let _guard = store::write_lock().await;
    let snap = Snapshot::load(&state.git_ns).await?;
    let t = now();
    let resource = parse_resource(&p.resource)?;
    // Item 1.
    let ns = bound_namespace_for(&snap, &resource)?.clone();
    // Item 2.
    let existing = lookup(&snap, &resource)?.cloned();
    if let Some(r) = &existing
        && matches!(
            r.state,
            RepoState::Active | RepoState::Orphaned | RepoState::Archived
        )
    {
        return Err(declared(
            ALREADY_MANAGED,
            format!("this VTC already manages {resource}"),
        ));
    }
    // Item 3 — `git.ns.admin`, or an explicit `own` on a `pendingCreate`
    // reservation (the person who ran the manual steps finishing the job).
    let reservation = existing
        .as_ref()
        .filter(|r| r.state == RepoState::PendingCreate)
        .map(|r| Scope::Repo(r.id.clone()));
    let owns_reservation = reservation.as_ref().is_some_and(|scope| {
        rules::explicit_admitted(&snap, &actor.did, Right::RepoOwn, scope, t).is_some()
    });
    for o in &p.owners {
        did_core("owners", o)?;
    }
    if p.owners.is_empty() {
        return Err(OpError::Malformed(
            "a repository always has an owner: name at least one".into(),
        ));
    }
    let st = settings(state).await;
    let is_admin =
        rules::effective_on(&snap, &actor.did, &ns.resource(), t).contains(&Right::NsAdmin);
    if !is_admin && !owns_reservation {
        return Err(OpError::PermissionDenied(format!(
            "adopting {resource} needs git.ns.admin on {}, or ownership of its reservation",
            ns.resource()
        )));
    }
    // Finishing one's own reservation is the second half of `repo/create`, a
    // normal-class action; adopting anything else is elevated (design §6).
    if !owns_reservation {
        consent_gate(state, &actor, "repo.adopt", None).await?;
    }
    refuse_while_withdrawing(
        state,
        &snap,
        &resource,
        existing.as_ref().map(|r| r.id.as_str()),
    )
    .await?;
    let actor_rights: Vec<Right> = rules::effective_on(&snap, &actor.did, &resource, t)
        .into_iter()
        .collect();
    let mut version = None;
    let mut owner_standing = Vec::new();
    let mut waiver = None;
    for o in &p.owners {
        let s = standing(state, o).await?;
        // Naming an owner is a grant of `own`, under the same fixed rules as
        // `git-ns/right/grant` — except on one's own reservation, whose
        // entitlement is the reservation itself. Naming oneself is a
        // self-grant (rule 7), waived only where single-administrator mode
        // finds nobody else who could make it (VTI-APV-022, `single_admin`).
        let passed = match &reservation {
            Some(scope) if owns_reservation => {
                rules::explicit_admitted(&snap, &actor.did, Right::RepoOwn, scope, t).ok_or_else(
                    || OpError::PermissionDenied("the reservation is not yours".into()),
                )?
            }
            _ => {
                let admit = |w: Option<&super::single_admin::Waivable>| {
                    rules::grant_admitted(
                        &snap,
                        &actor.did,
                        o,
                        Right::RepoOwn,
                        &resource,
                        s.member,
                        actor.member,
                        st.rules,
                        t,
                        w,
                    )
                };
                match admit(None) {
                    Ok(passed) => passed,
                    Err(Refusal::SelfGrant(m)) => {
                        let Some(w) = super::single_admin::waivable(
                            state,
                            &snap,
                            &ns,
                            &actor,
                            Right::RepoOwn,
                            &resource,
                            st.rules,
                        )
                        .await?
                        else {
                            return Err(Refusal::SelfGrant(m).into());
                        };
                        let passed = admit(Some(&w))?;
                        waiver = Some(w);
                        passed
                    }
                    Err(e) => return Err(e.into()),
                }
            }
        };
        // Fixed rule 5 binds the reservation path too: every owner is a member.
        rules::members_only(Right::RepoOwn, s.member, actor.member)?;
        version = check_policy(
            state,
            PolicyInput {
                action: "repo.adopt",
                actor: &actor,
                actor_rights: actor_rights.clone(),
                resource: &resource,
                right: Some(Right::RepoOwn),
                subject: Some((&s, vec![])),
                visibility: None,
                expires_at: None,
                namespace: Some(&ns),
                passed,
            },
        )
        .await?;
        owner_standing.push(s);
    }
    // Single-administrator mode: the step-up and the `Critical` row, both
    // before the write; either missing refuses it.
    let mark = match &waiver {
        Some(w) => Some(
            super::single_admin::authorize(
                state,
                w,
                super::single_admin::WaivedOp {
                    type_uri: <adopt::Payload as trust_tasks_rs::Payload>::TYPE_URI,
                    payload: &payload_json,
                    kind: "repo.adopt",
                },
            )
            .await?,
        ),
        None => None,
    };

    // Item 4.
    let mut repo = match existing {
        Some(mut r) => {
            // A detached repository gets no rights back beyond `owners`.
            if r.state != RepoState::PendingCreate {
                store::put_rights(
                    &state.git_ns,
                    &Scope::Repo(r.id.clone()),
                    &Default::default(),
                )
                .await?;
            }
            // Nor its forge id: whatever is at this name on the forge now may
            // not be the repository that was detached. The inspection this
            // adoption queues reports the real one.
            if r.state == RepoState::Detached {
                r.forge_id = None;
                r.forge_report = Default::default();
                r.sync = SyncStatus::new(SyncState::Unchecked);
            }
            r.namespace_id = ns.id.clone();
            r
        }
        None => Repo {
            id: new_id("repo"),
            namespace_id: ns.id.clone(),
            resource: resource.to_string(),
            forge_id: None,
            visibility: Visibility::Public,
            description: None,
            state: RepoState::Active,
            created_by: None,
            created_at: t,
            bootstrap: Bootstrap::default(),
            sync: SyncStatus::new(SyncState::Unchecked),
            failed_step: None,
            last_error: None,
            roles_digest: None,
            forge_report: Default::default(),
        },
    };
    repo.state = RepoState::Active;
    repo.roles_digest = None;
    repo.sync = SyncStatus::new(if ns.mode == Mode::Bridge {
        SyncState::Pending
    } else {
        SyncState::Unchecked
    });
    store::put_repo(&state.git_ns.ks, &repo).await?;
    let scope = Scope::Repo(repo.id.clone());
    let mut set = store::get_rights(&state.git_ns, &scope).await?;
    for s in &owner_standing {
        if !set
            .rows
            .iter()
            .any(|r| r.subject == s.did && r.right == Right::RepoOwn && r.is_live(t))
        {
            set.rows
                .retain(|r| !(r.subject == s.did && r.right == Right::RepoOwn));
            let mut row = new_row(&s.did, Right::RepoOwn, &actor.did, s.member);
            if s.did == actor.did {
                row.single_admin = mark.clone();
            }
            set.rows.push(row);
            audit(
                state,
                &actor.did,
                Some(&s.did),
                Audit {
                    action: "gitNs.right.granted",
                    namespace: Some(&ns.id),
                    resource: Some(resource.to_string()),
                    right: Some(Right::RepoOwn),
                    policy_version: version,
                    detail: Some("adopted".into()),
                },
            )
            .await;
        }
    }
    store::put_rights(&state.git_ns, &scope, &set).await?;
    audit(
        state,
        &actor.did,
        None,
        Audit {
            action: "gitNs.repo.adopted",
            namespace: Some(&ns.id),
            resource: Some(resource.to_string()),
            right: None,
            policy_version: version,
            detail: None,
        },
    )
    .await;
    // Item 5 — inspect, then bootstrap what is missing (on the result).
    if ns.mode == Mode::Bridge {
        bridge::enqueue(
            state,
            NewJob {
                namespace_id: ns.id.clone(),
                kind: JobKind::Inspect,
                payload: json!({
                    "namespace": ns.id,
                    "kind": "inspect",
                    "repo": resource.to_string(),
                }),
                repo_id: Some(repo.id.clone()),
                link_id: None,
            },
        )
        .await?;
    }
    let owners: Vec<String> = set
        .rows
        .iter()
        .filter(|r| r.right == Right::RepoOwn && r.is_live(t))
        .map(|r| r.subject.clone())
        .collect();
    let mut body = json!({ "repo": wire::repo_summary(&repo, &owners) });
    if let Some(w) = &waiver {
        super::single_admin::note(state, w, &ns, "repo.adopt").await;
        body["ext"] = super::single_admin::response_ext(w);
    }
    Ok(wire::into(body)?)
}

// ── git-ns/repo/transfer/0.1 ────────────────────────────────────────────────

pub async fn repo_transfer(
    state: &AppState,
    actor_did: &str,
    p: transfer::Payload,
) -> OpResult<transfer::Response> {
    let actor = standing(state, actor_did).await?;
    let _guard = store::write_lock().await;
    let snap = Snapshot::load(&state.git_ns).await?;
    let t = now();
    let resource = parse_resource(&p.resource)?;
    // Item 1.
    let repo = repo_at(&snap, &resource)?.clone();
    if !matches!(repo.state, RepoState::Active | RepoState::Orphaned) {
        return Err(declared(
            REPO_NOT_ACTIVE,
            format!(
                "{resource} is {}; only an active or orphaned repository changes hands",
                repo.state.as_str()
            ),
        ));
    }
    let ns = snap.namespace(&repo.namespace_id).cloned().ok_or_else(|| {
        declared(
            UNKNOWN_NAMESPACE,
            format!("{resource} lies in no bound namespace"),
        )
    })?;
    let scope = Scope::Repo(repo.id.clone());
    // Item 2 — an explicit record to hand over. Implied ownership has none.
    let Some(passed) = rules::explicit_admitted(&snap, &actor.did, Right::RepoOwn, &scope, t)
    else {
        return Err(declared(
            NOT_OWNER,
            format!(
                "you hold no explicit git.repo.own record on {resource} to hand over; a namespace \
                 admin names an owner with git-ns/right/grant instead"
            ),
        ));
    };
    let to = p.to.to_string();
    did_core("to", &to)?;
    if to == actor.did {
        return Err(declared(SELF_TRANSFER, "`to` is you"));
    }
    let to_standing = standing(state, &to).await?;
    // Fixed rule 5 of `git-ns/right/grant/0.3`: `own` goes only to a member.
    rules::members_only(Right::RepoOwn, to_standing.member, actor.member)?;
    consent_gate(state, &actor, "repo.transfer", Some(Right::RepoOwn)).await?;
    let version = check_policy(
        state,
        PolicyInput {
            action: "repo.transfer",
            actor: &actor,
            actor_rights: rules::effective_on(&snap, &actor.did, &resource, t)
                .into_iter()
                .collect(),
            resource: &resource,
            right: Some(Right::RepoOwn),
            subject: Some((
                &to_standing,
                rules::effective_on(&snap, &to, &resource, t)
                    .into_iter()
                    .collect(),
            )),
            visibility: None,
            expires_at: None,
            namespace: Some(&ns),
            passed,
        },
    )
    .await?;
    // Item 3 — one write: the grant and the revoke land together or not at
    // all, so the repository never has fewer owners than before.
    //
    // What is handed over is what the caller holds, expiry included: the
    // recipient ends with ownership at least as durable as the caller's and
    // never more. An expiring record does not count toward the last-owner
    // invariant (as `git-ns/namespace/reseat/0.3` states it for the last
    // admin, and this VTC applies to owners alike), so a permanent owner who
    // hands over to someone holding only an expiring record must leave them a
    // permanent one — or the repository is ownerless when it lapses.
    let mut set = store::get_rights(&state.git_ns, &scope).await?;
    let live_own =
        |r: &RightRow, who: &str| r.subject == who && r.right == Right::RepoOwn && r.is_live(t);
    // `None` is permanent; otherwise the latest expiry among live records.
    let durability = |who: &str| -> Option<Option<chrono::DateTime<chrono::Utc>>> {
        let rows: Vec<&RightRow> = set.rows.iter().filter(|r| live_own(r, who)).collect();
        if rows.is_empty() {
            None
        } else if rows.iter().any(|r| r.expires_at.is_none()) {
            Some(None)
        } else {
            Some(rows.iter().filter_map(|r| r.expires_at).max())
        }
    };
    let handed = durability(&actor.did).unwrap_or(None);
    let to_already = match (durability(&to), handed) {
        (None, _) => false,
        (Some(None), _) => true,
        (Some(Some(_)), None) => false,
        (Some(Some(has)), Some(gets)) => has >= gets,
    };
    if !to_already {
        set.rows
            .retain(|r| !(r.subject == to && r.right == Right::RepoOwn));
        let mut row = new_row(&to, Right::RepoOwn, &actor.did, to_standing.member);
        row.expires_at = handed;
        set.rows.push(row);
    }
    set.rows
        .retain(|r| !(r.subject == actor.did && r.right == Right::RepoOwn));
    store::put_rights(&state.git_ns, &scope, &set).await?;
    let mut repo = repo;
    if repo.state == RepoState::Orphaned {
        repo.state = RepoState::Active;
        store::put_repo(&state.git_ns.ks, &repo).await?;
    }
    if !to_already {
        audit(
            state,
            &actor.did,
            Some(&to),
            Audit {
                action: "gitNs.right.granted",
                namespace: Some(&ns.id),
                resource: Some(resource.to_string()),
                right: Some(Right::RepoOwn),
                policy_version: version,
                detail: Some("transfer".into()),
            },
        )
        .await;
    }
    audit(
        state,
        &actor.did,
        Some(&actor.did),
        Audit {
            action: "gitNs.right.revoked",
            namespace: Some(&ns.id),
            resource: Some(resource.to_string()),
            right: Some(Right::RepoOwn),
            policy_version: version,
            detail: Some("transfer".into()),
        },
    )
    .await;
    let owners: Vec<String> = set
        .rows
        .iter()
        .filter(|r| r.right == Right::RepoOwn && r.is_live(t))
        .map(|r| r.subject.clone())
        .collect();
    Ok(wire::into(
        json!({ "repo": wire::repo_summary(&repo, &owners) }),
    )?)
}

// ── git-ns/repo/archive/0.1 ─────────────────────────────────────────────────

pub async fn repo_archive(
    state: &AppState,
    actor_did: &str,
    p: archive::Payload,
) -> OpResult<archive::Response> {
    let actor = standing(state, actor_did).await?;
    let _guard = store::write_lock().await;
    let snap = Snapshot::load(&state.git_ns).await?;
    let t = now();
    let resource = parse_resource(&p.resource)?;
    // Item 1.
    let mut repo = repo_at(&snap, &resource)?.clone();
    if matches!(
        repo.state,
        RepoState::PendingCreate | RepoState::Detached | RepoState::Unmanaged
    ) {
        return Err(declared(
            REPO_NOT_ACTIVE,
            format!(
                "{resource} is {}; there is nothing to archive",
                repo.state.as_str()
            ),
        ));
    }
    let scope = Scope::Repo(repo.id.clone());
    // Item 2 — `own`, explicit or implied by `ns.admin`.
    let passed = rules::holds_admitted(&snap, &actor.did, Right::RepoOwn, &resource, t)?;
    let actor_rights = rules::effective_on(&snap, &actor.did, &resource, t);
    let owners_now = rules::owners(&snap, &repo.id, t);
    // Item 3 — repeating an archive is safe.
    if repo.state == RepoState::Archived {
        return Ok(wire::into(json!({
            "repo": wire::repo_summary(&repo, &owners_now),
            "rightsRevoked": 0,
        }))?);
    }
    let ns = snap.namespace(&repo.namespace_id).cloned();
    consent_gate(state, &actor, "repo.archive", None).await?;
    let version = check_policy(
        state,
        PolicyInput {
            action: "repo.archive",
            actor: &actor,
            actor_rights: actor_rights.into_iter().collect(),
            resource: &resource,
            right: None,
            subject: None,
            visibility: None,
            expires_at: None,
            namespace: ns.as_ref(),
            passed,
        },
    )
    .await?;
    // Item 4.
    repo.state = RepoState::Archived;
    store::put_repo(&state.git_ns.ks, &repo).await?;
    let mut set = store::get_rights(&state.git_ns, &scope).await?;
    let (gone, kept): (Vec<RightRow>, Vec<RightRow>) = set
        .rows
        .drain(..)
        .partition(|r| r.right == Right::CommitSign);
    set.rows = kept;
    store::put_rights(&state.git_ns, &scope, &set).await?;
    for row in &gone {
        audit(
            state,
            &actor.did,
            Some(&row.subject),
            Audit {
                action: "gitNs.right.revoked",
                namespace: Some(&repo.namespace_id),
                resource: Some(resource.to_string()),
                right: Some(Right::CommitSign),
                policy_version: version,
                detail: Some("archived".into()),
            },
        )
        .await;
    }
    audit(
        state,
        &actor.did,
        None,
        Audit {
            action: "gitNs.repo.archived",
            namespace: Some(&repo.namespace_id),
            resource: Some(resource.to_string()),
            right: None,
            policy_version: version,
            detail: None,
        },
    )
    .await;
    if let Some(ns) = ns.filter(|n| n.mode == Mode::Bridge) {
        bridge::enqueue(
            state,
            NewJob {
                namespace_id: ns.id.clone(),
                kind: JobKind::Archive,
                payload: json!({
                    "namespace": ns.id,
                    "kind": "archive",
                    "repo": resource.to_string(),
                }),
                repo_id: Some(repo.id.clone()),
                link_id: None,
            },
        )
        .await?;
    }
    Ok(wire::into(json!({
        "repo": wire::repo_summary(&repo, &owners_now),
        "rightsRevoked": gone.len(),
    }))?)
}

// ── git-ns/right/grant/0.3 ──────────────────────────────────────────────────
//
// 0.1 and 0.2 are not served: every grant is held to fixed rule 7, and every
// record answered carries its `breakGlass`.

pub async fn right_grant(
    state: &AppState,
    actor_did: &str,
    p: grant::Payload,
) -> OpResult<grant::Response> {
    right_grant_via(state, actor_did, p, None).await
}

/// Why a grant is being made other than by a direct request, and what must
/// still hold, checked under the same store lock as the write.
pub(crate) struct GrantVia<'a> {
    /// The policy input's `via` (`drift.adopt`).
    pub via: &'a str,
    /// Re-checked against the snapshot the grant is decided on; a refusal
    /// here is the grant's refusal, and nothing is written.
    pub still_holds: &'a (dyn Fn(&Snapshot) -> OpResult<()> + Send + Sync),
    /// The forge account whose link must still resolve to the grant's
    /// subject, checked under the same lock — the lock every link and unlink
    /// is written under — so that a relink cannot fall between check and write
    /// (drift/resolve 0.3, adopt step 6).
    pub linked_to: Option<&'a LinkedTo>,
    /// The document the requester actually signed, which a single-administrator
    /// waiver's step-up is bound to and its audit row names
    /// ([`super::single_admin`]) — the `drift/resolve` document, not the grant
    /// derived from it.
    pub op: super::single_admin::WaivedOp<'a>,
}

/// A forge account and the member it must be linked to.
pub(crate) struct LinkedTo {
    pub forge: String,
    pub id: String,
    pub member: String,
}

pub(crate) async fn right_grant_via(
    state: &AppState,
    actor_did: &str,
    p: grant::Payload,
    via: Option<GrantVia<'_>>,
) -> OpResult<grant::Response> {
    let (row, resource, waived) = right_grant_record(state, actor_did, p, via).await?;
    let mut body = json!({ "right": wire::right_record_full(&row, &resource, true) });
    if let Some(w) = &waived {
        body["ext"] = super::single_admin::response_ext(w);
    }
    Ok(wire::into(body)?)
}

async fn right_grant_record(
    state: &AppState,
    actor_did: &str,
    p: grant::Payload,
    via: Option<GrantVia<'_>>,
) -> OpResult<(RightRow, Resource, Option<super::single_admin::Waivable>)> {
    let actor = standing(state, actor_did).await?;
    // What a single-administrator waiver's step-up is bound to: the grant as
    // signed, or the document it was derived from.
    let own_payload = serde_json::to_value(&p).map_err(AppError::from)?;
    let waived_op = match &via {
        Some(v) => v.op,
        None => super::single_admin::WaivedOp {
            type_uri: <grant::Payload as trust_tasks_rs::Payload>::TYPE_URI,
            payload: &own_payload,
            kind: "right.grant",
        },
    };
    let _guard = store::write_lock().await;
    let snap = Snapshot::load(&state.git_ns).await?;
    if let Some(v) = &via {
        (v.still_holds)(&snap)?;
        if let Some(link) = v.linked_to {
            let now_linked = super::bridge::linked_accounts(state)
                .await?
                .into_iter()
                .find(|(_, forges)| forges.get(&link.forge).is_some_and(|a| a.id == link.id))
                .map(|(did, _)| did);
            match now_linked {
                Some(did) if did == link.member => {}
                Some(_) => return Err(super::drift::subject_changed(&link.forge, &link.id)),
                None => {
                    return Err(declared(
                        super::drift::ACCOUNT_NOT_LINKED,
                        format!(
                            "{} account {} is no longer linked to a member; it can only be \
                             reverted",
                            link.forge, link.id
                        ),
                    ));
                }
            }
        }
    }
    let t = now();
    let resource = parse_resource(&p.resource)?;
    let right = right_from_wire(&to_string_json(&p.right))?;
    let subject = p.subject.to_string();
    did_core("subject", &subject)?;

    // Item 2.
    let ns = bound_namespace_for(&snap, &resource)?.clone();
    // Item 3.
    let scope = if resource.is_namespace() {
        if ns.resource() != resource {
            // Contained by a namespace yet not a repository and not the
            // namespace itself: cannot happen with a two-segment resource,
            // kept as a refusal rather than a panic.
            return Err(declared(
                UNKNOWN_NAMESPACE,
                format!("{resource} is not a bound namespace"),
            ));
        }
        Scope::Namespace(ns.id.clone())
    } else {
        let repo = repo_at(&snap, &resource)?;
        if matches!(
            repo.state,
            RepoState::Archived | RepoState::Detached | RepoState::Unmanaged
        ) {
            return Err(declared(
                REPO_NOT_ACTIVE,
                format!("{resource} is {}; it takes no grants", repo.state.as_str()),
            ));
        }
        Scope::Repo(repo.id.clone())
    };
    // Item 4 — the fixed rules, in order, then policy. Rule 7 refuses an
    // elevated self-grant unless single-administrator mode finds nobody else
    // who could make it (VTI-APV-022, `single_admin`).
    let st = settings(state).await;
    let subject_standing = standing(state, &subject).await?;
    let admit = |waiver: Option<&super::single_admin::Waivable>| {
        rules::grant_admitted(
            &snap,
            &actor.did,
            &subject,
            right,
            &resource,
            subject_standing.member,
            actor.member,
            st.rules,
            t,
            waiver,
        )
    };
    let (passed, waiver) = match admit(None) {
        Ok(passed) => (passed, None),
        Err(Refusal::SelfGrant(m)) => {
            let Some(w) = super::single_admin::waivable(
                state, &snap, &ns, &actor, right, &resource, st.rules,
            )
            .await?
            else {
                return Err(Refusal::SelfGrant(m).into());
            };
            (admit(Some(&w))?, Some(w))
        }
        Err(e) => return Err(e.into()),
    };
    // VTI-ACL-037 / VTI-ACL-071, read from the ACL entry itself: a grant is a
    // delegation and is never wider than what the granter's own live entry
    // holds at a covering qualifier. The fixed rules above decide which
    // holdings carry grant authority; this is the floor under them, so a rule
    // that ever admitted too much still could not write past the granter.
    if let Some(q) = store::scope_qualifier(&state.git_ns.ks, &scope).await?
        && !crate::acl::get_acl_entry(&state.acl_ks, &actor.did)
            .await?
            .is_some_and(|e| crate::acl::resource_grant::granter_covers(&e, right, &q))
    {
        return Err(Refusal::Escalation(format!(
            "your ACL entry does not hold {right} at {resource} or a qualifier covering it, so \
             you cannot grant it (VTI-ACL-071)"
        ))
        .into());
    }
    consent_gate(state, &actor, "right.grant", Some(right)).await?;
    let expires_at = p.expires_at;
    let version = check_policy_via(
        state,
        PolicyInput {
            action: "right.grant",
            actor: &actor,
            actor_rights: rules::effective_on(&snap, &actor.did, &resource, t)
                .into_iter()
                .collect(),
            resource: &resource,
            right: Some(right),
            subject: Some((
                &subject_standing,
                rules::effective_on(&snap, &subject, &resource, t)
                    .into_iter()
                    .collect(),
            )),
            visibility: None,
            expires_at,
            namespace: Some(&ns),
            passed,
        },
        via.as_ref().map(|v| v.via),
    )
    .await?;
    // Item 5.
    if let Some(e) = expires_at
        && e <= t
    {
        return Err(declared(EXPIRY_IN_PAST, "`expiresAt` is not in the future"));
    }
    // Item 6 — a live record already there is returned unchanged; so is an
    // unratified (or still pending) break-glass record, whose ratification is
    // `git-ns/right/ratify`'s and never a side effect of a grant (0.3, item 6).
    let mut set = store::get_rights(&state.git_ns, &scope).await?;
    if let Some(existing) = set
        .rows
        .iter()
        .find(|r| r.subject == subject && r.right == right && r.is_recorded(t))
    {
        return Ok((existing.clone(), resource, None));
    }
    // Single-administrator mode: the step-up and the `Critical` row, both
    // before the write; either missing refuses it.
    let mark = match &waiver {
        Some(w) => Some(super::single_admin::authorize(state, w, waived_op).await?),
        None => None,
    };
    // Item 7.
    set.rows
        .retain(|r| !(r.subject == subject && r.right == right));
    let mut row = new_row(&subject, right, &actor.did, subject_standing.member);
    row.expires_at = expires_at.map(|e| e.with_nanosecond(0).unwrap_or(e));
    row.reason = p.reason.as_ref().map(|r| r.to_string());
    row.granter_was_member = actor.member;
    row.single_admin = mark;
    set.rows.push(row.clone());
    store::put_rights(&state.git_ns, &scope, &set).await?;
    // A named owner ends an orphaned repository's orphanhood.
    if let Scope::Repo(id) = &scope
        && right == Right::RepoOwn
        && let Some(mut repo) = snap.repo(id).cloned()
        && repo.state == RepoState::Orphaned
    {
        repo.state = RepoState::Active;
        store::put_repo(&state.git_ns.ks, &repo).await?;
    }
    audit(
        state,
        &actor.did,
        Some(&subject),
        Audit {
            action: "gitNs.right.granted",
            namespace: Some(&ns.id),
            resource: Some(resource.to_string()),
            right: Some(right),
            policy_version: version,
            detail: expires_at.map(|_| "expiring".to_string()),
        },
    )
    .await;
    if let Some(w) = &waiver {
        super::single_admin::note(state, w, &ns, waived_op.kind).await;
    }
    Ok((row, resource, waiver))
}

// ── git-ns/right/revoke/0.3 ─────────────────────────────────────────────────

pub async fn right_revoke(
    state: &AppState,
    actor_did: &str,
    p: revoke::Payload,
) -> OpResult<revoke::Response> {
    let (row, resource) = right_revoke_record(state, actor_did, p).await?;
    Ok(wire::into(
        json!({ "revoked": wire::right_record_full(&row, &resource, true) }),
    )?)
}

async fn right_revoke_record(
    state: &AppState,
    actor_did: &str,
    p: revoke::Payload,
) -> OpResult<(RightRow, Resource)> {
    let actor = standing(state, actor_did).await?;
    let _guard = store::write_lock().await;
    let snap = Snapshot::load(&state.git_ns).await?;
    let t = now();
    let resource = parse_resource(&p.resource)?;
    let right = right_from_wire(&to_string_json(&p.right))?;
    let subject = p.subject.to_string();
    // `revoke/0.3` pins `_shared/0.4`, whose `Did` is DID-core, so the payload
    // cannot carry anything else; checked again here as for a grant.
    did_core("subject", &subject)?;
    let not_granted = || {
        declared(
            NOT_GRANTED,
            format!(
                "no live record gives {subject} {right} on {resource}; implied rights are not \
                 records and cannot be revoked"
            ),
        )
    };
    // Item 1.
    let scope = scope_for(&snap, &resource)?.ok_or_else(not_granted)?;
    // A break-glass record still waiting for its `effectiveAt` matches: it is
    // revocable while it waits (`git-ns/right/revoke/0.3`, item 1).
    let row = snap
        .rows(&scope)
        .iter()
        .find(|r| r.subject == subject && r.right == right && r.is_recorded(t))
        .cloned()
        .ok_or_else(not_granted)?;
    // Item 2 — and, for an unratified break-glass record, any community
    // administrator (0.3, *Authorization*).
    let st = settings(state).await;
    let bg_passed = rules::revoke_break_glass_admitted(actor.community_admin, &row);
    // Whoever is entitled, taking back an unratified break-glass is neither
    // gated nor put to policy.
    let unratified_bg = row.is_unratified_break_glass();
    let passed = match bg_passed {
        Some(p) => p,
        None => rules::revoke_admitted(&snap, &actor.did, &row, &resource, st.rules, t)?,
    };
    // Item 3 — resignations too.
    match &scope {
        Scope::Repo(id) => {
            let unmanaged = snap
                .repo(id)
                .is_some_and(|r| r.state == RepoState::Unmanaged);
            if right == Right::RepoOwn && !unmanaged && rules::is_last_owner(&snap, id, &subject, t)
            {
                return Err(declared(
                    LAST_OWNER,
                    format!(
                        "{subject} is the last owner of {resource}; name another owner first, or \
                         transfer with git-ns/repo/transfer"
                    ),
                ));
            }
        }
        Scope::Namespace(id) => {
            let bound = snap
                .namespace(id)
                .is_some_and(|n| n.state == NamespaceState::Bound);
            if right == Right::NsAdmin && bound && rules::is_last_admin(&snap, id, &subject, t) {
                return Err(declared(
                    LAST_ADMIN,
                    format!(
                        "{subject} is the last git.ns.admin of {resource}; grant another first"
                    ),
                ));
            }
        }
    }
    // Taking back an unratified break-glass is never gated or refused by
    // policy: policy may narrow break-glass, never entrench one (0.3, item 4).
    if !unratified_bg {
        consent_gate(state, &actor, "right.revoke", Some(right)).await?;
    }
    let subject_standing = standing(state, &subject).await?;
    // Item 4.
    let version = if unratified_bg {
        let _ = passed;
        None
    } else {
        check_policy(
            state,
            PolicyInput {
                action: "right.revoke",
                actor: &actor,
                actor_rights: rules::effective_on(&snap, &actor.did, &resource, t)
                    .into_iter()
                    .collect(),
                resource: &resource,
                right: Some(right),
                subject: Some((&subject_standing, vec![right])),
                visibility: None,
                expires_at: None,
                namespace: snap.scope_namespace(&scope),
                passed,
            },
        )
        .await?
    };
    // Item 5.
    let mut set = store::get_rights(&state.git_ns, &scope).await?;
    set.rows
        .retain(|r| !(r.subject == subject && r.right == right));
    store::put_rights(&state.git_ns, &scope, &set).await?;
    // Item 6 — a break-glass record's end is announced and audited as its
    // break-glass was.
    if row.break_glass.is_some()
        && let Some(ns) = snap.scope_namespace(&scope)
    {
        let after = Snapshot::load(&state.git_ns).await?;
        if let Err(e) = super::break_glass::announce(
            state,
            &after,
            ns,
            &resource,
            &row,
            super::break_glass::Event::Revoked,
            &actor.did,
            p.reason.as_ref().map(|r| r.to_string()),
            None,
            None,
            version,
        )
        .await
        {
            // The revocation is durable; refusing now would say it failed.
            tracing::error!(error = %e, "the break-glass revocation's audit row could not be written");
        }
    }
    audit(
        state,
        &actor.did,
        Some(&subject),
        Audit {
            action: if row.break_glass.is_some() {
                "gitNs.right.breakGlassRevoked"
            } else {
                "gitNs.right.revoked"
            },
            namespace: snap.scope_namespace(&scope).map(|n| n.id.as_str()),
            resource: Some(resource.to_string()),
            right: Some(right),
            policy_version: version,
            // Why, for the record's history (a reseat reports how each admin
            // record ended): a resignation, and the revoker's reason.
            detail: match (actor.did == subject, p.reason.as_ref()) {
                (true, Some(r)) => Some(format!("resigned: {}", **r)),
                (true, None) => Some("resigned".to_string()),
                (false, Some(r)) => Some(r.to_string()),
                (false, None) => None,
            },
        },
    )
    .await;
    // An unratified break-glass taken back closes its ratification item,
    // naming who revoked it, whichever door they used (`vtc-action-list.md`
    // §8.2).
    if unratified_bg && let Some(mark) = row.break_glass.as_ref() {
        crate::admin_actions::queues::break_glass_ended(
            state,
            &resource.to_string(),
            right.as_str(),
            &subject,
            &super::wire::timestamp(mark.at),
            false,
            &actor.did,
        )
        .await;
    }
    Ok((row, resource))
}

// ── git-ns/account/link/0.1 ─────────────────────────────────────────────────

pub async fn account_link(
    state: &AppState,
    actor_did: &str,
    p: link::Payload,
) -> OpResult<link::Response> {
    // The account is recorded against this DID.
    did_core("member", actor_did)?;
    let actor = standing(state, actor_did).await?;
    // Item 1.
    if !actor.member {
        return Err(OpError::PermissionDenied(
            "linking a forge account is for members of this community".into(),
        ));
    }
    let forge = p.forge.to_string();
    // Item 2.
    let (ns, bridge_did) = {
        let snap = Snapshot::load(&state.git_ns).await?;
        let ns = snap
            .namespaces
            .iter()
            .find(|n| {
                n.forge == forge
                    && n.mode == Mode::Bridge
                    && n.state == NamespaceState::Bound
                    && n.bridge_did.is_some()
            })
            .cloned()
            .ok_or_else(|| {
                declared(
                    UNSUPPORTED_FORGE,
                    format!("this VTC has no bridge-mode namespace on {forge} to complete a link"),
                )
            })?;
        let did = ns.bridge_did.clone().unwrap_or_default();
        (ns, did)
    };
    consent_gate(state, &actor, "account.link", None).await?;
    // Item 3 — in-line: the response is where the member goes next.
    let job_id = new_id("job");
    let payload = json!({
        "jobId": job_id,
        "namespace": ns.id,
        "kind": "beginAccountLink",
        "subject": actor.did,
    });
    let ack = bridge::send_inline(state, &bridge_did, &payload).await?;
    let next = ack.next.ok_or_else(|| {
        OpError::Unavailable("the bridge accepted the link but returned nowhere to send you".into())
    })?;
    let _guard = store::write_lock().await;
    let attempt = LinkAttempt {
        id: new_id("lnk"),
        member: actor.did.clone(),
        forge: forge.clone(),
        namespace_id: ns.id.clone(),
        job_id: job_id.clone(),
        state: LinkState::Pending,
        created_at: now(),
        expires_at: next.expires_at,
        account: None,
        finished_at: None,
    };
    store::put_link(&state.git_ns.ks, &attempt).await?;
    bridge::record_inline_job(
        state,
        &bridge_did,
        &ns.id,
        JobKind::BeginAccountLink,
        payload,
        None,
        Some(attempt.id.clone()),
    )
    .await?;
    let mut out = json!({
        "linkId": attempt.id,
        "url": next.url,
        "expiresAt": wire::timestamp(next.expires_at),
    });
    if let Some(code) = next.user_code {
        out["userCode"] = json!(code.to_string());
    }
    Ok(wire::into(out)?)
}

// ── git-ns/account/link-status/0.1 ──────────────────────────────────────────

pub async fn account_link_status(
    state: &AppState,
    actor_did: &str,
    p: link_status::Payload,
) -> OpResult<link_status::Response> {
    let unknown = || {
        declared(
            UNKNOWN_LINK,
            "no link with this identifier was begun by you, or this VTC no longer remembers it",
        )
    };
    let _guard = store::write_lock().await;
    // The entitlement is having begun the link: anyone else gets the same
    // answer as for an identifier that does not exist.
    let mut attempt = store::get_link(&state.git_ns.ks, &p.link_id)
        .await?
        .filter(|a| a.member == actor_did)
        .ok_or_else(unknown)?;
    if attempt.state == LinkState::Pending && attempt.expires_at <= now() {
        attempt.state = LinkState::Expired;
        attempt.finished_at = Some(now());
        store::put_link(&state.git_ns.ks, &attempt).await?;
    }
    let mut out = json!({ "state": attempt.state.as_str() });
    if attempt.state == LinkState::Linked
        && let Some(a) = &attempt.account
    {
        out["account"] = json!({ "forge": a.forge, "id": a.id, "login": a.login });
    }
    Ok(wire::into(out)?)
}

// ── git-ns/account/unlink/0.1 ───────────────────────────────────────────────

/// Delete the binding of the caller's DID to their account on one forge, and
/// re-project so the bridge withdraws the roles it gave that account.
///
/// Numbered items are the specification's *Request*.
pub async fn account_unlink(
    state: &AppState,
    actor_did: &str,
    p: unlink::Payload,
) -> OpResult<unlink::Response> {
    did_core("member", actor_did)?;
    let actor = standing(state, actor_did).await?;
    // Item 1: the entitlement is being the DID the account is linked to. A
    // member whose access lapsed still holds their link and may remove it;
    // anyone with no link — a non-member included — is answered `notLinked`,
    // so the answer says nothing about who is a member.
    consent_gate(state, &actor, "account.unlink", None).await?;
    let forge = p.forge.to_string();
    let guard = p.account_id.as_ref().map(|id| id.to_string());

    // Items 2 and 3, in one edit of the member row, under the git-ns store
    // lock that serialises every link completion, so a link completing at
    // the same moment is either before (and unlinked, if it is the account
    // named) or after (and kept).
    let mut outcome: Result<super::model::ForgeAccount, String> =
        Err(format!("no account is linked to you on {forge}"));
    let write = store::write_lock().await;
    crate::members::storage::edit_member(&state.members_ks, &actor.did, |m| {
        if m.removed_at.is_some() {
            return false;
        }
        let Some(forges) = m
            .extensions
            .get_mut("forges")
            .and_then(serde_json::Value::as_object_mut)
        else {
            return false;
        };
        let Some(acct) = forges.get(&forge) else {
            return false;
        };
        let text = |k: &str| {
            acct.get(k)
                .and_then(serde_json::Value::as_str)
                .filter(|v| !v.is_empty())
                .map(str::to_string)
        };
        // An entry without an id and a login is not a link: the projection
        // never reads one (`bridge::linked_accounts`).
        let (Some(id), Some(login)) = (text("id"), text("login")) else {
            return false;
        };
        if guard.as_deref().is_some_and(|g| g != id) {
            outcome = Err(format!(
                "the account linked on {forge} is not {}",
                guard.as_deref().unwrap_or_default()
            ));
            return false;
        }
        forges.remove(&forge);
        let now_empty = forges.is_empty();
        if now_empty && let Some(o) = m.extensions.as_object_mut() {
            o.remove("forges");
        }
        outcome = Ok(super::model::ForgeAccount {
            forge: forge.clone(),
            id,
            login,
        });
        true
    })
    .await?;
    drop(write);
    let unlinked = outcome.map_err(|why| declared(NOT_LINKED, why))?;
    let t = now();

    // Item 4 — the next desired roles are computed from the records, which
    // no longer hold the binding, so every target it reached changes digest
    // and is queued now; the projector sends them on its next tick. No
    // `removeAccounts` (item 5): only the roles the bridge gave go.
    let affected: Vec<String> = Snapshot::load(&state.git_ns)
        .await?
        .namespaces
        .iter()
        .filter(|n| n.forge == forge && n.mode == Mode::Bridge && n.state == NamespaceState::Bound)
        .map(|n| n.id.clone())
        .collect();
    bridge::project_roles(state, false).await?;

    // Item 6 — the member and the forge, never the account. One row per
    // namespace whose roles change, so its admins see why in its activity;
    // one with no namespace where none does.
    let rows: Vec<Option<&str>> = if affected.is_empty() {
        vec![None]
    } else {
        affected.iter().map(|id| Some(id.as_str())).collect()
    };
    for ns in rows {
        audit(
            state,
            &actor.did,
            Some(&actor.did),
            Audit {
                action: "gitNs.account.unlinked",
                namespace: ns,
                resource: None,
                right: None,
                policy_version: None,
                detail: Some(forge.clone()),
            },
        )
        .await;
    }

    Ok(wire::into(json!({
        "unlinked": { "forge": unlinked.forge, "id": unlinked.id, "login": unlinked.login },
        "unlinkedAt": wire::timestamp(t),
    }))?)
}

/// Build the `beginBind` / `beginAccountLink` acknowledgement type's `next`,
/// for [`bridge::send_inline`]'s callers.
pub type JobAck = job_wire::Response;