fix: date dreamer conclusions to latest source observation

src/utils/agent_tools.py

@@ -24,7 +24,11 @@ from src.telemetry.events import (
     emit,
 )
 from src.utils import summarizer
-from src.utils.formatting import format_new_turn_with_timestamp, utc_now_iso
+from src.utils.formatting import (
+    format_datetime_utc,
+    format_new_turn_with_timestamp,
+    utc_now_iso,
+)
 from src.utils.representation import Representation
 from src.utils.types import ToolResult, embedding_call_purpose, get_current_iteration
 
@@ -1306,6 +1310,49 @@ def _normalize_observation_id(obs_id: str) -> str:
     return obs_id.strip()
 
 
+async def _latest_source_timestamp(
+    ctx: ToolContext,
+    observations: list[schemas.ObservationInput],
+) -> str | None:
+    """Latest ``message_created_at`` across all source observations in the batch.
+
+    Dreamer conclusions (deductive/inductive) are derived from existing
+    observations referenced by ``source_ids`` rather than from live messages. To
+    keep their timeline aligned with their evidence, we date them to the most
+    recent source observation instead of the dream-run time. Returns None if no
+    source_ids resolve to a usable timestamp (caller falls back to now).
+    """
+    source_ids: list[str] = []
+    for obs in observations:
+        if obs.source_ids:
+            source_ids.extend(obs.source_ids)
+    if not source_ids:
+        return None
+
+    async with tracked_db("create_observations.source_ts", read_only=True) as db:
+        docs = await crud.fetch_documents_by_ids(
+            db,
+            workspace_name=ctx.workspace_name,
+            observer=ctx.observer,
+            observed=ctx.observed,
+            document_ids=list(set(source_ids)),
+        )
+
+    latest: datetime | None = None
+    for doc in docs:
+        raw = doc.internal_metadata.get("message_created_at")
+        if not isinstance(raw, str):
+            continue
+        try:
+            parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+        except ValueError:
+            continue
+        if latest is None or parsed > latest:
+            latest = parsed
+
+    return format_datetime_utc(latest) if latest is not None else None
+
+
 async def _handle_create_observations_impl(
     ctx: ToolContext,
     tool_input: dict[str, Any],
@@ -1370,8 +1417,11 @@ async def _handle_create_observations_impl(
         message_ids = [msg.id for msg in ctx.current_messages]
         message_created_at = str(ctx.current_messages[-1].created_at)
     else:
+        # Dreamer path: no current messages. Backdate the conclusion to the latest source observation.
         message_ids = []
-        message_created_at = utc_now_iso()
+        message_created_at = (
+            await _latest_source_timestamp(ctx, observations)
+        ) or utc_now_iso()
 
     # Use lock to serialize database writes (prevents concurrent commit issues)
     async with ctx.db_lock: