From ae5f940a5d188737e6e7aa92e4812914c0250899 Mon Sep 17 00:00:00 2001 From: wuyang <5700876+banisherwy@user.noreply.gitee.com> Date: Sun, 26 Jul 2026 14:11:57 +0800 Subject: [PATCH] fix: deduplicate Work execution batches --- agent_platform/runtime/loop.py | 41 ++++++++++++++++++++--------- tests/test_agent_loop.py | 47 ++++++++++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 12 deletions(-) diff --git a/agent_platform/runtime/loop.py b/agent_platform/runtime/loop.py index d677bcc..c0b9e66 100644 --- a/agent_platform/runtime/loop.py +++ b/agent_platform/runtime/loop.py @@ -489,8 +489,9 @@ class AgentLoop: "tool_calls": tool_calls, } messages.append(assistant_message) - blocked_call_ids: set[str] = set() - mutation_planned = False + blocked_call_reasons: dict[str, str] = {} + batch_mutation_epoch = 0 + batch_execution_epochs: dict[str, int] = {} for call in tool_calls: function = call.get("function") or {} name = str(function.get("name", "")) @@ -499,10 +500,17 @@ class AgentLoop: except json.JSONDecodeError: arguments = {} if name in MUTATION_TOOLS: - mutation_planned = True + batch_mutation_epoch += 1 key = execution_call_key(name, arguments) if isinstance(arguments, dict) else None - if key and not mutation_planned and failed_execution_revisions.get(key) == mutation_count: - blocked_call_ids.add(str(call.get("id") or "")) + if not key: + continue + call_id = str(call.get("id") or "") + if failed_execution_revisions.get(key) == mutation_count and batch_mutation_epoch == 0: + blocked_call_reasons[call_id] = "failed_retry" + elif batch_execution_epochs.get(key) == batch_mutation_epoch: + blocked_call_reasons[call_id] = "batch_duplicate" + else: + batch_execution_epochs[key] = batch_mutation_epoch results = await self._execute_calls( calls=tool_calls, spec=spec, @@ -510,7 +518,7 @@ class AgentLoop: context=tool_context, depth=depth, read_only=read_only, - blocked_call_ids=blocked_call_ids, + blocked_call_reasons=blocked_call_reasons, ) execution_failures: list[str] = [] for call, result in zip(tool_calls, results, strict=True): @@ -529,7 +537,7 @@ class AgentLoop: if is_substantive_execution(name, arguments): successful_executions += 1 unresolved_execution_failure = None - elif is_substantive_execution(name, arguments): + elif is_substantive_execution(name, arguments) and not result.get("policy_duplicate", False): unresolved_execution_failure = f"{name}: {public_tool_summary(result, 1000)}" execution_failures.append(unresolved_execution_failure) key = execution_call_key(name, arguments) if isinstance(arguments, dict) else None @@ -588,9 +596,9 @@ class AgentLoop: context: ToolContext, depth: int, read_only: bool, - blocked_call_ids: set[str] | None = None, + blocked_call_reasons: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - blocked_call_ids = blocked_call_ids or set() + blocked_call_reasons = blocked_call_reasons or {} parsed: list[tuple[int, dict[str, Any], str, dict[str, Any], bool]] = [] for index, call in enumerate(calls): function = call.get("function") or {} @@ -622,7 +630,7 @@ class AgentLoop: context=context, depth=depth, read_only=read_only, - blocked_repeat=str(call.get("id") or "") in blocked_call_ids, + blocked_reason=blocked_call_reasons.get(str(call.get("id") or "")), ) index = 0 @@ -649,7 +657,7 @@ class AgentLoop: context: ToolContext, depth: int, read_only: bool, - blocked_repeat: bool = False, + blocked_reason: str | None = None, ) -> dict[str, Any]: call_id = str(call.get("id") or uuid.uuid4().hex) public_args = { @@ -661,7 +669,16 @@ class AgentLoop: ) if "__parse_error__" in arguments: result = {"ok": False, "error": arguments["__parse_error__"]} - elif blocked_repeat: + elif blocked_reason == "batch_duplicate": + result = { + "ok": False, + "policy_duplicate": True, + "error": ( + "Execution policy skipped this duplicate command because the same assistant response " + "already requested it without an intervening repair." + ), + } + elif blocked_reason == "failed_retry": result = { "ok": False, "error": ( diff --git a/tests/test_agent_loop.py b/tests/test_agent_loop.py index 9ba8b4e..cb0b0ca 100644 --- a/tests/test_agent_loop.py +++ b/tests/test_agent_loop.py @@ -433,6 +433,53 @@ async def test_unchanged_failed_command_is_blocked_until_a_repair(settings) -> N await store.close() +async def test_duplicate_commands_in_one_model_response_execute_once(settings) -> None: + batch = { + "choices": [ + { + "message": { + "role": "assistant", + "tool_calls": [ + tool_call("1", "write_file", {"path": "sort.py", "content": "print('ok')"}), + tool_call("2", "exec", {"command": "python3 sort.py"}), + tool_call("3", "exec", {"command": "python3 sort.py"}), + tool_call("4", "exec", {"command": "python3 sort.py"}), + ], + }, + "finish_reason": "tool_calls", + } + ] + } + final = {"choices": [{"message": {"role": "assistant", "content": "完成"}, "finish_reason": "stop"}]} + provider = ScriptedProvider([batch, final]) + registry = RecordingRegistry() + store = RuntimeStore(settings.database_url) + await store.initialize() + + async def callback(event_type, payload): + return None + + try: + answer = await AgentLoop(provider, registry, store, max_tool_output_chars=10_000).run( + spec=get_model_spec("work-light"), + messages=[{"role": "user", "content": "写脚本并运行"}], + identity=UserIdentity("u1", "", "", "user"), + raw_user_jwt="jwt", + chat_id="c1", + callback=callback, + ) + assert answer == "完成" + assert [name for name, _ in registry.calls] == ["write_file", "exec"] + duplicate_results = [ + message + for message in provider.requests[1]["messages"] + if message.get("role") == "tool" and "duplicate command" in message.get("content", "") + ] + assert len(duplicate_results) == 2 + finally: + await store.close() + + async def test_script_delivery_requires_successful_execution(settings) -> None: write = { "choices": [