tests/test_chunking.py

@@ -148,23 +148,39 @@ async def test_search_collapses_chunk_rows_to_best_chunk_per_note():
     from scribe.services import embeddings as emb
 
     note_a, note_b = MagicMock(id=1), MagicMock(id=2)
-    rows = [(note_a, 0.10), (note_b, 0.20), (note_a, 0.25), (note_a, 0.30)]
+    # Rows are (Note, distance, chunk_index, chunk_text) — the chunk columns
+    # ride along so the collapse can report WHICH passage won (#4243).
+    rows = [
+        (note_a, 0.10, 3, "the passage that actually matched"),
+        (note_b, 0.20, 0, "b's best"),
+        (note_a, 0.25, 7, "a worse chunk of a"),
+        (note_a, 0.30, 1, "a worse chunk of a"),
+    ]
     result = MagicMock()
     result.all.return_value = rows
     session, ctx = _session_ctx()
     session.execute = AsyncMock(return_value=result)
 
+    report: dict = {}
     with (
         patch.object(emb, "async_session", return_value=ctx),
         patch.object(emb, "get_embedding", AsyncMock(return_value=[0.0] * 384)),
     ):
         out = await emb.semantic_search_notes(
-            1, "a query", limit=8, demote_superseded=False
+            1, "a query", limit=8, demote_superseded=False, report=report
         )
 
     assert [note.id for _s, note in out] == [1, 2]
     assert out[0][0] == 1.0 - 0.10  # the BEST chunk's score, not a later one
 
+    # And the winning chunk is reported, not merely used for scoring. Without
+    # this a caller can only preview the head of the body — a span this query
+    # has already ranked lower than the one that won (#4243).
+    assert report["best_chunk"][1] == {
+        "index": 3, "text": "the passage that actually matched",
+    }
+    assert report["best_chunk"][2]["index"] == 0
+
 
 # --- the write path: one row per chunk (#280 step 3) -------------------------
 

tests/test_task_work_log_surface.py

@@ -322,9 +322,12 @@ async def test_logs_for_task_is_not_filtered_by_who_wrote_the_entry():
          patch("scribe.services.task_logs.async_session",
                MagicMock(return_value=session)):
         await logs_for_task(7, 42)
-    stmt = str(session.execute.await_args.args[0])
-    assert "task_logs.task_id" in stmt
-    assert "task_logs.user_id" not in stmt
+    # The WHERE clause specifically — `user_id` is a selected COLUMN of the
+    # row and appears in every SELECT, so asserting against the whole compiled
+    # statement tests the projection rather than the scoping.
+    where = str(session.execute.await_args.args[0].whereclause)
+    assert "task_logs.task_id" in where
+    assert "task_logs.user_id" not in where
 
 
 @pytest.mark.asyncio