fix(subagents): verify the returned row position in read_batch_item (#8191)

read_batch_item trusted list_items(offset=position) to land on position.
That only holds while item rows are dense: OFFSET-based paging skips
surviving rows over any gap (e.g. a future retention delete), so the
reader would serve the next item's report attributed to the requested
item key, and the tool layer would project next_position from the wrong
base. The browser projection (read_browser_batch_item) already verifies
items[0].position; align the model-facing reader with it and add the
same negative-position guard. Reading past a gap now returns not_found
instead of a shifted row, matching the browser path's semantics.

backend/packages/harness/deerflow/subagents/batch_service.py

@@ -365,11 +365,18 @@ class SubagentBatchService:
 
     async def read_batch_item(self, *, batch_id: str, user_id: str, thread_id: str, position: int) -> dict[str, Any] | None:
         """Read one current-thread result; never expose worker execution context."""
+        if position < 0:
+            return None
         batch = await self._repository.get_batch(batch_id, user_id=user_id)
         if batch is None or batch["thread_id"] != thread_id:
             return None
         items = await self._repository.list_items(batch_id, user_id=user_id, offset=position, limit=1, include_result=True)
-        if not items:
+        # ``list_items(offset=position)`` only lands on ``position`` while item
+        # rows are dense; after any retention gap it returns the next surviving
+        # row. Serving it would attribute one item's report to another item key
+        # (``next_position`` advances from the wrong base too). Verify the row's
+        # position, same as the browser projection below.
+        if not items or items[0].get("position") != position:
             return None
         fields = ("id", "item_key", "position", "status", "attempt", "result", "result_truncated", "error", "stop_reason", "acceptance_criteria", "acceptance_verdict")
         return {key: items[0].get(key) for key in fields}

backend/tests/test_batch_result_reader.py

@@ -433,3 +433,31 @@ async def test_runtime_identity_is_not_a_model_argument(env):
     trusted.server_info = SimpleNamespace(user=SimpleNamespace(identity="user-1"))
     batch = await submit(env)
     assert (await read(env.service, batch, runtime=trusted))["status"] == "ok"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("env", ["sqlite", "postgres"], indirect=True)
+async def test_read_batch_item_never_serves_a_gapped_position(env):
+    """``read_batch_item`` must verify the returned row's position like its
+    browser sibling: ``list_items(offset=position)`` only equals position
+    while rows are dense, so a retention gap must yield ``None`` — not the
+    next item's report attributed to the wrong item key."""
+    from sqlalchemy import delete as sa_delete
+
+    from deerflow.persistence.subagent_batches.model import SubagentBatchItemRow
+
+    batch = await submit(env)
+    await complete(env, batch)
+    async with get_session_factory()() as session:
+        await session.execute(sa_delete(SubagentBatchItemRow).where(SubagentBatchItemRow.batch_id == batch, SubagentBatchItemRow.position == 0))
+        await session.commit()
+
+    # Without the position check this returned the surviving item's report
+    # attributed to the deleted item's position.
+    assert await env.service.read_batch_item(batch_id=batch, user_id="user-1", thread_id="thread-1", position=0) is None
+    # The browser projection already defends against gaps; keep them aligned.
+    assert await env.service.read_browser_batch_item(batch_id=batch, user_id="user-1", thread_id="thread-1", position=0) is None
+    # Positions past the gap never resolve to a shifted row either (OFFSET
+    # semantics): the survivor is only reachable from a fresh dense batch.
+    assert await env.service.read_batch_item(batch_id=batch, user_id="user-1", thread_id="thread-1", position=1) is None
+    assert await env.service.read_batch_item(batch_id=batch, user_id="user-1", thread_id="thread-1", position=2) is None