fix(task-continuity): count distinct archived search sources (#7956)

* fix(task-continuity): count distinct archived search sources

* docs(task-continuity): explain archive search ranking trade-off

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -2719,8 +2719,9 @@ and SDK retries can add calls. See the
 
 With `task_continuity.enabled`, `history_search` searches the current task's active
 and compacted history. Its optional `role` accepts `user`, `assistant`, or `tool`
-and filters before the eight-result limit; omitting it or passing `null` preserves
-search across all roles.
+and filters before limiting results to eight distinct sources. Omitting `role` or
+passing `null` preserves search across all roles. Repeated archives of one source
+share one result slot.
 Search excerpts contain at most 600 characters around the earliest locatable
 matching occurrence that fits and include `excerpt_start` / `excerpt_end`, a
 zero-based, half-open character range

backend/packages/harness/deerflow/agents/task_continuity/AGENTS.md

@@ -17,8 +17,9 @@ unused slots are recalculated in the next batch.
 
 `history_search` accepts optional `role=user|assistant|tool`, mapped at the tool
 boundary to stored roles `human|ai|tool`; omission/null keeps all roles. Filter
-archived JSON payloads before SQLite FTS `LIMIT 8`, and active messages before
-the merged result limit. Preserve ranking, source IDs, returned roles, checkpoint
+archived JSON payloads before ranked cursor iteration. Stop at eight distinct
+source IDs, then merge active matches within that limit. Preserve ranking,
+source IDs, returned roles, checkpoint
 reachability, and user/thread isolation. Do not add a role parameter to `history_read`.
 Sync and async entry points share the implementation; async uses `run_file_io`.
 Regression coverage lives in `backend/tests/test_task_continuity.py`.

backend/packages/harness/deerflow/agents/task_continuity/archive.py

@@ -215,11 +215,16 @@ def lookup(state: dict, runtime, *, query: str | None = None, source_id: str | N
                 else:
                     match = " OR ".join('"' + term.replace('"', '""') + '"' for term in keywords)
                     role_filter = " AND json_extract(payload, '$.role') = ?" if role is not None else ""
-                    parameters = [match, *batches, *([role] if role is not None else []), 8]
-                    rows = db.execute(f"SELECT payload FROM sources WHERE sources MATCH ? AND batch IN ({placeholders}){role_filter} ORDER BY rank LIMIT ?", parameters)
+                    parameters = [match, *batches, *([role] if role is not None else [])]
+                    # Rank all matches in the bounded archive. A SQL row limit can hide distinct sources behind duplicates.
+                    rows = db.execute(f"SELECT payload FROM sources WHERE sources MATCH ? AND batch IN ({placeholders}){role_filter} ORDER BY rank", parameters)
                 for (payload,) in rows:
                     row = json.loads(payload)
-                    found[row["id"]] = row
+                    # Repeated archives share one result slot; keep FTS order
+                    # and stop reading once eight distinct sources are selected.
+                    found.setdefault(row["id"], row)
+                    if len(found) == 8:
+                        break
                 if excerpts and query is not None and found:
                     # A source may be archived at different lengths; select its first stored version, as history_read does.
                     ids = list(found)

backend/tests/test_task_continuity.py

@@ -1002,3 +1002,94 @@ async def test_active_search_excerpt_keeps_overlapping_fitting_occurrence(role_r
     read_arguments = {"runtime": role_runtime, "source_id": row["id"], "offset": row["excerpt_start"]}
     page = json.loads(await history_read.ainvoke(read_arguments) if async_mode else history_read.invoke(read_arguments))
     assert page["text"].startswith(row["excerpt"])
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("async_mode", [False, True])
+@pytest.mark.parametrize("role", [None, "user"])
+@pytest.mark.parametrize("repeated", [False, True])
+async def test_archived_search_counts_distinct_sources(role_runtime, async_mode, role, repeated):
+    import json
+
+    config = TaskContinuityConfig(enabled=True)
+    shared = HumanMessage(content="needle", id="shared")
+    other = HumanMessage(content="needle distinct evidence important constraint keep backup", id="other")
+    for i in range(8 if repeated else 1):
+        marker = HumanMessage(content=f"unrelated marker {i}", id=f"marker-{i}")
+        role_runtime.state["task_history"] = archive.capture(role_runtime.state, role_runtime, [shared, marker], config)
+    role_runtime.state["task_history"] = archive.capture(role_runtime.state, role_runtime, [other], config)
+    expected_ids = {row["id"] for row in archive.records([shared, other])}
+    for source_id in expected_ids:
+        assert json.loads(history_read.invoke({"runtime": role_runtime, "source_id": source_id}))["id"] == source_id
+
+    args = {"runtime": role_runtime, "query": "needle", "role": role}
+    result = json.loads(await history_search.ainvoke(args) if async_mode else history_search.invoke(args))
+
+    assert result["status"] == "available"
+    assert {row["id"] for row in result["results"]} == expected_ids
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("async_mode", [False, True])
+async def test_archived_search_retains_ranked_eight_source_limit(role_runtime, async_mode):
+    import json
+
+    config = TaskContinuityConfig(enabled=True)
+    shared = HumanMessage(content="needle", id="shared")
+    for i in range(8):
+        role_runtime.state["task_history"] = archive.capture(role_runtime.state, role_runtime, [shared, HumanMessage(content=f"unrelated marker {i}", id=f"marker-{i}")], config)
+    words = ["alpha", "bravo", "charlie", "delta", "echo", "foxtrot", "golf", "hotel", "india"]
+    # Equal query frequency with strictly increasing indexed document lengths
+    # gives a deterministic FTS ranking independent of insertion order.
+    others = [HumanMessage(content="needle " + " ".join(words[: i + 1]), id=f"other-{i}") for i in range(9)]
+    role_runtime.state["task_history"] = archive.capture(role_runtime.state, role_runtime, list(reversed(others)), config)
+
+    args = {"runtime": role_runtime, "query": "needle"}
+    rows = json.loads(await history_search.ainvoke(args) if async_mode else history_search.invoke(args))["results"]
+
+    assert [row["message_id"] for row in rows] == ["shared", *[message.id for message in others[:7]]]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("async_mode", [False, True])
+async def test_cumulative_graph_history_does_not_hide_distinct_archived_evidence(role_runtime, async_mode):
+    import json
+
+    graph = create_agent(model=StaticModel(), middleware=[compacting(TaskContinuityConfig(enabled=True))], state_schema=ThreadState, checkpointer=InMemorySaver())
+    config = {"configurable": {"thread_id": role_runtime.context["thread_id"]}}
+    shared = HumanMessage(content="needle", id="shared")
+    conversation = [shared, AIMessage(content="Acknowledged.", id="initial-reply")]
+    for i in range(10):
+        conversation.append(HumanMessage(content=f"Continue step {i}", id=f"step-{i}"))
+        args = ({"messages": list(conversation)}, config)
+        state = await graph.ainvoke(*args, context=role_runtime.context) if async_mode else graph.invoke(*args, context=role_runtime.context)
+        conversation.append(state["messages"][-1])
+    other = HumanMessage(content="needle distinct evidence important constraint keep backup", id="other")
+    conversation.extend([other, AIMessage(content="Recorded constraint.", id="other-reply"), HumanMessage(content="Proceed", id="final-user")])
+    args = ({"messages": conversation}, config)
+    state = await graph.ainvoke(*args, context=role_runtime.context) if async_mode else graph.invoke(*args, context=role_runtime.context)
+    role_runtime.state = state
+    assert not any(message.id in {"shared", "other"} for message in state["messages"])
+
+    search_args = {"runtime": role_runtime, "query": "needle"}
+    result = json.loads(await history_search.ainvoke(search_args) if async_mode else history_search.invoke(search_args))
+
+    assert result["status"] == "available"
+    assert {row["message_id"] for row in result["results"]} == {"shared", "other"}
+
+
+def test_repeated_archive_sources_keep_canonical_read_version(role_runtime):
+    import json
+
+    text = "needle " + "padding " * 200
+    message = HumanMessage(content=text, id="repeated-version")
+    for cap in (1000, 2000):
+        config = TaskContinuityConfig(enabled=True, max_record_chars=cap)
+        role_runtime.state["task_history"] = archive.capture(role_runtime.state, role_runtime, [message], config)
+
+    row = json.loads(history_search.invoke({"runtime": role_runtime, "query": "needle"}))["results"][0]
+    source = json.loads(history_read.invoke({"runtime": role_runtime, "source_id": row["id"]}))
+
+    assert row["truncated"] is source["truncated"] is True
+    assert source["text"] == text[:1000]
+    assert row["excerpt"] == source["text"][row["excerpt_start"] : row["excerpt_end"]]

docs/task-continuity.md

@@ -41,6 +41,7 @@ The standard lead-agent builders (including custom-agent bootstrap) and
 - `history_search`: keyword search over the current messages and compacted source
   batches reachable from the current checkpoint. English words and Chinese
   character bigrams are supported. Returns up to eight excerpts of at most 600 characters.
+  Repeated archives of one source share one result slot, preserving FTS ranking.
   The optional `role="user"`, `"assistant"`, or `"tool"` filters by message author
   type; omission or `null` preserves search across all roles. These values map to
   internal `human`, `ai`, and `tool` roles; returned role values stay unchanged.