diff --git a/src/utils/agent_tools.py b/src/utils/agent_tools.py index d4df768b..64cef96f 100644 --- a/src/utils/agent_tools.py +++ b/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: