From 2dd308cf3b37ce7e4827f2d11fa1395bae982339 Mon Sep 17 00:00:00 2001 From: wuyang <5700876+banisherwy@user.noreply.gitee.com> Date: Sun, 26 Jul 2026 14:31:48 +0800 Subject: [PATCH] fix: reject invalid and duplicate Work writes --- agent_platform/runtime/loop.py | 64 +++++++++++++++++++++++++++++++++- tests/test_agent_loop.py | 51 +++++++++++++++++++++++++++ 2 files changed, 114 insertions(+), 1 deletion(-) diff --git a/agent_platform/runtime/loop.py b/agent_platform/runtime/loop.py index 4729044..cdfbc58 100644 --- a/agent_platform/runtime/loop.py +++ b/agent_platform/runtime/loop.py @@ -1,6 +1,8 @@ from __future__ import annotations +import ast import asyncio +import hashlib import html import json import re @@ -319,6 +321,32 @@ def normalize_tool_arguments(name: str, arguments: dict[str, Any]) -> dict[str, return {**arguments, "content": normalized} +def write_call_fingerprint(name: str, arguments: dict[str, Any]) -> str | None: + if name != "write_file": + return None + path = arguments.get("path") + content = arguments.get("content") + if not isinstance(path, str) or not isinstance(content, str): + return None + digest = hashlib.sha256(content.encode("utf-8")).hexdigest() + return f"{PurePosixPath(path)}\0{digest}" + + +def python_source_error(arguments: dict[str, Any]) -> str | None: + path = arguments.get("path") + content = arguments.get("content") + if not isinstance(path, str) or not isinstance(content, str): + return None + if PurePosixPath(path).suffix.lower() != ".py": + return None + try: + ast.parse(content, filename=path) + except SyntaxError as exc: + location = f"line {exc.lineno}" if exc.lineno else "unknown line" + return f"{exc.msg} ({location})" + return None + + def is_substantive_execution(name: str, arguments: dict[str, Any]) -> bool: if name == "start_process": return True @@ -466,6 +494,7 @@ class AgentLoop: successful_executions = 0 unresolved_execution_failure: str | None = None failed_execution_revisions: dict[str, int] = {} + successful_write_fingerprints: set[str] = set() consecutive_checkpoint_rejections = 0 for iteration in range(max_iterations): await recorder.emit( @@ -544,6 +573,7 @@ class AgentLoop: blocked_call_reasons: dict[str, str] = {} batch_mutation_epoch = 0 batch_execution_epochs: dict[str, int] = {} + batch_write_fingerprints: set[str] = set() for call in tool_calls: function = call.get("function") or {} name = str(function.get("name", "")) @@ -551,12 +581,22 @@ class AgentLoop: arguments = json.loads(function.get("arguments") or "{}") except json.JSONDecodeError: arguments = {} + if isinstance(arguments, dict): + arguments = normalize_tool_arguments(name, arguments) if name in MUTATION_TOOLS: batch_mutation_epoch += 1 + fingerprint = write_call_fingerprint(name, arguments) if isinstance(arguments, dict) else None + call_id = str(call.get("id") or "") + if fingerprint and ( + fingerprint in successful_write_fingerprints or fingerprint in batch_write_fingerprints + ): + blocked_call_reasons[call_id] = "duplicate_write" + continue + if fingerprint: + batch_write_fingerprints.add(fingerprint) key = execution_call_key(name, arguments) if isinstance(arguments, dict) else None 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: @@ -580,10 +620,15 @@ class AgentLoop: arguments = json.loads(function.get("arguments") or "{}") except json.JSONDecodeError: arguments = {} + if isinstance(arguments, dict): + arguments = normalize_tool_arguments(name, arguments) if result.get("ok", False): successful_tools += 1 if name in MUTATION_TOOLS: mutation_count += 1 + fingerprint = write_call_fingerprint(name, arguments) if isinstance(arguments, dict) else None + if fingerprint: + successful_write_fingerprints.add(fingerprint) elif name in VERIFICATION_TOOLS and mutation_count: verified_mutation = mutation_count if is_substantive_execution(name, arguments): @@ -721,6 +766,15 @@ class AgentLoop: ) if "__parse_error__" in arguments: result = {"ok": False, "error": arguments["__parse_error__"]} + elif blocked_reason == "duplicate_write": + result = { + "ok": False, + "policy_duplicate": True, + "error": ( + "Execution policy skipped this write because the same path and identical content were " + "already written in this run. Make a real change before writing again." + ), + } elif blocked_reason == "batch_duplicate": result = { "ok": False, @@ -739,6 +793,14 @@ class AgentLoop: "a materially different diagnostic command." ), } + elif syntax_error := python_source_error(arguments): + result = { + "ok": False, + "error": ( + "Refused to write invalid complete Python source. " + f"Fix the content and call write_file again: {syntax_error}" + ), + } elif name not in TOOL_METADATA: result = {"ok": False, "error": f"Unknown tool: {name}"} elif read_only and name not in READ_ONLY_TOOL_NAMES: diff --git a/tests/test_agent_loop.py b/tests/test_agent_loop.py index 990e6ec..f1d29e7 100644 --- a/tests/test_agent_loop.py +++ b/tests/test_agent_loop.py @@ -11,6 +11,7 @@ from agent_platform.runtime.loop import ( AgentLoop, is_substantive_execution, normalize_write_file_content, + python_source_error, select_tool_specs, tool_event_details, ) @@ -39,6 +40,13 @@ def test_write_file_normalizes_double_escaped_plain_text() -> None: ) +def test_invalid_complete_python_source_is_detected_before_write() -> None: + error = python_source_error({"path": "benchmark.py", "content": "# Report\n- O(n²)\n"}) + assert error is not None + assert "invalid character" in error + assert python_source_error({"path": "benchmark.py", "content": "print('ok')\n"}) is None + + def test_read_only_shell_commands_do_not_satisfy_execution_evidence() -> None: assert not is_substantive_execution("exec", {"command": "cat report.txt"}) assert not is_substantive_execution("exec", {"command": "sed -n '1,20p' script.py"}) @@ -509,6 +517,49 @@ async def test_duplicate_commands_in_one_model_response_execute_once(settings) - await store.close() +async def test_duplicate_identical_writes_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", "write_file", {"path": "sort.py", "content": "print('ok')"}), + tool_call("3", "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 = provider.requests[1]["messages"][-2] + assert duplicate["role"] == "tool" + assert "same path and identical content" in duplicate["content"] + finally: + await store.close() + + async def test_script_delivery_requires_successful_execution(settings) -> None: write = { "choices": [