feat(storage): add BlobReferenceSource port + registry

Step 1 of docs/plan/derived-blobs.md. Makes "who references this blob
hash" an extension point instead of SQL hardcoded in two places
(dedup_gc's reap predicate and blobs_consistency's refcount recompute,
both naming storage.files and storage.chunk_manifests directly). Adding
a blob-owning table without teaching those two risks silent orphaning:
GC sees ref_count = 0 and reaps live content.

Behaviour is unchanged — this commit only introduces the port and the
two sources that reproduce today's SQL. Wiring follows.

Two levels, not one. add_reference bumps chunk_manifests.ref_count first
and only falls back to storage.blobs.ref_count, so a reference lands on
whichever counter its hash names and the two must be recomputed
separately. RefLevel is a parameter rather than a property of a source,
because storage.files legitimately contributes at both: a manifest-less
legacy row references a chunk, a CDC row references a Blob. The
NOT EXISTS guard on the chunk-level files term is load-bearing — for a
single-chunk file the whole-file hash equals its lone chunk's hash, so
without it the row is counted at both levels.

SQL fragments rather than a per-hash count. blobs_consistency recomputes
with one query per page, the expected count inlined as correlated
subqueries; asking each source for a count per hash would turn that into
sources x rows round-trips. So sources emit a fragment the registry sums
into the existing page query, and count_references exists only for the
on-demand path where the candidate set is already filtered to
ref_count = 0.

Fragments use their own aliases (cnt_f, cnt_m) rather than the sweeps'
outer-row aliases (b, m). A fragment reusing `m` would shadow the outer
alias in the manifest sweep and silently correlate against itself;
there is a test for it.

The SQL builders are free functions so the shape can be asserted without
constructing a pool — sqlx's connect_lazy still needs a Tokio context,
and the fragments are pure string assembly anyway.

9 unit tests. fmt, clippy --all-features --all-targets, build clean.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/application/ports/blob_reference_ports.rs

@@ -0,0 +1,278 @@
+//! `BlobReferenceSource` — the extension point that teaches ref-counting
+//! and the consistency jobs about a table holding blob references.
+//!
+//! Before this port, "who references this hash" was hardcoded SQL in two
+//! places (`dedup_gc`'s reap predicate and `blobs_consistency`'s refcount
+//! recompute), both naming `storage.files` and `storage.chunk_manifests`
+//! directly. Any new blob-owning table therefore risked silent orphaning:
+//! `dedup_gc` sees `ref_count = 0`, or a manifest with no `storage.files`
+//! row behind it, and reaps live content.
+//!
+//! See `docs/plan/derived-blobs.md` for the design and the coverage matrix.
+//!
+//! # Two levels, and why a source may span both
+//!
+//! [`DedupService::add_reference`] bumps `chunk_manifests.ref_count` first
+//! and only falls back to `storage.blobs.ref_count`. So a reference lands
+//! on whichever counter its hash names, and the two must be recomputed
+//! separately — mixing them double-counts, systematically:
+//!
+//! * A **Blob** (`chunk_manifests.file_hash`) is "the content of a file".
+//! * A **Chunk** (`storage.blobs.hash`) is a physical byte payload.
+//! * For a single-chunk Blob the two hashes are **equal**, because both are
+//!   BLAKE3 over the same bytes. That aliasing is why today's chunk-level
+//!   recompute carries a `NOT EXISTS` clause, and why every fragment here
+//!   must be level-correct rather than merely plausible.
+//!
+//! A source is not confined to one level: [`RefLevel::Chunk`] and
+//! [`RefLevel::Manifest`] fragments are requested independently, and
+//! `storage.files` legitimately contributes to both — a manifest-less
+//! legacy row references a chunk, a CDC row references a Blob.
+//!
+//! # Why SQL fragments rather than a per-hash count
+//!
+//! `blobs_consistency` recomputes refcounts with **one query per page**,
+//! the expected count inlined as correlated subqueries. Asking each source
+//! for a count per hash would turn that into `sources × rows` round-trips —
+//! a catastrophic regression on a table with millions of rows. So sources
+//! contribute a *fragment* that the registry sums into the existing page
+//! query, and [`BlobReferenceSource::count_references`] exists only for the
+//! on-demand path (`dedup_gc` checking a single reap candidate, where the
+//! candidate set is already filtered to `ref_count = 0`).
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+
+use crate::domain::errors::DomainError;
+
+/// Which counter a source's references land on.
+///
+/// Not a property of the source — see the module docs; the same source may
+/// contribute at both levels.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum RefLevel {
+    /// References a physical chunk. Feeds `storage.blobs.ref_count`.
+    Chunk,
+    /// References a Blob via its manifest. Feeds
+    /// `chunk_manifests.ref_count`.
+    Manifest,
+}
+
+impl RefLevel {
+    /// Both levels, for callers that sweep each in turn.
+    pub const ALL: [RefLevel; 2] = [RefLevel::Chunk, RefLevel::Manifest];
+
+    /// Stable name for logs and consistency-finding fields.
+    pub fn as_str(self) -> &'static str {
+        match self {
+            RefLevel::Chunk => "chunk",
+            RefLevel::Manifest => "manifest",
+        }
+    }
+}
+
+/// One table that holds references to blob hashes.
+///
+/// Implementors are registered on [`BlobReferenceRegistry`] during DI.
+/// Adding a blob-owning table **without** registering it is the failure
+/// this port exists to prevent.
+#[async_trait]
+pub trait BlobReferenceSource: Send + Sync {
+    /// Short stable identifier for logs and consistency-finding `source`
+    /// fields — `"files"`, `"chunks"`, `"content_derived"`, …
+    ///
+    /// Stable across releases: log aggregators key off it.
+    fn source_name(&self) -> &'static str;
+
+    /// A correlated-subquery fragment counting this source's references
+    /// **at `level`** to `outer_hash_expr`, or `None` when this source
+    /// holds no references at that level.
+    ///
+    /// `outer_hash_expr` is the SQL expression naming the hash of the row
+    /// being recomputed — `"b.hash"` when sweeping `storage.blobs`,
+    /// `"m.file_hash"` when sweeping `storage.chunk_manifests`. The
+    /// fragment must be a parenthesised scalar subquery so the registry can
+    /// join fragments with `+`.
+    ///
+    /// **Identifiers only.** `outer_hash_expr` is supplied by the sweep, never
+    /// by a request; no fragment may interpolate caller input.
+    fn ref_count_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String>;
+
+    /// Count of references this source holds on `blob_hash`, across both
+    /// levels.
+    ///
+    /// **On-demand path only** — `dedup_gc` checking a single reap
+    /// candidate. The consistency sweeps must use [`Self::ref_count_sql`];
+    /// calling this per row would turn one query per page into
+    /// `sources × rows` round-trips.
+    async fn count_references(&self, blob_hash: &str) -> Result<u64, DomainError>;
+
+    /// Iterate the hashes this source references, paged by the
+    /// implementation's natural cursor (typically a primary key).
+    ///
+    /// Used by `backend_consistency` to walk the backend against the union
+    /// of all sources. Returns the page plus the cursor to resume from,
+    /// `None` when exhausted.
+    async fn list_referenced_blobs(
+        &self,
+        cursor: Option<Vec<u8>>,
+        limit: usize,
+    ) -> Result<(Vec<String>, Option<Vec<u8>>), DomainError>;
+
+    /// Notification that `dedup_gc` reaped this blob.
+    ///
+    /// Sources maintaining a denormalised refcount can clean up here. Most
+    /// leave the default noop — the mapping row is normally deleted by the
+    /// owning service's `on_blob_deleted` hook instead.
+    fn on_blob_reaped(&self, _blob_hash: &str) {}
+}
+
+/// The set of registered [`BlobReferenceSource`]s.
+///
+/// Assembled once during DI and shared (`Arc`) by `dedup_gc` and the
+/// consistency jobs, so all three agree on what "referenced" means.
+#[derive(Default)]
+pub struct BlobReferenceRegistry {
+    sources: Vec<Arc<dyn BlobReferenceSource>>,
+}
+
+impl BlobReferenceRegistry {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Register a source. Order is irrelevant — fragments are summed and
+    /// counts added.
+    pub fn register(&mut self, source: Arc<dyn BlobReferenceSource>) {
+        self.sources.push(source);
+    }
+
+    pub fn sources(&self) -> &[Arc<dyn BlobReferenceSource>] {
+        &self.sources
+    }
+
+    /// The summed SQL expression counting every source's references at
+    /// `level` to `outer_hash_expr`.
+    ///
+    /// Returns `"0"` when no source contributes at this level, which keeps
+    /// the caller's query valid without a special case.
+    pub fn ref_count_expr(&self, level: RefLevel, outer_hash_expr: &str) -> String {
+        let fragments: Vec<String> = self
+            .sources
+            .iter()
+            .filter_map(|s| s.ref_count_sql(level, outer_hash_expr))
+            .collect();
+
+        if fragments.is_empty() {
+            "0".to_string()
+        } else {
+            fragments.join("\n + ")
+        }
+    }
+
+    /// Total references held on `hash` across every source.
+    ///
+    /// On-demand path only — see [`BlobReferenceSource::count_references`].
+    pub async fn total_references(&self, hash: &str) -> Result<u64, DomainError> {
+        let mut total = 0u64;
+        for source in &self.sources {
+            total = total.saturating_add(source.count_references(hash).await?);
+        }
+        Ok(total)
+    }
+
+    /// Fan out a reap notification to every source.
+    pub fn notify_reaped(&self, blob_hash: &str) {
+        for source in &self.sources {
+            source.on_blob_reaped(blob_hash);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    struct Stub {
+        name: &'static str,
+        chunk: Option<&'static str>,
+        manifest: Option<&'static str>,
+        count: u64,
+    }
+
+    #[async_trait]
+    impl BlobReferenceSource for Stub {
+        fn source_name(&self) -> &'static str {
+            self.name
+        }
+
+        fn ref_count_sql(&self, level: RefLevel, outer: &str) -> Option<String> {
+            let tmpl = match level {
+                RefLevel::Chunk => self.chunk?,
+                RefLevel::Manifest => self.manifest?,
+            };
+            Some(tmpl.replace("{outer}", outer))
+        }
+
+        async fn count_references(&self, _blob_hash: &str) -> Result<u64, DomainError> {
+            Ok(self.count)
+        }
+
+        async fn list_referenced_blobs(
+            &self,
+            _cursor: Option<Vec<u8>>,
+            _limit: usize,
+        ) -> Result<(Vec<String>, Option<Vec<u8>>), DomainError> {
+            Ok((Vec::new(), None))
+        }
+    }
+
+    fn registry() -> BlobReferenceRegistry {
+        let mut r = BlobReferenceRegistry::new();
+        r.register(Arc::new(Stub {
+            name: "a",
+            chunk: Some("(SELECT 1 WHERE {outer} = 'x')"),
+            manifest: None,
+            count: 2,
+        }));
+        r.register(Arc::new(Stub {
+            name: "b",
+            chunk: Some("(SELECT 2 WHERE {outer} = 'y')"),
+            manifest: Some("(SELECT 3 WHERE {outer} = 'z')"),
+            count: 5,
+        }));
+        r
+    }
+
+    #[test]
+    fn chunk_level_sums_every_contributing_source() {
+        let expr = registry().ref_count_expr(RefLevel::Chunk, "b.hash");
+        assert!(expr.contains("b.hash = 'x'"), "{expr}");
+        assert!(expr.contains("b.hash = 'y'"), "{expr}");
+        assert!(expr.contains('+'), "fragments must be summed: {expr}");
+    }
+
+    /// A source returning `None` for a level must contribute nothing there —
+    /// this is what keeps manifest-only tables out of the chunk recompute,
+    /// where they would double-count against the single-chunk hash alias.
+    #[test]
+    fn manifest_level_skips_non_contributing_sources() {
+        let expr = registry().ref_count_expr(RefLevel::Manifest, "m.file_hash");
+        assert!(expr.contains("m.file_hash = 'z'"), "{expr}");
+        assert!(!expr.contains('+'), "only one source contributes: {expr}");
+    }
+
+    /// An empty level must still yield a valid scalar expression, so callers
+    /// need no special case before a registry is fully populated.
+    #[test]
+    fn empty_level_yields_zero_literal() {
+        let r = BlobReferenceRegistry::new();
+        assert_eq!(r.ref_count_expr(RefLevel::Chunk, "b.hash"), "0");
+    }
+
+    #[tokio::test]
+    async fn total_references_adds_across_sources() {
+        assert_eq!(registry().total_references("deadbeef").await.unwrap(), 7);
+    }
+}

src/application/ports/mod.rs

@@ -1,6 +1,7 @@
 pub mod auth_ports;
 pub mod authorization_ports;
 pub mod blob_lifecycle;
+pub mod blob_reference_ports;
 pub mod blob_storage_ports;
 pub mod cache_ports;
 pub mod calendar_ports;

src/infrastructure/repositories/pg/blob_reference_sources.rs

@@ -0,0 +1,328 @@
+//! The two implicit blob-reference sources, made explicit.
+//!
+//! Before this module, "who references this hash" lived as hardcoded SQL
+//! inside `blobs_consistency`'s refcount recompute and `dedup_gc`'s reap
+//! predicate. These two implementations reproduce that SQL **exactly** —
+//! the fragments below sum to today's `actual_ref_count` expression — so
+//! the registry can be wired in without changing any observed count.
+//!
+//! See `docs/plan/derived-blobs.md` and
+//! [`crate::application::ports::blob_reference_ports`].
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use sqlx::{PgPool, Row};
+use uuid::Uuid;
+
+use crate::application::ports::blob_reference_ports::{BlobReferenceSource, RefLevel};
+use crate::domain::errors::DomainError;
+
+/// Aliases used inside the emitted fragments.
+///
+/// Deliberately distinct from the aliases the sweeps use for their outer
+/// row (`b` for `storage.blobs`, `m` for `storage.chunk_manifests`): a
+/// fragment reusing `m` would shadow the outer alias in the manifest-level
+/// sweep and silently correlate against itself.
+const FILES_ALIAS: &str = "cnt_f";
+const MANIFEST_ALIAS: &str = "cnt_m";
+
+/// Fragment for [`FilesReferenceSource`], as a free function so the SQL
+/// shape can be tested without constructing a pool — it is a property of
+/// the module, not of an instance.
+fn files_ref_sql(level: RefLevel, outer_hash_expr: &str) -> Option<String> {
+    let f = FILES_ALIAS;
+    match level {
+        // Legacy whole-file blobs only — CDC files are counted at the
+        // manifest level, and counting them here too would double up on the
+        // single-chunk hash alias.
+        RefLevel::Chunk => Some(format!(
+            "(SELECT COUNT(*) FROM storage.files {f}
+               WHERE {f}.blob_hash = {outer_hash_expr}
+                 AND NOT EXISTS (
+                     SELECT 1 FROM storage.chunk_manifests {MANIFEST_ALIAS}
+                      WHERE {MANIFEST_ALIAS}.file_hash = {f}.blob_hash
+                 ))"
+        )),
+        RefLevel::Manifest => Some(format!(
+            "(SELECT COUNT(*) FROM storage.files {f}
+               WHERE {f}.blob_hash = {outer_hash_expr})"
+        )),
+    }
+}
+
+/// Fragment for [`ChunksReferenceSource`]. See [`files_ref_sql`].
+fn chunks_ref_sql(level: RefLevel, outer_hash_expr: &str) -> Option<String> {
+    match level {
+        RefLevel::Chunk => {
+            let m = MANIFEST_ALIAS;
+            Some(format!(
+                "(SELECT COUNT(*) FROM storage.chunk_manifests {m}
+                   WHERE {outer_hash_expr} = ANY({m}.chunk_hashes))"
+            ))
+        }
+        // A manifest is never referenced by another manifest.
+        RefLevel::Manifest => None,
+    }
+}
+
+// ─── storage.files ───────────────────────────────────────────────────────
+
+/// References held by `storage.files.blob_hash`.
+///
+/// Contributes at **both** levels, which is why `RefLevel` is a parameter
+/// rather than a property of the source:
+///
+/// * [`RefLevel::Manifest`] — a CDC file's `blob_hash` names a manifest.
+/// * [`RefLevel::Chunk`] — a pre-CDC legacy file, whose `blob_hash` names a
+///   whole-file blob with no manifest behind it. The `NOT EXISTS` guard is
+///   load-bearing: for a single-chunk file the whole-file hash *equals* its
+///   lone chunk's hash, so without it the row would be counted at both
+///   levels.
+pub struct FilesReferenceSource {
+    pool: Arc<PgPool>,
+}
+
+impl FilesReferenceSource {
+    pub fn new(pool: Arc<PgPool>) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl BlobReferenceSource for FilesReferenceSource {
+    fn source_name(&self) -> &'static str {
+        "files"
+    }
+
+    fn ref_count_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String> {
+        files_ref_sql(level, outer_hash_expr)
+    }
+
+    async fn count_references(&self, blob_hash: &str) -> Result<u64, DomainError> {
+        // No level split here: the question is "how many file rows name this
+        // exact hash", and a hash names either a manifest or a legacy blob,
+        // never both at once from the caller's point of view.
+        let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM storage.files WHERE blob_hash = $1")
+            .bind(blob_hash)
+            .fetch_one(self.pool.as_ref())
+            .await
+            .map_err(|e| {
+                DomainError::internal_error("BlobRefSource", format!("files count: {e}"))
+            })?;
+        Ok(n.max(0) as u64)
+    }
+
+    async fn list_referenced_blobs(
+        &self,
+        cursor: Option<Vec<u8>>,
+        limit: usize,
+    ) -> Result<(Vec<String>, Option<Vec<u8>>), DomainError> {
+        // Paged by the file's own PK so the cursor is stable under concurrent
+        // inserts; `blob_hash` is not unique and would skip or repeat rows.
+        let after: Option<Uuid> = match cursor {
+            Some(bytes) => Some(decode_uuid_cursor(&bytes)?),
+            None => None,
+        };
+
+        let rows = sqlx::query(
+            "SELECT id, blob_hash FROM storage.files
+              WHERE ($1::uuid IS NULL OR id > $1)
+              ORDER BY id
+              LIMIT $2",
+        )
+        .bind(after)
+        .bind(limit as i64)
+        .fetch_all(self.pool.as_ref())
+        .await
+        .map_err(|e| DomainError::internal_error("BlobRefSource", format!("files page: {e}")))?;
+
+        let next = rows
+            .last()
+            .map(|r| r.get::<Uuid, _>("id").as_bytes().to_vec())
+            .filter(|_| rows.len() == limit);
+        let hashes = rows
+            .iter()
+            .map(|r| r.get::<String, _>("blob_hash"))
+            .collect();
+        Ok((hashes, next))
+    }
+}
+
+// ─── storage.chunk_manifests ─────────────────────────────────────────────
+
+/// References held by `storage.chunk_manifests.chunk_hashes[]`.
+///
+/// Chunk level only — a manifest never references another manifest, so
+/// [`RefLevel::Manifest`] yields `None` and this source contributes nothing
+/// to the manifest recompute.
+pub struct ChunksReferenceSource {
+    pool: Arc<PgPool>,
+}
+
+impl ChunksReferenceSource {
+    pub fn new(pool: Arc<PgPool>) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl BlobReferenceSource for ChunksReferenceSource {
+    fn source_name(&self) -> &'static str {
+        "chunks"
+    }
+
+    fn ref_count_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option<String> {
+        chunks_ref_sql(level, outer_hash_expr)
+    }
+
+    async fn count_references(&self, blob_hash: &str) -> Result<u64, DomainError> {
+        let n: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM storage.chunk_manifests WHERE $1 = ANY(chunk_hashes)",
+        )
+        .bind(blob_hash)
+        .fetch_one(self.pool.as_ref())
+        .await
+        .map_err(|e| DomainError::internal_error("BlobRefSource", format!("chunks count: {e}")))?;
+        Ok(n.max(0) as u64)
+    }
+
+    async fn list_referenced_blobs(
+        &self,
+        cursor: Option<Vec<u8>>,
+        limit: usize,
+    ) -> Result<(Vec<String>, Option<Vec<u8>>), DomainError> {
+        // Paged by the manifest PK, not by the unnested chunk hash: a single
+        // manifest expands to many hashes, so the page boundary has to fall
+        // between manifests or the cursor cannot be resumed unambiguously.
+        let after: Option<String> = match cursor {
+            Some(bytes) => Some(String::from_utf8(bytes).map_err(|e| {
+                DomainError::internal_error("BlobRefSource", format!("bad chunk cursor: {e}"))
+            })?),
+            None => None,
+        };
+
+        let rows = sqlx::query(
+            "SELECT file_hash, chunk_hashes FROM storage.chunk_manifests
+              WHERE ($1::text IS NULL OR file_hash > $1)
+              ORDER BY file_hash
+              LIMIT $2",
+        )
+        .bind(after)
+        .bind(limit as i64)
+        .fetch_all(self.pool.as_ref())
+        .await
+        .map_err(|e| DomainError::internal_error("BlobRefSource", format!("chunks page: {e}")))?;
+
+        let next = rows
+            .last()
+            .map(|r| r.get::<String, _>("file_hash").into_bytes())
+            .filter(|_| rows.len() == limit);
+        let hashes = rows
+            .iter()
+            .flat_map(|r| r.get::<Vec<String>, _>("chunk_hashes"))
+            .collect();
+        Ok((hashes, next))
+    }
+}
+
+fn decode_uuid_cursor(bytes: &[u8]) -> Result<Uuid, DomainError> {
+    let raw: [u8; 16] = bytes.try_into().map_err(|_| {
+        DomainError::internal_error(
+            "BlobRefSource",
+            format!("bad uuid cursor: expected 16 bytes, got {}", bytes.len()),
+        )
+    })?;
+    Ok(Uuid::from_bytes(raw))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::application::ports::blob_reference_ports::BlobReferenceRegistry;
+
+    /// The registry sums whatever the sources emit; these helpers exercise the
+    /// same code path without needing a pool, since `ref_count_sql` is pure.
+    fn summed(level: RefLevel, outer: &str) -> String {
+        let frags: Vec<String> = [files_ref_sql(level, outer), chunks_ref_sql(level, outer)]
+            .into_iter()
+            .flatten()
+            .collect();
+        if frags.is_empty() {
+            "0".to_string()
+        } else {
+            frags.join("\n + ")
+        }
+    }
+
+    /// The chunk-level expression must reproduce the two terms
+    /// `blobs_consistency` inlines today: legacy-only files (guarded by
+    /// NOT EXISTS) plus manifests citing the chunk.
+    #[test]
+    fn chunk_level_reproduces_todays_two_terms() {
+        let expr = summed(RefLevel::Chunk, "b.hash");
+        assert!(expr.contains("storage.files"), "{expr}");
+        assert!(
+            expr.contains("NOT EXISTS"),
+            "legacy term must keep the CDC guard: {expr}"
+        );
+        assert!(
+            expr.contains("= ANY(cnt_m.chunk_hashes)"),
+            "chunk term missing: {expr}"
+        );
+        assert!(expr.contains("b.hash"), "must correlate on the outer row");
+        assert!(expr.contains('+'), "both terms must be summed: {expr}");
+    }
+
+    /// Only `storage.files` references a manifest, so the manifest-level
+    /// expression is the single files term with no `NOT EXISTS` guard — the
+    /// guard exists to keep CDC rows *out* of the chunk level, and applying
+    /// it here would count nothing at all.
+    #[test]
+    fn manifest_level_is_files_only_and_unguarded() {
+        let expr = summed(RefLevel::Manifest, "m.file_hash");
+        assert!(expr.contains("storage.files"), "{expr}");
+        assert!(!expr.contains("NOT EXISTS"), "{expr}");
+        assert!(
+            !expr.contains("chunk_hashes"),
+            "chunks must not contribute at manifest level: {expr}"
+        );
+        assert!(!expr.contains('+'), "only one source contributes: {expr}");
+        assert!(expr.contains("m.file_hash"));
+    }
+
+    /// Fragments must not use the aliases the sweeps use for their outer row
+    /// (`b` for storage.blobs, `m` for chunk_manifests), or the manifest sweep
+    /// would shadow its own alias and silently correlate against itself.
+    #[test]
+    fn fragments_avoid_outer_row_aliases() {
+        for level in RefLevel::ALL {
+            let expr = summed(level, "m.file_hash");
+            for bad in [
+                "storage.files f",
+                "storage.files b",
+                "chunk_manifests m ",
+                "chunk_manifests b",
+            ] {
+                assert!(!expr.contains(bad), "alias collision at {level:?}: {expr}");
+            }
+        }
+    }
+
+    /// A source declining a level must drop out of the sum entirely, which is
+    /// what keeps manifest-only tables out of the chunk recompute where the
+    /// single-chunk hash alias would double-count them.
+    #[test]
+    fn chunks_source_declines_manifest_level() {
+        assert!(chunks_ref_sql(RefLevel::Manifest, "m.file_hash").is_none());
+        assert!(chunks_ref_sql(RefLevel::Chunk, "b.hash").is_some());
+    }
+
+    /// Guards the registry contract the sweeps rely on: an empty level still
+    /// yields a valid scalar expression.
+    #[test]
+    fn empty_registry_yields_zero_literal() {
+        let r = BlobReferenceRegistry::new();
+        assert_eq!(r.ref_count_expr(RefLevel::Manifest, "m.file_hash"), "0");
+    }
+}

src/infrastructure/repositories/pg/mod.rs

@@ -1,5 +1,6 @@
 mod address_book_pg_repository;
 mod app_password_pg_repository;
+pub mod blob_reference_sources;
 mod calendar_event_pg_repository;
 mod calendar_pg_repository;
 mod contact_group_pg_repository;