Hermes 调度中心：spawn 多 Agent 与 terminal + delegate_task 编排实战

# 场景：快速检查工具可用性
terminal(command="claude --version")
# → claude-code 1.0.x

terminal(command="codex --version")
# → codex-cli 2.0.x

terminal(command="opencode --version")
# → opencode 0.3.x

# 场景：检查项目结构
terminal(command="find src -name '*.py' | head -20", workdir="/opt/data/my-project")

# 场景：启动 Claude Code 作为后台 Agent
result = terminal(
    command='claude "Review all Python files in src/ for security issues"',
    background=True,
    workdir="/opt/data/my-project",
    timeout=600,  # 最多 10 分钟
)
# → session_id: sess_abc123

# 等待完成
process(action="wait", session_id="sess_abc123", timeout=600)

# 获取输出
output = process(action="log", session_id="sess_abc123")

# 方式一：直接命令行调用
terminal(
    command='claude "Generate a REST API endpoint for user registration '
            'with email validation, password hashing, and JWT token creation. '
            'Use FastAPI and follow the existing project patterns."',
    workdir="/opt/data/my-project",
    timeout=300,
)

# 方式二：带系统提示词
terminal(
    command='claude -p "You are an expert Python developer. '
            'Always write PEP8-compliant code with type hints. '
            'Include docstrings and error handling. '
            'Generate a REST API endpoint for user registration..."',
    workdir="/opt/data/my-project",
    timeout=300,
)

# 使用 PTY 模式进行交互式会话
pty_result = terminal(
    command="claude",
    pty=True,
    background=True,
    workdir="/opt/data/my-project",
)

session_id = pty_result["session_id"]

# 发送任务指令
process(action="submit", session_id=session_id,
        data="Create a data validation module using Pydantic")

# 等待响应并获取输出
process(action="wait", session_id=session_id, timeout=120)
output = process(action="log", session_id=session_id, limit=100)

# 可以继续对话
process(action="submit", session_id=session_id,
        data="Now add unit tests for all validators")

process(action="wait", session_id=session_id, timeout=120)
final_output = process(action="log", session_id=session_id)

# 完成后关闭
process(action="close", session_id=session_id)

# 在生产环境中，建议限制 Claude Code 的权限
terminal(
    command='claude --allowedTools="Edit,Read,Bash(git*,pytest*,python*)" '
            '"Implement the user authentication module"',
    workdir="/opt/data/my-project",
    timeout=300,
)

# 场景：批量修复代码风格问题
terminal(
    command='codex "Fix all PEP8 violations in the src/ directory. '
            'Apply black formatting and isort imports. '
            'Do not modify test files."',
    workdir="/opt/data/my-project",
    timeout=180,
)

# 场景：批量生成类型注解
terminal(
    command='codex "Add type hints to all functions in src/services/. '
            'Use typing module for complex types. '
            'Run mypy after to verify."',
    workdir="/opt/data/my-project",
    timeout=240,
)

# 先安装 OpenAI SDK
terminal(command="pip install openai", timeout=60)

# 然后执行批量处理脚本
write_file(
    path="/opt/data/my-project/batch_fix.py",
    content="""
import asyncio
from openai import AsyncOpenAI
import os
import glob

async def batch_deprecation_fix():
    client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])

    # 找到所有需要修复的文件
    files = glob.glob("src/**/*.py", recursive=True)
    deprecated_pattern = "asyncio.get_event_loop()"

    fixed_count = 0
    for filepath in files:
        with open(filepath) as f:
            content = f.read()

        if deprecated_pattern not in content:
            continue

        # 调用 Codex 修复
        response = await client.responses.create(
            model="codex",
            input=f"Fix deprecated asyncio usage in:
{content}
Replace asyncio.get_event_loop() with asyncio.get_running_loop().
Return the full fixed file content.",
        )

        fixed_content = response.output_text
        if fixed_content != content:
            with open(filepath, 'w') as f:
                f.write(fixed_content)
            fixed_count += 1
            print(f"Fixed: {filepath}")

    print(f"Total files fixed: {fixed_count}")

asyncio.run(batch_deprecation_fix())
"""
)

# 执行批量处理脚本
terminal(
    command="python batch_fix.py",
    workdir="/opt/data/my-project",
    timeout=600,
)

# 场景：批量生成模块文档
terminal(
    command='opencode "Generate comprehensive docstrings for all '
            'classes and functions in src/models/. '
            'Use Google docstring format."',
    workdir="/opt/data/my-project",
    timeout=120,
)

# 场景：生成 README
terminal(
    command='opencode "Generate a README.md for this project. '
            'Include: project description, installation instructions, '
            'usage examples, API reference, and contribution guidelines. '
            "Analyze the project structure first."',
    workdir="/opt/data/my-project",
    timeout=180,
)

# 场景：PR 审查
terminal(
    command='opencode "Review the changes in the latest git commit. '
            'Focus on: security issues, performance concerns, and '
            'code quality. Provide specific line-by-line feedback."',
    workdir="/opt/data/my-project",
    timeout=120,
)

# 场景：代码复杂度分析
terminal(
    command='opencode "Analyze the cyclomatic complexity of all '
            'functions in src/. Flag any function with complexity > 10. '
            'Suggest refactoring for the most complex ones."',
    workdir="/opt/data/my-project",
    timeout=120,
)

# delegate_task 允许 Hermes Agent 将子任务委派给专门的 subagent
# 适用于需要不同上下文或专长的子任务

# 场景 1：委派代码生成任务
result = delegate_task(
    prompt="Generate a FastAPI CRUD API for the User model. "
           "Include: create, read, update, delete endpoints. "
           "Use SQLAlchemy async. Add request validation with Pydantic. "
           "Follow REST best practices.",
    # subagent 会自动获得独立的上下文和工具访问权限
)
# → 返回生成的代码和元数据

# 场景 2：委派测试编写
test_result = delegate_task(
    prompt="Write comprehensive unit tests for the User API. "
           "Cover: happy paths, edge cases, error handling, "
           "authentication failures. Use pytest and pytest-asyncio. "
           "Aim for > 90% code coverage.",
)

# 场景 3：委派文档编写
doc_result = delegate_task(
    prompt="Write API documentation for the User endpoints. "
           "Use OpenAPI/Swagger format. Include request/response "
           "examples, error codes, and authentication requirements.",
)

# 组合策略：delegate_task 负责智能决策，terminal 负责执行

class TaskOrchestrator:
    """Hermes 调度中心的任务编排器"""

    def __init__(self, project_path: str):
        self.project_path = project_path
        self.task_results = {}

    async def execute_feature_pipeline(self, feature_spec: str):
        """端到端的功能开发流水线"""

        # Step 1: delegate_task — 架构设计
        design = delegate_task(
            prompt=f"Design the architecture for this feature: {feature_spec}. "
                   f"Output: module structure, class diagrams (text), "
                   f"API contracts, data models.",
        )
        self.task_results["design"] = design

        # Step 2: terminal — 创建目录结构
        terminal(
            command=f"mkdir -p {self.project_path}/src/features/{{api,services,models}}",
            workdir=self.project_path,
        )

        # Step 3: delegate_task — Claude Code 实现核心逻辑
        implementation = delegate_task(
            prompt=f"Implement the feature based on this design: {design}. "
                   f"Use Claude Code. Generate all source files. "
                   f"Follow existing project patterns.",
        )
        self.task_results["implementation"] = implementation

        # Step 4: terminal — 执行 Claude Code 实现
        terminal(
            command=f'claude "{implementation.instructions}"',
            workdir=self.project_path,
            timeout=300,
        )

        # Step 5: delegate_task — Codex 批量测试生成
        test_plan = delegate_task(
            prompt="Generate a test plan for the implemented feature. "
                   "List all test cases needed.",
        )

        # Step 6: terminal — Codex 执行测试生成
        terminal(
            command=f'codex "{test_plan.test_generation_prompt}"',
            workdir=self.project_path,
            timeout=180,
        )

        # Step 7: terminal — 运行测试验证
        test_result = terminal(
            command="pytest tests/ -v --cov=src/features "
                    "--cov-report=term-missing",
            workdir=self.project_path,
            timeout=120,
        )
        self.task_results["test_result"] = test_result

        # Step 8: delegate_task — OpenCode 文档生成
        delegate_task(
            prompt="Generate documentation for the new feature module. "
                   "Include API reference, usage examples, and "
                   "integration guide.",
        )

        return self.task_results

# 完整场景：从 Bug 报告到修复提交的全自动化

class BugFixPipeline:
    """自动化 Bug 修复流水线"""

    def __init__(self, repo_path: str):
        self.repo = repo_path

    async def fix_bug(self, bug_report: dict) -> PipelineResult:
        """执行完整的 Bug 修复流程"""

        # === Phase 1: 根因分析 (Claude Code) ===
        analysis = terminal(
            command=f'claude "Analyze this bug report and find the root cause '
                    f'in the codebase: {bug_report["description"]}. '
                    f"Relevant module: {bug_report['module']}. "
                    f'Output: root cause analysis, affected files, fix strategy."',
            workdir=self.repo,
            timeout=300,
        )

        affected_files = self.parse_affected_files(analysis.output)

        # === Phase 2: 生成修复补丁 (delegate_task → Claude Code) ===
        fix_plan = delegate_task(
            prompt=f"Generate fix patches for these files: {affected_files}. "
                   f"Bug root cause: {analysis.output}. "
                   f"Use Claude Code to implement the fix.",
        )

        # === Phase 3: 应用修复 (terminal → Claude Code) ===
        terminal(
            command=f'claude "{fix_plan.fix_instructions}"',
            workdir=self.repo,
            timeout=300,
        )

        # === Phase 4: 批量测试验证 (terminal → pytest) ===
        test_result = terminal(
            command=f"pytest {bug_report['test_path']} -v "
                    f"--tb=short",
            workdir=self.repo,
            timeout=120,
        )

        if not test_result.success:
            # === Phase 4b: 修复测试 (delegate_task → Codex) ===
            delegate_task(
                prompt=f"Fix the failing tests. "
                       f"Test output: {test_result.output}. "
                       f"Use Codex for batch test fixes.",
            )

            # 重新运行测试
            test_result = terminal(
                command=f"pytest {bug_report['test_path']} -v --tb=short",
                workdir=self.repo,
                timeout=120,
            )

        # === Phase 5: 代码审查 (terminal → OpenCode) ===
        review = terminal(
            command='opencode "Review the latest changes. '
                    'Focus on: correctness, edge cases, performance. '
                    'Approve or request changes."',
            workdir=self.repo,
            timeout=120,
        )

        # === Phase 6: 提交 (terminal → git) ===
        if review.approved:
            terminal(
                command=f'git add -A && '
                        f'git commit -m "fix: {bug_report["title"]}\n\n'
                        f'{bug_report["description"]}"',
                workdir=self.repo,
            )

        return PipelineResult(
            status="fixed" if review.approved else "review_failed",
            analysis=analysis,
            fix_plan=fix_plan,
            test_result=test_result,
            review=review,
        )

# 场景：同时处理多个独立的子任务

async def parallel_code_review(pr_numbers: list[int]):
    """并发审查多个 Pull Request"""

    sessions = {}

    # 为每个 PR 启动独立的审查任务
    for pr_num in pr_numbers:
        session = terminal(
            command=f'opencode "Review PR #{pr_num}. '
                    f"Focus on security and correctness. "
                    f'Output a review summary."',
            workdir="/opt/data/my-project",
            background=True,
            timeout=180,
        )
        sessions[pr_num] = session["session_id"]

    # 等待所有审查完成
    results = {}
    for pr_num, session_id in sessions.items():
        process(action="wait", session_id=session_id, timeout=180)
        output = process(action="log", session_id=session_id)
        results[pr_num] = self.parse_review(output)

    # 汇总报告
    return self.generate_review_report(results)

class ResilientOrchestrator:
    """带容错能力的编排器"""

    MAX_RETRIES = 3
    RETRY_DELAY = 5  # seconds

    async def execute_with_retry(self, command: str, **kwargs) -> TerminalResult:
        """带重试的命令执行"""
        last_error = None

        for attempt in range(self.MAX_RETRIES):
            try:
                result = terminal(command=command, **kwargs)
                if result.exit_code == 0:
                    return result

                last_error = Exception(
                    f"Exit code {result.exit_code}: {result.output}"
                )
            except Exception as e:
                last_error = e

            if attempt < self.MAX_RETRIES - 1:
                await asyncio.sleep(self.RETRY_DELAY * (attempt + 1))
                # 可选：在重试前调整策略
                command = self.adjust_command(command, attempt)

        raise last_error

    def adjust_command(self, command: str, attempt: int) -> str:
        """根据失败次数调整命令策略"""
        if attempt == 1:
            # 第一次重试：增加超时
            return command + " --timeout=600"
        if attempt == 2:
            # 第二次重试：缩小范围
            return command.replace("src/", "src/core/")
        return command

    async def handle_process_failure(self, session_id: str) -> str:
        """处理后台进程失败"""
        # 获取失败日志
        logs = process(action="log", session_id=session_id)

        # 分析失败原因
        error_type = self.analyze_error(logs)

        if error_type == "timeout":
            # 超时：重新启动，增加 timeout
            return "restart_with_longer_timeout"
        elif error_type == "oom":
            # OOM：缩小任务范围
            return "reduce_scope"
        elif error_type == "permission":
            # 权限问题：调整权限后重试
            return "fix_permissions_and_retry"
        else:
            # 未知错误：委派给专家 Agent 分析
            return delegate_task(
                prompt=f"Analyze this process failure and suggest recovery: {logs}"
            )

# 链式编排：每个步骤的输出是下一步的输入

class ChainedOrchestrator:
    """链式任务编排"""

    def __init__(self):
        self.chain_state = {}

    async def run_chain(self, steps: list[ChainStep]):
        """执行编排链"""
        current_context = {}

        for step in steps:
            # 可选：条件跳过
            if step.condition and not step.condition(current_context):
                continue

            # 构建命令，注入上一步的输出
            command = self.render_command(step.command, current_context)

            # 选择执行方式
            if step.tool == "claude_code":
                result = terminal(command=f'claude "{command}"', **step.kwargs)
            elif step.tool == "codex":
                result = terminal(command=f'codex "{command}"', **step.kwargs)
            elif step.tool == "opencode":
                result = terminal(command=f'opencode "{command}"', **step.kwargs)
            elif step.tool == "delegate":
                result = delegate_task(prompt=command)
            elif step.tool == "shell":
                result = terminal(command=command, **step.kwargs)

            # 更新上下文
            current_context[step.output_key] = result
            self.chain_state[step.name] = result

            # 失败处理
            if step.fail_fast and result.get("exit_code", 0) != 0:
                raise ChainException(
                    f"Chain failed at step: {step.name}",
                    state=self.chain_state,
                )

        return self.chain_state

# 使用示例
chain = ChainedOrchestrator()
result = await chain.run_chain([
    ChainStep(
        name="analyze",
        tool="claude_code",
        command="Analyze the bug and identify root cause",
        output_key="analysis",
        fail_fast=True,
    ),
    ChainStep(
        name="fix",
        tool="codex",
        command="Generate fix based on: {analysis}",
        output_key="fix",
        fail_fast=True,
    ),
    ChainStep(
        name="test",
        tool="shell",
        command="pytest tests/ -v",
        output_key="test_result",
        fail_fast=True,
    ),
    ChainStep(
        name="review",
        tool="opencode",
        command="Review the changes",
        output_key="review",
        fail_fast=False,  # 审查失败不阻断
    ),
    ChainStep(
        name="commit",
        tool="shell",
        command="git add -A && git commit -m 'auto-fix'",
        output_key="commit_result",
        fail_fast=True,
        condition=lambda ctx: ctx.get("review", {}).get("approved", True),
    ),
])

class ObservableOrchestrator:
    """带监控的可观测编排器"""

    def __init__(self):
        self.metrics = {
            "total_tasks": 0,
            "successful_tasks": 0,
            "failed_tasks": 0,
            "total_duration": 0,
            "tool_usage": defaultdict(int),
        }
        self.event_log = []

    def record_event(self, event: dict):
        """记录编排事件"""
        self.event_log.append({
            "timestamp": datetime.now().isoformat(),
            **event,
        })

    def execute_monitored(self, tool: str, command: str, **kwargs):
        """带监控的执行"""
        start_time = time.time()
        self.metrics["total_tasks"] += 1
        self.metrics["tool_usage"][tool] += 1

        self.record_event({"type": "start", "tool": tool, "command": command})

        try:
            if tool == "delegate":
                result = delegate_task(prompt=command, **kwargs)
            else:
                result = terminal(command=command, **kwargs)

            duration = time.time() - start_time
            self.metrics["successful_tasks"] += 1
            self.metrics["total_duration"] += duration

            self.record_event({
                "type": "success",
                "tool": tool,
                "duration": duration,
            })

            return result

        except Exception as e:
            duration = time.time() - start_time
            self.metrics["failed_tasks"] += 1

            self.record_event({
                "type": "failure",
                "tool": tool,
                "error": str(e),
                "duration": duration,
            })

            raise

    def get_dashboard_data(self) -> dict:
        """生成仪表盘数据"""
        return {
            "metrics": self.metrics,
            "recent_events": self.event_log[-20:],
            "avg_duration": (
                self.metrics["total_duration"] /
                max(self.metrics["successful_tasks"], 1)
            ),
            "success_rate": (
                self.metrics["successful_tasks"] /
                max(self.metrics["total_tasks"], 1) * 100
            ),
        }