offendingcommit
diff --git a/‎src/crud/collection.py‎
Lines changed: 22 additions & 0 deletions b/‎src/crud/collection.py‎
Lines changed: 22 additions & 0 deletions
diff --git a/‎src/deriver/enqueue.py‎
Lines changed: 5 additions & 29 deletions b/‎src/deriver/enqueue.py‎
Lines changed: 5 additions & 29 deletions
diff --git a/‎src/dreamer/dream_scheduler.py‎
Lines changed: 51 additions & 26 deletions b/‎src/dreamer/dream_scheduler.py‎
Lines changed: 51 additions & 26 deletions
diff --git a/‎src/dreamer/orchestrator.py‎
Lines changed: 31 additions & 1 deletion b/‎src/dreamer/orchestrator.py‎
Lines changed: 31 additions & 1 deletion
diff --git a/‎src/routers/workspaces.py‎
Lines changed: 1 addition & 13 deletions b/‎src/routers/workspaces.py‎
Lines changed: 1 addition & 13 deletions
@@ -81,6 +81,7 @@ async def get_collection(
     *,
     observer: str,
     observed: str,
+    with_for_update: bool = False,
 ) -> models.Collection:
     """
     Get a collection by observer/observed for a workspace.
@@ -90,13 +91,34 @@ async def get_collection(
         workspace_name: Name of the workspace
         observer: Name of the observing peer (owns the collection)
         observed: Name of the observed peer
+        with_for_update: If True, acquire a row-level lock (SELECT ... FOR UPDATE)
+            on the collection. Bypasses the cache so the lock is actually held
+            by the current transaction. Callers using this flag must wrap the
+            read and subsequent write in the same transaction (the lock is
+            released on commit/rollback).
 
     Returns:
         The collection if found
 
     Raises:
         ResourceNotFoundException: If the collection does not exist
     """
+    if with_for_update:
+        # Row-lock path: go direct to DB (skip cache) so the FOR UPDATE lock
+        # is actually acquired on the row in the current transaction. The
+        # cached dict path would return without issuing SELECT ... FOR UPDATE.
+        stmt = (
+            select(models.Collection)
+            .where(models.Collection.workspace_name == workspace_name)
+            .where(models.Collection.observer == observer)
+            .where(models.Collection.observed == observed)
+            .with_for_update()
+        )
+        collection = await db.scalar(stmt)
+        if collection is None:
+            raise ResourceNotFoundException("Collection not found")
+        return collection
+
     data = await _fetch_collection(db, workspace_name, observer, observed)
     if data is None:
         raise ResourceNotFoundException("Collection not found")
 
@@ -1,5 +1,4 @@
 import logging
-from datetime import datetime, timezone
 from typing import Any, Literal
 
 from sqlalchemy import exists, insert, select
@@ -436,27 +435,26 @@ async def enqueue_dream(
     observer: str,
     observed: str,
     dream_type: schemas.DreamType,
-    document_count: int,
     session_name: str | None = None,
 ) -> None:
     """
     Enqueue a dream task for immediate processing by the deriver.
 
+    Does not touch collection.internal_metadata["dream"] — both guard fields
+    are written atomically in process_dream on successful completion.
+
     Deduplication: If a dream with the same work_unit_key is already in-progress
-    (has an ActiveQueueSession), the enqueue is skipped to prevent running
-    multiple dreams concurrently for the same collection.
+    (has an ActiveQueueSession) or pending in the queue, the enqueue is skipped.
 
     Args:
         workspace_name: Name of the workspace
         observer: Name of the observer peer
         observed: Name of the observed peer
         dream_type: Type of dream to execute
-        document_count: Current document count for metadata update
         session_name: Name of the session to scope the dream to if specified
     """
     async with tracked_db("dream_enqueue") as db_session:
         try:
-            # Create the dream queue record
             dream_record = create_dream_record(
                 workspace_name,
                 observer=observer,
@@ -467,11 +465,6 @@ async def enqueue_dream(
 
             work_unit_key = dream_record["work_unit_key"]
 
-            # Check if a dream with this work_unit_key is currently in progress
-            # (has an ActiveQueueSession, meaning a worker is processing it)
-            # We only block on in-progress dreams, not pending ones - if there's
-            # a pending dream, we don't need to add another one anyway since
-            # the queue processor will pick it up.
             in_progress_check = select(
                 exists(
                     select(models.ActiveQueueSession.id).where(
@@ -491,7 +484,6 @@ async def enqueue_dream(
                 )
                 return
 
-            # Check if there's already a pending dream with the same work_unit_key
             pending_check = select(
                 exists(
                     select(QueueItem.id).where(
@@ -512,25 +504,9 @@ async def enqueue_dream(
                 )
                 return
 
-            # Insert into queue
             stmt = insert(QueueItem).returning(QueueItem)
             await db_session.execute(stmt, [dream_record])
-
-            # Update collection metadata (CRUD handles cache invalidation)
-            now_iso = datetime.now(timezone.utc).isoformat()
-            await crud.update_collection_internal_metadata(
-                db_session,
-                workspace_name,
-                observer,
-                observed,
-                update_data={
-                    "dream": {
-                        "last_dream_document_count": document_count,
-                        "last_dream_at": now_iso,
-                    }
-                },
-            )
-            # update_collection_internal_metadata commits already
+            await db_session.commit()
 
             logger.info(
                 "Enqueued dream task for %s/%s/%s (type: %s)",
 
@@ -4,7 +4,7 @@
 from logging import getLogger
 
 import sentry_sdk
-from sqlalchemy import func, select
+from sqlalchemy import exists, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from src import models
@@ -160,20 +160,19 @@ async def execute_dream(
         observer: str,
         observed: str,
     ) -> None:
-        """Execute the dream by enqueueing it and updating collection metadata."""
-        # Import here to avoid circular dependency
+        """Execute the dream by enqueueing it."""
         from src import crud
         from src.deriver.enqueue import enqueue_dream
         from src.utils.config_helpers import get_configuration
 
-        # Find the most recent session and get current document count
         async with tracked_db("dream_session_lookup") as db:
             stmt = (
                 select(models.Document.session_name)
                 .where(
                     models.Document.workspace_name == workspace_name,
                     models.Document.observer == observer,
                     models.Document.observed == observed,
+                    models.Document.level == "explicit",
                 )
                 .order_by(models.Document.created_at.desc())
                 .limit(1)
@@ -186,14 +185,6 @@ async def execute_dream(
                 )
                 return
 
-            # Get current document count at execution time (not stale from scheduling)
-            count_stmt = select(func.count(models.Document.id)).where(
-                models.Document.workspace_name == workspace_name,
-                models.Document.observer == observer,
-                models.Document.observed == observed,
-            )
-            current_document_count = int(await db.scalar(count_stmt) or 0)
-
             session = await crud.get_session(
                 db, workspace_name=workspace_name, session_name=session_name
             )
@@ -212,7 +203,6 @@ async def execute_dream(
             observer=observer,
             observed=observed,
             dream_type=dream_type,
-            document_count=current_document_count,
             session_name=session_name,
         )
 
@@ -231,13 +221,18 @@ async def check_and_schedule_dream(
     collection: models.Collection,
 ) -> bool:
     """
-    Check if a collection has reached the document threshold and schedule a timer-based dream.
+    From the moment a dream is scheduled until it completes or fails, no second
+    dream may be enqueued for the same (workspace, observer, observed) — and the
+    baseline count advances only when consolidation actually happened.
+
+    Check if a collection has reached the explicit-observation threshold and schedule a timer-based dream.
 
     This function only schedules a timer-based dream if:
     1. Dreams are enabled
-    2. Document threshold is reached
+    2. Explicit-observation threshold is reached (dreamer output does not count)
     3. Minimum hours between dreams have passed
-    4. No dream is already scheduled for this collection
+    4. No dream is already pending in the queue for this collection (in-flight check)
+    5. No dream is already scheduled for this collection
 
     Args:
         db: Database session
@@ -249,38 +244,36 @@ async def check_and_schedule_dream(
     if not settings.DREAM.ENABLED:
         return False
 
-    # Get dream metadata from internal_metadata
     dream_metadata = collection.internal_metadata.get("dream", {})
     last_dream_document_count = dream_metadata.get("last_dream_document_count", 0)
     last_dream_at = dream_metadata.get("last_dream_at")
 
-    # Count current documents in the collection
+    # Count explicit-level docs only: dreamer output (deductive/inductive/
+    # contradiction) would inflate the threshold and create a feedback loop.
     count_stmt = select(func.count(models.Document.id)).where(
         models.Document.workspace_name == collection.workspace_name,
         models.Document.observer == collection.observer,
         models.Document.observed == collection.observed,
+        models.Document.level == "explicit",
     )
-    current_document_count = int(await db.scalar(count_stmt) or 0)
+    current_explicit_count = int(await db.scalar(count_stmt) or 0)
 
-    # Calculate documents added since last dream
-    documents_since_last_dream = current_document_count - last_dream_document_count
+    documents_since_last_dream = current_explicit_count - last_dream_document_count
 
     logger.debug(
         "Dream check",
         extra={
             "workspace_name": collection.workspace_name,
             "observer": collection.observer,
             "observed": collection.observed,
-            "current_document_count": current_document_count,
+            "current_explicit_count": current_explicit_count,
             "last_dream_document_count": last_dream_document_count,
             "documents_since_last_dream": documents_since_last_dream,
             "document_threshold": settings.DREAM.DOCUMENT_THRESHOLD,
         },
     )
 
-    # Only schedule timer if document threshold is reached
     if documents_since_last_dream >= settings.DREAM.DOCUMENT_THRESHOLD:
-        # Check if we're within minimum hours between dreams
         if last_dream_at:
             try:
                 last_dream_time = datetime.fromisoformat(last_dream_at)
@@ -299,11 +292,43 @@ async def check_and_schedule_dream(
                     f"Invalid last_dream_at timestamp: {last_dream_at}, error: {e}"
                 )
 
+        # Queue is source of truth for in-flight dreams; mirrors
+        # uq_queue_dream_pending_work_unit_key.
+        enabled_dream_types = settings.DREAM.ENABLED_TYPES
+        pending_keys = [
+            construct_work_unit_key(
+                collection.workspace_name,
+                {
+                    "task_type": "dream",
+                    "observer": collection.observer,
+                    "observed": collection.observed,
+                    "dream_type": dream_type,
+                },
+            )
+            for dream_type in enabled_dream_types
+        ]
+        pending_exists = await db.scalar(
+            select(
+                exists(
+                    select(models.QueueItem.id).where(
+                        models.QueueItem.task_type == "dream",
+                        models.QueueItem.processed == False,  # noqa: E712
+                        models.QueueItem.work_unit_key.in_(pending_keys),
+                    )
+                )
+            )
+        )
+        if pending_exists:
+            logger.info(
+                "Skipping dream schedule for %s/%s: pending dream already in queue",
+                collection.observer,
+                collection.observed,
+            )
+            return False
+
         dream_scheduler = get_dream_scheduler()
         if dream_scheduler:
-            enabled_dream_types = settings.DREAM.ENABLED_TYPES
             for dream_type in enabled_dream_types:
-                # Include dream_type in key so each dream type can be tracked independently
                 dream_work_unit_key = construct_work_unit_key(
                     collection.workspace_name,
                     {
 
@@ -17,11 +17,13 @@
 import time
 import uuid
 from dataclasses import dataclass
+from datetime import datetime, timezone
 from typing import Any
 
 import sentry_sdk
+from sqlalchemy import func, select
 
-from src import crud
+from src import crud, models
 from src.config import settings
 from src.dependencies import tracked_db
 from src.dreamer.specialists import SPECIALISTS, SpecialistResult
@@ -323,6 +325,34 @@ async def process_dream(
                         + f"duration={result.total_duration_ms:.0f}ms"
                     )
 
+                    # Both guard fields advance together only on successful consolidation.
+                    now_iso = datetime.now(timezone.utc).isoformat()
+                    async with tracked_db("dream.guard_pair_write") as db:
+                        collection = await crud.get_collection(
+                            db,
+                            workspace_name,
+                            observer=payload.observer,
+                            observed=payload.observed,
+                            with_for_update=True,
+                        )
+                        count_stmt = select(func.count(models.Document.id)).where(
+                            models.Document.workspace_name == workspace_name,
+                            models.Document.observer == payload.observer,
+                            models.Document.observed == payload.observed,
+                            models.Document.level == "explicit",
+                        )
+                        current_explicit_count = int(await db.scalar(count_stmt) or 0)
+                        dream_meta = dict(collection.internal_metadata.get("dream", {}))
+                        dream_meta["last_dream_at"] = now_iso
+                        dream_meta["last_dream_document_count"] = current_explicit_count
+                        await crud.update_collection_internal_metadata(
+                            db,
+                            workspace_name,
+                            payload.observer,
+                            payload.observed,
+                            update_data={"dream": dream_meta},
+                        )
+
     except Exception as e:
         logger.error(
             f"Error processing dream task {payload.dream_type} for {payload.observer}/{payload.observed}: {str(e)}",
 
@@ -3,10 +3,9 @@
 from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query, Response
 from fastapi_pagination import Page
 from fastapi_pagination.ext.sqlalchemy import apaginate
-from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from src import crud, models, schemas
+from src import crud, schemas
 from src.config import settings
 from src.dependencies import db
 from src.deriver.enqueue import enqueue_deletion, enqueue_dream
@@ -201,7 +200,6 @@ async def schedule_dream(
     request: schemas.ScheduleDreamRequest = Body(
         ..., description="Dream scheduling parameters"
     ),
-    db: AsyncSession = db,
 ):
     """
     Manually schedule a dream task for a specific collection.
@@ -224,21 +222,11 @@ async def schedule_dream(
     observed = request.observed if request.observed is not None else request.observer
     dream_type = request.dream_type
 
-    # Count documents in the collection
-    count_stmt = select(func.count(models.Document.id)).where(
-        models.Document.workspace_name == workspace_id,
-        models.Document.observer == observer,
-        models.Document.observed == observed,
-    )
-    document_count = int(await db.scalar(count_stmt) or 0)
-
-    # Enqueue the dream task for immediate processing
     await enqueue_dream(
         workspace_id,
         observer=observer,
         observed=observed,
         dream_type=dream_type,
-        document_count=document_count,
         session_name=request.session_id,
     )