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

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.

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)}