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

src/crud/__init__.py

@@ -46,11 +46,15 @@
 )
 from .scope import (
     add_sessions_to_scope,
+    clear_scope_backfill_status,
     get_or_create_scopes,
     get_scope,
+    get_scope_backfill_status,
     get_scope_session_names,
     get_scopes,
+    invalidate_scope_peer_cache,
     remove_session_from_scope,
+    update_scope_backfill_status,
 )
 from .session import (
     SessionDeletionResult,
@@ -130,11 +134,15 @@
     "get_working_representation",
     # Scope
     "add_sessions_to_scope",
+    "clear_scope_backfill_status",
     "get_or_create_scopes",
     "get_scope",
+    "get_scope_backfill_status",
     "get_scope_session_names",
     "get_scopes",
+    "invalidate_scope_peer_cache",
     "remove_session_from_scope",
+    "update_scope_backfill_status",
     # Session
     "SessionDeletionResult",
     "get_sessions",

src/crud/document.py

@@ -441,6 +441,30 @@ def _normalize_content(content: str) -> str:
     return content.strip().lower()
 
 
+def _dedup_key(
+    content: str, level: str, session_name: str | None
+) -> tuple[str, str, str | None]:
+    """Build the exact-match dedup key for a document.
+
+    Dedup never crosses levels: a same-content document at a different level is
+    a different kind of record (an explicit fact is not interchangeable with a
+    deductive conclusion that happens to share its text).
+
+    For **explicit** documents dedup additionally never crosses sessions.
+    Explicit documents are session-pure records of what was derived from that
+    session's messages — the Scopes copy-by-session model depends on this — so
+    a repeat of the same fact in a different session must produce a new
+    document in that session rather than reinforce another session's row.
+    Derived levels (deductive/inductive/contradiction) are consolidations and
+    may still dedup across sessions.
+    """
+    return (
+        _normalize_content(content),
+        level,
+        session_name if level == "explicit" else None,
+    )
+
+
 async def create_documents(
     db: AsyncSession,
     documents: list[schemas.DocumentCreate],
@@ -477,9 +501,10 @@ async def create_documents(
     # exact-content dedup (independent of `deduplicate`): pre-fetch
     # existing live documents whose normalized content matches anything in this
     # batch, scoped to (workspace, observer, observed). The SQL normalization must
-    # mirror _normalize_content.
+    # mirror _normalize_content. Matching is further scoped per-document by
+    # level (always) and session (for explicit documents) via _dedup_key.
     batch_normalized: set[str] = {_normalize_content(d.content) for d in documents}
-    existing_by_normalized: dict[str, models.Document] = {}
+    existing_by_key: dict[tuple[str, str, str | None], models.Document] = {}
     if batch_normalized:
         # The `normalized_content_sql.in_(...)` filter below narrows to the
         # (workspace, observer, observed) partition via the single-column indexes,
@@ -507,29 +532,49 @@ async def create_documents(
             )
         )
         for existing_doc in existing_result.scalars():
-            # If multiple historical rows share normalized content, reinforcing
+            # If multiple historical rows share a dedup key, reinforcing
             # one is sufficient; keep the first.
-            existing_by_normalized.setdefault(
-                _normalize_content(existing_doc.content), existing_doc
+            existing_by_key.setdefault(
+                _dedup_key(
+                    existing_doc.content,
+                    existing_doc.level,
+                    existing_doc.session_name,
+                ),
+                existing_doc,
             )
 
-    # Tracks normalized content already accepted from this batch so exact
+    # Tracks dedup keys already accepted from this batch so exact
     # duplicates within a single inference call collapse to one document.
-    seen_in_batch: set[str] = set()
+    seen_in_batch: set[tuple[str, str, str | None]] = set()
 
     for doc in documents:
         try:
-            normalized_content = _normalize_content(doc.content)
+            # Session-purity invariant: an explicit document must always carry
+            # the session it was derived from. Refuse to write session-less
+            # explicit documents rather than silently minting global explicit
+            # memory (the Scopes copy-by-session model depends on explicit
+            # documents staying session-pure).
+            if doc.level == "explicit" and doc.session_name is None:
+                logger.error(
+                    "Refusing to create explicit document without session_name in %s/%s/%s (session-purity invariant): %r",
+                    workspace_name,
+                    observer,
+                    observed,
+                    doc.content[:80],
+                )
+                continue
+
+            dedup_key = _dedup_key(doc.content, doc.level, doc.session_name)
 
             # Exact-match dedup, always on:
             # 1) collapse exact duplicates within this batch (drop silently).
-            if normalized_content in seen_in_batch:
+            if dedup_key in seen_in_batch:
                 continue
-            seen_in_batch.add(normalized_content)
+            seen_in_batch.add(dedup_key)
 
             # 2) drop exact duplicates of an existing live document, recording
             #    the re-derivation as reinforcement on the existing row.
-            existing_match = existing_by_normalized.get(normalized_content)
+            existing_match = existing_by_key.get(dedup_key)
             if existing_match is not None:
                 # Reinforce the existing row. greatest(...) keeps the bump atomic
                 # server-side (concurrent workers can't lose an increment) while
@@ -1080,14 +1125,28 @@ async def is_rejected_duplicate(
     If the document is a duplicate AND the existing document is superior,
     increments the existing document's ``times_derived`` to record the
     reinforcement, then returns True.
+
+    Merges are scoped so they never cross document levels, and never cross
+    sessions for explicit-level documents (session-purity invariant: an
+    explicit document records what was derived from exactly one session, so
+    a near-duplicate from another session must not reinforce or replace it).
     """
+    filters: dict[str, Any] = {"level": doc.level}
+    if doc.level == "explicit":
+        if doc.session_name is None:
+            # create_documents refuses session-less explicit documents; if one
+            # reaches here anyway it has no valid merge partner.
+            return False
+        filters["session_name"] = doc.session_name
+
     # Step 1: Find potential duplicates using cosine similarity
     similar_docs = await query_documents(
         db=db,
         workspace_name=workspace_name,
         query=doc.content,
         observer=observer,
         observed=observed,
+        filters=filters,
         max_distance=0.05,
         top_k=1,
         embedding=doc.embedding,

src/crud/scope.py

@@ -5,16 +5,25 @@
 that observes its member sessions (``observe_others=true``) 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; backfill of pre-existing documents and reconciliation on removal land
-in a follow-up (DEV-1999).
+Messages ingested after a membership change flow to the scope via the normal
+deriver fan-out. Retroactive changes are handled by queue jobs (DEV-1999):
+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 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
@@ -31,6 +40,12 @@
 
 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"]
+
 # Peer-level configuration stamped on every scope peer at creation. `kind` is
 # the authoritative scope flag; `observe_me: false` ensures no representation
 # is ever formed *of* a scope peer.
@@ -235,8 +250,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. Membership only affects messages ingested
-    after this call (backfill is DEV-1999).
+    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
@@ -284,6 +300,23 @@ async def add_sessions_to_scope(
 
     await db.commit()
 
+    # Backfill (DEV-1999): 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,
+        )
+
     return await get_scope_session_names(db, workspace_name, scope_name)
 
 
@@ -297,8 +330,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``). Reconciliation of documents derived while the session was a
-    member lands in a follow-up (DEV-1999).
+    ``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 (DEV-1999).
 
     Args:
         db: Database session
@@ -310,6 +344,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(db, workspace_name, scope_name)
@@ -320,3 +356,124 @@ async def remove_session_from_scope(
         session_name=session_name,
         peer_names={scope_peer_name(scope_name)},
     )
+
+    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(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)