fix: date dreamer conclusions to latest source observation (#890)

* fix: date dreamer conclusions to latest source observation

* fix: correct and normalize dreamer conclusion timestamps

* fix: updating documentation
This commit is contained in:
ajspig 2026-07-15 10:36:48 -04:00 committed by GitHub
parent 5ad22840d8
commit 0842c8e21a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 60 additions and 3 deletions

View File

@ -24,7 +24,12 @@ 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,
parse_datetime_iso,
utc_now_iso,
)
from src.utils.representation import Representation
from src.utils.types import ToolResult, embedding_call_purpose, get_current_iteration
@ -1306,6 +1311,52 @@ 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.
Their logical timestamp is the point when the conclusion became possible
from its evidence, not when the dreamer happened to run, so we date
``internal_metadata["message_created_at"]`` to the most recent source
observation. The physical ``Document.created_at`` column remains the insert
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
latest: datetime | None = 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)),
)
for doc in docs:
raw = doc.internal_metadata.get("message_created_at")
if not isinstance(raw, str):
continue
try:
# always tz-aware, so the comparison below can't crash on mixed formats
parsed = parse_datetime_iso(raw)
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],
@ -1368,10 +1419,16 @@ async def _handle_create_observations_impl(
# Determine message context
if ctx.current_messages:
message_ids = [msg.id for msg in ctx.current_messages]
message_created_at = str(ctx.current_messages[-1].created_at)
# same ISO-8601 Z format as the dreamer path below
message_created_at = format_datetime_utc(ctx.current_messages[-1].created_at)
else:
# Dreamer path: no current messages. Backdate the conclusion to the
# latest source observation, which is when the inference became possible.
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: