From 24884e8120d13bb28f0a71de7f1f0ddcc102cf1d Mon Sep 17 00:00:00 2001 From: lilei <93921041+xihongshichaojidan8@users.noreply.github.com> Date: Mon, 9 Nov 2026 09:07:20 +0800 Subject: [PATCH] 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 --- README.md | 5 +- .../deerflow/agents/task_continuity/AGENTS.md | 5 +- .../agents/task_continuity/archive.py | 11 ++- backend/tests/test_task_continuity.py | 91 +++++++++++++++++++ docs/task-continuity.md | 1 + 5 files changed, 106 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 0c1220486e..272675321a 100644 --- a/README.md +++ b/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 diff --git a/backend/packages/harness/deerflow/agents/task_continuity/AGENTS.md b/backend/packages/harness/deerflow/agents/task_continuity/AGENTS.md index f604b3425e..1895397316 100644 --- a/backend/packages/harness/deerflow/agents/task_continuity/AGENTS.md +++ b/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`. diff --git a/backend/packages/harness/deerflow/agents/task_continuity/archive.py b/backend/packages/harness/deerflow/agents/task_continuity/archive.py index 6e85ff3ce9..82b9444dda 100644 --- a/backend/packages/harness/deerflow/agents/task_continuity/archive.py +++ b/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) diff --git a/backend/tests/test_task_continuity.py b/backend/tests/test_task_continuity.py index 5d815fbe13..96ccf6c0c3 100644 --- a/backend/tests/test_task_continuity.py +++ b/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"]] diff --git a/docs/task-continuity.md b/docs/task-continuity.md index 869a31070c..4617e28740 100644 --- a/docs/task-continuity.md +++ b/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.