diff --git a/src/crud/__init__.py b/src/crud/__init__.py index 5cbdfca1c..761080cf7 100644 --- a/src/crud/__init__.py +++ b/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", diff --git a/src/crud/document.py b/src/crud/document.py index ed3813409..9045c4b83 100644 --- a/src/crud/document.py +++ b/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,7 +1125,20 @@ 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, @@ -1088,6 +1146,7 @@ async def is_rejected_duplicate( query=doc.content, observer=observer, observed=observed, + filters=filters, max_distance=0.05, top_k=1, embedding=doc.embedding, diff --git a/src/crud/scope.py b/src/crud/scope.py index ac6cf2832..3db4fadc3 100644 --- a/src/crud/scope.py +++ b/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: {: {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', '{}') || {: }} + + 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(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 3f6b9586e..3189bd808 100644 --- a/src/crud/session.py +++ b/src/crud/session.py @@ -7,7 +7,18 @@ from cashews import NOT_NONE from nanoid import generate as generate_nanoid -from sqlalchemy import Select, and_, case, cast, delete, func, insert, select, update +from sqlalchemy import ( + Select, + and_, + case, + cast, + delete, + exists, + func, + insert, + select, + update, +) from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.engine import CursorResult from sqlalchemy.exc import IntegrityError @@ -276,8 +287,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 (backfill is DEV-1999). + # 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( @@ -308,6 +319,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 6e35d6149..998fbba2a 100644 --- a/src/deriver/consumer.py +++ b/src/deriver/consumer.py @@ -8,6 +8,10 @@ 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 @@ 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 05883306b..0bbd9a750 100644 --- a/src/deriver/enqueue.py +++ b/src/deriver/enqueue.py @@ -16,6 +16,7 @@ create_deletion_payload, create_dream_payload, create_payload, + create_scope_task_payload, ) from src.utils.work_unit import construct_work_unit_key @@ -403,6 +404,7 @@ def create_dream_record( delay_reason: str | None = None, documents_since_last_dream_at_schedule: int | None = None, document_threshold: int | None = None, + rebuild: bool = False, ) -> dict[str, Any]: """ Create a queue record for a dream task. @@ -417,6 +419,7 @@ def create_dream_record( delay_reason: what governed when it fires documents_since_last_dream_at_schedule: count snapshot at schedule time document_threshold: DOCUMENT_THRESHOLD snapshot at schedule time + rebuild: card_refresh only — rebuild the card without the prior card Returns: Queue record dictionary with workspace_name and other fields @@ -430,6 +433,7 @@ def create_dream_record( delay_reason=delay_reason, documents_since_last_dream_at_schedule=documents_since_last_dream_at_schedule, document_threshold=document_threshold, + rebuild=rebuild, ) return { @@ -452,6 +456,7 @@ async def enqueue_dream( delay_reason: str | None = None, documents_since_last_dream_at_schedule: int | None = None, document_threshold: int | None = None, + rebuild: bool = False, ) -> None: """ Enqueue a dream task for immediate processing by the deriver. @@ -461,6 +466,8 @@ async def enqueue_dream( Deduplication: If a dream with the same work_unit_key is already in-progress (has an ActiveQueueSession) or pending in the queue, the enqueue is skipped. + The work unit key includes the dream type, so e.g. a card_refresh dream + never collides with a pending omni dream for the same collection. Args: workspace_name: Name of the workspace @@ -468,6 +475,7 @@ async def enqueue_dream( observed: Name of the observed peer dream_type: Type of dream to execute session_name: Name of the session to scope the dream to if specified + rebuild: card_refresh only — rebuild the card without the prior card """ async with tracked_db("dream_enqueue") as db_session: try: @@ -481,6 +489,7 @@ async def enqueue_dream( delay_reason=delay_reason, documents_since_last_dream_at_schedule=documents_since_last_dream_at_schedule, document_threshold=document_threshold, + rebuild=rebuild, ) work_unit_key = dream_record["work_unit_key"] @@ -545,6 +554,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 000000000..286d6f1ee --- /dev/null +++ b/src/deriver/scope_backfill.py @@ -0,0 +1,534 @@ +"""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.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: + docs_copied, touched_observed = await _run_backfill( + workspace_name, scope_peer, session_name + ) + except Exception: + await _write_backfill_status( + workspace_name, scope_peer, session_name, state="failed" + ) + raise + + # 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]]: + """Run the copy itself. Returns (docs copied or restored, touched observed peers).""" + # 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: + 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/dreamer/orchestrator.py b/src/dreamer/orchestrator.py index 0b529cc68..00f001f9e 100644 --- a/src/dreamer/orchestrator.py +++ b/src/dreamer/orchestrator.py @@ -26,7 +26,11 @@ from src import crud, models from src.config import settings from src.dependencies import tracked_db -from src.dreamer.specialists import SPECIALISTS, SpecialistResult +from src.dreamer.specialists import ( + SPECIALISTS, + CardRefreshSpecialist, + SpecialistResult, +) from src.dreamer.surprisal import SurprisalScore # type: ignore from src.exceptions import SurprisalError from src.schemas import DreamType @@ -307,6 +311,151 @@ async def run_dream( ) +async def run_card_refresh_dream( + workspace_name: str, + observer: str, + observed: str, + session_name: str | None = None, + *, + rebuild: bool = False, + dream_type: str | None = None, + trigger_reason: str | None = None, + delay_reason: str | None = None, +) -> DreamResult | None: + """ + Run a lightweight card-only refresh dream. + + Runs a single CardRefreshSpecialist restricted to peer-card tools + (get_recent_observations, search_memory, update_peer_card) with a low + tool-iteration cap. It never creates or deletes observations. + + Args: + workspace_name: Workspace identifier + observer: Observer peer name + observed: Observed peer name + session_name: Session identifier if specified + rebuild: When True the existing card is NOT injected into the prompt + and the specialist rebuilds it solely from observations present in + the collection (used after removals). + """ + if not settings.DREAM.ENABLED: + return None + + run_id = generate_nanoid() + task_name = f"dream_orchestrator_{run_id}" + start_time = time.perf_counter() + + logger.info( + f"[{run_id}] Starting card-refresh dream for {workspace_name}/{observer}/{observed} (rebuild={rebuild})" + ) + + # Short-lived DB session for config resolution + async with tracked_db("dream.config") as db: + if session_name is not None: + session = await crud.get_session( + db, workspace_name=workspace_name, session_name=session_name + ) + else: + session = None + + workspace = await crud.get_workspace(db, workspace_name=workspace_name) + configuration = get_configuration(None, session, workspace) + if not configuration.dream.enabled: + logger.info( + f"[{run_id}] Dreams disabled for {workspace_name}/{session_name}, skipping card refresh" + ) + return None + if not configuration.peer_card.create: + logger.info( + f"[{run_id}] Peer card creation disabled for {workspace_name}, skipping card refresh" + ) + return None + + specialist_success = False + specialist_result: SpecialistResult | None = None + duration_ms = 0.0 + try: + specialist = CardRefreshSpecialist(rebuild=rebuild) + try: + specialist_result = await specialist.run( + workspace_name=workspace_name, + observer=observer, + observed=observed, + session_name=session_name, + configuration=configuration, + parent_run_id=run_id, + ) + logger.info( + f"[{run_id}] Card refresh completed: {specialist_result.content[:200]}..." + ) + accumulate_metric( + task_name, "card_refresh_result", specialist_result.content, "blob" + ) + specialist_success = specialist_result.success + except Exception as e: + # Exception (not BaseException) — CancelledError must propagate so + # the worker can shut down; the finally still emits the run event. + logger.error( + f"[{run_id}] Card refresh specialist failed: {e}", exc_info=True + ) + accumulate_metric(task_name, "card_refresh_error", str(e), "blob") + + duration_ms = (time.perf_counter() - start_time) * 1000 + accumulate_metric(task_name, "total_duration", duration_ms, "ms") + logger.info(f"[{run_id}] Card-refresh dream completed in {duration_ms:.0f}ms") + log_performance_metrics("dream_orchestrator", run_id) + finally: + # Emit DreamRunEvent unconditionally so analytics see a parent for the + # specialist event, mirroring run_dream. Card refresh is a + # deduction-family run, so its outcome rides on deduction_success. + if duration_ms == 0.0: + duration_ms = (time.perf_counter() - start_time) * 1000 + try: + emit( + DreamRunEvent( + run_id=run_id, + workspace_name=workspace_name, + session_name=session_name, + observer=observer, + observed=observed, + specialists_run=["card_refresh"], + deduction_success=specialist_success, + induction_success=False, + surprisal_enabled=False, + surprisal_conclusion_count=0, + total_iterations=( + specialist_result.iterations if specialist_result else 0 + ), + total_input_tokens=( + specialist_result.input_tokens if specialist_result else 0 + ), + total_output_tokens=( + specialist_result.output_tokens if specialist_result else 0 + ), + total_duration_ms=duration_ms, + dream_type=dream_type, + enabled_types_count=len(settings.DREAM.ENABLED_TYPES), + trigger_reason=trigger_reason, + delay_reason=delay_reason, + ) + ) + except Exception: # pragma: no cover - telemetry must not raise + logger.debug("Failed to emit DreamRunEvent", exc_info=True) + + return DreamResult( + run_id=run_id, + specialists_run=["card_refresh"], + deduction_success=specialist_success, + induction_success=False, + surprisal_enabled=False, + surprisal_conclusion_count=0, + total_iterations=specialist_result.iterations if specialist_result else 0, + total_duration_ms=duration_ms, + input_tokens=specialist_result.input_tokens if specialist_result else 0, + output_tokens=specialist_result.output_tokens if specialist_result else 0, + ) + + def _create_queries_from_surprisal( high_surprisal_obs: list[SurprisalScore], ) -> list[str]: @@ -401,6 +550,28 @@ async def process_dream( update_data={"dream": dream_meta}, ) + case DreamType.CARD_REFRESH: + # Card-only refresh: never touches observations and never + # advances the omni dream guard pair (last_dream_at / + # last_dream_document_count) — a card refresh must not delay + # or satisfy consolidation scheduling. + result = await run_card_refresh_dream( + workspace_name=workspace_name, + observer=payload.observer, + observed=payload.observed, + session_name=payload.session_name, + rebuild=payload.rebuild, + dream_type=payload.dream_type.value, + trigger_reason=payload.trigger_reason, + delay_reason=payload.delay_reason, + ) + if result is not None: + logger.info( + f"Card-refresh dream completed: run_id={result.run_id}, " + + f"iterations={result.total_iterations}, " + + f"duration={result.total_duration_ms:.0f}ms" + ) + except Exception as e: logger.error( f"Error processing dream task {payload.dream_type} for {payload.observer}/{payload.observed}: {str(e)}", diff --git a/src/dreamer/specialists.py b/src/dreamer/specialists.py index b0d44ec8f..00eadb7e7 100644 --- a/src/dreamer/specialists.py +++ b/src/dreamer/specialists.py @@ -32,6 +32,7 @@ from src.telemetry.logging import accumulate_metric, log_performance_metrics from src.telemetry.prometheus.metrics import TokenTypes from src.utils.agent_tools import ( + CARD_REFRESH_SPECIALIST_TOOLS, DEDUCTION_SPECIALIST_TOOLS, INDUCTION_SPECIALIST_TOOLS, create_tool_executor, @@ -70,6 +71,64 @@ class SpecialistResult: # Tool names to exclude when peer card creation is disabled PEER_CARD_TOOL_NAMES = {"update_peer_card"} +# Shared PEER CARD system-prompt section (identity-store taxonomy + rules). +# Used verbatim by DeductionSpecialist and CardRefreshSpecialist. +PEER_CARD_SYSTEM_SECTION = """ + +## PEER CARD (REQUIRED) + +The peer card is the target observee's identity store: stable identity markers that distinguish this entity from others and persist across interactions. Behavior, tendencies, transient state, and episodic facts belong in observations, not on the peer card. + +A peer can be anything with identity that changes over time — a human, an agent, a codebase, a team, an organization. Do not assume the target observee is human. Do not require any field; empty is the correct output when evidence is absent. + +### Allowed entry kinds + +Each entry must start with one of these four prefixes (exact case, followed by a space): + +- `IDENTITY: ...` — canonical name, kind, aliases, IDs + - `IDENTITY: Name: Alice` + - `IDENTITY: Kind: Python monorepo` + - `IDENTITY: Version: 4.2` + - `IDENTITY: Aliases: alice@example.com` +- `ATTRIBUTE: ...` — stable durable property of the entity (including explicitly stated standing preferences) + - `ATTRIBUTE: Location: NYC` + - `ATTRIBUTE: Language: Python` + - `ATTRIBUTE: Prefers tea` + - `ATTRIBUTE: Charter: ship Honcho infrastructure` +- `RELATIONSHIP: ...` — durable link to another entity + - `RELATIONSHIP: Spouse: Bob` + - `RELATIONSHIP: Maintainer: vineeth` + - `RELATIONSHIP: Members: vineeth, rajat` +- `INSTRUCTION: ...` — standing rule of engagement that the target observee has explicitly stated (do/don't for the observer). Only when explicit; never inferred from behavior. + - `INSTRUCTION: Call me Vee` + - `INSTRUCTION: Never push to main without review` + +### Rules + +1. **Stable.** If the value plausibly changes within six months absent a deliberate announcement, it does not belong on the card. Prefer leaving the card empty over filling it with volatile content. +2. **Subject is the target observee.** Every entry must be a fact about the target observee, not about another participant in the session. Never write facts about co-occurring peers into the card, no matter how frequently they appear in the messages. +3. **Evidence-grounded.** Only write what the target observee has explicitly stated, or what another participant has explicitly stated about the target observee with the target observee's assent. No "general knowledge" inferences (`"co-founder"` does not imply an age; mentioning a colleague does not imply a family relationship). +4. **Type-agnostic.** The target observee may not be human. Do not require name/age/location/family/occupation fields. +5. **No behavioral content.** TRAITs, behavioral tendencies, patterns, and inferred preferences belong in observations, not on the peer card. Do not write `TRAIT:` entries or behavioral `PREFERENCE:` entries — they will be rejected. +6. **No evidence bundles.** Each entry is one concise fact. No `e.g.` clauses, no parenthetical example lists, no semicolon-separated value dumps. + +### Migrating an existing peer card + +The CURRENT PEER CARD shown in the user message may contain entries from an older format that do not start with an allowed prefix (e.g. `Name: Alice`, `Lives in NYC`, `TRAIT: Analytical`, `PREFERENCE: Detailed explanations`). When you call `update_peer_card`, you are responsible for re-emitting the entries you want to keep — entries you omit are dropped, and entries without an allowed prefix are silently rejected. + +For each legacy entry: + +- If it is still a valid identity marker, re-emit it under the correct prefix and keep the original content where reasonable. Examples: + - `Name: Alice` → `IDENTITY: Name: Alice` + - `Lives in NYC` → `ATTRIBUTE: Location: NYC` + - `Works at Google` → `ATTRIBUTE: Employer: Google` + - `INSTRUCTION: Call me Vee` → keep as is (already correctly prefixed) +- Drop entries that violate the rules above: behavioral `TRAIT:` lines, inferred behavioral `PREFERENCE:` lines, one-off events, transient state, evidence bundles. Do not re-prefix them — they are not identity markers. + +When in doubt about a specific legacy entry, prefer migrating it (so valid info isn't lost) over dropping it. Splitting one dense legacy entry into multiple correctly-prefixed entries is fine and encouraged (e.g. a semicolon-separated `Tech Stack:` dump can become several `ATTRIBUTE:` lines, one per durable tool/platform). + +Call `update_peer_card` with the complete deduplicated list when there is a durable identity update to record, or when the existing card needs migration. Entries that do not start with one of the four allowed prefixes will be rejected. Keep concise (max 40 entries).""" + class BaseSpecialist(ABC): """Base class for agentic specialists.""" @@ -78,6 +137,10 @@ class BaseSpecialist(ABC): # Whether this specialist is allowed to write to the peer card. Defaults to True; # specialists that should never touch the card (e.g., induction) override to False. can_update_peer_card: bool = True + # Whether the current peer card is fetched and injected into the user prompt. + # Card-refresh runs in rebuild mode set this to False so the card is + # reconstructed solely from observations present in the collection. + inject_peer_card: bool = True # Subclasses can override to customize the peer card update instruction peer_card_update_instruction: str = ( "Only update this with durable identity markers via `update_peer_card`." @@ -218,9 +281,11 @@ async def run( configuration is None or configuration.peer_card.create ) - # Fetch current peer card to inject into prompt (saves a tool call) + # Fetch current peer card to inject into prompt (saves a tool call). + # Skipped when inject_peer_card is False (card-refresh rebuild + # mode): the card must be reconstructed from observations only. current_peer_card: list[str] | None = None - if peer_card_enabled: + if peer_card_enabled and self.inject_peer_card: current_peer_card = await crud.get_peer_card( db, workspace_name=workspace_name, @@ -490,61 +555,7 @@ def build_system_prompt( _ = observed peer_card_section = "" if peer_card_enabled: - peer_card_section = """ - -## PEER CARD (REQUIRED) - -The peer card is the target observee's identity store: stable identity markers that distinguish this entity from others and persist across interactions. Behavior, tendencies, transient state, and episodic facts belong in observations, not on the peer card. - -A peer can be anything with identity that changes over time — a human, an agent, a codebase, a team, an organization. Do not assume the target observee is human. Do not require any field; empty is the correct output when evidence is absent. - -### Allowed entry kinds - -Each entry must start with one of these four prefixes (exact case, followed by a space): - -- `IDENTITY: ...` — canonical name, kind, aliases, IDs - - `IDENTITY: Name: Alice` - - `IDENTITY: Kind: Python monorepo` - - `IDENTITY: Version: 4.2` - - `IDENTITY: Aliases: alice@example.com` -- `ATTRIBUTE: ...` — stable durable property of the entity (including explicitly stated standing preferences) - - `ATTRIBUTE: Location: NYC` - - `ATTRIBUTE: Language: Python` - - `ATTRIBUTE: Prefers tea` - - `ATTRIBUTE: Charter: ship Honcho infrastructure` -- `RELATIONSHIP: ...` — durable link to another entity - - `RELATIONSHIP: Spouse: Bob` - - `RELATIONSHIP: Maintainer: vineeth` - - `RELATIONSHIP: Members: vineeth, rajat` -- `INSTRUCTION: ...` — standing rule of engagement that the target observee has explicitly stated (do/don't for the observer). Only when explicit; never inferred from behavior. - - `INSTRUCTION: Call me Vee` - - `INSTRUCTION: Never push to main without review` - -### Rules - -1. **Stable.** If the value plausibly changes within six months absent a deliberate announcement, it does not belong on the card. Prefer leaving the card empty over filling it with volatile content. -2. **Subject is the target observee.** Every entry must be a fact about the target observee, not about another participant in the session. Never write facts about co-occurring peers into the card, no matter how frequently they appear in the messages. -3. **Evidence-grounded.** Only write what the target observee has explicitly stated, or what another participant has explicitly stated about the target observee with the target observee's assent. No "general knowledge" inferences (`"co-founder"` does not imply an age; mentioning a colleague does not imply a family relationship). -4. **Type-agnostic.** The target observee may not be human. Do not require name/age/location/family/occupation fields. -5. **No behavioral content.** TRAITs, behavioral tendencies, patterns, and inferred preferences belong in observations, not on the peer card. Do not write `TRAIT:` entries or behavioral `PREFERENCE:` entries — they will be rejected. -6. **No evidence bundles.** Each entry is one concise fact. No `e.g.` clauses, no parenthetical example lists, no semicolon-separated value dumps. - -### Migrating an existing peer card - -The CURRENT PEER CARD shown in the user message may contain entries from an older format that do not start with an allowed prefix (e.g. `Name: Alice`, `Lives in NYC`, `TRAIT: Analytical`, `PREFERENCE: Detailed explanations`). When you call `update_peer_card`, you are responsible for re-emitting the entries you want to keep — entries you omit are dropped, and entries without an allowed prefix are silently rejected. - -For each legacy entry: - -- If it is still a valid identity marker, re-emit it under the correct prefix and keep the original content where reasonable. Examples: - - `Name: Alice` → `IDENTITY: Name: Alice` - - `Lives in NYC` → `ATTRIBUTE: Location: NYC` - - `Works at Google` → `ATTRIBUTE: Employer: Google` - - `INSTRUCTION: Call me Vee` → keep as is (already correctly prefixed) -- Drop entries that violate the rules above: behavioral `TRAIT:` lines, inferred behavioral `PREFERENCE:` lines, one-off events, transient state, evidence bundles. Do not re-prefix them — they are not identity markers. - -When in doubt about a specific legacy entry, prefer migrating it (so valid info isn't lost) over dropping it. Splitting one dense legacy entry into multiple correctly-prefixed entries is fine and encouraged (e.g. a semicolon-separated `Tech Stack:` dump can become several `ATTRIBUTE:` lines, one per durable tool/platform). - -Call `update_peer_card` with the complete deduplicated list when there is a durable identity update to record, or when the existing card needs migration. Entries that do not start with one of the four allowed prefixes will be rejected. Keep concise (max 40 entries).""" + peer_card_section = PEER_CARD_SYSTEM_SECTION return f"""You are a deductive reasoning agent analyzing observations about the target observee. @@ -763,6 +774,113 @@ def build_user_prompt( Go.""" +class CardRefreshSpecialist(BaseSpecialist): + """ + Card-only maintenance specialist for the ``card_refresh`` dream type. + + Restricted to peer-card work: it may discover observations + (get_recent_observations, search_memory) and rewrite the peer card + (update_peer_card). It has NO observation-mutating tools — a card refresh + must never create or delete observations. + + Two modes: + - refresh (default): the current card is injected into the prompt and the + specialist folds in new identity markers. + - rebuild: the current card is NOT injected; the specialist reconstructs + the card solely from observations present in the collection. Used after + removals, where the old card may contain facts whose support was deleted. + + Not a singleton — instantiated per run because ``rebuild`` is per-dream + state. + """ + + name: str = "card_refresh" + peer_card_update_instruction: str = "Update this with `update_peer_card`. See the PEER CARD section in the system prompt for the allowed entry kinds and rules." + + # Low iteration ceiling for this lightweight, single-purpose run. + MAX_ITERATIONS_CEILING: int = 6 + + def __init__(self, *, rebuild: bool = False) -> None: + self.rebuild: bool = rebuild + # In rebuild mode the existing card is withheld from the prompt. + self.inject_peer_card: bool = not rebuild + + def get_tools(self, *, peer_card_enabled: bool = True) -> list[dict[str, Any]]: + if peer_card_enabled: + return CARD_REFRESH_SPECIALIST_TOOLS + # Defensive: a card refresh without card write access is a no-op, and + # the orchestrator skips the run entirely when peer cards are disabled. + return [ + t + for t in CARD_REFRESH_SPECIALIST_TOOLS + if t["name"] not in PEER_CARD_TOOL_NAMES + ] + + def get_model_config(self) -> ConfiguredModelSettings: + # Card refresh is a deduction-family task; reuse its model config. + return _require_specialist_model_config( + settings.DREAM.DEDUCTION_MODEL_CONFIG, + specialist_name="DREAM CARD_REFRESH", + ) + + def get_max_tokens(self) -> int: + return 8192 + + def get_max_iterations(self) -> int: + return min(self.MAX_ITERATIONS_CEILING, settings.DREAM.MAX_TOOL_ITERATIONS) + + def build_system_prompt( + self, observed: str, *, peer_card_enabled: bool = True + ) -> str: + _ = observed + _ = peer_card_enabled + rebuild_section = "" + if self.rebuild: + rebuild_section = """ + +## REBUILD MODE + +The existing peer card is deliberately NOT shown to you: it may contain entries whose supporting observations have since been removed. Build the card solely from the observations you find in the collection right now. Do not carry over or guess at prior card content — if an identity marker is not supported by a current observation, it does not go on the card.""" + + return f"""You are a peer-card maintenance agent for the target observee. + +## YOUR JOB + +Refresh the peer card and nothing else. You cannot create or delete observations — you have no tools for that. Your only write operation is `update_peer_card`. + +## PROCESS + +1. Survey the observation space: start with `get_recent_observations`, then use `search_memory` for targeted follow-ups (names, roles, locations, standing instructions). +2. Extract stable identity markers supported by the observations you found. +3. Call `update_peer_card` once with the complete deduplicated list. + +Keep it short — a handful of tool calls at most.{rebuild_section} +{PEER_CARD_SYSTEM_SECTION}""" + + def build_user_prompt( + self, + observed: str, + hints: list[str] | None, + peer_card: list[str] | None = None, + ) -> str: + _ = hints + target_observee_context = self._build_target_observee_context(observed) + peer_card_context = self._build_peer_card_context(peer_card) + + if self.rebuild: + return f"""{target_observee_context}Rebuild the peer card from scratch. + +The previous card is not shown and must not be assumed: reconstruct the card solely from observations currently in the collection. Start with `get_recent_observations`, verify with `search_memory` where needed, then call `update_peer_card` with the complete list. + +Go.""" + + return f"""{target_observee_context}{peer_card_context}Refresh the peer card. + +Review recent observations with `get_recent_observations` (and `search_memory` for targeted checks), then call `update_peer_card` with the complete deduplicated list if there is anything to add, correct, or migrate. If the card is already accurate and complete, finish without updating it. + +Go.""" + + # Singleton instances SPECIALISTS: dict[str, BaseSpecialist] = { "deduction": DeductionSpecialist(), diff --git a/src/routers/scopes.py b/src/routers/scopes.py index bb8860ee8..fc57d15ab 100644 --- a/src/routers/scopes.py +++ b/src/routers/scopes.py @@ -8,9 +8,10 @@ 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: backfill of pre-existing documents and reconciliation on removal land -in a follow-up (DEV-1999) — for now, scope membership only affects messages -ingested *after* the membership change. +Retroactive membership changes are handled asynchronously (DEV-1999): 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 @@ -107,8 +108,9 @@ async def add_sessions_to_scope( All named sessions must already exist (404 otherwise). Returns the scope's full member session list after the addition. - Note: membership only affects messages ingested after this call — backfill - of pre-existing documents lands in a follow-up (DEV-1999). + 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``. """ session_ids = await crud.add_sessions_to_scope( db, @@ -134,8 +136,10 @@ async def remove_session_from_scope( """ Remove a Session from a Scope. - Note: documents already derived while the session was a member are left in - place — reconciliation on removal lands in a follow-up (DEV-1999). + 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, @@ -158,3 +162,24 @@ async def get_scope_sessions( """Get the IDs of the Sessions that are members of a Scope.""" session_ids = await crud.get_scope_session_names(db, workspace_id, scope_id) return schemas.ScopeSessions(session_ids=session_ids) + + +@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/routers/workspaces.py b/src/routers/workspaces.py index 8b19fdfde..b7e15a716 100644 --- a/src/routers/workspaces.py +++ b/src/routers/workspaces.py @@ -231,6 +231,7 @@ async def schedule_dream( observed=observed, dream_type=dream_type, session_name=request.session_id, + rebuild=request.rebuild, # Manual route — explicit sentinels for the DreamRunEvent # scheduling-context fields. Auto-schedule threads concrete # threshold/delay reasons (see src/dreamer/dream_scheduler.py); diff --git a/src/schemas/__init__.py b/src/schemas/__init__.py index aa5dc01cb..d41a6f71b 100644 --- a/src/schemas/__init__.py +++ b/src/schemas/__init__.py @@ -39,6 +39,7 @@ ScopeCreate, ScopeSessions, ScopeSessionsAdd, + ScopeStatus, Session, SessionBase, SessionContext, @@ -139,6 +140,7 @@ "ScopeCreate", "ScopeSessions", "ScopeSessionsAdd", + "ScopeStatus", "Session", "SessionBase", "SessionContext", diff --git a/src/schemas/api.py b/src/schemas/api.py index 3bfa65301..4f937a576 100644 --- a/src/schemas/api.py +++ b/src/schemas/api.py @@ -366,10 +366,9 @@ class SessionCreate(SessionBase): default=None, description=( "Optional list of (unprefixed) scope names to add this session to. " - "Each scope is created if it does not exist yet. Note: scope " - "membership only affects messages ingested after the session is " - "added to the scope; backfill of pre-existing documents lands in a " - "follow-up (DEV-1999)." + "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." ), ) @@ -536,6 +535,18 @@ class ScopeSessions(BaseModel): session_ids: list[str] +class ScopeStatus(BaseModel): + """Per-session backfill/reconciliation job status for a scope (DEV-1999). + + ``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 # --------------------------------------------------------------------------- @@ -774,6 +785,14 @@ class ScheduleDreamRequest(BaseModel): session_id: str | None = Field( None, description="Session ID to scope the dream to if specified" ) + rebuild: bool = Field( + False, + description=( + "card_refresh dreams only: rebuild the peer card solely from " + "observations currently in the collection, without injecting the " + "existing card (use after removals)" + ), + ) # --------------------------------------------------------------------------- diff --git a/src/schemas/configuration.py b/src/schemas/configuration.py index b8291caba..53a8c11ac 100644 --- a/src/schemas/configuration.py +++ b/src/schemas/configuration.py @@ -17,6 +17,10 @@ class DreamType(str, Enum): """Types of dreams that can be triggered.""" OMNI = "omni" + # Lightweight card-only refresh: runs a single specialist restricted to + # peer-card tools. Used for event-driven refreshes (scope membership + # changes, cold starts) — never creates or deletes observations. + CARD_REFRESH = "card_refresh" class ReasoningConfiguration(BaseModel): diff --git a/src/telemetry/events/llm.py b/src/telemetry/events/llm.py index 8d74d2f63..1761e3334 100644 --- a/src/telemetry/events/llm.py +++ b/src/telemetry/events/llm.py @@ -34,6 +34,7 @@ class CallPurpose(str, Enum): DIALECTIC_ANSWER = "dialectic.answer" DREAM_DEDUCTION = "dream.deduction" DREAM_INDUCTION = "dream.induction" + DREAM_CARD_REFRESH = "dream.card_refresh" SUMMARY_SHORT = "summary.short" SUMMARY_LONG = "summary.long" diff --git a/src/utils/agent_tools.py b/src/utils/agent_tools.py index d4df768b2..7bde1e340 100644 --- a/src/utils/agent_tools.py +++ b/src/utils/agent_tools.py @@ -845,6 +845,18 @@ def _extract_pattern_snippet( TOOLS["create_observations_inductive"], ] +# Tools for the card-refresh specialist (card_refresh dream type). +# Card-only maintenance: discovery plus update_peer_card. Deliberately +# excludes every observation-mutating tool (create_observations*, +# delete_observations) — a card refresh must never touch observations. +CARD_REFRESH_SPECIALIST_TOOLS: list[dict[str, Any]] = [ + # Discovery tools + TOOLS["get_recent_observations"], + TOOLS["search_memory"], + # Action tool + TOOLS["update_peer_card"], +] + async def create_observations( observations: list[schemas.ObservationInput], @@ -1357,6 +1369,21 @@ async def _handle_create_observations_impl( ) ) continue + # Session-purity invariant: explicit observations record what was + # directly derived from a session's messages. Agents that are not + # processing messages (dreamer specialists, dialectic) must not mint + # them — consolidation output belongs at a derived level. + if not ctx.current_messages and validated.level == "explicit": + validation_failures.append( + ObservationFailure( + content_preview=validated.content[:50], + error=( + "Only message ingestion can create 'explicit' observations; " + "use a derived level (deductive/inductive/contradiction)" + ), + ) + ) + continue observations.append(validated) if not observations: diff --git a/src/utils/queue_payload.py b/src/utils/queue_payload.py index 605cf6150..59815ca57 100644 --- a/src/utils/queue_payload.py +++ b/src/utils/queue_payload.py @@ -66,6 +66,31 @@ class DreamPayload(BasePayload): delay_reason: str | None = None documents_since_last_dream_at_schedule: int | None = None document_threshold: int | None = None + # card_refresh only: when True the existing peer card is NOT injected into + # the specialist prompt and the card is rebuilt solely from observations + # currently in the collection (used after removals, where the old card may + # contain facts whose support was deleted). + 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): @@ -103,6 +128,7 @@ def create_dream_payload( delay_reason: str | None = None, documents_since_last_dream_at_schedule: int | None = None, document_threshold: int | None = None, + rebuild: bool = False, ) -> dict[str, Any]: """Create a dream payload.""" return DreamPayload( @@ -114,6 +140,23 @@ def create_dream_payload( delay_reason=delay_reason, documents_since_last_dream_at_schedule=documents_since_last_dream_at_schedule, document_threshold=document_threshold, + rebuild=rebuild, + ).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) diff --git a/src/utils/types.py b/src/utils/types.py index 33a2993d6..24a13ae67 100644 --- a/src/utils/types.py +++ b/src/utils/types.py @@ -251,7 +251,14 @@ async def post_commit(self) -> None: 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 6e0e25d46..2c05d58ec 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 af35c9990..003902f17 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -849,6 +849,7 @@ async def mock_tracked_db_context(_: str | None = None, *, read_only: bool = Fal "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/crud/test_document.py b/tests/crud/test_document.py index 1258b5229..1492e070b 100644 --- a/tests/crud/test_document.py +++ b/tests/crud/test_document.py @@ -1,4 +1,5 @@ import datetime +from unittest.mock import AsyncMock, patch import pytest from nanoid import generate as generate_nanoid @@ -924,3 +925,320 @@ async def test_create_documents( assert len(documents) == 2 assert documents[0].content in ["Observation 1", "Observation 2"] assert documents[1].content in ["Observation 1", "Observation 2"] + + +class TestSessionPurityInvariant: + """Regression tests for the explicit-document session-purity invariant. + + Explicit documents are session-pure records of what was derived from one + session's messages (the Scopes copy-by-session model depends on this): + + - an explicit document must always carry a non-null session_name + - dedup/merge (exact and semantic) must never cross document levels + - dedup/merge must never cross sessions for explicit documents + """ + + async def _setup( + self, + db_session: AsyncSession, + test_workspace: models.Workspace, + test_peer: models.Peer, + ) -> tuple[models.Peer, models.Session, models.Session]: + """Create an observed peer, two sessions, and the collection.""" + test_peer2 = models.Peer( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add(test_peer2) + session_a = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + session_b = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add_all([session_a, session_b]) + await db_session.flush() + + collection = models.Collection( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + db_session.add(collection) + await db_session.flush() + return test_peer2, session_a, session_b + + def _doc( + self, + content: str, + *, + session_name: str | None, + level: str = "explicit", + message_id: int = 1, + ) -> schemas.DocumentCreate: + return schemas.DocumentCreate( + content=content, + embedding=[0.1] * 1536, + session_name=session_name, + level=level, # pyright: ignore[reportArgumentType] + metadata=schemas.DocumentMetadata( + message_ids=[message_id], + message_created_at="2026-01-01T00:00:00Z", + ), + ) + + async def _live_docs( + self, + db_session: AsyncSession, + workspace_name: str, + observer: str, + observed: str, + ) -> list[models.Document]: + return list( + ( + await db_session.execute( + select(models.Document).where( + models.Document.workspace_name == workspace_name, + models.Document.observer == observer, + models.Document.observed == observed, + models.Document.deleted_at.is_(None), + ) + ) + ) + .scalars() + .all() + ) + + @pytest.mark.asyncio + async def test_explicit_without_session_is_refused( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """An explicit document with session_name=None must not be written; + derived levels remain allowed without a session (dream output).""" + test_workspace, test_peer = sample_data + test_peer2, _, _ = await self._setup(db_session, test_workspace, test_peer) + + accepted = await crud.create_documents( + db_session, + [ + self._doc("Global explicit fact", session_name=None), + self._doc( + "Dream-derived conclusion", + session_name=None, + level="deductive", + message_id=2, + ), + ], + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + + assert [d.content for d in accepted] == ["Dream-derived conclusion"] + live = await self._live_docs( + db_session, test_workspace.name, test_peer.name, test_peer2.name + ) + assert len(live) == 1 + assert live[0].level == "deductive" + + @pytest.mark.asyncio + async def test_exact_dedup_never_merges_explicit_across_sessions( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """The same explicit fact stated in two sessions produces two + session-pure documents; the other session's row is not reinforced.""" + test_workspace, test_peer = sample_data + test_peer2, session_a, session_b = await self._setup( + db_session, test_workspace, test_peer + ) + + await crud.create_documents( + db_session, + [self._doc("User likes coffee", session_name=session_a.name)], + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + accepted = await crud.create_documents( + db_session, + [ + self._doc( + "user likes coffee ", session_name=session_b.name, message_id=2 + ) + ], + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + + assert len(accepted) == 1 + live = await self._live_docs( + db_session, test_workspace.name, test_peer.name, test_peer2.name + ) + assert len(live) == 2 + assert {doc.session_name for doc in live} == {session_a.name, session_b.name} + assert all(doc.times_derived == 1 for doc in live) + + @pytest.mark.asyncio + async def test_exact_dedup_never_merges_across_levels( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """An explicit fact must not be dropped/reinforced against a derived + document that happens to share its content.""" + test_workspace, test_peer = sample_data + test_peer2, session_a, _ = await self._setup( + db_session, test_workspace, test_peer + ) + + await crud.create_documents( + db_session, + [ + self._doc( + "User likes coffee", session_name=session_a.name, level="deductive" + ) + ], + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + accepted = await crud.create_documents( + db_session, + [self._doc("User likes coffee", session_name=session_a.name, message_id=2)], + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + + assert len(accepted) == 1 + live = await self._live_docs( + db_session, test_workspace.name, test_peer.name, test_peer2.name + ) + assert len(live) == 2 + assert {doc.level for doc in live} == {"explicit", "deductive"} + assert all(doc.times_derived == 1 for doc in live) + + @pytest.mark.asyncio + async def test_exact_dedup_still_merges_derived_levels_across_sessions( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Derived levels are consolidations, not session-pure records: + cross-session exact dedup still reinforces the existing row.""" + test_workspace, test_peer = sample_data + test_peer2, session_a, session_b = await self._setup( + db_session, test_workspace, test_peer + ) + + await crud.create_documents( + db_session, + [ + self._doc( + "Probably a morning person", + session_name=session_a.name, + level="deductive", + ) + ], + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + accepted = await crud.create_documents( + db_session, + [ + self._doc( + "probably a morning person", + session_name=session_b.name, + level="deductive", + message_id=2, + ) + ], + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + + assert len(accepted) == 0 + live = await self._live_docs( + db_session, test_workspace.name, test_peer.name, test_peer2.name + ) + assert len(live) == 1 + assert live[0].times_derived == 2 + + @pytest.mark.asyncio + async def test_semantic_dedup_scoped_to_level_and_session( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """is_rejected_duplicate must constrain candidate search to the same + level, and to the same session for explicit documents.""" + test_workspace, test_peer = sample_data + test_peer2, session_a, _ = await self._setup( + db_session, test_workspace, test_peer + ) + + explicit_doc = self._doc("User likes coffee", session_name=session_a.name) + with patch( + "src.crud.document.query_documents", new=AsyncMock(return_value=[]) + ) as mock_query: + rejected = await is_rejected_duplicate( + db_session, + explicit_doc, + test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + assert rejected is False + assert mock_query.await_args is not None + assert mock_query.await_args.kwargs["filters"] == { + "level": "explicit", + "session_name": session_a.name, + } + + deductive_doc = self._doc( + "User likes coffee", session_name=None, level="deductive" + ) + with patch( + "src.crud.document.query_documents", new=AsyncMock(return_value=[]) + ) as mock_query: + rejected = await is_rejected_duplicate( + db_session, + deductive_doc, + test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + assert rejected is False + assert mock_query.await_args is not None + assert mock_query.await_args.kwargs["filters"] == {"level": "deductive"} + + @pytest.mark.asyncio + async def test_semantic_dedup_refuses_sessionless_explicit( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """A session-less explicit document has no valid merge partner: it is + never treated as a duplicate and no candidate search runs.""" + test_workspace, test_peer = sample_data + test_peer2, _, _ = await self._setup(db_session, test_workspace, test_peer) + + doc = self._doc("User likes coffee", session_name=None) + with patch( + "src.crud.document.query_documents", new=AsyncMock(return_value=[]) + ) as mock_query: + rejected = await is_rejected_duplicate( + db_session, + doc, + test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + assert rejected is False + mock_query.assert_not_awaited() diff --git a/tests/deriver/test_scope_backfill.py b/tests/deriver/test_scope_backfill.py new file mode 100644 index 000000000..c49994b6b --- /dev/null +++ b/tests/deriver/test_scope_backfill.py @@ -0,0 +1,702 @@ +"""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 select +from sqlalchemy.ext.asyncio import AsyncSession + +from src import 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 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, + configuration={"kind": "scope", "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 _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 _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 _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 _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 + + +# --------------------------------------------------------------------------- +# 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) + + 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 _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 _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 == 200 + + # 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 + 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 + backfill_status = response.json()["backfill_status"] + assert backfill_status[session_name]["state"] == "completed" + assert backfill_status[session_name]["docs_copied"] == 1 + + +# --------------------------------------------------------------------------- +# 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 == 200 + + 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 diff --git a/tests/dreamer/test_card_refresh.py b/tests/dreamer/test_card_refresh.py new file mode 100644 index 000000000..43d1edea8 --- /dev/null +++ b/tests/dreamer/test_card_refresh.py @@ -0,0 +1,335 @@ +"""Tests for the card_refresh dream type (DEV-2000, Scopes RFC prerequisite). + +Covers: +- queue plumbing: payload roundtrip, work-unit key isolation from omni, + enqueue alongside a pending omni dream +- process_dream dispatch of DreamType.CARD_REFRESH (and that it does NOT + advance the omni dream guard pair) +- specialist tool restriction (no observation-mutating tools) +- the low tool-iteration cap +- rebuild mode omitting the prior peer card from the prompt +""" + +from typing import Any +from unittest.mock import AsyncMock, patch + +import pytest +import pytest_asyncio +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from src import models +from src.config import settings +from src.deriver.enqueue import enqueue_dream +from src.dreamer.orchestrator import DreamResult, process_dream +from src.dreamer.specialists import CardRefreshSpecialist +from src.llm import HonchoLLMCallResponse +from src.schemas import DreamType +from src.utils.queue_payload import DreamPayload, create_dream_payload +from src.utils.work_unit import construct_work_unit_key, parse_work_unit_key + +OBSERVATION_MUTATION_TOOLS = { + "create_observations", + "create_observations_deductive", + "create_observations_inductive", + "delete_observations", +} + + +@pytest_asyncio.fixture +async def seeded_collection( + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +) -> models.Collection: + """Create a Collection with an empty dream metadata dict.""" + workspace, peer = sample_data + collection = models.Collection( + observer=peer.name, + observed=peer.name, + workspace_name=workspace.name, + internal_metadata={}, + ) + db_session.add(collection) + await db_session.commit() + await db_session.refresh(collection) + return collection + + +def _make_card_refresh_result() -> DreamResult: + return DreamResult( + run_id="test_run_card", + specialists_run=["card_refresh"], + deduction_success=True, + induction_success=False, + surprisal_enabled=False, + surprisal_conclusion_count=0, + total_iterations=2, + total_duration_ms=42.0, + input_tokens=10, + output_tokens=5, + ) + + +class TestQueuePlumbing: + def test_payload_roundtrip_carries_rebuild(self): + payload_dict = create_dream_payload( + DreamType.CARD_REFRESH, + observer="alice", + observed="bob", + rebuild=True, + ) + validated = DreamPayload(**payload_dict) + assert validated.dream_type == DreamType.CARD_REFRESH + assert validated.rebuild is True + + # Default is False, including for older payloads missing the field. + assert ( + DreamPayload(dream_type=DreamType.OMNI, observer="a", observed="b").rebuild + is False + ) + + def test_work_unit_key_does_not_collide_with_omni(self): + base = {"task_type": "dream", "observer": "alice", "observed": "bob"} + omni_key = construct_work_unit_key("ws", {**base, "dream_type": "omni"}) + card_key = construct_work_unit_key("ws", {**base, "dream_type": "card_refresh"}) + + assert omni_key != card_key + parsed = parse_work_unit_key(card_key) + assert parsed.task_type == "dream" + assert parsed.dream_type == "card_refresh" + assert parsed.observer == "alice" + assert parsed.observed == "bob" + + @pytest.mark.asyncio + async def test_enqueue_alongside_pending_omni( + self, + db_session: AsyncSession, + seeded_collection: models.Collection, + ): + """A pending omni dream must not dedupe away a card_refresh enqueue — + the work-unit keys differ by dream type.""" + await enqueue_dream( + seeded_collection.workspace_name, + observer=seeded_collection.observer, + observed=seeded_collection.observed, + dream_type=DreamType.OMNI, + ) + await enqueue_dream( + seeded_collection.workspace_name, + observer=seeded_collection.observer, + observed=seeded_collection.observed, + dream_type=DreamType.CARD_REFRESH, + rebuild=True, + ) + + items = ( + ( + await db_session.execute( + select(models.QueueItem).where( + models.QueueItem.workspace_name + == seeded_collection.workspace_name, + models.QueueItem.task_type == "dream", + models.QueueItem.processed == False, # noqa: E712 + ) + ) + ) + .scalars() + .all() + ) + assert len(items) == 2 + dream_types = {item.payload["dream_type"] for item in items} + assert dream_types == {"omni", "card_refresh"} + card_item = next( + item for item in items if item.payload["dream_type"] == "card_refresh" + ) + assert card_item.payload["rebuild"] is True + + +class TestProcessDreamDispatch: + @pytest.mark.asyncio + async def test_dispatches_card_refresh( + self, + seeded_collection: models.Collection, + ): + payload = DreamPayload( + dream_type=DreamType.CARD_REFRESH, + observer=seeded_collection.observer, + observed=seeded_collection.observed, + rebuild=True, + trigger_reason="manual", + ) + + with patch( + "src.dreamer.orchestrator.run_card_refresh_dream", + new=AsyncMock(return_value=_make_card_refresh_result()), + ) as mock_run: + await process_dream(payload, seeded_collection.workspace_name) + + assert mock_run.await_args is not None + kwargs = mock_run.await_args.kwargs + assert kwargs["workspace_name"] == seeded_collection.workspace_name + assert kwargs["observer"] == seeded_collection.observer + assert kwargs["observed"] == seeded_collection.observed + assert kwargs["rebuild"] is True + assert kwargs["dream_type"] == "card_refresh" + assert kwargs["trigger_reason"] == "manual" + + @pytest.mark.asyncio + async def test_card_refresh_does_not_advance_dream_guard( + self, + db_session: AsyncSession, + seeded_collection: models.Collection, + ): + """The omni guard pair (last_dream_at / last_dream_document_count) + must not move on a card refresh — it would delay real consolidation.""" + payload = DreamPayload( + dream_type=DreamType.CARD_REFRESH, + observer=seeded_collection.observer, + observed=seeded_collection.observed, + ) + + with patch( + "src.dreamer.orchestrator.run_card_refresh_dream", + new=AsyncMock(return_value=_make_card_refresh_result()), + ): + await process_dream(payload, seeded_collection.workspace_name) + + await db_session.refresh(seeded_collection) + dream_meta: dict[str, Any] = seeded_collection.internal_metadata.get( + "dream", {} + ) + assert "last_dream_at" not in dream_meta + assert "last_dream_document_count" not in dream_meta + + +class TestCardRefreshSpecialist: + def test_tools_exclude_observation_mutation(self): + for rebuild in (False, True): + specialist = CardRefreshSpecialist(rebuild=rebuild) + tool_names = {t["name"] for t in specialist.get_tools()} + assert tool_names == { + "get_recent_observations", + "search_memory", + "update_peer_card", + } + assert not tool_names & OBSERVATION_MUTATION_TOOLS + + def test_tools_without_peer_card_strip_update(self): + specialist = CardRefreshSpecialist() + tool_names = {t["name"] for t in specialist.get_tools(peer_card_enabled=False)} + assert "update_peer_card" not in tool_names + assert not tool_names & OBSERVATION_MUTATION_TOOLS + + def test_low_iteration_cap(self, monkeypatch: pytest.MonkeyPatch): + specialist = CardRefreshSpecialist() + assert specialist.get_max_iterations() == min( + 6, settings.DREAM.MAX_TOOL_ITERATIONS + ) + + monkeypatch.setattr(settings.DREAM, "MAX_TOOL_ITERATIONS", 4) + assert specialist.get_max_iterations() == 4 + + monkeypatch.setattr(settings.DREAM, "MAX_TOOL_ITERATIONS", 30) + assert specialist.get_max_iterations() == 6 + + def test_rebuild_flag_controls_card_injection(self): + assert CardRefreshSpecialist(rebuild=False).inject_peer_card is True + assert CardRefreshSpecialist(rebuild=True).inject_peer_card is False + + def test_rebuild_prompts_instruct_observation_only_build(self): + specialist = CardRefreshSpecialist(rebuild=True) + system_prompt = specialist.build_system_prompt("alice") + assert "REBUILD MODE" in system_prompt + assert "solely from the observations" in system_prompt + + user_prompt = specialist.build_user_prompt("alice", hints=None, peer_card=None) + assert "Rebuild the peer card" in user_prompt + assert "CURRENT PEER CARD" not in user_prompt + + async def _run_specialist( + self, specialist: CardRefreshSpecialist, stored_card: list[str] + ) -> tuple[AsyncMock, AsyncMock]: + """Run the specialist with a fully mocked LLM layer; returns the + (get_peer_card, honcho_llm_call) mocks for inspection.""" + mock_response = HonchoLLMCallResponse( + content="done", + input_tokens=10, + output_tokens=5, + finish_reasons=["stop"], + ) + mock_get_peer_card = AsyncMock(return_value=stored_card) + mock_llm_call = AsyncMock(return_value=mock_response) + + with ( + patch("src.dreamer.specialists.crud.get_peer", new=AsyncMock()), + patch( + "src.dreamer.specialists.crud.get_peer_card", + new=mock_get_peer_card, + ), + patch( + "src.dreamer.specialists.create_tool_executor", + new=AsyncMock(return_value=AsyncMock()), + ), + patch( + "src.dreamer.specialists.honcho_llm_call", + new=mock_llm_call, + ), + ): + result = await specialist.run( + workspace_name="workspace", + observer="alice", + observed="alice", + session_name=None, + ) + assert result.success is True + return mock_get_peer_card, mock_llm_call + + # Sentinel card entry that cannot collide with the prompt's own examples + # (the shared PEER CARD section contains e.g. "IDENTITY: Name: Alice"). + STORED_CARD: list[str] = [ + "IDENTITY: Name: Zorblax-Prime", + "ATTRIBUTE: Location: Ganymede", + ] + + @pytest.mark.asyncio + async def test_refresh_mode_injects_existing_card( + self, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setattr(settings.METRICS, "ENABLED", False) + + mock_get_peer_card, mock_llm_call = await self._run_specialist( + CardRefreshSpecialist(rebuild=False), self.STORED_CARD + ) + + mock_get_peer_card.assert_awaited_once() + assert mock_llm_call.await_args is not None + kwargs = mock_llm_call.await_args.kwargs + user_message = kwargs["messages"][1]["content"] + assert "IDENTITY: Name: Zorblax-Prime" in user_message + assert "CURRENT PEER CARD" in user_message + # Restricted tool offering and low iteration cap reach the LLM call. + tool_names = {t["name"] for t in kwargs["tools"]} + assert not tool_names & OBSERVATION_MUTATION_TOOLS + assert kwargs["max_tool_iterations"] == min( + 6, settings.DREAM.MAX_TOOL_ITERATIONS + ) + + @pytest.mark.asyncio + async def test_rebuild_mode_omits_existing_card( + self, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setattr(settings.METRICS, "ENABLED", False) + + mock_get_peer_card, mock_llm_call = await self._run_specialist( + CardRefreshSpecialist(rebuild=True), self.STORED_CARD + ) + + # The stored card is never even fetched, let alone injected. + mock_get_peer_card.assert_not_awaited() + assert mock_llm_call.await_args is not None + kwargs = mock_llm_call.await_args.kwargs + for message in kwargs["messages"]: + assert "IDENTITY: Name: Zorblax-Prime" not in message["content"] + # No CURRENT PEER CARD block in the user prompt (the system prompt's + # shared taxonomy section legitimately mentions the phrase). + assert "CURRENT PEER CARD" not in kwargs["messages"][1]["content"] diff --git a/tests/routes/test_workspaces.py b/tests/routes/test_workspaces.py index 0350b729d..5934ccfdf 100644 --- a/tests/routes/test_workspaces.py +++ b/tests/routes/test_workspaces.py @@ -9,6 +9,7 @@ from src import models from src.models import Peer, Workspace +from src.schemas import DreamType def test_get_or_create_workspace(client: TestClient): @@ -758,3 +759,52 @@ async def fake_enqueue_dream(*args: Any, **kwargs: Any) -> None: "Loop 4: enqueue_dream no longer accepts document_count; the baseline " "is written atomically with last_dream_at in process_dream." ) + + +@pytest.mark.asyncio +async def test_schedule_dream_card_refresh_forwards_rebuild( + client: TestClient, + db_session: AsyncSession, + sample_data: tuple[Workspace, Peer], +): + """POST /schedule_dream accepts dream_type=card_refresh and forwards the + rebuild flag to enqueue_dream (manual/event-driven card refreshes bypass + the volume gates by design).""" + workspace, peer = sample_data + + collection = models.Collection( + observer=peer.name, + observed=peer.name, + workspace_name=workspace.name, + internal_metadata={}, + ) + db_session.add(collection) + await db_session.commit() + + captured: dict[str, Any] = {} + + async def fake_enqueue_dream(*args: Any, **kwargs: Any) -> None: + captured["args"] = args + captured["kwargs"] = kwargs + + with ( + patch("src.routers.workspaces.settings.DREAM.ENABLED", True), + patch( + "src.routers.workspaces.enqueue_dream", + new=AsyncMock(side_effect=fake_enqueue_dream), + ), + ): + response = client.post( + f"/v3/workspaces/{workspace.name}/schedule_dream", + json={ + "observer": peer.name, + "observed": peer.name, + "dream_type": "card_refresh", + "rebuild": True, + }, + ) + + assert response.status_code == 204, response.text + assert "kwargs" in captured, "enqueue_dream was not called" + assert captured["kwargs"]["dream_type"] == DreamType.CARD_REFRESH + assert captured["kwargs"]["rebuild"] is True diff --git a/tests/telemetry/test_events.py b/tests/telemetry/test_events.py index bce0a38c8..a5d959e5a 100644 --- a/tests/telemetry/test_events.py +++ b/tests/telemetry/test_events.py @@ -239,6 +239,7 @@ def test_call_purpose_enum_values(self): assert CallPurpose.DIALECTIC_ANSWER.value == "dialectic.answer" assert CallPurpose.DREAM_DEDUCTION.value == "dream.deduction" assert CallPurpose.DREAM_INDUCTION.value == "dream.induction" + assert CallPurpose.DREAM_CARD_REFRESH.value == "dream.card_refresh" assert CallPurpose.SUMMARY_SHORT.value == "summary.short" assert CallPurpose.SUMMARY_LONG.value == "summary.long" diff --git a/tests/utils/test_agent_tools.py b/tests/utils/test_agent_tools.py index 12037bf7a..071b6308e 100644 --- a/tests/utils/test_agent_tools.py +++ b/tests/utils/test_agent_tools.py @@ -248,6 +248,36 @@ async def test_dialectic_context_forces_deductive( assert doc.level == "deductive" assert doc.source_ids == ["premise1", "premise2"] + async def test_non_deriver_context_rejects_explicit( + self, + db_session: AsyncSession, + make_tool_context: Callable[..., ToolContext], + ): + """Session-purity invariant: agents without current_messages (dreamer + specialists, dialectic) must not create explicit-level observations, + even when they pass level='explicit' to the generic tool.""" + ctx = make_tool_context(current_messages=None) + + result = await _handle_create_observations( + ctx, + { + "observations": [ + {"content": "Claims to be a doctor", "level": "explicit"}, + ] + }, + ) + + assert isinstance(result, str) + assert "ERROR" in result + assert "explicit" in result + + # Verify nothing landed in the DB + stmt = select(models.Document).where( + models.Document.content == "Claims to be a doctor" + ) + doc = (await db_session.execute(stmt)).scalar_one_or_none() + assert doc is None + async def test_source_ids_display_prefix_is_stripped( self, db_session: AsyncSession,