feat: scope backfill-by-copy and removal reconciliation jobs (#904)

* fix: enforce explicit-document session purity in dedup/merge paths

Audit for DEV-2000 (Scopes RFC prerequisite): explicit-level documents must
stay session-pure so scope memory can be built by copying explicit documents
between collections. Two classes of violation were possible:

- Exact-content and semantic dedup in crud/document.py matched candidates
  with no level or session scoping, so an explicit document could be
  reinforced by — or soft-deleted in favor of — a same-content document from
  a different session or a different level (silently merging cross-session
  derivations into one row).
- The generic create_observations tool handler accepted level='explicit'
  from agents with no message context (dreamer/dialectic), which would mint
  session-less explicit documents.

Enforcement (refuse, never rewrite):
- create_documents refuses explicit documents with a null session_name
- exact dedup keys on (content, level, session-for-explicit); derived levels
  keep cross-session consolidation
- is_rejected_duplicate scopes candidate search to the same level, and the
  same session for explicit documents
- the create_observations tool rejects explicit-level input outside message
  ingestion (deriver) context

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat: add card_refresh dream type for event-driven peer-card updates

Adds a lightweight dream variant (DEV-2000, Scopes RFC prerequisite) that
runs ONLY the peer-card update — for event-driven refreshes such as scope
membership changes and cold starts:

- DreamType.CARD_REFRESH alongside OMNI; dispatched by process_dream to a
  new run_card_refresh_dream orchestration
- CardRefreshSpecialist: restricted to get_recent_observations,
  search_memory, and update_peer_card (no observation-mutating tools), with
  a low tool-iteration cap of min(6, DREAM.MAX_TOOL_ITERATIONS)
- rebuild=True mode carried in the dream payload: the existing card is NOT
  injected into the prompt and the specialist rebuilds it solely from
  observations present in the collection (for use after removals)
- enqueue-able via the manual enqueue_dream path (bypasses volume gates);
  the work-unit key already embeds the dream type so a card refresh never
  collides with a pending omni dream. POST /v3/workspaces/{id}/schedule_dream
  accepts dream_type=card_refresh plus the rebuild flag
- card refreshes never advance the omni dream guard pair
  (last_dream_at / last_dream_document_count)
- shared PEER CARD prompt section extracted (verbatim) from
  DeductionSpecialist for reuse; CallPurpose gains dream.card_refresh

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat: reserve scope__ peer namespace with kind flag and guardrails

Introduce the scope peer namespace (scope__<name>) and the authoritative
{"kind": "scope"} configuration flag, plus the server-side guardrails:

- src/utils/scopes.py: single source of truth for the prefix, kind flag,
  and name helpers (scope_peer_name / is_scope_peer_name /
  scope_name_from_peer / validate_no_scope_peer_names)
- reject reserved-prefix names on peer get-or-create (422)
- reject scope peers as message authors in crud.create_messages (422)
- reject scope peers as chat/representation targets (422); a scope peer
  as the path-level observer is deferred to Phase 2b
- reject scope peers on the generic session-peer add/set/remove routes
  and the session-create peers mapping (422, directing to scopes routes)
- peers.list excludes scope peers by default; new PeerGet.kind option
  ("scope" | "all") switches the view via a configuration JSONB filter
- schemas: Scope / ScopeCreate / ScopeSessions(Add) and
  SessionCreate.scopes (unprefixed scope names, validated)

Part of DEV-1997 (Scopes RFC DEV-1970).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat: add scopes CRUD routes and session-create scopes wiring

New /v3/workspaces/{workspace_id}/scopes facade (workspace-level auth;
peer- and session-scoped keys are rejected):

  POST   ""                                   create-or-get (201/200)
  POST   /list                                 paginated scope list
  GET    /{scope_id}                           single scope
  POST   /{scope_id}/sessions                  add memberships
  DELETE /{scope_id}/sessions/{session_id}     remove membership
  GET    /{scope_id}/sessions                  list member session ids

- crud/scope.py: get_or_create_scopes stamps the backing peer with
  {"kind": "scope", "observe_me": false} and refuses to adopt a
  legacy peer occupying the reserved name without the flag (409)
- memberships are session_peers rows with observe_others=true /
  observe_me=false — identical to a hand-built observer peer
- SessionCreate.scopes: create-or-get each scope peer and add the
  membership at session creation (the no-backfill common path)
- crud/session.py: public upsert_session_peers wrapper so the facade
  bypasses the route-level guardrails without reaching into privates

Backfill of pre-existing documents and reconciliation on removal land in
DEV-1999; membership only affects messages ingested after the change.

Part of DEV-1997 (Scopes RFC DEV-1970).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* test: cover scopes facade, guardrails, and observer semantics

- create-or-get idempotency, list/get, name validation, legacy-collision
  rejection (409), auth scoping (workspace key ok, peer/session keys 401)
- reserved prefix rejected on peer create; peers.list kind filtering
- scope peers rejected as message authors, chat/representation targets,
  and on the generic session-peer routes
- membership add/list/remove with observe_others=true / observe_me=false
  row shape asserted via DB, and facade-less equivalence with a
  hand-built observer peer
- end-to-end litmus: after adding a session to a scope, the deriver
  enqueue fan-out includes the scope peer as an observer
- session creation with scopes: [a, b] creates both memberships

Part of DEV-1997 (Scopes RFC DEV-1970).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat: scope backfill-by-copy and removal reconciliation jobs

Retroactive scope membership changes (DEV-1999):

- New queue task types scope_backfill / scope_removal with payloads,
  work-unit keys ({task}:{workspace}:{scope_peer}:{session}), deduped
  enqueue (mirrors enqueue_dream), and consumer dispatch.
- Backfill copies a session's explicit documents from each sender's
  global (P, P) collection into the scope's (scope_peer, P) collection
  with internal_metadata.copied_from as the idempotency marker;
  soft-deleted copies from an earlier removal are restored, so
  add -> remove -> re-add converges on exactly one live copy. Completion
  enqueues one manual omni dream per touched collection.
- Removal soft-deletes the session's explicit documents in the scope's
  collections and cascades (fail-closed, transitively) to derived
  documents whose source_ids intersect anything removed, deletes the
  vectors from the external store, then enqueues a card_refresh dream
  with rebuild=True plus a manual omni dream per touched collection.
- Zero LLM re-derivation: explicit documents are session-pure (DEV-2000
  invariant); the only external call is re-embedding rows whose
  embedding column is NULL (external-store deployments).
- Per-session job status lives in the scope peer's internal_metadata
  under backfill_status, written via single-statement JSONB merges
  (concurrent-writer safe) and surfaced at
  GET /v3/workspaces/{w}/scopes/{scope_id}/status.
- Enqueued from the scopes add-sessions route and SessionCreate.scopes
  handling, only when the session already has messages.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* test: add scope backfill/removal test coverage, fix backfill status JSONB bug

Adds tests/deriver/test_scope_backfill.py covering the DEV-1999 scope
backfill-by-copy and removal reconciliation jobs implemented in a prior
commit: explicit-doc copying, copied_from idempotency (including
add->remove->re-add), multi-peer collection routing, removal cascade to
dependent derived docs, dream enqueues (manual omni on backfill;
card_refresh rebuild + omni on removal), the status route, and add-sessions
route wiring (backfill enqueued only when the session already has messages).

Fixes a real production bug surfaced by these tests: update_scope_backfill_status
passed json.dumps()'d strings through SQLAlchemy cast(..., JSONB), which
double-encodes (psycopg re-serializes the already-JSON string), producing a
JSONB string scalar instead of an object. Postgres's `||` between two
non-array jsonb scalars doesn't merge — it silently wraps both into a
2-element array, corrupting backfill_status into a list. This crashed
clear_scope_backfill_status's `#-` path delete (called on every removal)
with "path element is not an integer" once a session had ever completed a
backfill. Fixed by passing raw Python dicts to cast() instead, mirroring the
working pattern already used in update_collection_internal_metadata.

Also adds src.deriver.scope_backfill.tracked_db to conftest's tracked_db
patch list — the module was missing from that per-import-site allowlist, so
its DB work ran against the real configured database instead of the
isolated per-test database.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(crud): preserve cache invalidation across get_or_create retry

`get_or_create_peers` and `get_or_create_scopes` mutate existing rows, then
insert new ones inside `db.begin_nested()`. When a concurrent writer creates
one of those rows first, the insert raises IntegrityError and the function
retries.

`begin_nested()` autoflushes the pending UPDATEs *before* opening the
savepoint, so the rollback neither undoes them nor expires the now-clean ORM
state. The retry then compared already-updated values, found no change, and
dropped those peers from `changed_peers` — skipping the cache purge while the
row change committed anyway, leaving entries stale until the 300s TTL.

Carry the mutated names into the retry via `_pending_invalidation` so the
purge cannot be lost.

The scopes facade mirrors `get_or_create_peers`, so both copies carried this.
The peer path is pre-existing and runs on every message ingest.

Also add the missing /v3 prefix to `_SCOPES_ROUTE_GUIDANCE`, which pointed
callers at a 404.

Adds tests/crud/test_get_or_create_retry_invalidation.py, which drives a real
racing session and fails without this change.

* fix(scopes): make scope identity unforgeable, unblock non-pattern peer names

Three coupled changes to the scopes facade.

1. `PeerCreate` no longer gates internal lookups. It exists to validate a new,
   user-supplied peer id at the API boundary, but crud used it as a DTO for
   names that already exist, so any name outside RESOURCE_NAME_PATTERN raised a
   raw pydantic ValidationError — which is not a HonchoException, so it fell
   through to the catch-all handler as an HTTP 500. Adds `PeerSpec` (same
   fields, no charset pattern) as `PeerCreate`'s base, widens
   `get_or_create_peers` to accept it, and changes `get_peer` to take a plain
   str. All 13 construction sites converted; the create route keeps full
   validation.

   This unbreaks the Dreamer: DreamScheduler passes `collection.observer`
   straight into the specialist preflight, and scope peers have
   `observe_others=true`, so every `(scope.x, peer)` dream died there — the
   feature scopes exist to enable. It also fixes a pre-existing bug unrelated
   to scopes: a peer named `alice.smith` (legal before d429de0e5338, which
   validated names by length alone) 500s on message create, session peer add,
   and peer update.

2. The `kind` flag moves from `configuration` to `internal_metadata`.
   `configuration` is user-writable — `PeerCreate`/`PeerUpdate` accept a
   free-form dict and `update_peer` replaces it wholesale — so a legitimate
   `{"observe_me": true}` update silently dropped the flag, and a forged
   `{"kind": "scope"}` injected an ordinary peer into `POST /scopes/list`.
   `internal_metadata` appears in no API schema. `observe_me: false` stays in
   `configuration`, where it belongs.

3. Scope identity requires prefix AND flag, via `is_scope_peer()` and
   `scope_peer_clause()`. Neither half is forgeable: the prefix sits outside
   RESOURCE_NAME_PATTERN, `internal_metadata` is unreachable. Usage-site guards
   become flag-based so a legacy peer merely occupying the namespace keeps
   working rather than 422-ing on its own traffic; peer create and update stay
   name-based, since those must stop new names entering the namespace.
   `update_peer` now returns 422 instead of 500.

Also swaps the reserved prefix from `scope__` to `scope.`: `_` is inside
RESOURCE_NAME_PATTERN, so any tenant could already own a `scope__x` peer.

No DB migration — `internal_metadata` already exists on `peers`, and no scope
peers exist in any deployment yet.

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

* fix(scopes): validate peer names on create, close namespace squatting and upsert race

Addresses three review findings against 48047a6a.

1. `PeerSpec` let API callers create invalid and reserved-prefix peers.
   Widening `get_or_create_peers` to accept a pattern-free schema fixed the
   lookup 500s but also removed validation from the *insert* path, and
   request-controlled names reach it via message authors, session peer maps, and
   the chat observer path — none of which carry a charset pattern of their own.
   Confirmed: `POST /sessions/{id}/messages` with `peer_id: "scope.x"` returned
   201 and minted an unflagged squatter, after which `POST /scopes {id: x}` was
   permanently 409-blocked — namespace denial of service by any caller able to
   post a message. `peer_id: "not a valid name!@#"` was likewise created.

   Fixed by validating only names about to be INSERTed
   (`_validate_new_peer_names`), so already-existing names — legacy dotted
   names, scope peers — still resolve without a spurious 422. That keeps the
   Dreamer fix intact, since it reads through `get_peer`.

2. Existing reserved-prefix squatters could not be updated. The name-based guard
   on `PUT /peers/{peer_id}` refused every `scope.` name, contradicting the
   invariant that an unflagged squatter stays a normal peer. Now flag-based, so
   behavior is three-way: a real scope is refused, an existing unflagged peer
   updates, and a missing reserved-prefix name is refused by (1) rather than
   minted.

3. Scope checks raced with get-or-create and the membership upsert. The
   route-level guards run before peers are resolved, so a scope created
   concurrently in that window would be attached by the generic path with a
   default `SessionPeerConfig()`, clobbering its observer membership config.
   Adds `_reject_resolved_scope_peers`, which runs on the resolved rows in the
   same transaction as the upsert — no window, no extra query. The early checks
   stay for better error messages.

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

* fix(scopes): guard scope membership config, move checks to the mutation point

Addresses a second review pass against 10655792.

1. Scope membership configuration was directly user-mutable.
   `PUT /sessions/{id}/peers/{peer_id}/config` had no scope guard at all, and
   `crud.set_peer_config` resolved the peer only to discard the row. Confirmed:
   posting `{"observe_others": false, "observe_me": true}` for a scope returned
   204 and persisted, which silently stops all fan-out into the scope and makes
   Honcho form a representation *of* a scope — neither of which is reachable
   through the facade. Deterministic, no race required. Now checked on the row
   `get_peer` already returns, so it costs nothing and cannot race.

2. Empty and over-long names were still 500s. Removing the charset pattern from
   `PeerSpec` fixed one trap but left its length bounds, and request-bound peer
   names carry no length limits of their own — so `peer_id: ""` or a 513-char
   name reached `PeerSpec(...)` and raised a raw pydantic ValidationError that
   the catch-all turned into a 500. `PeerSpec` now carries no constraints at all
   (matching its documented purpose) and every rule for a new name lives in
   `_validate_new_peer_names` on the insert path.

3. Resolved-row protection generalized. The previous pass applied it only to
   membership upserts, leaving check-then-use windows elsewhere: peer update
   could have a concurrently-created scope's configuration replaced wholesale
   (create-path validation does not fire for a peer that now exists), the chat
   observer get-or-create could resolve a fresh scope as its observer, and the
   generic session-peer removal could silently detach a scope from its sessions.
   Each now inspects the resolved peer immediately before acting; the redundant
   name-level guard on the update route is dropped in favor of the race-free one.

`remove_peers_from_session` grows an internal `_allow_scope_peers` flag because
the scopes facade ends membership through that same path and must not be blocked
by its own guard.

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

* test(scopes): enumerate every peer-touching route against a scope policy

Three review passes each found the same class of defect: a route nobody had
checked, rather than logic that was subtly wrong. One of them — `PUT
/sessions/{id}/peers/{id}/config`, which let any caller set a scope to
`observe_others=false` and silently stop all fan-out into it — predated this work
entirely, because the guardrail set was assembled guardrail-by-guardrail instead
of derived from the route list. Sampling review cannot close that kind of gap;
enumeration can.

Derives every route through which a peer name can reach the system and requires
each to be classified as GUARDED or EXEMPT-with-a-reason, so a newly added
peer-touching route fails the suite until someone classifies it. Detection is the
union of path shape and a walk of the dependant tree (including sub-dependency
`Form(...)` params and nested request-body models), because neither signal alone
suffices: parameter names miss `POST /sessions/{id}/peers`, whose peer names are
dict keys, and path shape misses `messages/upload`, whose `peer_id` arrives as a
form field behind a parser dependency.

Both invariants are then asserted behaviorally, by calling the routes rather than
inspecting annotations — the guards deliberately live in crud, which is what makes
`messages/upload` guarded for free via `crud.create_messages`:

- a real scope is refused on all 11 guarded routes, and the rejection must name
  the scope, so an unrelated 422 (a malformed body) cannot pass the assertion;
- an *unflagged* peer merely occupying the reserved namespace is unaffected. That
  half regressed once already when `update_peer` used a name-based check.

Mutation-tested all three failure modes: disabling the `set_peer_config` guard
fails the guarded test naming that route; regressing `update_peer` to name-based
fails the squatter test; adding an unclassified peer route fails the enumeration.

Covers the HTTP surface only. Peer names also reach the system through the
deriver, dreamer, and queue, which have no route table to enumerate — noted in
the module docstring rather than implied to be covered.

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

* fix(scopes): refuse a scope in every observed position; enumerate per position

Addresses a fourth review pass against 62681e4d. The headline finding is that the
previous commit's enumeration test had the wrong *model*, not a missing entry.

1. Manual conclusions could create knowledge about a scope. `POST /conclusions`
   validated only that observer_id and observed_id exist, so a scope as
   `observed_id` persisted a conclusion about a peer carrying observe_me=false and
   created an (observer, scope) collection for it. Confirmed: 201, and it read
   back. `POST /schedule_dream` had the same hole via `observed`.

   The fix is positional, because the invariant is:

       A scope may be an OBSERVER. A scope may never be OBSERVED.

   A scope as `observer_id` is how scoped conclusions are stored and must keep
   working (verified still 201); as `observed_id` it is now refused. Same split
   applied to schedule_dream `observed`, the peer-card `target` (which also covers
   a scope's self-card, since target-omitted collapses observed to peer_id), and
   session-context `peer_target`.

2. Chat target and both representation roles kept check-to-use races. Only the
   chat path-level observer was re-checked on its resolved row; the target was
   checked by name and then resolved without inspecting scope identity. Both are
   now checked at the dialectic preflight, where observer and observed are already
   resolved — an absent name has already failed by then, and an existing squatter
   cannot retroactively become a scope.

3. Generic membership removal was still racy. The adjacent SELECT narrowed the
   window but could not close it under READ COMMITTED. The UPDATE now carries its
   own correlated NOT EXISTS against scope_peer_clause(), so Postgres evaluates
   the exclusion as part of the statement and a scope committed after the advisory
   check still cannot be detached.

4. New-name validation ran after the name reached Postgres. A NUL byte passed the
   request schemas and PeerSpec, then raised psycopg.DataError inside the lookup —
   a 500. Values that cannot correspond to a stored row by construction (NUL
   bytes, over-length names) are now refused before the query.
   (Over-length names already returned 422; only the wasted query was real there.)

The enumeration test is rekeyed from (method, path) to (method, path, position).
A binary per-route verdict cannot express finding 1 at all: `POST /conclusions` is
one route with two positions and opposite verdicts. Detection widens to observer /
observed / target / peer_target / peer_perspective, which surfaced four routes the
previous version never saw — conclusions, schedule_dream, queue/status, and
session context.

Also registers `src.routers.workspaces.tracked_db` in the conftest patch list; the
new guard there would otherwise have run against the real configured database
instead of the per-test one.

Mutation-tested: disabling the conclusions observed-guard fails the positional
test naming that position; adding an unclassified `observed_id` param fails
enumeration.

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

* fix(scopes): refuse future scopes in observed positions, preserve scope membership

Addresses a fifth review pass against 14136e5b. All six findings reproduced
locally before fixing.

1. High — generic peer replacement removed scope memberships.
   `set_peers_for_session` soft-deleted every active SessionPeer row, and the
   request-level guard only inspected names *present* in the replacement map. A
   caller detached a scope by simply omitting it, never naming it — so no
   request-level guard could ever see it. Reproduced: scope sessions went
   ['<id>'] -> [] on a 200. The exclusion now lives in the UPDATE itself
   (correlated NOT EXISTS against scope_peer_clause), so replacement means
   "replace ordinary peers" regardless of request contents or concurrent creation.

2. High — peer cards could be pre-seeded for future scopes.
   `set_peer_card` resolves only the observer and writes a JSONB key derived from
   an unchecked observed name, and the route guard rejected only *existing*
   flagged scopes. Reproduced: PUT card with target=scope.<missing> returned 200,
   creating that scope then returned 201, and the card described the real scope.

3. High — dreams could be queued for future scopes.
   The route checked `observed` in a read-only session that closed before
   `enqueue_dream`, and a missing reserved name passes any is-it-a-scope check.
   Reproduced: 204 with observed=scope.<missing>.

   2 and 3 share a root cause, so they share a fix: a new `reject_scope_observed`
   that is stricter than `reject_scope_peers` in exactly one case — a *missing*
   reserved name is refused, because nothing on these paths creates the peer, so
   nothing else would ever catch it. Existing unflagged squatters still pass.
   Both guards moved to the mutation point: card validation into
   `crud.set_peer_card` (same transaction as the JSONB write, so Dreamer and
   agent-tool callers are covered), dream validation into `enqueue_dream` (same
   transaction as the queue insert). The redundant route-level checks are dropped
   rather than left as weaker duplicates.

4. Medium — prefixed NUL names still reached PostgreSQL.
   `reject_scope_peers` filtered for the reserved prefix and sent matches to a
   text comparison, so "scope.future\0name" raised psycopg.DataError — a 500.
   Both guards now share `_reserved_name_candidates`, which materializes the input
   once and rejects impossible values before any SQL. Materializing matters
   independently: the message-author path passes a generator, and validation
   iterates separately from the prefix filter, so a generator would be
   half-consumed. `_reject_impossible_peer_names` now takes a Collection so the
   type checker enforces that.

5. Medium — representation kept a check-to-use race.
   The previous commit claimed both representation roles were rechecked after
   resolution; that was wrong — only the dialectic preflight got that check, and
   the representation route never goes through it. It now opens one short
   read-only session *after* the embedding call, checks both positions, and passes
   that same session to `get_working_representation`, so no connection is held
   across external work and a scope committed later cannot have conclusions in the
   collection being read.

6. Low — policy coverage was not exhaustive. `sender_id` reaches CRUD as
   `observed` but was missing from the detected parameter set. ALLOW cases could
   also not carry builders, so the suite never proved the other half of the
   contract — that legitimate scope *observers* keep working, which a guard
   rejecting scopes everywhere would satisfy. Both fixed; observer positions on
   conclusions, dreams, cards, session context and queue status are now asserted
   behaviorally.

Deliberately not implemented: the scope-creation backstop scanning for
pre-existing card keys and queue items naming a future backing peer. Reasoning is
recorded in `get_or_create_scopes` — no new such state can be created now, any
pre-existing row is coincidental since `scope.` was never a meaningful namespace,
the consequence is inert, and detecting card keys means a full table scan per
scope creation.

Mutation-tested each new guard: removing the replacement exclusion fails both
membership-preservation tests; weakening either observed guard to existing-only
fails the pre-seeding tests.

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

* fix(scopes): exclude scope memberships from the session observer limit

Scope memberships carry observe_others=true, so every scope counted against
SESSION_OBSERVERS_LIMIT (default 10) — capping scopes-per-session at the limit
minus the session's real observers, and reporting the failure as
`400 Cannot create session <name> with 11 observers. ... Observers are peers
with 'observe_others' set to true.` on a membership call. Wrong on three counts:
the ceiling is undocumented and contradicts RFC §5.1 ("sessions belong to any
number of scopes"), the message describes session creation, and it leaks the
word "observer" through a facade whose entire job is hiding observers (RFC
goal 5). The limit exists to bound per-observer deriver fan-out for real peers;
a scope costs document rows, not LLM calls (RFC §5.2), so it does not belong in
that budget.

Excluded from both halves of the check in `_get_or_add_peers_to_session`: the
incoming names via a flag-based lookup, existing memberships via a correlated
NOT EXISTS on `scope_peer_clause()` — the same pattern the replacement and
removal paths already use, so the exclusion holds regardless of concurrent
scope creation. The early `count_observers_in_config(session.peer_names)` check
in `get_or_create_session` is left alone: `peer_names` cannot contain a scope,
and `scopes` is a separate field.

`reject_scope_peers` is split into a `scope_peer_names()` query helper plus a
two-line raiser so the observer count reuses the authoritative name-AND-flag
predicate instead of growing a third copy of it. Still costs nothing on the
common path — no reserved-prefix name in the input means no query at all.

Also caps `SessionCreate.scopes` at 100, matching `ScopeSessionsAdd.session_ids`.
This belongs in the same commit: the observer limit was the only thing bounding
that list, so removing it turns an unbounded `scopes` array into a peer row and
a membership row per element, committed — the single-request path to the
cardinality anti-pattern RFC §8 warns about. Partly answers OQ6: no per-session
cap, 100 per request.

Tests: a session joins SESSION_OBSERVERS_LIMIT + 2 scopes through both the
facade and session creation; real observers over the limit still 400, so the
carve-out cannot quietly disable the limit; 101 scopes is a 422.

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

* fix(scopes): skip semantic retrieval when the embedding precompute failed

Addresses three open review comments.

1. Major — the representation read could embed inside its DB session. The
   route's precompute is suppressed, and both
   `RepresentationManager.get_working_representation` and
   `crud.query_documents` fall back to embedding when a query arrives without
   one, so a failed precompute meant an external call inside the read session
   this branch opens for the scope re-check — the connection-holding rule the
   route's own comment claimed to satisfy. The innermost fallback also only
   catches ValueError, so a provider outage surfaced as a 500. The semantic
   query is now passed only when an embedding exists, degrading to
   derived+recent retrieval. (`crud.query_documents` embedding inside a caller's
   session predates this branch and is left alone.)

2. Minor — `test_resolved_scope_peer_rejected_at_membership_upsert` described a
   race it does not perform. It creates an already-flagged scope and calls crud
   directly; the unflagged → flagged transition is not simulated. Docstring now
   says what the test actually pins.

3. Minor — `test_empty_replacement_preserves_scope_membership` asserted only
   half its docstring. It passed if the empty PUT left every ordinary
   membership intact; now asserts the ordinary peer's left_at is set.

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

* fix(scopes): close auth and observed-position gaps, paginate membership

Review response for #884.

Security:
- gate `SessionCreate.scopes` behind a workspace-level key; the session-create
  route is self-authorizing, so a peer- or session-scoped token could mint scope
  peers and join sessions to scopes it had no access to via `POST /scopes`
- refuse a reserved-but-nonexistent name in two observed positions that used the
  permissive guard: chat `target` and session-context `peer_target`. Both let a
  caller act on `scope.X` before it existed, then create the scope

Facade:
- exclude scope peers from `GET /sessions/{id}/peers` and refuse the membership
  -config read for a real scope, matching its write side
- replace `GET /scopes/{id}/sessions` with `POST /scopes/{id}/sessions/list`
  returning `Page[Session]`; the add route now returns 204. Membership was
  unbounded on both, while every other list surface paginates
- rename `crud.get_scope` to `get_scope_or_raise`

Tests:
- add a missing-name axis to the route-policy table (`Case.refuse_missing`), which
  is what surfaced the two guard gaps above
- delete 14 hand-written tests the table now enumerates; 52 -> 39 functions in
  test_scopes.py with more cases covered
- tighten the squatter assertion from `!= 422` to `< 400`, which was passing on 5xx
- assert the FastAPI-internals traversal still derives positions, so a framework
  upgrade can't silently empty the suite

Docs:
- drop internal ticket and RFC references from the published OpenAPI descriptions
  and surrounding comments; state the behavior instead
- move implementation reasoning out of the `PUT /peers/{id}` docstring, which
  FastAPI publishes, into a comment

* test(scopes): assert exact statuses for permissive missing-name cases

Follow-up review pass on #884.

- add `Case.missing_status` so a permissive missing-name position asserts the
  status it should actually get (404, or 200 for the no-op removal) instead of
  `!= 422`, which also passed on a 5xx — the same hole already closed in the
  squatter assertion
- require it whenever `refuse_missing` is False, and require its absence when
  True, so the policy table can't drift from the assertion
- repoint a stale allow-reason at POST /scopes/{scope_id}/sessions/list; the GET
  it named was removed
- document the membership list's ordering under `reverse`

* chore: clean up stale docstring language

* fix: don't backfill a session that left the scope

scope_backfill and scope_removal carry different work-unit keys, so
nothing orders them: a removal enqueued right after the add — or one
that lands while the backfill is embedding — sweeps the scope before
the copies exist, leaving a departed session's documents live in the
scope forever.

_run_backfill now re-checks SessionPeer membership inside the write
transaction and returns None; process_scope_backfill then skips both
the dream enqueues and the status write, so a skipped backfill can't
resurrect the status entry removal just cleared.

Adds coverage for the skip, the NULL-embedding re-embed path, and the
failed-status write. Handler-driven tests now stand up the membership
row the guard requires.

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

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

src/crud/__init__.py

@@ -49,12 +49,16 @@ from .representation import (
 )
 from .scope import (
     add_sessions_to_scope,
+    clear_scope_backfill_status,
     get_or_create_scopes,
+    get_scope_backfill_status,
     get_scope_or_raise,
     get_scope_sessions,
     get_scopes,
+    invalidate_scope_peer_cache,
     remove_session_from_scope,
     resolve_scope_peers,
+    update_scope_backfill_status,
 )
 from .session import (
     SessionDeletionResult,
@@ -137,12 +141,16 @@ __all__ = [
     "get_working_representation",
     # Scope
     "add_sessions_to_scope",
+    "clear_scope_backfill_status",
     "get_or_create_scopes",
+    "get_scope_backfill_status",
     "get_scope_or_raise",
     "get_scope_sessions",
     "get_scopes",
+    "invalidate_scope_peer_cache",
     "remove_session_from_scope",
     "resolve_scope_peers",
+    "update_scope_backfill_status",
     # Session
     "SessionDeletionResult",
     "get_sessions",

src/crud/scope.py

@@ -7,17 +7,26 @@ authoritative, user-unwritable flag) and ``{"observe_me": false}`` in
 and never speaks.
 See ``src/utils/scopes.py`` for the namespace helpers.
 
-Membership only affects messages ingested *after* a session is added to a
-scope. Conclusions already derived are neither backfilled on add nor
-reconciled on removal.
+Messages ingested after a membership change flow to the scope via the normal
+deriver fan-out. Retroactive changes are handled by queue jobs: adding a session
+that already has messages enqueues a ``scope_backfill`` task (copy of the
+session's explicit documents into the scope's collections) and removing a
+session enqueues a ``scope_removal`` task (soft-delete of the session's
+documents plus dependent derived documents).
 """
 
 from collections.abc import Sequence
+from datetime import datetime, timezone
 from logging import getLogger
+from typing import Any, Literal
+from typing import cast as py_cast
 
-from sqlalchemy import Select, select
+from sqlalchemy import Select, Text, cast, select, update
+from sqlalchemy.dialects.postgresql import ARRAY, JSONB, array
+from sqlalchemy.engine import CursorResult
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.sql.functions import func
 
 from src import models, schemas
 from src.cache.client import safe_cache_delete
@@ -38,11 +47,21 @@ from .workspace import get_or_create_workspace
 
 logger = getLogger(__name__)
 
+# Key inside the scope peer's internal_metadata that holds per-session
+# backfill job status: {<session_name>: {state, updated_at[, docs_copied]}}.
+BACKFILL_STATUS_KEY = "backfill_status"
+
+ScopeBackfillState = Literal["pending", "completed", "failed"]
+
 # Internal metadata stamped on every scope peer at creation. `kind` is the
 # authoritative scope flag and lives here — NOT in `configuration` — because
 # `configuration` is user-writable (`PeerCreate`/`PeerUpdate` accept a free-form
 # dict, and `update_peer` replaces it wholesale), so a user could forge or clear
 # the flag. `internal_metadata` appears in no API schema at all.
+#
+# Backfill job status shares this field under BACKFILL_STATUS_KEY below. Every
+# write to it must be a JSONB merge scoped to that key, never a wholesale
+# replacement, or the `kind` flag goes with it and the peer stops being a scope.
 SCOPE_PEER_INTERNAL_METADATA: dict[str, str] = {
     "kind": SCOPE_KIND,
 }
@@ -355,9 +374,9 @@ async def add_sessions_to_scope(
 
     Each membership is a ``session_peers`` row for the scope peer with
     ``observe_others=true, observe_me=false`` — exactly what a hand-built
-    observer peer would carry. No backfill happens here: membership only affects
-    messages ingested after this call, and conclusions already derived are left
-    as they are.
+    observer peer would carry. Sessions that already have messages get a
+    ``scope_backfill`` queue task so their existing explicit documents are
+    copied into the scope's collections; fresh sessions need nothing.
 
     Args:
         db: Database session
@@ -400,6 +419,23 @@ async def add_sessions_to_scope(
 
     await db.commit()
 
+    # Backfill: only sessions that already have messages need it.
+    # Imported lazily: src.deriver.enqueue imports crud at module level.
+    from src.deriver.enqueue import enqueue_scope_backfill
+
+    msg_result = await db.execute(
+        select(models.Message.session_name)
+        .where(models.Message.workspace_name == workspace_name)
+        .where(models.Message.session_name.in_(requested))
+        .distinct()
+    )
+    for session_with_messages in sorted({row[0] for row in msg_result.all()}):
+        await enqueue_scope_backfill(
+            workspace_name,
+            scope_peer=scope_peer_name(scope_name),
+            session_name=session_with_messages,
+        )
+
 
 async def remove_session_from_scope(
     db: AsyncSession,
@@ -411,8 +447,9 @@ async def remove_session_from_scope(
     Remove a session from a scope by ending the scope peer's membership.
 
     Ends the membership the same way the generic remove-peer path does (sets
-    ``left_at``). Conclusions derived while the session was a member are left in
-    place — nothing reconciles them.
+    ``left_at``), then enqueues a ``scope_removal`` reconciliation task that
+    soft-deletes the session's documents from the scope's collections along
+    with any derived documents whose support left the scope.
 
     Args:
         db: Database session
@@ -424,6 +461,8 @@ async def remove_session_from_scope(
         ResourceNotFoundException: If the scope or session does not exist
     """
     # Lazy import for the same circular-import reason as add_sessions_to_scope.
+    from src.deriver.enqueue import enqueue_scope_removal
+
     from .session import remove_peers_from_session
 
     await get_scope_or_raise(db, workspace_name, scope_name)
@@ -436,3 +475,124 @@ async def remove_session_from_scope(
         # This *is* the supported path for ending scope membership.
         _allow_scope_peers=True,
     )
+
+    await enqueue_scope_removal(
+        workspace_name,
+        scope_peer=scope_peer_name(scope_name),
+        session_name=session_name,
+    )
+
+
+async def update_scope_backfill_status(
+    db: AsyncSession,
+    workspace_name: str,
+    scope_peer: str,
+    session_name: str,
+    *,
+    state: ScopeBackfillState,
+    docs_copied: int | None = None,
+) -> None:
+    """
+    Merge one session's backfill status into the scope peer's internal_metadata.
+
+    Uses a single-statement nested JSONB merge so concurrent writers updating
+    *different* session keys never clobber each other (the merge is computed
+    from the row's current committed value under the row lock, never from a
+    stale Python-side read):
+
+        internal_metadata || {"backfill_status":
+            coalesce(internal_metadata->'backfill_status', '{}') || {<session>: <entry>}}
+
+    Does not commit; the caller owns the transaction. Callers should
+    invalidate the peer cache after committing (``peer_cache_key``).
+    """
+    entry: dict[str, Any] = {
+        "state": state,
+        "updated_at": datetime.now(timezone.utc).isoformat(),
+    }
+    if docs_copied is not None:
+        entry["docs_copied"] = docs_copied
+
+    # NOTE: the JSONB operand must be a Python dict, not a json.dumps() string.
+    # cast(<already-serialized-str>, JSONB) double-encodes: psycopg's JSONB
+    # bind adapter serializes the *string* again, producing a JSONB string
+    # scalar instead of an object. `||` between two non-array jsonb scalars
+    # doesn't merge keys — it silently wraps both sides into a 2-element
+    # array, corrupting backfill_status into a list and later crashing
+    # clear_scope_backfill_status's `#-` path delete (which then sees an
+    # array where it expects an object).
+    merged_status = func.coalesce(
+        models.Peer.internal_metadata.op("->")(BACKFILL_STATUS_KEY),
+        cast({}, JSONB),
+    ).op("||")(cast({session_name: entry}, JSONB))
+
+    stmt = (
+        update(models.Peer)
+        .where(models.Peer.workspace_name == workspace_name)
+        .where(models.Peer.name == scope_peer)
+        .values(
+            internal_metadata=models.Peer.internal_metadata.op("||")(
+                func.jsonb_build_object(BACKFILL_STATUS_KEY, merged_status)
+            )
+        )
+    )
+    result = py_cast(CursorResult[Any], await db.execute(stmt))
+    if result.rowcount == 0:
+        raise ResourceNotFoundException(
+            f"Scope peer {scope_peer} not found in workspace {workspace_name}"
+        )
+
+
+async def clear_scope_backfill_status(
+    db: AsyncSession,
+    workspace_name: str,
+    scope_peer: str,
+    session_name: str,
+) -> None:
+    """
+    Drop one session's entry from the scope peer's backfill status.
+
+    Called by the removal reconciliation job: once a session leaves a scope its
+    backfill status is moot, and clearing it keeps add→remove→re-add cycles
+    honest (the re-add starts from a fresh ``pending``). Single-statement
+    ``#-`` delete, safe against concurrent per-key merges. Does not commit.
+    Missing peers are a no-op (removal is idempotent).
+    """
+    stmt = (
+        update(models.Peer)
+        .where(models.Peer.workspace_name == workspace_name)
+        .where(models.Peer.name == scope_peer)
+        .values(
+            internal_metadata=models.Peer.internal_metadata.op("#-")(
+                cast(array([BACKFILL_STATUS_KEY, session_name]), ARRAY(Text))
+            )
+        )
+    )
+    await db.execute(stmt)
+
+
+async def invalidate_scope_peer_cache(workspace_name: str, scope_peer: str) -> None:
+    """Invalidate the cached peer row after a status write commits."""
+    await safe_cache_delete(peer_cache_key(workspace_name, scope_peer))
+
+
+async def get_scope_backfill_status(
+    db: AsyncSession,
+    workspace_name: str,
+    scope_name: str,
+) -> dict[str, Any]:
+    """
+    Read the per-session backfill job status map for a scope.
+
+    Returns:
+        ``{<session_name>: {state, updated_at[, docs_copied]}}`` (empty when no
+        backfill has ever been enqueued for the scope)
+
+    Raises:
+        ResourceNotFoundException: If the scope does not exist
+    """
+    peer = await get_scope_or_raise(db, workspace_name, scope_name)
+    status = peer.internal_metadata.get(BACKFILL_STATUS_KEY, {})
+    if not isinstance(status, dict):
+        return {}
+    return py_cast(dict[str, Any], status)

src/crud/session.py

@@ -320,8 +320,8 @@ async def get_or_create_session(
 
     # Add the session to any requested scopes: create-or-get each scope peer
     # and record an observer membership (observe_others=true, observe_me=false).
-    # No backfill happens here — membership only affects messages ingested
-    # after this point.
+    # If the session already has messages, a backfill task is enqueued after
+    # commit (below) so its existing documents are copied into the scope.
     scopes_result = None
     if session.scopes:
         scopes_result = await get_or_create_scopes(
@@ -352,6 +352,30 @@ async def get_or_create_session(
     if scopes_result is not None:
         await scopes_result.post_commit()
 
+    # Backfill (DEV-1999): a pre-existing session added to scopes at
+    # create-or-get time may already have messages; those need a
+    # backfill-by-copy task per scope. Fresh sessions need nothing.
+    if session.scopes:
+        has_messages = await db.scalar(
+            select(
+                exists(
+                    select(models.Message.id)
+                    .where(models.Message.workspace_name == workspace_name)
+                    .where(models.Message.session_name == session.name)
+                )
+            )
+        )
+        if has_messages:
+            # Imported lazily: src.deriver.enqueue imports crud at module level.
+            from src.deriver.enqueue import enqueue_scope_backfill
+
+            for scope_name in session.scopes:
+                await enqueue_scope_backfill(
+                    workspace_name,
+                    scope_peer=scope_peer_name(scope_name),
+                    session_name=session.name,
+                )
+
     # Only update cache if session data changed or was newly created
     if needs_cache_update:
         cache_key = session_cache_key(workspace_name, session.name)

src/deriver/consumer.py

@@ -8,6 +8,10 @@ from sqlalchemy import select
 from src import crud, models
 from src.dependencies import tracked_db
 from src.deriver.deriver import process_representation_tasks_batch
+from src.deriver.scope_backfill import (
+    process_scope_backfill,
+    process_scope_removal,
+)
 from src.dreamer import process_dream
 from src.exceptions import ResourceNotFoundException, ValidationException
 from src.models import Message
@@ -26,6 +30,8 @@ from src.utils.queue_payload import (
     DeletionPayload,
     DreamPayload,
     ReconcilerPayload,
+    ScopeBackfillPayload,
+    ScopeRemovalPayload,
     SummaryPayload,
     WebhookPayload,
 )
@@ -150,6 +156,36 @@ async def process_item(queue_item: models.QueueItem) -> None:
                 raise ValueError(f"Invalid payload structure: {str(e)}") from e
             await process_deletion(validated, workspace_name)
 
+    elif task_type == "scope_backfill":
+        with sentry_sdk.start_transaction(
+            name="process_scope_backfill_task", op="deriver"
+        ):
+            try:
+                validated = ScopeBackfillPayload(**queue_payload)
+            except ValidationError as e:
+                logger.error(
+                    "Invalid scope_backfill payload received: %s. Payload: %s",
+                    str(e),
+                    queue_payload,
+                )
+                raise ValueError(f"Invalid payload structure: {str(e)}") from e
+            await process_scope_backfill(validated, workspace_name)
+
+    elif task_type == "scope_removal":
+        with sentry_sdk.start_transaction(
+            name="process_scope_removal_task", op="deriver"
+        ):
+            try:
+                validated = ScopeRemovalPayload(**queue_payload)
+            except ValidationError as e:
+                logger.error(
+                    "Invalid scope_removal payload received: %s. Payload: %s",
+                    str(e),
+                    queue_payload,
+                )
+                raise ValueError(f"Invalid payload structure: {str(e)}") from e
+            await process_scope_removal(validated, workspace_name)
+
     else:
         raise ValueError(f"Invalid task type: {task_type}")
 

src/deriver/enqueue.py

@@ -16,6 +16,7 @@ from src.utils.queue_payload import (
     create_deletion_payload,
     create_dream_payload,
     create_payload,
+    create_scope_task_payload,
 )
 from src.utils.work_unit import construct_work_unit_key
 
@@ -569,6 +570,168 @@ async def enqueue_dream(
             raise
 
 
+def create_scope_task_record(
+    workspace_name: str,
+    *,
+    task_type: Literal["scope_backfill", "scope_removal"],
+    scope_peer: str,
+    session_name: str,
+) -> dict[str, Any]:
+    """
+    Create a queue record for a scope backfill / removal task (DEV-1999).
+
+    Args:
+        workspace_name: Name of the workspace
+        task_type: "scope_backfill" or "scope_removal"
+        scope_peer: Prefixed name of the peer backing the scope
+        session_name: Name of the session whose membership changed
+
+    Returns:
+        Queue record dictionary ready for insertion into the queue
+    """
+    payload = create_scope_task_payload(
+        task_type, scope_peer=scope_peer, session_name=session_name
+    )
+    return {
+        "work_unit_key": construct_work_unit_key(workspace_name, payload),
+        "payload": payload,
+        "session_id": None,
+        "task_type": task_type,
+        "workspace_name": workspace_name,
+        "message_id": None,
+    }
+
+
+async def _enqueue_scope_task(
+    workspace_name: str,
+    *,
+    task_type: Literal["scope_backfill", "scope_removal"],
+    scope_peer: str,
+    session_name: str,
+) -> None:
+    """
+    Enqueue a scope backfill / removal task, deduplicating like enqueue_dream.
+
+    If a task with the same work_unit_key is already in progress (has an
+    ActiveQueueSession) or pending in the queue, the enqueue is skipped — the
+    handlers are idempotent, so a queued task already covers this membership
+    change.
+
+    For backfill tasks, the scope peer's per-session status entry is written
+    to "pending" in the same transaction as the queue insert, so the status
+    surface never claims a job exists that was never enqueued (or vice versa).
+    """
+    async with tracked_db("scope_task_enqueue") as db_session:
+        try:
+            record = create_scope_task_record(
+                workspace_name,
+                task_type=task_type,
+                scope_peer=scope_peer,
+                session_name=session_name,
+            )
+            work_unit_key = record["work_unit_key"]
+
+            is_in_progress = await db_session.scalar(
+                select(
+                    exists(
+                        select(models.ActiveQueueSession.id).where(
+                            models.ActiveQueueSession.work_unit_key == work_unit_key
+                        )
+                    )
+                )
+            )
+            if is_in_progress:
+                logger.debug(
+                    "Skipping %s enqueue - already in progress: %s",
+                    task_type,
+                    work_unit_key,
+                )
+                return
+
+            is_pending = await db_session.scalar(
+                select(
+                    exists(
+                        select(QueueItem.id).where(
+                            QueueItem.work_unit_key == work_unit_key,
+                            QueueItem.processed == False,  # noqa: E712
+                        )
+                    )
+                )
+            )
+            if is_pending:
+                logger.debug(
+                    "%s already pending in queue: %s", task_type, work_unit_key
+                )
+                return
+
+            stmt = insert(QueueItem).returning(QueueItem)
+            await db_session.execute(stmt, [record])
+
+            if task_type == "scope_backfill":
+                await crud.update_scope_backfill_status(
+                    db_session,
+                    workspace_name,
+                    scope_peer,
+                    session_name,
+                    state="pending",
+                )
+
+            await db_session.commit()
+            await crud.invalidate_scope_peer_cache(workspace_name, scope_peer)
+
+            logger.info(
+                "Enqueued %s task for %s/%s/%s",
+                task_type,
+                workspace_name,
+                scope_peer,
+                session_name,
+            )
+
+        except Exception as e:
+            logger.exception("Failed to enqueue %s task!", task_type)
+            if settings.SENTRY.ENABLED:
+                import sentry_sdk
+
+                sentry_sdk.capture_exception(e)
+            raise
+
+
+async def enqueue_scope_backfill(
+    workspace_name: str,
+    *,
+    scope_peer: str,
+    session_name: str,
+) -> None:
+    """
+    Enqueue a backfill-by-copy task for a session newly added to a scope.
+
+    Only call this for sessions that already have messages — fresh sessions
+    need nothing (the deriver fan-out covers everything ingested after the
+    membership change).
+    """
+    await _enqueue_scope_task(
+        workspace_name,
+        task_type="scope_backfill",
+        scope_peer=scope_peer,
+        session_name=session_name,
+    )
+
+
+async def enqueue_scope_removal(
+    workspace_name: str,
+    *,
+    scope_peer: str,
+    session_name: str,
+) -> None:
+    """Enqueue a removal reconciliation task for a session removed from a scope."""
+    await _enqueue_scope_task(
+        workspace_name,
+        task_type="scope_removal",
+        scope_peer=scope_peer,
+        session_name=session_name,
+    )
+
+
 def create_deletion_record(
     workspace_name: str,
     deletion_type: Literal["session", "observation", "workspace"],

src/deriver/scope_backfill.py

@@ -0,0 +1,558 @@
+"""Scope membership reconciliation jobs (DEV-1999, part of the Scopes RFC).
+
+Two queue task handlers, dispatched from ``consumer.process_item``:
+
+- ``scope_backfill`` — a session was added to a scope that already had
+  messages. Copy the session's *explicit* documents from each sender peer's
+  global ``(P, P)`` collection into the scope's ``(scope_peer, P)`` collection,
+  then enqueue a manual omni dream per touched collection to rebuild the
+  scope's higher-order layer and card.
+- ``scope_removal`` — a session was removed from a scope. Soft-delete the
+  session's explicit documents from the scope's collections, cascade the
+  soft-delete to derived documents whose support left the scope (fail-closed),
+  then enqueue a ``card_refresh`` dream with ``rebuild=True`` plus a manual
+  omni dream per touched collection.
+
+Zero LLM re-derivation: explicit-level documents are session-pure and
+identical across observer collections (the DEV-2000 invariant), so retroactive
+membership is pure row copying. The only external call is an embedding lookup
+for source rows whose embedding column is NULL — embedding API only, never an
+LLM.
+
+DB sessions are never held across embedding or vector-store calls: the
+handlers run in phases (plan → embed → write → sync), each phase opening its
+own short-lived session.
+"""
+
+import logging
+from dataclasses import dataclass
+from typing import Any
+
+from sqlalchemy import select, update
+from sqlalchemy.dialects.postgresql import array
+from sqlalchemy.sql.functions import func
+
+from src import crud, models
+from src.config import settings
+from src.crud.scope import ScopeBackfillState
+from src.crud.session import is_peer_in_session
+from src.dependencies import tracked_db
+from src.embedding_client import embedding_client
+from src.schemas import DreamType
+from src.utils.queue_payload import ScopeBackfillPayload, ScopeRemovalPayload
+from src.utils.scopes import is_scope_peer_name
+from src.vector_store import VectorRecord, get_external_vector_store
+
+logger = logging.getLogger(__name__)
+
+# internal_metadata key linking a backfilled copy to the global document it
+# was copied from. The presence of this key is the idempotency marker.
+COPIED_FROM_KEY = "copied_from"
+
+
+def _store_embeddings_in_postgres() -> bool:
+    """Whether document embeddings are persisted to the postgres column.
+
+    True when TYPE=pgvector OR still migrating (dual-write) — mirrors
+    ``crud.document.create_documents``.
+    """
+    return (
+        settings.VECTOR_STORE.TYPE == "pgvector" or not settings.VECTOR_STORE.MIGRATED
+    )
+
+
+def _embedding_as_list(embedding: Any) -> list[float] | None:
+    """Normalize a pgvector column value (numpy array or None) to a list."""
+    if embedding is None:
+        return None
+    return list(embedding)
+
+
+@dataclass
+class _CopySpec:
+    """A planned copy of one global explicit document into a scope collection.
+
+    ``restore_document_id`` points at an existing soft-deleted copy to restore
+    when set; for new copies it is filled with the inserted row's id after the
+    write phase.
+    """
+
+    observed: str
+    source_id: str
+    content: str
+    embedding: list[float] | None
+    internal_metadata: dict[str, Any]
+    times_derived: int
+    source_ids: list[str] | None
+    session_name: str
+    restore_document_id: str | None = None
+
+
+# ---------------------------------------------------------------------------
+# Backfill (session added to a scope)
+# ---------------------------------------------------------------------------
+
+
+async def process_scope_backfill(
+    payload: ScopeBackfillPayload, workspace_name: str
+) -> None:
+    """Process a ``scope_backfill`` queue task.
+
+    Idempotent: a live copy (matched by ``internal_metadata.copied_from``) is
+    never duplicated, and a soft-deleted copy left by an earlier removal is
+    restored rather than re-inserted, so add→remove→re-add converges on
+    exactly one live copy per source document.
+    """
+    scope_peer = payload.scope_peer
+    session_name = payload.session_name
+
+    try:
+        result = await _run_backfill(workspace_name, scope_peer, session_name)
+    except Exception:
+        await _write_backfill_status(
+            workspace_name, scope_peer, session_name, state="failed"
+        )
+        raise
+
+    if result is None:
+        # The session left the scope before this task was drained: removal
+        # already cleared the copies and the status entry, so there is nothing
+        # to backfill and nothing to record.
+        logger.info(
+            "Scope backfill skipped for %s/%s/%s: membership is no longer active",
+            workspace_name,
+            scope_peer,
+            session_name,
+        )
+        return
+    docs_copied, touched_observed = result
+
+    # One manual (gate-bypassing) omni dream per touched collection builds the
+    # scope's higher-order layer and card. enqueue_dream dedupes on the
+    # work-unit key, so a batch of backfills collapses to one dream each.
+    from src.deriver.enqueue import enqueue_dream
+
+    for observed in sorted(touched_observed):
+        await enqueue_dream(
+            workspace_name,
+            observer=scope_peer,
+            observed=observed,
+            dream_type=DreamType.OMNI,
+            trigger_reason="scope_backfill",
+        )
+
+    await _write_backfill_status(
+        workspace_name,
+        scope_peer,
+        session_name,
+        state="completed",
+        docs_copied=docs_copied,
+    )
+
+    logger.info(
+        "Scope backfill complete for %s/%s/%s: %d documents copied across %d collections",
+        workspace_name,
+        scope_peer,
+        session_name,
+        docs_copied,
+        len(touched_observed),
+    )
+
+
+async def _run_backfill(
+    workspace_name: str, scope_peer: str, session_name: str
+) -> tuple[int, set[str]] | None:
+    """Run the copy itself.
+
+    Returns (docs copied or restored, touched observed peers), or ``None`` if
+    the session is no longer a member of the scope.
+    """
+    # Phase 1 (DB): plan. Read the session's explicit documents from each
+    # sender's global (P, P) collection, and any existing copies (live or
+    # soft-deleted) already in the scope's collections.
+    plans: list[_CopySpec] = []
+    async with tracked_db("scope_backfill.plan") as db:
+        source_result = await db.execute(
+            select(models.Document).where(
+                models.Document.workspace_name == workspace_name,
+                models.Document.session_name == session_name,
+                models.Document.level == "explicit",
+                models.Document.observer == models.Document.observed,
+                models.Document.deleted_at.is_(None),
+            )
+        )
+        source_docs = [
+            doc
+            for doc in source_result.scalars().all()
+            # Scope peers never speak and are never observed, but stay
+            # defensive: never treat another scope's rows as a source.
+            if not is_scope_peer_name(doc.observer)
+        ]
+
+        if not source_docs:
+            return 0, set()
+
+        # Existing copies in the scope's collections for this session, keyed
+        # by (observed, copied_from). Includes soft-deleted rows: those are
+        # restore candidates, not blockers.
+        copies_result = await db.execute(
+            select(models.Document).where(
+                models.Document.workspace_name == workspace_name,
+                models.Document.observer == scope_peer,
+                models.Document.session_name == session_name,
+                models.Document.internal_metadata.has_key(COPIED_FROM_KEY),
+            )
+        )
+        live_copies: set[tuple[str, str]] = set()
+        soft_deleted_copies: dict[tuple[str, str], str] = {}
+        for copy_doc in copies_result.scalars().all():
+            key = (copy_doc.observed, str(copy_doc.internal_metadata[COPIED_FROM_KEY]))
+            if copy_doc.deleted_at is None:
+                live_copies.add(key)
+            else:
+                soft_deleted_copies.setdefault(key, copy_doc.id)
+
+        for source in source_docs:
+            key = (source.observed, source.id)
+            if key in live_copies:
+                continue
+            plans.append(
+                _CopySpec(
+                    observed=source.observed,
+                    source_id=source.id,
+                    content=source.content,
+                    embedding=_embedding_as_list(source.embedding),
+                    internal_metadata=dict(source.internal_metadata),
+                    times_derived=source.times_derived,
+                    source_ids=list(source.source_ids)
+                    if source.source_ids is not None
+                    else None,
+                    session_name=session_name,
+                    restore_document_id=soft_deleted_copies.get(key),
+                )
+            )
+
+    if not plans:
+        return 0, set()
+
+    # Phase 2 (no DB): fill missing embeddings. Source rows have NULL
+    # embeddings on external-store deployments (and soft-deleted copies may
+    # have lost their vectors) — re-embed via the embedding API only; no LLM.
+    missing = [spec for spec in plans if spec.embedding is None]
+    if missing:
+        logger.info(
+            "Scope backfill re-embedding %d documents with NULL embeddings for %s/%s/%s (embedding API only, no LLM)",
+            len(missing),
+            workspace_name,
+            scope_peer,
+            session_name,
+        )
+        embeddings = await embedding_client.simple_batch_embed(
+            [spec.content for spec in missing]
+        )
+        for spec, embedding in zip(missing, embeddings, strict=True):
+            spec.embedding = embedding
+
+    # Phase 3 (DB): write the copies.
+    store_in_postgres = _store_embeddings_in_postgres()
+    touched_observed = {spec.observed for spec in plans}
+    new_rows: list[models.Document] = []
+    async with tracked_db("scope_backfill.write") as db:
+        # scope_backfill and scope_removal carry different work-unit keys, so
+        # nothing orders them: a removal enqueued right after the add (or one
+        # that landed while phase 2 was embedding) can sweep the scope before
+        # these copies exist. Re-checking membership here, in the transaction
+        # that inserts, keeps a removed session from being copied back in.
+        if not await is_peer_in_session(db, workspace_name, session_name, scope_peer):
+            return None
+
+        for observed in sorted(touched_observed):
+            await crud.get_or_create_collection(
+                db, workspace_name, observer=scope_peer, observed=observed
+            )
+
+        restore_ids: list[str] = []
+        for spec in plans:
+            if spec.restore_document_id is not None:
+                restore_ids.append(spec.restore_document_id)
+                continue
+            row = models.Document(
+                workspace_name=workspace_name,
+                observer=scope_peer,
+                observed=spec.observed,
+                content=spec.content,
+                level="explicit",
+                times_derived=spec.times_derived,
+                internal_metadata={
+                    **spec.internal_metadata,
+                    COPIED_FROM_KEY: spec.source_id,
+                },
+                session_name=spec.session_name,
+                source_ids=spec.source_ids,
+                embedding=spec.embedding if store_in_postgres else None,
+            )
+            row.sync_state = "pending"
+            new_rows.append(row)
+
+        db.add_all(new_rows)
+        if restore_ids:
+            await db.execute(
+                update(models.Document)
+                .where(models.Document.id.in_(restore_ids))
+                .values(deleted_at=None, sync_state="pending")
+            )
+        await db.commit()
+
+        # IDs are generated client-side and remain accessible post-commit
+        # (expire_on_commit=False), mirroring crud.document.create_documents.
+        for spec, row in zip(
+            [s for s in plans if s.restore_document_id is None], new_rows, strict=True
+        ):
+            spec.restore_document_id = row.id
+
+    copied_ids = [
+        spec.restore_document_id
+        for spec in plans
+        if spec.restore_document_id is not None
+    ]
+
+    # Phase 4: sync to the external vector store (or mark synced in pgvector
+    # mode). Failures leave rows in sync_state='pending' for the reconciler.
+    await _sync_copies_to_vector_store(workspace_name, scope_peer, plans, copied_ids)
+
+    return len(plans), touched_observed
+
+
+async def _sync_copies_to_vector_store(
+    workspace_name: str,
+    scope_peer: str,
+    plans: list[_CopySpec],
+    copied_ids: list[str],
+) -> None:
+    """Mirror the document-create sync path for the backfilled copies."""
+    external_vector_store = get_external_vector_store()
+
+    if external_vector_store is None:
+        # pgvector mode: embeddings live in the postgres column; nothing to sync.
+        async with tracked_db("scope_backfill.mark_synced") as db:
+            await db.execute(
+                update(models.Document)
+                .where(models.Document.id.in_(copied_ids))
+                .values(sync_state="synced", last_sync_at=func.now(), sync_attempts=0)
+            )
+            await db.commit()
+        return
+
+    by_observed: dict[str, list[_CopySpec]] = {}
+    for spec in plans:
+        by_observed.setdefault(spec.observed, []).append(spec)
+
+    synced_ids: list[str] = []
+    failed_ids: list[str] = []
+    for observed, specs in by_observed.items():
+        namespace = external_vector_store.get_vector_namespace(
+            "document", workspace_name, scope_peer, observed
+        )
+        records = [
+            VectorRecord(
+                id=spec.restore_document_id,
+                embedding=spec.embedding,
+                metadata={
+                    "workspace_name": workspace_name,
+                    "observer": scope_peer,
+                    "observed": observed,
+                    "session_name": spec.session_name,
+                    "level": "explicit",
+                },
+            )
+            for spec in specs
+            if spec.restore_document_id is not None and spec.embedding is not None
+        ]
+        ids = [record.id for record in records]
+        try:
+            await external_vector_store.upsert_many(namespace, records)
+            synced_ids.extend(ids)
+        except Exception:
+            logger.exception(
+                "Failed to upsert backfilled vectors to %s; leaving docs pending for the reconciler",
+                namespace,
+            )
+            failed_ids.extend(ids)
+
+    async with tracked_db("scope_backfill.sync_state") as db:
+        if synced_ids:
+            await db.execute(
+                update(models.Document)
+                .where(models.Document.id.in_(synced_ids))
+                .values(sync_state="synced", last_sync_at=func.now(), sync_attempts=0)
+            )
+        if failed_ids:
+            await db.execute(
+                update(models.Document)
+                .where(models.Document.id.in_(failed_ids))
+                .values(
+                    sync_attempts=models.Document.sync_attempts + 1,
+                    last_sync_at=func.now(),
+                )
+            )
+        await db.commit()
+
+
+# ---------------------------------------------------------------------------
+# Removal reconciliation (session removed from a scope)
+# ---------------------------------------------------------------------------
+
+
+async def process_scope_removal(
+    payload: ScopeRemovalPayload, workspace_name: str
+) -> None:
+    """Process a ``scope_removal`` queue task.
+
+    Soft-deletes (fail-closed) rather than hard-deletes: the reconciler's
+    soft-delete sweep handles vector cleanup and eventual hard deletion, and a
+    later re-add restores the same rows.
+    """
+    scope_peer = payload.scope_peer
+    session_name = payload.session_name
+
+    removed_by_observed: dict[str, list[str]] = {}
+    async with tracked_db("scope_removal") as db:
+        observed_result = await db.execute(
+            select(models.Collection.observed).where(
+                models.Collection.workspace_name == workspace_name,
+                models.Collection.observer == scope_peer,
+            )
+        )
+        for observed in [row[0] for row in observed_result.all()]:
+            explicit_stmt = (
+                update(models.Document)
+                .where(
+                    models.Document.workspace_name == workspace_name,
+                    models.Document.observer == scope_peer,
+                    models.Document.observed == observed,
+                    models.Document.session_name == session_name,
+                    models.Document.level == "explicit",
+                    models.Document.deleted_at.is_(None),
+                )
+                .values(deleted_at=func.now())
+                .returning(models.Document.id)
+            )
+            frontier = [row[0] for row in (await db.execute(explicit_stmt)).all()]
+            all_removed = list(frontier)
+
+            # Fail-closed cascade: soft-delete derived documents whose support
+            # (source_ids) intersects anything removed, transitively — a
+            # deduction resting on removed evidence must leave with it, and so
+            # must an induction resting on that deduction.
+            while frontier:
+                derived_stmt = (
+                    update(models.Document)
+                    .where(
+                        models.Document.workspace_name == workspace_name,
+                        models.Document.observer == scope_peer,
+                        models.Document.observed == observed,
+                        models.Document.level != "explicit",
+                        models.Document.deleted_at.is_(None),
+                        models.Document.source_ids.has_any(array(frontier)),
+                    )
+                    .values(deleted_at=func.now())
+                    .returning(models.Document.id)
+                )
+                frontier = [row[0] for row in (await db.execute(derived_stmt)).all()]
+                all_removed.extend(frontier)
+
+            if all_removed:
+                removed_by_observed[observed] = all_removed
+
+        # The session left the scope, so its backfill status entry is moot;
+        # clearing it keeps a later re-add starting from a fresh "pending".
+        await crud.clear_scope_backfill_status(
+            db, workspace_name, scope_peer, session_name
+        )
+        await db.commit()
+    await crud.invalidate_scope_peer_cache(workspace_name, scope_peer)
+
+    # Delete the vectors eagerly so recall can't surface removed memory while
+    # waiting for the reconciler sweep (which remains the backstop on failure).
+    external_vector_store = get_external_vector_store()
+    if external_vector_store is not None:
+        for observed, removed_ids in removed_by_observed.items():
+            namespace = external_vector_store.get_vector_namespace(
+                "document", workspace_name, scope_peer, observed
+            )
+            try:
+                await external_vector_store.delete_many(namespace, removed_ids)
+            except Exception:
+                logger.exception(
+                    "Failed to delete removed vectors from %s; reconciler sweep will retry",
+                    namespace,
+                )
+
+    # Rebuild what remains: the card must be regenerated from remaining
+    # evidence only (rebuild=True drops the stale card from the prompt), and a
+    # manual omni dream rebuilds the higher-order structure.
+    from src.deriver.enqueue import enqueue_dream
+
+    for observed in sorted(removed_by_observed):
+        await enqueue_dream(
+            workspace_name,
+            observer=scope_peer,
+            observed=observed,
+            dream_type=DreamType.CARD_REFRESH,
+            rebuild=True,
+            trigger_reason="scope_removal",
+        )
+        await enqueue_dream(
+            workspace_name,
+            observer=scope_peer,
+            observed=observed,
+            dream_type=DreamType.OMNI,
+            trigger_reason="scope_removal",
+        )
+
+    logger.info(
+        "Scope removal reconciliation complete for %s/%s/%s: %d documents soft-deleted across %d collections",
+        workspace_name,
+        scope_peer,
+        session_name,
+        sum(len(ids) for ids in removed_by_observed.values()),
+        len(removed_by_observed),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Shared helpers
+# ---------------------------------------------------------------------------
+
+
+async def _write_backfill_status(
+    workspace_name: str,
+    scope_peer: str,
+    session_name: str,
+    *,
+    state: ScopeBackfillState,
+    docs_copied: int | None = None,
+) -> None:
+    """Best-effort status write in its own short-lived session."""
+    try:
+        async with tracked_db("scope_backfill.status") as db:
+            await crud.update_scope_backfill_status(
+                db,
+                workspace_name,
+                scope_peer,
+                session_name,
+                state=state,
+                docs_copied=docs_copied,
+            )
+            await db.commit()
+        await crud.invalidate_scope_peer_cache(workspace_name, scope_peer)
+    except Exception:
+        # Never mask the underlying failure (or fail a completed backfill)
+        # over a status bookkeeping write.
+        logger.exception(
+            "Failed to write scope backfill status %s for %s/%s/%s",
+            state,
+            workspace_name,
+            scope_peer,
+            session_name,
+        )

src/routers/scopes.py

@@ -8,9 +8,10 @@ that keeps the observer/observed mechanics hidden.
 All scopes routes require a workspace-level (or admin) key: scopes are an
 app-level admin surface, so peer- and session-scoped keys are rejected.
 
-Note: scope membership only affects messages ingested *after* the membership
-change. Conclusions already derived are neither backfilled on add nor
-reconciled on removal.
+Retroactive membership changes are handled asynchronously: adding a session
+that already has messages enqueues a backfill-by-copy job, removing a session
+enqueues a removal-reconciliation job. Track backfill progress via
+``GET /scopes/{scope_id}/status``.
 """
 
 import logging
@@ -109,8 +110,9 @@ async def add_sessions_to_scope(
     is already a member is a no-op. List the resulting membership with
     `POST /scopes/{scope_id}/sessions/list`.
 
-    Note: membership applies only to messages ingested after this call;
-    conclusions already derived are not backfilled.
+    Note: any added session that already has messages triggers an asynchronous
+    backfill-by-copy of its existing documents into the scope; track progress
+    via ``GET /scopes/{scope_id}/status``.
     """
     await crud.add_sessions_to_scope(
         db,
@@ -135,8 +137,10 @@ async def remove_session_from_scope(
     """
     Remove a Session from a Scope.
 
-    Note: conclusions already derived while the session was a member are left in
-    place.
+    Note: documents copied/derived while the session was a member are
+    reconciled asynchronously — the session's explicit copies are soft-deleted
+    from the scope, dependent derived documents follow (fail-closed), and the
+    scope's card is rebuilt from the remaining evidence.
     """
     await crud.remove_session_from_scope(
         db,
@@ -171,3 +175,24 @@ async def get_scope_sessions(
             workspace_name=workspace_id, scope_name=scope_id, reverse=reverse
         ),
     )
+
+
+@router.get(
+    "/{scope_id}/status",
+    response_model=schemas.ScopeStatus,
+    dependencies=[Depends(require_auth(workspace_name="workspace_id"))],
+)
+async def get_scope_status(
+    workspace_id: str = Path(...),
+    scope_id: str = Path(...),
+    db: AsyncSession = read_db,
+):
+    """
+    Get the backfill/reconciliation job status for a Scope.
+
+    Returns a per-session map of the backfill job state (pending / completed /
+    failed) with the number of documents copied once complete. Empty when no
+    backfill has ever been enqueued for the scope.
+    """
+    backfill_status = await crud.get_scope_backfill_status(db, workspace_id, scope_id)
+    return schemas.ScopeStatus(backfill_status=backfill_status)

src/schemas/__init__.py

@@ -39,6 +39,7 @@ from src.schemas.api import (
     Scope,
     ScopeCreate,
     ScopeSessionsAdd,
+    ScopeStatus,
     Session,
     SessionBase,
     SessionContext,
@@ -140,6 +141,7 @@ __all__ = [
     "Scope",
     "ScopeCreate",
     "ScopeSessionsAdd",
+    "ScopeStatus",
     "Session",
     "SessionBase",
     "SessionContext",

src/schemas/api.py

@@ -428,9 +428,9 @@ class SessionCreate(SessionBase):
         max_length=100,
         description=(
             "Optional list of (unprefixed) scope names to add this session to. "
-            "Each scope is created if it does not exist yet. Membership applies "
-            "only to messages ingested after the session is added to the scope; "
-            "conclusions already derived are not backfilled."
+            "Each scope is created if it does not exist yet. If the session "
+            "already has messages, its existing documents are backfilled into "
+            "the scope asynchronously."
         ),
     )
 
@@ -591,6 +591,18 @@ class ScopeSessionsAdd(BaseModel):
     )
 
 
+class ScopeStatus(BaseModel):
+    """Per-session backfill/reconciliation job status for a scope.
+
+    ``backfill_status`` maps each session that has had a backfill enqueued to
+    its current job state: ``{state, updated_at[, docs_copied]}`` where
+    ``state`` is ``pending``/``completed``/``failed`` and ``docs_copied`` is
+    present once a backfill completes.
+    """
+
+    backfill_status: dict[str, dict[str, Any]] = Field(default_factory=dict)
+
+
 # ---------------------------------------------------------------------------
 # Conclusion schemas
 # ---------------------------------------------------------------------------

src/utils/queue_payload.py

@@ -73,6 +73,26 @@ class DreamPayload(BasePayload):
     rebuild: bool = False
 
 
+class ScopeBackfillPayload(BasePayload):
+    """Payload for scope backfill tasks (session added to a scope, DEV-1999).
+
+    workspace_name lives in the QueueItem's dedicated column, mirroring the
+    other task payloads.
+    """
+
+    task_type: Literal["scope_backfill"] = "scope_backfill"
+    scope_peer: str
+    session_name: str
+
+
+class ScopeRemovalPayload(BasePayload):
+    """Payload for scope removal reconciliation tasks (session removed from a scope)."""
+
+    task_type: Literal["scope_removal"] = "scope_removal"
+    scope_peer: str
+    session_name: str
+
+
 class DeletionPayload(BasePayload):
     """Payload for deletion tasks."""
 
@@ -124,6 +144,22 @@ def create_dream_payload(
     ).model_dump(mode="json", exclude_none=True)
 
 
+def create_scope_task_payload(
+    task_type: Literal["scope_backfill", "scope_removal"],
+    *,
+    scope_peer: str,
+    session_name: str,
+) -> dict[str, Any]:
+    """Create a scope backfill / removal payload."""
+    if task_type == "scope_backfill":
+        return ScopeBackfillPayload(
+            scope_peer=scope_peer, session_name=session_name
+        ).model_dump(mode="json", exclude_none=True)
+    return ScopeRemovalPayload(
+        scope_peer=scope_peer, session_name=session_name
+    ).model_dump(mode="json", exclude_none=True)
+
+
 def create_deletion_payload(
     deletion_type: Literal["session", "observation", "workspace"],
     resource_id: str,

src/utils/types.py

@@ -251,7 +251,14 @@ class GetOrCreateResult(Generic[T]):
 
 
 TaskType = Literal[
-    "webhook", "summary", "representation", "dream", "deletion", "reconciler"
+    "webhook",
+    "summary",
+    "representation",
+    "dream",
+    "deletion",
+    "reconciler",
+    "scope_backfill",
+    "scope_removal",
 ]
 VectorSyncState = Literal["synced", "pending", "failed"]
 DocumentLevel = Literal["explicit", "deductive", "inductive", "contradiction"]

src/utils/work_unit.py

@@ -74,6 +74,15 @@ def construct_work_unit_key(
             raise ValueError("reconciler_type is required for reconciler tasks")
         return f"reconciler:{reconciler_type}"
 
+    if task_type in ("scope_backfill", "scope_removal"):
+        scope_peer = payload.get("scope_peer")
+        session_name = payload.get("session_name")
+        if not scope_peer or not session_name:
+            raise ValueError(
+                f"scope_peer and session_name are required for {task_type} tasks"
+            )
+        return f"{task_type}:{workspace_name}:{scope_peer}:{session_name}"
+
     raise ValueError(f"Invalid task type: {task_type}")
 
 
@@ -183,4 +192,19 @@ def parse_work_unit_key(work_unit_key: str) -> ParsedWorkUnit:
             observed=None,
         )
 
+    if task_type in ("scope_backfill", "scope_removal"):
+        # {task_type}:{workspace}:{scope_peer}:{session}
+        if len(parts) != 4:
+            raise ValueError(
+                f"Invalid work_unit_key format for task_type {task_type}: {work_unit_key}"
+            )
+        return ParsedWorkUnit(
+            task_type=task_type,
+            workspace_name=parts[1],
+            session_name=parts[3],
+            # The scope peer is the observer of every collection the task touches.
+            observer=parts[2],
+            observed=None,
+        )
+
     raise ValueError(f"Invalid task type in work_unit_key: {task_type}")

tests/conftest.py

@@ -957,6 +957,7 @@ def mock_tracked_db(request: pytest.FixtureRequest):
         "src.dialectic.core.tracked_db",
         "src.dreamer.specialists.tracked_db",
         "src.dreamer.surprisal.tracked_db",
+        "src.deriver.scope_backfill.tracked_db",
     ]
     with ExitStack() as stack:
         for target in tracked_db_targets:

tests/deriver/test_scope_backfill.py

@@ -0,0 +1,958 @@
+"""Tests for scope backfill-by-copy and removal reconciliation (DEV-1999).
+
+A scope is an observer peer (``scope.<name>``). Adding a session with
+pre-existing messages to a scope enqueues a ``scope_backfill`` task; the
+handler (``src.deriver.scope_backfill``) copies each observed peer's
+explicit-level documents from their global ``(P, P)`` collection into the
+scope's ``(scope_peer, P)`` collection, stamping ``copied_from`` for
+idempotency, then enqueues a manual omni dream. Removal enqueues
+``scope_removal``, which soft-deletes the copies (cascading to dependent
+derived documents) and enqueues a card_refresh (rebuild) + omni dream.
+
+These tests exercise the handlers directly (``process_scope_backfill`` /
+``process_scope_removal``) against real Collection/Document/Peer rows,
+mirroring the fixture style in tests/crud/test_document.py and
+tests/dreamer/test_card_refresh.py: rows are created directly via
+``db_session`` (never through the cache-backed ``crud.get_or_create_collection``,
+which the ``mock_crud_collection_operations`` autouse fixture stubs out to an
+unpersisted object for every other test). Fixture data must be *committed*
+(not merely flushed) because the handlers run their DB work through
+``tracked_db``, which in tests opens a separate session bound to the same
+engine (see ``mock_tracked_db_context`` in conftest.py) — a different
+connection that cannot see another session's uncommitted writes.
+"""
+
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+from nanoid import generate as generate_nanoid
+from sqlalchemy import func, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src import crud, models
+from src.deriver.scope_backfill import (
+    COPIED_FROM_KEY,
+    process_scope_backfill,
+    process_scope_removal,
+)
+from src.schemas import DreamType
+from src.utils.queue_payload import ScopeBackfillPayload, ScopeRemovalPayload
+from src.utils.scopes import is_scope_peer, scope_peer_name
+
+_EMBEDDING_DIM = 1536
+
+
+def _embedding(seed: float = 0.5) -> list[float]:
+    return [seed] * _EMBEDDING_DIM
+
+
+async def _create_peer(db_session: AsyncSession, workspace_name: str) -> models.Peer:
+    peer = models.Peer(name=str(generate_nanoid()), workspace_name=workspace_name)
+    db_session.add(peer)
+    await db_session.commit()
+    return peer
+
+
+async def _create_scope_peer(
+    db_session: AsyncSession, workspace_name: str, scope_name: str
+) -> models.Peer:
+    peer = models.Peer(
+        name=scope_peer_name(scope_name),
+        workspace_name=workspace_name,
+        # The authoritative kind flag lives in internal_metadata, which is not
+        # user-writable; `configuration` carries only the observe_me knob.
+        internal_metadata={"kind": "scope"},
+        configuration={"observe_me": False},
+    )
+    db_session.add(peer)
+    await db_session.commit()
+    return peer
+
+
+async def _create_session(
+    db_session: AsyncSession, workspace_name: str
+) -> models.Session:
+    session = models.Session(name=str(generate_nanoid()), workspace_name=workspace_name)
+    db_session.add(session)
+    await db_session.commit()
+    return session
+
+
+async def _join_scope(
+    db_session: AsyncSession, workspace_name: str, session_name: str, scope_peer: str
+) -> None:
+    """Record the scope peer's membership, as the scopes routes do.
+
+    The backfill handler refuses to copy into a scope the session has left, so
+    tests driving the handler directly must stand the membership row up.
+    """
+    db_session.add(
+        models.SessionPeer(
+            workspace_name=workspace_name,
+            session_name=session_name,
+            peer_name=scope_peer,
+        )
+    )
+    await db_session.commit()
+
+
+async def _create_collection(
+    db_session: AsyncSession, workspace_name: str, observer: str, observed: str
+) -> models.Collection:
+    collection = models.Collection(
+        workspace_name=workspace_name, observer=observer, observed=observed
+    )
+    db_session.add(collection)
+    await db_session.commit()
+    return collection
+
+
+async def _create_document(
+    db_session: AsyncSession,
+    workspace_name: str,
+    *,
+    observer: str,
+    observed: str,
+    session_name: str | None,
+    content: str = "some observation",
+    level: str = "explicit",
+    embedding: list[float] | None = None,
+    internal_metadata: dict[str, Any] | None = None,
+    source_ids: list[str] | None = None,
+) -> models.Document:
+    doc = models.Document(
+        workspace_name=workspace_name,
+        observer=observer,
+        observed=observed,
+        content=content,
+        level=level,
+        session_name=session_name,
+        embedding=embedding if embedding is not None else _embedding(),
+        internal_metadata=internal_metadata or {},
+        source_ids=source_ids,
+    )
+    db_session.add(doc)
+    await db_session.commit()
+    return doc
+
+
+async def _get_docs(
+    db_session: AsyncSession,
+    workspace_name: str,
+    *,
+    observer: str,
+    observed: str | None = None,
+    include_deleted: bool = True,
+) -> list[models.Document]:
+    stmt = select(models.Document).where(
+        models.Document.workspace_name == workspace_name,
+        models.Document.observer == observer,
+    )
+    if observed is not None:
+        stmt = stmt.where(models.Document.observed == observed)
+    if not include_deleted:
+        stmt = stmt.where(models.Document.deleted_at.is_(None))
+    result = await db_session.execute(stmt)
+    return list(result.scalars().all())
+
+
+async def _dream_items(
+    db_session: AsyncSession, workspace_name: str
+) -> list[models.QueueItem]:
+    result = await db_session.execute(
+        select(models.QueueItem).where(
+            models.QueueItem.workspace_name == workspace_name,
+            models.QueueItem.task_type == "dream",
+        )
+    )
+    return list(result.scalars().all())
+
+
+# ---------------------------------------------------------------------------
+# 1. Backfill copies exactly the target session's explicit docs
+# ---------------------------------------------------------------------------
+
+
+async def test_backfill_copies_only_target_session_explicit_docs(
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+):
+    test_workspace, sender = sample_data
+    workspace_name = test_workspace.name
+    scope_name = str(generate_nanoid())
+    scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
+
+    target_session = await _create_session(db_session, workspace_name)
+    other_session = await _create_session(db_session, workspace_name)
+    await _join_scope(db_session, workspace_name, target_session.name, scope_peer.name)
+
+    await _create_collection(
+        db_session, workspace_name, observer=sender.name, observed=sender.name
+    )
+    # Destination collection: crud.get_or_create_collection is stubbed to an
+    # unpersisted object by the autouse mock_crud_collection_operations
+    # fixture, so the scope's own collection must already exist for the
+    # copied Document rows' FK to resolve.
+    await _create_collection(
+        db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+    )
+
+    # In-scope: the target session's explicit doc.
+    target_doc = await _create_document(
+        db_session,
+        workspace_name,
+        observer=sender.name,
+        observed=sender.name,
+        session_name=target_session.name,
+        content="target session explicit fact",
+        embedding=_embedding(0.7),
+    )
+    # Out-of-scope: another session's explicit doc.
+    await _create_document(
+        db_session,
+        workspace_name,
+        observer=sender.name,
+        observed=sender.name,
+        session_name=other_session.name,
+        content="other session explicit fact",
+    )
+    # Out-of-scope: a derived (non-explicit) doc for the target session.
+    await _create_document(
+        db_session,
+        workspace_name,
+        observer=sender.name,
+        observed=sender.name,
+        session_name=target_session.name,
+        content="deductive fact",
+        level="deductive",
+    )
+
+    await process_scope_backfill(
+        ScopeBackfillPayload(
+            scope_peer=scope_peer.name, session_name=target_session.name
+        ),
+        workspace_name,
+    )
+
+    copies = await _get_docs(
+        db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+    )
+    assert len(copies) == 1
+    copy = copies[0]
+    assert copy.content == "target session explicit fact"
+    assert copy.level == "explicit"
+    assert copy.session_name == target_session.name
+    assert copy.internal_metadata[COPIED_FROM_KEY] == target_doc.id
+    assert list(copy.embedding) == pytest.approx(  # pyright: ignore[reportUnknownMemberType]
+        _embedding(0.7)
+    )
+    assert copy.deleted_at is None
+
+
+# ---------------------------------------------------------------------------
+# 2. Idempotency
+# ---------------------------------------------------------------------------
+
+
+async def test_backfill_processed_twice_is_idempotent(
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+):
+    test_workspace, sender = sample_data
+    workspace_name = test_workspace.name
+    scope_name = str(generate_nanoid())
+    scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
+    session = await _create_session(db_session, workspace_name)
+    await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
+    await _create_collection(
+        db_session, workspace_name, observer=sender.name, observed=sender.name
+    )
+    await _create_document(
+        db_session,
+        workspace_name,
+        observer=sender.name,
+        observed=sender.name,
+        session_name=session.name,
+    )
+
+    await _create_collection(
+        db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+    )
+
+    payload = ScopeBackfillPayload(
+        scope_peer=scope_peer.name, session_name=session.name
+    )
+    await process_scope_backfill(payload, workspace_name)
+    await process_scope_backfill(payload, workspace_name)
+
+    copies = await _get_docs(
+        db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+    )
+    assert len(copies) == 1
+
+
+async def test_add_remove_readd_converges_on_one_live_copy(
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+):
+    test_workspace, sender = sample_data
+    workspace_name = test_workspace.name
+    scope_name = str(generate_nanoid())
+    scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
+    session = await _create_session(db_session, workspace_name)
+    await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
+    await _create_collection(
+        db_session, workspace_name, observer=sender.name, observed=sender.name
+    )
+    await _create_document(
+        db_session,
+        workspace_name,
+        observer=sender.name,
+        observed=sender.name,
+        session_name=session.name,
+    )
+
+    await _create_collection(
+        db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+    )
+
+    backfill_payload = ScopeBackfillPayload(
+        scope_peer=scope_peer.name, session_name=session.name
+    )
+    removal_payload = ScopeRemovalPayload(
+        scope_peer=scope_peer.name, session_name=session.name
+    )
+
+    # add
+    await process_scope_backfill(backfill_payload, workspace_name)
+    # remove
+    await process_scope_removal(removal_payload, workspace_name)
+    live = await _get_docs(
+        db_session,
+        workspace_name,
+        observer=scope_peer.name,
+        observed=sender.name,
+        include_deleted=False,
+    )
+    assert live == []
+    # re-add
+    await process_scope_backfill(backfill_payload, workspace_name)
+
+    all_copies = await _get_docs(
+        db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+    )
+    live_copies = [d for d in all_copies if d.deleted_at is None]
+    assert len(all_copies) == 1  # restored, not duplicated
+    assert len(live_copies) == 1
+
+
+async def test_backfill_skips_a_session_that_left_the_scope(
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+):
+    """A removal that lands first must not be undone by a queued backfill.
+
+    scope_backfill and scope_removal carry different work-unit keys, so nothing
+    orders them: add-then-remove can leave a backfill queued after removal has
+    already swept the scope.
+    """
+    test_workspace, sender = sample_data
+    workspace_name = test_workspace.name
+    scope_name = str(generate_nanoid())
+    scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
+    session = await _create_session(db_session, workspace_name)
+    await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
+    await _create_collection(
+        db_session, workspace_name, observer=sender.name, observed=sender.name
+    )
+    await _create_collection(
+        db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+    )
+    await _create_document(
+        db_session,
+        workspace_name,
+        observer=sender.name,
+        observed=sender.name,
+        session_name=session.name,
+    )
+
+    # The session leaves the scope before the queued backfill is drained.
+    await db_session.execute(
+        update(models.SessionPeer)
+        .where(
+            models.SessionPeer.workspace_name == workspace_name,
+            models.SessionPeer.session_name == session.name,
+            models.SessionPeer.peer_name == scope_peer.name,
+        )
+        .values(left_at=func.now())
+    )
+    await db_session.commit()
+
+    await process_scope_backfill(
+        ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name),
+        workspace_name,
+    )
+
+    assert (
+        await _get_docs(
+            db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+        )
+        == []
+    )
+    # No status entry either: removal cleared it, and a skipped backfill must
+    # not resurrect the session in the scope's status map.
+    # Names held as plain strings: expire_all() below would make reading them
+    # off the ORM instances trigger a lazy reload mid-assertion.
+    scope_peer_name_str, session_name = scope_peer.name, session.name
+    db_session.expire_all()
+    peer = await db_session.scalar(
+        select(models.Peer)
+        .where(models.Peer.workspace_name == workspace_name)
+        .where(models.Peer.name == scope_peer_name_str)
+    )
+    assert peer is not None
+    assert session_name not in peer.internal_metadata.get("backfill_status", {})
+
+
+# ---------------------------------------------------------------------------
+# 3. Multi-peer session
+# ---------------------------------------------------------------------------
+
+
+async def test_backfill_multi_peer_session_copies_into_right_collections(
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+):
+    test_workspace, peer_a = sample_data
+    workspace_name = test_workspace.name
+    peer_b = await _create_peer(db_session, workspace_name)
+    scope_name = str(generate_nanoid())
+    scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
+    session = await _create_session(db_session, workspace_name)
+    await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
+
+    for peer in (peer_a, peer_b):
+        await _create_collection(
+            db_session, workspace_name, observer=peer.name, observed=peer.name
+        )
+        await _create_collection(
+            db_session, workspace_name, observer=scope_peer.name, observed=peer.name
+        )
+        await _create_document(
+            db_session,
+            workspace_name,
+            observer=peer.name,
+            observed=peer.name,
+            session_name=session.name,
+            content=f"fact about {peer.name}",
+        )
+
+    await process_scope_backfill(
+        ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name),
+        workspace_name,
+    )
+
+    copies_a = await _get_docs(
+        db_session, workspace_name, observer=scope_peer.name, observed=peer_a.name
+    )
+    copies_b = await _get_docs(
+        db_session, workspace_name, observer=scope_peer.name, observed=peer_b.name
+    )
+    assert len(copies_a) == 1
+    assert copies_a[0].content == f"fact about {peer_a.name}"
+    assert len(copies_b) == 1
+    assert copies_b[0].content == f"fact about {peer_b.name}"
+
+
+# ---------------------------------------------------------------------------
+# 4. Removal cascade
+# ---------------------------------------------------------------------------
+
+
+async def test_removal_cascades_to_dependent_derived_docs_only(
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+):
+    test_workspace, sender = sample_data
+    workspace_name = test_workspace.name
+    scope_name = str(generate_nanoid())
+    scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
+    session = await _create_session(db_session, workspace_name)
+    await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
+    await _create_collection(
+        db_session, workspace_name, observer=sender.name, observed=sender.name
+    )
+    await _create_collection(
+        db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+    )
+    await _create_document(
+        db_session,
+        workspace_name,
+        observer=sender.name,
+        observed=sender.name,
+        session_name=session.name,
+    )
+
+    await process_scope_backfill(
+        ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name),
+        workspace_name,
+    )
+    [copy] = await _get_docs(
+        db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+    )
+
+    # A derived doc resting on the copy's evidence -> must be cascaded.
+    dependent = await _create_document(
+        db_session,
+        workspace_name,
+        observer=scope_peer.name,
+        observed=sender.name,
+        session_name=None,
+        content="deduction resting on removed evidence",
+        level="deductive",
+        source_ids=[copy.id],
+    )
+    # An unrelated derived doc in the same collection -> must survive.
+    unrelated = await _create_document(
+        db_session,
+        workspace_name,
+        observer=scope_peer.name,
+        observed=sender.name,
+        session_name=None,
+        content="unrelated deduction",
+        level="deductive",
+        source_ids=["some-other-doc-id-not-removed"],
+    )
+
+    copy_id, dependent_id, unrelated_id = copy.id, dependent.id, unrelated.id
+
+    await process_scope_removal(
+        ScopeRemovalPayload(scope_peer=scope_peer.name, session_name=session.name),
+        workspace_name,
+    )
+
+    # process_scope_removal runs on a separate tracked_db session (a
+    # different connection). Query raw columns rather than full ORM entities
+    # so this session's identity map (holding the pre-removal `copy` /
+    # `dependent` / `unrelated` instances) can't hand back stale, expired
+    # attributes.
+    result = await db_session.execute(
+        select(models.Document.id, models.Document.deleted_at).where(
+            models.Document.workspace_name == workspace_name,
+            models.Document.observer == scope_peer.name,
+            models.Document.observed == sender.name,
+        )
+    )
+    deleted_at_by_id = {row[0]: row[1] for row in result.all()}
+    assert deleted_at_by_id[copy_id] is not None
+    assert deleted_at_by_id[dependent_id] is not None
+    assert deleted_at_by_id[unrelated_id] is None
+
+
+# ---------------------------------------------------------------------------
+# 5. Dream enqueues
+# ---------------------------------------------------------------------------
+
+
+async def test_backfill_enqueues_manual_omni_dream(
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+):
+    test_workspace, sender = sample_data
+    workspace_name = test_workspace.name
+    scope_name = str(generate_nanoid())
+    scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
+    session = await _create_session(db_session, workspace_name)
+    await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
+    await _create_collection(
+        db_session, workspace_name, observer=sender.name, observed=sender.name
+    )
+    await _create_collection(
+        db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+    )
+    await _create_document(
+        db_session,
+        workspace_name,
+        observer=sender.name,
+        observed=sender.name,
+        session_name=session.name,
+    )
+
+    await process_scope_backfill(
+        ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name),
+        workspace_name,
+    )
+
+    dreams = await _dream_items(db_session, workspace_name)
+    assert len(dreams) == 1
+    payload = dreams[0].payload
+    assert payload["dream_type"] == DreamType.OMNI.value
+    assert payload["observer"] == scope_peer.name
+    assert payload["observed"] == sender.name
+    assert payload["trigger_reason"] == "scope_backfill"
+    assert payload.get("rebuild", False) is False
+
+
+async def test_removal_enqueues_card_refresh_rebuild_and_omni_dream(
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+):
+    """Removal's own dream enqueues, isolated from backfill's.
+
+    The scope's copy is created directly (as if an earlier backfill already
+    ran and its dream was drained by the deriver) rather than by calling
+    process_scope_backfill first: enqueue_dream dedupes on work_unit_key, so
+    a still-pending omni dream from an immediately-preceding backfill would
+    silently swallow removal's own omni enqueue and make this test couple to
+    that unrelated dedup behavior instead of testing removal in isolation.
+    """
+    test_workspace, sender = sample_data
+    workspace_name = test_workspace.name
+    scope_name = str(generate_nanoid())
+    scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
+    session = await _create_session(db_session, workspace_name)
+    await _create_collection(
+        db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+    )
+    await _create_document(
+        db_session,
+        workspace_name,
+        observer=scope_peer.name,
+        observed=sender.name,
+        session_name=session.name,
+        internal_metadata={COPIED_FROM_KEY: "some-source-doc-id"},
+    )
+
+    await process_scope_removal(
+        ScopeRemovalPayload(scope_peer=scope_peer.name, session_name=session.name),
+        workspace_name,
+    )
+
+    dreams = await _dream_items(db_session, workspace_name)
+    removal_dreams = [
+        d for d in dreams if d.payload.get("trigger_reason") == "scope_removal"
+    ]
+    assert len(removal_dreams) == 2
+
+    by_type = {d.payload["dream_type"]: d.payload for d in removal_dreams}
+    assert DreamType.CARD_REFRESH.value in by_type
+    assert DreamType.OMNI.value in by_type
+    card_refresh_payload = by_type[DreamType.CARD_REFRESH.value]
+    assert card_refresh_payload["rebuild"] is True
+    assert card_refresh_payload["observer"] == scope_peer.name
+    assert card_refresh_payload["observed"] == sender.name
+
+
+# ---------------------------------------------------------------------------
+# 6. Status endpoint
+# ---------------------------------------------------------------------------
+
+
+async def test_status_reflects_pending_then_completed(
+    client: TestClient,
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+):
+    test_workspace, sender = sample_data
+    workspace_name = test_workspace.name
+    scope_name = str(generate_nanoid())
+
+    response = client.post(
+        f"/v3/workspaces/{workspace_name}/scopes", json={"id": scope_name}
+    )
+    assert response.status_code == 201
+    scope_peer_full_name = scope_peer_name(scope_name)
+
+    session_name = str(generate_nanoid())
+    response = client.post(
+        f"/v3/workspaces/{workspace_name}/sessions",
+        json={"id": session_name, "peers": {sender.name: {}}},
+    )
+    assert response.status_code == 201
+
+    message = models.Message(
+        workspace_name=workspace_name,
+        session_name=session_name,
+        peer_name=sender.name,
+        content="hello from before the scope existed",
+        public_id=generate_nanoid(),
+        seq_in_session=1,
+        token_count=5,
+    )
+    db_session.add(message)
+    await db_session.commit()
+
+    # The message's explicit document (normally produced by the deriver) —
+    # created directly since the deriver isn't run in this test.
+    await _create_collection(
+        db_session, workspace_name, observer=sender.name, observed=sender.name
+    )
+    await _create_document(
+        db_session,
+        workspace_name,
+        observer=sender.name,
+        observed=sender.name,
+        session_name=session_name,
+    )
+
+    response = client.post(
+        f"/v3/workspaces/{workspace_name}/scopes/{scope_name}/sessions",
+        json={"session_ids": [session_name]},
+    )
+    assert response.status_code == 204, response.text
+
+    # Destination collection: crud.get_or_create_collection is stubbed to an
+    # unpersisted object by the autouse fixture, so it must pre-exist.
+    await _create_collection(
+        db_session, workspace_name, observer=scope_peer_full_name, observed=sender.name
+    )
+
+    status_url = f"/v3/workspaces/{workspace_name}/scopes/{scope_name}/status"
+    response = client.get(status_url)
+    assert response.status_code == 200, response.text
+    backfill_status = response.json()["backfill_status"]
+    assert backfill_status[session_name]["state"] == "pending"
+
+    # Simulate the deriver picking up the enqueued task.
+    await process_scope_backfill(
+        ScopeBackfillPayload(
+            scope_peer=scope_peer_full_name, session_name=session_name
+        ),
+        workspace_name,
+    )
+
+    response = client.get(status_url)
+    assert response.status_code == 200, response.text
+    backfill_status = response.json()["backfill_status"]
+    assert backfill_status[session_name]["state"] == "completed"
+    assert backfill_status[session_name]["docs_copied"] == 1
+
+
+async def test_backfill_re_embeds_sources_with_null_embeddings(
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+):
+    """Source rows carry no embedding on external-store deployments.
+
+    Phase 2 re-embeds those (embedding API only) and pairs results back with
+    strict=True, so a mis-pairing would raise rather than silently mismatch.
+    """
+    test_workspace, sender = sample_data
+    workspace_name = test_workspace.name
+    scope_name = str(generate_nanoid())
+    scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
+    session = await _create_session(db_session, workspace_name)
+    await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
+    await _create_collection(
+        db_session, workspace_name, observer=sender.name, observed=sender.name
+    )
+    await _create_collection(
+        db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+    )
+    source = await _create_document(
+        db_session,
+        workspace_name,
+        observer=sender.name,
+        observed=sender.name,
+        session_name=session.name,
+        content="fact whose vector lives in the external store",
+    )
+    source.embedding = None
+    await db_session.commit()
+
+    await process_scope_backfill(
+        ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name),
+        workspace_name,
+    )
+
+    [copy] = await _get_docs(
+        db_session, workspace_name, observer=scope_peer.name, observed=sender.name
+    )
+    assert copy.embedding is not None
+    assert len(copy.embedding) == _EMBEDDING_DIM
+
+
+async def test_backfill_failure_records_failed_status(
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+    monkeypatch: pytest.MonkeyPatch,
+):
+    test_workspace, _ = sample_data
+    workspace_name = test_workspace.name
+    scope_name = str(generate_nanoid())
+    scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
+    session = await _create_session(db_session, workspace_name)
+
+    # Plain strings: the ORM instances are expired below (see the same guard in
+    # test_backfill_status_writes_preserve_the_scope_kind_flag).
+    scope_peer_name_str, session_name = scope_peer.name, session.name
+
+    async def boom(*_args: Any, **_kwargs: Any) -> None:
+        raise RuntimeError("copy phase blew up")
+
+    monkeypatch.setattr("src.deriver.scope_backfill._run_backfill", boom)
+
+    with pytest.raises(RuntimeError):
+        await process_scope_backfill(
+            ScopeBackfillPayload(
+                scope_peer=scope_peer_name_str, session_name=session_name
+            ),
+            workspace_name,
+        )
+
+    db_session.expire_all()
+    peer = await db_session.scalar(
+        select(models.Peer)
+        .where(models.Peer.workspace_name == workspace_name)
+        .where(models.Peer.name == scope_peer_name_str)
+    )
+    assert peer is not None
+    assert peer.internal_metadata["backfill_status"][session_name]["state"] == "failed"
+
+
+# ---------------------------------------------------------------------------
+# 7. Route wiring: add-sessions enqueues backfill only when messages exist
+# ---------------------------------------------------------------------------
+
+
+async def test_add_sessions_enqueues_backfill_only_when_session_has_messages(
+    client: TestClient,
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+):
+    test_workspace, sender = sample_data
+    workspace_name = test_workspace.name
+    scope_name = str(generate_nanoid())
+    assert (
+        client.post(
+            f"/v3/workspaces/{workspace_name}/scopes", json={"id": scope_name}
+        ).status_code
+        == 201
+    )
+
+    # Session with a pre-existing message.
+    session_with_messages = str(generate_nanoid())
+    assert (
+        client.post(
+            f"/v3/workspaces/{workspace_name}/sessions",
+            json={"id": session_with_messages, "peers": {sender.name: {}}},
+        ).status_code
+        == 201
+    )
+    message = models.Message(
+        workspace_name=workspace_name,
+        session_name=session_with_messages,
+        peer_name=sender.name,
+        content="already said something",
+        public_id=generate_nanoid(),
+        seq_in_session=1,
+        token_count=5,
+    )
+    db_session.add(message)
+    await db_session.commit()
+
+    # Empty session, no messages.
+    empty_session = str(generate_nanoid())
+    assert (
+        client.post(
+            f"/v3/workspaces/{workspace_name}/sessions",
+            json={"id": empty_session},
+        ).status_code
+        == 201
+    )
+
+    response = client.post(
+        f"/v3/workspaces/{workspace_name}/scopes/{scope_name}/sessions",
+        json={"session_ids": [session_with_messages, empty_session]},
+    )
+    assert response.status_code == 204, response.text
+
+    result = await db_session.execute(
+        select(models.QueueItem).where(
+            models.QueueItem.workspace_name == workspace_name,
+            models.QueueItem.task_type == "scope_backfill",
+        )
+    )
+    backfill_items = list(result.scalars().all())
+    assert len(backfill_items) == 1
+    assert backfill_items[0].payload["session_name"] == session_with_messages
+
+
+# ---------------------------------------------------------------------------
+# The scope `kind` flag and the backfill status map share the scope peer's
+# internal_metadata. Every write to that column must be a JSONB merge scoped to
+# the backfill key; a wholesale assignment would drop the flag and silently turn
+# the peer back into an ordinary one — invisible until some later read stopped
+# recognising it as a scope.
+# ---------------------------------------------------------------------------
+
+
+async def test_backfill_status_writes_preserve_the_scope_kind_flag(
+    db_session: AsyncSession,
+    sample_data: tuple[models.Workspace, models.Peer],
+):
+    """Status writes must not clobber the authoritative kind flag.
+
+    Both live in internal_metadata, so this pins the one property that makes
+    them able to coexist. Covers the whole lifecycle, because a wholesale write
+    could be introduced at any single step: pending, completed, then cleared.
+    """
+    test_workspace, _ = sample_data
+    workspace_name = test_workspace.name
+    scope_name = str(generate_nanoid())
+    session_name = str(generate_nanoid())
+    await _create_scope_peer(db_session, workspace_name, scope_name)
+    # Held as a plain string: the ORM instance is expired below on every check,
+    # so reading an attribute off it would trigger a reload mid-assertion.
+    backing_peer = scope_peer_name(scope_name)
+
+    async def assert_still_a_scope(stage: str) -> dict[str, Any]:
+        db_session.expire_all()
+        refreshed = await db_session.scalar(
+            select(models.Peer)
+            .where(models.Peer.workspace_name == workspace_name)
+            .where(models.Peer.name == backing_peer)
+        )
+        assert refreshed is not None
+        assert is_scope_peer(refreshed.name, refreshed.internal_metadata), (
+            f"the peer stopped being a scope after {stage}: "
+            f"internal_metadata={refreshed.internal_metadata!r}"
+        )
+        # And the facade still resolves it, which is what actually breaks:
+        # get_scope_or_raise 404s on a peer that has lost the flag.
+        resolved = await crud.get_scope_or_raise(db_session, workspace_name, scope_name)
+        assert resolved.name == backing_peer
+        return refreshed.internal_metadata
+
+    await assert_still_a_scope("creation")
+
+    await crud.update_scope_backfill_status(
+        db_session,
+        workspace_name,
+        backing_peer,
+        session_name,
+        state="pending",
+    )
+    await db_session.commit()
+    metadata = await assert_still_a_scope("a pending status write")
+    assert metadata["backfill_status"][session_name]["state"] == "pending"
+
+    await crud.update_scope_backfill_status(
+        db_session,
+        workspace_name,
+        backing_peer,
+        session_name,
+        state="completed",
+        docs_copied=3,
+    )
+    await db_session.commit()
+    metadata = await assert_still_a_scope("a completed status write")
+    assert metadata["backfill_status"][session_name]["docs_copied"] == 3
+
+    await crud.clear_scope_backfill_status(
+        db_session, workspace_name, backing_peer, session_name
+    )
+    await db_session.commit()
+    metadata = await assert_still_a_scope("clearing the status")
+    assert session_name not in metadata.get("backfill_status", {})