From 4d9f1d7aa03d98531e986dbfca1c79532deb7655 Mon Sep 17 00:00:00 2001 From: Daoyuan Li <94409450+DaoyuanLi2816@users.noreply.github.com> Date: Mon, 5 Oct 2026 01:10:07 -0700 Subject: [PATCH] fix(extensions): exclude unadmitted team waits from active budget (#6298) * fix(extensions): exclude unadmitted team waits from active budget * docs(extensions): explain frozen team input reservation --- backend/tests/test_agent_teams_plugin.py | 71 +++++++++++++++++++ .../deerflow-extension-agent-teams/AGENTS.md | 3 + .../deerflow-extension-agent-teams/README.md | 2 + .../deerflow_extension_agent_teams/service.py | 9 ++- 4 files changed, 84 insertions(+), 1 deletion(-) diff --git a/backend/tests/test_agent_teams_plugin.py b/backend/tests/test_agent_teams_plugin.py index 95206f6185..0f790e9d6d 100644 --- a/backend/tests/test_agent_teams_plugin.py +++ b/backend/tests/test_agent_teams_plugin.py @@ -258,6 +258,77 @@ async def test_ambiguous_admission_retries_same_key_and_frozen_input(plugin, mon assert len(runs.starts) == 1 +async def queue_active_budget_jobs(actions, runs): + jobs = [] + for index in range(5): + owner = "bob" if index == 4 else "alice" + team = await actions["create"]( + {"request_id": f"team-{index}", "name": "Release", "goal": "Check readiness", "members": [{"name": "Research", "agent": "researcher"}, {"name": "Review", "agent": "reviewer"}]}, + context(runs, owner), + ) + for member in team["members"]: + if len(jobs) == 9: + break + payload = {"team_id": team["id"], "member_id": member["id"], "text": "Check", "request_id": f"job-{len(jobs)}"} + sent = await actions["send"](payload, context(runs, owner)) + jobs.append({**sent, "owner": owner, "team_id": team["id"], "member_id": member["id"], "thread_id": member["thread_id"]}) + return jobs + + +@pytest.mark.asyncio +@pytest.mark.parametrize("wait_kind", ["interrupt", "clarification"]) +async def test_unadmitted_conversation_waits_do_not_consume_other_teams_active_budget(plugin, wait_kind): + from deerflow.agents.middlewares.clarification_middleware import ClarificationMiddleware + + _, actions, service = plugin + runs = Runs() + jobs = await queue_active_budget_jobs(actions, runs) + for job in jobs[:8]: + state = runs.threads[job["thread_id"]] + if wait_kind == "interrupt": + state.update(next=["tools"], interrupts=[{"value": "Approve?"}]) + else: + request = SimpleNamespace(tool_call={"name": "ask_clarification", "id": "outside", "args": {"question": "Which environment?", "clarification_type": "missing_info"}}, runtime=None) + command = ClarificationMiddleware().wrap_tool_call(request, lambda _: pytest.fail("tool handler should be intercepted")) + state["values"]["messages"].extend(m.model_dump() for m in command.update["messages"]) + # A later request on the same waiting conversation must still remain queued. + blocked = jobs[0] + await actions["send"]({"team_id": blocked["team_id"], "member_id": blocked["member_id"], "text": "Later", "request_id": "later"}, context(runs)) + for _ in range(2): + await service.tick() + assert [thread for thread, _ in runs.starts] == [jobs[-1]["thread_id"]] + for job in jobs[:8]: + stored = await service.db("get", job["owner"], job["team_id"]) + assert all(item["status"] == "queued" and item["input"] is None for item in stored["jobs"]) + healthy = await actions["get"]({"team_id": jobs[-1]["team_id"]}, context(runs, "bob")) + assert healthy["jobs"][0]["status"] == "running" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("lost_ack", [False, True]) +async def test_active_budget_retains_real_and_ambiguous_admissions(plugin, monkeypatch, lost_ack): + _, actions, service = plugin + runs = Runs() + jobs = await queue_active_budget_jobs(actions, runs) + if lost_ack: + start = runs.start + + async def uncertain(**kwargs): + await start(**kwargs) + raise TimeoutError("admitted but acknowledgement lost") + + monkeypatch.setattr(runs, "start", uncertain) + for _ in range(2): + await service.tick() + assert len(runs.starts) == 8 + assert jobs[-1]["id"] not in runs.runs + for job in jobs[:8]: + stored = await service.db("get", job["owner"], job["team_id"]) + assert all(item["input"] is not None for item in stored["jobs"]) + healthy = await service.db("get", "bob", jobs[-1]["team_id"]) + assert healthy["jobs"][0]["status"] == "queued" and healthy["jobs"][0]["input"] is None + + @pytest.mark.asyncio async def test_cancel_stops_a_running_request_and_pending_delete_is_rejected(plugin): _, actions, service = plugin diff --git a/examples/deerflow-extension-agent-teams/AGENTS.md b/examples/deerflow-extension-agent-teams/AGENTS.md index cb0bbda2cd..48b3419d80 100644 --- a/examples/deerflow-extension-agent-teams/AGENTS.md +++ b/examples/deerflow-extension-agent-teams/AGENTS.md @@ -15,6 +15,9 @@ Unknown admission outcomes must retain ownership of the pending job. Never automatically approve interruptions, infer success from a terminal run alone, or serialize a host capability. Only an authenticated action or native mention rebinds after restart. +Release a new job's temporary active-budget reservation only if its input remains +unfrozen after advancement; conversation waits must not block unrelated teams. +Keep frozen inputs charged even when admission acknowledgement is lost. Reconnect ensures every member thread before persisting readiness, serialized with deletion and task updates. Count only requests toward the 100-request cap; diff --git a/examples/deerflow-extension-agent-teams/README.md b/examples/deerflow-extension-agent-teams/README.md index f4d410f999..fae4c6691e 100644 --- a/examples/deerflow-extension-agent-teams/README.md +++ b/examples/deerflow-extension-agent-teams/README.md @@ -99,6 +99,8 @@ own teams. Sharing a host conversation does **not** share its team's private dat fairness and finer-grained locking are outside this example's current scope. Each conversation is serialized, including result receipts. A busy host conversation stays queued without cancelling its run. + Queued jobs blocked by an unanswered question or interrupt do not consume the + active-job budget before admission; an unknown admission outcome still does. - Each owner can have 20 teams. A team accepts at most 100 requests, with separate space for up to 100 result receipts (200 total job entries); a handoff chain accepts at most 12 requests. Start a new user request for another chain, or a diff --git a/examples/deerflow-extension-agent-teams/deerflow_extension_agent_teams/service.py b/examples/deerflow-extension-agent-teams/deerflow_extension_agent_teams/service.py index ccf99eb3ac..2b13411ca2 100644 --- a/examples/deerflow-extension-agent-teams/deerflow_extension_agent_teams/service.py +++ b/examples/deerflow-extension-agent-teams/deerflow_extension_agent_teams/service.py @@ -392,7 +392,8 @@ class Teams: if job["status"] in TERMINAL or job["thread_id"] in occupied: continue occupied.add(job["thread_id"]) - if job["status"] == "queued" and job["input"] is None: + reserved = job["status"] == "queued" and job["input"] is None + if reserved: if active >= 8: continue active += 1 @@ -414,6 +415,11 @@ class Teams: # Disconnect rather than releasing the member's slot. self.handles.pop((owner, team_id), None) break + finally: + # A conversation wait did not attempt admission. Frozen + # inputs retain the slot, including unknown outcomes. + if reserved and job["input"] is None: + active -= 1 async def save_job(self, owner, team_id, job): def update(team): @@ -443,6 +449,7 @@ class Teams: "Use the exact registered tool names in your tool list. Your final answer is automatically returned to the requester; " "do not send a separate peer request just to report completion. Do not wait in a polling loop.\n" ) + json.dumps({"request": job["text"], "your_member_id": job["member_id"], "source_thread": job["source"], "kind": job["kind"], **shared(team)}, ensure_ascii=False) + # Freeze and persist input before start(): tick releases reservations only while input is None. job["input"] = {"messages": [{"role": "user", "id": "team-" + job["id"], "content": content}]} await self.save_job(owner, team["id"], job) if job["run_id"] is None: