diff --git a/src/crud/__init__.py b/src/crud/__init__.py index 7ce4c249..38be14da 100644 --- a/src/crud/__init__.py +++ b/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", diff --git a/src/crud/scope.py b/src/crud/scope.py index a2ef1588..d5b92c31 100644 --- a/src/crud/scope.py +++ b/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: {: {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', '{}') || {: }} + + 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(, 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: + ``{: {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) diff --git a/src/crud/session.py b/src/crud/session.py index 1fc884ec..efa3f664 100644 --- a/src/crud/session.py +++ b/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) diff --git a/src/deriver/consumer.py b/src/deriver/consumer.py index 708d05f6..118135ca 100644 --- a/src/deriver/consumer.py +++ b/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}") diff --git a/src/deriver/enqueue.py b/src/deriver/enqueue.py index f62a3db3..2f70e787 100644 --- a/src/deriver/enqueue.py +++ b/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"], diff --git a/src/deriver/scope_backfill.py b/src/deriver/scope_backfill.py new file mode 100644 index 00000000..81017e28 --- /dev/null +++ b/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, + ) diff --git a/src/routers/scopes.py b/src/routers/scopes.py index 9d9087c0..de0d2865 100644 --- a/src/routers/scopes.py +++ b/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) diff --git a/src/schemas/__init__.py b/src/schemas/__init__.py index 32e144b3..b4211b99 100644 --- a/src/schemas/__init__.py +++ b/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", diff --git a/src/schemas/api.py b/src/schemas/api.py index 411f2d14..55a6121a 100644 --- a/src/schemas/api.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/src/utils/queue_payload.py b/src/utils/queue_payload.py index 5d616b1e..59815ca5 100644 --- a/src/utils/queue_payload.py +++ b/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, diff --git a/src/utils/types.py b/src/utils/types.py index 33a2993d..24a13ae6 100644 --- a/src/utils/types.py +++ b/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"] diff --git a/src/utils/work_unit.py b/src/utils/work_unit.py index 6e0e25d4..2c05d58e 100644 --- a/src/utils/work_unit.py +++ b/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}") diff --git a/tests/conftest.py b/tests/conftest.py index b57e3786..f99859b3 100644 --- a/tests/conftest.py +++ b/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: diff --git a/tests/deriver/test_scope_backfill.py b/tests/deriver/test_scope_backfill.py new file mode 100644 index 00000000..06464c6f --- /dev/null +++ b/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.``). 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", {})