diff --git a/crates/tinymemory-api/src/provider/records.rs b/crates/tinymemory-api/src/provider/records.rs
index f1f6641..1514e51 100644
--- a/crates/tinymemory-api/src/provider/records.rs
+++ b/crates/tinymemory-api/src/provider/records.rs
@@ -350,7 +350,9 @@ pub trait MemoryMaintenance: Send + Sync {
     /// Idempotent, and by construction rather than by bookkeeping — the ingest
     /// gate answers `already_ingested` for a document the tree already holds, so
     /// a second pass writes nothing and an interrupted pass loses nothing. That
-    /// is also why `limit` bounds cost rather than carrying a cursor.
+    /// is also why `limit` bounds cost rather than carrying a cursor: a pass asks
+    /// that gate before it spends, so a document already filed is reported but
+    /// never charged, and calling again advances past it (openhuman#6051).
     ///
     /// Expensive on purpose to call explicitly: a pass is one read and one set
     /// of chunk embeddings per document. A driver must not run this on its own
diff --git a/crates/tinymemory-bus/src/provider/types.rs b/crates/tinymemory-bus/src/provider/types.rs
index faa3acf..f21cfb7 100644
--- a/crates/tinymemory-bus/src/provider/types.rs
+++ b/crates/tinymemory-bus/src/provider/types.rs
@@ -786,23 +786,28 @@ pub struct FlushOutcome {
 /// Four counters rather than one, because "did nothing" has three very
 /// different causes a caller has to be able to tell apart: the tree already
 /// held everything (`already_present`), nothing could be addressed
-/// (`skipped`), or there was nothing to look at (`scanned: 0`). Collapsing
-/// them would make an account whose scope could not be resolved read exactly
-/// like one that is fully backfilled.
+/// (`skipped`), or there was nothing to file at all (`scanned: 0` with the
+/// other two at zero). Collapsing them would make an account whose scope
+/// could not be resolved read exactly like one that is fully backfilled.
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub struct BackfillTreesOutcome {
-    /// Documents examined this pass.
+    /// Documents charged against the limit this pass: read and filed, or on a
+    /// dry run the ones a real pass would read and file. A document the tree
+    /// already holds is reported under `already_present` instead.
     pub scanned: u64,
     /// Documents that produced new memory-tree rows.
     pub ingested: u64,
     /// Documents the tree already held. Not a failure: this is what makes a
-    /// repeated pass readable as "nothing left to do".
+    /// repeated pass readable as "nothing left to do". Recognised before any
+    /// budget is spent, so they never stand between a pass and the documents
+    /// behind them (openhuman#6051).
     pub already_present: u64,
     /// Documents left alone — no resolvable scope, or a tolerated failure.
     /// Never filed under a guess.
     pub skipped: u64,
-    /// Whether the pass stopped on its limit with documents still unexamined.
-    /// The caller resumes by calling again; there is no cursor to carry.
+    /// Whether the pass stopped on its limit with documents still waiting to
+    /// be filed. The caller resumes by calling again; there is no cursor to
+    /// carry, because a document already filed costs the next pass nothing.
     pub more_pending: bool,
     /// Bounded, human-readable reasons behind `skipped`.
     pub notes: Vec<String>,
@@ -811,11 +816,12 @@ pub struct BackfillTreesOutcome {
 /// How much of the backfill to attempt, and whether to write at all.
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub struct BackfillTreesRequest {
-    /// Documents to examine at most. `None` leaves the bound to the driver.
+    /// Documents to read and file at most. `None` leaves the bound to the
+    /// driver.
     ///
     /// A bound rather than a cursor because the work is idempotent: the pass
-    /// re-reads what it already treed and the ingest gate answers
-    /// `already_ingested`, so resuming is just calling again.
+    /// asks the ingest gate before it spends, a document the tree already holds
+    /// costs nothing, and so resuming is just calling again (openhuman#6051).
     pub limit: Option<u64>,
     /// Report what a real pass would examine, and write nothing.
     ///
diff --git a/crates/tinymemory-core/src/backfill.rs b/crates/tinymemory-core/src/backfill.rs
index 57215c6..3fbaaf4 100644
--- a/crates/tinymemory-core/src/backfill.rs
+++ b/crates/tinymemory-core/src/backfill.rs
@@ -14,13 +14,22 @@
 //! the `{toolkit}:{connection_id}` identity: two call sites owning one rule is
 //! exactly what produced #6007.
 //!
-//! # Idempotent by construction, not by bookkeeping
+//! # Idempotent by construction, resumable by asking first
 //!
 //! The ingest pipeline answers `already_ingested` when its transaction persists
 //! nothing, so running this twice writes nothing the second time. There is no
 //! watermark to keep and no way for an interrupted run to corrupt anything —
-//! the worst case is repeated work. `limit` exists to bound *cost*, not to
-//! guarantee correctness.
+//! the worst case is repeated work.
+//!
+//! `limit` bounds *cost*: it counts the documents a pass reads and files, not
+//! the documents it looks at. A document the tree already holds is recognised
+//! by asking the ingest gate first — one keyed lookup, by the funnel's own
+//! identity — and is never charged. That is what makes "call again" a real
+//! resume story. The first version charged the limit before the gate could
+//! answer, and because `list_documents` yields newest first — exactly the
+//! documents the sync path had already filed — a large account re-examined the
+//! same `limit` filed documents on every pass and never reached the rest
+//! (openhuman#6051).
 //!
 //! # Why it costs what it costs
 //!
@@ -37,7 +46,7 @@ use crate::sources::SourceKind;
 use crate::store::MemoryClientRef;
 use crate::Config;
 
-/// Documents examined per pass when the caller names no bound.
+/// Documents read and filed per pass when the caller names no bound.
 ///
 /// Deliberately modest: a pass is resumable (just call again), and a caller
 /// that wants the whole account can say so. The default protects the operator
@@ -53,17 +62,22 @@ const MAX_NOTES: usize = 20;
 /// What one backfill pass examined and wrote.
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct BackfillReport {
-    /// Documents examined.
+    /// Documents charged against the limit: read and filed, or — on a dry run
+    /// — the ones a real pass would read and file. A document the tree already
+    /// holds is not counted here.
     pub scanned: u64,
     /// Documents that produced new memory-tree rows.
     pub ingested: u64,
     /// Documents the tree already held. Not a failure — this is the counter
-    /// that makes a repeated run readable as "nothing left to do".
+    /// that makes a repeated run readable as "nothing left to do". Recognised
+    /// before any budget is spent, so they never stand between a pass and the
+    /// documents behind them.
     pub already_present: u64,
     /// Documents left alone: no resolvable scope, or a tolerated read/ingest
     /// failure. Never filed under a guess.
     pub skipped: u64,
-    /// Whether the pass stopped on its limit with documents still unexamined.
+    /// Whether the pass stopped on its limit with documents still waiting to be
+    /// filed.
     pub more_pending: bool,
     /// Bounded, human-readable reasons behind `skipped`.
     pub notes: Vec<String>,
@@ -86,9 +100,10 @@ struct Target {
 
 /// Walk the connector namespaces, feeding stored documents into the memory tree.
 ///
-/// `dry_run` reports what a real pass would examine without reading any content
-/// or writing anything, which is the only honest way to show an operator the
-/// size of the job before they pay for it.
+/// `dry_run` reports what a real pass would read and file — and what the tree
+/// already holds — without reading any content or writing anything, which is
+/// the only honest way to show an operator the size of the job before they pay
+/// for it.
 pub async fn backfill_connector_trees(
     config: &Config,
     client: &MemoryClientRef,
@@ -123,10 +138,6 @@ pub async fn backfill_connector_trees(
             .unwrap_or_default();
 
         for document in documents {
-            if report.scanned >= limit {
-                report.more_pending = true;
-                break 'targets;
-            }
             let Some(key) = document.get("key").and_then(serde_json::Value::as_str) else {
                 // A document row with no key cannot be read back or addressed
                 // in the tree; counting it as skipped keeps `scanned` honest.
@@ -137,6 +148,59 @@ pub async fn backfill_connector_trees(
                 ));
                 continue;
             };
+
+            // Ask before spending. A document the tree already holds costs one
+            // keyed lookup and none of the limit; only a document that still
+            // needs reading and filing is charged, so a pass that stops on its
+            // limit resumes past everything already filed when called again.
+            // The probe has to come before the limit check, or a pass could not
+            // tell "more to file" from "more already filed".
+            match crate::engine::connector_item_already_treed(
+                config,
+                &target.toolkit,
+                &target.connection_id,
+                key,
+            ) {
+                Ok(Some(true)) => {
+                    report.already_present = report.already_present.saturating_add(1);
+                    // A filed document is the only kind this loop handles
+                    // without awaiting anything, and on a large account they
+                    // come in long runs — every document the sync path has
+                    // already treed. Hand the runtime a turn per document so
+                    // a pass over tens of thousands of them does not hold its
+                    // worker thread for the whole run.
+                    tokio::task::yield_now().await;
+                    continue;
+                }
+                Ok(Some(false)) => {}
+                // The scope was built from the registry above, so this is close
+                // to unreachable — but the funnel would refuse the same item,
+                // and it is counted the way that refusal is.
+                Ok(None) => {
+                    report.skipped = report.skipped.saturating_add(1);
+                    continue;
+                }
+                Err(error) => {
+                    let rendered = format!("{error:#}");
+                    crate::corruption::escalate_or_count(
+                        "connector tree backfill",
+                        config,
+                        error,
+                        &failures,
+                    )?;
+                    report.skipped = report.skipped.saturating_add(1);
+                    report.note(format!(
+                        "{}: the tree gate could not be read ({rendered})",
+                        target.namespace
+                    ));
+                    continue;
+                }
+            }
+
+            if report.scanned >= limit {
+                report.more_pending = true;
+                break 'targets;
+            }
             report.scanned = report.scanned.saturating_add(1);
             if dry_run {
                 continue;
diff --git a/crates/tinymemory-core/src/backfill_tests.rs b/crates/tinymemory-core/src/backfill_tests.rs
index 7a7bb16..a57e7e6 100644
--- a/crates/tinymemory-core/src/backfill_tests.rs
+++ b/crates/tinymemory-core/src/backfill_tests.rs
@@ -190,6 +190,10 @@ async fn a_stored_document_is_filed_into_the_tree_and_never_twice() {
         second.already_present, 1,
         "and must say why it wrote nothing: {second:?}"
     );
+    assert_eq!(
+        second.scanned, 0,
+        "a document the tree holds is never charged against the limit: {second:?}"
+    );
     assert_eq!(
         crate::store::chunks::store::count_chunks(&*config).expect("count chunks"),
         chunks_after_first,
@@ -238,3 +242,195 @@ async fn a_bounded_pass_reports_that_more_is_pending() {
         "a pass that stopped on its limit must not read as complete: {report:?}"
     );
 }
+
+/// The bug behind openhuman#6051. A pass that stops on its limit has to be
+/// resumable by calling again, and it was not: a document the tree already
+/// held was charged against the limit before the gate could say so, so a
+/// profile whose newest 500 documents were already filed re-examined those
+/// same 500 on every click and never reached the rest. Already-filed documents
+/// are recognised first and cost nothing, so each bounded pass files new
+/// documents until none are left.
+#[tokio::test]
+async fn already_filed_documents_do_not_charge_the_limit_so_bounded_passes_converge() {
+    let (_dir, config, client) = workspace(&[composio_source("src_gmail", "gmail", "conn-1")]);
+    for key in ["msg-1", "msg-2", "msg-3"] {
+        store_document(&client, "source:gmail:conn-1", key, "Q3 roadmap.").await;
+    }
+
+    let first = super::backfill_connector_trees(&*config, &client, Some(2), false)
+        .await
+        .expect("first bounded pass");
+    assert_eq!(
+        first.ingested, 2,
+        "the first pass files up to its limit: {first:?}"
+    );
+    assert!(
+        first.more_pending,
+        "one document is still waiting: {first:?}"
+    );
+
+    let second = super::backfill_connector_trees(&*config, &client, Some(2), false)
+        .await
+        .expect("second bounded pass");
+    assert_eq!(
+        second.already_present, 2,
+        "the documents the first pass filed are recognised: {second:?}"
+    );
+    assert_eq!(
+        second.ingested, 1,
+        "and they must not have spent the budget the waiting one needed: {second:?}"
+    );
+    assert!(
+        !second.more_pending,
+        "nothing is waiting once the last document is filed: {second:?}"
+    );
+
+    let third = super::backfill_connector_trees(&*config, &client, Some(2), false)
+        .await
+        .expect("third bounded pass");
+    assert_eq!(
+        (third.scanned, third.ingested, third.already_present),
+        (0, 0, 3),
+        "a fully filed store charges nothing and says so: {third:?}"
+    );
+    assert!(
+        !third.more_pending,
+        "and does not ask to be run again: {third:?}"
+    );
+}
+
+/// Targets are walked in registry order, and a limit reached inside one used
+/// to end the pass before the next was looked at — so an account whose first
+/// namespace alone held more filed documents than the limit walled off every
+/// namespace behind it, including the legacy `skill-` one this backfill exists
+/// for. With filed documents free, the walk carries on into the next target.
+#[tokio::test]
+async fn a_later_target_is_reached_once_the_earlier_one_is_fully_filed() {
+    let (_dir, config, client) = workspace(&[
+        composio_source("src_notion", "notion", "conn-n"),
+        composio_source("src_gmail", "gmail", "conn-g"),
+    ]);
+    store_document(&client, "source:notion:conn-n", "page-1", "Roadmap page.").await;
+    store_document(&client, "source:gmail:conn-g", "msg-1", "Q3 roadmap.").await;
+
+    let first = super::backfill_connector_trees(&*config, &client, Some(1), false)
+        .await
+        .expect("first bounded pass");
+    assert_eq!(
+        (first.ingested, first.more_pending),
+        (1, true),
+        "the first target fills the whole budget: {first:?}"
+    );
+
+    let second = super::backfill_connector_trees(&*config, &client, Some(1), false)
+        .await
+        .expect("second bounded pass");
+    assert_eq!(
+        (second.already_present, second.ingested, second.more_pending),
+        (1, 1, false),
+        "the filed first target costs nothing, so the second is reached: {second:?}"
+    );
+
+    let treed = crate::store::chunks::store::list_chunks(
+        &*config,
+        &crate::store::chunks::ListChunksQuery {
+            source_id: Some("gmail:conn-g:msg-1".into()),
+            limit: Some(8),
+            ..Default::default()
+        },
+    )
+    .expect("list chunks by source id");
+    assert!(
+        !treed.is_empty(),
+        "the second target's document must actually be in the tree"
+    );
+}
+
+/// The preview is what the operator confirms against, so it has to count the
+/// documents that would be filed — not everything in the store, most of which
+/// may already be there. Once nothing is waiting it says so with `scanned: 0`,
+/// which is the signal a host uses for "nothing to repair".
+#[tokio::test]
+async fn a_dry_run_tells_already_filed_documents_from_waiting_ones() {
+    let (_dir, config, client) = workspace(&[composio_source("src_gmail", "gmail", "conn-1")]);
+    store_document(&client, "source:gmail:conn-1", "msg-1", "Q3 roadmap.").await;
+    store_document(&client, "source:gmail:conn-1", "msg-2", "Launch date.").await;
+
+    let filed_one = super::backfill_connector_trees(&*config, &client, Some(1), false)
+        .await
+        .expect("file one document");
+    assert_eq!(filed_one.ingested, 1, "{filed_one:?}");
+    let chunks_after = crate::store::chunks::store::count_chunks(&*config).expect("count chunks");
+
+    let preview = super::backfill_connector_trees(&*config, &client, None, true)
+        .await
+        .expect("dry run with one document waiting");
+    assert_eq!(
+        (
+            preview.scanned,
+            preview.already_present,
+            preview.more_pending
+        ),
+        (1, 1, false),
+        "the preview counts the waiting document and names the filed one: {preview:?}"
+    );
+    assert_eq!(
+        crate::store::chunks::store::count_chunks(&*config).expect("count chunks"),
+        chunks_after,
+        "a dry run must leave the tree untouched"
+    );
+
+    super::backfill_connector_trees(&*config, &client, None, false)
+        .await
+        .expect("file the rest");
+    let done = super::backfill_connector_trees(&*config, &client, None, true)
+        .await
+        .expect("dry run with nothing waiting");
+    assert_eq!(
+        (done.scanned, done.already_present, done.more_pending),
+        (0, 2, false),
+        "a fully filed store previews as nothing to do: {done:?}"
+    );
+}
+
+/// The probe's failure policy is the funnel's (openhuman#5820): an ordinary
+/// store error skips that document with a note and moves on, charging nothing,
+/// and only corruption aborts the pass. Here the tree's directory is replaced
+/// by a file, so the gate cannot even be opened — a plain I/O error, nothing a
+/// corruption classifier would match.
+#[tokio::test]
+async fn a_gate_read_failure_is_tolerated_per_document_and_charges_nothing() {
+    let (dir, config, client) = workspace(&[composio_source("src_gmail", "gmail", "conn-1")]);
+    store_document(&client, "source:gmail:conn-1", "msg-1", "Q3 roadmap.").await;
+    store_document(&client, "source:gmail:conn-1", "msg-2", "Launch date.").await;
+
+    let tree_dir = dir.path().join("workspace").join("memory_tree");
+    let _ = std::fs::remove_dir_all(&tree_dir);
+    std::fs::write(&tree_dir, b"not a directory").expect("occupy the tree path");
+
+    let report = super::backfill_connector_trees(&*config, &client, Some(1), false)
+        .await
+        .expect("a gate that cannot be read is tolerated, not fatal");
+    assert_eq!(
+        (
+            report.scanned,
+            report.ingested,
+            report.already_present,
+            report.skipped
+        ),
+        (0, 0, 0, 2),
+        "every document is skipped and none is charged: {report:?}"
+    );
+    assert!(
+        !report.more_pending,
+        "a skip is not a document left waiting: {report:?}"
+    );
+    assert!(
+        report
+            .notes
+            .iter()
+            .any(|note| note.contains("tree gate could not be read")),
+        "the skip must say why: {:?}",
+        report.notes
+    );
+}
diff --git a/crates/tinymemory-core/src/engine/mod.rs b/crates/tinymemory-core/src/engine/mod.rs
index ff4e874..7bd68d4 100644
--- a/crates/tinymemory-core/src/engine/mod.rs
+++ b/crates/tinymemory-core/src/engine/mod.rs
@@ -72,6 +72,9 @@ pub use sync::{
 };
 // Crate-private seam for `crate::sources::sync` (openhuman#5820); not host surface.
 pub(crate) use sync::run_source_pipeline_core;
+// Crate-private seam for `crate::backfill` (openhuman#6051): the ingest gate
+// asked by the funnel's own identity, so the walk never re-derives it.
+pub(crate) use sync::connector_item_already_treed;
 // The audit type, under the seam path OpenHuman already names
 // (`memory::tinycortex::SyncAuditEntry` embeds it in an RPC response type).
 // The type itself is core-owned (#18 §B1a); only the address is preserved.
diff --git a/crates/tinymemory-core/src/engine/sync.rs b/crates/tinymemory-core/src/engine/sync.rs
index 3d69612..56a711f 100644
--- a/crates/tinymemory-core/src/engine/sync.rs
+++ b/crates/tinymemory-core/src/engine/sync.rs
@@ -147,21 +147,21 @@ pub async fn ingest_connector_item_into_tree(
     title: &str,
     content: &str,
 ) -> anyhow::Result<Option<crate::ingest_pipeline::IngestResult>> {
-    let toolkit = toolkit.trim().to_ascii_lowercase();
-    let connection_id = connection_id.trim();
-    // A blank toolkit/connection would yield a scope with no platform prefix
-    // (`":conn"`), which no retrieval kind matches; skip rather than write an
-    // unreachable tree. The caller's own store still holds the item.
-    if toolkit.is_empty() || connection_id.is_empty() {
+    // The caller's own store still holds a scopeless item; it is only the tree
+    // that skips it.
+    let Some(identity) = connector_item_identity(toolkit, connection_id, item_id) else {
         tracing::debug!(
             item_id = %item_id,
             "[tinycortex:sync] skipping memory-tree ingest: item has no toolkit/connection scope"
         );
         return Ok(None);
-    }
-    let tree_scope = format!("{toolkit}:{connection_id}");
-    let source_id = format!("{tree_scope}:{item_id}");
-    let owner = format!("{toolkit}-sync:{connection_id}");
+    };
+    let ConnectorItemIdentity {
+        tree_scope,
+        source_id,
+        owner,
+        toolkit,
+    } = identity;
     let input = tinycortex::memory::ingest::canonicalize::document::DocumentInput {
         provider: format!("composio:{toolkit}"),
         title: title.to_string(),
@@ -182,6 +182,84 @@ pub async fn ingest_connector_item_into_tree(
     .map_err(|error| anyhow::anyhow!("memory-tree ingest failed for source `{source_id}`: {error}"))
 }
 
+/// The tree identity of one connector item, derived once for every reader and
+/// writer of it.
+///
+/// Three names that must agree with each other and with what the sync path
+/// wrote: the tree scope, the per-item source id under it, and the owner. They
+/// are built here and nowhere else — [`ingest_connector_item_into_tree`] files
+/// under them and [`connector_item_already_treed`] asks the gate by them, so a
+/// backfill that probes before it files cannot probe by one name and file under
+/// another. Two call sites owning this rule is what produced openhuman#6007.
+struct ConnectorItemIdentity {
+    /// `{toolkit}:{connection_id}` — the `path_scope` retrieval resolves by
+    /// platform prefix, and the literal prefix OpenHuman counts a source's
+    /// ingest by.
+    tree_scope: String,
+    /// `{tree_scope}:{item_id}` — the ingest gate's key, one per item so each
+    /// message admits independently.
+    source_id: String,
+    /// `{toolkit}-sync:{connection_id}`.
+    owner: String,
+    /// The normalised toolkit, for the ingest tag and provider name.
+    toolkit: String,
+}
+
+/// Derives the identity, or `None` when either scope half is blank.
+///
+/// A blank toolkit/connection would yield a scope with no platform prefix
+/// (`":conn"`), which no retrieval kind matches; callers skip rather than write
+/// — or probe for — an unreachable tree.
+fn connector_item_identity(
+    toolkit: &str,
+    connection_id: &str,
+    item_id: &str,
+) -> Option<ConnectorItemIdentity> {
+    let toolkit = toolkit.trim().to_ascii_lowercase();
+    let connection_id = connection_id.trim();
+    if toolkit.is_empty() || connection_id.is_empty() {
+        return None;
+    }
+    let tree_scope = format!("{toolkit}:{connection_id}");
+    Some(ConnectorItemIdentity {
+        source_id: format!("{tree_scope}:{item_id}"),
+        owner: format!("{toolkit}-sync:{connection_id}"),
+        tree_scope,
+        toolkit,
+    })
+}
+
+/// Whether the memory tree already holds a connector item, asked by the same
+/// identity [`ingest_connector_item_into_tree`] files it under (openhuman#6051).
+///
+/// Answers `Ok(None)` for an item with no resolvable scope — the item the funnel
+/// would skip — and the ingest gate's answer otherwise. That is the gate's
+/// best-effort read, not its transactional claim: a concurrent ingest can file
+/// the item between this answer and a caller's own ingest, in which case that
+/// ingest answers `already_ingested` and nothing is written twice. Callers use
+/// it to decide what to *spend* on — a backfill pass that charged its limit for
+/// documents the tree already held could never advance past them — while
+/// correctness stays with the claim inside the ingest.
+///
+/// # Errors
+/// Returns the store's error when the gate cannot be read.
+pub(crate) fn connector_item_already_treed(
+    config: &Config,
+    toolkit: &str,
+    connection_id: &str,
+    item_id: &str,
+) -> anyhow::Result<Option<bool>> {
+    let Some(identity) = connector_item_identity(toolkit, connection_id, item_id) else {
+        return Ok(None);
+    };
+    crate::store::chunks::store::is_source_ingested(
+        config,
+        crate::store::chunks::types::SourceKind::Document,
+        &identity.source_id,
+    )
+    .map(Some)
+}
+
 /// [`ingest_connector_item_into_tree`] plus the failure policy every connector
 /// sync path needs, so no path has to reach for `crate::corruption` itself.
 ///
diff --git a/crates/tinymemory-core/src/engine/sync_tests.rs b/crates/tinymemory-core/src/engine/sync_tests.rs
index 1411f4d..8120842 100644
--- a/crates/tinymemory-core/src/engine/sync_tests.rs
+++ b/crates/tinymemory-core/src/engine/sync_tests.rs
@@ -769,3 +769,73 @@ async fn the_shared_funnel_skips_either_blank_scope_half() {
         );
     }
 }
+
+/// The gate probe the backfill spends by (openhuman#6051) answers through the
+/// same identity the funnel files under: nothing before the ingest, the item
+/// after it — including when the caller spells the scope halves differently
+/// from the writer, since both normalise through one derivation — and `None`
+/// for the scopeless item the funnel itself would skip.
+#[tokio::test]
+async fn the_gate_probe_agrees_with_the_funnel_it_files_through() {
+    use tinymemory_api::host::test_support::TestHostConfig;
+    use tinymemory_api::host::MemoryHostConfig;
+
+    crate::test_seams::init();
+    let workspace = tempfile::tempdir().expect("workspace");
+    let mut host = TestHostConfig::default();
+    host.workspace_dir = workspace.path().join("workspace");
+    let config = host.to_arc();
+
+    assert_eq!(
+        super::connector_item_already_treed(&*config, "gmail", "conn-1", "msg-1")
+            .expect("the gate answers"),
+        Some(false),
+        "an item the tree has never seen is not treed"
+    );
+
+    super::ingest_connector_item_into_tree(
+        &*config,
+        "gmail",
+        "conn-1",
+        "msg-1",
+        "Quarterly planning",
+        "Let's finalise the Q3 roadmap.",
+    )
+    .await
+    .expect("ingest")
+    .expect("a scoped item reaches the pipeline");
+
+    assert_eq!(
+        super::connector_item_already_treed(&*config, "gmail", "conn-1", "msg-1")
+            .expect("the gate answers"),
+        Some(true),
+        "the funnel's own write is recognised"
+    );
+    assert_eq!(
+        super::connector_item_already_treed(&*config, " Gmail ", " conn-1 ", "msg-1")
+            .expect("the gate answers"),
+        Some(true),
+        "the probe normalises the scope the way the funnel did, so it asks by the key the \
+         funnel wrote"
+    );
+    assert_eq!(
+        super::connector_item_already_treed(&*config, "gmail", "conn-1", "msg-2")
+            .expect("the gate answers"),
+        Some(false),
+        "a different item under the same scope is its own key"
+    );
+
+    for (toolkit, connection_id, blank_half) in [
+        ("   ", "conn-1", "toolkit"),
+        ("gmail", "   ", "connection_id"),
+    ] {
+        assert_eq!(
+            super::connector_item_already_treed(&*config, toolkit, connection_id, "msg-1")
+                .unwrap_or_else(|error| panic!(
+                    "a blank {blank_half} is a skip, not an error: {error:#}"
+                )),
+            None,
+            "a blank {blank_half} has no tree identity to ask by"
+        );
+    }
+}