security: harden registry trust and ingestion

CHANGELOG.md

@@ -1,5 +1,18 @@
 # Changelog
 
+## 0.3.0 - 2026-09-13
+
+- Snapshot authenticated SQLite bytes into a private file before queries and sign
+  sealed receipt bytes, removing mutable-path time-of-check/time-of-use windows.
+- Require consumer-supplied reviewer trust roots during signed verification and
+  activation; support retired reviewer keys with OpenSSH validity epochs.
+- Hash the exact source descriptor stream consumed by parsers and add configurable
+  expanded-byte, record, database-growth, query-output and diff-output limits.
+- Bind identity decisions to entity metadata, redact descriptions from typed diffs,
+  include diff attribution and keep CrUX job IDs stable across output-only limits.
+- Migrate writer stores to schema/rules version 3 while allowing activation over
+  pinned version 1 and version 2 generations for revocation checks.
+
 ## 0.2.0 - 2026-09-13
 
 - Open generations from an authenticated immutable SQLite file descriptor and

Cargo.lock

@@ -78,14 +78,14 @@ checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470"
 
 [[package]]
 name = "argand-atomic"
-version = "0.2.0"
+version = "0.3.0"
 dependencies = [
  "tempfile",
 ]
 
 [[package]]
 name = "argand-site-registry"
-version = "0.2.0"
+version = "0.3.0"
 dependencies = [
  "anyhow",
  "argand-atomic",

Cargo.toml

@@ -4,7 +4,7 @@ resolver = "3"
 members = ["crates/argand-atomic", "crates/argand-site-registry"]
 
 [workspace.package]
-version = "0.2.0"
+version = "0.3.0"
 authors = ["Nic Weyand"]
 edition = "2024"
 license = "AGPL-3.0-or-later"

README.md

@@ -121,8 +121,8 @@ have no dataset, review, signing or activation authority.
 
 The code remains **AGPL-3.0-or-later**; the complete license is in [LICENSE](LICENSE).
 Original attribution is retained. [UPSTREAM.json](UPSTREAM.json) records the signed
-Argand extraction revision and original file hashes. Version 0.2 advances the
+Argand extraction revision and original file hashes. Version 0.3 advances the
-standalone database to schema version 2; the migration and consumer rules are
+standalone database to schema version 3; the migration and consumer rules are
 documented in [CONSUMERS.md](docs/CONSUMERS.md). Argand still uses its embedded copy;
 its switch to a pinned upstream release is a coordinated downstream change. This
 repository does not operate a public approved-link dataset.

crates/argand-site-registry/README.md

@@ -271,7 +271,9 @@ argand-site-registry sign --generation "$ARGAND_SITE_DATA/generation-2" \
   --allowed-reviewers /secure/reviewer-allowed-signers
 argand-site-registry activate --generation "$ARGAND_SITE_DATA/generation-2" \
   --current "$ARGAND_SITE_DATA/current.json" \
-  --allowed-signers /secure/registry-allowed-signers --identity registry-publisher
+  --allowed-signers /secure/registry-allowed-signers \
+  --allowed-reviewers /secure/reviewer-allowed-signers \
+  --identity registry-publisher
 ```
 
 Before signing, the CLI replays every stored reviewer signature against the supplied
@@ -280,7 +282,8 @@ operator-controlled SSH release key. The external release allowed-signers
 file follows OpenSSH syntax: `registry-publisher ssh-ed25519 PUBLIC_KEY`. Neither
 keys nor the trust file should come from the downloaded dataset. Consumers can
 open `Registry::open(path, trusted_receipt_sha256)` once and reuse its indexed
-queries, or verify a publisher with `release::verify_signed` first. A hash proves
+queries, or verify both a publisher and the retained reviewer proofs with
+`release::verify_signed` first. A hash proves
 integrity only relative to a trusted pin. Signature verification authenticates
 the publisher, not the truth of a source assertion.
 
@@ -313,7 +316,7 @@ downloads. No scheduled job signs, approves, renews approvals or activates links
 
 ## Storage and operating limits
 
-Migrations `migrations/001.sql` and `002.sql` own schema version 2. `sources`, `records` and
+Migrations `migrations/001.sql`, `002.sql` and `003.sql` own schema version 3. `sources`, `records` and
 `facts` preserve snapshot/native IDs, licenses, retrieval times and confidence;
 `reviews` and their cryptographic `review_auth` proofs are append-only. Complete source selection is latest retrieval time per
 provider/scope, with digest as the deterministic tie break. Use the **same scope**
@@ -331,21 +334,28 @@ join. The canonical label rule prefers labels, then English,
 then language/text order. Original labels/aliases are kept. Popularity has its
 own source, target, observation period and audience scope and never creates an
 ownership edge. All derivations bind input fact IDs, PSL identity and the
-`argand.site-rules/v2` contract through their generation receipt. A v1 writer store
+`argand.site-rules/v3` contract through their generation receipt. A v1 or v2 writer store
 migrates in place while retaining review history. Any legacy unauthenticated
-decision makes release signing fail closed; start a reviewed v2 store from the
+decision makes release signing fail closed; start a reviewed v3 store from the
 pinned source inputs rather than deleting historical decisions.
 
 Imports use transactions of 256 relevant records with durable replay checkpoints.
-Restart replays the compressed stream and skips committed records. Large source
+Restart replays the compressed stream and skips committed records. The exact open
-records are capped at 16 MiB; SQLite has an 8 MiB page cache and disk-backed sorts.
+descriptor stream is hashed, and any integrity or resource failure removes all
+partially committed rows for that source. Large source records are capped at 16 MiB;
+imports also have finite expanded-byte, record and database-growth ceilings. Override
+them with `import --maximum-expanded-bytes`, `--maximum-records` and
+`--maximum-database-growth-bytes` when a reviewed source requires different bounds.
+SQLite has an 8 MiB page cache and disk-backed sorts.
 Builds stream a canonical sorted copy and never load the full registry into RAM.
 Names and entity metadata exposed by lookup are capped at 256 facts each, with
 uncapped totals; the complete assertions remain available in the database/export.
-Lookup returns at most 100 edges and reports all pre-limit ambiguity counts.
+Lookup returns at most 100 edges, reports all pre-limit ambiguity counts and caps
+aggregate serialized candidate data at 64 MiB. Typed diff events are capped at
+16 MiB and the full stream at 1 GiB; descriptions are redacted and the header
+contains source attribution.
 The full store/history and each generation consume disk; there is no automatic
-pruning. The original compressed source is hashed before/after import, so expect
+pruning. These bounds are not a full-dump throughput claim.
-extra sequential disk reads. These bounds are not a full-dump throughput claim.
 
 The `observation` module validates and deterministically normalizes future crawler
 evidence for redirects, canonical links, hreflang, JSON-LD sameAs, sitemaps and

crates/argand-site-registry/migrations/003.sql

@@ -0,0 +1,3 @@
+-- By Nic Weyand! Bind identity decisions to all displayed evidence fields.
+UPDATE registry_metadata SET rules='argand.site-rules/v3' WHERE singleton=1;
+PRAGMA user_version=3;

crates/argand-site-registry/src/adapters/curlie.rs

@@ -14,7 +14,9 @@ pub(super) struct Curlie;
 
 impl SourceAdapter for Curlie {
     fn ingest(&self, input: &mut dyn BufRead, sink: &mut dyn RecordSink) -> anyhow::Result<()> {
-        let gzip = flate2::read::MultiGzDecoder::new(input);
+        const MAXIMUM_EXPANDED_ARCHIVE_BYTES: u64 = 1024 * 1024 * 1024 * 1024;
+        let gzip =
+            flate2::read::MultiGzDecoder::new(input).take(MAXIMUM_EXPANDED_ARCHIVE_BYTES + 1);
         let mut archive = tar::Archive::new(gzip);
         let mut entries = 0_u64;
         for entry in archive.entries()? {
@@ -97,6 +99,7 @@ impl SourceAdapter for Curlie {
                 "unexpected archive trailing data"
             );
         }
+        ensure!(gzip.limit() > 0, "Curlie archive expansion cap exceeded");
         ensure!(entries > 0, "Curlie archive has no entries");
         Ok(())
     }

crates/argand-site-registry/src/catalog.rs

@@ -315,15 +315,18 @@ impl Registry {
             },
         )?;
         let mut observations = Vec::new();
+        let mut output_bytes = 0;
         for row in rows {
             let (fact, source, target, value, derived) = row?;
-            observations.push(PopularityObservation {
+            let observation = PopularityObservation {
                 source,
                 target,
                 value: serde_json::from_str(&value)?,
                 domain: serde_json::from_str(&derived)?,
                 provenance: evidence::fact(&self.db, &fact)?,
-            });
+            };
+            crate::query::account_output(&mut output_bytes, &observation)?;
+            observations.push(observation);
         }
         Ok(PopularityLookup {
             input: input.into(),
@@ -349,12 +352,14 @@ impl Registry {
         );
         let mut statement = self.db.prepare("SELECT f.id FROM facts f JOIN selected_sources s ON s.id=f.source_id WHERE f.predicate='category' AND json_extract(f.value,'$.category_id')=?1 ORDER BY f.id")?;
         let mut metadata = Vec::new();
+        let mut output_bytes = 0;
         for id in statement.query_map([category_id], |row| row.get::<_, String>(0))? {
             let mut fact = evidence::fact(&self.db, &id?)?;
             if let Some(value) = fact["value"].as_object_mut() {
                 value.remove("description");
                 value.insert("description_redacted".into(), json!(true));
             }
+            crate::query::account_output(&mut output_bytes, &fact)?;
             metadata.push(fact);
         }
         let total_members = self.db.query_row("SELECT count(*) FROM edges e WHERE e.entity IN(SELECT f.subject FROM facts f JOIN selected_sources s ON s.id=f.source_id WHERE f.predicate='category_membership' AND json_extract(f.value,'$.category_id')=?1)",[category_id],|row|crate::store::unsigned(row,0))?;
@@ -384,10 +389,14 @@ impl Registry {
         let fingerprints = statement
             .query_map(params, |row| row.get::<_, String>(0))?
             .collect::<Result<Vec<_>, _>>()?;
-        fingerprints
+        let mut candidates = Vec::new();
-            .iter()
+        let mut output_bytes = 0;
-            .map(|fingerprint| self.candidate(fingerprint))
+        for fingerprint in fingerprints {
-            .collect()
+            let candidate = self.candidate(&fingerprint)?;
+            crate::query::account_output(&mut output_bytes, &candidate)?;
+            candidates.push(candidate);
+        }
+        Ok(candidates)
     }
 
     fn property_edges<P: rusqlite::Params>(

crates/argand-site-registry/src/cli.rs

@@ -98,6 +98,12 @@ enum Command {
         input: PathBuf,
         #[arg(long)]
         manifest: PathBuf,
+        #[arg(long)]
+        maximum_expanded_bytes: Option<u64>,
+        #[arg(long)]
+        maximum_records: Option<u64>,
+        #[arg(long)]
+        maximum_database_growth_bytes: Option<u64>,
     },
     /// Build a new immutable generation; output must not exist.
     Build {
@@ -260,6 +266,8 @@ enum Command {
         #[arg(long)]
         allowed_signers: PathBuf,
         #[arg(long)]
+        allowed_reviewers: PathBuf,
+        #[arg(long)]
         identity: String,
     },
     /// Run configured acquisition/import/build; never approves or activates.
@@ -269,6 +277,7 @@ enum Command {
     },
 }
 
+#[allow(clippy::too_many_lines)] // One exhaustive command dispatch table is easier to audit.
 pub(super) async fn run() -> anyhow::Result<()> {
     let value = match Args::parse().command {
         command @ (Command::Download { .. }
@@ -287,9 +296,17 @@ pub(super) async fn run() -> anyhow::Result<()> {
             database,
             input,
             manifest,
-        } => {
+            maximum_expanded_bytes,
-            serde_json::json!({"source_id":registry::store::import(&mut registry::store::open(&database)?,&registry::read_json(&manifest)?,&input)?})
+            maximum_records,
-        }
+            maximum_database_growth_bytes,
+        } => import_source(
+            &database,
+            &input,
+            &manifest,
+            maximum_expanded_bytes,
+            maximum_records,
+            maximum_database_growth_bytes,
+        )?,
         Command::Build { database, output } => {
             ensure!(database.is_file(), "import database does not exist");
             let receipt = registry::build::build(&registry::store::open(&database)?, &output)?;
@@ -303,15 +320,15 @@ pub(super) async fn run() -> anyhow::Result<()> {
             signature,
             allowed_reviewers,
             identity,
-        } => {
+        } => record_review(
-            let (review, authentication) = registry::review::authenticate(
+            &database,
-                &decision,
+            &generation,
-                &signature,
+            &pin,
-                &allowed_reviewers,
+            &decision,
-                &identity,
+            &signature,
-            )?;
+            &allowed_reviewers,
-            serde_json::json!({"review_sequence":registry::review::record_authenticated(&registry::store::open(&database)?,&Registry::open(&generation,&pin)?,&review,&authentication)?,"authenticated_reviewer":identity,"rebuild_required":true})
+            &identity,
-        }
+        )?,
         Command::Export {
             generation,
             pin,
@@ -354,11 +371,15 @@ pub(super) async fn run() -> anyhow::Result<()> {
             generation,
             current,
             allowed_signers,
+            allowed_reviewers,
             identity,
-        } => {
+        } => activate_release(
-            registry::release::activate(&generation, &current, &allowed_signers, &identity)?;
+            &generation,
-            serde_json::json!({"current":current})
+            &current,
-        }
+            &allowed_signers,
+            &allowed_reviewers,
+            &identity,
+        )?,
         Command::Update { config } => {
             let config = read_config(&config)?;
             serde_json::json!({"candidate":registry::update::run(&config).await?})
@@ -367,6 +388,74 @@ pub(super) async fn run() -> anyhow::Result<()> {
     write_json(&value)
 }
 
+fn record_review(
+    database: &std::path::Path,
+    generation: &std::path::Path,
+    pin: &str,
+    decision: &std::path::Path,
+    signature: &std::path::Path,
+    allowed_reviewers: &std::path::Path,
+    identity: &str,
+) -> anyhow::Result<serde_json::Value> {
+    let (review, authentication) =
+        registry::review::authenticate(decision, signature, allowed_reviewers, identity)?;
+    let sequence = registry::review::record_authenticated(
+        &registry::store::open(database)?,
+        &Registry::open(generation, pin)?,
+        &review,
+        &authentication,
+    )?;
+    Ok(
+        serde_json::json!({"review_sequence":sequence,"authenticated_reviewer":identity,"rebuild_required":true}),
+    )
+}
+
+fn activate_release(
+    generation: &std::path::Path,
+    current: &std::path::Path,
+    allowed_signers: &std::path::Path,
+    allowed_reviewers: &std::path::Path,
+    identity: &str,
+) -> anyhow::Result<serde_json::Value> {
+    registry::release::activate(
+        generation,
+        current,
+        allowed_signers,
+        identity,
+        allowed_reviewers,
+    )?;
+    Ok(serde_json::json!({"current":current}))
+}
+
+fn import_source(
+    database: &std::path::Path,
+    input: &std::path::Path,
+    manifest_path: &std::path::Path,
+    maximum_expanded_bytes: Option<u64>,
+    maximum_records: Option<u64>,
+    maximum_database_growth_bytes: Option<u64>,
+) -> anyhow::Result<serde_json::Value> {
+    let manifest = registry::read_json(manifest_path)?;
+    let mut limits = registry::store::ImportLimits::for_manifest(&manifest);
+    if let Some(value) = maximum_expanded_bytes {
+        limits.maximum_expanded_bytes = value;
+    }
+    if let Some(value) = maximum_records {
+        limits.maximum_records = value;
+    }
+    if let Some(value) = maximum_database_growth_bytes {
+        limits.maximum_database_growth_bytes = value;
+    }
+    Ok(
+        serde_json::json!({"source_id":registry::store::import_with_limits(
+            &mut registry::store::open(database)?,
+            &manifest,
+            input,
+            limits,
+        )?}),
+    )
+}
+
 fn write_json(value: &serde_json::Value) -> anyhow::Result<()> {
     writeln!(
         std::io::stdout().lock(),

crates/argand-site-registry/src/crux.rs

@@ -41,7 +41,7 @@ pub struct CruxDownload {
 /// Returns authentication, billing, schema, size, expiry, or transport failures.
 pub async fn download(cache: &Path, request: &CruxDownload) -> anyhow::Result<CachedSource> {
     let query = query(request)?;
-    let key = crate::digest(&serde_json::to_vec(request)?);
+    let key = job_key(request)?;
     let root = cache.join("crux").join(&key);
     fs::create_dir_all(&root)?;
     let lock = OpenOptions::new() // atomic-writes: allow advisory lock inode must remain stable
@@ -175,13 +175,14 @@ async fn download_pages(
     job_id: &str,
     part: &Path,
 ) -> anyhow::Result<()> {
+    const HEADER: &[u8] = b"origin,rank,yyyymm,country_code\n";
     let result_url = format!(
         "https://bigquery.googleapis.com/bigquery/v2/projects/{}/queries/{job_id}",
         request.project
     );
     let mut output = File::create(part)?; // atomic-writes: allow unpublished replay file, synced and renamed before manifest publication
-    output.write_all(b"origin,rank,yyyymm,country_code\n")?;
+    output.write_all(HEADER)?;
-    let mut written = 29_u64;
+    let mut written = u64::try_from(HEADER.len())?;
     let mut count = 0_u64;
     let mut page_token = String::new();
     let mut pending = 0_u32;
@@ -280,6 +281,15 @@ fn query(request: &CruxDownload) -> anyhow::Result<String> {
     ))
 }
 
+fn job_key(request: &CruxDownload) -> anyhow::Result<String> {
+    Ok(crate::digest(&serde_json::to_vec(&(
+        &request.project,
+        &request.month,
+        &request.country,
+        request.maximum_bytes_billed,
+    ))?))
+}
+
 async fn bounded_json(mut response: reqwest::Response) -> anyhow::Result<Value> {
     ensure!(
         response.status().is_success(),
@@ -338,6 +348,11 @@ mod tests {
             maximum_output_bytes: 1_000_000,
         };
         let sql = query(&request)?;
+        let original_job = job_key(&request)?;
+        request.maximum_output_bytes += 1;
+        assert_eq!(job_key(&request)?, original_job);
+        request.maximum_bytes_billed += 1;
+        assert_ne!(job_key(&request)?, original_job);
         assert!(sql.contains("`chrome-ux-report.country_gb.202608`"));
         assert!(sql.contains("experimental.popularity.rank"));
         request.maximum_bytes_billed = 0;

crates/argand-site-registry/src/diff.rs

@@ -6,6 +6,35 @@ use rusqlite::{Connection, OptionalExtension};
 use serde_json::{Value, json};
 use std::io::Write;
 
+const MAXIMUM_EVENT_BYTES: usize = 16 * 1024 * 1024;
+const MAXIMUM_OUTPUT_BYTES: u64 = 1024 * 1024 * 1024;
+
+struct Output<'a> {
+    inner: &'a mut dyn Write,
+    written: u64,
+}
+
+impl Output<'_> {
+    fn line(&mut self, value: &Value) -> anyhow::Result<()> {
+        let bytes = serde_json::to_vec(value)?;
+        anyhow::ensure!(
+            bytes.len() <= MAXIMUM_EVENT_BYTES,
+            "diff event exceeds 16 MiB"
+        );
+        self.written = self
+            .written
+            .checked_add(u64::try_from(bytes.len())? + 1)
+            .ok_or_else(|| anyhow::anyhow!("diff output size overflow"))?;
+        anyhow::ensure!(
+            self.written <= MAXIMUM_OUTPUT_BYTES,
+            "diff output exceeds 1 GiB"
+        );
+        self.inner.write_all(&bytes)?;
+        self.inner.write_all(b"\n")?;
+        Ok(())
+    }
+}
+
 struct Table {
     subject: &'static str,
     scan: &'static str,
@@ -70,21 +99,17 @@ const TABLES: &[Table] = &[
 /// # Errors
 /// Returns database, JSON, or output failures.
 pub fn write(old: &Registry, new: &Registry, output: &mut dyn Write) -> anyhow::Result<()> {
-    writeln!(
+    let mut output = Output {
-        output,
+        inner: output,
-        "{}",
+        written: 0,
-        json!({"schema":"argand.site-diff/v2","type":"header","old":old.identity,"new":new.identity})
+    };
-    )?;
+    output.line(&json!({"schema":"argand.site-diff/v3","type":"header","old":old.identity,"new":new.identity,"attribution":crate::release::attribution(),"descriptions_included":false}))?;
     let mut changes = 0_u64;
     for table in TABLES {
-        changes += removed_or_changed(table, &old.db, &new.db, output)?;
+        changes += removed_or_changed(table, &old.db, &new.db, &mut output)?;
-        changes += added(table, &new.db, &old.db, output)?;
+        changes += added(table, &new.db, &old.db, &mut output)?;
     }
-    writeln!(
+    output.line(&json!({"schema":"argand.site-diff/v3","type":"summary","changes":changes}))?;
-        output,
-        "{}",
-        json!({"schema":"argand.site-diff/v2","type":"summary","changes":changes})
-    )?;
     Ok(())
 }
 
@@ -92,7 +117,7 @@ fn removed_or_changed(
     table: &Table,
     from: &Connection,
     to: &Connection,
-    output: &mut dyn Write,
+    output: &mut Output<'_>,
 ) -> anyhow::Result<u64> {
     let mut count = 0;
     let mut scan = from.prepare(table.scan)?;
@@ -129,7 +154,7 @@ fn added(
     table: &Table,
     from: &Connection,
     to: &Connection,
-    output: &mut dyn Write,
+    output: &mut Output<'_>,
 ) -> anyhow::Result<u64> {
     let mut count = 0;
     let mut scan = from.prepare(table.scan)?;
@@ -152,7 +177,7 @@ fn added(
 }
 
 fn event(
-    output: &mut dyn Write,
+    output: &mut Output<'_>,
     subject: &str,
     change: &str,
     key: &str,
@@ -161,14 +186,54 @@ fn event(
 ) -> anyhow::Result<()> {
     let parse = |value: Option<&str>| -> anyhow::Result<Option<Value>> {
         value
-            .map(serde_json::from_str)
+            .map(|raw| {
+                let mut value: Value = serde_json::from_str(raw)?;
+                redact_descriptions(&mut value);
+                Ok(value)
+            })
             .transpose()
-            .map_err(Into::into)
     };
-    writeln!(
+    output.line(&json!({"schema":"argand.site-diff/v3","type":"change","subject":subject,"change":change,"key":key,"before":parse(before)?,"after":parse(after)?}))?;
-        output,
-        "{}",
-        json!({"schema":"argand.site-diff/v2","type":"change","subject":subject,"change":change,"key":key,"before":parse(before)?,"after":parse(after)?})
-    )?;
     Ok(())
 }
+
+fn redact_descriptions(value: &mut Value) {
+    match value {
+        Value::Object(object) => {
+            if object.get("predicate").and_then(Value::as_str) == Some("description") {
+                object.insert("value".into(), Value::Null);
+                object.insert("description_redacted".into(), Value::Bool(true));
+            }
+            if object.remove("description").is_some() {
+                object.insert("description_redacted".into(), Value::Bool(true));
+            }
+            for child in object.values_mut() {
+                redact_descriptions(child);
+            }
+        }
+        Value::Array(values) => {
+            for child in values {
+                redact_descriptions(child);
+            }
+        }
+        _ => {}
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn redacts_description_facts_and_nested_fields() {
+        let mut value = json!({
+            "predicate": "description",
+            "value": {"text": "secret"},
+            "nested": {"description": "also secret"}
+        });
+        redact_descriptions(&mut value);
+        assert!(value["value"].is_null());
+        assert_eq!(value["description_redacted"], true);
+        assert!(value["nested"].get("description").is_none());
+    }
+}

crates/argand-site-registry/src/generation.rs

@@ -13,7 +13,7 @@ use std::{
 };
 
 #[cfg(target_os = "linux")]
-use std::os::{fd::AsRawFd, unix::fs::OpenOptionsExt};
+use std::os::unix::fs::OpenOptionsExt;
 
 impl Registry {
     /// Opens only a complete, externally pinned generation.
@@ -21,6 +21,18 @@ impl Registry {
     /// # Errors
     /// Rejects altered receipts/databases and unsupported contracts.
     pub fn open(path: &Path, expected_pin: &str) -> anyhow::Result<Self> {
+        Self::open_with_rules(path, expected_pin, false)
+    }
+
+    pub(crate) fn open_previous(path: &Path, expected_pin: &str) -> anyhow::Result<Self> {
+        Self::open_with_rules(path, expected_pin, true)
+    }
+
+    fn open_with_rules(
+        path: &Path,
+        expected_pin: &str,
+        allow_legacy_rules: bool,
+    ) -> anyhow::Result<Self> {
         ensure!(
             crate::model::valid_digest(expected_pin),
             "provide a full trusted receipt SHA-256"
@@ -39,9 +51,10 @@ impl Registry {
                 && crate::digest(&attribution) == receipt.attribution_sha256,
             "registry license or attribution digest mismatch"
         );
+        let rules_supported = crate::store::supported_rule_version(&receipt.rules)
+            || (allow_legacy_rules && crate::store::legacy_rule_version(&receipt.rules));
         ensure!(
-            receipt.schema == "argand.site-registry/v1"
+            receipt.schema == "argand.site-registry/v1" && rules_supported,
-                && crate::store::supported_rule_version(&receipt.rules),
             "unsupported registry contract"
         );
         let database = path.join("registry.sqlite");
@@ -127,27 +140,38 @@ pub(crate) fn open_no_follow(_path: &Path) -> anyhow::Result<File> {
 
 #[cfg(target_os = "linux")]
 fn open_authenticated_database(path: &Path, expected: &str) -> anyhow::Result<(Connection, File)> {
-    let mut file = open_no_follow(path)?;
+    let mut source = open_no_follow(path)?;
     ensure!(
-        file.metadata()?.is_file(),
+        source.metadata()?.is_file(),
         "database must be a regular file"
     );
+    // SQLite's immutable mode trusts that its backing inode cannot change. Copy
+    // the authenticated bytes into a mode-0700 private directory, open them,
+    // then unlink the directory so no writer can alter the verified snapshot.
+    let directory = tempfile::tempdir()?;
+    let snapshot = directory.path().join("registry.sqlite");
+    let mut file = OpenOptions::new() // atomic-writes: allow private create-new verified snapshot
+        .read(true)
+        .write(true)
+        .create_new(true)
+        .mode(0o600)
+        .open(&snapshot)?;
     let mut hash = Sha256::new();
     let mut buffer = [0; 8192];
     loop {
-        let count = file.read(&mut buffer)?;
+        let count = source.read(&mut buffer)?;
         if count == 0 {
             break;
         }
         hash.update(&buffer[..count]);
+        std::io::Write::write_all(&mut file, &buffer[..count])?;
     }
     ensure!(
         format!("{:x}", hash.finalize()) == expected,
         "registry database digest mismatch"
     );
     file.rewind()?;
-    let descriptor = format!("/proc/self/fd/{}", file.as_raw_fd());
+    let mut uri = url::Url::from_file_path(&snapshot)
-    let mut uri = url::Url::from_file_path(descriptor)
         .map_err(|()| anyhow::anyhow!("cannot construct immutable SQLite URI"))?;
     uri.query_pairs_mut()
         .append_pair("mode", "ro")
@@ -156,6 +180,7 @@ fn open_authenticated_database(path: &Path, expected: &str) -> anyhow::Result<(C
         | rusqlite::OpenFlags::SQLITE_OPEN_URI
         | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX;
     let db = Connection::open_with_flags(uri.as_str(), flags)?;
+    drop(directory);
     Ok((db, file))
 }
 

crates/argand-site-registry/src/identity.rs

@@ -51,6 +51,7 @@ pub fn propose(registry: &Registry, left: &str, right: &str) -> anyhow::Result<E
         fingerprint: crate::digest(&serde_json::to_vec(&(
             &entities,
             &signatures,
+            &names,
             crate::store::RULE_VERSION,
         ))?),
         entities,

crates/argand-site-registry/src/query.rs

@@ -8,12 +8,25 @@ use serde::Serialize;
 use serde_json::{Value, json};
 use std::fs::File;
 
+const MAXIMUM_QUERY_OUTPUT_BYTES: usize = 64 * 1024 * 1024;
+
+pub(crate) fn account_output<T: Serialize>(used: &mut usize, value: &T) -> anyhow::Result<()> {
+    *used = used
+        .checked_add(serde_json::to_vec(value)?.len())
+        .ok_or_else(|| anyhow::anyhow!("query output size overflow"))?;
+    ensure!(
+        *used <= MAXIMUM_QUERY_OUTPUT_BYTES,
+        "query output exceeds 64 MiB"
+    );
+    Ok(())
+}
+
 pub use crate::resolution::{Resolution, ResolutionCounts, ResolutionStatus};
 
 /// Open verified generation. Hashes are checked once, outside the query path.
 pub struct Registry {
     pub(crate) db: Connection,
-    // Keep the authenticated inode alive for the SQLite /proc/self/fd reader.
+    // Keep the private authenticated snapshot alive with the SQLite reader.
     pub(crate) _database: File,
     /// External receipt pin supplied by the caller.
     pub identity: String,
@@ -162,10 +175,13 @@ impl Registry {
         let ids = stmt
             .query_map(params![key, limit], |r| r.get::<_, String>(0))?
             .collect::<Result<Vec<_>, _>>()?;
-        let candidates = ids
+        let mut candidates = Vec::new();
-            .iter()
+        let mut output_bytes = 0;
-            .map(|id| self.candidate(id))
+        for id in ids {
-            .collect::<anyhow::Result<Vec<_>>>()?;
+            let candidate = self.candidate(&id)?;
+            account_output(&mut output_bytes, &candidate)?;
+            candidates.push(candidate);
+        }
         Ok(Lookup {
             query: key,
             total_entities,
@@ -192,7 +208,7 @@ impl Registry {
             let signer:Option<String>=r.get(10)?;
             Ok(json!({"sequence":sequence,"decision":r.get::<_,String>(1)?,"reviewer":r.get::<_,String>(2)?,"reason":r.get::<_,String>(3)?,"evidence":r.get::<_,String>(4)?,"retrieved_at":r.get::<_,String>(5)?,"expires_at":r.get::<_,String>(6)?,"role":r.get::<_,String>(7)?,"locale":r.get::<_,String>(8)?,"country":r.get::<_,String>(9)?,"authentication":signer.map(|identity|json!({"identity":identity,"signature_sha256":r.get::<_,String>(11).unwrap_or_default(),"namespace":r.get::<_,String>(12).unwrap_or_default()})),"source":"argand_operator_review","source_identifier":format!("{}:{sequence}",fingerprint),"license":"CC0-1.0","license_url":"https://creativecommons.org/publicdomain/zero/1.0/","confidence":9000}))
         }).optional()?;
-        Ok(Candidate {
+        let candidate = Candidate {
             identity_provenance: Vec::new(),
             entity: crate::evidence::entity(&self.db, &entity, &name)?,
             property_scopes: crate::evidence::scopes(
@@ -217,6 +233,9 @@ impl Registry {
             provenance,
             review,
             eligible,
-        })
+        };
+        let mut output_bytes = 0;
+        account_output(&mut output_bytes, &candidate)?;
+        Ok(candidate)
     }
 }

crates/argand-site-registry/src/release.rs

@@ -8,7 +8,6 @@ use std::{
     fs::{self, File, OpenOptions},
     io::{BufWriter, Write},
     path::Path,
-    process::Command,
 };
 
 /// Source terms shipped and authenticated with every generation.
@@ -86,19 +85,16 @@ pub fn sign(
     pin: &str,
     allowed_reviewers: &Path,
 ) -> anyhow::Result<()> {
+    let receipt = crate::ssh::sealed_input(&generation.join("COMPLETE.json"), 1024 * 1024)?;
+    ensure!(crate::digest(&receipt.bytes) == pin, "receipt pin mismatch");
     let registry = Registry::open(generation, pin)?;
     crate::review::verify_all(&registry.db, allowed_reviewers)?;
-    ensure!(
+    crate::ssh::sign(
-        !generation.join("COMPLETE.json.sig").exists(),
+        "argand-site-registry",
-        "signature already exists"
+        &receipt.bytes,
-    );
+        key,
-    let status = Command::new("ssh-keygen")
+        &generation.join("COMPLETE.json.sig"),
-        .args(["-Y", "sign", "-n", "argand-site-registry", "-f"])
+    )?;
-        .arg(key)
-        .arg(generation.join("COMPLETE.json"))
-        .status()?;
-    ensure!(status.success(), "SSH signing failed");
-    File::open(generation.join("COMPLETE.json.sig"))?.sync_all()?;
     File::open(generation)?.sync_all()?;
     Ok(())
 }
@@ -111,6 +107,7 @@ pub fn verify_signed(
     generation: &Path,
     signers: &Path,
     identity: &str,
+    allowed_reviewers: &Path,
 ) -> anyhow::Result<Registry> {
     let receipt = crate::ssh::sealed_input(&generation.join("COMPLETE.json"), 1024 * 1024)?;
     let signature = crate::ssh::sealed_input(&generation.join("COMPLETE.json.sig"), 64 * 1024)?;
@@ -124,7 +121,7 @@ pub fn verify_signed(
         identity,
     )?;
     let registry = Registry::open(generation, &pin)?;
-    crate::review::ensure_all_authenticated(&registry.db)?;
+    crate::review::verify_all(&registry.db, allowed_reviewers)?;
     Ok(registry)
 }
 
@@ -138,8 +135,9 @@ pub fn activate(
     current: &Path,
     signers: &Path,
     identity: &str,
+    allowed_reviewers: &Path,
 ) -> anyhow::Result<()> {
-    let registry = verify_signed(generation, signers, identity)?;
+    let registry = verify_signed(generation, signers, identity, allowed_reviewers)?;
     let parent = current
         .parent()
         .context("current pointer needs a parent directory")?;
@@ -165,7 +163,7 @@ pub fn activate(
                     .context("invalid current pointer")?,
             "rollback would discard revocations; rebuild using the current review log"
         );
-        let old = Registry::open(
+        let old = Registry::open_previous(
             Path::new(
                 previous["generation"]
                     .as_str()

crates/argand-site-registry/src/review.rs

@@ -259,10 +259,6 @@ pub fn verify_all(db: &Connection, allowed_reviewers: &Path) -> anyhow::Result<(
     verify_rows(db, Some(&allowed.bytes))
 }
 
-pub(crate) fn ensure_all_authenticated(db: &Connection) -> anyhow::Result<()> {
-    verify_rows(db, None)
-}
-
 fn verify_rows(db: &Connection, allowed_reviewers: Option<&[u8]>) -> anyhow::Result<()> {
     let missing: u64 = db.query_row(
         "SELECT count(*) FROM reviews r LEFT JOIN review_auth a USING(sequence) WHERE a.sequence IS NULL",
@@ -300,12 +296,13 @@ fn verify_rows(db: &Connection, allowed_reviewers: Option<&[u8]>) -> anyhow::Res
         };
         validate_authentication(&review, &authentication)?;
         if let Some(allowed) = allowed_reviewers {
-            crate::ssh::verify(
+            crate::ssh::verify_at(
                 SIGNATURE_NAMESPACE,
                 &authentication.decision_json,
                 &authentication.signature,
                 allowed,
                 &authentication.signer,
+                Some(review.reviewed_at),
             )?;
         }
     }

crates/argand-site-registry/src/ssh.rs

@@ -35,6 +35,24 @@ pub(crate) fn verify(
     signature: &[u8],
     allowed_signers: &[u8],
     identity: &str,
+) -> anyhow::Result<()> {
+    verify_at(
+        namespace,
+        message,
+        signature,
+        allowed_signers,
+        identity,
+        None,
+    )
+}
+
+pub(crate) fn verify_at(
+    namespace: &str,
+    message: &[u8],
+    signature: &[u8],
+    allowed_signers: &[u8],
+    identity: &str,
+    verify_time: Option<chrono::DateTime<chrono::Utc>>,
 ) -> anyhow::Result<()> {
     // OpenSSH requires paths for its signature and trust file. Private,
     // create-new temporary copies bind the command to the exact bytes already
@@ -49,11 +67,15 @@ pub(crate) fn verify(
     allowed_file.write_all(allowed_signers)?;
     allowed_file.flush()?;
 
-    let status = Command::new("ssh-keygen")
+    let mut command = Command::new("ssh-keygen");
-        .args(["-Y", "verify", "-n", namespace, "-f"])
+    command.args(["-Y", "verify", "-n", namespace, "-f"]);
-        .arg(allowed_file.path())
+    command.arg(allowed_file.path()).arg("-I").arg(identity);
-        .arg("-I")
+    if let Some(time) = verify_time {
-        .arg(identity)
+        command
+            .arg("-O")
+            .arg(format!("verify-time={}", time.format("%Y%m%d%H%M%SZ")));
+    }
+    let status = command
         .arg("-s")
         .arg(signature_file.path())
         .stdin(Stdio::from(message_file.reopen()?))
@@ -62,3 +84,29 @@ pub(crate) fn verify(
     ensure!(status.success(), "untrusted SSH signature");
     Ok(())
 }
+
+pub(crate) fn sign(
+    namespace: &str,
+    message: &[u8],
+    key: &Path,
+    output: &Path,
+) -> anyhow::Result<()> {
+    ensure!(!output.exists(), "signature already exists");
+    let directory = tempfile::tempdir()?;
+    let message_path = directory.path().join("message");
+    let mut message_file = std::fs::OpenOptions::new() // atomic-writes: allow private create-new signing input
+        .write(true)
+        .create_new(true)
+        .open(&message_path)?;
+    message_file.write_all(message)?;
+    message_file.sync_all()?;
+    let status = Command::new("ssh-keygen")
+        .args(["-Y", "sign", "-n", namespace, "-f"])
+        .arg(key)
+        .arg(&message_path)
+        .status()?;
+    ensure!(status.success(), "SSH signing failed");
+    let signature = sealed_input(&message_path.with_extension("sig"), 64 * 1024)?;
+    argand_atomic::create_durable(output, &signature.bytes)?;
+    Ok(())
+}

crates/argand-site-registry/src/store.rs

@@ -7,15 +7,17 @@ use crate::{
 };
 use anyhow::{Context, ensure};
 use rusqlite::{Connection, OptionalExtension, params};
+use sha2::{Digest, Sha256};
 use std::{
-    fs::File,
+    cell::RefCell,
     io::{BufReader, Read},
     path::Path,
+    rc::Rc,
     time::Duration,
 };
 
 /// Adapter/normalization contract recorded in all generation identities.
-pub const RULE_VERSION: &str = "argand.site-rules/v2";
+pub const RULE_VERSION: &str = "argand.site-rules/v3";
 
 /// Whether a signed immutable generation uses a reader-compatible rule contract.
 #[must_use]
@@ -23,6 +25,10 @@ pub fn supported_rule_version(version: &str) -> bool {
     version == RULE_VERSION
 }
 
+pub(crate) fn legacy_rule_version(version: &str) -> bool {
+    matches!(version, "argand.site-rules/v1" | "argand.site-rules/v2")
+}
+
 /// Opens or migrates the local assertion store with bounded page cache.
 ///
 /// # Errors
@@ -39,14 +45,21 @@ pub fn open(path: &Path) -> anyhow::Result<Connection> {
             db.execute_batch("BEGIN IMMEDIATE")?;
             db.execute_batch(include_str!("../migrations/001.sql"))?;
             db.execute_batch(include_str!("../migrations/002.sql"))?;
+            db.execute_batch(include_str!("../migrations/003.sql"))?;
             db.execute_batch("COMMIT")?;
         }
         1 => {
             db.execute_batch("BEGIN IMMEDIATE")?;
             db.execute_batch(include_str!("../migrations/002.sql"))?;
+            db.execute_batch(include_str!("../migrations/003.sql"))?;
             db.execute_batch("COMMIT")?;
         }
-        2 => {}
+        2 => {
+            db.execute_batch("BEGIN IMMEDIATE")?;
+            db.execute_batch(include_str!("../migrations/003.sql"))?;
+            db.execute_batch("COMMIT")?;
+        }
+        3 => {}
         _ => anyhow::bail!("unsupported registry schema {version}"),
     }
     let rules: String = db.query_row(
@@ -67,6 +80,35 @@ pub(crate) fn configure(db: &Connection) -> anyhow::Result<()> {
     Ok(())
 }
 
+/// Hard resource ceilings applied to one source import.
+#[derive(Clone, Copy, Debug)]
+pub struct ImportLimits {
+    /// Maximum bytes produced by the manifest-declared compression layer.
+    pub maximum_expanded_bytes: u64,
+    /// Maximum source records accepted in this snapshot.
+    pub maximum_records: u64,
+    /// Maximum logical SQLite growth caused by this snapshot.
+    pub maximum_database_growth_bytes: u64,
+}
+
+impl ImportLimits {
+    /// Computes finite defaults scaled to the pinned compressed source size.
+    #[must_use]
+    pub fn for_manifest(manifest: &SourceManifest) -> Self {
+        Self {
+            maximum_expanded_bytes: manifest
+                .bytes
+                .saturating_mul(100)
+                .clamp(64 * 1024 * 1024, 1024 * 1024 * 1024 * 1024),
+            maximum_records: 500_000_000,
+            maximum_database_growth_bytes: manifest
+                .bytes
+                .saturating_mul(8)
+                .clamp(1024 * 1024 * 1024, 1024 * 1024 * 1024 * 1024),
+        }
+    }
+}
+
 /// Verifies a source object before and after parsing, publishing completion last.
 /// Incomplete transactions are rolled back; committed batches replay idempotently.
 ///
@@ -76,9 +118,33 @@ pub fn import(
     db: &mut Connection,
     manifest: &SourceManifest,
     path: &Path,
+) -> anyhow::Result<String> {
+    import_with_limits(db, manifest, path, ImportLimits::for_manifest(manifest))
+}
+
+/// Imports with explicit resource ceilings, primarily for constrained operators.
+///
+/// # Errors
+/// Returns source integrity, resource, parser, filesystem, or database errors.
+pub fn import_with_limits(
+    db: &mut Connection,
+    manifest: &SourceManifest,
+    path: &Path,
+    limits: ImportLimits,
 ) -> anyhow::Result<String> {
     manifest.validate()?;
-    verify_input(manifest, path)?;
+    ensure!(
+        limits.maximum_expanded_bytes > 0
+            && limits.maximum_records > 0
+            && limits.maximum_database_growth_bytes > 0,
+        "import limits must be positive"
+    );
+    let mut file = crate::generation::open_no_follow(path)?;
+    let metadata = file.metadata()?;
+    ensure!(
+        metadata.is_file() && metadata.len() == manifest.bytes,
+        "source length/type mismatch"
+    );
     let id = manifest.id()?;
     let declaration = serde_json::to_string(manifest)?;
     db.execute("INSERT OR IGNORE INTO sources(id,source,scope,retrieved_at,manifest) VALUES(?1,?2,?3,?4,?5)", params![id, manifest.source.key(),manifest.scope,manifest.retrieved_at.to_rfc3339(),declaration])?;
@@ -88,14 +154,39 @@ pub fn import(
         |r| Ok((unsigned(r, 0)?, r.get(1)?)),
     )?;
     if complete {
+        let mut hash = Sha256::new();
+        let bytes = std::io::copy(&mut file, &mut HashWriter(&mut hash))?;
+        ensure!(
+            bytes == manifest.bytes && format!("{:x}", hash.finalize()) == manifest.sha256,
+            "source digest mismatch"
+        );
         return Ok(id);
     }
-    let file = File::open(path)?;
+    let observation = Rc::new(RefCell::new(StreamObservation::default()));
-    let reader: Box<dyn Read> = match manifest.compression {
+    let observed = ObservedReader {
-        Compression::None => Box::new(file),
+        inner: file,
-        Compression::Gzip => Box::new(flate2::read::MultiGzDecoder::new(file)),
+        observation: Rc::clone(&observation),
-        Compression::Bzip2 => Box::new(bzip2::read::MultiBzDecoder::new(file)),
     };
+    let reader: Box<dyn Read> = match manifest.compression {
+        Compression::None => Box::new(observed),
+        Compression::Gzip => Box::new(flate2::read::MultiGzDecoder::new(observed)),
+        Compression::Bzip2 => Box::new(bzip2::read::MultiBzDecoder::new(observed)),
+    };
+    let mut reader = BufReader::new(ExpandedReader {
+        inner: reader,
+        read: 0,
+        maximum: limits.maximum_expanded_bytes,
+    });
+    let database_before = database_bytes(db)?;
+    let page_size: u64 = db.query_row("PRAGMA page_size", [], |row| unsigned(row, 0))?;
+    let original_max_pages: u64 =
+        db.query_row("PRAGMA max_page_count", [], |row| unsigned(row, 0))?;
+    let import_max_pages = database_before
+        .saturating_add(limits.maximum_database_growth_bytes)
+        .checked_div(page_size)
+        .context("invalid SQLite page size")?
+        .min(original_max_pages);
+    db.pragma_update(None, "max_page_count", i64::try_from(import_max_pages)?)?;
     db.execute_batch("BEGIN IMMEDIATE")?;
     let result = {
         let mut sink = SqlSink {
@@ -103,15 +194,23 @@ pub fn import(
             source: &id,
             ordinal: 0,
             checkpoint,
+            limits,
+            database_before,
         };
-        let parsed =
+        let parsed = adapters::adapter(manifest.format).ingest(&mut reader, &mut sink);
-            adapters::adapter(manifest.format).ingest(&mut BufReader::new(reader), &mut sink);
         parsed.and_then(|()| {
+            std::io::copy(&mut reader, &mut std::io::sink())?;
             ensure!(
                 sink.ordinal > 0 && sink.ordinal >= checkpoint,
                 "source empty or shorter than checkpoint"
             );
-            verify_input(manifest, path)?;
+            let observed = observation.borrow();
+            ensure!(
+                observed.bytes == manifest.bytes
+                    && format!("{:x}", observed.hash.clone().finalize()) == manifest.sha256,
+                "source digest mismatch"
+            );
+            sink.ensure_database_growth()?;
             sink.db.execute(
                 "UPDATE sources SET checkpoint=?2,complete=1 WHERE id=?1",
                 params![id, i64::try_from(sink.ordinal)?],
@@ -123,35 +222,57 @@ pub fn import(
         Ok(()) => db.execute_batch("COMMIT")?,
         Err(error) => {
             db.execute_batch("ROLLBACK")?;
+            discard_incomplete_source(db, &id)?;
+            db.pragma_update(None, "max_page_count", i64::try_from(original_max_pages)?)?;
             return Err(error);
         }
     }
+    db.pragma_update(None, "max_page_count", i64::try_from(original_max_pages)?)?;
     Ok(id)
 }
 
-fn verify_input(manifest: &SourceManifest, path: &Path) -> anyhow::Result<()> {
+fn discard_incomplete_source(db: &Connection, id: &str) -> anyhow::Result<()> {
-    let metadata = std::fs::symlink_metadata(path)?;
+    db.execute_batch("BEGIN IMMEDIATE")?;
-    ensure!(
+    db.execute("DELETE FROM facts WHERE source_id=?1", [id])?;
-        metadata.is_file() && metadata.len() == manifest.bytes,
+    db.execute("DELETE FROM records WHERE source_id=?1", [id])?;
-        "source length/type mismatch"
+    db.execute("DELETE FROM sources WHERE id=?1 AND complete=0", [id])?;
-    );
+    db.execute_batch("COMMIT")?;
-    ensure!(
-        crate::file_digest(path)? == manifest.sha256,
-        "source digest mismatch"
-    );
     Ok(())
 }
 
+fn database_bytes(db: &Connection) -> anyhow::Result<u64> {
+    let page_count: u64 = db.query_row("PRAGMA page_count", [], |row| unsigned(row, 0))?;
+    let page_size: u64 = db.query_row("PRAGMA page_size", [], |row| unsigned(row, 0))?;
+    Ok(page_count.saturating_mul(page_size))
+}
+
 struct SqlSink<'a> {
     db: &'a Connection,
     source: &'a str,
     ordinal: u64,
     checkpoint: u64,
+    limits: ImportLimits,
+    database_before: u64,
+}
+
+impl SqlSink<'_> {
+    fn ensure_database_growth(&self) -> anyhow::Result<()> {
+        ensure!(
+            database_bytes(self.db)?.saturating_sub(self.database_before)
+                <= self.limits.maximum_database_growth_bytes,
+            "source database growth cap exceeded"
+        );
+        Ok(())
+    }
 }
 
 impl RecordSink for SqlSink<'_> {
     fn emit(&mut self, record: Record) -> anyhow::Result<()> {
         self.ordinal += 1;
+        ensure!(
+            self.ordinal <= self.limits.maximum_records,
+            "source record cap exceeded"
+        );
         if self.ordinal <= self.checkpoint {
             return Ok(());
         }
@@ -184,6 +305,7 @@ impl RecordSink for SqlSink<'_> {
                 ])?;
         }
         if self.ordinal.is_multiple_of(256) {
+            self.ensure_database_growth()?;
             self.db.execute(
                 "UPDATE sources SET checkpoint=?2 WHERE id=?1",
                 params![self.source, i64::try_from(self.ordinal)?],
@@ -194,6 +316,63 @@ impl RecordSink for SqlSink<'_> {
     }
 }
 
+#[derive(Default)]
+struct StreamObservation {
+    hash: Sha256,
+    bytes: u64,
+}
+
+struct ObservedReader<R> {
+    inner: R,
+    observation: Rc<RefCell<StreamObservation>>,
+}
+
+impl<R: Read> Read for ObservedReader<R> {
+    fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
+        let count = self.inner.read(buffer)?;
+        let mut observation = self.observation.borrow_mut();
+        observation.hash.update(&buffer[..count]);
+        observation.bytes = observation.bytes.saturating_add(count as u64);
+        Ok(count)
+    }
+}
+
+struct ExpandedReader {
+    inner: Box<dyn Read>,
+    read: u64,
+    maximum: u64,
+}
+
+struct HashWriter<'a>(&'a mut Sha256);
+
+impl std::io::Write for HashWriter<'_> {
+    fn write(&mut self, buffer: &[u8]) -> std::io::Result<usize> {
+        self.0.update(buffer);
+        Ok(buffer.len())
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+impl Read for ExpandedReader {
+    fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
+        let remaining = self.maximum.saturating_sub(self.read);
+        if remaining == 0 {
+            let mut probe = [0_u8; 1];
+            if self.inner.read(&mut probe)? != 0 {
+                return Err(std::io::Error::other("source expansion cap exceeded"));
+            }
+            return Ok(0);
+        }
+        let bound = usize::try_from(remaining.min(buffer.len() as u64)).unwrap_or(buffer.len());
+        let count = self.inner.read(&mut buffer[..bound])?;
+        self.read = self.read.saturating_add(count as u64);
+        Ok(count)
+    }
+}
+
 /// Returns the manifest of a complete source snapshot.
 ///
 /// # Errors

crates/argand-site-registry/tests/cli.rs

@@ -337,6 +337,8 @@ fn release_lifecycle(
             text(&root.join("current.json"))?,
             "--allowed-signers",
             text(&root.join("allowed_signers"))?,
+            "--allowed-reviewers",
+            text(&root.join("allowed_signers"))?,
             "--identity",
             "fixture"
         ])
@@ -391,6 +393,24 @@ fn sign_and_activate(root: &Path, approved: &Path, approved_pin: &str) -> anyhow
             text(&current)?,
             "--allowed-signers",
             text(&allowed)?,
+            "--allowed-reviewers",
+            text(&wrong_reviewers)?,
+            "--identity",
+            "fixture"
+        ])
+        .is_err()
+    );
+    assert!(
+        run(&[
+            "activate",
+            "--generation",
+            text(approved)?,
+            "--current",
+            text(&current)?,
+            "--allowed-signers",
+            text(&allowed)?,
+            "--allowed-reviewers",
+            text(&allowed)?,
             "--identity",
             "untrusted"
         ])
@@ -404,6 +424,8 @@ fn sign_and_activate(root: &Path, approved: &Path, approved_pin: &str) -> anyhow
         text(&current)?,
         "--allowed-signers",
         text(&allowed)?,
+        "--allowed-reviewers",
+        text(&allowed)?,
         "--identity",
         "fixture",
     ])?;

crates/argand-site-registry/tests/failures.rs

@@ -6,7 +6,7 @@ use argand_site_registry::{
     download::CachedSource,
     model::{Compression, Format, Source},
     query::Registry,
-    store,
+    store::{self, ImportLimits},
 };
 use serde_json::json;
 use std::{fs, io::Write, process::Command};
@@ -23,7 +23,7 @@ fn version_one_store_migrates_without_losing_review_history() -> anyhow::Result<
     let migrated = store::open(&path)?;
     assert_eq!(
         migrated.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))?,
-        2
+        3
     );
     assert_eq!(
         migrated.query_row("SELECT rules FROM registry_metadata", [], |row| row
@@ -83,7 +83,8 @@ fn externally_signed_unauthenticated_reviews_are_rejected() -> anyhow::Result<()
         argand_site_registry::release::verify_signed(
             &root.path().join("generation"),
             &root.path().join("allowed"),
-            "fixture"
+            "fixture",
+            &root.path().join("allowed")
         )
         .is_err()
     );
@@ -123,6 +124,168 @@ fn unsigned_sqlite_sidecars_and_generation_symlinks_are_rejected() -> anyhow::Re
     Ok(())
 }
 
+#[test]
+fn opened_registry_uses_private_authenticated_database_bytes() -> anyhow::Result<()> {
+    let root = tempfile::tempdir()?;
+    let db = common::fixture(root.path())?;
+    let registry = common::build(&db, root.path(), "private-snapshot")?;
+    let mutable = rusqlite::Connection::open(root.path().join("private-snapshot/registry.sqlite"))?;
+    mutable.execute(
+        "UPDATE properties SET url='https://changed.invalid/' WHERE url='https://facebook.com/'",
+        [],
+    )?;
+    assert_eq!(
+        registry.lookup("FB", 1)?.candidates[0].url,
+        "https://facebook.com/"
+    );
+    Ok(())
+}
+
+#[test]
+fn import_limits_rollback_all_partial_source_rows() -> anyhow::Result<()> {
+    let root = tempfile::tempdir()?;
+    let mut db = store::open(&root.path().join("limited.sqlite"))?;
+    let compressed = common::gzip(common::CRUX.as_bytes())?;
+    let input = root.path().join("limited.csv.gz");
+    fs::write(&input, &compressed)?;
+    let mut manifest = common::manifest(Source::Crux, Format::CruxCsv, &compressed)?;
+    manifest.compression = Compression::Gzip;
+    assert!(
+        store::import_with_limits(
+            &mut db,
+            &manifest,
+            &input,
+            ImportLimits {
+                maximum_expanded_bytes: 16,
+                maximum_records: 100,
+                maximum_database_growth_bytes: 1024 * 1024,
+            },
+        )
+        .is_err()
+    );
+    assert_eq!(
+        db.query_row("SELECT count(*) FROM sources", [], |row| row
+            .get::<_, i64>(0))?,
+        0
+    );
+    Ok(())
+}
+
+#[test]
+fn retired_reviewer_keys_verify_history_only_with_a_validity_epoch() -> anyhow::Result<()> {
+    let root = tempfile::tempdir()?;
+    let db = common::fixture(root.path())?;
+    let generation = common::build(&db, root.path(), "reviewer-epoch")?;
+    let fingerprint = generation.lookup("FB", 1)?.candidates[0]
+        .fingerprint
+        .clone();
+    let key = root.path().join("reviewer");
+    assert!(
+        Command::new("ssh-keygen")
+            .args(["-q", "-t", "ed25519", "-N", "", "-f"])
+            .arg(&key)
+            .status()?
+            .success()
+    );
+    let decision = root.path().join("decision.json");
+    fs::write(
+        &decision,
+        serde_json::to_vec(&json!({
+            "fingerprint":fingerprint,"decision":"revoke","reviewer":"retired",
+            "reason":"synthetic key epoch test","evidence":"synthetic:key-epoch",
+            "reviewed_at":"2026-09-12T12:00:00Z","expires_at":"2026-09-12T12:00:00Z",
+            "role":"unspecified","locale":"","country":""
+        }))?,
+    )?;
+    assert!(
+        Command::new("ssh-keygen")
+            .args(["-Y", "sign", "-n", "argand-site-registry-review", "-f"])
+            .arg(&key)
+            .arg(&decision)
+            .status()?
+            .success()
+    );
+    let public = fs::read_to_string(key.with_extension("pub"))?;
+    let allowed = root.path().join("allowed");
+    fs::write(&allowed, format!("retired {public}"))?;
+    let (review, authentication) = argand_site_registry::review::authenticate(
+        &decision,
+        &decision.with_extension("json.sig"),
+        &allowed,
+        "retired",
+    )?;
+    argand_site_registry::review::record_authenticated(&db, &generation, &review, &authentication)?;
+    fs::write(
+        &allowed,
+        format!("retired valid-before=\"20260913000000Z\" {public}"),
+    )?;
+    assert!(argand_site_registry::review::verify_all(&db, &allowed).is_ok());
+    fs::write(
+        &allowed,
+        format!("retired valid-before=\"20260912000000Z\" {public}"),
+    )?;
+    assert!(argand_site_registry::review::verify_all(&db, &allowed).is_err());
+    Ok(())
+}
+
+#[test]
+fn activation_accepts_a_pinned_v1_previous_generation() -> anyhow::Result<()> {
+    let root = tempfile::tempdir()?;
+    let db = common::fixture(root.path())?;
+    let old = common::build(&db, root.path(), "legacy-current")?;
+    drop(old);
+    let receipt_path = root.path().join("legacy-current/COMPLETE.json");
+    let mut receipt: serde_json::Value = argand_site_registry::read_json(&receipt_path)?;
+    receipt["rules"] = json!("argand.site-rules/v1");
+    fs::write(&receipt_path, serde_json::to_vec_pretty(&receipt)?)?;
+    let old_pin = argand_site_registry::file_digest(&receipt_path)?;
+    let current = root.path().join("current.json");
+    fs::write(
+        &current,
+        serde_json::to_vec_pretty(&json!({
+            "schema":"argand.site-current/v1",
+            "generation":root.path().join("legacy-current").canonicalize()?,
+            "receipt_sha256":old_pin,
+            "signer":"legacy-publisher",
+            "revocation_sequence":0
+        }))?,
+    )?;
+
+    let new = common::build(&db, root.path(), "current-rules")?;
+    let key = root.path().join("publisher");
+    assert!(
+        Command::new("ssh-keygen")
+            .args(["-q", "-t", "ed25519", "-N", "", "-f"])
+            .arg(&key)
+            .status()?
+            .success()
+    );
+    let allowed = root.path().join("allowed-publisher");
+    fs::write(
+        &allowed,
+        format!(
+            "publisher {}",
+            fs::read_to_string(key.with_extension("pub"))?
+        ),
+    )?;
+    argand_site_registry::release::sign(
+        &root.path().join("current-rules"),
+        &key,
+        &new.identity,
+        &allowed,
+    )?;
+    argand_site_registry::release::activate(
+        &root.path().join("current-rules"),
+        &current,
+        &allowed,
+        "publisher",
+        &allowed,
+    )?;
+    let activated: serde_json::Value = argand_site_registry::read_json(&current)?;
+    assert_eq!(activated["receipt_sha256"], new.identity);
+    Ok(())
+}
+
 #[test]
 fn removed_entity_websites_retire_the_previous_selection() -> anyhow::Result<()> {
     let root = tempfile::tempdir()?;
@@ -263,6 +426,8 @@ fn typed_diff_reports_review_only_changes() -> anyhow::Result<()> {
     assert!(rows.iter().any(|row| {
         row["type"] == "change" && row["subject"] == "review" && row["change"] == "added"
     }));
+    assert_eq!(rows[0]["descriptions_included"], false);
+    assert!(rows[0]["attribution"]["curlie"].is_object());
     assert_eq!(rows.last().and_then(|row| row["changes"].as_u64()), Some(1));
     Ok(())
 }

crates/argand-site-registry/tests/identity.rs

@@ -171,3 +171,28 @@ fn name_collision_remains_ambiguous_until_review_and_changes_invalidate_link() -
     assert!(review::record(&db, &baseline, &tampered).is_err());
     Ok(())
 }
+
+#[test]
+fn metadata_changes_invalidate_identity_fingerprint() -> anyhow::Result<()> {
+    let root = tempfile::tempdir()?;
+    let mut db = common::fixture(root.path())?;
+    let baseline = common::build(&db, root.path(), "metadata-baseline")?;
+    let wiki = baseline.lookup("FB", 1)?.candidates[0].entity_id.clone();
+    let curlie = baseline.lookup("Facebook directory listing", 1)?.candidates[0]
+        .entity_id
+        .clone();
+    let before = identity::propose(&baseline, &wiki, &curlie)?;
+
+    let mut entity = common::entity("Q355", "Facebook", &["FB"], &["https://facebook.com/"]);
+    entity["claims"]["P17"] = json!([{"id":"Q355$country","rank":"normal","mainsnak":{"property":"P17","snaktype":"value","datavalue":{"type":"wikibase-entityid","value":{"id":"Q30"}}}}]);
+    let bytes = serde_json::to_vec(&json!({"entities":{"Q355":entity}}))?;
+    let mut source = common::manifest(Source::Wikidata, Format::WikidataEntities, &bytes)?;
+    source.retrieved_at += chrono::Duration::hours(1);
+    let input = root.path().join("metadata-change.json");
+    std::fs::write(&input, &bytes)?;
+    store::import(&mut db, &source, &input)?;
+    let changed = common::build(&db, root.path(), "metadata-changed")?;
+    let after = identity::propose(&changed, &wiki, &curlie)?;
+    assert_ne!(before.fingerprint, after.fingerprint);
+    Ok(())
+}

crates/argand-site-registry/tests/registry.rs

@@ -324,11 +324,12 @@ fn compressed_dumps_resume_and_corruption_fail_closed() -> anyhow::Result<()> {
     m.compression = Compression::Gzip;
     let path = dir.path().join("dump.gz");
     std::fs::write(&path, &bytes)?;
-    // Interrupt the sink after its first committed batch; rerun the exact input.
+    // A handled sink error removes committed rows whose complete stream digest
+    // was not established. A process crash still leaves the durable checkpoint.
     db.execute_batch("CREATE TRIGGER simulated_crash BEFORE INSERT ON records WHEN NEW.ordinal=270 BEGIN SELECT RAISE(ABORT,'simulated interruption'); END;")?;
     assert!(store::import(&mut db, &m, &path).is_err());
-    let checkpoint: i64 = db.query_row("SELECT checkpoint FROM sources", [], |r| r.get(0))?;
+    let partial: i64 = db.query_row("SELECT count(*) FROM sources", [], |r| r.get(0))?;
-    assert_eq!(checkpoint, 256);
+    assert_eq!(partial, 0);
     db.execute_batch("DROP TRIGGER simulated_crash")?;
     store::import(&mut db, &m, &path)?;
     let records: i64 = db.query_row("SELECT count(*) FROM records", [], |r| r.get(0))?;
@@ -340,12 +341,12 @@ fn compressed_dumps_resume_and_corruption_fail_closed() -> anyhow::Result<()> {
     let mut truncated = common::manifest(Source::Wikidata, Format::WikidataDump, &damaged)?;
     truncated.compression = Compression::Gzip;
     assert!(store::import(&mut db, &truncated, &path).is_err());
-    let complete: i64 = db.query_row(
+    let retained: i64 = db.query_row(
-        "SELECT complete FROM sources WHERE id=?1",
+        "SELECT count(*) FROM sources WHERE id=?1",
         [truncated.id()?],
         |r| r.get(0),
     )?;
-    assert_eq!(complete, 0);
+    assert_eq!(retained, 0);
     Ok(())
 }
 

docs/CONSUMERS.md

@@ -2,9 +2,11 @@
 
 ## Rust library
 
-Use `argand_site_registry::query::Registry::open(generation, trusted_pin)` once per
+Use `release::verify_signed(generation, publisher_signers, publisher_identity,
-immutable generation and reuse the reader. `lookup(query, limit)` returns evidence
+reviewer_signers)` once per production generation and reuse the returned reader.
-and complete ambiguity counts; `resolve_explained` returns a reviewed candidate or
+`Registry::open(generation, trusted_pin)` is the lower-level path when the pin
+distributor is also trusted for the complete review decision. `lookup(query,
+limit)` returns evidence and complete ambiguity counts; `resolve_explained` returns a reviewed candidate or
 a typed abstention reason with counts. Exact reverse views cover entity IDs,
 URLs/domains, popularity and Curlie categories. Check the compiled example and API
 docs for exact types. `selection_context` binds the full alternative set for
@@ -39,9 +41,10 @@ omits descriptions and includes fact provenance plus an attribution envelope.
 It is an assertion export, not a self-contained signed list of admitted routes.
 Raw SQL inspection is useful for audit; it does not implement resolution policy.
 
-Code version 0.2.0 uses schema version 2 and `argand.site-rules/v2`.
+Code version 0.3.0 uses schema version 3 and `argand.site-rules/v3`.
-It adds authenticated reviewer proofs, exact immutable SQLite opening, typed diffs
+It adds reviewer-trust enforcement for consumers, private authenticated SQLite
-and audit/evaluation APIs. Schema/rule contracts remain versioned independently in
+snapshots, exact-stream import checks, bounded outputs and metadata-bound identity
+decisions. Schema/rule contracts remain versioned independently in
 receipts. Unsupported contracts fail closed. Pin source releases,
 compile consumers and replay fixed fixtures before upgrades. Preserve import and
 review history; never mutate complete generations to migrate them.

docs/PUBLISHING.md

@@ -14,6 +14,12 @@ one accepted principal and public key per line:
 operator@example.org ssh-ed25519 REVIEWER_PUBLIC_KEY
 ```
 
+For rotation, retain an old public key with an OpenSSH `valid-before` option
+covering its signed decision times. New decisions are checked at append time, so
+an expired key cannot submit backdated reviews; historical release verification
+uses each authenticated `reviewed_at`. Remove a retired key only after no retained
+generation or review log depends on it.
+
 Distribute the release publisher public key to consumers through an independent
 authenticated channel. Do not put private keys, production trust files or source
 datasets in Git or CI. The isolated source CI runner has none of these files.
@@ -62,6 +68,7 @@ argand-site-registry sign --generation /data/registry/reviewed \
 argand-site-registry activate --generation /data/registry/reviewed \
   --current /data/registry/current.json \
   --allowed-signers /secure/release-allowed-signers \
+  --allowed-reviewers /secure/reviewer-allowed-signers \
   --identity registry-publisher
 ```
 

docs/TRUST.md

@@ -18,9 +18,11 @@ writer retains their exact decision/signature bytes in the append-only review lo
 Generations bind the database, license document and attribution to a completion
 receipt. Consumers provide a trusted hash or verify an external publisher key.
 Release signing re-verifies every stored decision against an external reviewer
-trust file. Activation checks publisher signatures, rejects structurally incomplete
+trust file. Activation checks publisher signatures, re-verifies every review
-review proofs and refuses rollback that loses distributed revocations. Updates
+against a separately supplied reviewer trust file and refuses rollback that loses
-build candidates and cannot approve, sign or activate them.
+distributed revocations. Reviewer validity epochs are evaluated at decision time
+for retained history while new decisions must pass the trust policy at append
+time. Updates build candidates and cannot approve, sign or activate them.
 
 ## What a publisher must establish
 
@@ -47,8 +49,11 @@ enforced quorum.
 
 ## What consumers must preserve
 
-Authenticate a release before opening it. Keep the full receipt pin and required
+Authenticate a release before opening it. Consumers that rely on reviewer
-attribution with caches and exports. Use `resolve` for reviewed destinations, keep
+separation must use `release::verify_signed` or `activate` with independently
+distributed publisher and reviewer trust files; a receipt pin alone delegates the
+whole release decision to whoever distributed that pin. Keep the full receipt pin
+and required attribution with caches and exports. Use `resolve` for reviewed destinations, keep
 null as abstention, and enforce application-specific malware/content/navigation
 policy separately. A verified signature authenticates the publisher, not the truth
 of every assertion. An official website may later be compromised.