chore(conductor): Mark track 'Saved Tool Presets' as complete

src/ai_client.py

@@ -98,6 +98,8 @@ tool_log_callback: Optional[Callable[[str, str], None]] = None
 
 _local_storage = threading.local()
 
+_tool_approval_modes: dict[str, str] = {}
+
 def get_current_tier() -> Optional[str]:
  """Returns the current tier from thread-local storage."""
  return getattr(_local_storage, "current_tier", None)
@@ -468,6 +470,35 @@ def set_agent_tools(tools: dict[str, bool]) -> None:
  _agent_tools = tools
  _CACHED_ANTHROPIC_TOOLS = None
 
+def set_tool_preset(preset_name: Optional[str]) -> None:
+ """Loads a tool preset and applies it via set_agent_tools."""
+ global _agent_tools, _CACHED_ANTHROPIC_TOOLS, _tool_approval_modes
+ _tool_approval_modes = {}
+ if not preset_name or preset_name == "None":
+  # Enable all tools if no preset
+  _agent_tools = {name: True for name in mcp_client.TOOL_NAMES}
+  _agent_tools[TOOL_NAME] = True
+ else:
+  try:
+   from src.tool_presets import ToolPresetManager
+   manager = ToolPresetManager()
+   presets = manager.load_all()
+   if preset_name in presets:
+    preset = presets[preset_name]
+    new_tools = {name: False for name in mcp_client.TOOL_NAMES}
+    new_tools[TOOL_NAME] = False
+    for cat in preset.categories.values():
+     for tool_entry in cat:
+      if isinstance(tool_entry, dict) and "name" in tool_entry:
+       name = tool_entry["name"]
+       new_tools[name] = True
+       _tool_approval_modes[name] = tool_entry.get("mode", "ask")
+    _agent_tools = new_tools
+  except Exception as e:
+   sys.stderr.write(f"[ERROR] Failed to set tool preset '{preset_name}': {e}\n")
+   sys.stderr.flush()
+ _CACHED_ANTHROPIC_TOOLS = None
+
 def _build_anthropic_tools() -> list[dict[str, Any]]:
  mcp_tools: list[dict[str, Any]] = []
  for spec in mcp_client.MCP_TOOL_SPECS:
@@ -621,22 +652,29 @@ async def _execute_single_tool_call_async(
  tool_executed = False
  events.emit("tool_execution", payload={"status": "started", "tool": name, "args": args, "round": r_idx})
  
+ # Check for auto approval mode
+ approval_mode = _tool_approval_modes.get(name, "ask")
+
  # Check for run_powershell
- if name == TOOL_NAME and pre_tool_callback:
+ if name == TOOL_NAME:
   scr = cast(str, args.get("script", ""))
   _append_comms("OUT", "tool_call", {"name": TOOL_NAME, "id": call_id, "script": scr})
-  # pre_tool_callback is synchronous and might block for HITL
-  res = await asyncio.to_thread(pre_tool_callback, scr, base_dir, qa_callback)
-  if res is None:
-   out = "USER REJECTED: tool execution cancelled"
-  else:
-   out = res
-  tool_executed = True
+  if approval_mode == "auto":
+   out = await asyncio.to_thread(_run_script, scr, base_dir, qa_callback, patch_callback)
+   tool_executed = True
+  elif pre_tool_callback:
+   # pre_tool_callback is synchronous and might block for HITL
+   res = await asyncio.to_thread(pre_tool_callback, scr, base_dir, qa_callback)
+   if res is None:
+    out = "USER REJECTED: tool execution cancelled"
+   else:
+    out = res
+   tool_executed = True
 
  if not tool_executed:
   if name and name in mcp_client.TOOL_NAMES:
    _append_comms("OUT", "tool_call", {"name": name, "id": call_id, "args": args})
-   if name in mcp_client.MUTATING_TOOLS and pre_tool_callback:
+   if name in mcp_client.MUTATING_TOOLS and approval_mode != "auto" and pre_tool_callback:
     desc = f"# MCP MUTATING TOOL: {name}\n" + "\n".join(f"#   {k}: {repr(v)}" for k, v in args.items())
     _res = await asyncio.to_thread(pre_tool_callback, desc, base_dir, qa_callback)
     out = "USER REJECTED: tool execution cancelled" if _res is None else await mcp_client.async_dispatch(name, args)
@@ -2161,21 +2199,13 @@ def _add_bleed_derived(d: dict[str, Any], sys_tok: int = 0, tool_tok: int = 0) -
  d["tool_tokens"] = tool_tok
  d["history_tokens"] = max(0, cur - sys_tok - tool_tok)
  return d
-
-def _is_mutating_tool(name: str) -> bool:
- """Returns True if the tool name is considered a mutating tool."""
- return name in mcp_client.MUTATING_TOOLS or name == TOOL_NAME
-
-def _confirm_and_run(script: str, base_dir: str, qa_callback: Optional[Callable[[str], str]] = None, patch_callback: Optional[Callable[[str, str], Optional[str]]] = None) -> Optional[str]:
- """
- Wrapper for the confirm_and_run_callback. 
- This is what the providers call to trigger HITL approval.
- """
- if confirm_and_run_callback:
-  return confirm_and_run_callback(script, base_dir, qa_callback, patch_callback)
- # Fallback to direct execution if no callback registered (headless default)
- from src import shell_runner
- return shell_runner.run_powershell(script, base_dir, qa_callback=qa_callback, patch_callback=patch_callback)
+# Check for tool preset in environment variable (headless mode)
+if os.environ.get("SLOP_TOOL_PRESET"):
+ try:
+  set_tool_preset(os.environ["SLOP_TOOL_PRESET"])
+ except Exception as _e:
+  sys.stderr.write(f"[DEBUG] Failed to auto-set tool preset from env: {_e}\n")
+  sys.stderr.flush()
 
 def get_history_bleed_stats(md_content: Optional[str] = None) -> dict[str, Any]:
  if _provider == "anthropic":