From b5bcf64741cfbf8dda823bcf9a8488bbe0f463ec Mon Sep 17 00:00:00 2001 From: Wenchao An Date: Wed, 14 Oct 2026 06:05:16 +0800 Subject: [PATCH] fix(extensions): harden team history cursors and cancelled results (#6832) * fix(extensions): harden team history cursors and cancelled results * refactor(extensions): deduplicate team history cursor anchors --- backend/tests/test_agent_teams_plugin.py | 84 +++++++++++++++++++ .../deerflow-extension-agent-teams/AGENTS.md | 2 + .../deerflow-extension-agent-teams/README.md | 2 + .../deerflow_extension_agent_teams/history.py | 5 +- 4 files changed, 91 insertions(+), 2 deletions(-) diff --git a/backend/tests/test_agent_teams_plugin.py b/backend/tests/test_agent_teams_plugin.py index 94c14629b5..dfecb44694 100644 --- a/backend/tests/test_agent_teams_plugin.py +++ b/backend/tests/test_agent_teams_plugin.py @@ -1922,3 +1922,87 @@ async def test_history_result_pages_and_reads_bound_the_complete_escaped_json(pl assert first_cursor with pytest.raises(ValueError, match="cursor|Cursor"): await service.read_history({"team_id": team["id"], "cursor": first_cursor}, caller) + + +@pytest.mark.asyncio +async def test_history_result_cursors_support_non_hex_job_ids_and_expiry(plugin, monkeypatch): + from deerflow_extension_agent_teams import service as module + + _, actions, service = plugin + runs = Runs() + team = await create(actions, runs) + member = team["members"][0] + caller = context(runs, thread=member["thread_id"]) + identity = module.identity + monkeypatch.setattr(module, "identity", lambda *parts: "request-" + identity(*parts)) + jobs = [] + for index in range(2): + sent = await scheduling_send(actions, runs, "alice", team, member, str(index)) + stored = await service.db("get", "alice", team["id"]) + await service.finish("alice", stored, stored["jobs"][-1], "completed", "Retained evidence") + jobs.append(sent) + payload = {"team_id": team["id"], "limit": 1} + first = await service.search_results(payload, caller) + assert first["items"][0]["job_id"] == jobs[1]["id"] + await scheduling_send(actions, runs, "alice", team, member, "later") + stored = await service.db("get", "alice", team["id"]) + await service.finish("alice", stored, stored["jobs"][-1], "completed", "New evidence") + continuation = {**payload, "cursor": first["next_cursor"]} + second = await service.search_results(continuation, caller) + assert [item["job_id"] for item in second["items"]] == [jobs[0]["id"]] + assert second["next_cursor"] is None and not second["cursor_expired"] + await service.db("change", "alice", team["id"], lambda t: t.update(jobs=[job for job in t["jobs"] if job["id"] != jobs[1]["id"]])) + expired = await service.search_results(continuation, caller) + assert expired["cursor_expired"] and expired["items"] == [] and expired["next_cursor"] is None + + +@pytest.mark.asyncio +async def test_history_result_cursors_issued_before_anchor_hashing_expire(plugin): + _, actions, service = plugin + runs = Runs() + team = await create(actions, runs) + member = team["members"][0] + caller = context(runs, thread=member["thread_id"]) + for index in range(2): + await scheduling_send(actions, runs, "alice", team, member, str(index)) + stored = await service.db("get", "alice", team["id"]) + await service.finish("alice", stored, stored["jobs"][-1], "completed", "Retained evidence") + payload = {"team_id": team["id"], "limit": 1} + first = await service.search_results(payload, caller) + fingerprint = first["next_cursor"].split(":")[0] + legacy_cursor = f"{fingerprint}:{first['items'][0]['job_id']}" + expired = await service.search_results({**payload, "cursor": legacy_cursor}, caller) + assert expired["cursor_expired"] and expired["items"] == [] and expired["next_cursor"] is None + + +@pytest.mark.asyncio +async def test_history_cancelled_results_are_unavailable_before_and_after_admission(plugin): + _, actions, service = plugin + runs = Runs() + team = await create(actions, runs) + member = team["members"][0] + caller = context(runs, thread=member["thread_id"]) + unavailable = {"available": False, "text": None, "truncated": None, "reason": "cancelled"} + for admitted in (False, True): + sent = await scheduling_send(actions, runs, "alice", team, member, str(admitted)) + if admitted: + await service.tick() + assert sent["id"] in runs.runs + await actions["cancel"]({"team_id": team["id"], "job_id": sent["id"]}, context(runs)) + if admitted: + await service.tick() + await service.tick() + result = await service.read_result({"team_id": team["id"], "job_id": sent["id"]}, caller) + assert result["status"] == "cancelled" and result["result"] == unavailable + found = await service.search_results({"team_id": team["id"]}, caller) + assert len(found["items"]) == 2 and all(item["result"] == unavailable for item in found["items"]) + notice = "Request cancelled; checkpoint retained" + history = await service.read_history({"team_id": team["id"], "job_id": sent["id"]}, caller) + assert any(item["kind"] == "cancelled" and item["text"] == notice for item in history["items"]) + stored = await service.db("get", "alice", team["id"]) + assert stored["jobs"][-1]["result"] == notice + await service.db("change", "alice", team["id"], lambda t: t["jobs"][-1].pop("result")) + legacy = await service.read_result({"team_id": team["id"], "job_id": sent["id"]}, caller) + assert legacy["result"] == unavailable + filtered = await service.search_results({"team_id": team["id"], "query": notice}, caller) + assert filtered["items"] == [] diff --git a/examples/deerflow-extension-agent-teams/AGENTS.md b/examples/deerflow-extension-agent-teams/AGENTS.md index 55155ed1f7..e96745315d 100644 --- a/examples/deerflow-extension-agent-teams/AGENTS.md +++ b/examples/deerflow-extension-agent-teams/AGENTS.md @@ -17,6 +17,8 @@ to record IDs, not shifting offsets. Report expired anchors instead of silently resetting pagination. Bound the complete serialized JSON response, including metadata and escaping, to 24,000 UTF-8 bytes. Preserve result-truncation provenance; legacy unknown flags stay null. Queries never reconnect or admit Agent runs. +Hash cursor anchors independently of job-ID syntax. Cancelled jobs have no +available answer; their cancellation notices remain shared history events. Persist run input and idempotency keys before admission. Serialize by thread and record a terminal result and its follow-up receipt in one SQLite transaction. diff --git a/examples/deerflow-extension-agent-teams/README.md b/examples/deerflow-extension-agent-teams/README.md index ed2652780c..d8534c0c36 100644 --- a/examples/deerflow-extension-agent-teams/README.md +++ b/examples/deerflow-extension-agent-teams/README.md @@ -118,6 +118,8 @@ whether the original collected answer was truncated; legacy answers report `result.truncated=null` when that provenance is unknown. `result.available=false` distinguishes unfinished work, cancellation without an answer, and a legacy answer no longer retained. Inspect the source conversation for further evidence. +Cancelled requests have no available answer at either cancellation timing; +their cancellation notices remain in shared history. These tools expose retained task data, not hidden reasoning, execution inputs, approval responses, attachment bytes or an unbounded transcript archive. diff --git a/examples/deerflow-extension-agent-teams/deerflow_extension_agent_teams/history.py b/examples/deerflow-extension-agent-teams/deerflow_extension_agent_teams/history.py index 4864a78674..d697853b64 100644 --- a/examples/deerflow-extension-agent-teams/deerflow_extension_agent_teams/history.py +++ b/examples/deerflow-extension-agent-teams/deerflow_extension_agent_teams/history.py @@ -33,6 +33,8 @@ def options(team, payload, validate_text): def stored_result(team, job): + if job["status"] == "cancelled": + return None if "result" in job: return job["result"] return next((message["text"] for message in reversed(team["messages"]) if message["job_id"] == job["id"] and message["kind"] in TERMINAL), None) @@ -84,9 +86,9 @@ def page(team, payload, validate_text, *, results=False): searchable = record["text"] + ("\n" + (stored_result(team, record) or "") if results else "") if query and query not in searchable.casefold(): continue + anchor = token(record_job_id, record["kind"]) if results: item = job_record(team, record) - anchor = record["id"] else: content, cut = excerpt(record["text"], 6000) job = jobs.get(record_job_id) @@ -100,7 +102,6 @@ def page(team, payload, validate_text, *, results=False): "truncated": True if cut else (job.get("result_truncated") if job else None) if record["kind"] in TERMINAL else False, "conversation": {"thread_id": job["thread_id"] if job else record.get("source_thread"), "run_id": record.get("run_id")}, } - anchor = token(record_job_id, record["kind"]) rows.append((anchor, item)) response = { "team_id": team["id"],