From a6413c75ad67b7804b5bcba4f0e4a87b4cc96f6d Mon Sep 17 00:00:00 2001 From: Gumayusy <2732381414@qq.com> Date: Thu, 22 Oct 2026 11:45:58 +0800 Subject: [PATCH] fix: include active messages in history search limit (#7169) * fix: include active messages in history search limit * fix: deduplicate active and archived source reads * fix: deduplicate active history matches --- .../agents/task_continuity/archive.py | 23 +++++--- backend/tests/test_task_continuity.py | 55 +++++++++++++++++++ 2 files changed, 71 insertions(+), 7 deletions(-) diff --git a/backend/packages/harness/deerflow/agents/task_continuity/archive.py b/backend/packages/harness/deerflow/agents/task_continuity/archive.py index 5a3612cca9..6e85ff3ce9 100644 --- a/backend/packages/harness/deerflow/agents/task_continuity/archive.py +++ b/backend/packages/harness/deerflow/agents/task_continuity/archive.py @@ -199,6 +199,10 @@ def lookup(state: dict, runtime, *, query: str | None = None, source_id: str | N found = {} history = normalize_task_history(state.get("task_history")) status = "unavailable" if history.get("status") == "unavailable" else "available" + active_matches = [] + for row in active: + if (source_id and row["id"] == source_id) or (query is not None and (role is None or row["role"] == role) and any(t in row["text"].casefold() for t in keywords)): + active_matches.append(row) if batches: try: with closing(sqlite3.connect(path.as_uri() + "?mode=ro", uri=True, timeout=2)) as db: @@ -211,8 +215,8 @@ 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 [])] - rows = db.execute(f"SELECT payload FROM sources WHERE sources MATCH ? AND batch IN ({placeholders}){role_filter} ORDER BY rank LIMIT 8", parameters) + 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) for (payload,) in rows: row = json.loads(payload) found[row["id"]] = row @@ -229,12 +233,17 @@ def lookup(state: dict, runtime, *, query: str | None = None, source_id: str | N status = "unavailable" elif history.get("scope") is not None and history["scope"] != owner: status = "scope_unavailable" - for row in active: - if (source_id and row["id"] == source_id) or (query is not None and (role is None or row["role"] == role) and any(t in row["text"].casefold() for t in keywords)): - found[row["id"]] = row - results = list(found.values())[:8] + archived_matches = list(found.values()) + archived_ids = {item["id"] for item in archived_matches} + active_by_id = {row["id"]: row for row in active} + archived_matches = [active_by_id.get(row["id"], row) for row in archived_matches] + active_only_by_id = {} + for row in active_matches: + if row["id"] not in archived_ids: + active_only_by_id.setdefault(row["id"], row) + active_only = list(active_only_by_id.values()) + results = archived_matches[: max(0, 8 - len(active_only))] + active_only[:8] if excerpts and query is not None: - active_by_id = {row["id"]: row for row in active} excerpts_result = [] for row in results: # Active text takes precedence when reading by ID, even if only archived FTS matched the source. diff --git a/backend/tests/test_task_continuity.py b/backend/tests/test_task_continuity.py index b0bd0ca448..5d815fbe13 100644 --- a/backend/tests/test_task_continuity.py +++ b/backend/tests/test_task_continuity.py @@ -91,6 +91,61 @@ def test_old_checkpoint_cannot_see_future_batch(scoped): assert archive.lookup(old, scoped, query="Citrine")["results"] +def test_active_search_results_participate_in_archive_result_limit(scoped): + archived = [HumanMessage(content=f"needle archived {i}", id=f"archived-{i}") for i in range(8)] + history = archive.capture({}, scoped, archived, TaskContinuityConfig(enabled=True)) + active = HumanMessage(content="needle current", id="current") + + result = archive.lookup({"task_history": history, "messages": [active]}, scoped, query="needle") + + assert len(result["results"]) == 8 + assert any(row["message_id"] == "current" for row in result["results"]) + assert sum(row["role"] == "human" for row in result["results"]) == 8 + + +def test_active_search_results_deduplicate_against_archived_matches(scoped): + archived = [HumanMessage(content=f"needle archived {i}", id=f"archived-{i}") for i in range(8)] + history = archive.capture({}, scoped, archived, TaskContinuityConfig(enabled=True)) + active = [archived[0], HumanMessage(content="needle current", id="current")] + + result = archive.lookup({"task_history": history, "messages": active}, scoped, query="needle") + + assert len(result["results"]) == 8 + message_ids = [row["message_id"] for row in result["results"]] + assert message_ids.count("archived-0") == 1 + assert "current" in message_ids + + +def test_read_by_id_deduplicates_active_and_archived_source(scoped): + message = HumanMessage(content="needle archived and active", id="shared") + history = archive.capture({}, scoped, [message], TaskContinuityConfig(enabled=True)) + source_id = archive.records([message])[0]["id"] + + result = archive.lookup( + {"task_history": history, "messages": [message]}, + scoped, + source_id=source_id, + ) + + assert [row["id"] for row in result["results"]] == [source_id] + + +def test_active_only_matches_deduplicate_before_result_limit(scoped): + archived = [HumanMessage(content="needle archived", id="archived")] + history = archive.capture({}, scoped, archived, TaskContinuityConfig(enabled=True)) + duplicate_a = HumanMessage(content="needle duplicate") + duplicate_b = HumanMessage(content="needle duplicate") + + result = archive.lookup( + {"task_history": history, "messages": [duplicate_a, duplicate_b]}, + scoped, + query="needle", + ) + + assert len(result["results"]) == 2 + assert {row["message_id"] for row in result["results"]} == {None, "archived"} + + def test_retention_is_explicit_and_duplicate_capture_is_idempotent(scoped): config = TaskContinuityConfig(enabled=True, max_batches=1) old = {"task_history": archive.capture({}, scoped, conversation(), config)}