feat: focus Work tools by task intent

agent_platform/runtime/loop.py

@@ -33,6 +33,8 @@ relevant verification, and report concrete results. Use tools instead of inventi
 command output. A claim that a file was created, changed, executed, or tested must be backed by a
 successful tool result from this run. For scripts, reports, code, and other deliverables, create real
 files under /workspace, inspect them after writing, run relevant checks, and cite their exact paths.
+For a code-and-report task, follow this order: write a real source file, read it, run that exact source
+file with a complete command, write the report using only the measured output, then read the report.
 Write source files with real line breaks rather than literal `\\n` escape sequences. If any command
 fails, inspect the failure, repair the underlying problem, and successfully rerun the relevant check;
 never replace failed output with invented numbers, placeholders, or a hand-written success report.
@@ -113,6 +115,32 @@ READ_ONLY_EXEC_COMMAND = re.compile(
     r"^(?:cat|cut|diff|find|git\s+(?:diff|show|status)|grep|head|jq|ls|pwd|rg|sed\s+-n|stat|tail|wc)\b",
     re.IGNORECASE,
 )
+PROCESS_REQUEST = re.compile(
+    r"(服务器|服务|后台|常驻|网站|网页|\b(?:server|service|daemon|background|web\s*app)\b)",
+    re.IGNORECASE,
+)
+GIT_REQUEST = re.compile(
+    r"(仓库|提交|分支|\b(?:git|repo(?:sitory)?|commit|branch|pull request)\b)",
+    re.IGNORECASE,
+)
+MEMORY_REQUEST = re.compile(
+    r"(记住|记忆|偏好|忘记|\b(?:remember|memory|preference|forget)\b)",
+    re.IGNORECASE,
+)
+DELEGATION_REQUEST = re.compile(
+    r"(多\s*(?:agent|代理)|子代理|分工|并行调研|大型|架构|重构|迁移|"
+    r"\b(?:multi[- ]agent|sub[- ]agent|delegate|parallel research|architecture|migration|refactor)\b)",
+    re.IGNORECASE,
+)
+CORE_WORK_TOOLS = {
+    "workspace_status",
+    "list_files",
+    "read_file",
+    "search_files",
+    "write_file",
+    "apply_patch",
+    "exec",
+}
 
 
 def latest_user_text(messages: list[dict[str, Any]]) -> str:
@@ -300,6 +328,26 @@ def is_substantive_execution(name: str, arguments: dict[str, Any]) -> bool:
     return bool(command and not READ_ONLY_EXEC_COMMAND.match(command))
 
 
+def select_tool_specs(
+    specs: list[dict[str, Any]],
+    request_text: str,
+    *,
+    depth: int,
+) -> list[dict[str, Any]]:
+    allowed = set(CORE_WORK_TOOLS)
+    if depth == 0:
+        allowed.add("update_plan")
+    if PROCESS_REQUEST.search(request_text):
+        allowed.update({"start_process", "poll_process", "cancel_process"})
+    if GIT_REQUEST.search(request_text):
+        allowed.update({"git_status", "git_diff"})
+    if MEMORY_REQUEST.search(request_text):
+        allowed.update({"remember", "recall_memory", "forget_memory"})
+    if depth == 0 and DELEGATION_REQUEST.search(request_text):
+        allowed.add("delegate_task")
+    return [spec for spec in specs if str((spec.get("function") or {}).get("name", "")) in allowed]
+
+
 @dataclass(slots=True)
 class RunRecorder:
     store: RuntimeStore
@@ -403,8 +451,12 @@ class AgentLoop:
         read_only: bool,
     ) -> str:
         max_iterations = min(spec.max_iterations, 8 if depth else spec.max_iterations)
-        available_specs = self.tools.specs(read_only=read_only, allow_delegate=depth == 0)
         request_text = latest_user_text(messages)
+        available_specs = select_tool_specs(
+            self.tools.specs(read_only=read_only, allow_delegate=depth == 0),
+            request_text,
+            depth=depth,
+        )
         artifact_required = depth == 0 and bool(ARTIFACT_REQUEST.search(request_text))
         action_required = depth == 0 and bool(ACTION_REQUEST.search(request_text))
         execution_required = depth == 0 and bool(EXECUTION_REQUEST.search(request_text))

agent_platform/runtime/tools.py

@@ -92,7 +92,7 @@ TOOL_METADATA: dict[str, ToolMetadata] = {
     ),
     "write_file": ToolMetadata(
         "write_file",
-        "Write the complete UTF-8 contents of a workspace file.",
+        "Write complete UTF-8 file contents. Use actual line breaks between source lines, not literal \\\\n text.",
         object_schema(
             {
                 "path": {"type": "string", "maxLength": 4096},
@@ -115,7 +115,8 @@ TOOL_METADATA: dict[str, ToolMetadata] = {
     ),
     "exec": ToolMetadata(
         "exec",
-        "Run a shell command in the isolated workspace.",
+        "Run a complete non-interactive shell command. Name the script, pass -c code, or invoke a test; "
+        "bare python, node, or shells are rejected.",
         object_schema(
             {
                 "command": {"type": "string"},

tests/test_agent_loop.py

@@ -11,6 +11,7 @@ from agent_platform.runtime.loop import (
     AgentLoop,
     is_substantive_execution,
     normalize_write_file_content,
+    select_tool_specs,
     tool_event_details,
 )
 from agent_platform.runtime.tools import TOOL_METADATA
@@ -45,6 +46,34 @@ def test_read_only_shell_commands_do_not_satisfy_execution_evidence() -> None:
     assert is_substantive_execution("start_process", {"command": "python3 server.py"})
 
 
+def test_simple_script_task_receives_a_focused_tool_menu() -> None:
+    specs = [metadata.openai_spec() for metadata in TOOL_METADATA.values()]
+    selected = select_tool_specs(specs, "写脚本对比排序算法，给一个报告", depth=0)
+    names = {spec["function"]["name"] for spec in selected}
+    assert names == {
+        "workspace_status",
+        "list_files",
+        "read_file",
+        "search_files",
+        "write_file",
+        "apply_patch",
+        "exec",
+        "update_plan",
+    }
+
+
+def test_complex_requests_enable_only_relevant_optional_tools() -> None:
+    specs = [metadata.openai_spec() for metadata in TOOL_METADATA.values()]
+    selected = select_tool_specs(
+        specs,
+        "重构 Git 仓库里的 web server，并记住这个偏好，分工并行调研",
+        depth=0,
+    )
+    names = {spec["function"]["name"] for spec in selected}
+    assert {"delegate_task", "git_status", "git_diff", "start_process", "poll_process"} <= names
+    assert {"remember", "recall_memory", "forget_memory"} <= names
+
+
 class ScriptedProvider:
     def __init__(self, responses: list[dict[str, Any]]) -> None:
         self.responses = responses