From 953849b54221eb2507f93fae08851440b819a0dc Mon Sep 17 00:00:00 2001 From: shixi-li Date: Tue, 28 Jul 2026 02:24:12 +0800 Subject: [PATCH 1/2] fix: validate source_ids against real documents on the write path Dream specialists cite evidence via source_ids, but nothing validated that a cited id refers to a document that actually exists: validate_level_fields only checks the list is non-empty, the tool schema only checks minItems, and create_observations persisted the ids unchecked. A fabricated id became permanent false provenance, broke get_children_of_document traversal, and silently degraded _latest_source_timestamp into a wrong timestamp. - resolve cited source_ids against existing documents inside create_observations before persistence: unresolvable ids are stripped, and an observation left with fewer real sources than its level requires (1, or 2 for contradiction, matching validate_level_fields) is rejected and surfaced as an ObservationFailure instead of being stored - tell specialists that search_messages results carry no citable id: the message-snippet header, the source_ids schema descriptions (mirroring delete_observations' exact-[id:xxx] rule), and the deduction/induction RULES now all state that message text must go in premises/sources and invented ids are discarded Fixes #939 --- src/dreamer/specialists.py | 14 ++-- src/utils/agent_tools.py | 112 ++++++++++++++++++++++++++-- tests/utils/test_agent_tools.py | 127 +++++++++++++++++++++++++++++++- 3 files changed, 238 insertions(+), 15 deletions(-) diff --git a/src/dreamer/specialists.py b/src/dreamer/specialists.py index 61e8a9d3..6ccea7b4 100644 --- a/src/dreamer/specialists.py +++ b/src/dreamer/specialists.py @@ -611,10 +611,11 @@ Use `create_observations_deductive`. 1. Don't explain your reasoning - just call tools 2. Create observations based on what you ACTUALLY FIND, not what you expect 3. Always include source_ids linking to the observations you're synthesizing -4. Empty or missing source_ids will be rejected -5. Delete outdated observations - don't leave duplicates -6. Quality over quantity - fewer good deductions beat many weak ones -7. When you are finished, do not output a summary of what you did - output only the token DONE""" +4. Copy source_ids exactly from the [id:xxx] shown in observation results - `search_messages` results have no ID and cannot be cited; invented IDs are discarded +5. Empty or missing source_ids will be rejected +6. Delete outdated observations - don't leave duplicates +7. Quality over quantity - fewer good deductions beat many weak ones +8. When you are finished, do not output a summary of what you did - output only the token DONE""" def build_user_prompt( self, @@ -745,8 +746,9 @@ Use `create_observations_inductive`. 3. Confidence based on evidence count: 2=low, 3-4=medium, 5+=high 4. Look for HOW things change over time, not just static facts 5. Include source_ids - always link back to evidence -6. Empty or missing source_ids will be rejected -7. When you are finished, do not output a summary of what you did - output only the token DONE""" +6. Copy source_ids exactly from the [id:xxx] shown in observation results - `search_messages` results have no ID and cannot be cited; invented IDs are discarded +7. Empty or missing source_ids will be rejected +8. When you are finished, do not output a summary of what you did - output only the token DONE""" def build_user_prompt( self, diff --git a/src/utils/agent_tools.py b/src/utils/agent_tools.py index 9f214009..30c1d9f7 100644 --- a/src/utils/agent_tools.py +++ b/src/utils/agent_tools.py @@ -106,7 +106,11 @@ def _base_observation_properties() -> dict[str, Any]: "description": ( "Document IDs of source or premise observations. Required and " + "must be non-empty for deductive, inductive, and contradiction " - + "observations." + + "observations. Copy the exact ID shown in [id:xxx] format from " + + "observation results (e.g. for '[id:abc123XYZ]' pass " + + "'abc123XYZ'). Message search results carry no ID and cannot " + + "be cited here; IDs that do not match an existing observation " + + "are discarded." ), }, "premises": { @@ -223,7 +227,7 @@ def _deductive_observation_item_schema() -> dict[str, Any]: "type": "array", "items": {"type": "string"}, "minItems": 1, - "description": "Required non-empty list of source observation IDs supporting the deduction", + "description": "Required non-empty list of source observation IDs supporting the deduction. Copy the exact ID shown in [id:xxx] format from observation results; message search results carry no ID and cannot be cited", }, "premises": { "type": "array", @@ -249,7 +253,7 @@ def _inductive_observation_item_schema() -> dict[str, Any]: "type": "array", "items": {"type": "string"}, "minItems": 2, - "description": "Required list of at least two source observation IDs supporting the pattern", + "description": "Required list of at least two source observation IDs supporting the pattern. Copy the exact ID shown in [id:xxx] format from observation results; message search results carry no ID and cannot be cited", }, "sources": { "type": "array", @@ -867,6 +871,88 @@ CARD_REFRESH_SPECIALIST_TOOLS: list[dict[str, Any]] = [ ] +# Levels whose source_ids must resolve to real documents before persistence. +_SOURCE_GROUNDED_LEVELS: tuple[str, ...] = ("deductive", "inductive", "contradiction") + + +async def _filter_ungrounded_source_ids( + observations: list[schemas.ObservationInput], + *, + workspace_name: str, + observer: str, + observed: str, +) -> tuple[list[schemas.ObservationInput], list[ObservationFailure]]: + """Drop cited source_ids that resolve to no existing document. + + Derived observations must cite the documents they are built on, but models + sometimes fabricate ids (or paste message text) to satisfy the tool + schema, and nothing downstream re-checks them: a dangling id becomes + permanent false provenance, breaks ``get_child_observations`` traversal, + and silently skews ``_latest_source_timestamp``. Resolve every cited id + against real documents before persistence: fabricated ids are stripped, + and an observation left with fewer real sources than + ``validate_level_fields`` requires for its level (1, or 2 for + contradiction) is rejected as an ``ObservationFailure`` rather than + stored. + + Returns (grounded observations, failures for ungrounded observations). + """ + cited_ids: set[str] = set() + for obs in observations: + if obs.level in _SOURCE_GROUNDED_LEVELS and obs.source_ids: + cited_ids.update(obs.source_ids) + if not cited_ids: + return observations, [] + + async with tracked_db("create_observations.ground_sources", read_only=True) as db: + docs = await crud.fetch_documents_by_ids( + db, + workspace_name=workspace_name, + observer=observer, + observed=observed, + document_ids=list(cited_ids), + ) + # Collect ids inside the session scope — the ORM objects expire when + # it closes, and a refresh outside it has no session to run on. + resolved_ids = {doc.id for doc in docs} + + grounded: list[schemas.ObservationInput] = [] + failed: list[ObservationFailure] = [] + for obs in observations: + if obs.level not in _SOURCE_GROUNDED_LEVELS or not obs.source_ids: + grounded.append(obs) + continue + real_ids = [sid for sid in obs.source_ids if sid in resolved_ids] + dropped = len(obs.source_ids) - len(real_ids) + # Mirror the per-level minimums enforced by validate_level_fields. + min_sources = 2 if obs.level == "contradiction" else 1 + if len(real_ids) < min_sources: + failed.append( + ObservationFailure( + content_preview=obs.content[:50], + error=( + f"{dropped} of {len(obs.source_ids)} source_ids do not " + f"resolve to existing observations; '{obs.level}' requires " + f"at least {min_sources} real source(s) " + f"(cite only ids shown as [id:xxx] in observation results)" + ), + ) + ) + continue + if dropped: + logger.warning( + "Dropped %d unresolvable source_ids from %s observation in %s/%s/%s", + dropped, + obs.level, + workspace_name, + observer, + observed, + ) + obs = obs.model_copy(update={"source_ids": real_ids}) + grounded.append(obs) + return grounded, failed + + async def create_observations( observations: list[schemas.ObservationInput], observer: str, @@ -911,6 +997,21 @@ async def create_observations( logger.info("No non-empty observations to create") return ObservationsCreatedResult(created_count=0, created_levels=[], failed=[]) + # Ground cited source_ids against real documents before persistence — + # fabricated ids are stripped and observations left without the required + # real sources are rejected (see _filter_ungrounded_source_ids). + normalized_observations, failed = await _filter_ungrounded_source_ids( + normalized_observations, + workspace_name=workspace_name, + observer=observer, + observed=observed, + ) + if not normalized_observations: + logger.info("No observations with resolvable source_ids to create") + return ObservationsCreatedResult( + created_count=0, created_levels=[], failed=failed + ) + # Ensure collection exists (short DB scope) async with tracked_db("create_observations.collection") as db: await crud.get_or_create_collection( @@ -942,7 +1043,6 @@ async def create_observations( # Build document objects with pre-computed embeddings documents: list[schemas.DocumentCreate] = [] - failed: list[ObservationFailure] = [] for i, obs in enumerate(normalized_observations): embedding: list[float] if embeddings_by_index is not None: @@ -2364,7 +2464,9 @@ def _format_message_snippets( ) output = ( - f"Found {total_matches} matching messages in {len(snippets)} conversation snippets {desc}:\n\n" + f"Found {total_matches} matching messages in {len(snippets)} conversation snippets {desc}.\n" + + "These are raw messages with no observation ID - do not cite them in " + + "source_ids; use their text in premises/sources instead:\n\n" + "\n\n".join(snippet_texts) ) # `[0]` extracts the truncated text — telemetry signal is discarded here diff --git a/tests/utils/test_agent_tools.py b/tests/utils/test_agent_tools.py index 71b93eb0..1f8506b6 100644 --- a/tests/utils/test_agent_tools.py +++ b/tests/utils/test_agent_tools.py @@ -217,9 +217,12 @@ class TestCreateObservations: async def test_dialectic_context_forces_deductive( self, db_session: AsyncSession, + tool_test_data: Any, make_tool_context: Callable[..., ToolContext], ): """Dialectic context (no current_messages) forces observations to be deductive.""" + *_, documents = tool_test_data + source_ids = [documents[0].id, documents[1].id] ctx = make_tool_context(current_messages=None) result = await _handle_create_observations( @@ -228,7 +231,7 @@ class TestCreateObservations: "observations": [ { "content": "Inferred preference for quiet spaces", - "source_ids": ["premise1", "premise2"], + "source_ids": list(source_ids), "premises": [ "User mentioned working in libraries", "User avoids noisy cafes", @@ -248,7 +251,7 @@ class TestCreateObservations: doc = (await db_session.execute(stmt)).scalar_one_or_none() assert doc is not None assert doc.level == "deductive" - assert doc.source_ids == ["premise1", "premise2"] + assert doc.source_ids == source_ids async def test_non_deriver_context_rejects_explicit( self, @@ -283,10 +286,12 @@ class TestCreateObservations: async def test_source_ids_display_prefix_is_stripped( self, db_session: AsyncSession, + tool_test_data: Any, make_tool_context: Callable[..., ToolContext], ): """Models sometimes copy the '[id:xxx]' display format into source_ids; the prefix must be stripped so provenance links reference real IDs.""" + *_, documents = tool_test_data ctx = make_tool_context(current_messages=None) result = await _handle_create_observations( @@ -295,7 +300,10 @@ class TestCreateObservations: "observations": [ { "content": "Inferred preference for early mornings", - "source_ids": ["id:premise1", "ID:premise2"], + "source_ids": [ + f"id:{documents[0].id}", + f"ID:{documents[1].id}", + ], "premises": [ "User schedules meetings before 9am", "User mentions waking at 5:30", @@ -312,7 +320,118 @@ class TestCreateObservations: ) doc = (await db_session.execute(stmt)).scalar_one_or_none() assert doc is not None - assert doc.source_ids == ["premise1", "premise2"] + assert doc.source_ids == [documents[0].id, documents[1].id] + + async def test_fabricated_source_ids_reject_ungrounded_observation( + self, + db_session: AsyncSession, + make_tool_context: Callable[..., ToolContext], + ): + """An observation whose cited source_ids resolve to no existing + documents must be rejected and surfaced as a failure, not persisted + with false provenance.""" + ctx = make_tool_context(current_messages=None) + + result = await _handle_create_observations( + ctx, + { + "observations": [ + { + "content": "Conclusion citing invented evidence", + "source_ids": [ + "fabricated-id-that-does-not-exist-1", + "fabricated-id-that-does-not-exist-2", + ], + "premises": ["Premise that was never observed"], + }, + ] + }, + ) + + assert "Created 0 observations" in result + assert "source_ids" in str(result) + + stmt = select(models.Document).where( + models.Document.content == "Conclusion citing invented evidence" + ) + doc = (await db_session.execute(stmt)).scalar_one_or_none() + assert doc is None + + async def test_fabricated_source_ids_stripped_when_real_source_remains( + self, + db_session: AsyncSession, + tool_test_data: Any, + make_tool_context: Callable[..., ToolContext], + ): + """Fabricated ids are dropped while the observation survives on its + remaining real sources.""" + *_, documents = tool_test_data + real_id = documents[0].id + ctx = make_tool_context(current_messages=None) + + result = await _handle_create_observations( + ctx, + { + "observations": [ + { + "content": "Conclusion citing mixed evidence", + "source_ids": [real_id, "fabricated-id-that-does-not-exist"], + "premises": ["User likes coffee"], + }, + ] + }, + ) + + assert "Created 1 observations" in result + + stmt = select(models.Document).where( + models.Document.content == "Conclusion citing mixed evidence" + ) + doc = (await db_session.execute(stmt)).scalar_one_or_none() + assert doc is not None + assert doc.source_ids == [real_id] + + async def test_mixed_batch_keeps_grounded_rejects_ungrounded( + self, + db_session: AsyncSession, + tool_test_data: Any, + make_tool_context: Callable[..., ToolContext], + ): + """A batch mixing grounded and ungrounded observations persists only + the grounded one and reports the other as failed.""" + *_, documents = tool_test_data + real_id = documents[1].id + ctx = make_tool_context(current_messages=None) + + result = await _handle_create_observations( + ctx, + { + "observations": [ + { + "content": "Grounded conclusion", + "source_ids": [real_id], + "premises": ["User works remotely"], + }, + { + "content": "Ungrounded conclusion", + "source_ids": ["fabricated-id-that-does-not-exist"], + "premises": ["Premise that was never observed"], + }, + ] + }, + ) + + assert "Created 1 observations" in result + assert "Failed 1" in result + + stmt = select(models.Document).where( + models.Document.content.in_( + ["Grounded conclusion", "Ungrounded conclusion"] + ) + ) + docs = (await db_session.execute(stmt)).scalars().all() + assert len(docs) == 1 + assert docs[0].content == "Grounded conclusion" async def test_empty_observations_list_returns_error( self, make_tool_context: Callable[..., ToolContext] From 2131e6d338294933605576191b6b9e5f7b85c83b Mon Sep 17 00:00:00 2001 From: Shixi Li Date: Tue, 28 Jul 2026 11:59:21 +0800 Subject: [PATCH 2/2] test(dreamer): cover partially grounded contradiction --- tests/utils/test_agent_tools.py | 42 +++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/tests/utils/test_agent_tools.py b/tests/utils/test_agent_tools.py index 1f8506b6..60721955 100644 --- a/tests/utils/test_agent_tools.py +++ b/tests/utils/test_agent_tools.py @@ -391,6 +391,48 @@ class TestCreateObservations: assert doc is not None assert doc.source_ids == [real_id] + async def test_contradiction_rejected_when_only_one_real_source_remains( + self, + db_session: AsyncSession, + tool_test_data: Any, + make_tool_context: Callable[..., ToolContext], + ): + """A contradiction must still have two real sources after fabricated + source IDs are removed.""" + *_, documents = tool_test_data + real_id = documents[0].id + ctx = make_tool_context(current_messages=None) + + result = await _handle_create_observations( + ctx, + { + "observations": [ + { + "content": "Conflicting claims with one invented citation", + "level": "contradiction", + "source_ids": [ + real_id, + "fabricated-id-that-does-not-exist", + ], + "sources": [ + "The user said they prefer tea", + "The user said they prefer coffee", + ], + }, + ] + }, + ) + + assert "Created 0 observations" in result + assert "Failed 1" in result + assert "requires at least 2 real source(s)" in str(result) + + stmt = select(models.Document).where( + models.Document.content == "Conflicting claims with one invented citation" + ) + doc = (await db_session.execute(stmt)).scalar_one_or_none() + assert doc is None + async def test_mixed_batch_keeps_grounded_rejects_ungrounded( self, db_session: AsyncSession,