diff --git a/src/dreamer/specialists.py b/src/dreamer/specialists.py index 28ff4577..4c312889 100644 --- a/src/dreamer/specialists.py +++ b/src/dreamer/specialists.py @@ -607,10 +607,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, @@ -741,8 +742,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 de07e38f..7f50f4f3 100644 --- a/src/utils/agent_tools.py +++ b/src/utils/agent_tools.py @@ -120,7 +120,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": { @@ -237,7 +241,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", @@ -263,7 +267,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", @@ -964,6 +968,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, @@ -1010,6 +1096,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( @@ -1043,7 +1144,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: @@ -2486,7 +2586,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 346296ef..8f7a9f06 100644 --- a/tests/utils/test_agent_tools.py +++ b/tests/utils/test_agent_tools.py @@ -218,9 +218,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( @@ -229,7 +232,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", @@ -249,7 +252,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, @@ -284,10 +287,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( @@ -296,7 +301,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", @@ -313,7 +321,160 @@ 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_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, + 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]