Agent Product = Model（智能） + Harness（環境）
                  │                  │
                  │                  ├── Tools（工具）
                  │                  ├── Knowledge（知識）
                  │                  ├── Observation（觀測）
                  │                  ├── Action Interfaces（行動介面）
                  │                  └── Permissions（權限）
                  │
                  └── 由訓練而得的感知、推理、行動能力

❌ 提示詞水管工「Agent」          ✅ 真正的 Agent
══════════════════════          ══════════════════
┌──────┐                       ┌──────────────┐
│ Node │→ if/else              │   Model      │
│ Graph│→ prompt chain         │  (trained)   │
│      │→ hardcoded routing    │              │
│      │→ rule trees           │  決策 + 推理  │
└──────┘                       └──────┬───────┘
                                      │
脆弱、不可擴展、                 ┌──────┴───────┐
無法泛化                        │   Harness    │
                               │  Tools +     │
                               │  Knowledge + │
                               │  Permissions │
                               └──────────────┘
                                  靈活、可組合、
                                  跨領域泛化

模型做決策    ←→    Harness 執行
模型做推理    ←→    Harness 提供上下文
模型是駕駛者  ←→    Harness 是載具

程式開發 Agent 的 Harness = IDE + 終端機 + 檔案系統
農業 Agent 的 Harness     = 感測器 + 灌溉控制 + 氣象資料
飯店 Agent 的 Harness     = 訂房系統 + 客戶通訊 + 設施管理 API

Agent（智能、決策者）永遠是模型。Harness 因領域而變。Agent 跨領域泛化。

Claude Code 本質：

  一個 Agent Loop
  + 工具（bash, read, write, edit, glob, grep, browser...）
  + 按需 Skill 載入
  + Context 壓縮
  + Subagent 派生
  + 帶相依圖的任務系統
  + 非同步郵箱的團隊協調
  + Worktree 隔離的平行執行
  + 權限治理

User Request
    │
    ▼
┌─────────────────────────────────────────┐
│              Agent Loop                  │
│  ┌──────────────────────────────────┐   │
│  │ 1. 組裝 messages[]               │   │
│  │ 2. 呼叫 LLM API                  │   │
│  │ 3. 檢查 stop_reason              │   │
│  │    ├─ "tool_use" → 執行工具       │   │
│  │    │   ├─ 取得 tool_result        │   │
│  │    │   └─ 加入 messages[] → 回到 1│   │
│  │    └─ "end_turn" → 回傳結果       │   │
│  └──────────────────────────────────┘   │
└─────────────────────────────────────────┘
    │
    ▼
Response to User

# 抽象 LLM Gateway 設計
import os
from anthropic import Anthropic

# 基本配置（支援替換 base_url）
client = Anthropic(
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    base_url=os.getenv("ANTHROPIC_BASE_URL", "https://api.anthropic.com"),
)

MODEL = os.getenv("MODEL", "claude-sonnet-4-6")

┌───────────────────────────────────┐
│          LLM Gateway              │
│  ┌─────────────────────────────┐  │
│  │   Model Router              │  │
│  │   ├─ 簡單任務 → DeepSeek     │  │
│  │   ├─ 複雜任務 → Claude       │  │
│  │   ├─ 程式碼 → GPT-4o        │  │
│  │   └─ 本地機密 → 私有模型     │  │
│  ├─────────────────────────────┤  │
│  │   Rate Limiter              │  │
│  │   Token Counter             │  │
│  │   Cost Tracker              │  │
│  │   Retry / Fallback          │  │
│  └─────────────────────────────┘  │
└───────────────────────────────────┘

# Tool 定義（JSON Schema 格式）
TOOLS = [
    {
        "name": "bash",
        "description": "執行 shell 命令。用於編譯、測試、Git 操作等。",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "要執行的 shell 命令"
                }
            },
            "required": ["command"]
        }
    },
    {
        "name": "read_file",
        "description": "讀取檔案內容",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "檔案路徑"},
                "start_line": {"type": "integer", "description": "起始行號"},
                "end_line": {"type": "integer", "description": "結束行號"}
            },
            "required": ["path"]
        }
    },
    {
        "name": "write_file",
        "description": "寫入檔案",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "檔案路徑"},
                "content": {"type": "string", "description": "檔案內容"}
            },
            "required": ["path", "content"]
        }
    }
]

# Dispatch Map（名稱 → 處理函式）
TOOL_HANDLERS = {
    "bash":       handle_bash,
    "read_file":  handle_read_file,
    "write_file": handle_write_file,
    "edit_file":  handle_edit_file,
    "glob":       handle_glob,
    "grep":       handle_grep,
    "todo_write": handle_todo_write,
    "subagent":   handle_subagent,
}

1. 定義 Tool Schema（JSON Schema）
2. 實作 Handler 函式
3. 註冊到 TOOL_HANDLERS dict
4. Agent Loop 無需修改 — 自動 dispatch

# Task 資料結構
{
    "id": "task-001",
    "title": "實作使用者登入 API",
    "status": "in_progress",       # todo / in_progress / done / blocked
    "assignee": "backend-agent",
    "dependencies": ["task-000"],   # 相依任務 ID
    "worktree": "/worktrees/task-001",
    "created_at": "2026-05-02T10:00:00Z",
    "updated_at": "2026-05-02T10:30:00Z"
}

Task 相依圖範例：

task-001 (DB Schema)
    │
    ├── task-002 (Entity 實作) ──┐
    │                            ├── task-004 (API 整合測試)
    └── task-003 (API 實作)  ────┘
                                      │
                                 task-005 (部署)

專案根目錄/
├── .git/                      # 共用 Git 倉庫
├── main-workspace/            # 主工作區
├── worktrees/
│   ├── task-001/              # Task 1 獨立工作區
│   │   ├── src/
│   │   └── ...
│   ├── task-002/              # Task 2 獨立工作區
│   │   ├── src/
│   │   └── ...
│   └── task-003/              # Task 3 獨立工作區

# 建立隔離的 Worktree
git worktree add worktrees/task-001 -b feature/task-001

# 在隔離環境中執行任務
cd worktrees/task-001
# Agent 在此目錄中讀寫檔案、執行測試

# 完成後合併回主分支
git checkout main
git merge feature/task-001
git worktree remove worktrees/task-001

┌─────────────────────────────────────────┐
│              Memory 架構                 │
│                                         │
│  ┌─────────────┐   ┌─────────────────┐  │
│  │  短期記憶    │   │  長期記憶        │  │
│  │  (Context)  │   │  (Persistent)   │  │
│  │             │   │                 │  │
│  │ • messages[]│   │ • Skills 文件    │  │
│  │ • Tool 結果  │   │ • Task 歷史      │  │
│  │ • 當前 Todo  │   │ • 架構決策紀錄   │  │
│  │             │   │ • 錯誤修復經驗   │  │
│  └──────┬──────┘   └────────┬────────┘  │
│         │                    │           │
│         ▼                    ▼           │
│  Context Compact      On-Demand Load    │
│  (三層壓縮)           (tool_result 注入) │
└─────────────────────────────────────────┘

{"from": "lead", "to": "backend", "type": "request", "task_id": "task-002", "content": "實作 UserService.login() 方法", "timestamp": "2026-05-02T10:00:00Z"}
{"from": "backend", "to": "lead", "type": "response", "task_id": "task-002", "content": "已完成，通過 5 個單元測試", "timestamp": "2026-05-02T10:15:00Z"}
{"from": "lead", "to": "qa", "type": "request", "task_id": "task-003", "content": "對 UserService.login() 進行整合測試", "timestamp": "2026-05-02T10:16:00Z"}

import os
from anthropic import Anthropic

client = Anthropic()
MODEL = os.getenv("MODEL", "claude-sonnet-4-6")

SYSTEM = "你是一位資深軟體工程師。使用 bash 工具執行命令。"

TOOLS = [{
    "name": "bash",
    "description": "執行 shell 命令",
    "input_schema": {
        "type": "object",
        "properties": {
            "command": {"type": "string", "description": "Shell 命令"}
        },
        "required": ["command"]
    }
}]

def handle_bash(command: str) -> str:
    """安全執行 shell 命令"""
    import subprocess
    try:
        result = subprocess.run(
            command, shell=True, capture_output=True,
            text=True, timeout=30, cwd=os.getcwd()
        )
        output = result.stdout + result.stderr
        return output[:10000]  # 限制輸出長度
    except subprocess.TimeoutExpired:
        return "ERROR: 命令執行超時（30 秒）"
    except Exception as e:
        return f"ERROR: {e}"

TOOL_HANDLERS = {"bash": handle_bash}

def agent_loop(user_message: str):
    """核心 Agent 迴圈"""
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = client.messages.create(
            model=MODEL,
            system=SYSTEM,
            max_tokens=4096,
            messages=messages,
            tools=TOOLS,
        )
        
        # 將 assistant 回覆加入歷史
        messages.append({
            "role": "assistant",
            "content": response.content
        })
        
        # 如果模型決定停止（不再呼叫工具），結束迴圈
        if response.stop_reason != "tool_use":
            # 印出最終文字回覆
            for block in response.content:
                if hasattr(block, "text"):
                    print(block.text)
            return
        
        # 執行模型請求的所有工具
        results = []
        for block in response.content:
            if block.type == "tool_use":
                print(f"[Tool] {block.name}: {block.input}")
                output = TOOL_HANDLERS[block.name](**block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        
        # 將工具結果作為 user 訊息送回
        messages.append({"role": "user", "content": results})

# 使用
if __name__ == "__main__":
    agent_loop("列出當前目錄下的所有 Python 檔案，並統計行數")

User: "列出所有 Python 檔案"
    │
    ▼
Agent Loop 開始
    │
    ▼
LLM 決定呼叫 bash tool
    │
    ▼
執行: find . -name "*.py" | wc -l
    │
    ▼
tool_result: "42"
    │
    ▼
LLM 收到結果，決定再呼叫 bash
    │
    ▼
執行: find . -name "*.py" -exec wc -l {} +
    │
    ▼
tool_result: "共 3,247 行"
    │
    ▼
LLM 決定停止 (stop_reason = "end_turn")
    │
    ▼
回覆: "找到 42 個 Python 檔案，共 3,247 行程式碼"

# 新增 read_file 和 write_file 工具
def handle_read_file(path: str, start_line: int = None, end_line: int = None) -> str:
    """讀取檔案，支援行號範圍"""
    try:
        with open(path, "r", encoding="utf-8") as f:
            lines = f.readlines()
        if start_line and end_line:
            lines = lines[start_line - 1:end_line]
        return "".join(lines)[:10000]
    except FileNotFoundError:
        return f"ERROR: 檔案不存在: {path}"
    except Exception as e:
        return f"ERROR: {e}"

def handle_write_file(path: str, content: str) -> str:
    """寫入檔案（含目錄自動建立）"""
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return f"OK: 已寫入 {path}（{len(content)} 字元）"

def handle_edit_file(path: str, old_string: str, new_string: str) -> str:
    """精確替換檔案中的字串"""
    with open(path, "r", encoding="utf-8") as f:
        content = f.read()
    if old_string not in content:
        return f"ERROR: 找不到要替換的字串"
    if content.count(old_string) > 1:
        return f"ERROR: 找到多個匹配，請提供更多上下文"
    new_content = content.replace(old_string, new_string, 1)
    with open(path, "w", encoding="utf-8") as f:
        f.write(new_content)
    return f"OK: 已替換"

# 擴展 Dispatch Map — 迴圈完全不變
TOOL_HANDLERS = {
    "bash":       handle_bash,
    "read_file":  handle_read_file,
    "write_file": handle_write_file,
    "edit_file":  handle_edit_file,
}

class PermissionGateway:
    """三層防護：拒絕清單 → 規則匹配 → 人工審批"""
    
    # 第一層：硬性拒絕清單（永遠不允許）
    DENY_LIST = [
        r"rm\s+-rf\s+/",          # 遞迴刪除根目錄
        r"DROP\s+DATABASE",        # 刪除資料庫
        r"curl.*\|\s*bash",        # 遠端程式碼執行
        r"chmod\s+777\s+/",        # 授予全域寫入
        r">\s*/etc/passwd",        # 覆蓋系統檔案
    ]
    
    # 第二層：規則引擎（根據上下文決定）
    RULE_ENGINE = [
        # 格式：(pattern, condition, verdict)
        (r"git push",        "branch == 'main'",    "REQUIRE_APPROVAL"),
        (r"kubectl apply",   "namespace == 'prod'", "REQUIRE_APPROVAL"),
        (r"\.env",           "operation == 'write'","DENY"),
        (r"secrets/",        "any",                 "DENY"),
    ]
    
    def check(self, tool_name: str, tool_input: dict, context: dict) -> tuple:
        """三閘門權限檢查，回傳 (allowed: bool, reason: str)"""
        
        # Gate 1: 硬性拒絕清單
        cmd = tool_input.get("command", "") or tool_input.get("path", "")
        for pattern in self.DENY_LIST:
            if re.search(pattern, cmd, re.IGNORECASE):
                return False, f"[DENY_LIST] 禁止操作: {pattern}"
        
        # Gate 2: 規則引擎
        for pattern, condition, verdict in self.RULE_ENGINE:
            if re.search(pattern, cmd, re.IGNORECASE):
                if self._eval_condition(condition, context):
                    if verdict == "DENY":
                        return False, f"[RULE] 規則禁止: {condition}"
                    elif verdict == "REQUIRE_APPROVAL":
                        return self._request_human_approval(cmd)
        
        # Gate 3: 預設允許（白名單外需人工確認）
        return True, "OK"
    
    def _request_human_approval(self, command: str) -> tuple:
        """非同步人工審批"""
        print(f"\n⚠️ 需要審批：{command}")
        decision = input("允許此操作？[y/n]: ").strip().lower()
        return (True, "已核准") if decision == "y" else (False, "已拒絕")

使用者 Prompt
    │
    ▼
┌─────────────────────────────────────────┐
│  Gate 1: Deny List（硬性拒絕）            │
│  • 硬編碼的永久禁止清單                   │
│  • 不可繞過，不可設定                     │
│  匹配 → BLOCKED（立即）                  │
└─────────────┬───────────────────────────┘
              │ 通過
              ▼
┌─────────────────────────────────────────┐
│  Gate 2: Rule Engine（情境規則）          │
│  • 根據上下文（分支、環境、操作類型）     │
│  • 可設定、可擴展                         │
│  匹配 → DENY 或 REQUIRE_APPROVAL         │
└─────────────┬───────────────────────────┘
              │ 通過
              ▼
┌─────────────────────────────────────────┐
│  Gate 3: Human Approval（人工審批）       │
│  • 高風險操作需人工確認                   │
│  • 非同步等待決策                         │
│  批准 → ALLOWED / 拒絕 → BLOCKED         │
└─────────────────────────────────────────┘

from typing import Callable, Optional
from dataclasses import dataclass

@dataclass
class HookContext:
    """Hook 執行上下文"""
    tool_name: str
    tool_input: dict
    tool_result: Optional[str] = None
    agent_name: str = "main"
    session_id: str = ""

class HookSystem:
    """Agent Loop 的 Hooks 管理器"""
    
    def __init__(self):
        self._hooks: dict[str, list[Callable]] = {
            "UserPromptSubmit":  [],   # 使用者提交 Prompt 後
            "PreToolUse":        [],   # Tool 執行前
            "PostToolUse":       [],   # Tool 執行後
            "Stop":              [],   # Agent 正常停止時
            "SubagentStop":      [],   # Subagent 停止時
        }
    
    def register(self, event: str, handler: Callable):
        """註冊 Hook Handler"""
        if event not in self._hooks:
            raise ValueError(f"未知 Hook 事件: {event}")
        self._hooks[event].append(handler)
    
    def fire(self, event: str, ctx: HookContext) -> Optional[str]:
        """觸發 Hook，回傳 None 表示繼續，回傳字串表示覆寫結果"""
        for handler in self._hooks.get(event, []):
            result = handler(ctx)
            if result is not None:
                return result  # 第一個非 None 結果覆寫後續
        return None

# === 企業級 Hook 範例 ===

# 1. PreToolUse：自動安全掃描
def security_scan_hook(ctx: HookContext) -> Optional[str]:
    """在 bash 工具執行前掃描危險命令"""
    if ctx.tool_name == "bash":
        cmd = ctx.tool_input.get("command", "")
        if "DROP TABLE" in cmd.upper():
            return "HOOK_BLOCKED: 偵測到 DROP TABLE，已阻止執行"
    return None  # 允許繼續

# 2. PostToolUse：成本追蹤
def cost_tracking_hook(ctx: HookContext):
    """在每次 Tool 執行後記錄成本"""
    import json, time
    entry = {
        "timestamp": time.time(),
        "tool": ctx.tool_name,
        "agent": ctx.agent_name,
        "result_len": len(ctx.tool_result or ""),
    }
    with open("tool_usage.jsonl", "a") as f:
        f.write(json.dumps(entry) + "\n")

# 3. Stop：任務完成通知
def notification_hook(ctx: HookContext):
    """Agent 完成任務時發送 Slack 通知"""
    import subprocess
    subprocess.run([
        "curl", "-X", "POST", "-H", "Content-type: application/json",
        "--data", f'{{"text": "Agent [{ctx.agent_name}] 已完成任務"}}',
        os.getenv("SLACK_WEBHOOK_URL", "")
    ], capture_output=True)

# === 整合到 Agent Loop ===
hooks = HookSystem()
hooks.register("PreToolUse",  security_scan_hook)
hooks.register("PostToolUse", cost_tracking_hook)
hooks.register("Stop",        notification_hook)

def agent_loop_with_hooks(user_message: str):
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM,
            max_tokens=4096, messages=messages, tools=TOOLS,
        )
        messages.append({"role": "assistant", "content": response.content})
        
        if response.stop_reason != "tool_use":
            # 觸發 Stop Hook
            hooks.fire("Stop", HookContext(tool_name="", tool_input={}))
            break
        
        results = []
        for block in response.content:
            if block.type == "tool_use":
                ctx = HookContext(tool_name=block.name, tool_input=block.input)
                
                # 觸發 PreToolUse Hook
                blocked = hooks.fire("PreToolUse", ctx)
                if blocked:
                    output = blocked  # Hook 阻止執行
                else:
                    output = TOOL_HANDLERS[block.name](**block.input)
                
                # 觸發 PostToolUse Hook
                ctx.tool_result = output
                hooks.fire("PostToolUse", ctx)
                
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        
        messages.append({"role": "user", "content": results})

class TodoManager:
    """管理 Agent 的任務計畫"""
    
    def __init__(self):
        self.todos = []  # [{"id": int, "task": str, "status": str}]
    
    def write(self, todos: list) -> str:
        """建立 / 更新任務清單"""
        self.todos = todos
        return self._format()
    
    def update(self, todo_id: int, status: str) -> str:
        """更新單一任務狀態"""
        for todo in self.todos:
            if todo["id"] == todo_id:
                todo["status"] = status
                return self._format()
        return f"ERROR: 找不到 ID {todo_id}"
    
    def _format(self) -> str:
        """格式化輸出，方便 LLM 閱讀"""
        lines = ["=== 任務清單 ==="]
        for t in self.todos:
            icon = {"todo": "⬜", "in_progress": "🔄", "done": "✅"}
            lines.append(f"{icon.get(t['status'], '?')} [{t['id']}] {t['task']}")
        return "\n".join(lines)

# 關鍵機制：Nag Reminder
# 在每次 tool_result 後附加未完成任務提醒
def inject_nag_reminder(results: list, todo_manager: TodoManager) -> list:
    """注入任務提醒，防止 Agent 偏離計畫"""
    pending = [t for t in todo_manager.todos if t["status"] != "done"]
    if pending:
        reminder = "\n⚠️ 待完成任務：\n"
        for t in pending:
            reminder += f"  - [{t['id']}] {t['task']}\n"
        reminder += "請繼續執行下一個待完成任務。"
        results.append({
            "type": "tool_result",
            "tool_use_id": "nag",
            "content": reminder,
        })
    return results

def handle_subagent(task: str, allowed_tools: list = None) -> str:
    """啟動子 Agent，使用獨立的對話歷史"""
    
    # 關鍵：子 Agent 有全新的 messages[]
    sub_messages = [{"role": "user", "content": task}]
    
    # 可選：限制子 Agent 可用的工具集
    sub_tools = TOOLS if not allowed_tools else [
        t for t in TOOLS if t["name"] in allowed_tools
    ]
    
    # 子 Agent 獨立迴圈
    while True:
        response = client.messages.create(
            model=MODEL,
            system="你是專門處理子任務的助手。完成後簡潔回報結果。",
            max_tokens=4096,
            messages=sub_messages,
            tools=sub_tools,
        )
        
        sub_messages.append({
            "role": "assistant",
            "content": response.content
        })
        
        if response.stop_reason != "tool_use":
            # 提取子 Agent 的最終回覆
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return "子任務完成（無文字回覆）"
        
        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = TOOL_HANDLERS[block.name](**block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        sub_messages.append({"role": "user", "content": results})

主 Agent Context                    子 Agent Context
┌─────────────────┐                ┌──────────────────┐
│ 對話歷史 (大)    │                │ 全新 messages[]   │
│ 多個任務記錄     │    spawn       │ 僅包含子任務      │
│ 各種 Tool 結果   │ ──────────►    │ 乾淨的 Context    │
│                 │                │ 專注完成一件事    │
│  ◄──── 結果摘要 ─┤                │                  │
└─────────────────┘                └──────────────────┘

優點：
• 子任務的噪音不會汙染主 Context
• 失敗的子任務可獨立重試
• 不同子任務可用不同模型 / 權限

# Skills 目錄結構
# skills/
# ├── spring-boot/
# │   └── SKILL.md          # Spring Boot 開發指南
# ├── react/
# │   └── SKILL.md          # React 開發指南
# ├── database-migration/
# │   └── SKILL.md          # DB Migration 指南
# └── security-review/
#     └── SKILL.md          # 安全審查指南

def handle_load_skill(skill_name: str) -> str:
    """按需載入知識文件"""
    skill_path = f"skills/{skill_name}/SKILL.md"
    if not os.path.exists(skill_path):
        available = os.listdir("skills")
        return f"ERROR: 找不到 skill '{skill_name}'。可用: {available}"
    
    with open(skill_path, "r", encoding="utf-8") as f:
        content = f.read()
    
    return f"=== Skill: {skill_name} ===\n{content}"

<!-- skills/spring-boot-upgrade/SKILL.md -->
# Spring Boot 升級指南

## 從 Spring Boot 2.x 升級到 3.x

### 前置檢查
1. 確認 Java 版本 >= 17
2. 檢查所有依賴的 Jakarta EE 相容性
3. 掃描 `javax.*` → `jakarta.*` 的 import 變更

### 升級步驟
1. 更新 parent pom 版本
2. 批次替換 javax → jakarta
3. 更新 Spring Security 配置
4. 執行完整測試套件

### 常見問題
- Hibernate 6.x 的 HQL 語法變更
- Spring Security 的 SecurityFilterChain 新寫法
- Actuator endpoint 路徑變更

def compact_context(messages: list, model: str = MODEL) -> list:
    """三層壓縮策略"""
    
    total_tokens = estimate_tokens(messages)
    max_tokens = 200000  # Claude 的 context window
    
    if total_tokens < max_tokens * 0.5:
        return messages  # 還不需要壓縮
    
    # L1：摘要舊對話（保留最近 10 輪）
    if total_tokens < max_tokens * 0.7:
        old_messages = messages[:-20]  # 舊的部分
        recent_messages = messages[-20:]  # 近期保留
        
        summary = client.messages.create(
            model=model,
            system="請摘要以下對話的關鍵資訊，保留所有重要的決策和結論。",
            max_tokens=2000,
            messages=[{
                "role": "user",
                "content": f"請摘要：\n{format_messages(old_messages)}"
            }]
        )
        
        return [{
            "role": "user",
            "content": f"[對話摘要]\n{summary.content[0].text}"
        }] + recent_messages
    
    # L2：移除工具執行細節，僅保留結論
    # L3：僅保留當前任務描述 + 最近結果
    # ...（遞進壓縮）

class MemorySystem:
    """三層記憶架構：選取 → 擷取 → 整合"""
    
    def __init__(self, memory_file: str = ".agent_memory.json"):
        self.memory_file = memory_file
        self.memories = self._load()
    
    # --- 第一層：選取（Selection）---
    # 決定「哪些資訊值得記住」
    MEMORY_TRIGGERS = [
        "使用者偏好",
        "架構決策",
        "已知錯誤和修復方式",
        "環境配置",
        "重要約束條件",
    ]
    
    def should_remember(self, content: str) -> bool:
        """判斷此內容是否值得記住"""
        return any(trigger in content for trigger in self.MEMORY_TRIGGERS)
    
    # --- 第二層：擷取（Extraction）---
    # 決定「以何種形式儲存」
    def extract_and_store(self, content: str, category: str):
        """擷取關鍵資訊並結構化存儲"""
        entry = {
            "category": category,          # 類別
            "content": content[:500],       # 內容摘要
            "created_at": datetime.now().isoformat(),
            "access_count": 0,             # 存取次數（用於遺忘策略）
        }
        self.memories.setdefault(category, []).append(entry)
        self._save()
    
    # --- 第三層：整合（Consolidation）---
    # 定期合併相似記憶，防止記憶庫膨脹
    def consolidate(self):
        """整合相似記憶，保持記憶庫精簡"""
        for category, entries in self.memories.items():
            if len(entries) > 10:
                # 使用 LLM 摘要並合併
                summary_prompt = f"請整合以下 {category} 類別的記憶，保留關鍵資訊：\n"
                for e in entries:
                    summary_prompt += f"- {e['content']}\n"
                
                consolidated = client.messages.create(
                    model=MODEL,
                    system="你是記憶整合助手，請將多條記憶合併為簡潔的摘要。",
                    max_tokens=500,
                    messages=[{"role": "user", "content": summary_prompt}]
                ).content[0].text
                
                self.memories[category] = [{
                    "category": category,
                    "content": consolidated,
                    "created_at": datetime.now().isoformat(),
                    "access_count": 0,
                }]
        self._save()
    
    def recall(self, query: str) -> str:
        """從記憶庫中取回相關記憶"""
        all_memories = []
        for entries in self.memories.values():
            all_memories.extend(entries)
        
        if not all_memories:
            return ""
        
        # 簡單關鍵字匹配（生產環境可用向量搜尋）
        relevant = [m for m in all_memories if any(w in m["content"] for w in query.split())]
        
        if not relevant:
            return ""
        
        result = "=== 相關記憶 ===\n"
        for m in relevant[:5]:
            result += f"[{m['category']}] {m['content']}\n"
        return result

新資訊輸入
    │
    ▼
Selection（值得記住嗎？）
    │ 是
    ▼
Extraction（以何種結構儲存？）
    │
    ▼
Storage（存入 .agent_memory.json）
    │
    ▼                         ← 定期執行
Consolidation（記憶超過 10 條時整合）
    │
    ▼
Recall（按需從記憶庫取回注入 Context）

class SystemPromptBuilder:
    """動態 System Prompt 組裝器"""
    
    def __init__(self):
        self.sections = {
            "identity":    "",   # 角色定義（固定）
            "tools":       "",   # 可用工具說明（半固定）
            "context":     "",   # 當前任務上下文（動態）
            "skills":      "",   # 按需載入的技能（動態）
            "memory":      "",   # 長期記憶摘要（動態）
            "constraints": "",   # 當前約束條件（動態）
        }
    
    def set_identity(self, role: str, expertise: list):
        """設定角色身份（通常固定）"""
        self.sections["identity"] = f"""你是 {role}。
專業領域：{', '.join(expertise)}。
你的目標是高效、安全地完成使用者交付的任務。"""
    
    def set_active_skills(self, skill_names: list):
        """注入當前激活的技能知識"""
        skill_content = []
        for name in skill_names:
            path = f"skills/{name}/SKILL.md"
            if os.path.exists(path):
                with open(path) as f:
                    skill_content.append(f"## Skill: {name}\n{f.read()[:2000]}")
        
        self.sections["skills"] = "\n\n".join(skill_content)
    
    def inject_memory(self, memory: MemorySystem, current_task: str):
        """注入相關記憶"""
        recalled = memory.recall(current_task)
        self.sections["memory"] = recalled
    
    def set_task_context(self, task_desc: str, progress: str):
        """設定當前任務上下文"""
        self.sections["context"] = f"""## 當前任務
{task_desc}

## 進度
{progress}"""
    
    def build(self) -> str:
        """組裝最終 System Prompt"""
        parts = []
        
        # 固定部分（永遠包含）
        if self.sections["identity"]:
            parts.append(self.sections["identity"])
        if self.sections["tools"]:
            parts.append(self.sections["tools"])
        
        # 動態部分（非空時包含）
        for section in ["context", "skills", "memory", "constraints"]:
            if self.sections[section]:
                parts.append(self.sections[section])
        
        return "\n\n---\n\n".join(parts)

# 使用範例
builder = SystemPromptBuilder()
builder.set_identity("資深 Java 工程師", ["Spring Boot", "Microservices", "TDD"])
builder.set_active_skills(["spring-boot-3", "security-checklist"])
builder.inject_memory(memory_system, current_task="實作使用者登入 API")
builder.set_task_context("實作 JWT 認證模組", "已完成 Entity 層，正在實作 Service 層")

DYNAMIC_SYSTEM = builder.build()

import json
from datetime import datetime

class TaskManager:
    """檔案式任務管理器，支援相依圖"""
    
    TASK_FILE = ".tasks.jsonl"
    
    def create(self, title: str, deps: list = None) -> dict:
        task = {
            "id": f"task-{self._next_id():03d}",
            "title": title,
            "status": "todo",
            "dependencies": deps or [],
            "created_at": datetime.now().isoformat(),
        }
        self._append(task)
        return task
    
    def update_status(self, task_id: str, status: str) -> dict:
        tasks = self._load_all()
        for t in tasks:
            if t["id"] == task_id:
                # 檢查相依任務是否都已完成
                if status == "in_progress":
                    for dep_id in t.get("dependencies", []):
                        dep = next((x for x in tasks if x["id"] == dep_id), None)
                        if dep and dep["status"] != "done":
                            return {"error": f"相依任務 {dep_id} 尚未完成"}
                t["status"] = status
                t["updated_at"] = datetime.now().isoformat()
                self._save_all(tasks)
                return t
        return {"error": f"找不到任務 {task_id}"}
    
    def get_ready_tasks(self) -> list:
        """取得所有相依已滿足、可執行的任務"""
        tasks = self._load_all()
        done_ids = {t["id"] for t in tasks if t["status"] == "done"}
        return [
            t for t in tasks
            if t["status"] == "todo"
            and all(d in done_ids for d in t.get("dependencies", []))
        ]

import threading
import queue

class BackgroundRunner:
    """背景任務執行器"""
    
    def __init__(self):
        self.notify_queue = queue.Queue()
    
    def run_in_background(self, task_id: str, command: str):
        """在 daemon thread 中執行命令"""
        def _run():
            import subprocess
            result = subprocess.run(
                command, shell=True,
                capture_output=True, text=True
            )
            self.notify_queue.put({
                "task_id": task_id,
                "command": command,
                "stdout": result.stdout,
                "stderr": result.stderr,
                "returncode": result.returncode,
            })
        
        thread = threading.Thread(target=_run, daemon=True)
        thread.start()
        return f"背景任務 {task_id} 已啟動: {command}"
    
    def check_notifications(self) -> list:
        """檢查已完成的背景任務"""
        notifications = []
        while not self.notify_queue.empty():
            notifications.append(self.notify_queue.get_nowait())
        return notifications

Agent 主迴圈
    │
    ├── 啟動背景任務：mvn test（耗時 2 分鐘）
    │       └── daemon thread 執行中...
    │
    ├── 同時繼續：撰寫下一個類別
    │
    ├── 檢查通知：測試完成！3 個失敗
    │
    └── 修復失敗的測試

import json
import time
import threading
from datetime import datetime, timedelta

class CronJob:
    """持久化排程任務定義"""
    def __init__(self, job_id: str, cron_expr: str, task: str, enabled: bool = True):
        self.id = job_id
        self.cron_expr = cron_expr   # "0 2 * * *" = 每天凌晨 2 點
        self.task = task             # 要執行的任務描述
        self.enabled = enabled
        self.last_run = None
        self.next_run = None

class CronScheduler:
    """耐久性排程器：基於檔案的任務佇列 + 輪詢執行"""
    
    CRON_FILE = ".cron_jobs.json"
    POLL_INTERVAL = 30  # 每 30 秒檢查一次
    
    def __init__(self):
        self.jobs = self._load_jobs()
        self._running = False
    
    def add_job(self, cron_expr: str, task: str) -> str:
        """新增排程任務"""
        job_id = f"cron-{int(time.time())}"
        job = {
            "id": job_id,
            "cron_expr": cron_expr,
            "task": task,
            "enabled": True,
            "created_at": datetime.now().isoformat(),
            "last_run": None,
            "next_run": self._calc_next_run(cron_expr),
        }
        self.jobs[job_id] = job
        self._save_jobs()
        return job_id
    
    def remove_job(self, job_id: str):
        """移除排程任務"""
        if job_id in self.jobs:
            del self.jobs[job_id]
            self._save_jobs()
    
    def start(self, agent_executor):
        """啟動排程輪詢 daemon"""
        self._running = True
        
        def poll_loop():
            while self._running:
                now = datetime.now().isoformat()
                for job_id, job in list(self.jobs.items()):
                    if not job.get("enabled"):
                        continue
                    if job.get("next_run") and job["next_run"] <= now:
                        print(f"[Cron] 觸發排程任務: {job['task']}")
                        
                        # 在背景執行 Agent 任務
                        threading.Thread(
                            target=agent_executor,
                            args=(job["task"],),
                            daemon=True
                        ).start()
                        
                        # 更新執行記錄
                        job["last_run"] = now
                        job["next_run"] = self._calc_next_run(job["cron_expr"])
                        self._save_jobs()
                
                time.sleep(self.POLL_INTERVAL)
        
        thread = threading.Thread(target=poll_loop, daemon=True)
        thread.start()
        return thread
    
    def _calc_next_run(self, cron_expr: str) -> str:
        """解析 cron 表達式，計算下次執行時間（簡化版）"""
        # 生產環境建議使用 croniter 套件
        # 這裡用簡單的 daily 示意
        next_run = datetime.now() + timedelta(hours=24)
        return next_run.isoformat()
    
    def _load_jobs(self) -> dict:
        if os.path.exists(self.CRON_FILE):
            with open(self.CRON_FILE) as f:
                return json.load(f)
        return {}
    
    def _save_jobs(self):
        with open(self.CRON_FILE, "w") as f:
            json.dump(self.jobs, f, indent=2, ensure_ascii=False)

# 企業應用範例
scheduler = CronScheduler()

# 每天凌晨 2 點掃描安全漏洞
scheduler.add_job("0 2 * * *", "對整個程式碼庫執行 SAST 掃描，產出安全報告")

# 每週一早上 9 點產生技術債報告
scheduler.add_job("0 9 * * 1", "分析所有 TODO/FIXME 評論，產出技術債報告")

# 每小時檢查 CI 狀態
scheduler.add_job("0 * * * *", "檢查最近 1 小時內失敗的 CI 建置，並嘗試自動修復")

# 啟動排程器
scheduler.start(agent_executor=lambda task: agent_loop(task))

class AgentTeam:
    """Agent 團隊管理器"""
    
    def __init__(self, team_dir: str = ".team"):
        self.team_dir = team_dir
        os.makedirs(team_dir, exist_ok=True)
    
    def register_teammate(self, name: str, role: str, tools: list):
        """註冊隊友"""
        config = {
            "name": name,
            "role": role,
            "allowed_tools": tools,
            "status": "idle",
        }
        with open(f"{self.team_dir}/{name}.json", "w") as f:
            json.dump(config, f, indent=2)
    
    def send_message(self, to: str, message: dict):
        """透過 JSONL Mailbox 發送訊息"""
        mailbox = f"{self.team_dir}/{to}.mailbox.jsonl"
        message["timestamp"] = datetime.now().isoformat()
        with open(mailbox, "a") as f:
            f.write(json.dumps(message, ensure_ascii=False) + "\n")
    
    def read_messages(self, agent_name: str) -> list:
        """讀取信箱中的訊息"""
        mailbox = f"{self.team_dir}/{agent_name}.mailbox.jsonl"
        if not os.path.exists(mailbox):
            return []
        with open(mailbox, "r") as f:
            return [json.loads(line) for line in f if line.strip()]

Lead Agent                     Backend Agent
    │                              │
    ├─── REQUEST ──────────────►   │
    │   {type: "request",          │
    │    task: "實作 login API"}    │
    │                              │
    │                              ├── 讀取程式碼
    │                              ├── 實作功能
    │                              ├── 執行測試
    │                              │
    │   ◄──────── RESPONSE ────────┤
    │   {type: "response",         │
    │    status: "done",           │
    │    summary: "完成，5 測試通過"}│
    │                              │

def autonomous_cycle(agent_name: str, team: AgentTeam, task_mgr: TaskManager):
    """自治 Agent 的 idle cycle"""
    while True:
        # 1. 檢查信箱是否有直接指派的任務
        messages = team.read_messages(agent_name)
        if messages:
            for msg in messages:
                if msg["type"] == "request":
                    execute_task(msg)
                    team.send_message(msg["from"], {
                        "type": "response",
                        "task_id": msg["task_id"],
                        "status": "done"
                    })
            continue
        
        # 2. 主動掃描任務板，認領可執行的任務
        ready_tasks = task_mgr.get_ready_tasks()
        my_tasks = [t for t in ready_tasks if matches_my_skills(t, agent_name)]
        
        if my_tasks:
            task = my_tasks[0]
            task_mgr.update_status(task["id"], "in_progress")
            execute_task(task)
            task_mgr.update_status(task["id"], "done")
            continue
        
        # 3. 沒有任務，idle 等待
        time.sleep(5)

def create_isolated_workspace(task_id: str) -> str:
    """為任務建立隔離的 Git Worktree"""
    import subprocess
    
    branch_name = f"agent/{task_id}"
    worktree_path = f"worktrees/{task_id}"
    
    # 建立 worktree
    subprocess.run(
        f"git worktree add {worktree_path} -b {branch_name}",
        shell=True, check=True
    )
    
    return worktree_path

def cleanup_workspace(task_id: str):
    """清理完成的 Worktree"""
    import subprocess
    worktree_path = f"worktrees/{task_id}"
    
    subprocess.run(f"git worktree remove {worktree_path}", shell=True)
    subprocess.run(f"git branch -d agent/{task_id}", shell=True)

1. 建立任務 → task_mgr.create("實作登入 API")
2. 建立隔離空間 → create_isolated_workspace("task-001")
3. Agent 在 worktrees/task-001/ 中工作
4. 執行測試（在隔離環境中）
5. 完成後：
   a. git merge 回主分支
   b. 清理 worktree
   c. 更新任務狀態為 done

def handle_bash_with_recovery(command: str, max_retries: int = 3) -> str:
    """帶錯誤恢復的 bash 執行"""
    import subprocess
    
    result = subprocess.run(
        command, shell=True,
        capture_output=True, text=True, timeout=60
    )
    
    output = result.stdout + result.stderr
    
    if result.returncode != 0:
        # 提供結構化的錯誤資訊，讓模型能分析並修復
        return (
            f"COMMAND_FAILED (exit code: {result.returncode})\n"
            f"STDOUT:\n{result.stdout[:3000]}\n"
            f"STDERR:\n{result.stderr[:3000]}\n"
            f"HINT: 請分析錯誤原因並嘗試修復。"
        )
    
    return output[:10000]

Agent 執行 mvn test
    │
    ▼
測試失敗：3 errors
    │
    ▼
Agent 分析錯誤訊息
    │
    ▼
read_file 讀取失敗的測試
    │
    ▼
read_file 讀取對應的原始碼
    │
    ▼
edit_file 修復程式碼
    │
    ▼
bash: mvn test（重新測試）
    │
    ▼
全部通過 ✅

import subprocess
import json

class MCPClient:
    """MCP 客戶端：透過標準協議連接外部 MCP Server"""
    
    def __init__(self, server_command: str):
        """啟動 MCP Server 子程序"""
        self.process = subprocess.Popen(
            server_command.split(),
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        self._msg_id = 0
        self._initialize()
    
    def _send(self, method: str, params: dict = None) -> dict:
        """透過 JSON-RPC 2.0 與 MCP Server 通訊"""
        self._msg_id += 1
        request = {
            "jsonrpc": "2.0",
            "id": self._msg_id,
            "method": method,
            "params": params or {},
        }
        payload = json.dumps(request) + "\n"
        self.process.stdin.write(payload.encode())
        self.process.stdin.flush()
        
        response = self.process.stdout.readline()
        return json.loads(response)
    
    def _initialize(self):
        """初始化 MCP 連接"""
        self._send("initialize", {
            "protocolVersion": "2024-11-05",
            "clientInfo": {"name": "learn-claude-code", "version": "1.0"},
        })
    
    def list_tools(self) -> list:
        """列出 MCP Server 提供的所有工具"""
        result = self._send("tools/list")
        return result.get("result", {}).get("tools", [])
    
    def call_tool(self, tool_name: str, arguments: dict) -> str:
        """呼叫 MCP 工具"""
        result = self._send("tools/call", {
            "name": tool_name,
            "arguments": arguments,
        })
        content = result.get("result", {}).get("content", [])
        return "\n".join(c.get("text", "") for c in content)

class MCPToolPool:
    """統一工具池：內建 Tool + MCP 動態 Tool 的無縫整合"""
    
    NAMING_PREFIX = "mcp__{server}__{tool}"
    
    def __init__(self, builtin_tools: list, builtin_handlers: dict):
        self.tools = list(builtin_tools)         # 內建工具定義
        self.handlers = dict(builtin_handlers)   # 內建工具處理器
        self.mcp_clients = {}
    
    def connect_mcp_server(self, server_name: str, command: str):
        """連接 MCP Server 並整合其工具"""
        client = MCPClient(command)
        self.mcp_clients[server_name] = client
        
        # 取得 MCP Server 的工具清單
        mcp_tools = client.list_tools()
        
        for tool in mcp_tools:
            # 使用命名前綴避免衝突：mcp__github__create_issue
            safe_name = self.NAMING_PREFIX.format(
                server=server_name,
                tool=self._normalize_name(tool["name"])
            )
            
            # 加入工具定義（與內建工具格式一致）
            self.tools.append({
                "name": safe_name,
                "description": tool.get("description", ""),
                "input_schema": tool.get("inputSchema", {}),
            })
            
            # 動態建立 Handler
            def make_handler(client, original_name):
                return lambda **kwargs: client.call_tool(original_name, kwargs)
            
            self.handlers[safe_name] = make_handler(client, tool["name"])
    
    def _normalize_name(self, name: str) -> str:
        """標準化工具名稱，移除特殊字元"""
        import re
        return re.sub(r'[^a-zA-Z0-9_]', '_', name).lower()

# 企業 MCP Server 整合範例
tool_pool = MCPToolPool(TOOLS, TOOL_HANDLERS)

# 連接 GitHub MCP Server（管理 Issue / PR）
tool_pool.connect_mcp_server("github", "npx @modelcontextprotocol/server-github")

# 連接資料庫 MCP Server
tool_pool.connect_mcp_server("postgres", "npx @modelcontextprotocol/server-postgres")

# 連接 Jira MCP Server（企業自建）
tool_pool.connect_mcp_server("jira", "python mcp_servers/jira_server.py")

# Agent Loop 無需任何改動，自動支援所有 MCP 工具
print(f"可用工具數量：{len(tool_pool.tools)} 個")
# 輸出：可用工具數量：42 個（5 內建 + 37 MCP 動態）

選 MCP Server：
✅ 需要與多個 Agent / 多個團隊共享工具
✅ 工具需要獨立部署、版本管理
✅ 整合的系統已有 MCP Server 生態
✅ 需要支援多種傳輸協議（stdio / HTTP / WebSocket）

選手工 Tool Handler：
✅ 單一 Agent 使用的簡單工具
✅ 快速原型開發
✅ 需要與 Agent 緊密耦合的業務邏輯
✅ 敏感資訊不適合透過網路傳輸

s20 完整 Agent 工具清單（27+ 內建）：

核心工具（s01-s02）：
  bash, read_file, write_file, edit_file, glob, grep

計畫與隔離（s05-s06）：
  todo_write, todo_update, subagent_spawn

知識管理（s07, s10）：
  load_skill, list_skills, update_system_prompt

記憶（s09）：
  memory_store, memory_recall, memory_consolidate

任務系統（s12）：
  task_create, task_update, task_list, task_get_ready

背景執行（s13）：
  background_run, background_check

排程（s14）：
  cron_add, cron_remove, cron_list

團隊協作（s15-s16）：
  team_send, team_read, team_register

工作區（s18）：
  worktree_create, worktree_remove

MCP 整合（s19）：
  mcp_connect, mcp_list_tools（動態）

+ MCP 動態工具（無限擴充）

使用者 Prompt
    │
    ▼ UserPromptSubmit Hook（Prompt 掃描）
┌──────────────────────────────────────────────────┐
│                  Agent Loop (s01)                │
│                                                  │
│  ┌─────────────────────────────────────────────┐ │
│  │ System Prompt（動態組裝 s10）                │ │
│  │ = Identity + Skills + Memory + Context      │ │
│  └─────────────────────────────────────────────┘ │
│                                                  │
│  ┌─────────────────────────────────────────────┐ │
│  │ Tool Pool（s02 + s19）                       │ │
│  │ = 27 內建工具 + N 個 MCP 動態工具            │ │
│  └─────────────────────────────────────────────┘ │
│                  │                               │
│                  ▼ PreToolUse Hook               │
│          Permission Gateway (s03)                │
│          Deny List → Rules → Approval            │
│                  │                               │
│                  ▼                               │
│          Tool Execution                          │
│                  │                               │
│                  ▼ PostToolUse Hook              │
│          Audit Log + Cost Track                  │
│                                                  │
│  Memory System (s09) ← 自動記憶關鍵資訊          │
│  Context Compact (s08) ← Token > 70% 時觸發     │
│  Task Manager (s12) ← 追蹤多步驟任務            │
│  Cron Scheduler (s14) ← 定時觸發                │
│  Agent Teams (s15-16) ← 複雜任務委派            │
│  Worktree (s18) ← 每任務獨立目錄                │
└──────────────────────────────────────────────────┘
    │
    ▼ Stop Hook（完成通知、資源清理）
Response to User

Agent Product = Model（智能） + Harness（環境）

Harness = Tools + Knowledge + Observation + Action Interfaces + Permissions

    Tools:          檔案 I/O、Shell、網路、資料庫、瀏覽器
    Knowledge:      產品文件、領域參考、API 規格、風格指南
    Observation:    git diff、錯誤日誌、瀏覽器狀態、感測器數據
    Action:         CLI 命令、API 呼叫、UI 互動
    Permissions:    沙箱、審批流程、信任邊界

┌──────────────────────────────────────────┐
│                                          │
│   模型 = 駕駛員（決策者）                 │
│   Harness = 車輛（執行環境）              │
│                                          │
│   模型決定去哪裡、怎麼開                  │
│   Harness 提供方向盤、油門、煞車          │
│                                          │
│   你的工作：打造一輛好車                  │
│   模型的工作：開好這輛車                  │
│                                          │
└──────────────────────────────────────────┘

Harness 工程師的工作流：

  ┌─────────┐    ┌─────────┐     ┌─────────┐    ┌─────────┐    ┌─────────┐
  │ 1.Tools │ →  │2.Knowledge│ → │3.Context│ →  │4.Perms  │ →  │ 5.Data  │
  │         │    │         │     │         │    │         │    │         │
  │ 建構    │    │ 策展    │      │ 管理    │    │ 治理    │     │ 收集    │
  │ 行動能力│    │ 領域知識│      │ 記憶窗口│     │ 安全邊界│     │ 訓練素材│
  └─────────┘    └─────────┘     └─────────┘    └─────────┘    └─────────┘
                              ↑                              │
                              └──────── 持續反饋迴圈 ─────────┘

# ✅ 好的 Tool 設計：原子、可組合、描述清晰
{
    "name": "db_query",
    "description": "對指定資料庫執行唯讀 SQL 查詢。僅支援 SELECT，不可執行 DDL/DML。結果以 JSON 陣列回傳，最多 100 筆。",
    "input_schema": {
        "type": "object",
        "properties": {
            "database": {
                "type": "string",
                "enum": ["app_db", "audit_db", "analytics_db"],
                "description": "目標資料庫名稱"
            },
            "sql": {
                "type": "string",
                "description": "SELECT SQL 語句"
            }
        },
        "required": ["database", "sql"]
    }
}

# ❌ 不好的 Tool 設計：太泛、無限制、描述模糊
{
    "name": "database",
    "description": "操作資料庫",
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {"type": "string"}
        }
    }
}

# 多層安全機制設計

class ToolPermissionManager:
    """Tool 權限管理器"""
    
    # 1. 白名單機制：限制可用工具
    ALLOWED_TOOLS = {
        "junior_agent": ["read_file", "grep", "glob"],
        "senior_agent": ["read_file", "write_file", "edit_file", "bash", "grep", "glob"],
        "admin_agent":  ["read_file", "write_file", "edit_file", "bash", "grep", "glob", "db_query"],
    }
    
    # 2. 路徑限制：限制可存取的檔案範圍
    ALLOWED_PATHS = [
        "src/",
        "tests/",
        "docs/",
    ]
    BLOCKED_PATHS = [
        ".env",
        "secrets/",
        "credentials/",
        "*.pem",
        "*.key",
    ]
    
    # 3. 命令黑名單：禁止危險命令
    BLOCKED_COMMANDS = [
        "rm -rf /",
        "DROP DATABASE",
        "DROP TABLE",
        "DELETE FROM",  # 不允許不帶 WHERE 的 DELETE
        "curl.*|bash",  # 禁止遠端程式碼執行
        "chmod 777",
    ]
    
    # 4. 審批流程：需要人工確認的操作
    REQUIRE_APPROVAL = [
        "git push",
        "docker push",
        "kubectl apply",
        "terraform apply",
        "npm publish",
    ]
    
    def check_permission(self, agent_role: str, tool_name: str, 
                         tool_input: dict) -> tuple:
        """檢查權限，回傳 (allowed: bool, reason: str)"""
        
        # 檢查工具白名單
        if tool_name not in self.ALLOWED_TOOLS.get(agent_role, []):
            return False, f"角色 {agent_role} 無權使用工具 {tool_name}"
        
        # 檢查路徑限制
        if "path" in tool_input:
            path = tool_input["path"]
            if any(path.startswith(bp) or path.endswith(bp.lstrip("*")) 
                   for bp in self.BLOCKED_PATHS):
                return False, f"禁止存取路徑: {path}"
        
        # 檢查命令黑名單
        if tool_name == "bash":
            import re
            cmd = tool_input.get("command", "")
            for pattern in self.BLOCKED_COMMANDS:
                if re.search(pattern, cmd, re.IGNORECASE):
                    return False, f"禁止執行命令: {cmd}"
        
        # 檢查是否需要審批
        if tool_name == "bash":
            cmd = tool_input.get("command", "")
            for pattern in self.REQUIRE_APPROVAL:
                if pattern in cmd:
                    return self._request_approval(cmd)
        
        return True, "OK"
    
    def _request_approval(self, command: str) -> tuple:
        """請求人工審批"""
        print(f"\n⚠️ 需要審批的操作: {command}")
        approval = input("是否允許？(y/n): ").strip().lower()
        if approval == "y":
            return True, "已核准"
        return False, "使用者拒絕"

SYSTEM_PROMPT = """你是一位資深軟體工程師，負責在企業級 Java 專案中進行開發。

## 工作規則
1. 在修改程式碼前，先使用 read_file 了解現有程式碼
2. 使用 todo_write 規劃任務步驟後再開始執行
3. 每次修改後執行測試（mvn test 或 npm test）
4. 遵循專案的 coding style（見 .editorconfig）

## 安全規則
- 不得修改 .env、credentials 等機密檔案
- 不得執行 rm -rf、DROP TABLE 等危險命令
- SQL 查詢必須使用參數化查詢，禁止字串拼接

## 程式碼品質
- Java 使用 Google Java Style
- 所有 public 方法需有 JavaDoc
- 單元測試覆蓋率需達 80%

## 錯誤處理
- 遇到編譯錯誤：讀取錯誤訊息 → 分析原因 → 修復 → 重新編譯
- 遇到測試失敗：讀取測試程式碼 → 讀取實作程式碼 → 修復 → 重跑測試
- 最多重試 3 次，若仍失敗則報告問題
"""

def execute_with_retry(tool_name: str, tool_input: dict, 
                        max_retries: int = 3) -> str:
    """帶 retry 的工具執行"""
    for attempt in range(max_retries):
        result = TOOL_HANDLERS[tool_name](**tool_input)
        
        if "ERROR" not in result and "FAILED" not in result:
            return result
        
        if attempt < max_retries - 1:
            # 附加重試資訊讓模型知道這是第幾次嘗試
            result += f"\n\n[重試 {attempt + 1}/{max_retries}] 請分析錯誤原因並調整策略。"
    
    return result + "\n\n[已達最大重試次數] 請回報此問題給使用者。"

# 輸出長度限制
def truncate_output(output: str, max_chars: int = 10000) -> str:
    if len(output) <= max_chars:
        return output
    half = max_chars // 2
    return (
        output[:half] + 
        f"\n\n... [截斷 {len(output) - max_chars} 字元] ...\n\n" + 
        output[-half:]
    )

# Token 預算控制
def check_token_budget(messages: list, max_budget: int = 150000) -> bool:
    current = estimate_tokens(messages)
    if current > max_budget:
        # 觸發 context compact
        return False
    return True

# 迴圈次數限制
MAX_LOOP_ITERATIONS = 50

def agent_loop_with_guardrails(user_message: str):
    messages = [{"role": "user", "content": user_message}]
    iteration = 0
    
    while iteration < MAX_LOOP_ITERATIONS:
        iteration += 1
        
        if not check_token_budget(messages):
            messages = compact_context(messages)
        
        response = client.messages.create(
            model=MODEL, system=SYSTEM_PROMPT,
            max_tokens=4096, messages=messages, tools=TOOLS,
        )
        
        # ... 正常迴圈邏輯 ...
    
    print(f"⚠️ 已達最大迴圈次數 ({MAX_LOOP_ITERATIONS})，強制停止")

載入優先順序（高 → 低）：

1. 全域設定（~/.claude/CLAUDE.md）
   └── 適用所有專案的個人偏好、通用規則

2. 專案根目錄（./CLAUDE.md）
   └── 整個專案的 Agent 行為規範

3. 子目錄（./src/CLAUDE.md、./tests/CLAUDE.md）
   └── 特定模組的局部覆蓋規則

規則合併原則：
  子目錄 > 專案根目錄 > 全域設定
  衝突時，較具體的規則優先

<!-- 專案根目錄 CLAUDE.md -->
# 專案 Agent 行為規範

## 專案背景
這是一個金融業核心交易系統，使用 Java 21 + Spring Boot 3.2。
所有程式碼變更必須符合 PCI-DSS 合規要求。

## 技術規範
- 架構：Clean Architecture（Controller → Service → Repository）
- 框架：Spring Boot 3.2、Spring Security 6、JPA/Hibernate 6
- 測試：JUnit 5、Mockito、AssertJ（覆蓋率 >= 80%）
- 程式碼風格：Google Java Style Guide

## 強制性行為規則
1. **閱讀優先**：修改任何檔案前必須先用 read_file 了解現有實作
2. **先計畫後執行**：複雜任務必須先用 todo_write 列出步驟
3. **修改後必測**：每次程式碼修改後執行 `mvn test -q`
4. **禁止刪除測試**：不可刪除現有測試，只能新增

## 安全紅線（絕對禁止）
- 不可修改 .env、application-prod.yml 等機密配置
- SQL 查詢必須使用參數化，禁止字串拼接
- 不可在程式碼中硬編碼 API Key、密碼、Token
- 不可直接操作生產資料庫

## 命名規範
- 類別：PascalCase
- 方法：camelCase，動詞開頭（findById, createOrder）
- 常數：UPPER_SNAKE_CASE
- DB 欄位：snake_case

## 錯誤處理 SOP
遇到編譯錯誤：
  1. 讀取完整錯誤訊息
  2. read_file 讀取出錯檔案（含上下文 ±20 行）
  3. 分析根因
  4. edit_file 修復
  5. 重新編譯驗證
  最多重試 3 次，若仍失敗則回報問題

## 常用指令
- 編譯：`mvn clean compile -q`
- 測試：`mvn test -q`
- 安全掃描：`mvn dependency-check:check -q`

✅ 適合放入 CLAUDE.md：
  - 專案架構描述（不常變）
  - 絕對禁止的操作（安全紅線）
  - 必須遵守的流程（閱讀 → 計畫 → 執行 → 測試）
  - 命名規範和程式碼風格
  - 常用指令速查

❌ 不適合放入 CLAUDE.md：
  - 詳細的 API 文件（應放 Skills）
  - 當前任務說明（應放 Prompt）
  - 頻繁更動的配置
  - 過長的內容（建議 < 500 行）

# 1. Clone 專案
git clone https://github.com/shareAI-lab/learn-claude-code
cd learn-claude-code

# 2. 建立虛擬環境（建議）
python -m venv .venv

# Windows
.venv\Scripts\activate

# Linux / macOS
source .venv/bin/activate

# 3. 安裝依賴
pip install -r requirements.txt

# 4. 設定環境變數
cp .env.example .env

# Anthropic API Key（必要）
ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxxx

# 模型選擇（可選，預設 claude-sonnet-4-6）
MODEL=claude-sonnet-4-6

# 如果使用替代 API 端點
# ANTHROPIC_BASE_URL=https://your-proxy.company.com/v1

# 執行最簡單的 Agent（s01）
python agents/s01_agent_loop.py

# 輸入測試指令
> 請列出當前目錄下的所有 Python 檔案

learn-claude-code/
├── agents/                    # 12 個 Session 的 Python 實作
│   ├── s01_agent_loop.py      # Session 1: 基礎 Agent Loop
│   ├── s02_tool_use.py        # Session 2: Tool Dispatch
│   ├── s03_todo_write.py      # Session 3: TodoWrite 計畫
│   ├── s04_subagent.py        # Session 4: Subagent 隔離
│   ├── s05_skills.py          # Session 5: Skills 按需載入
│   ├── s06_context_compact.py # Session 6: Context 壓縮
│   ├── s07_tasks.py           # Session 7: Task 持久化
│   ├── s08_background.py      # Session 8: 背景任務
│   ├── s09_agent_teams.py     # Session 9: Agent 團隊
│   ├── s10_team_protocols.py  # Session 10: 團隊協定
│   ├── s11_autonomous.py      # Session 11: 自治 Agent
│   ├── s12_worktree_task_isolation.py  # Session 12: Worktree 隔離
│   └── s_full.py              # 完整版：所有機制整合
├── docs/
│   ├── en/                    # 英文文件
│   ├── zh/                    # 中文文件
│   └── ja/                    # 日文文件
├── skills/                    # Skill 知識文件
├── web/                       # Next.js 互動學習平台
├── tests/                     # 測試
├── .env.example               # 環境變數範本
└── requirements.txt           # Python 依賴

Week 1：基礎（獨自完成）
├── Day 1-2：s01 Agent Loop — 理解核心迴圈
├── Day 3-4：s02 Tool Use — 新增工具到 dispatch map
└── Day 5  ：s03 TodoWrite — 加入計畫能力

Week 2：進階（配對學習）
├── Day 1-2：s04 Subagent — 子任務隔離
├── Day 3  ：s05 Skills — 按需知識載入
└── Day 4-5：s06 Context Compact — 壓縮策略

Week 3：持久化與團隊（團隊討論）
├── Day 1-2：s07 Tasks + s08 Background
├── Day 3-4：s09-s10 Agent Teams
└── Day 5  ：s11-s12 Autonomous + Worktree

Week 4：整合與企業客製（實戰）
└── 基於 s_full.py 開發企業級 Agent

#!/usr/bin/env python3
"""
Session 01: The Agent Loop
格言："One loop & Bash is all you need"

這是最基礎的 Agent — 一個 while 迴圈 + 一個 bash 工具。
模型決定何時呼叫工具、何時停止。程式碼只負責執行。
"""

import os
import subprocess
from anthropic import Anthropic

client = Anthropic()
MODEL = os.getenv("MODEL", "claude-sonnet-4-6")

# System Prompt：定義 Agent 的角色和行為規則
SYSTEM = """你是一位資深軟體工程師。
你可以使用 bash 工具執行 shell 命令來完成任務。
在執行命令前，先解釋你要做什麼以及為什麼。
"""

# Tool 定義：使用 JSON Schema 描述工具的名稱、用途、參數
TOOLS = [{
    "name": "bash",
    "description": "在當前目錄執行 shell 命令。適用於查看檔案、執行程式、Git 操作等。",
    "input_schema": {
        "type": "object",
        "properties": {
            "command": {
                "type": "string",
                "description": "要執行的 shell 命令"
            }
        },
        "required": ["command"]
    }
}]

def handle_bash(command: str) -> str:
    """執行 shell 命令並回傳輸出"""
    try:
        result = subprocess.run(
            command, shell=True,
            capture_output=True, text=True,
            timeout=30, cwd=os.getcwd()
        )
        output = result.stdout + result.stderr
        return output[:10000] if output else "(無輸出)"
    except subprocess.TimeoutExpired:
        return "ERROR: 命令執行超時（30 秒限制）"

TOOL_HANDLERS = {"bash": handle_bash}

def agent_loop():
    """互動式 Agent 主迴圈"""
    print("=== Agent Loop (s01) ===")
    print("輸入任務指令，Agent 會使用 bash 工具完成。輸入 'exit' 離開。\n")
    
    messages = []
    
    while True:
        user_input = input("You> ").strip()
        if user_input.lower() in ("exit", "quit"):
            break
        
        messages.append({"role": "user", "content": user_input})
        
        # Agent 內部迴圈：持續執行直到模型決定停止
        while True:
            response = client.messages.create(
                model=MODEL, system=SYSTEM,
                max_tokens=4096, messages=messages, tools=TOOLS,
            )
            
            messages.append({
                "role": "assistant",
                "content": response.content
            })
            
            # 模型決定停止
            if response.stop_reason != "tool_use":
                for block in response.content:
                    if hasattr(block, "text"):
                        print(f"\nAgent> {block.text}\n")
                break
            
            # 執行模型請求的工具
            results = []
            for block in response.content:
                if block.type == "tool_use":
                    print(f"  [執行] {block.name}: {block.input}")
                    output = TOOL_HANDLERS[block.name](**block.input)
                    print(f"  [結果] {output[:200]}...")
                    results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": output,
                    })
            
            messages.append({"role": "user", "content": results})

if __name__ == "__main__":
    agent_loop()

=== Agent Loop (s01) ===
輸入任務指令，Agent 會使用 bash 工具完成。輸入 'exit' 離開。

You> 查看這個專案用了哪些 Python 套件
  [執行] bash: {'command': 'cat requirements.txt'}
  [結果] anthropic>=0.40.0\npython-dotenv>=1.0.0...

Agent> 這個專案使用了以下 Python 套件：
- anthropic：Anthropic API 客戶端
- python-dotenv：環境變數管理

"""
Session 02: Tool Use
格言："Adding a tool means adding one handler"

新增工具 = 新增一個 handler + 一個 schema
Agent Loop 完全不需要修改
"""

# 新增工具定義
NEW_TOOLS = [
    {
        "name": "read_file",
        "description": "讀取檔案內容。可指定行號範圍。",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "檔案的相對路徑"},
                "start_line": {"type": "integer", "description": "起始行號（1-based，可選）"},
                "end_line": {"type": "integer", "description": "結束行號（1-based，可選）"}
            },
            "required": ["path"]
        }
    },
    {
        "name": "write_file",
        "description": "建立或覆寫檔案。目錄不存在時自動建立。",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "檔案路徑"},
                "content": {"type": "string", "description": "檔案完整內容"}
            },
            "required": ["path", "content"]
        }
    },
    {
        "name": "edit_file",
        "description": "精確替換檔案中的一段文字。old_string 必須在檔案中出現恰好一次。",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "檔案路徑"},
                "old_string": {"type": "string", "description": "要被替換的原始文字（含上下文）"},
                "new_string": {"type": "string", "description": "替換後的新文字"}
            },
            "required": ["path", "old_string", "new_string"]
        }
    },
    {
        "name": "glob",
        "description": "用 glob 模式搜尋檔案路徑",
        "input_schema": {
            "type": "object",
            "properties": {
                "pattern": {"type": "string", "description": "glob 模式，例如 '**/*.java'"}
            },
            "required": ["pattern"]
        }
    },
    {
        "name": "grep",
        "description": "在檔案中搜尋文字或正規表達式",
        "input_schema": {
            "type": "object",
            "properties": {
                "pattern": {"type": "string", "description": "搜尋模式（支援 regex）"},
                "path": {"type": "string", "description": "搜尋路徑（可為目錄）"},
                "include": {"type": "string", "description": "只搜尋匹配的檔案，如 '*.java'"}
            },
            "required": ["pattern"]
        }
    }
]

# 新增 Handler 函式
import glob as glob_module

def handle_read_file(path: str, start_line: int = None, end_line: int = None) -> str:
    try:
        with open(path, "r", encoding="utf-8") as f:
            lines = f.readlines()
        if start_line and end_line:
            selected = lines[max(0, start_line - 1):end_line]
            header = f"[{path} L{start_line}-{end_line}]\n"
            return header + "".join(selected)
        return "".join(lines)[:15000]
    except FileNotFoundError:
        return f"ERROR: 檔案不存在: {path}"

def handle_write_file(path: str, content: str) -> str:
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return f"OK: 已寫入 {path}（{len(content)} 字元）"

def handle_edit_file(path: str, old_string: str, new_string: str) -> str:
    try:
        with open(path, "r", encoding="utf-8") as f:
            content = f.read()
    except FileNotFoundError:
        return f"ERROR: 檔案不存在: {path}"
    
    count = content.count(old_string)
    if count == 0:
        return "ERROR: 找不到要替換的字串"
    if count > 1:
        return f"ERROR: 找到 {count} 個匹配，請提供更精確的上下文"
    
    new_content = content.replace(old_string, new_string, 1)
    with open(path, "w", encoding="utf-8") as f:
        f.write(new_content)
    return "OK: 替換完成"

def handle_glob(pattern: str) -> str:
    matches = sorted(glob_module.glob(pattern, recursive=True))
    if not matches:
        return f"找不到匹配 '{pattern}' 的檔案"
    return "\n".join(matches[:100])

def handle_grep(pattern: str, path: str = ".", include: str = None) -> str:
    import subprocess
    cmd = f'grep -rn "{pattern}" {path}'
    if include:
        cmd += f' --include="{include}"'
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    output = result.stdout
    return output[:10000] if output else f"找不到匹配 '{pattern}' 的內容"

# 擴展 Dispatch Map
TOOL_HANDLERS.update({
    "read_file":  handle_read_file,
    "write_file": handle_write_file,
    "edit_file":  handle_edit_file,
    "glob":       handle_glob,
    "grep":       handle_grep,
})

# Agent Loop 完全不變 — 新工具自動可用

"""
Session 03: TodoWrite
格言："An agent without a plan drifts"

加入 todo_write 工具 + nag reminder 機制
讓 Agent 先規劃再執行，大幅提升完成率
"""

class TodoManager:
    def __init__(self):
        self.todos = []
    
    def write(self, todos: list) -> str:
        self.todos = [
            {"id": i + 1, "task": t["task"], "status": t.get("status", "todo")}
            for i, t in enumerate(todos)
        ]
        return self._format()
    
    def update(self, todo_id: int, status: str) -> str:
        for t in self.todos:
            if t["id"] == todo_id:
                t["status"] = status
                return self._format()
        return f"ERROR: 找不到任務 #{todo_id}"
    
    def _format(self) -> str:
        icons = {"todo": "⬜", "in_progress": "🔄", "done": "✅", "skipped": "⏭️"}
        lines = ["📋 任務清單："]
        for t in self.todos:
            lines.append(f"  {icons.get(t['status'], '?')} #{t['id']} {t['task']}")
        pending = sum(1 for t in self.todos if t["status"] in ("todo", "in_progress"))
        lines.append(f"\n  進度：{len(self.todos) - pending}/{len(self.todos)} 完成")
        return "\n".join(lines)
    
    def get_nag_reminder(self) -> str:
        """產生提醒訊息"""
        pending = [t for t in self.todos if t["status"] in ("todo", "in_progress")]
        if not pending:
            return ""
        lines = ["\n⚠️ 尚未完成的任務："]
        for t in pending:
            lines.append(f"  - #{t['id']} {t['task']} [{t['status']}]")
        lines.append("請繼續執行下一個任務。")
        return "\n".join(lines)

todo_mgr = TodoManager()

# 註冊 todo_write tool
TODO_TOOL = {
    "name": "todo_write",
    "description": "建立或更新任務計畫清單。在開始複雜任務前，先用此工具規劃步驟。",
    "input_schema": {
        "type": "object",
        "properties": {
            "todos": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "task": {"type": "string"},
                        "status": {"type": "string", "enum": ["todo", "in_progress", "done", "skipped"]}
                    },
                    "required": ["task", "status"]
                }
            }
        },
        "required": ["todos"]
    }
}

def handle_todo_write(todos: list) -> str:
    return todo_mgr.write(todos)

# 在 tool_result 注入 nag reminder
def agent_loop_with_planning(user_message: str):
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM,
            max_tokens=4096, messages=messages, tools=TOOLS,
        )
        messages.append({"role": "assistant", "content": response.content})
        
        if response.stop_reason != "tool_use":
            for block in response.content:
                if hasattr(block, "text"):
                    print(block.text)
            return
        
        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = TOOL_HANDLERS[block.name](**block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        
        # 🔑 關鍵：注入 Nag Reminder
        nag = todo_mgr.get_nag_reminder()
        if nag:
            results[-1]["content"] += nag
        
        messages.append({"role": "user", "content": results})

# 專門的測試工具
TEST_TOOL = {
    "name": "run_tests",
    "description": "執行專案的測試套件。支援指定測試檔案或測試類別。",
    "input_schema": {
        "type": "object",
        "properties": {
            "test_path": {
                "type": "string",
                "description": "測試檔案或目錄路徑（可選，預設執行全部測試）"
            },
            "test_class": {
                "type": "string",
                "description": "Java 測試類別名稱（可選）"
            },
            "framework": {
                "type": "string",
                "enum": ["maven", "gradle", "pytest", "jest"],
                "description": "測試框架"
            }
        },
        "required": ["framework"]
    }
}

def handle_run_tests(framework: str, test_path: str = None, 
                     test_class: str = None) -> str:
    """執行測試並回傳結構化結果"""
    cmd_map = {
        "maven": "mvn test",
        "gradle": "./gradlew test",
        "pytest": "python -m pytest",
        "jest": "npx jest",
    }
    
    cmd = cmd_map.get(framework, "mvn test")
    
    if test_class and framework == "maven":
        cmd += f" -Dtest={test_class}"
    elif test_path and framework == "pytest":
        cmd += f" {test_path}"
    
    result = subprocess.run(
        cmd, shell=True, capture_output=True,
        text=True, timeout=300  # 測試可能耗時較長
    )
    
    output = result.stdout + result.stderr
    
    # 結構化輸出
    status = "PASSED" if result.returncode == 0 else "FAILED"
    return (
        f"TEST_RESULT: {status}\n"
        f"EXIT_CODE: {result.returncode}\n"
        f"OUTPUT:\n{output[-5000:]}"  # 取最後 5000 字元（通常包含結果摘要）
    )

# 在 System Prompt 中加入錯誤修復指引
ERROR_RECOVERY_PROMPT = """
## 錯誤修復流程

當你遇到錯誤時，請遵循以下步驟：

### 編譯錯誤
1. 仔細閱讀錯誤訊息，找出出錯的檔案和行號
2. 使用 read_file 讀取出錯的程式碼（含上下文）
3. 分析錯誤原因
4. 使用 edit_file 修復
5. 重新編譯驗證

### 測試失敗
1. 閱讀測試失敗的詳細資訊（expected vs actual）
2. read_file 讀取測試程式碼
3. read_file 讀取被測試的實作程式碼
4. 判斷是測試錯誤還是實作錯誤
5. 修復並重跑測試

### 執行時期錯誤
1. 閱讀 stack trace
2. 定位到拋出例外的程式碼
3. 檢查輸入資料和狀態
4. 修復問題

### 注意事項
- 最多重試 3 次同一個修復
- 如果 3 次都無法解決，回報問題讓使用者決定
- 不要盲目修改，先理解再動手
"""

# 範例：自動修復流程
SYSTEM_WITH_RECOVERY = SYSTEM + ERROR_RECOVERY_PROMPT

User: "在 src/main/java/com/example 下建立 Calculator.java，包含加減乘除方法，
       並建立對應的 JUnit 測試。所有測試必須通過。"

Agent 計畫（TodoWrite）：
  ⬜ #1 建立 Calculator.java
  ⬜ #2 建立 CalculatorTest.java
  ⬜ #3 執行測試
  ⬜ #4 修復任何失敗的測試

Agent 執行：
  🔄 #1 write_file → Calculator.java ✅
  🔄 #2 write_file → CalculatorTest.java ✅
  🔄 #3 run_tests → 2 FAILED
     │
     ├── 分析：除法方法未處理除以零
     ├── read_file Calculator.java
     ├── edit_file → 加入除以零檢查
     ├── run_tests → 1 FAILED
     │
     ├── 分析：測試期望的例外類型不對
     ├── read_file CalculatorTest.java
     ├── edit_file → 修正例外類型
     ├── run_tests → ALL PASSED ✅
     │
  ✅ #4 所有測試通過

#!/usr/bin/env python3
"""
實作 Hooks 機制（對應 s04）
讓 Agent Loop 具備 PreToolUse / PostToolUse / Stop 三種擴展點
"""

import os
import json
import time
import subprocess
from anthropic import Anthropic

client = Anthropic()
MODEL = os.getenv("MODEL", "claude-sonnet-4-6")

# =========================================================
# Step 1: 定義 Hook Registry
# =========================================================
_HOOKS = {
    "PreToolUse":  [],
    "PostToolUse": [],
    "Stop":        [],
}

def on(event: str):
    """裝飾器：將函式註冊為 Hook Handler"""
    def decorator(fn):
        _HOOKS[event].append(fn)
        return fn
    return decorator

def fire(event: str, **kwargs) -> dict:
    """觸發 Hook，收集所有 Handler 的回傳值"""
    results = {}
    for handler in _HOOKS.get(event, []):
        result = handler(**kwargs)
        if result:
            results.update(result)
    return results

# =========================================================
# Step 2: 實作各 Hook
# =========================================================

@on("PreToolUse")
def security_guard(tool_name: str, tool_input: dict, **_) -> dict | None:
    """安全防護：阻止危險命令"""
    if tool_name == "bash":
        cmd = tool_input.get("command", "")
        DANGEROUS = ["rm -rf /", "DROP DATABASE", "chmod 777 /", "curl | bash"]
        for d in DANGEROUS:
            if d.lower() in cmd.lower():
                print(f"🔴 [Security] 阻止危險命令: {cmd}")
                return {"blocked": True, "reason": f"危險命令: {d}"}
    return None

@on("PreToolUse")
def audit_logger(tool_name: str, tool_input: dict, **_):
    """審計日誌：記錄所有工具呼叫"""
    entry = {
        "timestamp": time.time(),
        "event": "PRE_TOOL",
        "tool": tool_name,
        "input_summary": str(tool_input)[:200],
    }
    with open("audit.jsonl", "a") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")

@on("PostToolUse")
def token_tracker(tool_name: str, result: str, **_):
    """成本追蹤：累計 Token 使用量（估算）"""
    estimated_tokens = len(result) // 4
    entry = {
        "timestamp": time.time(),
        "event": "POST_TOOL",
        "tool": tool_name,
        "estimated_tokens": estimated_tokens,
    }
    with open("cost_tracking.jsonl", "a") as f:
        f.write(json.dumps(entry) + "\n")

@on("Stop")
def completion_notifier(**_):
    """任務完成通知（可整合 Slack / Teams / Email）"""
    print("✅ [Hook] Agent 任務完成")
    # 實際整合：發送 Slack Webhook、觸發下游系統等

# =========================================================
# Step 3: 整合到 Agent Loop（僅需在工具呼叫前後插入 fire()）
# =========================================================

def handle_bash(command: str) -> str:
    result = subprocess.run(
        command, shell=True, capture_output=True,
        text=True, timeout=30
    )
    return (result.stdout + result.stderr)[:10000]

TOOL_HANDLERS = {"bash": handle_bash}
TOOLS = [{
    "name": "bash",
    "description": "執行 shell 命令",
    "input_schema": {
        "type": "object",
        "properties": {"command": {"type": "string"}},
        "required": ["command"]
    }
}]

def agent_loop_with_hooks(user_message: str):
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = client.messages.create(
            model=MODEL, max_tokens=4096,
            messages=messages, tools=TOOLS,
        )
        messages.append({"role": "assistant", "content": response.content})
        
        if response.stop_reason != "tool_use":
            fire("Stop")  # ← Stop Hook
            for block in response.content:
                if hasattr(block, "text"):
                    print(block.text)
            return
        
        results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            
            # ↓ PreToolUse Hook
            hook_result = fire("PreToolUse",
                               tool_name=block.name,
                               tool_input=block.input)
            
            if hook_result.get("blocked"):
                output = f"BLOCKED: {hook_result.get('reason', '安全防護阻止')}"
            else:
                output = TOOL_HANDLERS[block.name](**block.input)
            
            # ↓ PostToolUse Hook
            fire("PostToolUse", tool_name=block.name, result=output)
            
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": output,
            })
        
        messages.append({"role": "user", "content": results})

if __name__ == "__main__":
    agent_loop_with_hooks("列出當前目錄下的所有 Python 檔案並統計行數")

未使用 Hooks：                    使用 Hooks 後：
┌──────────────┐                 ┌──────────────┐
│ Agent Loop   │                 │ Agent Loop   │
│              │                 │ ↕ PreToolUse │
│  if cmd dangerous:             │   ├─ security_guard
│    return error                │   └─ audit_logger
│  if cmd ran:                   │ ↕ PostToolUse│
│    log_to_file(...)            │   └─ token_tracker
│  if done:                      │ ↕ Stop       │
│    send_notification(...)      │   └─ completion_notifier
└──────────────┘                 └──────────────┘
迴圈程式碼越來越複雜              迴圈程式碼保持乾淨
新增功能需修改核心迴圈             新增功能只需掛鉤

# Code Generation 專用 System Prompt
CODE_GEN_SYSTEM = """你是企業級 Java 後端開發專家。

## 程式碼規範
- 使用 Clean Architecture（Controller → Service → Repository）
- 所有 public 方法需有 JavaDoc
- 使用 Lombok 簡化 boilerplate
- Entity 使用 JPA 註解
- DTO 與 Entity 分離
- 使用 MapStruct 做物件轉換
- 例外處理使用自訂 BusinessException

## 命名規範
- 類別：PascalCase（UserService、OrderController）
- 方法：camelCase（findById、createOrder）
- 常數：UPPER_SNAKE_CASE（MAX_RETRY_COUNT）
- 資料表：snake_case（user_account、order_detail）

## 安全規範
- SQL 必須使用參數化查詢
- 密碼必須 BCrypt 加密
- API 需有 JWT 驗證
- 輸入必須驗證（@Valid + @NotNull 等）

## 工作流程
1. 分析需求
2. 規劃程式碼結構（TodoWrite）
3. 生成程式碼（由底層往上：Entity → Repository → Service → Controller）
4. 生成測試
5. 執行測試確認通過
"""

Prompt: "建立使用者管理模組，包含：
  - User Entity（id, username, email, password, status, createdAt）
  - UserRepository（JPA）
  - UserService（CRUD + 密碼加密）
  - UserController（RESTful API）
  - UserDTO / CreateUserRequest / UpdateUserRequest
  - 完整的 JUnit 測試"

Agent 自動規劃並依序產生：
1. User.java（Entity + JPA 註解）
2. UserRepository.java（extends JpaRepository）
3. UserDTO.java + CreateUserRequest.java + UpdateUserRequest.java
4. UserMapper.java（MapStruct）
5. UserService.java（含 BCrypt 密碼加密）
6. UserController.java（REST + Swagger 註解）
7. UserServiceTest.java（Mockito）
8. UserControllerTest.java（MockMvc）
9. 執行 mvn test → 全部通過

# Code Review Agent 的 System Prompt
CODE_REVIEW_SYSTEM = """你是資深程式碼審查員。審查時請關注：

## 審查維度（依優先順序）
1. **安全性**（Critical）
   - SQL Injection
   - XSS
   - 硬編碼密碼 / Token
   - 不安全的反序列化
   - 路徑遍歷

2. **正確性**（High）
   - 邏輯錯誤
   - 邊界條件
   - Null 處理
   - 併發問題
   - 資源洩漏

3. **效能**（Medium）
   - N+1 查詢
   - 不必要的迴圈
   - 大量物件建立
   - 缺少索引

4. **可維護性**（Low）
   - 命名清晰度
   - 方法長度（< 30 行）
   - 類別職責（SRP）
   - 重複程式碼

## 輸出格式
對每個問題產出：
- 嚴重等級：Critical / High / Medium / Low
- 檔案和行號
- 問題描述
- 建議修復方式
- 修復後的程式碼片段
"""

def code_review_agent(file_paths: list) -> str:
    """對指定檔案執行 Code Review"""
    
    # 讀取所有待審查的檔案
    file_contents = []
    for path in file_paths:
        with open(path, "r") as f:
            file_contents.append(f"--- {path} ---\n{f.read()}")
    
    prompt = f"""請審查以下程式碼：

{chr(10).join(file_contents)}

請依照安全性 > 正確性 > 效能 > 可維護性的優先順序，列出所有發現的問題。"""
    
    # 使用 subagent 執行，避免汙染主 context
    return handle_subagent(prompt, allowed_tools=["read_file", "grep"])

TEST_GEN_SYSTEM = """你是測試工程師，負責為 Java 程式碼撰寫完整的單元測試。

## 測試框架
- JUnit 5
- Mockito（Mock 依賴）
- AssertJ（流暢斷言）

## 測試策略
1. **正常路徑**：驗證預期行為
2. **邊界條件**：空值、零、最大值、空集合
3. **錯誤路徑**：例外情境、無效輸入
4. **安全測試**：SQL Injection 嘗試、XSS payload

## 命名規範
- 測試方法：should_預期結果_when_條件
  例如：should_throwException_when_userNotFound

## 覆蓋率目標
- 行覆蓋率 >= 80%
- 分支覆蓋率 >= 70%
- 關鍵業務邏輯 >= 90%
"""

REFACTOR_SYSTEM = """你是重構專家，負責改善現有程式碼品質。

## 重構原則
1. **保持行為不變**：重構不改變外部行為
2. **小步前進**：每次只做一個重構
3. **測試保護**：重構前確認測試通過，重構後再跑一次
4. **編譯確認**：每步驟後編譯確認

## 常見重構手法
- Extract Method：將長方法拆分
- Extract Class：將大類別拆分
- Move Method：將方法移到更合適的類別
- Replace Conditional with Polymorphism
- Introduce Parameter Object
- Replace Magic Number with Constant

## 工作流程
1. 閱讀現有程式碼
2. 執行測試確認全部通過
3. 規劃重構步驟（TodoWrite）
4. 依序執行每個重構
5. 每步驟後執行測試
6. 提交可編譯、可測試的結果
"""

DOC_GEN_SYSTEM = """你是技術文件撰寫專家。

## 文件類型
1. **API 文件**：Swagger / OpenAPI 格式
2. **架構文件**：系統架構圖 + 元件說明
3. **開發指南**：環境建置 + 開發流程
4. **變更日誌**：版本變更紀錄

## 文件規範
- 使用 Markdown 格式
- 包含 Mermaid 圖表
- 提供程式碼範例
- 標示版本和最後更新時間

## 自動生成流程
1. 掃描專案結構（glob + grep）
2. 讀取關鍵程式碼
3. 分析架構和模式
4. 產出文件
"""

def decompose_task(high_level_task: str) -> list:
    """
    Lead Agent 的任務拆解策略
    
    原則：
    1. 每個子任務在 50% context 內可完成
    2. 明確定義輸入/輸出
    3. 建立清晰的相依關係
    """
    
    # 範例：拆解「建立使用者管理模組」
    tasks = [
        {
            "id": "task-001",
            "title": "設計使用者管理 API 規格",
            "assignee": "architect",
            "dependencies": [],
            "deliverables": ["api-spec.yaml"],
        },
        {
            "id": "task-002",
            "title": "建立 User Entity 和 Repository",
            "assignee": "backend",
            "dependencies": ["task-001"],
            "deliverables": ["User.java", "UserRepository.java"],
        },
        {
            "id": "task-003",
            "title": "實作 UserService（CRUD + 業務邏輯）",
            "assignee": "backend",
            "dependencies": ["task-002"],
            "deliverables": ["UserService.java"],
        },
        {
            "id": "task-004",
            "title": "實作 UserController（REST API）",
            "assignee": "backend",
            "dependencies": ["task-003"],
            "deliverables": ["UserController.java"],
        },
        {
            "id": "task-005",
            "title": "建立使用者管理前端頁面",
            "assignee": "frontend",
            "dependencies": ["task-001"],  # 只需要 API 規格即可開始
            "deliverables": ["UserList.vue", "UserForm.vue"],
        },
        {
            "id": "task-006",
            "title": "撰寫單元測試和整合測試",
            "assignee": "qa",
            "dependencies": ["task-003", "task-004"],
            "deliverables": ["UserServiceTest.java", "UserControllerTest.java"],
        },
        {
            "id": "task-007",
            "title": "安全掃描",
            "assignee": "security",
            "dependencies": ["task-004"],
            "deliverables": ["security-report.md"],
        },
    ]
    return tasks

class MailboxProtocol:
    """JSONL Mailbox 通訊協定"""
    
    def __init__(self, team_dir: str = ".team"):
        self.team_dir = team_dir
        os.makedirs(team_dir, exist_ok=True)
    
    def send_request(self, from_agent: str, to_agent: str, 
                     task_id: str, content: str):
        """發送任務請求"""
        msg = {
            "id": f"msg-{int(time.time()*1000)}",
            "from": from_agent,
            "to": to_agent,
            "type": "request",
            "task_id": task_id,
            "content": content,
            "status": "pending",
            "timestamp": datetime.now().isoformat(),
        }
        self._append(to_agent, msg)
        return msg["id"]
    
    def send_response(self, from_agent: str, to_agent: str,
                      task_id: str, status: str, result: str):
        """回覆任務結果"""
        msg = {
            "from": from_agent,
            "to": to_agent,
            "type": "response",
            "task_id": task_id,
            "status": status,  # "done" / "failed" / "blocked"
            "result": result,
            "timestamp": datetime.now().isoformat(),
        }
        self._append(to_agent, msg)
    
    def read_inbox(self, agent_name: str) -> list:
        """讀取信箱中的未處理訊息"""
        mailbox_path = f"{self.team_dir}/{agent_name}.mailbox.jsonl"
        if not os.path.exists(mailbox_path):
            return []
        
        with open(mailbox_path, "r") as f:
            messages = [json.loads(line) for line in f if line.strip()]
        
        # 回傳未處理的訊息
        return [m for m in messages if m.get("status") == "pending"]
    
    def _append(self, agent_name: str, msg: dict):
        path = f"{self.team_dir}/{agent_name}.mailbox.jsonl"
        with open(path, "a") as f:
            f.write(json.dumps(msg, ensure_ascii=False) + "\n")

Lead → Backend Mailbox:
  {"type": "request", "task_id": "task-003", 
   "content": "請根據 api-spec.yaml 實作 UserService，包含 CRUD 和密碼加密"}

Backend 處理:
  1. 讀取 api-spec.yaml
  2. 實作 UserService.java
  3. 執行單元測試
  4. 回報結果

Backend → Lead Mailbox:
  {"type": "response", "task_id": "task-003", "status": "done",
   "result": "已完成 UserService.java，6 個方法，12 個測試全部通過"}

Lead 收到回報:
  → 更新 task-003 狀態為 done
  → 檢查 task-006（QA）的相依是否滿足
  → 分派 task-006 給 QA Agent

class AgentLifecycle:
    """Agent 生命週期管理"""
    
    STATES = ["idle", "working", "waiting", "shutdown"]
    
    def __init__(self, agent_name: str):
        self.name = agent_name
        self.state = "idle"
        self.current_task = None
        self.events = []  # 事件日誌
    
    def transition(self, new_state: str, reason: str = ""):
        """狀態轉換"""
        old_state = self.state
        self.state = new_state
        event = {
            "agent": self.name,
            "from": old_state,
            "to": new_state,
            "reason": reason,
            "timestamp": datetime.now().isoformat(),
        }
        self.events.append(event)
        print(f"[{self.name}] {old_state} → {new_state}: {reason}")
    
    def claim_task(self, task_id: str):
        """認領任務"""
        self.current_task = task_id
        self.transition("working", f"開始處理 {task_id}")
    
    def complete_task(self):
        """完成任務"""
        task_id = self.current_task
        self.current_task = None
        self.transition("idle", f"完成 {task_id}")
    
    def request_shutdown(self):
        """請求關機"""
        self.transition("shutdown", "收到關機指令")

                    ┌───────────┐
         建立       │   idle    │◄────────────────┐
        ─────►     │  (閒置)   │                  │
                    └─────┬─────┘                  │
                          │ 認領任務                │ 完成任務
                          ▼                        │
                    ┌───────────┐                  │
                    │  working  │──────────────────┘
                    │  (工作中) │
                    └─────┬─────┘
                          │ 等待相依
                          ▼
                    ┌───────────┐
                    │  waiting  │──► 相依完成 ──► working
                    │  (等待)   │
                    └─────┬─────┘
                          │ 關機指令
                          ▼
                    ┌───────────┐
                    │ shutdown  │
                    │  (關機)   │
                    └───────────┘

Step 1: 使用者下達需求
─────────────────────
Prompt: "建立訂單管理系統，需要：
  - 訂單 CRUD API（Spring Boot）
  - 訂單列表/詳情/建立頁面（Vue 3）
  - 訂單狀態流轉（草稿→已提交→已確認→已完成/已取消）
  - 分頁 + 搜尋功能"

Step 2: Lead Agent 拆解任務
────────────────────────────
📋 任務清單：
  ⬜ #1 [Architect] 設計 API 規格（OpenAPI）
  ⬜ #2 [Backend]   建立 Order Entity + Repository
  ⬜ #3 [Backend]   實作 OrderService（含狀態機）
  ⬜ #4 [Backend]   實作 OrderController
  ⬜ #5 [Frontend]  建立 OrderList.vue
  ⬜ #6 [Frontend]  建立 OrderDetail.vue
  ⬜ #7 [Frontend]  建立 OrderForm.vue
  ⬜ #8 [QA]        撰寫 API 整合測試
  ⬜ #9 [QA]        撰寫 E2E 測試

Step 3: 各 Agent 依相依順序執行
──────────────────────────────
Architect Agent 產出 → api-spec.yaml
Backend Agent 依序產出：
  → Order.java（Entity + 狀態枚舉）
  → OrderRepository.java
  → OrderService.java（含狀態轉換邏輯）
  → OrderController.java
Frontend Agent（與 Backend 平行開始）：
  → orderApi.ts（根據 API 規格生成）
  → OrderList.vue（含分頁、搜尋）
  → OrderDetail.vue（含狀態流轉按鈕）
  → OrderForm.vue（含表單驗證）

REVERSE_ENG_SYSTEM = """你是逆向工程專家，負責分析舊系統程式碼。

## 分析流程
1. **結構探索**：掃描目錄結構，了解模組劃分
2. **進入點分析**：找到 main 方法、Controller、Servlet
3. **呼叫鏈追蹤**：從進入點追蹤到資料庫
4. **業務邏輯擷取**：識別核心業務規則
5. **資料模型分析**：分析 Entity / DAO / SQL
6. **文件產出**：產生架構文件和業務規則文件

## 產出物
- 系統架構圖（Mermaid）
- 模組清單與職責
- API 列表
- 資料庫 Schema 文件
- 業務規則文件
- 技術債清單
"""

Prompt: "分析 /legacy-app 目錄下的舊系統程式碼，產出：
  1. 系統架構文件
  2. API 清單
  3. 資料庫 Schema
  4. 核心業務邏輯文件
  5. 重構建議"

Agent 執行：
  1. glob("**/*.java") → 找到 387 個 Java 檔案
  2. grep("@Controller|@RestController|@RequestMapping") → 找到 23 個 Controller
  3. grep("@Entity|@Table") → 找到 45 個 Entity
  4. 逐一讀取關鍵檔案
  5. 產出分析報告

產出範例（架構圖）：

技術債清單（自動產出）：
┌────┬────────────────────────────┬──────┬──────┐
│ #  │ 問題                       │ 嚴重 │ 影響 │
├────┼────────────────────────────┼──────┼──────┤
│ 1  │ SQL 字串拼接（32 處）       │ High │ 安全 │
│ 2  │ 硬編碼密碼（5 處）          │ High │ 安全 │
│ 3  │ 無單元測試（覆蓋率 0%）     │ High │ 品質 │
│ 4  │ 巨型 Service（>500 行 x12）│ Med  │ 維護 │
│ 5  │ 已棄用 API 使用（Java 8）   │ Med  │ 升級 │
│ 6  │ 無日誌記錄（Exception catch 後空）│ Med │ 維運 │
└────┴────────────────────────────┴──────┴──────┘

UPGRADE_SYSTEM = """你是 Framework 升級專家。

## Spring Boot 2.x → 3.x 升級步驟

### Phase 1：評估
1. 掃描 pom.xml，列出所有依賴
2. 檢查 Java 版本（需 >= 17）
3. 分析 javax → jakarta 的影響範圍
4. 評估第三方套件相容性

### Phase 2：升級
1. 更新 parent pom 版本
2. 批次替換 javax → jakarta import
3. 更新 Spring Security 配置
4. 更新 Hibernate/JPA 配置
5. 處理已移除的 API

### Phase 3：驗證
1. 編譯專案
2. 執行全部測試
3. 逐一修復失敗
4. 執行整合測試
5. 產出升級報告
"""

Prompt: "將此專案從 Spring Boot 2.7 升級到 3.2，Java 11 升級到 21"

Agent 自動執行：

📋 任務清單：
  ✅ #1 掃描現有依賴（找到 47 個依賴）
  ✅ #2 分析 javax import（找到 156 處）
  ✅ #3 更新 pom.xml
     - spring-boot-starter-parent: 2.7.18 → 3.2.5
     - java.version: 11 → 21
  ✅ #4 批次替換 javax → jakarta
     - javax.persistence → jakarta.persistence（89 處）
     - javax.validation → jakarta.validation（34 處）
     - javax.servlet → jakarta.servlet（33 處）
  ✅ #5 更新 Spring Security 配置
     - WebSecurityConfigurerAdapter → SecurityFilterChain
  ✅ #6 編譯 → 12 錯誤
  ✅ #7 修復編譯錯誤
  ✅ #8 執行測試 → 5 失敗
  ✅ #9 修復測試
  ✅ #10 全部通過 → 產出升級報告

BATCH_REWRITE_SYSTEM = """你是批次系統改寫專家。

## 舊系統特徵
- Shell Script / COBOL / 舊版 Java
- FTP 檔案交換
- crontab 排程
- 固定格式文字檔案（Fixed-length / CSV / EBCDIC）

## 改寫策略
1. 分析現有批次流程（排程、輸入輸出、相依）
2. 設計新架構（Spring Batch / Apache Camel）
3. 實作 Reader → Processor → Writer
4. 建立測試資料
5. 比對新舊系統輸出

## 改寫原則
- 保持業務邏輯不變
- 使用 Spring Batch 的 chunk-oriented processing
- FTP → SFTP（安全升級）
- crontab → Spring Scheduler / Kubernetes CronJob
- 固定格式 → 結構化格式（JSON / Parquet）
- 加入監控和告警
"""

舊系統流程：
  1. crontab 每日 02:00 觸發
  2. Shell script 從 FTP 下載對帳檔
  3. COBOL 程式解析固定格式檔案
  4. 逐筆比對資料庫
  5. 產出差異報表
  6. FTP 上傳報表

新系統設計（Agent 產出）：

┌─────────────────────────────────────────┐
│         Spring Batch Job                 │
│                                         │
│  ┌──────────┐  ┌──────────┐  ┌────────┐│
│  │  SFTP    │  │ Processor│  │ Writer ││
│  │  Reader  │→│ (比對)   │→│ (報表) ││
│  └──────────┘  └──────────┘  └────────┘│
│                                         │
│  Step 1: Download from SFTP             │
│  Step 2: Parse → ReconciliationItem     │
│  Step 3: Compare with DB                │
│  Step 4: Generate Report                │
│  Step 5: Upload Report via SFTP         │
│  Step 6: Send Notification              │
└─────────────────────────────────────────┘

排程：Kubernetes CronJob
監控：Prometheus + Grafana
告警：PagerDuty / Teams Webhook

import os

DB_QUERY_TOOL = {
    "name": "db_query",
    "description": "對指定資料庫執行唯讀 SQL 查詢（僅 SELECT）。回傳 JSON 格式結果，最多 100 筆。",
    "input_schema": {
        "type": "object",
        "properties": {
            "database": {
                "type": "string",
                "enum": ["app_db", "audit_db", "analytics_db"],
                "description": "目標資料庫"
            },
            "sql": {
                "type": "string",
                "description": "SELECT SQL 語句（禁止 DDL/DML）"
            }
        },
        "required": ["database", "sql"]
    }
}

DB_SCHEMA_TOOL = {
    "name": "db_schema",
    "description": "查詢資料庫 Schema 資訊（表結構、欄位、索引）",
    "input_schema": {
        "type": "object",
        "properties": {
            "database": {"type": "string", "enum": ["app_db", "audit_db"]},
            "table_name": {"type": "string", "description": "資料表名稱（可選，不填則列出所有表）"}
        },
        "required": ["database"]
    }
}

def handle_db_query(database: str, sql: str) -> str:
    """安全的資料庫查詢"""
    import re
    
    # 安全檢查：僅允許 SELECT
    sql_upper = sql.strip().upper()
    if not sql_upper.startswith("SELECT"):
        return "ERROR: 僅允許 SELECT 查詢"
    
    # 禁止危險操作
    dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE", "EXEC"]
    for keyword in dangerous:
        if re.search(rf'\b{keyword}\b', sql_upper):
            return f"ERROR: 禁止使用 {keyword} 操作"
    
    # 限制結果筆數
    if "LIMIT" not in sql_upper and "FETCH FIRST" not in sql_upper:
        sql = sql.rstrip(";") + " LIMIT 100"
    
    # 執行查詢（示意）
    # conn = get_connection(database)
    # cursor = conn.cursor()
    # cursor.execute(sql)
    # results = cursor.fetchall()
    # return json.dumps(results, default=str)
    
    return "（資料庫查詢結果）"

def handle_db_schema(database: str, table_name: str = None) -> str:
    """查詢 Schema 資訊"""
    if table_name:
        # 查詢特定表結構
        sql = f"""
        SELECT column_name, data_type, is_nullable, column_default
        FROM information_schema.columns
        WHERE table_name = '{table_name}'
        ORDER BY ordinal_position
        """
    else:
        # 列出所有表
        sql = """
        SELECT table_name, table_type
        FROM information_schema.tables
        WHERE table_schema = 'public'
        ORDER BY table_name
        """
    return handle_db_query(database, sql)

┌───────────────────────────────────────┐
│          Agent Tool Layer              │
│                                       │
│  db_query ─── Connection Pool         │
│              ├── DB2    (JDBC)        │
│              ├── Oracle (JDBC)        │
│              └── PostgreSQL (JDBC)    │
│                                       │
│  db_schema ── Metadata Cache          │
│              └── 5 分鐘 TTL           │
│                                       │
│  db_migration ── Flyway / Liquibase   │
│              └── Version Control      │
└───────────────────────────────────────┘

MQ_TOOL = {
    "name": "mq_publish",
    "description": "發送訊息到指定的 Kafka Topic 或 RabbitMQ Queue",
    "input_schema": {
        "type": "object",
        "properties": {
            "broker": {"type": "string", "enum": ["kafka", "rabbitmq"]},
            "topic": {"type": "string", "description": "Topic 或 Queue 名稱"},
            "message": {"type": "string", "description": "訊息內容（JSON 格式）"},
            "key": {"type": "string", "description": "訊息 Key（Kafka 用，可選）"}
        },
        "required": ["broker", "topic", "message"]
    }
}

MQ_CONSUME_TOOL = {
    "name": "mq_consume",
    "description": "從指定的 Topic/Queue 讀取最近的訊息（用於除錯）",
    "input_schema": {
        "type": "object",
        "properties": {
            "broker": {"type": "string", "enum": ["kafka", "rabbitmq"]},
            "topic": {"type": "string"},
            "count": {"type": "integer", "description": "讀取筆數（最多 10）", "maximum": 10}
        },
        "required": ["broker", "topic"]
    }
}

Agent 開發微服務時的 MQ 整合流程：

1. Agent 分析業務需求 → 決定使用事件驅動架構
2. 設計 Event Schema（OrderCreatedEvent、PaymentCompletedEvent）
3. 生成 Producer 程式碼（Spring Kafka Template）
4. 生成 Consumer 程式碼（@KafkaListener）
5. 使用 mq_publish tool 測試訊息發送
6. 使用 mq_consume tool 驗證訊息接收

MICROSERVICE_SKILL = """
# 微服務開發 Skill

## 服務間通訊
- 同步：REST API（OpenFeign / WebClient）
- 異步：Kafka Event（Event-Driven）
- 服務發現：Consul / Eureka

## 共用元件
- API Gateway（Spring Cloud Gateway）
- Config Server（Spring Cloud Config）
- Circuit Breaker（Resilience4j）
- Distributed Tracing（Micrometer + Zipkin）

## 資料管理
- 每個服務獨立資料庫（Database per Service）
- 分散式交易（Saga Pattern）
- Event Sourcing + CQRS（複雜場景）

## 部署
- Docker Container
- Kubernetes Deployment
- Helm Chart
"""

CI_TOOL = {
    "name": "ci_trigger",
    "description": "觸發 CI/CD Pipeline 並等待結果",
    "input_schema": {
        "type": "object",
        "properties": {
            "platform": {"type": "string", "enum": ["github_actions", "jenkins"]},
            "pipeline": {"type": "string", "description": "Pipeline 名稱"},
            "parameters": {"type": "object", "description": "Pipeline 參數"}
        },
        "required": ["platform", "pipeline"]
    }
}

def handle_ci_trigger(platform: str, pipeline: str, 
                      parameters: dict = None) -> str:
    """觸發 CI/CD Pipeline"""
    if platform == "github_actions":
        # 使用 GitHub CLI
        cmd = f"gh workflow run {pipeline}"
        if parameters:
            for k, v in parameters.items():
                cmd += f" -f {k}={v}"
        return handle_bash(cmd)
    
    elif platform == "jenkins":
        # 使用 Jenkins CLI
        import urllib.request
        jenkins_url = os.getenv("JENKINS_URL")
        token = os.getenv("JENKINS_TOKEN")
        # ... Jenkins API 呼叫
        return "Pipeline 已觸發"

# .github/workflows/agent-ci.yml
name: Agent CI Pipeline

on:
  workflow_dispatch:
    inputs:
      task_id:
        description: 'Agent Task ID'
        required: true
      branch:
        description: 'Branch to build'
        required: true

jobs:
  build-and-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          ref: ${{ github.event.inputs.branch }}
      
      - name: Setup Java 21
        uses: actions/setup-java@v4
        with:
          java-version: '21'
          distribution: 'temurin'
      
      - name: Build and Test
        run: mvn clean verify
      
      - name: Security Scan
        run: mvn dependency-check:check
      
      - name: Notify Agent
        if: always()
        run: |
          echo "Task: ${{ github.event.inputs.task_id }}"
          echo "Status: ${{ job.status }}"

class RBACManager:
    """角色基礎存取控制"""
    
    ROLES = {
        "viewer": {
            "description": "僅能讀取",
            "tools": ["read_file", "glob", "grep", "db_query"],
            "paths": ["**/*"],
            "actions": ["read"],
        },
        "developer": {
            "description": "一般開發人員",
            "tools": [
                "read_file", "write_file", "edit_file",
                "glob", "grep", "bash", "run_tests",
                "todo_write", "db_query", "db_schema",
            ],
            "paths": ["src/**", "tests/**", "docs/**"],
            "actions": ["read", "write", "execute"],
            "blocked_paths": [".env", "secrets/**", "*.key", "*.pem"],
        },
        "tech_lead": {
            "description": "技術主管",
            "tools": ["*"],  # 所有工具
            "paths": ["**/*"],
            "actions": ["read", "write", "execute", "deploy"],
            "require_approval": ["ci_trigger", "db_migration"],
        },
        "admin": {
            "description": "系統管理員",
            "tools": ["*"],
            "paths": ["**/*"],
            "actions": ["*"],
        },
    }
    
    def __init__(self, role: str):
        if role not in self.ROLES:
            raise ValueError(f"未知角色: {role}")
        self.role = role
        self.config = self.ROLES[role]
    
    def can_use_tool(self, tool_name: str) -> bool:
        allowed = self.config["tools"]
        return "*" in allowed or tool_name in allowed
    
    def can_access_path(self, path: str) -> bool:
        import fnmatch
        blocked = self.config.get("blocked_paths", [])
        for pattern in blocked:
            if fnmatch.fnmatch(path, pattern):
                return False
        return True
    
    def needs_approval(self, tool_name: str) -> bool:
        return tool_name in self.config.get("require_approval", [])

                    企業內網
┌──────────────────────────────────────────────────┐
│                                                  │
│  Agent Runtime                                   │
│  ┌─────────────────────────────────────────────┐ │
│  │ Tool Pool（MCPToolPool）                      │ │
│  │ ├── 內建工具（bash, read_file, etc.）         │ │
│  │ └── MCP 動態工具（按需發現）                  │ │
│  └──────────────────┬──────────────────────────┘ │
│                     │ MCP Protocol (JSON-RPC 2.0) │
│     ┌───────────────┼────────────────┐           │
│     ▼               ▼                ▼           │
│  MCP Server 1   MCP Server 2    MCP Server 3     │
│  [GitHub]       [PostgreSQL]    [Jira 企業版]    │
│  - 建立 Issue   - 執行查詢     - 建立 Ticket     │
│  - Code Review  - Schema 瀏覽  - 更新 Sprint     │
│  - PR 管理      - Migration    - 讀取任務板      │
└──────────────────────────────────────────────────┘

# mcp_servers/internal_db_server.py
# 企業自建 MCP Server：封裝內部資料庫存取

from mcp.server import Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
import asyncio
import json

# 建立 MCP Server 實例
server = Server("internal-db-server")

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """宣告此 MCP Server 提供的工具清單"""
    return [
        types.Tool(
            name="query_app_db",
            description="查詢應用程式資料庫（唯讀 SELECT）",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "SELECT SQL 語句（禁止 DDL/DML）"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "最大回傳筆數（預設 50，最大 200）",
                        "default": 50
                    }
                },
                "required": ["sql"]
            }
        ),
        types.Tool(
            name="get_table_schema",
            description="查詢指定資料表的欄位結構",
            inputSchema={
                "type": "object",
                "properties": {
                    "table_name": {"type": "string"}
                },
                "required": ["table_name"]
            }
        ),
    ]

@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict
) -> list[types.TextContent]:
    """執行工具呼叫並回傳結果"""
    
    if name == "query_app_db":
        sql = arguments["sql"]
        limit = min(arguments.get("limit", 50), 200)
        
        # 安全檢查
        sql_upper = sql.strip().upper()
        if not sql_upper.startswith("SELECT"):
            return [types.TextContent(
                type="text",
                text="ERROR: 僅允許 SELECT 查詢"
            )]
        
        # 加入 LIMIT 限制
        if "LIMIT" not in sql_upper:
            sql = sql.rstrip(";") + f" LIMIT {limit}"
        
        # 執行查詢（使用連線池）
        # results = await db_pool.fetch(sql)
        # return [types.TextContent(type="text", text=json.dumps(results))]
        return [types.TextContent(type="text", text="（資料庫查詢結果）")]
    
    elif name == "get_table_schema":
        table = arguments["table_name"]
        schema_sql = f"""
        SELECT column_name, data_type, is_nullable, column_default
        FROM information_schema.columns
        WHERE table_name = '{table}'
        ORDER BY ordinal_position
        """
        # results = await db_pool.fetch(schema_sql)
        return [types.TextContent(type="text", text=f"Schema for {table}")]
    
    raise ValueError(f"未知工具: {name}")

async def main():
    """以 stdio 模式啟動 MCP Server"""
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream, write_stream,
            InitializationOptions(
                server_name="internal-db-server",
                server_version="1.0.0",
            )
        )

if __name__ == "__main__":
    asyncio.run(main())

{
  "mcpServers": {
    "internal-db": {
      "command": "python",
      "args": ["mcp_servers/internal_db_server.py"],
      "env": {
        "DB_HOST": "internal-postgres.company.com",
        "DB_PORT": "5432",
        "DB_NAME": "app_db",
        "DB_USER": "agent_readonly"
      }
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "${GITHUB_TOKEN}"
      }
    },
    "jira": {
      "command": "python",
      "args": ["mcp_servers/jira_server.py"],
      "env": {
        "JIRA_URL": "https://company.atlassian.net",
        "JIRA_TOKEN": "${JIRA_API_TOKEN}"
      }
    }
  }
}

class PromptInjectionGuard:
    """Prompt Injection 防護機制"""
    
    # 常見 Prompt Injection 模式
    INJECTION_PATTERNS = [
        r"ignore\s+(previous|above|all)\s+instructions",
        r"you\s+are\s+now\s+a",
        r"new\s+instructions?:",
        r"system\s*:\s*",
        r"forget\s+(everything|all|previous)",
        r"override\s+(system|instructions)",
        r"jailbreak",
        r"DAN\s+mode",
        r"pretend\s+you\s+are",
    ]
    
    @classmethod
    def check_input(cls, text: str) -> tuple:
        """檢查輸入是否包含注入嘗試"""
        import re
        for pattern in cls.INJECTION_PATTERNS:
            if re.search(pattern, text, re.IGNORECASE):
                return False, f"偵測到可疑的 Prompt Injection 模式: {pattern}"
        return True, "OK"
    
    @classmethod
    def sanitize_tool_output(cls, output: str) -> str:
        """清理 Tool 輸出中的潛在注入內容"""
        # 將可能被解讀為指令的內容用引號包裹
        if "SYSTEM:" in output.upper() or "INSTRUCTION:" in output.upper():
            return f"[Tool Output - Sanitized]\n```\n{output}\n```"
        return output
    
    @classmethod
    def check_file_content(cls, content: str, file_path: str) -> tuple:
        """檢查讀取的檔案內容是否包含注入嘗試"""
        safe, reason = cls.check_input(content)
        if not safe:
            return False, f"檔案 {file_path} 包含可疑內容: {reason}"
        return True, "OK"

def secure_agent_loop(user_message: str):
    """帶安全防護的 Agent Loop"""
    
    # 1. 檢查使用者輸入
    safe, reason = PromptInjectionGuard.check_input(user_message)
    if not safe:
        print(f"⚠️ 安全警告: {reason}")
        return
    
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM,
            max_tokens=4096, messages=messages, tools=TOOLS,
        )
        
        messages.append({"role": "assistant", "content": response.content})
        
        if response.stop_reason != "tool_use":
            break
        
        results = []
        for block in response.content:
            if block.type == "tool_use":
                # 2. 權限檢查
                allowed, reason = rbac.can_use_tool(block.name)
                if not allowed:
                    results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": f"PERMISSION_DENIED: {reason}",
                        "is_error": True,
                    })
                    continue
                
                # 3. 執行工具
                output = TOOL_HANDLERS[block.name](**block.input)
                
                # 4. 清理輸出
                output = PromptInjectionGuard.sanitize_tool_output(output)
                
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        
        messages.append({"role": "user", "content": results})

class ToolSecurityPolicy:
    """Tool 安全策略"""
    
    def __init__(self):
        self.audit_log = []
    
    def pre_execute_check(self, tool_name: str, tool_input: dict, 
                          agent_context: dict) -> tuple:
        """執行前安全檢查"""
        checks = [
            self._check_path_traversal,
            self._check_command_injection,
            self._check_sensitive_data_access,
            self._check_rate_limit,
        ]
        
        for check in checks:
            passed, reason = check(tool_name, tool_input, agent_context)
            if not passed:
                self._log_security_event("BLOCKED", tool_name, tool_input, reason)
                return False, reason
        
        self._log_security_event("ALLOWED", tool_name, tool_input, "")
        return True, "OK"
    
    def _check_path_traversal(self, tool_name, tool_input, ctx):
        """檢查路徑遍歷攻擊"""
        if "path" in tool_input:
            path = tool_input["path"]
            # 禁止 .. 路徑遍歷
            if ".." in path:
                return False, f"禁止路徑遍歷: {path}"
            # 禁止絕對路徑（限制在專案目錄內）
            if os.path.isabs(path):
                return False, f"禁止絕對路徑: {path}"
        return True, "OK"
    
    def _check_command_injection(self, tool_name, tool_input, ctx):
        """檢查命令注入"""
        if tool_name == "bash":
            cmd = tool_input.get("command", "")
            # 檢查危險的命令串接
            dangerous_chars = [";", "&&", "||", "|", "`", "$(",]
            for char in dangerous_chars:
                if char in cmd:
                    # 允許 pipe 但需要更嚴格的檢查
                    if char == "|":
                        continue  # pipe 在某些場景是合法的
                    return False, f"禁止命令中包含 '{char}'"
        return True, "OK"
    
    def _check_sensitive_data_access(self, tool_name, tool_input, ctx):
        """檢查敏感資料存取"""
        sensitive_patterns = [
            ".env", "password", "secret", "token", "credential",
            ".key", ".pem", ".p12", ".jks",
        ]
        
        if "path" in tool_input:
            path = tool_input["path"].lower()
            for pattern in sensitive_patterns:
                if pattern in path:
                    return False, f"禁止存取敏感檔案: {tool_input['path']}"
        
        return True, "OK"
    
    def _check_rate_limit(self, tool_name, tool_input, ctx):
        """速率限制"""
        # 最近 1 分鐘內的操作次數
        import time
        now = time.time()
        recent = [e for e in self.audit_log if now - e.get("timestamp", 0) < 60]
        if len(recent) > 100:
            return False, "操作過於頻繁，請稍後再試（1 分鐘限 100 次）"
        return True, "OK"
    
    def _log_security_event(self, action, tool_name, tool_input, reason):
        """記錄安全事件"""
        import time
        self.audit_log.append({
            "timestamp": time.time(),
            "action": action,
            "tool": tool_name,
            "input_summary": str(tool_input)[:200],
            "reason": reason,
        })

SECURITY_SCAN_TOOL = {
    "name": "security_scan",
    "description": "對程式碼執行安全掃描（SAST）",
    "input_schema": {
        "type": "object",
        "properties": {
            "scan_type": {
                "type": "string",
                "enum": ["sast", "dependency", "secret"],
                "description": "掃描類型"
            },
            "target": {
                "type": "string",
                "description": "掃描目標路徑"
            }
        },
        "required": ["scan_type"]
    }
}

def handle_security_scan(scan_type: str, target: str = ".") -> str:
    """執行安全掃描"""
    if scan_type == "sast":
        # 使用 SonarQube / Semgrep
        return handle_bash(f"semgrep --config=auto {target} --json")
    
    elif scan_type == "dependency":
        # 使用 OWASP Dependency Check
        return handle_bash("mvn dependency-check:check")
    
    elif scan_type == "secret":
        # 使用 git-secrets / trufflehog
        return handle_bash(f"trufflehog filesystem {target} --json")
    
    return "ERROR: 不支援的掃描類型"

import json
import time
from datetime import datetime

class AuditLogger:
    """操作審計日誌"""
    
    def __init__(self, log_file: str = "audit.jsonl"):
        self.log_file = log_file
    
    def log(self, event_type: str, agent_name: str, 
            tool_name: str, tool_input: dict, 
            result_summary: str, user_id: str = None):
        """記錄操作事件"""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "event_type": event_type,  # TOOL_CALL, PERMISSION_DENIED, ERROR, etc.
            "agent_name": agent_name,
            "user_id": user_id or "system",
            "tool_name": tool_name,
            "tool_input": self._sanitize_input(tool_input),
            "result_summary": result_summary[:500],
            "session_id": os.getenv("SESSION_ID", "unknown"),
        }
        
        with open(self.log_file, "a") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    
    def _sanitize_input(self, tool_input: dict) -> dict:
        """移除敏感資訊"""
        sanitized = {}
        sensitive_keys = ["password", "token", "secret", "api_key"]
        for k, v in tool_input.items():
            if any(s in k.lower() for s in sensitive_keys):
                sanitized[k] = "***REDACTED***"
            else:
                sanitized[k] = str(v)[:200]
        return sanitized
    
    def query(self, agent_name: str = None, 
              start_time: str = None, end_time: str = None) -> list:
        """查詢審計日誌"""
        with open(self.log_file, "r") as f:
            entries = [json.loads(line) for line in f if line.strip()]
        
        if agent_name:
            entries = [e for e in entries if e["agent_name"] == agent_name]
        if start_time:
            entries = [e for e in entries if e["timestamp"] >= start_time]
        if end_time:
            entries = [e for e in entries if e["timestamp"] <= end_time]
        
        return entries

# 金融業合規 Agent 配置
FINANCIAL_COMPLIANCE_CONFIG = {
    # 資料分級
    "data_classification": {
        "public": ["API 文件", "README"],
        "internal": ["架構文件", "開發規範"],
        "confidential": ["客戶資料", "交易記錄"],
        "restricted": ["加密金鑰", "認證憑證"],
    },
    
    # Agent 存取限制
    "access_rules": {
        # Agent 不可直接接觸客戶個資
        "no_direct_pii_access": True,
        # 所有 DB 查詢必須審計
        "audit_all_db_queries": True,
        # 程式碼不可包含硬編碼憑證
        "no_hardcoded_credentials": True,
        # 所有外部 API 呼叫必須加密（TLS 1.2+）
        "enforce_tls": True,
    },
    
    # 合規標準
    "compliance_standards": [
        "OWASP Top 10",
        "PCI-DSS（信用卡資料）",
        "ISO 27001",
        "金管會資安管理辦法",
    ],
    
    # 自動合規檢查
    "auto_checks": [
        "dependency_vulnerability_scan",
        "static_code_analysis",
        "secret_detection",
        "license_compliance",
    ],
}

┌──────────────────────────────────────────────┐
│              安全邊界（Trust Boundary）         │
│                                              │
│  ┌──────────────────┐   ┌────────────────┐   │
│  │  Agent Runtime    │   │  Audit System  │   │
│  │  ┌────────────┐  │   │                │   │
│  │  │ Permission │  │   │  • 操作日誌    │   │
│  │  │ Manager    │──┼───│  • 異常偵測    │   │
│  │  └────────────┘  │   │  • 合規報告    │   │
│  │  ┌────────────┐  │   │  • 告警通知    │   │
│  │  │ Data       │  │   └────────────────┘   │
│  │  │ Sanitizer  │  │                        │
│  │  └────────────┘  │   ┌────────────────┐   │
│  │  ┌────────────┐  │   │  Secret Vault  │   │
│  │  │ Rate       │  │   │  (HashiCorp    │   │
│  │  │ Limiter    │  │   │   Vault)       │   │
│  │  └────────────┘  │   └────────────────┘   │
│  └──────────────────┘                        │
│                                              │
│  ⛔ Agent 不可存取：                           │
│  • 生產資料庫（使用脫敏資料）                   │
│  • 客戶個資（PII）                             │
│  • 加密金鑰（透過 Vault API 間接使用）          │
│  • 網路設備配置                                │
└──────────────────────────────────────────────┘

1. 需求分析（5 分鐘）
   - 閱讀需求文件
   - 確認範圍和驗收標準
   - 決定使用哪些 Agent

2. 撰寫 Prompt（10 分鐘）
   - 描述需求（What）
   - 指定技術規範（How）
   - 定義驗收條件（Done）

3. Agent 執行（自動，15-60 分鐘）
   - Agent 規劃任務
   - 依序執行開發
   - 自動測試和修復

4. 人工 Review（20 分鐘）
   - 檢查架構決策
   - 驗證業務邏輯
   - 確認安全合規
   - 驗證測試覆蓋率

5. 提交（5 分鐘）
   - Git commit with meaningful message
   - 建立 Pull Request
   - 觸發 CI/CD

## 任務描述
[用 1-3 句話說明要做什麼]

## 技術規範
- 語言/框架：[Java 21 / Spring Boot 3.2]
- 架構模式：[Clean Architecture / 三層式]
- 資料庫：[PostgreSQL / JPA]

## 具體需求
1. [需求 1]
2. [需求 2]
3. [需求 3]

## 驗收條件
- [ ] 所有測試通過
- [ ] 覆蓋率 >= 80%
- [ ] 無 SAST 高風險項目
- [ ] API 有 Swagger 文件

## 注意事項
- [例外處理方式]
- [命名規範]
- [其他限制]

建立 StringUtils.java 工具類別，包含：
- isEmpty(String s): 檢查空字串（含 null 和空白）
- truncate(String s, int maxLen): 截斷字串
- toCamelCase(String s): snake_case 轉 camelCase
每個方法都要有 JavaDoc 和完整的 JUnit 測試。

為現有的 Order 模組新增「訂單匯出 Excel」功能：

技術規範：
- 使用 Apache POI 4.x
- Spring Boot 3.2 / Java 21

需求：
1. 新增 OrderExportService，支援條件查詢後匯出
2. 匯出欄位：訂單編號、客戶名稱、金額、狀態、建立日期
3. 支援日期範圍篩選
4. 新增 REST API：GET /api/orders/export?startDate=&endDate=
5. 回傳 Content-Type: application/vnd.openxmlformats-officedocument.spreadsheetml.sheet

驗收條件：
- 匯出 10,000 筆資料在 5 秒內完成
- 有完整的單元測試
- 有錯誤處理（無資料時回傳 204）

設計並實作「即時通知系統」微服務：

架構要求：
- 獨立 Spring Boot 微服務
- 使用 Kafka 接收事件
- 支援 Email / SMS / Push 三種通知管道
- 使用 Strategy Pattern 管理管道
- 通知模板使用 Thymeleaf

功能需求：
1. Kafka Consumer 監聽 notification-topic
2. NotificationEvent 包含：type, channel, recipient, templateId, variables
3. 每種管道有獨立的 Sender 實作
4. 通知記錄持久化到 PostgreSQL
5. 提供 REST API 查詢通知歷史

非功能需求：
- 重試機制：失敗後最多重試 3 次（指數退避）
- 限流：每秒最多 100 封 Email
- 監控：Prometheus metrics（發送量、失敗率、延遲）

請依序產出：
1. 專案結構和 pom.xml
2. Entity + Repository
3. Kafka Consumer
4. NotificationService + Strategy
5. 各管道 Sender
6. REST Controller
7. 完整測試

常見 Debug 方式：

1. 查看 Agent 的任務規劃
   - 檢查 TodoWrite 的輸出是否合理
   - 任務是否遺漏重要步驟

2. 檢查 Tool 執行結果
   - 每次 Tool call 的輸入和輸出
   - 是否有 ERROR 被忽略

3. Context 分析
   - 當前 messages[] 的 Token 用量
   - 是否因 Context 過長導致「遺忘」

4. 重現問題
   - 記錄導致問題的 Prompt
   - 在乾淨的 Context 中重試
   - 使用 subagent 隔離問題

5. 常用 Debug 指令
   > "請解釋你剛才為什麼選擇這個做法"
   > "請列出你目前的任務清單和進度"
   > "請重新規劃任務，考慮 XXX 限制"

✅ DO（好的做法）
─────────────────
1. 明確指定技術堆疊和版本
2. 提供具體的驗收條件
3. 複雜任務先請 Agent 規劃（Plan Mode）
4. 用範例說明預期輸出
5. 指出「不要」做什麼（避免過度設計）

❌ DON'T（壞的做法）
─────────────────
1. 模糊描述：「幫我做個登入功能」
2. 過於龐大：一次塞入 10 個功能
3. 無驗收標準：「做好就好」
4. 矛盾指令：同時要求簡潔和詳盡
5. 跳過 Review：直接合併 Agent 產出

判斷何時需要多個 Agent：

單一 Agent 適用場景：
  • 修改 1-3 個檔案
  • 單一模組的 CRUD
  • Bug 修復
  • 小型重構

Multi-Agent 適用場景：
  • 跨模組開發（前後端 + DB）
  • 大型功能（> 10 個檔案）
  • 需要不同專業知識
  • 需要平行處理

Agent 拆分原則：
  1. 按職責拆分（SRP for Agents）
  2. 每個 Agent 有明確的角色和工具集
  3. 通訊透過 Mailbox，不共享 Context
  4. 相依關係清晰（DAG）
  5. 可獨立失敗和重試

Token 消耗優化策略：

1. 選擇合適的模型
   ┌────────────────────────────────────────┐
   │ 任務類型           建議模型    相對成本 │
   │ 架構設計           Opus       $$$      │
   │ 程式碼開發         Sonnet     $$       │
   │ 文件產生           Haiku/DeepSeek $    │
   │ 格式轉換           Haiku      $        │
   └────────────────────────────────────────┘

2. 減少不必要的 Context
   - 使用行號範圍讀取檔案（非全檔）
   - 及時 compact（50% 時觸發）
   - Subagent 隔離避免 Context 膨脹

3. 避免無效重試
   - 設定 max_retries = 3
   - 設定 max_loop_iterations = 50
   - 失敗時回報而非無限重試

4. 批次處理
   - 相似任務批次處理（降低 overhead）
   - 使用 background task 平行執行

5. 預估成本
   - 每個 Prompt 約消耗 1K-5K input tokens
   - 每個 Tool call 約消耗 500-2K tokens
   - 完整開發任務約 50K-200K tokens
   - 按 Sonnet 費率：約 $0.15-$0.60 / 任務

# CLAUDE.md 企業模板（最佳實踐版）

## 角色定位（Role）
你是 [公司名稱] 的企業級 AI 開發助理，專注於 [領域]。
本地 repo 路徑：/home/agent/workspace
主要程式語言：TypeScript + Python

## 硬性規則（Hard Rules）— 永遠遵守，不得例外
- 禁止直接推送到 main/master 分支
- 禁止刪除 production 環境資源
- 所有外部 API 呼叫必須使用 .env 中的配置，不得硬編碼憑證
- 資料庫操作前必須確認備份存在

## 工作流程（Workflow）
1. 接收任務時，先閱讀 docs/architecture.md 了解系統架構
2. 修改前先執行 git status 確認工作目錄乾淨
3. 每完成一個子任務，更新 TODO.md 中的進度
4. PR 前必須通過 npm run lint && npm run test

## 目錄結構慣例（Project Structure）
src/
  api/         # REST API 路由層
  services/    # 業務邏輯層（禁止在此層直接呼叫 DB）
  repositories/# 資料存取層
  models/      # TypeORM Entity 定義

## 常用指令備忘（Quick Commands）
- 啟動開發環境：docker-compose up -d
- 執行測試：npm run test:watch
- 查看 log：docker logs -f app --tail=100

# prompt_cache_config.py
# 正確設定 Prompt Caching（企業 Harness 應在此層實作）

import anthropic

client = anthropic.Anthropic()

def build_cached_request(user_message: str, tools: list) -> dict:
    """建構帶 Prompt Cache 標記的請求"""
    
    # System Prompt 主體（超過 1024 tokens 才觸發快取）
    system_content = [
        {
            "type": "text",
            "text": load_base_system_prompt(),  # 基礎角色定義（穩定、不變）
            "cache_control": {"type": "ephemeral"}  # 標記為可快取
        },
        {
            "type": "text",
            "text": load_project_context(),  # 專案背景（每日更新一次）
            "cache_control": {"type": "ephemeral"}  # 另一個快取層
        },
        {
            "type": "text",
            "text": f"今日日期：{get_today()}｜任務背景：{get_session_context()}"
            # 不加 cache_control：每次請求都變化，不值得快取
        }
    ]
    
    return {
        "model": "claude-sonnet-4-6",
        "max_tokens": 8192,
        "system": system_content,
        "tools": tools,
        "messages": [{"role": "user", "content": user_message}]
    }

def load_base_system_prompt() -> str:
    """載入穩定的基礎 System Prompt（快取命中率高）"""
    # 這部分應包含：角色定義、工具使用規則、輸出格式要求
    # 避免包含任何動態內容（時間、用戶名稱、任務ID）
    with open("prompts/base_system.md") as f:
        return f.read()

def load_project_context() -> str:
    """載入專案背景（每日重新快取）"""
    # 這部分包含：架構摘要、程式碼慣例、API 端點清單
    # 可以在 CI/CD 時自動更新
    with open("prompts/project_context.md") as f:
        return f.read()

# token_budget_manager.py
# 企業 Agent 的 Token 預算控制層

class TokenBudgetManager:
    """
    Token 預算三層架構：
    - Session 預算：單次 Agent 任務的上限
    - Daily 預算：每日總消耗上限（接入帳務系統）
    - 模型選擇策略：依任務複雜度動態選擇模型
    """
    
    MODEL_PRICING = {
        "claude-opus-4-8":    {"input": 15.0, "output": 75.0},   # per 1M tokens
        "claude-sonnet-4-6":  {"input": 3.0,  "output": 15.0},
        "claude-haiku-4-5-20251001": {"input": 0.8, "output": 4.0},
    }
    
    def __init__(self, session_budget_usd: float = 1.0):
        self.session_budget = session_budget_usd
        self.spent = 0.0
        self.call_count = 0
    
    def select_model(self, task_complexity: str) -> str:
        """依複雜度選擇最符合成本效益的模型"""
        mapping = {
            "high":   "claude-opus-4-8",         # 架構設計、複雜推理
            "medium": "claude-sonnet-4-6",        # 一般開發任務
            "low":    "claude-haiku-4-5-20251001" # 格式轉換、文件生成
        }
        return mapping.get(task_complexity, "claude-sonnet-4-6")
    
    def record_usage(self, model: str, input_tokens: int, output_tokens: int):
        """記錄 Token 使用量並計算費用"""
        pricing = self.MODEL_PRICING[model]
        cost = (input_tokens / 1_000_000 * pricing["input"] +
                output_tokens / 1_000_000 * pricing["output"])
        self.spent += cost
        self.call_count += 1
        
        if self.spent > self.session_budget * 0.8:
            print(f"[警告] 已消耗 {self.spent:.4f} USD，接近預算上限")
    
    def check_budget(self) -> bool:
        """檢查是否仍有剩餘預算"""
        return self.spent < self.session_budget
    
    def summary(self) -> dict:
        return {
            "total_calls": self.call_count,
            "total_spent_usd": round(self.spent, 4),
            "budget_remaining_usd": round(self.session_budget - self.spent, 4),
            "budget_utilization_pct": round(self.spent / self.session_budget * 100, 1)
        }

通用 Agent Harness 架構（跨領域）：

  ┌───────────────────────────────────────────────────┐
  │                    Agent Loop                     │
  │                                                   │
  │  感知(Observe) → 推理(Reason) → 行動(Act) → 回饋  │
  └───────────────────┬───────────────────────────────┘
                      │
  ┌───────────────────┴───────────────────────────────┐
  │                    Harness                        │
  │                                                   │
  │  ┌──────────┐ ┌──────────┐ ┌──────────┐          │
  │  │  Tools   │ │Knowledge │ │ Context  │          │
  │  │ (領域)   │ │ (領域)   │ │ Manager  │          │
  │  └──────────┘ └──────────┘ └──────────┘          │
  │  ┌──────────┐ ┌──────────┐ ┌──────────┐          │
  │  │Permission│ │  Task    │ │ Memory   │          │
  │  │ Boundary │ │ System   │ │ System   │          │
  │  └──────────┘ └──────────┘ └──────────┘          │
  └───────────────────────────────────────────────────┘

被動 Agent（Chat → Response）：

  使用者 → "幫我寫一個函式" → Agent 執行 → 完成 → 結束

常駐 Agent（Always-On）：

  ┌─────────────────────────────────────────────────┐
  │                                                 │
  │  ┌──────┐   ┌──────┐   ┌──────┐   ┌──────┐    │
  │  │Cron  │   │Event │   │IM    │   │Webhook│   │
  │  │定時  │   │事件  │   │即時  │   │外部   │    │
  │  │巡檢  │   │觸發  │   │通訊  │   │觸發   │    │
  │  └──┬───┘   └──┬───┘   └──┬───┘   └──┬───┘    │
  │     └──────────┴──────────┴──────────┘         │
  │                    │                            │
  │              ┌─────┴─────┐                      │
  │              │ Agent Core│                      │
  │              │ (Model +  │                      │
  │              │  Harness) │                      │
  │              └─────┬─────┘                      │
  │                    │                            │
  │     ┌──────────────┼──────────────┐             │
  │     │              │              │             │
  │  ┌──┴──┐      ┌────┴───┐    ┌────┴──┐         │
  │  │Memory│      │Actions │    │Report │         │
  │  │記憶  │      │行動    │    │報告   │          │
  │  └─────┘      └────────┘    └───────┘          │
  │                                                 │
  └─────────────────────────────────────────────────┘

# 安裝
npm i -g @shareai-lab/kode

# 使用
kode "重構 UserService，將驗證邏輯抽出為獨立類別"

import { Agent, Tool, Skill } from '@shareai-lab/kode-agent-sdk';

// 建立自訂 Agent
const agent = new Agent({
  model: 'claude-sonnet-4-6',
  tools: [readFile, writeFile, bash, dbQuery],
  skills: ['./skills/company-coding-style'],
  permissions: {
    allowedPaths: ['src/', 'tests/'],
    blockedCommands: ['rm -rf', 'DROP TABLE'],
    requireApproval: ['git push', 'deploy'],
  },
});

// 執行任務
const result = await agent.run('分析並修復 OrderService 中的效能問題');

learn-claude-code 的 Agent：         claw0 的 Agent：

  User → Prompt → Agent → Done       Agent ──────────────────┐
                                      │ Heartbeat（健康檢查）  │
                                      │ Cron（排程巡檢）       │
                                      │ IM（即時通訊整合）     │
                                      │ Memory（跨 Session）   │
                                      │ Soul（人格與風格）     │
                                      └────────────────────────┘
                                      永遠在線，主動行動

## 任務：實作使用者認證模組

### 技術規格
- Spring Boot 3.2 + Spring Security 6
- JWT Token（Access Token + Refresh Token）
- BCrypt 密碼加密
- PostgreSQL

### 功能需求
1. POST /api/auth/register — 使用者註冊
2. POST /api/auth/login — 登入取得 Token
3. POST /api/auth/refresh — 刷新 Token
4. POST /api/auth/logout — 登出（Token 黑名單）

### 資料模型
- User: id, username, email, passwordHash, roles, createdAt, updatedAt
- RefreshToken: id, userId, token, expiresAt, createdAt

### 安全要求
- Access Token 有效期 15 分鐘
- Refresh Token 有效期 7 天
- 密碼需符合複雜度（>= 8 字元，含大小寫和數字）
- 登入失敗 5 次鎖定 30 分鐘

### 驗收條件
- [ ] 所有 API 可正常運作
- [ ] 單元測試覆蓋率 >= 80%
- [ ] 包含整合測試
- [ ] 無 OWASP Top 10 弱點

## 問題描述
OrderService.calculateTotal() 在處理折扣時計算錯誤。

## 重現步驟
1. 建立訂單包含 3 個商品
2. 套用 20% 折扣
3. 預期總金額：800，實際：1000（折扣未生效）

## 相關檔案
- src/main/java/com/example/service/OrderService.java
- src/test/java/com/example/service/OrderServiceTest.java

## 請求
1. 找到 Bug 原因
2. 修復問題
3. 新增測試覆蓋此情境
4. 確認所有既有測試仍通過

## 重構目標
將 UserService.java（目前 520 行）重構為符合 SRP 的多個類別。

## 重構範圍
- src/main/java/com/example/service/UserService.java

## 限制
- 不改變 public API 介面
- 所有既有測試必須通過
- 每次重構後編譯確認

## 預期結果
- UserService（核心 CRUD，< 100 行）
- UserValidationService（驗證邏輯）
- UserNotificationService（通知邏輯）
- UserStatisticsService（統計邏輯）

## 分析目標
分析 /legacy-app/src 目錄下的舊系統程式碼，產出以下文件：

1. **系統架構文件**
   - 模組劃分和職責
   - 系統架構圖（Mermaid）
   - 模組間依賴關係

2. **API 清單**
   - 所有 REST API 端點
   - 請求/回應格式
   - 認證方式

3. **資料庫文件**
   - ER 圖（Mermaid）
   - 表結構說明
   - 關鍵索引

4. **業務規則文件**
   - 核心業務邏輯說明
   - 狀態流轉規則
   - 驗證規則

5. **技術債清單**
   - 安全問題
   - 效能問題
   - 可維護性問題
   - 建議修復優先順序

# ===== 基礎操作 =====

# 啟動 Agent（各 Session）
python agents/s01_agent_loop.py       # 基礎迴圈
python agents/s02_tool_use.py         # 多工具
python agents/s03_todo_write.py       # 計畫能力
python agents/s04_subagent.py         # 子 Agent
python agents/s05_skills.py           # Skills 載入
python agents/s06_context_compact.py  # Context 壓縮
python agents/s07_tasks.py            # 任務管理
python agents/s08_background.py       # 背景任務
python agents/s09_agent_teams.py      # Agent 團隊
python agents/s10_team_protocols.py   # 團隊協定
python agents/s11_autonomous.py       # 自治 Agent
python agents/s12_worktree_task_isolation.py  # Worktree 隔離
python agents/s_full.py               # 完整版

# ===== Web 平台 =====
cd web && npm install && npm run dev  # http://localhost:3000

# ===== 環境管理 =====
cp .env.example .env                  # 建立環境變數檔
pip install -r requirements.txt       # 安裝依賴
python -m pytest tests/               # 執行測試

# ===== Git Worktree（s12 使用）=====
git worktree add worktrees/task-001 -b feature/task-001  # 建立隔離空間
git worktree list                     # 列出所有 worktree
git worktree remove worktrees/task-001  # 清理 worktree