didbot_policy_cedar/

shared.rs

//! One request per subject, shared by every statement asked about it.
//!
//! The server registers each statement of a Cedar document with an
//! evaluator of its own, and a write asks every one of them about the same
//! subject in turn. Most of what judging costs is building Cedar's context
//! and entities from the record, and that depends only on the subject and
//! the shapes. So each thread keeps the request it prepared last, beside a
//! copy of the subject it came from, and hands it to the next statement
//! asked about an equal subject under the same shapes.
//!
//! An entry answers only a subject equal to its own in every field, so it
//! only ever hands out the request that subject would have been built into
//! anyway. Each thread has its own entry, so a write judged on one thread
//! never reads another thread's. An entry is replaced whole, and only once
//! the new request is built, so a panic partway leaves the previous entry,
//! which still answers only its own subject.
//!
//! An entry holds the record the write carried, so it is kept for as long
//! as the reuse it exists for can happen and no longer. [`KEEP`] is that
//! long, and the next miss on any thread drops every entry past it — which
//! is why each thread's entry is reachable from [`SLOTS`] rather than only
//! from the thread itself. A subject bigger than [`KEEP_BELOW`] is judged
//! from a request built the same way and then not kept at all, so what one
//! thread holds between writes is bounded by that rather than by the
//! largest record the process has ever judged.

use std::sync::{Arc, Mutex, Weak};

use didbot_policy::{
    AccountKind, Asked, Change, ClientId, DeploymentAttributes, Diff, Found, Lookup, Subject,
    SubjectAttributes, SubjectKind, Universal, WriteAction,
};
use serde_json::Value;
use time::{Duration, OffsetDateTime};

use crate::request::{AccountAttributes, Action, Operation, Prepared, Principal, Request};
use crate::schema::Shapes;

/// How long an entry is kept after the subject that filled it.
///
/// Measured on the clock the subjects themselves carry —
/// `Universal::now`, the instant the caller is judging at — so this holds
/// wherever this crate runs, the browser policy checker included.
///
/// A write asks every statement about one subject in one burst, so the
/// reuse an entry exists for happens within microseconds of it being
/// filled. Seconds rather than milliseconds because a write descheduled
/// partway through its statements should still find its own entry;
/// anything longer is a copy of an account's record held for no further
/// gain.
const KEEP: Duration = Duration::seconds(5);

/// The largest subject an entry is kept for, measured over the strings and
/// record bodies it owns.
///
/// Reuse within one write happens whatever the size — the request is built
/// once either way. This is about what is still held afterwards: a record
/// at the 1 MiB an atproto record can reach, multiplied by the worker
/// threads a write can land on, is real memory on a 2 GiB host. 64 KiB
/// covers an ordinary record comfortably and bounds the rest.
const KEEP_BELOW: usize = 64 * 1024;

/// One thread's entry, reachable from that thread and from [`SLOTS`].
type Slot = Arc<Mutex<Option<Entry>>>;

/// Every live thread's slot. Weak, so a thread that has ended takes its
/// entry with it and leaves a link [`sweep`] drops.
static SLOTS: Mutex<Vec<Weak<Mutex<Option<Entry>>>>> = Mutex::new(Vec::new());

thread_local! {
    static LAST: Slot = registered();
}

struct Entry {
    /// Held, so no other `Shapes` can take this address while the entry
    /// compares against it.
    shapes: Arc<Shapes>,
    subject: Owned,
    prepared: Arc<Prepared>,
    /// The instant the subject that filled this entry was judged at. See
    /// [`KEEP`].
    used: OffsetDateTime,
}

/// A slot of this thread's own, listed in [`SLOTS`].
fn registered() -> Slot {
    let slot: Slot = Arc::new(Mutex::new(None));
    if let Ok(mut slots) = SLOTS.lock() {
        slots.push(Arc::downgrade(&slot));
    }
    slot
}

/// Drops every entry filled more than [`KEEP`] before `now`, and forgets
/// the slots of threads that have ended.
///
/// Run on a miss rather than on every call: a miss has just paid to build a
/// Cedar request, so one lock per live thread beside it is not measurable.
fn sweep(now: OffsetDateTime) {
    let Ok(mut slots) = SLOTS.lock() else {
        return;
    };
    slots.retain(|held| {
        let Some(slot) = held.upgrade() else {
            return false;
        };
        // `try_lock`: a thread busy in its own slot is left to the next
        // sweep rather than blocking this one behind it.
        if let Ok(mut slot) = slot.try_lock() {
            if slot.as_ref().is_some_and(|entry| now - entry.used >= KEEP) {
                // Taken rather than assigned, so the entry drops after the
                // lock is released.
                let _expired = slot.take();
            }
        }
        true
    });
}

/// The instant `subject` is being judged at.
///
/// Every subject carries one, so this is a clock that works the same in a
/// server and in the browser policy checker, and one an entry can be aged
/// against without reading a clock of this crate's own.
fn judged_at(subject: &Subject<'_>) -> OffsetDateTime {
    match subject {
        Subject::Write { universal, .. }
        | Subject::Blob { universal, .. }
        | Subject::Grant { universal, .. }
        | Subject::Token { universal, .. }
        | Subject::Create { universal, .. } => universal.now,
    }
}

/// The account as Cedar's principal, with what the subject says about it.
fn principal<'a>(account: &'a str, attributes: SubjectAttributes<'a>) -> Principal<'a> {
    Principal::Account {
        did: account,
        admitted_by: None,
        attributes: AccountAttributes {
            kind: Some(attributes.kind.as_str()),
            handle: attributes.handle,
            depth: attributes.depth,
        },
    }
}

/// A `record.write`'s operation, from the write's action.
fn operation(action: WriteAction) -> Operation {
    match action {
        WriteAction::Create => Operation::Create,
        WriteAction::Update => Operation::Update,
        WriteAction::Delete => Operation::Delete,
    }
}

/// The request `subject` asks, prepared against `shapes`.
pub(crate) fn prepared(subject: &Subject<'_>, shapes: &Arc<Shapes>) -> Arc<Prepared> {
    let now = judged_at(subject);
    let held = LAST.try_with(|slot| {
        let slot = slot
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        slot.as_ref()
            .filter(|entry| Arc::ptr_eq(&entry.shapes, shapes) && entry.subject.is(subject))
            .map(|entry| Arc::clone(&entry.prepared))
    });
    if let Ok(Some(prepared)) = held {
        return prepared;
    }
    let prepared = Arc::new(with_request(subject, |request| request.prepare(shapes)));
    tracing::trace!("prepared a cedar request for a new subject");
    sweep(now);
    let subject = Owned::of(subject);
    let keep = subject.weight() <= KEEP_BELOW;
    let entry = keep.then(|| Entry {
        shapes: Arc::clone(shapes),
        subject,
        prepared: Arc::clone(&prepared),
        used: now,
    });
    if !keep {
        tracing::trace!("the subject is too big to keep its prepared request");
    }
    // The previous entry is handed back by `replace`, so it drops outside
    // the lock. An oversized subject replaces it with nothing rather than
    // leaving whatever was there.
    let _previous = LAST.try_with(|slot| {
        let mut slot = slot
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        std::mem::replace(&mut *slot, entry)
    });
    prepared
}

/// The request `subject` asks, handed to `f`. A write's `value` and
/// `existing` are the records themselves, exactly as the store holds them.
pub(crate) fn with_request<R>(subject: &Subject<'_>, f: impl FnOnce(&Request<'_>) -> R) -> R {
    let request = match subject {
        Subject::Write {
            account,
            attributes,
            universal,
            collection,
            client_id,
            action,
            before,
            after,
            ..
        } => Request {
            principal: principal(account, *attributes),
            now: universal.now,
            action: Action::RecordWrite {
                collection,
                rkey: None,
                space: None,
                operation: operation(*action),
                client: client_id.map(ClientId::as_str),
                value: *after,
                existing: *before,
            },
        },
        Subject::Blob {
            account,
            attributes,
            universal,
            mime_type,
            sniffed,
            size,
            ..
        } => Request {
            principal: principal(account, *attributes),
            now: universal.now,
            action: Action::BlobWrite {
                mime_type,
                // Both are carried by every judgment this engine answers —
                // see `CedarEvaluator::evaluate`, which skips the one that
                // runs before a blob's bytes are read.
                sniffed: sniffed.unwrap_or_default(),
                size: size.unwrap_or_default(),
                space: None,
            },
        },
        Subject::Grant {
            universal,
            client_id,
            requested_scopes,
        } => Request {
            principal: Principal::Unknown,
            now: universal.now,
            action: Action::OauthAuthorize {
                client_id: client_id.as_str(),
                scopes: requested_scopes,
            },
        },
        Subject::Token {
            account,
            attributes,
            universal,
            client_id,
            scopes,
        } => Request {
            principal: principal(account, *attributes),
            now: universal.now,
            action: Action::OauthToken {
                client_id: client_id.as_str(),
                grant: None,
                scopes,
            },
        },
        Subject::Create {
            account,
            attributes,
            universal,
            kind,
            asked,
            pool,
            zone,
            name,
            label,
        } => Request {
            // An account this server does not host, the operator, is still
            // who is asking; it just has no attributes here.
            principal: attributes.map_or(
                Principal::Account {
                    did: account,
                    admitted_by: None,
                    attributes: AccountAttributes::default(),
                },
                |attributes| principal(account, attributes),
            ),
            now: universal.now,
            action: Action::AccountCreate {
                kind: kind.as_str(),
                asked: asked.as_str(),
                pool,
                zone,
                name: *name,
                label: *label,
            },
        },
    };
    f(&request)
}

/// A [`Subject`] copied out of the borrows it arrived in. Both directions
/// name every field, so a field added to `Subject` does not compile here
/// until it is copied and compared too.
#[derive(Debug)]
enum Owned {
    Write {
        account: String,
        attributes: Attributes,
        universal: Facts,
        client_id: Option<ClientId>,
        collection: String,
        action: WriteAction,
        changes: Vec<OwnedChange>,
        before: Option<Value>,
        after: Option<Value>,
        lookups: Vec<OwnedLookup>,
    },
    Blob {
        account: String,
        attributes: Attributes,
        universal: Facts,
        client_id: Option<ClientId>,
        mime_type: String,
        sniffed: Option<String>,
        size: Option<u64>,
    },
    Grant {
        universal: Facts,
        client_id: ClientId,
        requested_scopes: Vec<String>,
    },
    Token {
        account: String,
        attributes: Attributes,
        universal: Facts,
        client_id: ClientId,
        scopes: Vec<String>,
    },
    Create {
        account: String,
        attributes: Option<Attributes>,
        universal: Facts,
        kind: Kind,
        asked: Asked,
        pool: String,
        zone: String,
        name: Option<String>,
        label: Option<String>,
    },
}

#[derive(Debug)]
struct Attributes {
    kind: Kind,
    handle: Option<String>,
    depth: Option<u32>,
}

#[derive(Debug, Clone)]
enum Kind {
    Known(AccountKind),
    Unknown(String),
}

#[derive(Debug)]
struct Facts {
    pds_did: String,
    pds_hostname: String,
    operator_did: Option<String>,
    now: OffsetDateTime,
}

#[derive(Debug)]
struct OwnedChange {
    path: String,
    before: Option<Value>,
    after: Option<Value>,
}

#[derive(Debug)]
struct OwnedLookup {
    name: String,
    account: String,
    handle: Option<String>,
    uri: String,
    found: OwnedFound,
}

#[derive(Debug)]
enum OwnedFound {
    Records(Value),
    Nothing,
    Unknown(String),
}

impl Owned {
    fn of(subject: &Subject<'_>) -> Self {
        match *subject {
            Subject::Write {
                account,
                attributes,
                universal,
                client_id,
                collection,
                action,
                diff: Diff { changes },
                before,
                after,
                lookups,
            } => Owned::Write {
                account: account.to_owned(),
                attributes: Attributes::of(attributes),
                universal: Facts::of(universal),
                client_id: client_id.cloned(),
                collection: collection.to_owned(),
                action,
                changes: changes
                    .iter()
                    .map(
                        |&Change {
                             path,
                             before,
                             after,
                         }| OwnedChange {
                            path: path.to_owned(),
                            before: before.cloned(),
                            after: after.cloned(),
                        },
                    )
                    .collect(),
                before: before.cloned(),
                after: after.cloned(),
                lookups: lookups
                    .iter()
                    .map(
                        |&Lookup {
                             name,
                             account,
                             handle,
                             uri,
                             found,
                         }| OwnedLookup {
                            name: name.to_owned(),
                            account: account.to_owned(),
                            handle: handle.map(str::to_owned),
                            uri: uri.to_owned(),
                            found: match found {
                                Found::Records(records) => OwnedFound::Records(records.clone()),
                                Found::Nothing => OwnedFound::Nothing,
                                Found::Unknown(why) => OwnedFound::Unknown(why.to_owned()),
                            },
                        },
                    )
                    .collect(),
            },
            Subject::Blob {
                account,
                attributes,
                universal,
                client_id,
                mime_type,
                sniffed,
                size,
            } => Owned::Blob {
                account: account.to_owned(),
                attributes: Attributes::of(attributes),
                universal: Facts::of(universal),
                client_id: client_id.cloned(),
                mime_type: mime_type.to_owned(),
                sniffed: sniffed.map(ToOwned::to_owned),
                size,
            },
            Subject::Grant {
                universal,
                client_id,
                requested_scopes,
            } => Owned::Grant {
                universal: Facts::of(universal),
                client_id: client_id.clone(),
                requested_scopes: requested_scopes.iter().map(|s| (*s).to_owned()).collect(),
            },
            Subject::Token {
                account,
                attributes,
                universal,
                client_id,
                scopes,
            } => Owned::Token {
                account: account.to_owned(),
                attributes: Attributes::of(attributes),
                universal: Facts::of(universal),
                client_id: client_id.clone(),
                scopes: scopes.iter().map(|s| (*s).to_owned()).collect(),
            },
            Subject::Create {
                account,
                attributes,
                universal,
                kind,
                asked,
                pool,
                zone,
                name,
                label,
            } => Owned::Create {
                account: account.to_owned(),
                attributes: attributes.map(Attributes::of),
                universal: Facts::of(universal),
                kind: Kind::of(kind),
                asked,
                pool: pool.to_owned(),
                zone: zone.to_owned(),
                name: name.map(str::to_owned),
                label: label.map(str::to_owned),
            },
        }
    }

    /// The subject this was copied from, handed to `f`.
    fn view<R>(&self, f: impl FnOnce(&Subject<'_>) -> R) -> R {
        match self {
            Owned::Write {
                account,
                attributes,
                universal,
                client_id,
                collection,
                action,
                changes,
                before,
                after,
                lookups,
            } => {
                let changes: Vec<Change<'_>> = changes
                    .iter()
                    .map(|change| Change {
                        path: &change.path,
                        before: change.before.as_ref(),
                        after: change.after.as_ref(),
                    })
                    .collect();
                let lookups: Vec<Lookup<'_>> = lookups
                    .iter()
                    .map(|lookup| Lookup {
                        name: &lookup.name,
                        account: &lookup.account,
                        handle: lookup.handle.as_deref(),
                        uri: &lookup.uri,
                        found: match &lookup.found {
                            OwnedFound::Records(records) => Found::Records(records),
                            OwnedFound::Nothing => Found::Nothing,
                            OwnedFound::Unknown(why) => Found::Unknown(why),
                        },
                    })
                    .collect();
                f(&Subject::Write {
                    account,
                    attributes: attributes.view(),
                    universal: universal.view(),
                    client_id: client_id.as_ref(),
                    collection,
                    action: *action,
                    diff: Diff { changes: &changes },
                    before: before.as_ref(),
                    after: after.as_ref(),
                    lookups: &lookups,
                })
            }
            Owned::Blob {
                account,
                attributes,
                universal,
                client_id,
                mime_type,
                sniffed,
                size,
            } => f(&Subject::Blob {
                account,
                attributes: attributes.view(),
                universal: universal.view(),
                client_id: client_id.as_ref(),
                mime_type,
                sniffed: sniffed.as_deref(),
                size: *size,
            }),
            Owned::Grant {
                universal,
                client_id,
                requested_scopes,
            } => {
                let requested_scopes: Vec<&str> =
                    requested_scopes.iter().map(String::as_str).collect();
                f(&Subject::Grant {
                    universal: universal.view(),
                    client_id,
                    requested_scopes: &requested_scopes,
                })
            }
            Owned::Token {
                account,
                attributes,
                universal,
                client_id,
                scopes,
            } => {
                let scopes: Vec<&str> = scopes.iter().map(String::as_str).collect();
                f(&Subject::Token {
                    account,
                    attributes: attributes.view(),
                    universal: universal.view(),
                    client_id,
                    scopes: &scopes,
                })
            }
            Owned::Create {
                account,
                attributes,
                universal,
                kind,
                asked,
                pool,
                zone,
                name,
                label,
            } => f(&Subject::Create {
                account,
                attributes: attributes.as_ref().map(Attributes::view),
                universal: universal.view(),
                kind: kind.view(),
                asked: *asked,
                pool,
                zone,
                name: name.as_deref(),
                label: label.as_deref(),
            }),
        }
    }

    fn is(&self, subject: &Subject<'_>) -> bool {
        self.view(|own| own == subject)
    }

    /// Roughly how many bytes of text and record body this owns, for
    /// [`KEEP_BELOW`].
    ///
    /// The variable part only: a `Subject`'s fixed fields are a handful of
    /// words each, and what separates one subject from another by orders of
    /// magnitude is the record a write carries.
    fn weight(&self) -> usize {
        match self {
            Owned::Write {
                account,
                collection,
                changes,
                before,
                after,
                ..
            } => {
                account.len()
                    + collection.len()
                    + changes
                        .iter()
                        .map(|change| {
                            change.path.len() + weigh(&change.before) + weigh(&change.after)
                        })
                        .sum::<usize>()
                    + weigh(before)
                    + weigh(after)
            }
            Owned::Blob {
                account,
                mime_type,
                sniffed,
                ..
            } => account.len() + mime_type.len() + sniffed.as_ref().map_or(0, String::len),
            Owned::Grant {
                requested_scopes, ..
            } => requested_scopes.iter().map(String::len).sum(),
            Owned::Token {
                account, scopes, ..
            } => account.len() + scopes.iter().map(String::len).sum::<usize>(),
            Owned::Create {
                account,
                pool,
                zone,
                name,
                ..
            } => account.len() + pool.len() + zone.len() + name.as_ref().map_or(0, String::len),
        }
    }
}

/// Roughly how many bytes of text `value` holds.
///
/// Not the allocator's answer — a `Value`'s own words are not counted, and
/// neither is a map's or a vector's spare capacity. It is the record's size
/// to within a constant factor, which is what [`KEEP_BELOW`] is a threshold
/// on.
fn weigh(value: &Option<Value>) -> usize {
    fn of(value: &Value) -> usize {
        match value {
            Value::Null | Value::Bool(_) | Value::Number(_) => 8,
            Value::String(text) => text.len(),
            Value::Array(items) => items.iter().map(of).sum(),
            Value::Object(fields) => fields
                .iter()
                .map(|(name, held)| name.len() + of(held))
                .sum(),
        }
    }
    value.as_ref().map_or(0, of)
}

impl Attributes {
    fn of(
        SubjectAttributes {
            kind,
            handle,
            depth,
        }: SubjectAttributes<'_>,
    ) -> Self {
        Self {
            kind: Kind::of(kind),
            handle: handle.map(str::to_owned),
            depth,
        }
    }

    fn view(&self) -> SubjectAttributes<'_> {
        SubjectAttributes {
            kind: self.kind.view(),
            handle: self.handle.as_deref(),
            depth: self.depth,
        }
    }
}

impl Kind {
    fn of(kind: SubjectKind<'_>) -> Self {
        match kind {
            SubjectKind::Known(kind) => Kind::Known(kind),
            SubjectKind::Unknown(kind) => Kind::Unknown(kind.to_owned()),
        }
    }

    fn view(&self) -> SubjectKind<'_> {
        match self {
            Kind::Known(kind) => SubjectKind::Known(*kind),
            Kind::Unknown(kind) => SubjectKind::Unknown(kind),
        }
    }
}

impl Facts {
    fn of(
        Universal {
            deployment:
                DeploymentAttributes {
                    pds_did,
                    pds_hostname,
                    operator_did,
                },
            now,
        }: Universal<'_>,
    ) -> Self {
        Self {
            pds_did: pds_did.to_owned(),
            pds_hostname: pds_hostname.to_owned(),
            operator_did: operator_did.map(ToOwned::to_owned),
            now,
        }
    }

    fn view(&self) -> Universal<'_> {
        Universal {
            deployment: DeploymentAttributes {
                pds_did: &self.pds_did,
                pds_hostname: &self.pds_hostname,
                operator_did: self.operator_did.as_deref(),
            },
            now: self.now,
        }
    }
}

#[cfg(test)]
mod tests {
    use std::collections::{BTreeMap, HashMap};
    use std::panic::{catch_unwind, AssertUnwindSafe};

    use didbot_policy::{CompileOutcome, CompiledId, Evaluator, Outcome};
    use rand::rngs::StdRng;
    use rand::seq::IndexedRandom;
    use rand::{Rng, SeedableRng};
    use serde_json::json;
    use time::Duration;

    use super::*;
    use crate::{outcome, CedarEvaluator, Document, FieldType, Shape};

    const POST: &str = "app.bsky.feed.post";
    const PROFILE: &str = "app.bsky.actor.profile";
    const REPO: &str = "sh.tangled.repo";
    const APP: &str = "https://app.example/client-metadata.json";

    /// The fixtures and the workshop's statements, each under an id no
    /// other uses, so any few of them make a document. Between them they
    /// fire, error at evaluation, need a principal the gate does not carry,
    /// and read what only a shape types.
    const STATEMENTS: &[&str] = &[
        include_str!("../../didbot-pds/tests/fixtures/c1-no-posting.cedar"),
        include_str!("../../didbot-pds/tests/fixtures/c2-one-client.cedar"),
        include_str!("../../didbot-pds/tests/fixtures/c3-no-late-edits.cedar"),
        include_str!("../../didbot-pds/tests/fixtures/c4-public-only.cedar"),
        r#"@id("subtree")
        @reason("Nothing under the operator posts.")
        forbid(principal in Didbot::Account::"did:plc:operator", action == Didbot::Action::"record.write", resource)
        when { resource.collection == "app.bsky.feed.post" };"#,
        r#"@id("not-the-agent")
        @reason("The agent signs in nowhere.")
        forbid(principal == Didbot::Account::"did:plc:agent", action in [Didbot::Action::"oauth.authorize", Didbot::Action::"oauth.token"], resource);"#,
        r#"@id("no-crypto-names")
        @reason("Display names may not mention crypto.")
        forbid(principal, action == Didbot::Action::"record.write", resource)
        when { resource.collection == "app.bsky.actor.profile" && context has value && context.value.displayNmae like "*crypto*" };"#,
        r#"@id("no-secret-repos")
        @reason("Repositories may not be named secret.")
        forbid(principal, action == Didbot::Action::"record.write", resource)
        when { resource.collection == "sh.tangled.repo" && context has value && context.value.name == "secret" };"#,
        r#"@id("no-app-writes")
        @reason("This app may not write here.")
        forbid(principal, action == Didbot::Action::"record.write", resource)
        when { context has client && context.client.id like "https://app.example/*" };"#,
        r#"@id("no-app-sign-in")
        @reason("This app may not sign in here.")
        forbid(principal, action == Didbot::Action::"oauth.authorize", resource)
        when { resource.id like "https://app.example/*" };"#,
        r#"@id("unguarded")
        @reason("This app may not name repositories.")
        forbid(principal, action == Didbot::Action::"record.write", resource)
        when { resource.collection == "sh.tangled.repo" && context has value && context.client.id like "https://app.example/*" };"#,
        r#"@id("no-chat-tokens")
        @reason("Chat apps get no tokens.")
        forbid(principal, action == Didbot::Action::"oauth.token", resource)
        when { resource.id like "https://chat.example/*" };"#,
        r#"@id("no-refresh")
        @reason("Tokens are not refreshed.")
        forbid(principal, action == Didbot::Action::"oauth.token", resource)
        when { context.grant == "refresh_token" };"#,
        r#"@id("no-secret-text")
        @reason("Nothing says secret.")
        forbid(principal, action == Didbot::Action::"record.write", resource)
        when { context has value && context.value has text && context.value.text like "*secret*" };"#,
    ];

    fn now() -> OffsetDateTime {
        time::macros::datetime!(2026-09-11 12:00 UTC)
    }

    /// Posts described, as the workshop describes them.
    fn described() -> Arc<Shapes> {
        let mut shapes = Shapes::new();
        shapes
            .describe(
                POST,
                Shape::new()
                    .field("text", FieldType::String)
                    .field("createdAt", FieldType::Datetime)
                    .optional("langs", FieldType::Set(Box::new(FieldType::String))),
            )
            .expect("the post shape is a valid record type");
        Arc::new(shapes)
    }

    /// One compiled document, and what judging it afresh needs.
    struct Judge {
        evaluator: Arc<CedarEvaluator>,
        compiled: CompiledId,
        document: Arc<Document>,
        shapes: Arc<Shapes>,
    }

    impl Judge {
        fn new(
            evaluator: &Arc<CedarEvaluator>,
            shapes: &Arc<Shapes>,
            source: &str,
        ) -> Option<Self> {
            let compiled = match evaluator.compile(&json!({ "source": source })) {
                CompileOutcome::Compiled(id) | CompileOutcome::CompiledWithWarnings(id, _) => id,
                CompileOutcome::Rejected(_) => return None,
            };
            Some(Self {
                evaluator: Arc::clone(evaluator),
                compiled,
                document: evaluator.document(compiled).expect("just compiled"),
                shapes: Arc::clone(shapes),
            })
        }

        /// Through the evaluator, sharing the request between statements.
        /// An evaluation neither path can complete -- a statement's own
        /// check failing, never this call's `compiled` lookup, which was
        /// just compiled -- becomes the same `Outcome::Reject` the two
        /// paths would refuse the write with, so they stay comparable.
        fn evaluate(&self, subject: &Owned) -> Outcome {
            subject
                .view(|subject| self.evaluator.evaluate(subject, self.compiled))
                .unwrap_or_else(|error| Outcome::Reject {
                    reason: error.to_string(),
                })
        }

        /// A request built for this call alone, the way every statement
        /// built its own before they shared one.
        fn fresh(&self, subject: &Owned) -> Outcome {
            subject.view(|subject| {
                outcome(with_request(subject, |request| {
                    self.document.decide(request, &self.shapes)
                }))
                .unwrap_or_else(|error| Outcome::Reject {
                    reason: error.to_string(),
                })
            })
        }
    }

    /// Documents of one to three statements over both sets of shapes,
    /// several to an evaluator, the way a server compiles them.
    fn judges(rng: &mut StdRng) -> Vec<Judge> {
        let mut judges = Vec::new();
        for shapes in [described(), Arc::new(Shapes::new())] {
            let loadable: Vec<&str> = STATEMENTS
                .iter()
                .copied()
                .filter(|source| Document::load(source, &shapes).is_ok())
                .collect();
            assert!(loadable.len() >= STATEMENTS.len() - 1, "{loadable:?}");
            for _ in 0..8 {
                let evaluator = Arc::new(CedarEvaluator::new(Arc::clone(&shapes)));
                for _ in 0..rng.random_range(1..=3) {
                    let size = rng.random_range(1..=3);
                    let source = loadable
                        .choose_multiple(rng, size)
                        .copied()
                        .collect::<Vec<_>>();
                    judges.extend(Judge::new(&evaluator, &shapes, &source.join("\n")));
                }
            }
        }
        judges
    }

    fn records() -> Vec<Value> {
        let post = |text: &str, age: Duration| {
            json!({
                "text": text,
                "createdAt": (now() - age).format(&time::format_description::well_known::Rfc3339).unwrap(),
                "langs": ["en"],
            })
        };
        vec![
            post("hello", Duration::ZERO),
            post("hello", Duration::days(2)),
            post("secret plans", Duration::hours(1)),
            json!({ "displayName": "Agent" }),
            json!({ "displayName": "crypto fan" }),
            json!({ "name": "secret" }),
            json!({ "name": "open" }),
            json!({ "text": "not a time", "createdAt": "yesterday" }),
            // Cedar has no floats, so this record cannot become a context.
            json!({ "text": "hello", "score": 1.5 }),
        ]
    }

    fn leaves(path: &str, value: &Value, out: &mut BTreeMap<String, Value>) {
        match value {
            Value::Object(map) if !map.is_empty() => {
                for (key, child) in map {
                    let child_path = if path.is_empty() {
                        key.clone()
                    } else {
                        format!("{path}.{key}")
                    };
                    leaves(&child_path, child, out);
                }
            }
            Value::Array(items) if !items.is_empty() => {
                for (index, child) in items.iter().enumerate() {
                    leaves(&format!("{path}[{index}]"), child, out);
                }
            }
            leaf => {
                out.insert(path.to_owned(), leaf.clone());
            }
        }
    }

    fn diff(before: Option<&Value>, after: Option<&Value>) -> Vec<OwnedChange> {
        let (mut old, mut new) = (BTreeMap::new(), BTreeMap::new());
        if let Some(before) = before {
            leaves("", before, &mut old);
        }
        if let Some(after) = after {
            leaves("", after, &mut new);
        }
        let mut paths: Vec<&String> = old.keys().chain(new.keys()).collect();
        paths.sort_unstable();
        paths.dedup();
        paths
            .into_iter()
            .filter(|path| old.get(*path) != new.get(*path))
            .map(|path| OwnedChange {
                path: path.clone(),
                before: old.get(path).cloned(),
                after: new.get(path).cloned(),
            })
            .collect()
    }

    fn client(id: &str) -> ClientId {
        ClientId::parse(id).expect("a canonical client id")
    }

    fn facts(rng: &mut StdRng) -> Facts {
        let (pds_did, pds_hostname) = *[
            ("did:web:pds.example", "pds.example"),
            ("did:web:b.example", "b.example"),
        ]
        .choose(rng)
        .unwrap();
        Facts {
            pds_did: pds_did.to_owned(),
            pds_hostname: pds_hostname.to_owned(),
            operator_did: None,
            now: now()
                + *[Duration::ZERO, Duration::nanoseconds(1), Duration::days(1)]
                    .choose(rng)
                    .unwrap(),
        }
    }

    fn attributes(rng: &mut StdRng) -> Attributes {
        Attributes {
            kind: [
                Kind::Known(AccountKind::Agent),
                Kind::Known(AccountKind::Host),
                Kind::Unknown("robot".to_owned()),
            ]
            .choose(rng)
            .unwrap()
            .clone(),
            handle: [None, Some("agent.example".to_owned())]
                .choose(rng)
                .unwrap()
                .clone(),
            depth: *[None, Some(1), Some(2)].choose(rng).unwrap(),
        }
    }

    fn subject(rng: &mut StdRng, records: &[Value]) -> Owned {
        let clients = [
            APP,
            "https://other-app.example/client-metadata.json",
            "https://quernstone.example/client-metadata.json",
            "https://chat.example/app",
        ];
        let account = (*["did:plc:agent", "did:plc:other"].choose(rng).unwrap()).to_owned();
        let scopes = [&["atproto"][..], &["atproto", "transition:generic"], &[]]
            .choose(rng)
            .unwrap()
            .iter()
            .map(|s| (*s).to_owned())
            .collect();
        let client_id = client(clients.choose(rng).unwrap());
        match rng.random_range(0..6) {
            0 => Owned::Grant {
                universal: facts(rng),
                client_id,
                requested_scopes: scopes,
            },
            1 => Owned::Token {
                account,
                attributes: attributes(rng),
                universal: facts(rng),
                client_id,
                scopes,
            },
            _ => {
                let (before, after, action) = match rng.random_range(0..3) {
                    0 => (None, records.choose(rng), WriteAction::Create),
                    1 => (
                        records.choose(rng),
                        records.choose(rng),
                        WriteAction::Update,
                    ),
                    _ => (records.choose(rng), None, WriteAction::Delete),
                };
                Owned::Write {
                    account,
                    attributes: attributes(rng),
                    universal: facts(rng),
                    client_id: rng.random_bool(0.6).then_some(client_id),
                    collection: (*[POST, PROFILE, REPO, "app.example.other"]
                        .choose(rng)
                        .unwrap())
                    .to_owned(),
                    action,
                    changes: diff(before, after),
                    before: before.cloned(),
                    after: after.cloned(),
                    lookups: Vec::new(),
                }
            }
        }
    }

    /// Which of the evaluator's answers a run has seen.
    #[derive(Default, Debug)]
    struct Seen {
        allowed: usize,
        forbidden: usize,
        not_carried: usize,
        errored: usize,
        not_built: usize,
    }

    impl Seen {
        fn count(&mut self, outcome: &Outcome) {
            match outcome {
                Outcome::Allow => self.allowed += 1,
                Outcome::Reject { reason } if reason.contains("does not carry") => {
                    self.not_carried += 1;
                }
                Outcome::Reject { reason } if reason.contains("could not complete") => {
                    self.errored += 1;
                }
                Outcome::Reject { reason } if reason.contains("could not be built") => {
                    self.not_built += 1;
                }
                Outcome::Reject { .. } => self.forbidden += 1,
                Outcome::Freeze { .. } => panic!("a cedar document never freezes"),
            }
        }

        fn every_kind(&self) -> bool {
            [
                self.allowed,
                self.forbidden,
                self.not_carried,
                self.errored,
                self.not_built,
            ]
            .iter()
            .all(|&n| n > 0)
        }
    }

    /// Random documents asked about random subjects in a random order,
    /// mostly in runs on one subject the way a write asks every statement
    /// in turn. Every answer, reason text included, is the one a request
    /// built for that call alone gives.
    #[test]
    fn every_statement_answers_as_a_fresh_build_would() {
        let mut rng = StdRng::seed_from_u64(0x0c3d_a75e);
        let judges = judges(&mut rng);
        let records = records();
        let subjects: Vec<Owned> = (0..80).map(|_| subject(&mut rng, &records)).collect();
        let mut fresh = HashMap::new();
        let mut seen = Seen::default();
        let mut on = 0;
        for _ in 0..4000 {
            if rng.random_bool(0.2) {
                on = rng.random_range(0..subjects.len());
            }
            let at = rng.random_range(0..judges.len());
            let judge = &judges[at];
            let want = fresh
                .entry((at, on))
                .or_insert_with(|| judge.fresh(&subjects[on]));
            let got = judge.evaluate(&subjects[on]);
            assert_eq!(&got, want, "subject {:?}", subjects[on]);
            seen.count(&got);
        }
        assert!(seen.every_kind(), "{seen:?}");
    }

    /// Two writes that differ only in the record's text, alternating on
    /// one thread: each is judged on its own text.
    #[test]
    fn a_request_never_answers_for_the_next_write() {
        let shapes = described();
        let evaluator = Arc::new(CedarEvaluator::new(Arc::clone(&shapes)));
        let judge = Judge::new(&evaluator, &shapes, STATEMENTS[13]).expect("loads");
        let write = |text: &str| Owned::Write {
            account: "did:plc:agent".to_owned(),
            attributes: Attributes {
                kind: Kind::Known(AccountKind::Agent),
                handle: None,
                depth: None,
            },
            universal: Facts {
                pds_did: "did:web:pds.example".to_owned(),
                pds_hostname: "pds.example".to_owned(),
                now: now(),
                operator_did: None,
            },
            client_id: None,
            collection: POST.to_owned(),
            action: WriteAction::Create,
            changes: diff(None, Some(&json!({ "text": text }))),
            before: None,
            after: Some(json!({ "text": text })),
            lookups: Vec::new(),
        };
        let (open, secret) = (write("hello"), write("secret plans"));
        for _ in 0..3 {
            assert_eq!(judge.evaluate(&open), Outcome::Allow);
            assert_eq!(
                judge.evaluate(&secret),
                Outcome::Reject {
                    reason: "Nothing says secret.".to_owned()
                }
            );
        }
    }

    /// A create of `text`, as a subject.
    fn posting(text: &str) -> Owned {
        Owned::Write {
            account: "did:plc:agent".to_owned(),
            attributes: Attributes {
                kind: Kind::Known(AccountKind::Agent),
                handle: None,
                depth: None,
            },
            universal: Facts {
                pds_did: "did:web:pds.example".to_owned(),
                pds_hostname: "pds.example".to_owned(),
                now: now(),
                operator_did: None,
            },
            client_id: None,
            collection: POST.to_owned(),
            action: WriteAction::Create,
            changes: diff(None, Some(&json!({ "text": text }))),
            before: None,
            after: Some(json!({ "text": text })),
            lookups: Vec::new(),
        }
    }

    /// A record past [`KEEP_BELOW`] is judged from a request built the same
    /// way, and is not what the thread goes on holding afterwards. An
    /// ordinary one still is, which is what the reuse is for.
    #[test]
    fn a_record_too_big_to_keep_is_judged_all_the_same() {
        let shapes = described();
        let ordinary = posting("hello");
        let first = ordinary.view(|subject| prepared(subject, &shapes));
        let again = ordinary.view(|subject| prepared(subject, &shapes));
        assert!(
            Arc::ptr_eq(&first, &again),
            "an ordinary write built its request twice"
        );

        let huge = posting(&"o".repeat(KEEP_BELOW + 1));
        let first = huge.view(|subject| prepared(subject, &shapes));
        let again = huge.view(|subject| prepared(subject, &shapes));
        assert!(
            first.built.is_ok(),
            "a record too big to keep was not judged"
        );
        assert!(
            !Arc::ptr_eq(&first, &again),
            "a record too big to keep was kept anyway"
        );
    }

    /// A worker that judged one write and then went quiet does not go on
    /// holding that write's record. The sweep any later miss runs is what
    /// takes it, from whichever thread that miss happened on.
    #[test]
    fn an_idle_workers_record_is_dropped_by_a_sweep_from_another_thread() {
        let shapes = described();
        let (ready, judged) = std::sync::mpsc::channel();
        let (stop, wait) = std::sync::mpsc::channel::<()>();
        let worker = std::thread::spawn(move || {
            let write = posting("a record nobody asks about again");
            let held = write.view(|subject| prepared(subject, &shapes));
            ready.send((held, LAST.with(Arc::clone))).expect("sent");
            // Alive, and judging nothing further.
            wait.recv().ok();
        });

        let (held, slot) = judged.recv().expect("the worker judged its write");
        assert_eq!(
            Arc::strong_count(&held),
            2,
            "the worker did not keep the request it prepared"
        );

        // This one entry is aged rather than the clock, so a sweep running
        // for some other test's write is not disturbed by this one.
        slot.lock()
            .expect("not poisoned")
            .as_mut()
            .expect("held")
            .used -= KEEP;
        sweep(now() + Duration::days(7));
        assert_eq!(
            Arc::strong_count(&held),
            1,
            "an idle worker still holds the record it judged"
        );

        stop.send(()).ok();
        worker.join().expect("the worker finished");
    }

    /// The same edit judged by two engines, one that types `createdAt`
    /// as a datetime and one that does not. Each gets the request its own
    /// shapes build, however the two alternate.
    #[test]
    fn engines_with_different_shapes_never_share_a_request() {
        let described = described();
        let undescribed = Arc::new(Shapes::new());
        let typed = Judge::new(
            &Arc::new(CedarEvaluator::new(Arc::clone(&described))),
            &described,
            STATEMENTS[2],
        )
        .expect("loads");
        let untyped = Judge::new(
            &Arc::new(CedarEvaluator::new(Arc::clone(&undescribed))),
            &undescribed,
            STATEMENTS[2],
        )
        .expect("loads");
        let records = records();
        let edit = Owned::Write {
            account: "did:plc:agent".to_owned(),
            attributes: Attributes {
                kind: Kind::Known(AccountKind::Agent),
                handle: None,
                depth: None,
            },
            universal: Facts {
                pds_did: "did:web:pds.example".to_owned(),
                pds_hostname: "pds.example".to_owned(),
                now: now(),
                operator_did: None,
            },
            client_id: None,
            collection: POST.to_owned(),
            action: WriteAction::Update,
            changes: diff(Some(&records[1]), Some(&records[0])),
            before: Some(records[1].clone()),
            after: Some(records[0].clone()),
            lookups: Vec::new(),
        };
        let (want_typed, want_untyped) = (typed.fresh(&edit), untyped.fresh(&edit));
        assert_ne!(want_typed, want_untyped, "the shapes change the answer");
        for _ in 0..3 {
            assert_eq!(typed.evaluate(&edit), want_typed);
            assert_eq!(untyped.evaluate(&edit), want_untyped);
        }
    }

    /// Writes judged on eight threads at once, against the same
    /// evaluators: each thread's answers are its own subjects'. Each
    /// thread asks every statement about one write in turn, then asks each
    /// statement about every write in turn.
    #[test]
    fn concurrent_writes_are_each_judged_on_their_own_subject() {
        let mut rng = StdRng::seed_from_u64(0x7ead);
        let judges = judges(&mut rng);
        let records = records();
        let subjects: Vec<Owned> = (0..64).map(|_| subject(&mut rng, &records)).collect();
        let fresh: Vec<Vec<Outcome>> = subjects
            .iter()
            .map(|subject| judges.iter().map(|judge| judge.fresh(subject)).collect())
            .collect();
        std::thread::scope(|scope| {
            for thread in 0..8 {
                let (judges, subjects, fresh) = (&judges, &subjects, &fresh);
                let own: Vec<usize> = (thread..subjects.len()).step_by(8).collect();
                scope.spawn(move || {
                    for &on in &own {
                        for (at, judge) in judges.iter().enumerate() {
                            assert_eq!(judge.evaluate(&subjects[on]), fresh[on][at]);
                        }
                    }
                    for (at, judge) in judges.iter().enumerate() {
                        for &on in &own {
                            assert_eq!(judge.evaluate(&subjects[on]), fresh[on][at]);
                        }
                    }
                });
            }
        });
    }

    /// A thread that panics partway through judging a write goes on to
    /// judge the next one on its own subject.
    #[test]
    fn a_panic_mid_judgment_leaves_nothing_behind_for_the_next_write() {
        let mut rng = StdRng::seed_from_u64(0x0bad);
        let judges = judges(&mut rng);
        let records = records();
        let subjects: Vec<Owned> = (0..16).map(|_| subject(&mut rng, &records)).collect();
        for pair in subjects.windows(2) {
            let (first, next) = (&pair[0], &pair[1]);
            let panicked = catch_unwind(AssertUnwindSafe(|| {
                judges[0].evaluate(first);
                panic!("a judgment dies partway");
            }));
            assert!(panicked.is_err());
            for judge in &judges {
                assert_eq!(judge.evaluate(next), judge.fresh(next));
                assert_eq!(judge.evaluate(first), judge.fresh(first));
            }
        }
    }
}