Fix/edit tool bug

.flocks/plugins/workflows/loop_host_forensics_fast/workflow.md

@@ -51,14 +51,14 @@
 
 ### 3. 单台巡检（`inspect_host`）
 
-- 工具/模型：Python + `ssh_host_cmd` 预检 + `task`（`subagent_type=host-forensics-fast`）
+- 工具/模型：Python + `ssh_host_cmd` 预检 + `delegate_task`（`subagent_type=host-forensics-fast`）
 - 输入：`hosts`、`host_idx`、`ssh_user`、`per_host_dir`、`batch_report_path`、`triage_results`
 - 处理逻辑：
   - 取当前 `hosts[host_idx]`，计算 `ssh_target`，并归一化出 `ssh_host` / `ssh_user`。
   - 先用 `ssh_host_cmd("echo FLOCKS_SSH_OK")` 做轻量 SSH 预检。
   - 若预检失败：按错误文本归类（如 `auth_failed`、`connect_timeout`、`connection_refused` 等），直接写入索引与单机报告。
   - 若预检通过：构造 prompt，明确要求子 Agent 调用 SSH 工具时分别传 `host` 和 `username`。
-  - 调用 `tool.run_safe('task', ...)` 执行巡检；若仅因超时失败，则自动重试 1 次。
+  - 调用 `tool.run_safe('delegate_task', ...)` 执行巡检；若仅因超时失败，则自动重试 1 次。
   - 将本轮完整输出立即写入 `host_triage/NNNN_slug.md`。
   - 从子 Agent 输出中提取 `Verdict`，未识别时回退为 `UNKNOWN`。
   - 向 `triage_results` 仅追加轻量字段：`{host, ssh_user, ssh_target, ssh_host, success, verdict, failure_category, inspect_attempts, error, per_host_md}`。

flocks/acp/agent.py

@@ -94,6 +94,31 @@ def to_locations(tool_name: str, input_data: Dict[str, Any]) -> List[ToolCallLoc
         return []
 
 
+def edit_diff_text(
+    input_data: Dict[str, Any],
+    metadata: Any,
+) -> tuple[str, str]:
+    """Extract edit text for ACP diff rendering from result metadata or edits[]."""
+    if isinstance(metadata, dict):
+        filediff = metadata.get("filediff")
+        if isinstance(filediff, dict):
+            before = filediff.get("before")
+            after = filediff.get("after")
+            if isinstance(before, str) and isinstance(after, str):
+                return before, after
+
+    edits = input_data.get("edits")
+    if isinstance(edits, list) and len(edits) == 1 and isinstance(edits[0], dict):
+        old_text = edits[0].get("oldString", "")
+        new_text = edits[0].get("newString", "")
+        return (
+            old_text if isinstance(old_text, str) else "",
+            new_text if isinstance(new_text, str) else "",
+        )
+
+    return "", ""
+
+
 def parse_uri(uri: str) -> Dict[str, Any]:
     """
     Parse URI into file or text content
@@ -503,8 +528,10 @@ async def _handle_tool_part_update(self, session_id: str, part: Dict[str, Any])
             # Add diff content for edit tools
             if kind == "edit":
                 file_path = input_data.get("filePath", "")
-                old_text = input_data.get("oldString", "")
-                new_text = input_data.get("newString", input_data.get("content", ""))
+                old_text, new_text = edit_diff_text(
+                    input_data,
+                    state.get("metadata"),
+                )
                 content.append({
                     "type": "diff",
                     "path": file_path,
@@ -854,8 +881,10 @@ async def _process_message(self, message: Dict[str, Any]) -> None:
 
                     if kind == "edit":
                         file_path = input_data.get("filePath", "")
-                        old_text = input_data.get("oldString", "")
-                        new_text = input_data.get("newString", input_data.get("content", ""))
+                        old_text, new_text = edit_diff_text(
+                            input_data,
+                            state.get("metadata"),
+                        )
                         content.append({
                             "type": "diff",
                             "path": file_path,

flocks/agent/agents/hephaestus/prompt_builder.py

@@ -29,7 +29,6 @@ def inject(
         available_agents=available_agents,
         available_tools=tools,
         available_skills=skills,
-        use_task_system=False,
     )
 
 
@@ -39,6 +38,8 @@ def build_hephaestus_prompt(
     available_skills: List["AvailableSkill"],
     use_task_system: bool = False,
 ) -> str:
+    del use_task_system
+
     from flocks.agent.prompt_utils import (
         build_agent_selection_table,
         build_key_triggers_section,
@@ -62,7 +63,7 @@ def build_hephaestus_prompt(
     oracle_section = build_oracle_section(available_agents)
     hard_blocks = build_hard_blocks_section()
     anti_patterns = build_anti_patterns_section()
-    todo_discipline = _todo_discipline_section(use_task_system)
+    todo_discipline = _todo_discipline_section()
 
     template = """You are Hephaestus, an autonomous deep worker for software engineering.
 
@@ -245,44 +246,7 @@ def build_hephaestus_prompt(
     return prompt
 
 
-def _todo_discipline_section(use_task_system: bool) -> str:
-    if use_task_system:
-        return """## Task Discipline (NON-NEGOTIABLE)
-
-**Track ALL multi-step work with tasks. This is your execution backbone.**
-
-### When to Create Tasks (MANDATORY)
-
-| Trigger | Action |
-|---------|--------|
-| 2+ step task | `TaskCreate` FIRST, atomic breakdown |
-| Uncertain scope | `TaskCreate` to clarify thinking |
-| Complex single task | Break down into trackable steps |
-
-### Workflow (STRICT)
-
-1. **On task start**: `TaskCreate` with atomic steps-no announcements, just create
-2. **Before each step**: `TaskUpdate(status="in_progress")` (ONE at a time)
-3. **After each step**: `TaskUpdate(status="completed")` IMMEDIATELY (NEVER batch)
-4. **Scope changes**: Update tasks BEFORE proceeding
-
-### Why This Matters
-
-- **Execution anchor**: Tasks prevent drift from original request
-- **Recovery**: If interrupted, tasks enable seamless continuation
-- **Accountability**: Each task = explicit commitment to deliver
-
-### Anti-Patterns (BLOCKING)
-
-| Violation | Why It Fails |
-|-----------|--------------|
-| Skipping tasks on multi-step work | Steps get forgotten, user has no visibility |
-| Batch-completing multiple tasks | Defeats real-time tracking purpose |
-| Proceeding without `in_progress` | No indication of current work |
-| Finishing without completing tasks | Task appears incomplete |
-
-**NO TASKS ON MULTI-STEP WORK = INCOMPLETE WORK.**"""
-
+def _todo_discipline_section() -> str:
     return """## Todo Discipline (NON-NEGOTIABLE)
 
 **Track ALL multi-step work with todos. This is your execution backbone.**

flocks/agent/agents/rex/prompt_builder.py

@@ -30,7 +30,6 @@ def inject(
         available_tools=tools,
         available_skills=skills,
         available_workflows=workflows or [],
-        use_task_system=False,
     )
 
 
@@ -49,6 +48,7 @@ def build_dynamic_rex_prompt(
     )
 
     _ = available_tools
+    del use_task_system
 
     key_triggers = build_key_triggers_section(available_agents, available_skills)
     agent_selection = build_agent_selection_table(available_agents)
@@ -58,12 +58,8 @@ def build_dynamic_rex_prompt(
     im_send_section = _build_im_send_pointer_section()
     anti_patterns = _build_rex_anti_patterns_section()
     command_guidance_section = _build_command_guidance_section()
-    task_management_section = _task_management_section(use_task_system)
-    todo_hook_note = (
-        "YOUR TASK CREATION WOULD BE TRACKED BY HOOK([SYSTEM REMINDER - TASK CONTINUATION])"
-        if use_task_system
-        else "YOUR TODO CREATION WOULD BE TRACKED BY HOOK([SYSTEM REMINDER - TODO CONTINUATION])"
-    )
+    task_management_section = _task_management_section()
+    todo_hook_note = "YOUR TODO CREATION WOULD BE TRACKED BY HOOK([SYSTEM REMINDER - TODO CONTINUATION])"
 
     template = """<Role>
 You are "Rex" - Powerful AI orchestrator for security operations.
@@ -143,13 +139,12 @@ def build_dynamic_rex_prompt(
 - Match existing codebase patterns when editing.
 - Fix bugs minimally; do not refactor during a bugfix unless required.
 - Keep search bounded: stop when you have enough context, when results repeat, or when direct evidence already answers the question.
-- For independent parallel branches whose results are needed this turn, emit multiple foreground `delegate_task` / `task` tool calls in the same assistant turn. The runtime executes those sibling tool calls concurrently and returns all tool results before you continue.
+- For independent parallel branches whose results are needed this turn, emit multiple foreground `delegate_task` tool calls in the same assistant turn. The runtime executes those sibling tool calls concurrently and returns all tool results before you continue.
 - Do not use `run_in_background=true`; background subagent execution is disabled.
 
 ## 5. Verify
 
- - Use `lsp` for symbol-aware checks when useful, and run relevant tests on changed files before considering the work complete.
-- Run relevant build or test commands before finalizing when the affected area has them.
+- After code changes, run the lint/typecheck/tests. If tests fail, iterate until they pass before finalizing.
 - Verification evidence is mandatory: clean diagnostics, successful commands, or an explicit note about pre-existing failures.
 - Verify delegated work against expected behavior, codebase patterns, and any `must-do` / `must-not-do` requirements.
 
@@ -278,55 +273,42 @@ def _build_clarification_protocol() -> str:
 ```"""
 
 
-def _task_management_section(use_task_system: bool) -> str:
-    title = "Task Management" if use_task_system else "Todo Management"
-    unit = "tasks" if use_task_system else "todos"
-    create_action = "`TaskCreate`" if use_task_system else '`todo(action="write")`'
-    progress_action = (
-        '`TaskUpdate(status="in_progress")`'
-        if use_task_system
-        else "mark `in_progress`"
-    )
-    complete_action = (
-        '`TaskUpdate(status="completed")`'
-        if use_task_system
-        else "mark `completed`"
-    )
+def _task_management_section() -> str:
     clarification_protocol = _build_clarification_protocol()
 
-    return f"""<Task_Management>
-## {title}
+    return f"""<Todo_Management>
+## Todo Management
 
-Use {unit} as the primary coordination mechanism for non-trivial execution work.
+Use todos as the primary coordination mechanism for non-trivial execution work.
 
 ### When They Are Mandatory
 
 | Trigger | Action |
 |---------|--------|
-| Multi-step work (2+ steps) | Create {unit} first |
-| Uncertain scope | Create {unit} to structure the work |
-| User request with multiple items | Create {unit} first |
-| Complex single task | Break it into {unit} |
+| Multi-step work (2+ steps) | Create todos first |
+| Uncertain scope | Create todos to structure the work |
+| User request with multiple items | Create todos first |
+| Complex single task | Break it into todos |
 
 ### Operating Rules
 
-1. Start with {create_action} before implementation work begins.
-2. ONLY add {unit} when the user wants execution, not when they only want analysis or planning.
-3. Before each step, {progress_action}. Keep only one item in progress.
-4. After each step, {complete_action} immediately. Never batch updates.
-5. If scope changes, update the {unit} before continuing.
+1. Start with `todo(action="write")` before implementation work begins.
+2. ONLY add todos when the user wants execution, not when they only want analysis or planning.
+3. Before each step, mark it `in_progress`. Keep only one item in progress.
+4. After each step, mark it `completed` immediately. Never batch updates.
+5. If scope changes, update the todos before continuing.
 
 ### Failure Modes
 
 | Violation | Why It Breaks the Workflow |
 |-----------|----------------------------|
-| Skipping {unit} on non-trivial work | The user loses progress visibility and steps get dropped |
-| Batch-completing multiple {unit} | Real-time tracking becomes meaningless |
+| Skipping todos on non-trivial work | The user loses progress visibility and steps get dropped |
+| Batch-completing multiple todos | Real-time tracking becomes meaningless |
 | Proceeding without an in-progress item | It is unclear what is being worked on |
 | Finishing without closing items | The work appears incomplete |
 
 {clarification_protocol}
-</Task_Management>"""
+</Todo_Management>"""
 
 
 def _build_security_priority_section(available_agents: List["AvailableAgent"]) -> str:

flocks/agent/registry.py

@@ -40,6 +40,7 @@
 )
 import flocks.agent.delegatable_settings as delegatable_settings
 from flocks.agent.toolset import agent_declares_tool
+from flocks.agent.tool_permissions import permission_items_to_ruleset
 from flocks.agent.prompt_utils import categorize_tools
 from flocks.agent.agent_factory import (
     scan_and_load,
@@ -172,6 +173,7 @@ def _storage_custom_agent_to_info(agent_data: Dict[str, Any]) -> Optional[AgentI
         native=False,
         hidden=agent_data.get("hidden", False),
         delegatable=agent_data.get("delegatable"),
+        permission=permission_items_to_ruleset(agent_data.get("permission")),
         tools=agent_data.get("tools", []),
         tags=agent_data.get("tags", []),
     )

flocks/agent/tool_permissions.py

@@ -0,0 +1,86 @@
+"""Helpers for keeping agent tool selections and permission rules aligned."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List
+
+from flocks.permission.rule import PermissionLevel, PermissionRule, PermissionScope
+
+QUESTION_TOOL_NAME = "question"
+TOOLS_MANAGED_PERMISSION_SOURCE = "agent_tools"
+
+
+def normalize_permission_items(value: Any) -> List[Dict[str, Any]]:
+    """Return stored permission rules in API/storage dict form."""
+    if not isinstance(value, list):
+        return []
+
+    normalized: List[Dict[str, Any]] = []
+    for item in value:
+        if not isinstance(item, dict):
+            continue
+
+        permission = item.get("permission")
+        action = item.get("action") or item.get("level")
+        action = getattr(action, "value", action)
+        pattern = item.get("pattern") or "*"
+        if not permission or action not in {"allow", "ask", "deny"}:
+            continue
+
+        rule = dict(item)
+        rule["permission"] = str(permission)
+        rule["action"] = str(action)
+        rule["pattern"] = str(pattern)
+        normalized.append(rule)
+
+    return normalized
+
+
+def sync_question_permission_with_tools(
+    permissions: Any,
+    tools: List[str],
+) -> List[Dict[str, Any]]:
+    """Make the Agent tools checkbox an effective question allow/deny toggle."""
+    normalized = normalize_permission_items(permissions)
+    without_managed_question = [
+        rule
+        for rule in normalized
+        if not (
+            rule.get("permission") == QUESTION_TOOL_NAME
+            and rule.get("pattern", "*") == "*"
+            and rule.get("source") == TOOLS_MANAGED_PERMISSION_SOURCE
+        )
+    ]
+
+    if QUESTION_TOOL_NAME in set(tools):
+        return without_managed_question
+
+    if any(
+        rule.get("permission") == QUESTION_TOOL_NAME
+        and rule.get("action") == "deny"
+        and rule.get("pattern", "*") == "*"
+        for rule in without_managed_question
+    ):
+        return without_managed_question
+
+    return without_managed_question + [
+        {
+            "permission": QUESTION_TOOL_NAME,
+            "action": "deny",
+            "pattern": "*",
+            "source": TOOLS_MANAGED_PERMISSION_SOURCE,
+        }
+    ]
+
+
+def permission_items_to_ruleset(value: Any) -> List[PermissionRule]:
+    """Convert stored permission dicts to the internal PermissionRule shape."""
+    return [
+        PermissionRule(
+            permission=item["permission"],
+            level=PermissionLevel(item["action"]),
+            scope=PermissionScope.PATTERN,
+            pattern=item.get("pattern") or "*",
+        )
+        for item in normalize_permission_items(value)
+    ]

flocks/cli/commands/import_.py

@@ -98,6 +98,17 @@ def _normalize_part_data(
     metadata = normalized.get("metadata")
     metadata_dict = metadata if isinstance(metadata, dict) else {}
 
+    if part_type == "subtask":
+        normalized["type"] = "text"
+        normalized["text"] = ""
+        normalized["ignored"] = True
+        normalized["metadata"] = {
+            **metadata_dict,
+            "legacyPartType": "subtask",
+        }
+        part_type = "text"
+        metadata_dict = normalized["metadata"]
+
     if "content" in normalized and "text" not in normalized:
         normalized["text"] = normalized.get("content", "")
 
@@ -135,10 +146,6 @@ def _normalize_part_data(
         )
     elif part_type == "agent":
         normalized.setdefault("name", metadata_dict.get("name") or normalized.get("content") or "agent")
-    elif part_type == "subtask":
-        normalized.setdefault("prompt", metadata_dict.get("prompt") or normalized.get("content", ""))
-        normalized.setdefault("description", metadata_dict.get("description") or "")
-        normalized.setdefault("agent", metadata_dict.get("agent") or "agent")
     elif part_type == "retry":
         normalized.setdefault("attempt", metadata_dict.get("attempt") or 1)
         normalized.setdefault("error", metadata_dict.get("error") or {})