feat(hanzo-mcp): v0.10.2 - DAG auto-backgrounding and zsh tool

- DAG commands that exceed timeout now auto-background instead of failing
- Backgrounded processes return SUCCESS status so DAG continues
- Register backgrounded processes with ProcessManager for ps tool tracking
- Async log capture for backgrounded process output
- Enable standalone zsh tool for simple single-command execution
- All I/O uses aiofiles for non-blocking async operations
- Kill any backgrounded process without affecting DAG completion

pkg/hanzo-mcp/hanzo_mcp/tools/shell/__init__.py

@@ -20,6 +20,7 @@ from hanzo_mcp.tools.common.permissions import PermissionManager
 # Core tools
 from hanzo_mcp.tools.shell.dag_tool import DagTool, create_dag_tool
 from hanzo_mcp.tools.shell.ps_tool import PsTool, ps_tool
+from hanzo_mcp.tools.shell.zsh_tool import ZshTool, zsh_tool
 
 # Convenience tools
 from hanzo_mcp.tools.shell.open import open_tool
@@ -29,8 +30,10 @@ from hanzo_mcp.tools.shell.uvx_tool import uvx_tool
 __all__ = [
     "DagTool",
     "PsTool",
+    "ZshTool",
     "create_dag_tool",
     "ps_tool",
+    "zsh_tool",
     "get_shell_tools",
     "register_shell_tools",
 ]
@@ -57,8 +60,9 @@ def get_shell_tools(
     uvx_tool.permission_manager = permission_manager
 
     return [
-        dag,        # DAG execution (replaces bash, zsh, shell, exec, batch)
-        ps_tool,    # Process management (replaces process)
+        dag,        # DAG execution for complex workflows
+        zsh_tool,   # Simple single-command zsh (with auto-backgrounding)
+        ps_tool,    # Process management
         npx_tool,   # Node packages
         uvx_tool,   # Python packages
         open_tool,  # Open files/URLs

pkg/hanzo-mcp/hanzo_mcp/tools/shell/dag_tool.py

@@ -7,10 +7,15 @@ Supports serial (default), parallel, and complex mixed execution graphs.
 import asyncio
 import os
 import shutil
+import uuid
+import tempfile
 from typing import Any, Union, List, Dict, Optional, Annotated, override
 from dataclasses import dataclass, field
 from enum import Enum
 from datetime import datetime
+from pathlib import Path
+
+import aiofiles
 
 from pydantic import Field
 from mcp.server import FastMCP
@@ -19,6 +24,7 @@ from mcp.server.fastmcp import Context as MCPContext
 from hanzo_mcp.tools.common.base import BaseTool
 from hanzo_mcp.tools.common.context import create_tool_context
 from hanzo_mcp.tools.common.auto_timeout import auto_timeout
+from hanzo_mcp.tools.shell.base_process import ProcessManager
 
 
 class NodeStatus(Enum):
@@ -179,26 +185,38 @@ Named with dependencies:
       {"id": "test", "run": "make test", "after": ["build"]},
   ])
 
-Uses zsh for shell execution."""
+Uses zsh for shell execution.
+
+AUTO-BACKGROUNDING: Commands that exceed timeout are automatically
+backgrounded. Use ps tool to monitor: ps --logs <id>, ps --kill <id>"""
 
     async def _run_shell(
-        self, 
-        cmd: str, 
-        shell: str, 
+        self,
+        cmd: str,
+        shell: str,
         cwd: Optional[str] = None,
         env: Optional[Dict[str, str]] = None,
         timeout: int = 120,
     ) -> DagResult:
-        """Run a shell command asynchronously."""
+        """Run a shell command with auto-backgrounding on timeout.
+
+        Commands that exceed the timeout are automatically backgrounded
+        and can be monitored with the ps tool. All I/O is async.
+        """
         start_time = datetime.now()
         node_id = f"shell_{id(cmd)}"
-        
+        process_manager = ProcessManager()
+
         try:
             # Merge environment
             run_env = os.environ.copy()
             if env:
                 run_env.update(env)
-            
+
+            # Create log file for potential backgrounding
+            process_id = f"dag_{uuid.uuid4().hex[:8]}"
+            log_file = await process_manager.create_log_file(process_id)
+
             proc = await asyncio.create_subprocess_exec(
                 shell, "-c", cmd,
                 stdout=asyncio.subprocess.PIPE,
@@ -206,38 +224,58 @@ Uses zsh for shell execution."""
                 cwd=cwd,
                 env=run_env,
             )
-            
-            stdout, stderr = await asyncio.wait_for(
-                proc.communicate(), 
-                timeout=timeout
-            )
-            
-            duration = int((datetime.now() - start_time).total_seconds() * 1000)
-            exit_code = proc.returncode or 0
-            
-            return DagResult(
-                node_id=node_id,
-                command=cmd,
-                stdout=stdout.decode("utf-8", errors="replace"),
-                stderr=stderr.decode("utf-8", errors="replace"),
-                status=NodeStatus.SUCCESS if exit_code == 0 else NodeStatus.FAILED,
-                exit_code=exit_code,
-                duration_ms=duration,
-                node_type="shell",
-            )
-            
-        except asyncio.TimeoutError:
-            duration = int((datetime.now() - start_time).total_seconds() * 1000)
-            return DagResult(
-                node_id=node_id,
-                command=cmd,
-                stdout="",
-                stderr=f"Command timed out after {timeout}s",
-                status=NodeStatus.FAILED,
-                exit_code=124,
-                duration_ms=duration,
-                node_type="shell",
-            )
+
+            try:
+                stdout, stderr = await asyncio.wait_for(
+                    proc.communicate(),
+                    timeout=timeout
+                )
+
+                duration = int((datetime.now() - start_time).total_seconds() * 1000)
+                exit_code = proc.returncode or 0
+
+                return DagResult(
+                    node_id=node_id,
+                    command=cmd,
+                    stdout=stdout.decode("utf-8", errors="replace"),
+                    stderr=stderr.decode("utf-8", errors="replace"),
+                    status=NodeStatus.SUCCESS if exit_code == 0 else NodeStatus.FAILED,
+                    exit_code=exit_code,
+                    duration_ms=duration,
+                    node_type="shell",
+                )
+
+            except asyncio.TimeoutError:
+                # Auto-background: register process and return status
+                duration = int((datetime.now() - start_time).total_seconds() * 1000)
+
+                # Write initial status to log (async)
+                async with aiofiles.open(log_file, "w") as f:
+                    await f.write(f"[dag] Command backgrounded after {timeout}s timeout\n")
+                    await f.write(f"[dag] Command: {cmd}\n")
+                    await f.write(f"[dag] PID: {proc.pid}\n")
+                    await f.write(f"[dag] Started: {start_time.isoformat()}\n")
+                    await f.write("-" * 40 + "\n")
+
+                # Register with ProcessManager for ps tool
+                process_manager.add_process(process_id, proc, str(log_file))
+
+                # Start background log capture (non-blocking task)
+                asyncio.create_task(self._capture_background_output(proc, log_file))
+
+                return DagResult(
+                    node_id=node_id,
+                    command=cmd,
+                    stdout=f"[backgrounded] Process {process_id} (PID {proc.pid}) running in background.\n"
+                           f"Use: ps --logs {process_id}  # view output\n"
+                           f"Use: ps --kill {process_id}  # stop process",
+                    stderr="",
+                    status=NodeStatus.SUCCESS,  # Backgrounding is success, not failure
+                    exit_code=0,
+                    duration_ms=duration,
+                    node_type="shell",
+                )
+
         except Exception as e:
             duration = int((datetime.now() - start_time).total_seconds() * 1000)
             return DagResult(
@@ -251,6 +289,36 @@ Uses zsh for shell execution."""
                 node_type="shell",
             )
 
+    async def _capture_background_output(
+        self,
+        proc: asyncio.subprocess.Process,
+        log_file: Path
+    ) -> None:
+        """Capture output from backgrounded process to log file (async)."""
+        try:
+            async with aiofiles.open(log_file, "a") as f:
+                # Read stdout/stderr concurrently
+                async def read_stream(stream, prefix: str):
+                    if stream:
+                        while True:
+                            line = await stream.readline()
+                            if not line:
+                                break
+                            await f.write(f"{prefix}{line.decode('utf-8', errors='replace')}")
+                            await f.flush()
+
+                await asyncio.gather(
+                    read_stream(proc.stdout, ""),
+                    read_stream(proc.stderr, "[stderr] "),
+                )
+
+                # Write completion status
+                await proc.wait()
+                await f.write(f"\n[dag] Process exited with code {proc.returncode}\n")
+        except Exception:
+            # Silently handle errors in background capture
+            pass
+
     async def _run_tool(
         self, 
         tool_name: str, 

pkg/hanzo-mcp/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "hanzo-mcp"
-version = "0.10.1"
+version = "0.10.2"
 description = "The Zen of Hanzo MCP: One server to rule them all. The ultimate MCP that orchestrates all others."
 readme = "README.md"
 requires-python = ">=3.12"