diff --git a/docs/pipeline-design.md b/docs/pipeline-design.md
new file mode 100644
index 00000000..8009de4d
--- /dev/null
+++ b/docs/pipeline-design.md
@@ -0,0 +1,335 @@
+# 自动开发管道：多 Agent Plan + Review 设计方案
+
+## 目标流程
+
+```
+用户消息
+  → Step 1: Plan Agent 生成方案
+  → Step 2: N 个 Review Agent 并行审查方案 → 汇总 → 不通过则回 Step 1（最多 2 轮）
+  → Step 3: Implement Agent 写代码 + 跑测试
+  → Step 4: N 个 Review Agent 并行审查代码 → 汇总 → 不通过则回 Step 3（最多 2 轮）
+  → Step 5: Push + PR
+```
+
+简单问答、代码探索等非修改任务不走此管道，沿用现有单次 execute() 路径。
+
+---
+
+## 架构概览
+
+```
+event-handler.ts
+  │
+  ├── 简单任务 → claudeExecutor.execute()       (现有路径)
+  │
+  └── 代码修改 → PipelineOrchestrator.run()      (新增)
+                    │
+                    ├── Step planAgent       → 1x query()
+                    ├── Step planReview      → Nx query() 并行
+                    ├── Step implementAgent  → 1x query()
+                    ├── Step codeReview      → Nx query() 并行
+                    └── Step pushAgent       → 1x query()
+```
+
+### 新增文件结构
+
+```
+src/pipeline/
+  types.ts          # PipelinePhase, PipelineState, ReviewVerdict 等类型定义
+  orchestrator.ts   # PipelineOrchestrator — 状态机驱动的多步编排
+  reviewer.ts       # 并行 review 调度 + 结果聚合
+  prompts.ts        # 各角色 system prompt
+```
+
+---
+
+## 核心设计
+
+### 1. 状态机 (orchestrator.ts)
+
+```typescript
+type PipelinePhase =
+  | 'plan'
+  | 'plan_review'
+  | 'implement'
+  | 'code_review'
+  | 'push'
+  | 'done'
+  | 'failed';
+
+interface PipelineState {
+  phase: PipelinePhase;
+  userPrompt: string;
+  workingDir: string;
+  plan?: string;              // Step 1 输出
+  planReview?: ReviewResult;  // Step 2 输出
+  implementSummary?: string;  // Step 3 输出 (diff 摘要)
+  codeReview?: ReviewResult;  // Step 4 输出
+  pushResult?: string;        // Step 5 输出 (PR 链接等)
+  retries: Record<string, number>;
+}
+
+interface PipelineCallbacks {
+  onPhaseChange?: (phase: PipelinePhase, detail?: string) => Promise<void>;
+  onStreamUpdate?: (text: string) => Promise<void>;
+}
+
+class PipelineOrchestrator {
+  async run(
+    prompt: string,
+    workingDir: string,
+    callbacks: PipelineCallbacks,
+  ): Promise<PipelineState> {
+    let state: PipelineState = {
+      phase: 'plan',
+      userPrompt: prompt,
+      workingDir,
+      retries: {},
+    };
+
+    while (state.phase !== 'done' && state.phase !== 'failed') {
+      await callbacks.onPhaseChange?.(state.phase);
+
+      switch (state.phase) {
+        case 'plan':
+          state = await this.doPlan(state, callbacks);
+          break;
+        case 'plan_review':
+          state = await this.doReview(state, 'plan', callbacks);
+          break;
+        case 'implement':
+          state = await this.doImplement(state, callbacks);
+          break;
+        case 'code_review':
+          state = await this.doReview(state, 'code', callbacks);
+          break;
+        case 'push':
+          state = await this.doPush(state, callbacks);
+          break;
+      }
+    }
+
+    return state;
+  }
+}
+```
+
+**状态转移规则:**
+
+```
+plan ──成功──→ plan_review
+plan ──失败──→ failed
+
+plan_review ──通过──→ implement
+plan_review ──拒绝 & retries < 2──→ plan (携带反馈)
+plan_review ──拒绝 & retries >= 2──→ failed
+
+implement ──成功──→ code_review
+implement ──失败──→ failed
+
+code_review ──通过──→ push
+code_review ──拒绝 & retries < 2──→ implement (携带反馈)
+code_review ──拒绝 & retries >= 2──→ failed
+
+push ──成功──→ done
+push ──失败──→ failed (但代码已写好，报告手动步骤)
+```
+
+### 2. 并行 Review (reviewer.ts)
+
+```typescript
+interface ReviewAgent {
+  role: string;
+  systemPrompt: string;
+}
+
+interface ReviewVerdict {
+  role: string;
+  approved: boolean;
+  summary: string;
+  issues: string[];
+}
+
+interface ReviewResult {
+  approved: boolean;
+  verdicts: ReviewVerdict[];
+  consolidated: string;
+}
+
+async function parallelReview(
+  agents: ReviewAgent[],
+  content: string,
+  workingDir: string,
+): Promise<ReviewResult> {
+  const promises = agents.map(agent =>
+    claudeExecutor.execute({
+      sessionKey: `review-${agent.role}-${Date.now()}`,
+      prompt: `审查以下内容，严格按格式输出：\n\n${content}`,
+      workingDir,
+      // review agent 独立运行，不继承会话、不保存摘要
+    })
+  );
+
+  const results = await Promise.allSettled(promises);
+  const verdicts = results.map((r, i) => parseVerdict(r, agents[i]));
+
+  // 聚合策略：任一 agent 明确 REJECTED → 整体不通过
+  const approved = verdicts.every(v => v.approved);
+  const consolidated = verdicts
+    .map(v => `### [${v.role}] ${v.approved ? 'APPROVED' : 'REJECTED'}\n${v.summary}`)
+    .join('\n\n');
+
+  return { approved, verdicts, consolidated };
+}
+```
+
+**输出解析策略:**
+- 要求 review agent 输出第一行为 `APPROVED` 或 `REJECTED`
+- 解析失败时默认 `REJECTED`（宁可多审一轮，不放过问题）
+- 单个 agent 超时/崩溃 → 该 agent 视为弃权，不阻塞整体
+
+### 3. 角色化 System Prompts (prompts.ts)
+
+```typescript
+export const PLAN_AGENT_PROMPT = `你是一个技术方案设计师。
+根据用户需求，分析现有代码，输出结构化的实施方案。
+
+输出格式：
+## 需求理解
+(一句话总结)
+
+## 影响范围
+(列出需要修改的文件和原因)
+
+## 实施步骤
+(编号列表，每步具体到函数级别)
+
+## 风险点
+(可能出问题的地方)
+`;
+
+export const REVIEW_AGENTS: ReviewAgent[] = [
+  {
+    role: 'correctness',
+    systemPrompt: `你是代码正确性审查员。
+审查维度：逻辑错误、边界条件、类型安全、错误处理遗漏。
+不关注代码风格。
+输出格式：第一行 APPROVED 或 REJECTED，后跟问题列表。`,
+  },
+  {
+    role: 'security',
+    systemPrompt: `你是安全审查员。
+审查维度：注入漏洞、权限绕过、敏感信息泄露、不安全的依赖。
+不关注功能正确性。
+输出格式：第一行 APPROVED 或 REJECTED，后跟问题列表。`,
+  },
+  {
+    role: 'architecture',
+    systemPrompt: `你是架构审查员。
+审查维度：与现有代码风格一致性、抽象层次合理性、可维护性、接口设计。
+不关注具体实现细节。
+输出格式：第一行 APPROVED 或 REJECTED，后跟问题列表。`,
+  },
+];
+
+export const IMPLEMENT_AGENT_PROMPT = `你是一个高级开发工程师。
+根据已审批的技术方案，精确实施代码修改。
+
+规则：
+- 严格按方案执行，不擅自扩展范围
+- 写完代码后运行测试
+- 测试失败则修复，最多重试 2 轮
+- 不要 git add . 或 git add -A
+- 不要提交 .env 等敏感文件
+`;
+```
+
+### 4. 飞书卡片集成
+
+新增 `buildPipelineCard()` 显示管道进度：
+
+```
+┌──────────────────────────────────────┐
+│ 🤖 Claude Code - 自动开发管道        │  (蓝色 header)
+├──────────────────────────────────────┤
+│ 指令: 添加用户注册功能...            │
+├──────────────────────────────────────┤
+│ ✅ 方案设计    → 已完成              │
+│ ✅ 方案审查    → 3/3 通过            │
+│ 🔄 代码实现    → 执行中...           │
+│ ⬚ 代码审查                          │
+│ ⬚ 推送 & PR                         │
+├──────────────────────────────────────┤
+│ ⏳ 阶段 3/5 | ⏱️ 45s                │
+└──────────────────────────────────────┘
+```
+
+### 5. event-handler.ts 集成
+
+```typescript
+async function executeClaudeTask(prompt, chatId, userId, messageId, rootId) {
+  // ... 现有的 session / thread 逻辑 ...
+
+  const isPipelineTask = detectCodeTask(prompt);
+
+  if (isPipelineTask) {
+    const orchestrator = new PipelineOrchestrator();
+    const finalState = await orchestrator.run(prompt, session.workingDir, {
+      onPhaseChange: async (phase) => {
+        if (progressMsgId) {
+          await feishuClient.updateCard(progressMsgId, buildPipelineCard(prompt, phase));
+        }
+      },
+      onStreamUpdate,
+    });
+    // 根据 finalState 构建最终结果卡片
+  } else {
+    // 现有单次 execute() 逻辑
+  }
+}
+```
+
+`detectCodeTask()` 可以用关键词匹配 + 让第一轮 plan agent 自行判断是否需要走管道。
+
+---
+
+## 可靠性保障
+
+| 风险 | 对策 |
+|------|------|
+| Review Agent 输出格式不可控 | 强制首行 APPROVED/REJECTED，解析失败默认 REJECTED |
+| 单个 review agent 超时/崩溃 | `Promise.allSettled`；失败 agent 视为弃权，不阻塞 |
+| Plan 反复不通过死循环 | 每个 phase 硬限 2 次重试，超出则 failed，报告分歧 |
+| Pipeline 中途进程崩溃 | PipelineState 可序列化到 DB，重启后从上一个完成的 phase 恢复 |
+| 并行 review 资源消耗 | Review agent 设低预算 (maxBudgetUsd: 0.5, maxTurns: 10) |
+| 用户等待焦虑 | 每个 phase 切换更新卡片，review 阶段显示 "2/3 agents 完成" |
+| Review 反馈传递失真 | 把 consolidated review 原文注入下一轮 plan/implement 的 prompt |
+
+---
+
+## 成本估算
+
+单次完整管道（无重试）：
+
+| 步骤 | query 次数 |
+|------|-----------|
+| Plan | 1 |
+| Plan Review | 3 (并行) |
+| Implement | 1 |
+| Code Review | 3 (并行) |
+| Push | 1 |
+| **合计** | **9** |
+
+有重试最多 9 + 2×(1+3) = **17 次**。建议在管道触发前向用户确认。
+
+---
+
+## 实施建议
+
+分两步：
+
+1. **Phase A**: Pipeline 状态机 + 阶段卡片（不带并行 review，先用单 agent self-review）
+   - 验证多步编排的稳定性、卡片更新、错误恢复
+2. **Phase B**: 接入并行 Review Agent
+   - 调优各角色 prompt，校准 approved/rejected 阈值
+   - 性能调优（并行度、超时、预算）
diff --git a/src/claude/__tests__/executor.test.ts b/src/claude/__tests__/executor.test.ts
index c3460442..5a31fd08 100644
--- a/src/claude/__tests__/executor.test.ts
+++ b/src/claude/__tests__/executor.test.ts
@@ -59,6 +59,16 @@ function setupMessages(messages: Array<Record<string, unknown>>) {
   });
 }
 
+/** Shorthand for building an ExecuteInput with defaults */
+function makeInput(overrides: Record<string, unknown> = {}) {
+  return {
+    sessionKey: 'chat1:user1',
+    prompt: 'test prompt',
+    workingDir: '/tmp/work',
+    ...overrides,
+  };
+}
+
 beforeEach(() => {
   vi.clearAllMocks();
   // Default: yield a simple success result
@@ -94,15 +104,11 @@ describe('ClaudeExecutor', () => {
       ]);
 
       const externalCallback = vi.fn();
-      const resultPromise = executor.execute(
-        'chat1:user1', 'test prompt', '/tmp/work',
-        undefined, undefined, externalCallback,
-      );
-
-      // 模拟 MCP tool 在迭代过程中调用 onWorkspaceChanged
-      // 由于 mock 的 async iterator 是同步 resolve 的，这里需要在 query 构建后触发
-      // 实际上 capturedOnWorkspaceChanged 会在 createWorkspaceMcpServer 调用时被捕获
-      // 手动触发
+      const resultPromise = executor.execute(makeInput({
+        onWorkspaceChanged: externalCallback,
+      }));
+
+      // 手动触发 workspace 变更
       if (capturedOnWorkspaceChanged) {
         capturedOnWorkspaceChanged('/new/workspace');
       }
@@ -116,10 +122,9 @@ describe('ClaudeExecutor', () => {
     });
 
     it('should not set needsRestart when workspace does not change', async () => {
-      const result = await executor.execute(
-        'chat1:user1', 'test prompt', '/tmp/work',
-        undefined, undefined, vi.fn(),
-      );
+      const result = await executor.execute(makeInput({
+        onWorkspaceChanged: vi.fn(),
+      }));
 
       expect(result.needsRestart).toBeFalsy();
       expect(result.newWorkingDir).toBeUndefined();
@@ -137,10 +142,9 @@ describe('ClaudeExecutor', () => {
         { type: 'result', subtype: 'error', session_id: 'sess-1', errors: ['something failed'], duration_ms: 50 },
       ]);
 
-      const promise = executor.execute(
-        'chat1:user1', 'test', '/tmp/work',
-        undefined, undefined, vi.fn(),
-      );
+      const promise = executor.execute(makeInput({
+        onWorkspaceChanged: vi.fn(),
+      }));
       capturedCb?.('/new/dir');
       const result = await promise;
 
@@ -152,11 +156,9 @@ describe('ClaudeExecutor', () => {
 
   describe('disableWorkspaceTool', () => {
     it('should not create MCP server when disableWorkspaceTool is true', async () => {
-      await executor.execute(
-        'chat1:user1', 'test', '/tmp/work',
-        undefined, undefined, undefined,
-        { disableWorkspaceTool: true },
-      );
+      await executor.execute(makeInput({
+        disableWorkspaceTool: true,
+      }));
 
       // createWorkspaceMcpServer should NOT be called
       expect(mockCreateWorkspaceMcpServer).not.toHaveBeenCalled();
@@ -167,10 +169,9 @@ describe('ClaudeExecutor', () => {
     });
 
     it('should create MCP server when disableWorkspaceTool is not set', async () => {
-      await executor.execute(
-        'chat1:user1', 'test', '/tmp/work',
-        undefined, undefined, vi.fn(),
-      );
+      await executor.execute(makeInput({
+        onWorkspaceChanged: vi.fn(),
+      }));
 
       expect(mockCreateWorkspaceMcpServer).toHaveBeenCalledTimes(1);
       const queryCallOptions = mockQuery.mock.calls[0][0].options;
@@ -180,7 +181,7 @@ describe('ClaudeExecutor', () => {
 
   describe('options overrides', () => {
     it('should use default maxTurns and maxBudgetUsd', async () => {
-      await executor.execute('chat1:user1', 'test', '/tmp/work');
+      await executor.execute(makeInput());
 
       const opts = mockQuery.mock.calls[0][0].options;
       expect(opts.maxTurns).toBe(50);
@@ -188,11 +189,10 @@ describe('ClaudeExecutor', () => {
     });
 
     it('should override maxTurns and maxBudgetUsd from options', async () => {
-      await executor.execute(
-        'chat1:user1', 'test', '/tmp/work',
-        undefined, undefined, undefined,
-        { maxTurns: 5, maxBudgetUsd: 0.5 },
-      );
+      await executor.execute(makeInput({
+        maxTurns: 5,
+        maxBudgetUsd: 0.5,
+      }));
 
       const opts = mockQuery.mock.calls[0][0].options;
       expect(opts.maxTurns).toBe(5);
@@ -202,10 +202,7 @@ describe('ClaudeExecutor', () => {
 
   describe('workspace changed callback wrapping', () => {
     it('should not wrap when onWorkspaceChanged is undefined', async () => {
-      await executor.execute(
-        'chat1:user1', 'test', '/tmp/work',
-        undefined, undefined, undefined,
-      );
+      await executor.execute(makeInput());
 
       // createWorkspaceMcpServer should be called with undefined (no wrapping)
       expect(mockCreateWorkspaceMcpServer).toHaveBeenCalledWith(undefined);
@@ -213,10 +210,9 @@ describe('ClaudeExecutor', () => {
 
     it('should wrap when onWorkspaceChanged is provided', async () => {
       const cb = vi.fn();
-      await executor.execute(
-        'chat1:user1', 'test', '/tmp/work',
-        undefined, undefined, cb,
-      );
+      await executor.execute(makeInput({
+        onWorkspaceChanged: cb,
+      }));
 
       // createWorkspaceMcpServer should be called with a wrapper function (not the original cb)
       const passedCb = mockCreateWorkspaceMcpServer.mock.calls[0][0];
diff --git a/src/claude/executor.ts b/src/claude/executor.ts
index 556a6abb..09e3ff9d 100644
--- a/src/claude/executor.ts
+++ b/src/claude/executor.ts
@@ -14,6 +14,20 @@ import type { ClaudeResult, ExecuteOptions, ProgressCallback } from './types.js'
 // SDK 会自动管理工具执行、权限、流式输出等
 // ============================================================
 
+/** 执行 Claude query 的完整参数 */
+export interface ExecuteInput extends ExecuteOptions {
+  sessionKey: string;
+  prompt: string;
+  workingDir: string;
+  resumeSessionId?: string;
+  onProgress?: ProgressCallback;
+  onWorkspaceChanged?: (newDir: string) => void;
+  onStreamUpdate?: (text: string) => Promise<void>;
+  historySummaries?: string;
+  /** 覆盖 system prompt（用于 pipeline 各角色独立 prompt） */
+  systemPromptOverride?: string;
+}
+
 /** 工作区管理系统提示词 */
 const WORKSPACE_SYSTEM_PROMPT = `你正在通过飞书消息与用户交互。请保持回复简洁，适合在聊天消息中阅读。
 
@@ -41,7 +55,33 @@ const WORKSPACE_SYSTEM_PROMPT = `你正在通过飞书消息与用户交互。
 - 用户明确表示要在当前目录操作
 
 **重要：调用 setup_workspace 后，系统将自动重启以加载项目配置（CLAUDE.md 等）。
-请在调用后仅输出简短确认（如"工作区已就绪，正在重新加载项目配置..."），不要继续执行后续任务。**`;
+请在调用后仅输出简短确认（如"工作区已就绪，正在重新加载项目配置..."），不要继续执行后续任务。**
+
+## 自动开发流程
+
+当用户给出明确的代码修改任务（写功能、修 bug、重构等）时，自动按以下流程执行：
+
+1. 理解需求，确认工作目录和代码结构
+2. 检查当前分支：\`git branch\`。如果在 main/master/develop 上，先创建特性分支：\`git checkout -b feat/<描述性名称>\`
+3. 编写/修改代码
+4. 发现项目测试命令（查看 package.json scripts、Makefile 等）并运行测试
+   - 如果没有测试命令，跳过测试步骤并在报告中说明
+5. 如果测试失败：分析错误 → 修复 → 重新测试
+   - 最多重试 2 轮
+   - 如果相同测试以相同方式连续失败 2 次，停止重试，向用户说明根因
+6. 推送前预检：
+   - \`git remote -v\` 确认 origin 存在
+   - \`gh auth status\` 确认 GitHub CLI 已认证
+   - 如果任一检查失败，跳过推送/PR 步骤，报告已完成的工作和需要手动处理的部分
+7. 测试通过后：\`git add\` 相关文件 → \`git commit\` → \`git push -u origin\` → \`gh pr create\`
+8. 最后汇报：改了什么、测试结果、PR 链接
+
+规则：
+- commit message 格式遵循项目约定（查看 git log --oneline -5 学习风格）
+- 不要 git add . 或 git add -A，只添加本次变更的文件
+- 不要提交 .env、credentials 等敏感文件
+- 如果某步骤失败且无法自动修复，停下来向用户说明情况
+- 如果用户只是提问、审查代码或做探索性修改，不需要走这个流程。不确定时问用户："需要我提交这些改动并创建 PR 吗？"`;
 
 export class ClaudeExecutor {
   /** 运行中的 query 实例 (用于 abort) */
@@ -49,24 +89,14 @@ export class ClaudeExecutor {
 
   /**
    * 执行 Claude Agent SDK query
-   *
-   * @param sessionKey  会话标识 (chatId:userId)
-   * @param prompt      用户输入的指令
-   * @param workingDir  工作目录
-   * @param resumeSessionId  可选：恢复之前的会话
-   * @param onProgress  进度回调
-   * @param onWorkspaceChanged  工作区变更回调 (MCP 工具 clone 后更新 session)
-   * @param options     可选参数 (maxTurns, maxBudgetUsd, disableWorkspaceTool)
    */
-  async execute(
-    sessionKey: string,
-    prompt: string,
-    workingDir: string,
-    resumeSessionId?: string,
-    onProgress?: ProgressCallback,
-    onWorkspaceChanged?: (newDir: string) => void,
-    options?: ExecuteOptions,
-  ): Promise<ClaudeResult> {
+  async execute(input: ExecuteInput): Promise<ClaudeResult> {
+    const {
+      sessionKey, prompt, workingDir, resumeSessionId,
+      onProgress, onWorkspaceChanged, onStreamUpdate, historySummaries,
+      systemPromptOverride, disableWorkspaceTool, maxTurns, maxBudgetUsd,
+    } = input;
+
     const startTime = Date.now();
     const abortController = new AbortController();
 
@@ -96,10 +126,17 @@ export class ClaudeExecutor {
     // 每次 query 创建独立的 MCP 服务器实例，通过闭包绑定当前 session 的回调
     // 确保多 chat 并发执行时互不干扰
     // restart 时通过 disableWorkspaceTool 完全移除 setup_workspace，防止无限循环
-    const mcpServers = options?.disableWorkspaceTool
+    const mcpServers = disableWorkspaceTool
       ? undefined
       : { 'workspace-manager': createWorkspaceMcpServer(onWorkspaceChangedWrapped) };
 
+    // 构建 systemPrompt.append 内容
+    // pipeline 模式使用独立的 system prompt，不需要工作区管理指引
+    const baseAppend = systemPromptOverride ?? WORKSPACE_SYSTEM_PROMPT;
+    const promptAppend = historySummaries
+      ? baseAppend + `\n\n## 历史会话摘要\n以下是该用户之前的会话记录，帮助你了解项目上下文：\n${historySummaries}`
+      : baseAppend;
+
     // 构建 SDK query
     const q = query({
       prompt,
@@ -118,14 +155,14 @@ export class ClaudeExecutor {
         // 权限：acceptEdits 自动接受文件编辑，canUseTool 自动批准其余工具调用
         // 注意：不使用 bypassPermissions，因为 root 用户下会被拒绝
         permissionMode: 'acceptEdits',
-        canUseTool: async (toolName: string, input: Record<string, unknown>) => {
-          logger.info({ toolName, inputKeys: Object.keys(input) }, 'canUseTool called — auto allowing');
+        canUseTool: async (toolName: string, inputObj: Record<string, unknown>) => {
+          logger.info({ toolName, inputKeys: Object.keys(inputObj) }, 'canUseTool called — auto allowing');
           return { behavior: 'allow' as const };
         },
 
         // 预算和限制
-        maxTurns: options?.maxTurns ?? 50,
-        maxBudgetUsd: options?.maxBudgetUsd ?? 5,
+        maxTurns: maxTurns ?? 50,
+        maxBudgetUsd: maxBudgetUsd ?? 5,
 
         // 会话续接
         ...(resumeSessionId ? { resume: resumeSessionId } : {}),
@@ -134,7 +171,7 @@ export class ClaudeExecutor {
         systemPrompt: {
           type: 'preset',
           preset: 'claude_code',
-          append: WORKSPACE_SYSTEM_PROMPT,
+          append: promptAppend,
         },
 
         // 加载项目设置 (CLAUDE.md 等)
@@ -152,6 +189,12 @@ export class ClaudeExecutor {
     let sessionId: string | undefined;
     let resultMessage: SDKMessage | undefined;
 
+    // 流式更新状态
+    let lastStreamTime = Date.now();
+    let lastStreamLen = 0;
+    let lastStreamPromise: Promise<void> | undefined;
+    let streamFailed = 0;
+
     try {
       // 遍历 SDK 流式消息
       for await (const message of q) {
@@ -178,6 +221,17 @@ export class ClaudeExecutor {
                 }
               }
             }
+
+            // 流式卡片更新（节流：3秒 或 500字符，连续失败 3 次后停止）
+            if (onStreamUpdate && streamFailed < 3) {
+              const now = Date.now();
+              const newChars = output.length - lastStreamLen;
+              if (now - lastStreamTime >= 3000 || newChars >= 500) {
+                lastStreamTime = now;
+                lastStreamLen = output.length;
+                lastStreamPromise = onStreamUpdate(output).catch(() => { streamFailed++; });
+              }
+            }
             break;
 
           case 'result':
@@ -192,7 +246,6 @@ export class ClaudeExecutor {
     } catch (err) {
       this.runningQueries.delete(sessionKey);
 
-
       const durationMs = Date.now() - startTime;
       const errorMsg = err instanceof Error ? err.message : String(err);
       logger.error({ sessionKey, err: errorMsg }, 'Claude Agent SDK query error');
@@ -206,6 +259,9 @@ export class ClaudeExecutor {
       };
     }
 
+    // 等待最后一个流式更新完成，防止与最终卡片更新竞态
+    if (lastStreamPromise) await lastStreamPromise.catch(() => {});
+
     this.runningQueries.delete(sessionKey);
     const durationMs = Date.now() - startTime;
 
diff --git a/src/feishu/__tests__/message-builder.test.ts b/src/feishu/__tests__/message-builder.test.ts
index 5592445b..341ea3d6 100644
--- a/src/feishu/__tests__/message-builder.test.ts
+++ b/src/feishu/__tests__/message-builder.test.ts
@@ -1,5 +1,5 @@
 import { describe, it, expect } from 'vitest';
-import { buildProgressCard, buildResultCard, buildStatusCard } from '../message-builder.js';
+import { buildProgressCard, buildResultCard, buildStreamingCard, buildPipelineCard, buildStatusCard } from '../message-builder.js';
 
 describe('buildProgressCard', () => {
   it('should build a card with the given prompt', () => {
@@ -80,6 +80,89 @@ describe('buildResultCard', () => {
   });
 });
 
+describe('buildStreamingCard', () => {
+  it('should display prompt, content, and elapsed time', () => {
+    const card = buildStreamingCard('do something', 'partial output here', 15) as any;
+    expect(card.header.template).toBe('blue');
+    expect(card.header.title.content).toContain('执行中');
+    expect(card.elements[0].text.content).toContain('do something');
+    expect(card.elements[2].text.content).toContain('partial output here');
+    const note = card.elements[4];
+    expect(note.elements[0].content).toContain('15s');
+  });
+
+  it('should show last 2500 chars of long content', () => {
+    const longContent = 'A'.repeat(1000) + 'B'.repeat(2500);
+    const card = buildStreamingCard('test', longContent, 5) as any;
+    const displayed = card.elements[2].text.content;
+    expect(displayed).toContain('...');
+    expect(displayed).not.toContain('A');
+    expect(displayed).toContain('B');
+  });
+
+  it('should show placeholder when content is empty', () => {
+    const card = buildStreamingCard('test', '', 0) as any;
+    expect(card.elements[2].text.content).toContain('正在处理...');
+  });
+});
+
+describe('buildPipelineCard', () => {
+  it('should show in-progress phase with correct marker', () => {
+    const card = buildPipelineCard('task', 'implement', 3, 5, 30) as any;
+    expect(card.header.template).toBe('blue');
+    expect(card.header.title.content).toContain('自动开发管道');
+    const phasesContent = card.elements[2].text.content as string;
+    expect(phasesContent).toContain('✅ 1. 方案设计');
+    expect(phasesContent).toContain('✅ 2. 方案审查');
+    expect(phasesContent).toContain('🔄 3. 代码实现 ← 当前');
+    expect(phasesContent).toContain('⬚ 4. 代码审查');
+    expect(phasesContent).toContain('⬚ 5. 推送 & PR');
+    const note = card.elements[card.elements.length - 1];
+    expect(note.elements[0].content).toContain('阶段 3/5');
+    expect(note.elements[0].content).toContain('30s');
+  });
+
+  it('should show all checkmarks when done', () => {
+    const card = buildPipelineCard('task', 'done', 6, 5, 120, 0.72) as any;
+    expect(card.header.template).toBe('green');
+    expect(card.header.title.content).toContain('管道完成');
+    const phasesContent = card.elements[2].text.content as string;
+    expect(phasesContent).toContain('✅ 1.');
+    expect(phasesContent).toContain('✅ 5.');
+    expect(phasesContent).not.toContain('⬚');
+    expect(phasesContent).not.toContain('🔄');
+    const note = card.elements[card.elements.length - 1];
+    expect(note.elements[0].content).toContain('✅ 完成');
+    expect(note.elements[0].content).toContain('$0.7200');
+  });
+
+  it('should show failure state correctly', () => {
+    const card = buildPipelineCard('task', 'failed', 3, 5, 60) as any;
+    expect(card.header.template).toBe('red');
+    expect(card.header.title.content).toContain('管道失败');
+    const phasesContent = card.elements[2].text.content as string;
+    expect(phasesContent).toContain('✅ 1.');
+    expect(phasesContent).toContain('✅ 2.');
+    expect(phasesContent).toContain('❌ 3.');
+    expect(phasesContent).toContain('⬚ 4.');
+  });
+
+  it('should include detail section when provided', () => {
+    const card = buildPipelineCard('task', 'implement', 3, 5, 10, undefined, 'some detail') as any;
+    // detail should appear between phases and note
+    const allTexts = card.elements.map((e: any) => e.text?.content ?? '').join(' ');
+    expect(allTexts).toContain('some detail');
+  });
+
+  it('should truncate long detail to 2000 chars', () => {
+    const longDetail = 'X'.repeat(3000);
+    const card = buildPipelineCard('task', 'plan', 1, 5, 5, undefined, longDetail) as any;
+    const allTexts = card.elements.map((e: any) => e.text?.content ?? '').join(' ');
+    expect(allTexts).toContain('...');
+    expect(allTexts.length).toBeLessThan(3000 + 500);
+  });
+});
+
 describe('buildStatusCard', () => {
   it('should display working dir, status, and pending tasks', () => {
     const card = buildStatusCard('/home/user/project', 'idle', 3) as any;
diff --git a/src/feishu/event-handler.ts b/src/feishu/event-handler.ts
index a5e0cd11..a2bddf6b 100644
--- a/src/feishu/event-handler.ts
+++ b/src/feishu/event-handler.ts
@@ -4,7 +4,9 @@ import { isUserAllowed, containsDangerousCommand } from '../utils/security.js';
 import { sessionManager } from '../session/manager.js';
 import { taskQueue } from '../session/queue.js';
 import { claudeExecutor } from '../claude/executor.js';
-import { buildProgressCard, buildResultCard, buildStatusCard } from './message-builder.js';
+import { buildProgressCard, buildResultCard, buildStreamingCard, buildPipelineCard, buildStatusCard } from './message-builder.js';
+import { PipelineOrchestrator } from '../pipeline/orchestrator.js';
+import { PHASE_META, TOTAL_PHASES } from '../pipeline/types.js';
 import { feishuClient } from './client.js';
 import { config } from '../config.js';
 import { setupWorkspace } from '../workspace/manager.js';
@@ -342,6 +344,27 @@ async function handleSlashCommand(
     return true;
   }
 
+  // /dev <task> - 自动开发管道（绕过 taskQueue，使用 acquireSession 并发保护）
+  if (trimmed.startsWith('/dev ')) {
+    const task = trimmed.slice('/dev '.length).trim();
+    if (!task) {
+      const reply = '⚠️ 用法: `/dev <开发任务描述>`';
+      if (threadRootMsgId) {
+        await feishuClient.replyTextInThread(threadRootMsgId, reply);
+      } else {
+        await feishuClient.replyText(messageId, reply);
+      }
+      return true;
+    }
+    // 安全检查
+    if (containsDangerousCommand(task)) {
+      await feishuClient.replyText(messageId, '⚠️ 检测到危险命令，已拒绝执行');
+      return true;
+    }
+    await executePipelineTask(task, chatId, userId, messageId, rootId);
+    return true;
+  }
+
   // /help - 帮助
   if (trimmed === '/help') {
     const helpText = [
@@ -352,6 +375,7 @@ async function handleSlashCommand(
       '**可用命令:**',
       '`/project <path>` - 切换工作目录',
       '`/workspace <url|path> [branch]` - 创建隔离工作区 (自动 clone + 创建分支)',
+      '`/dev <task>` - 自动开发管道 (方案→审查→实现→审查→推送)',
       '`/status` - 查看当前会话状态',
       '`/reset` - 重置会话',
       '`/stop` - 中断当前执行',
@@ -412,6 +436,23 @@ async function ensureThread(
   return undefined;
 }
 
+/**
+ * 并发保护：原子地尝试获取会话锁（CAS: idle → busy）
+ * @returns true 如果成功获取锁，false 如果已被占用
+ */
+async function acquireSession(
+  chatId: string,
+  userId: string,
+  messageId: string,
+): Promise<boolean> {
+  // 原子 CAS：UPDATE ... WHERE status != 'busy'，单条 SQL 防止 TOCTOU 竞态
+  if (!sessionManager.tryAcquire(chatId, userId)) {
+    await feishuClient.replyText(messageId, '⏳ 当前会话正在执行任务，请等待完成或使用 /stop 中断');
+    return false;
+  }
+  return true;
+}
+
 /**
  * 执行 Claude Agent SDK 任务
  * 支持 workspace 变更后自动 restart：第一次 query 触发 setup_workspace 后，
@@ -442,6 +483,25 @@ async function executeClaudeTask(
   // 标记会话为忙碌
   sessionManager.setStatus(chatId, userId, 'busy');
 
+  // 获取历史摘要用于注入 system prompt
+  const summaries = sessionManager.getRecentSummaries(chatId, userId, 5);
+  let historySummaries: string | undefined;
+  if (summaries.length > 0) {
+    let combined = summaries.join('\n');
+    if (combined.length > 3000) {
+      combined = combined.slice(-3000);
+    }
+    historySummaries = combined;
+  }
+
+  // 构造流式卡片更新回调
+  const executeStartTime = Date.now();
+  const onStreamUpdate = async (text: string) => {
+    if (!progressMsgId) return;
+    const elapsed = Math.floor((Date.now() - executeStartTime) / 1000);
+    await feishuClient.updateCard(progressMsgId, buildStreamingCard(prompt, text, elapsed));
+  };
+
   // workspace 变更回调: MCP 工具 clone 后自动更新 session.workingDir
   const onWorkspaceChanged = (newDir: string) => {
     sessionManager.setWorkingDir(chatId, userId, newDir);
@@ -454,14 +514,16 @@ async function executeClaudeTask(
 
   try {
     // 第一次 query：可能触发 workspace setup
-    const result = await claudeExecutor.execute(
+    const result = await claudeExecutor.execute({
       sessionKey,
       prompt,
-      session.workingDir,
-      session.conversationId,
+      workingDir: session.workingDir,
+      resumeSessionId: session.conversationId,
       onProgress,
       onWorkspaceChanged,
-    );
+      onStreamUpdate,
+      historySummaries,
+    });
 
     // 检测是否需要 restart（workspace 变更后重新执行以加载 CLAUDE.md）
     if (result.needsRestart && result.newWorkingDir) {
@@ -501,15 +563,15 @@ async function executeClaudeTask(
       // - 不传 resumeSessionId（全新 session）
       // - 不传 onWorkspaceChanged（不触发二次 restart）
       // - disableWorkspaceTool: 完全移除 setup_workspace MCP tool，防止无限循环
-      const restartResult = await claudeExecutor.execute(
+      const restartResult = await claudeExecutor.execute({
         sessionKey,
         prompt,
-        result.newWorkingDir,
-        undefined,
+        workingDir: result.newWorkingDir,
         onProgress,
-        undefined,
-        { disableWorkspaceTool: true },
-      );
+        onStreamUpdate,
+        historySummaries,
+        disableWorkspaceTool: true,
+      });
 
       // 保存 restart query 的 session_id 用于下次续接
       // 如果 restart query 失败未返回 sessionId，用第一次 query 的作为 fallback
@@ -538,6 +600,18 @@ async function executeClaudeTask(
       prompt, result, result.durationMs, result.costUsd,
       progressMsgId, threadRootMsgId, chatId,
     );
+
+    // 保存会话摘要（取输出末尾 500 字符作为摘要）
+    if (result.success && result.output && result.output.length > 100) {
+      try {
+        const date = new Date().toISOString().slice(0, 10);
+        const tail = result.output.slice(-500).trim();
+        const summary = `[${date}] dir: ${session.workingDir} | ${tail}`;
+        sessionManager.saveSummary(chatId, userId, session.workingDir, summary);
+      } catch (err) {
+        logger.warn({ err }, 'Failed to save session summary');
+      }
+    }
   } catch (err) {
     logger.error({ err }, 'Error executing Claude Agent SDK query');
     const errorReply = `❌ 执行出错: ${(err as Error).message}`;
@@ -597,6 +671,149 @@ async function sendResultCard(
   }
 }
 
+/**
+ * 执行自动开发管道（/dev 命令触发）
+ * 绕过 taskQueue，使用 acquireSession 进行并发保护
+ */
+async function executePipelineTask(
+  prompt: string,
+  chatId: string,
+  userId: string,
+  messageId: string,
+  rootId?: string,
+): Promise<void> {
+  if (!await acquireSession(chatId, userId, messageId)) return;
+
+  const session = sessionManager.getOrCreate(chatId, userId);
+  const pipelineStartTime = Date.now();
+
+  // 确保话题存在
+  const threadRootMsgId = await ensureThread(chatId, userId, messageId, rootId);
+
+  // 发送管道初始卡片
+  let progressMsgId: string | undefined;
+  const initialCard = buildPipelineCard(prompt, 'plan', 1, TOTAL_PHASES, 0);
+  if (threadRootMsgId) {
+    progressMsgId = await feishuClient.replyCardInThread(threadRootMsgId, initialCard);
+  }
+  if (!progressMsgId) {
+    progressMsgId = await feishuClient.sendCard(chatId, initialCard);
+  }
+
+  // 获取历史摘要
+  const summaries = sessionManager.getRecentSummaries(chatId, userId, 5);
+  let historySummaries: string | undefined;
+  if (summaries.length > 0) {
+    let combined = summaries.join('\n');
+    if (combined.length > 3000) {
+      combined = combined.slice(-3000);
+    }
+    historySummaries = combined;
+  }
+
+  try {
+    const orchestrator = new PipelineOrchestrator();
+
+    // 跟踪当前 phase 供 onStreamUpdate 使用
+    let currentPipelinePhase: string = 'plan';
+    let currentPhaseIndex = 1;
+
+    const pipelineResult = await orchestrator.run(
+      prompt,
+      session.workingDir,
+      {
+        onPhaseChange: async (state) => {
+          currentPipelinePhase = state.phase;
+          currentPhaseIndex = PHASE_META[state.phase]?.index ?? currentPhaseIndex;
+          if (!progressMsgId) return;
+          const elapsed = Math.floor((Date.now() - pipelineStartTime) / 1000);
+          await feishuClient.updateCard(
+            progressMsgId,
+            buildPipelineCard(
+              prompt,
+              state.phase,
+              currentPhaseIndex,
+              TOTAL_PHASES,
+              elapsed,
+              state.totalCostUsd || undefined,
+            ),
+          );
+        },
+        onStreamUpdate: async (text: string) => {
+          if (!progressMsgId) return;
+          const elapsed = Math.floor((Date.now() - pipelineStartTime) / 1000);
+          // 使用 pipeline 卡片 + detail 区域展示流式输出，保留阶段进度
+          const tail = text.length > 2000 ? '...\n' + text.slice(-2000) : text;
+          await feishuClient.updateCard(
+            progressMsgId,
+            buildPipelineCard(prompt, currentPipelinePhase, currentPhaseIndex, TOTAL_PHASES, elapsed, undefined, tail),
+          );
+        },
+      },
+      historySummaries,
+    );
+
+    // 最终结果卡片
+    const totalElapsed = Math.floor((Date.now() - pipelineStartTime) / 1000);
+    const costStr = pipelineResult.totalCostUsd
+      ? ` | 💰 $${pipelineResult.totalCostUsd.toFixed(4)}`
+      : '';
+
+    // 失败时用 failedAtPhase 定位实际失败的阶段
+    const failedIndex = pipelineResult.state.failedAtPhase
+      ? PHASE_META[pipelineResult.state.failedAtPhase]?.index ?? TOTAL_PHASES
+      : TOTAL_PHASES;
+
+    const finalCard = buildPipelineCard(
+      prompt,
+      pipelineResult.success ? 'done' : 'failed',
+      pipelineResult.success ? TOTAL_PHASES + 1 : failedIndex,
+      TOTAL_PHASES,
+      totalElapsed,
+      pipelineResult.totalCostUsd || undefined,
+      pipelineResult.summary.slice(0, 2500),
+    );
+
+    if (progressMsgId) {
+      await feishuClient.updateCard(progressMsgId, finalCard);
+    } else if (threadRootMsgId) {
+      await feishuClient.replyCardInThread(threadRootMsgId, finalCard);
+    } else {
+      await feishuClient.sendCard(chatId, finalCard);
+    }
+
+    // 如果摘要太长，额外发送完整文本
+    if (pipelineResult.summary.length > 2500) {
+      if (threadRootMsgId) {
+        await feishuClient.replyTextInThread(threadRootMsgId, pipelineResult.summary);
+      } else {
+        await feishuClient.sendText(chatId, pipelineResult.summary);
+      }
+    }
+
+    // 保存摘要
+    if (pipelineResult.summary.length > 100) {
+      try {
+        const date = new Date().toISOString().slice(0, 10);
+        const tail = pipelineResult.summary.slice(-500).trim();
+        const summary = `[${date}] [pipeline] dir: ${session.workingDir} | ${tail}`;
+        sessionManager.saveSummary(chatId, userId, session.workingDir, summary);
+      } catch (err) {
+        logger.warn({ err }, 'Failed to save pipeline summary');
+      }
+    }
+  } catch (err) {
+    logger.error({ err }, 'Error executing pipeline');
+    await feishuClient.replyText(messageId, `❌ 管道执行出错: ${(err as Error).message}`);
+  } finally {
+    try {
+      sessionManager.setStatus(chatId, userId, 'idle');
+    } catch (err) {
+      logger.error({ err, chatId, userId }, 'Failed to reset session status');
+    }
+  }
+}
+
 /**
  * 解析飞书消息 (使用 SDK 类型化的事件数据)
  */
diff --git a/src/feishu/message-builder.ts b/src/feishu/message-builder.ts
index 9448e2dc..3e093737 100644
--- a/src/feishu/message-builder.ts
+++ b/src/feishu/message-builder.ts
@@ -3,6 +3,9 @@
  * 用于构建执行状态卡片、结果卡片等
  */
 
+import { PHASE_META, TOTAL_PHASES } from '../pipeline/types.js';
+import type { PipelinePhase } from '../pipeline/types.js';
+
 /** 构建 "执行中" 状态卡片 */
 export function buildProgressCard(prompt: string, statusText: string = '正在处理...'): Record<string, unknown> {
   return {
@@ -81,6 +84,147 @@ export function buildResultCard(
   };
 }
 
+/** 构建 "执行中" 流式更新卡片（显示实时输出） */
+export function buildStreamingCard(
+  prompt: string,
+  content: string,
+  elapsedSec: number,
+): Record<string, unknown> {
+  // 显示输出末尾 2500 字符，让用户看到最新进展
+  const maxLen = 2500;
+  const displayContent = content.length > maxLen
+    ? '...\n' + content.slice(-maxLen)
+    : content;
+
+  return {
+    config: { wide_screen_mode: true },
+    header: {
+      title: { tag: 'plain_text', content: '🤖 Claude Code - 执行中' },
+      template: 'blue',
+    },
+    elements: [
+      {
+        tag: 'div',
+        text: {
+          tag: 'lark_md',
+          content: `**指令:** ${escapeMarkdown(truncate(prompt, 200))}`,
+        },
+      },
+      { tag: 'hr' },
+      {
+        tag: 'div',
+        text: {
+          tag: 'lark_md',
+          content: displayContent.trim() || '⏳ 正在处理...',
+        },
+      },
+      { tag: 'hr' },
+      {
+        tag: 'note',
+        elements: [
+          {
+            tag: 'plain_text',
+            content: `⏳ 执行中 | ⏱️ ${elapsedSec}s`,
+          },
+        ],
+      },
+    ],
+  };
+}
+
+/** 管道中的可执行阶段（排除 done/failed） */
+const PIPELINE_PHASES: PipelinePhase[] = ['plan', 'plan_review', 'implement', 'code_review', 'push'];
+
+/** 构建管道进度卡片 */
+export function buildPipelineCard(
+  prompt: string,
+  phase: string,
+  phaseIndex: number,
+  totalPhases: number,
+  elapsedSec: number,
+  costUsd?: number,
+  detail?: string,
+): Record<string, unknown> {
+
+  const isDone = phase === 'done';
+  const isFailed = phase === 'failed';
+
+  const phaseLines = PIPELINE_PHASES.map((key) => {
+    const meta = PHASE_META[key];
+    const idx = meta.index;
+    if (isDone) return `✅ ${idx}. ${meta.label}`;
+    if (isFailed && idx >= phaseIndex) {
+      return idx === phaseIndex ? `❌ ${idx}. ${meta.label}` : `⬚ ${idx}. ${meta.label}`;
+    }
+    if (idx < phaseIndex) return `✅ ${idx}. ${meta.label}`;
+    if (idx === phaseIndex) return `🔄 ${idx}. ${meta.label} ← 当前`;
+    return `⬚ ${idx}. ${meta.label}`;
+  });
+
+  const headerTemplate = isDone ? 'green' : isFailed ? 'red' : 'blue';
+  const headerTitle = isDone
+    ? '🤖 Claude Code - 管道完成'
+    : isFailed
+      ? '🤖 Claude Code - 管道失败'
+      : '🤖 Claude Code - 自动开发管道';
+
+  const elements: Record<string, unknown>[] = [
+    {
+      tag: 'div',
+      text: {
+        tag: 'lark_md',
+        content: `**指令:** ${escapeMarkdown(truncate(prompt, 200))}`,
+      },
+    },
+    { tag: 'hr' },
+    {
+      tag: 'div',
+      text: {
+        tag: 'lark_md',
+        content: phaseLines.join('\n'),
+      },
+    },
+  ];
+
+  if (detail) {
+    elements.push({ tag: 'hr' });
+    const maxLen = 2000;
+    const displayDetail = detail.length > maxLen
+      ? '...\n' + detail.slice(-maxLen)
+      : detail;
+    elements.push({
+      tag: 'div',
+      text: {
+        tag: 'lark_md',
+        content: displayDetail.trim(),
+      },
+    });
+  }
+
+  elements.push({ tag: 'hr' });
+
+  const costStr = costUsd ? ` | 💰 $${costUsd.toFixed(4)}` : '';
+  const statusIcon = isDone ? '✅ 完成' : isFailed ? '❌ 失败' : `⏳ 阶段 ${phaseIndex}/${totalPhases}`;
+  elements.push({
+    tag: 'note',
+    elements: [
+      {
+        tag: 'plain_text',
+        content: `${statusIcon} | ⏱️ ${elapsedSec}s${costStr}`,
+      },
+    ],
+  });
+
+  return {
+    config: { wide_screen_mode: true },
+    header: {
+      title: { tag: 'plain_text', content: headerTitle },
+      template: headerTemplate,
+    },
+    elements,
+  };
+}
+
 /** 构建状态查询结果卡片 */
 export function buildStatusCard(
   workingDir: string,
diff --git a/src/pipeline/__tests__/orchestrator.test.ts b/src/pipeline/__tests__/orchestrator.test.ts
new file mode 100644
index 00000000..5f7d4214
--- /dev/null
+++ b/src/pipeline/__tests__/orchestrator.test.ts
@@ -0,0 +1,609 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+import type { ClaudeResult } from '../../claude/types.js';
+
+vi.mock('../../claude/executor.js', () => ({
+  claudeExecutor: {
+    execute: vi.fn(),
+  },
+}));
+
+vi.mock('../../utils/logger.js', () => ({
+  logger: {
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+    debug: vi.fn(),
+  },
+}));
+
+import { PipelineOrchestrator } from '../orchestrator.js';
+import { claudeExecutor } from '../../claude/executor.js';
+
+const mockExecute = vi.mocked(claudeExecutor.execute);
+
+function makeResult(overrides: Partial<ClaudeResult> = {}): ClaudeResult {
+  return {
+    success: true,
+    output: 'mock output',
+    durationMs: 100,
+    costUsd: 0.01,
+    ...overrides,
+  };
+}
+
+const noopCallbacks = { onPhaseChange: vi.fn() };
+
+describe('PipelineOrchestrator', () => {
+  let orchestrator: PipelineOrchestrator;
+
+  beforeEach(() => {
+    orchestrator = new PipelineOrchestrator();
+    vi.clearAllMocks();
+    mockExecute.mockReset();
+  });
+
+  // ============================================================
+  // 完整流程 — 正常路径 (全部 APPROVED)
+  // ============================================================
+
+  describe('happy path', () => {
+    it('should complete full pipeline: plan → review → implement → review → push → done', async () => {
+      // 5 calls: plan, plan_review, implement, code_review, push
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: '## 需求理解\nTest plan' }))      // plan
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED\n没有问题' }))            // plan_review
+        .mockResolvedValueOnce(makeResult({ output: '## 实现摘要\n修改了 foo.ts' }))   // implement
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED\n代码质量良好' }))        // code_review
+        .mockResolvedValueOnce(makeResult({ output: '## 推送结果\nPR: #123' }));       // push
+
+      const result = await orchestrator.run('添加登录功能', '/tmp/work', noopCallbacks);
+
+      expect(result.success).toBe(true);
+      expect(result.state.phase).toBe('done');
+      expect(mockExecute).toHaveBeenCalledTimes(5);
+      expect(result.totalCostUsd).toBeCloseTo(0.05);
+    });
+
+    it('should pass historySummaries only to the plan step', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'implemented' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'pushed' }));
+
+      await orchestrator.run('task', '/tmp', noopCallbacks, '历史摘要内容');
+
+      // plan step (call 0) should include historySummaries
+      const planCall = mockExecute.mock.calls[0][0];
+      expect(planCall.historySummaries).toBe('历史摘要内容');
+
+      // subsequent steps should not have historySummaries
+      for (let i = 1; i < 5; i++) {
+        expect(mockExecute.mock.calls[i][0].historySummaries).toBeUndefined();
+      }
+    });
+
+    it('should pass systemPromptOverride to each step', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'implemented' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'pushed' }));
+
+      await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      for (let i = 0; i < 5; i++) {
+        expect(mockExecute.mock.calls[i][0].systemPromptOverride).toBeDefined();
+        expect(typeof mockExecute.mock.calls[i][0].systemPromptOverride).toBe('string');
+      }
+    });
+
+    it('should call onPhaseChange for every phase transition + final state', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'pushed' }));
+
+      const onPhaseChange = vi.fn();
+      await orchestrator.run('task', '/tmp', { onPhaseChange });
+
+      // 5 phases in loop + 1 final notify = 6
+      expect(onPhaseChange).toHaveBeenCalledTimes(6);
+
+      const phases = onPhaseChange.mock.calls.map((c: unknown[]) => (c[0] as { phase: string }).phase);
+      expect(phases).toEqual(['plan', 'plan_review', 'implement', 'code_review', 'push', 'done']);
+    });
+  });
+
+  // ============================================================
+  // Plan 失败
+  // ============================================================
+
+  describe('plan failure', () => {
+    it('should fail if plan agent returns success: false', async () => {
+      mockExecute.mockResolvedValueOnce(makeResult({ success: false, error: 'Agent error' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.success).toBe(false);
+      expect(result.state.phase).toBe('failed');
+      expect(result.state.failureReason).toContain('方案设计失败');
+      expect(mockExecute).toHaveBeenCalledTimes(1);
+    });
+
+    it('should fail if plan agent returns empty output', async () => {
+      mockExecute.mockResolvedValueOnce(makeResult({ output: '' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.success).toBe(false);
+      expect(result.state.phase).toBe('failed');
+      expect(result.state.failedAtPhase).toBe('plan');
+    });
+  });
+
+  // ============================================================
+  // Plan Review — REJECTED + 重试
+  // ============================================================
+
+  describe('plan review rejection and retry', () => {
+    it('should retry plan once when plan review rejects', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan v1' }))                       // plan
+        .mockResolvedValueOnce(makeResult({ output: 'REJECTED\n方案不完整' }))           // plan_review → REJECTED
+        .mockResolvedValueOnce(makeResult({ output: 'plan v2 (improved)' }))             // plan (retry)
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))                       // plan_review → APPROVED
+        .mockResolvedValueOnce(makeResult({ output: 'implemented' }))                    // implement
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))                       // code_review
+        .mockResolvedValueOnce(makeResult({ output: 'pushed' }));                        // push
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.success).toBe(true);
+      expect(result.state.phase).toBe('done');
+      expect(mockExecute).toHaveBeenCalledTimes(7);
+
+      // Verify retry prompt includes review feedback
+      const retryPlanCall = mockExecute.mock.calls[2][0];
+      expect(retryPlanCall.prompt).toContain('方案不完整');
+    });
+
+    it('should fail after 2 consecutive plan review rejections', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan v1' }))
+        .mockResolvedValueOnce(makeResult({ output: 'REJECTED\n问题1' }))
+        .mockResolvedValueOnce(makeResult({ output: 'plan v2' }))
+        .mockResolvedValueOnce(makeResult({ output: 'REJECTED\n问题2' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.success).toBe(false);
+      expect(result.state.phase).toBe('failed');
+      expect(result.state.failureReason).toContain('方案审查');
+      expect(result.state.failureReason).toContain('2 次未通过');
+      expect(mockExecute).toHaveBeenCalledTimes(4);
+    });
+  });
+
+  // ============================================================
+  // Implement 失败
+  // ============================================================
+
+  describe('implement failure', () => {
+    it('should fail if implement agent returns success: false', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ success: false, error: 'compile error' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.success).toBe(false);
+      expect(result.state.phase).toBe('failed');
+      expect(result.state.failureReason).toContain('代码实现失败');
+      expect(result.state.failedAtPhase).toBe('implement');
+    });
+  });
+
+  // ============================================================
+  // Code Review — REJECTED + 重试
+  // ============================================================
+
+  describe('code review rejection and retry', () => {
+    it('should retry implement once when code review rejects', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl v1' }))                     // implement
+        .mockResolvedValueOnce(makeResult({ output: 'REJECTED\n安全漏洞' }))           // code_review → REJECTED
+        .mockResolvedValueOnce(makeResult({ output: 'impl v2 (fixed)' }))              // implement (retry)
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))                     // code_review → APPROVED
+        .mockResolvedValueOnce(makeResult({ output: 'pushed' }));                      // push
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.success).toBe(true);
+      expect(mockExecute).toHaveBeenCalledTimes(7);
+
+      // Verify retry prompt includes code review feedback
+      const retryImplCall = mockExecute.mock.calls[4][0];
+      expect(retryImplCall.prompt).toContain('安全漏洞');
+    });
+
+    it('should fail after 2 consecutive code review rejections', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl v1' }))
+        .mockResolvedValueOnce(makeResult({ output: 'REJECTED\nbug A' }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl v2' }))
+        .mockResolvedValueOnce(makeResult({ output: 'REJECTED\nbug B' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.success).toBe(false);
+      expect(result.state.phase).toBe('failed');
+      expect(result.state.failureReason).toContain('代码审查');
+    });
+  });
+
+  // ============================================================
+  // Review agent 自身失败 — fail-closed（管道失败）
+  // ============================================================
+
+  describe('review agent failure (fail-closed)', () => {
+    it('should fail pipeline when plan review agent crashes', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ success: false, error: 'agent crash' }));  // plan_review crashes
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.success).toBe(false);
+      expect(result.state.phase).toBe('failed');
+      expect(result.state.failedAtPhase).toBe('plan_review');
+      expect(result.state.failureReason).toContain('审查 agent 执行失败');
+      expect(mockExecute).toHaveBeenCalledTimes(2);
+    });
+
+    it('should fail pipeline when code review agent crashes', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'implemented' }))
+        .mockResolvedValueOnce(makeResult({ success: false, error: 'timeout' }));      // code_review crashes
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.success).toBe(false);
+      expect(result.state.phase).toBe('failed');
+      expect(result.state.failedAtPhase).toBe('code_review');
+      expect(result.state.failureReason).toContain('审查 agent 执行失败');
+      expect(mockExecute).toHaveBeenCalledTimes(4);
+    });
+  });
+
+  // ============================================================
+  // Push 失败 — 不算管道失败
+  // ============================================================
+
+  describe('push failure', () => {
+    it('should mark done (not failed) when push fails — code is already written', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'implemented' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ success: false, error: 'git push failed' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.success).toBe(true);
+      expect(result.state.phase).toBe('done');
+      expect(result.state.pushOutput).toContain('推送失败');
+      expect(result.state.pushOutput).toContain('代码修改已完成');
+    });
+  });
+
+  // ============================================================
+  // Verdict 解析
+  // ============================================================
+
+  describe('verdict parsing', () => {
+    it('should parse APPROVED on first line', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED\nsome suggestions' }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'pushed' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+      expect(result.success).toBe(true);
+    });
+
+    it('should parse REJECTED on first line', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'REJECTED\n- 缺少错误处理' }))
+        .mockResolvedValueOnce(makeResult({ output: 'plan v2' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'pushed' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+      expect(result.success).toBe(true);
+      // plan was retried
+      expect(mockExecute).toHaveBeenCalledTimes(7);
+    });
+
+    it('should default to REJECTED when verdict is unparseable', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        // unparseable output — no APPROVED or REJECTED keyword
+        .mockResolvedValueOnce(makeResult({ output: '我觉得还行吧' }))
+        .mockResolvedValueOnce(makeResult({ output: 'plan v2' }))
+        // second unparseable → max retries
+        .mockResolvedValueOnce(makeResult({ output: '差不多可以了' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.success).toBe(false);
+      expect(result.state.retries['plan_review']).toBe(2);
+    });
+
+    it('should detect APPROVED in body text when first line is not a verdict', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        // APPROVED appears in body but not as sole keyword on first line
+        .mockResolvedValueOnce(makeResult({ output: '审查意见\n\n结论：APPROVED，方案合理' }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'pushed' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+      expect(result.success).toBe(true);
+    });
+  });
+
+  // ============================================================
+  // 成本累计
+  // ============================================================
+
+  describe('cost tracking', () => {
+    it('should accumulate costs across all phases', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan', costUsd: 0.10 }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED', costUsd: 0.05 }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl', costUsd: 0.50 }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED', costUsd: 0.05 }))
+        .mockResolvedValueOnce(makeResult({ output: 'pushed', costUsd: 0.02 }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.totalCostUsd).toBeCloseTo(0.72);
+    });
+
+    it('should handle undefined costUsd gracefully', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan', costUsd: undefined }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED', costUsd: undefined }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl', costUsd: 0.50 }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED', costUsd: undefined }))
+        .mockResolvedValueOnce(makeResult({ output: 'pushed', costUsd: undefined }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.totalCostUsd).toBeCloseTo(0.50);
+    });
+  });
+
+  // ============================================================
+  // 摘要生成
+  // ============================================================
+
+  describe('summary generation', () => {
+    it('should include plan, implement, and push output in success summary', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'my plan details' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'implementation report' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'PR #42 created' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.summary).toContain('管道执行完成');
+      expect(result.summary).toContain('my plan details');
+      expect(result.summary).toContain('implementation report');
+      expect(result.summary).toContain('PR #42 created');
+    });
+
+    it('should include failure reason in failed summary', async () => {
+      mockExecute.mockResolvedValueOnce(makeResult({ success: false, error: 'timeout' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.summary).toContain('管道执行失败');
+      expect(result.summary).toContain('方案设计失败');
+    });
+  });
+
+  // ============================================================
+  // workingDir 传递
+  // ============================================================
+
+  describe('workingDir propagation', () => {
+    it('should pass workingDir to every step', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'pushed' }));
+
+      await orchestrator.run('task', '/my/project', noopCallbacks);
+
+      for (let i = 0; i < 5; i++) {
+        expect(mockExecute.mock.calls[i][0].workingDir).toBe('/my/project');
+      }
+    });
+  });
+
+  // ============================================================
+  // onStreamUpdate 回调
+  // ============================================================
+
+  describe('stream update callback', () => {
+    it('should forward onStreamUpdate to each step', async () => {
+      const onStreamUpdate = vi.fn();
+
+      mockExecute.mockImplementation(async (opts) => {
+        // Simulate calling onStreamUpdate during execution
+        await opts.onStreamUpdate?.('partial output');
+        return makeResult({ output: opts.prompt.includes('审查') ? 'APPROVED' : 'output' });
+      });
+
+      await orchestrator.run('task', '/tmp', {
+        onPhaseChange: vi.fn(),
+        onStreamUpdate,
+      });
+
+      // Should have been called for each step
+      expect(onStreamUpdate).toHaveBeenCalledWith('partial output');
+    });
+  });
+
+  // ============================================================
+  // failedAtPhase 追踪
+  // ============================================================
+
+  describe('failedAtPhase tracking', () => {
+    it('should set failedAtPhase to plan when plan fails', async () => {
+      mockExecute.mockResolvedValueOnce(makeResult({ success: false, error: 'err' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.state.failedAtPhase).toBe('plan');
+    });
+
+    it('should set failedAtPhase to implement when implement fails', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ success: false, error: 'compile error' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.state.failedAtPhase).toBe('implement');
+    });
+
+    it('should set failedAtPhase to plan_review when plan review max retries exceeded', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan v1' }))
+        .mockResolvedValueOnce(makeResult({ output: 'REJECTED\n问题1' }))
+        .mockResolvedValueOnce(makeResult({ output: 'plan v2' }))
+        .mockResolvedValueOnce(makeResult({ output: 'REJECTED\n问题2' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.state.failedAtPhase).toBe('plan_review');
+    });
+
+    it('should set failedAtPhase to code_review when code review max retries exceeded', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl v1' }))
+        .mockResolvedValueOnce(makeResult({ output: 'REJECTED\nbug A' }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl v2' }))
+        .mockResolvedValueOnce(makeResult({ output: 'REJECTED\nbug B' }));
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.state.failedAtPhase).toBe('code_review');
+    });
+  });
+
+  // ============================================================
+  // MAX_ITERATIONS 循环保护
+  // ============================================================
+
+  describe('max iterations protection', () => {
+    it('should fail pipeline when iterations exceed MAX_ITERATIONS', async () => {
+      // 制造无限循环：review 始终 REJECTED，但 retries 永远不增长
+      // 实际上 MAX_RETRIES=2 会先触发，所以我们用一个更直接的方式：
+      // 不断 REJECTED → retry → REJECTED → retry... 直到 MAX_ITERATIONS
+      // 每轮消耗 2 calls (plan + review)，MAX_RETRIES=2 在第 4 calls 后触发
+      // 所以正常情况下 MAX_ITERATIONS 不会先触发
+      // 为了测试 MAX_ITERATIONS，模拟一个极端情况：大量 mock 返回值
+      const mocks: ReturnType<typeof makeResult>[] = [];
+      for (let i = 0; i < 25; i++) {
+        mocks.push(makeResult({ output: `plan v${i}` }));
+        mocks.push(makeResult({ output: 'REJECTED\nredo' }));
+      }
+      for (const m of mocks) {
+        mockExecute.mockResolvedValueOnce(m);
+      }
+
+      const result = await orchestrator.run('task', '/tmp', noopCallbacks);
+
+      expect(result.success).toBe(false);
+      expect(result.state.phase).toBe('failed');
+      // 应该在 MAX_RETRIES 或 MAX_ITERATIONS 处终止
+      expect(mockExecute.mock.calls.length).toBeLessThanOrEqual(20);
+    });
+  });
+
+  // ============================================================
+  // onPhaseChange 回调异常不中断管道
+  // ============================================================
+
+  describe('callback error resilience', () => {
+    it('should continue pipeline when onPhaseChange throws', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'pushed' }));
+
+      const failingCallback = vi.fn().mockRejectedValue(new Error('callback exploded'));
+
+      const result = await orchestrator.run('task', '/tmp', {
+        onPhaseChange: failingCallback,
+      });
+
+      expect(result.success).toBe(true);
+      expect(result.state.phase).toBe('done');
+      // 回调仍然被调用了（只是异常被吞掉）
+      expect(failingCallback).toHaveBeenCalled();
+    });
+
+    it('should complete pipeline when onPhaseChange throws on final notification', async () => {
+      mockExecute
+        .mockResolvedValueOnce(makeResult({ output: 'plan' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'impl' }))
+        .mockResolvedValueOnce(makeResult({ output: 'APPROVED' }))
+        .mockResolvedValueOnce(makeResult({ output: 'pushed' }));
+
+      // 只在最后一次调用时抛异常（final notification）
+      let callCount = 0;
+      const onPhaseChange = vi.fn().mockImplementation(async () => {
+        callCount++;
+        if (callCount === 6) throw new Error('final callback error');
+      });
+
+      const result = await orchestrator.run('task', '/tmp', { onPhaseChange });
+
+      expect(result.success).toBe(true);
+      expect(result.state.phase).toBe('done');
+    });
+  });
+});
diff --git a/src/pipeline/orchestrator.ts b/src/pipeline/orchestrator.ts
new file mode 100644
index 00000000..19fc2af5
--- /dev/null
+++ b/src/pipeline/orchestrator.ts
@@ -0,0 +1,423 @@
+import { claudeExecutor } from '../claude/executor.js';
+import { logger } from '../utils/logger.js';
+import type { ClaudeResult } from '../claude/types.js';
+import type {
+  PipelinePhase,
+  PipelineState,
+  PipelineCallbacks,
+  PipelineResult,
+} from './types.js';
+import {
+  PLAN_SYSTEM_PROMPT,
+  PLAN_REVIEW_SYSTEM_PROMPT,
+  IMPLEMENT_SYSTEM_PROMPT,
+  CODE_REVIEW_SYSTEM_PROMPT,
+  PUSH_SYSTEM_PROMPT,
+} from './prompts.js';
+
+// ============================================================
+// Pipeline Orchestrator — 状态机驱动的多步开发管道
+//
+// Phase A: 每步使用单独的 Claude Agent SDK query()
+// Review 步骤暂用单 agent 自审，Phase B 将替换为并行多 agent
+// ============================================================
+
+const MAX_RETRIES = 2;
+
+export class PipelineOrchestrator {
+  /**
+   * 执行完整管道
+   */
+  async run(
+    prompt: string,
+    workingDir: string,
+    callbacks: PipelineCallbacks,
+    historySummaries?: string,
+  ): Promise<PipelineResult> {
+    const startTime = Date.now();
+
+    let state: PipelineState = {
+      phase: 'plan',
+      userPrompt: prompt,
+      workingDir,
+      retries: {},
+      phaseDurations: {},
+      totalCostUsd: 0,
+    };
+
+    logger.info({ prompt: prompt.slice(0, 100), workingDir }, 'Pipeline started');
+
+    // 最大迭代保护：5 phases * 2 retries * 2 safety margin
+    const MAX_ITERATIONS = 20;
+    let iterations = 0;
+
+    while (state.phase !== 'done' && state.phase !== 'failed') {
+      if (++iterations > MAX_ITERATIONS) {
+        state = { ...state, phase: 'failed', failedAtPhase: state.phase, failureReason: '管道超过最大迭代次数，可能存在循环' };
+        break;
+      }
+
+      const currentPhase = state.phase;
+      const phaseStart = Date.now();
+
+      try { await callbacks.onPhaseChange?.(state); } catch (err) {
+        logger.warn({ err, phase: currentPhase }, 'onPhaseChange callback failed');
+      }
+
+      logger.info({ phase: currentPhase, retries: state.retries }, 'Pipeline phase starting');
+
+      switch (currentPhase) {
+        case 'plan':
+          state = await this.doPlan(state, callbacks, historySummaries);
+          break;
+        case 'plan_review':
+          state = await this.doReview(state, 'plan_review', callbacks);
+          break;
+        case 'implement':
+          state = await this.doImplement(state, callbacks);
+          break;
+        case 'code_review':
+          state = await this.doReview(state, 'code_review', callbacks);
+          break;
+        case 'push':
+          state = await this.doPush(state, callbacks);
+          break;
+      }
+
+      state.phaseDurations[currentPhase] = Date.now() - phaseStart;
+    }
+
+    // 最终通知
+    try { await callbacks.onPhaseChange?.(state); } catch (err) {
+      logger.warn({ err, phase: state.phase }, 'Final onPhaseChange callback failed');
+    }
+
+    const durationMs = Date.now() - startTime;
+    const summary = this.buildSummary(state);
+
+    logger.info(
+      { phase: state.phase, durationMs, costUsd: state.totalCostUsd },
+      'Pipeline finished',
+    );
+
+    return {
+      success: state.phase === 'done',
+      state,
+      summary,
+      durationMs,
+      totalCostUsd: state.totalCostUsd,
+    };
+  }
+
+  // ============================================================
+  // Phase: Plan
+  // ============================================================
+
+  private async doPlan(
+    state: PipelineState,
+    callbacks: PipelineCallbacks,
+    historySummaries?: string,
+  ): Promise<PipelineState> {
+    // 如果是重试，带上上次 review 的反馈
+    let prompt = state.userPrompt;
+    if (state.planReviewFeedback) {
+      prompt = `${state.userPrompt}\n\n---\n上一版方案被审查拒绝，请根据以下反馈修改方案：\n${state.planReviewFeedback}`;
+    }
+
+    const result = await this.executeStep(
+      `pipeline-plan-${Date.now()}`,
+      prompt,
+      state.workingDir,
+      PLAN_SYSTEM_PROMPT,
+      historySummaries,
+      callbacks.onStreamUpdate,
+    );
+
+    const totalCostUsd = state.totalCostUsd + (result.costUsd ?? 0);
+
+    if (!result.success || !result.output) {
+      return {
+        ...state,
+        totalCostUsd,
+        phase: 'failed',
+        failedAtPhase: 'plan',
+        failureReason: `方案设计失败: ${result.error || '无输出'}`,
+      };
+    }
+
+    return {
+      ...state,
+      totalCostUsd,
+      phase: 'plan_review',
+      plan: result.output,
+    };
+  }
+
+  // ============================================================
+  // Phase: Review (plan_review 和 code_review 共用)
+  // ============================================================
+
+  private async doReview(
+    state: PipelineState,
+    reviewPhase: 'plan_review' | 'code_review',
+    callbacks: PipelineCallbacks,
+  ): Promise<PipelineState> {
+    const isPlanReview = reviewPhase === 'plan_review';
+    const systemPrompt = isPlanReview
+      ? PLAN_REVIEW_SYSTEM_PROMPT
+      : CODE_REVIEW_SYSTEM_PROMPT;
+
+    // 构建 review prompt
+    let prompt: string;
+    if (isPlanReview) {
+      prompt = `请审查以下实施方案：\n\n${state.plan}`;
+    } else {
+      // 代码审查：让 reviewer 自己 git diff
+      prompt = [
+        `请审查当前工作目录中的代码变更。`,
+        ``,
+        `原始需求: ${state.userPrompt}`,
+        ``,
+        `实施方案: ${state.plan?.slice(0, 1500) || '(无方案)'}`,
+        ``,
+        `实现报告: ${state.implementOutput?.slice(0, 1500) || '(无报告)'}`,
+        ``,
+        `请运行 git diff 查看实际变更，然后给出审查意见。`,
+      ].join('\n');
+    }
+
+    const result = await this.executeStep(
+      `pipeline-${reviewPhase}-${Date.now()}`,
+      prompt,
+      state.workingDir,
+      systemPrompt,
+      undefined,
+      callbacks.onStreamUpdate,
+    );
+
+    const totalCostUsd = state.totalCostUsd + (result.costUsd ?? 0);
+
+    if (!result.success) {
+      // review agent 自身失败（超时/崩溃等）— fail-closed，不跳过审查
+      logger.warn({ reviewPhase, error: result.error }, 'Review agent failed, treating as pipeline failure');
+      return {
+        ...state,
+        totalCostUsd,
+        phase: 'failed',
+        failedAtPhase: reviewPhase,
+        failureReason: `${isPlanReview ? '方案' : '代码'}审查 agent 执行失败: ${result.error || '未知错误'}`,
+      };
+    }
+
+    const verdict = this.parseVerdict(result.output);
+
+    if (verdict.approved) {
+      logger.info({ reviewPhase }, 'Review approved');
+      return {
+        ...state,
+        totalCostUsd,
+        phase: isPlanReview ? 'implement' : 'push',
+        ...(isPlanReview
+          ? { planReviewFeedback: undefined }
+          : { codeReviewFeedback: undefined }),
+      };
+    }
+
+    // REJECTED — 检查重试次数
+    const retryKey = reviewPhase;
+    const retryCount = (state.retries[retryKey] ?? 0) + 1;
+
+    if (retryCount >= MAX_RETRIES) {
+      logger.warn({ reviewPhase, retryCount }, 'Review rejected, max retries reached');
+      return {
+        ...state,
+        totalCostUsd,
+        phase: 'failed',
+        failedAtPhase: reviewPhase,
+        retries: { ...state.retries, [retryKey]: retryCount },
+        failureReason: `${isPlanReview ? '方案' : '代码'}审查连续 ${retryCount} 次未通过:\n${verdict.feedback}`,
+      };
+    }
+
+    logger.info({ reviewPhase, retryCount, feedback: verdict.feedback.slice(0, 200) }, 'Review rejected, retrying');
+
+    // 回退到上一步重做
+    return {
+      ...state,
+      totalCostUsd,
+      phase: isPlanReview ? 'plan' : 'implement',
+      retries: { ...state.retries, [retryKey]: retryCount },
+      ...(isPlanReview
+        ? { planReviewFeedback: verdict.feedback }
+        : { codeReviewFeedback: verdict.feedback }),
+    };
+  }
+
+  // ============================================================
+  // Phase: Implement
+  // ============================================================
+
+  private async doImplement(
+    state: PipelineState,
+    callbacks: PipelineCallbacks,
+  ): Promise<PipelineState> {
+    let prompt = `请按照以下已审批方案实施代码修改：\n\n${state.plan}`;
+
+    // 如果是重试，带上 code review 反馈
+    if (state.codeReviewFeedback) {
+      prompt += `\n\n---\n上一版实现被代码审查拒绝，请根据以下反馈修改：\n${state.codeReviewFeedback}`;
+    }
+
+    const result = await this.executeStep(
+      `pipeline-implement-${Date.now()}`,
+      prompt,
+      state.workingDir,
+      IMPLEMENT_SYSTEM_PROMPT,
+      undefined,
+      callbacks.onStreamUpdate,
+    );
+
+    const totalCostUsd = state.totalCostUsd + (result.costUsd ?? 0);
+
+    if (!result.success) {
+      return {
+        ...state,
+        totalCostUsd,
+        phase: 'failed',
+        failedAtPhase: 'implement',
+        failureReason: `代码实现失败: ${result.error || '无输出'}`,
+      };
+    }
+
+    return {
+      ...state,
+      totalCostUsd,
+      phase: 'code_review',
+      implementOutput: result.output,
+    };
+  }
+
+  // ============================================================
+  // Phase: Push
+  // ============================================================
+
+  private async doPush(
+    state: PipelineState,
+    callbacks: PipelineCallbacks,
+  ): Promise<PipelineState> {
+    const prompt = [
+      `请将当前工作目录中的代码变更提交并推送。`,
+      ``,
+      `原始需求: ${state.userPrompt}`,
+      ``,
+      `实现摘要: ${state.implementOutput?.slice(0, 1000) || '(无)'}`,
+    ].join('\n');
+
+    const result = await this.executeStep(
+      `pipeline-push-${Date.now()}`,
+      prompt,
+      state.workingDir,
+      PUSH_SYSTEM_PROMPT,
+      undefined,
+      callbacks.onStreamUpdate,
+    );
+
+    const totalCostUsd = state.totalCostUsd + (result.costUsd ?? 0);
+
+    if (!result.success) {
+      // push 失败不算完全失败，代码已经写好了
+      return {
+        ...state,
+        totalCostUsd,
+        phase: 'done',
+        pushOutput: `推送失败，但代码修改已完成: ${result.error || result.output}`,
+      };
+    }
+
+    return {
+      ...state,
+      totalCostUsd,
+      phase: 'done',
+      pushOutput: result.output,
+    };
+  }
+
+  // ============================================================
+  // 工具方法
+  // ============================================================
+
+  /**
+   * 执行单步 Claude query（底层调用 claudeExecutor）
+   */
+  private async executeStep(
+    sessionKey: string,
+    prompt: string,
+    workingDir: string,
+    systemPrompt: string,
+    historySummaries?: string,
+    onStreamUpdate?: (text: string) => Promise<void>,
+  ): Promise<ClaudeResult> {
+    return claudeExecutor.execute({
+      sessionKey,
+      prompt,
+      workingDir,
+      // pipeline 每步独立，不 resume 上一步的 session
+      resumeSessionId: undefined,
+      onStreamUpdate,
+      historySummaries,
+      systemPromptOverride: systemPrompt,
+    });
+  }
+
+  /**
+   * 解析 review 输出中的 APPROVED/REJECTED
+   */
+  private parseVerdict(output: string): { approved: boolean; feedback: string } {
+    const lines = output.trim().split('\n');
+    const firstLine = lines[0]?.trim().toUpperCase() ?? '';
+
+    if (firstLine === 'APPROVED' || firstLine.startsWith('APPROVED')) {
+      return { approved: true, feedback: lines.slice(1).join('\n').trim() };
+    }
+
+    if (firstLine === 'REJECTED' || firstLine.startsWith('REJECTED')) {
+      return { approved: false, feedback: lines.slice(1).join('\n').trim() };
+    }
+
+    // 无法解析 → 搜索全文（使用词边界避免匹配 "NOT APPROVED" 等子串）
+    if (/\bAPPROVED\b/i.test(output) && !/\bREJECTED\b/i.test(output)) {
+      return { approved: true, feedback: output };
+    }
+
+    // 默认 REJECTED（宁可多审一轮）
+    logger.warn({ firstLine: lines[0] }, 'Could not parse review verdict, defaulting to REJECTED');
+    return { approved: false, feedback: output };
+  }
+
+  /**
+   * 构建最终摘要文本
+   */
+  private buildSummary(state: PipelineState): string {
+    if (state.phase === 'failed') {
+      const parts: string[] = ['## 管道执行失败'];
+      if (state.failureReason) parts.push(`\n**原因:** ${state.failureReason}`);
+      if (state.plan) parts.push(`\n**方案:**\n${state.plan.slice(0, 1000)}`);
+      if (state.implementOutput) parts.push(`\n**已完成的实现:**\n${state.implementOutput.slice(0, 1000)}`);
+      return parts.join('\n');
+    }
+
+    const parts: string[] = ['## 管道执行完成'];
+
+    if (state.plan) {
+      parts.push(`\n**方案:**\n${state.plan.slice(0, 800)}`);
+    }
+    if (state.implementOutput) {
+      parts.push(`\n**实现:**\n${state.implementOutput.slice(0, 800)}`);
+    }
+    if (state.pushOutput) {
+      parts.push(`\n**推送:**\n${state.pushOutput.slice(0, 500)}`);
+    }
+
+    return parts.join('\n');
+  }
+}
diff --git a/src/pipeline/prompts.ts b/src/pipeline/prompts.ts
new file mode 100644
index 00000000..a13dad21
--- /dev/null
+++ b/src/pipeline/prompts.ts
@@ -0,0 +1,123 @@
+// ============================================================
+// Pipeline 各角色 System Prompt
+// ============================================================
+
+/** Plan Agent: 生成实施方案 */
+export const PLAN_SYSTEM_PROMPT = `你是一个技术方案设计师。根据用户需求，分析现有代码结构，输出结构化的实施方案。
+
+严格按以下格式输出：
+
+## 需求理解
+(一句话总结用户要做什么)
+
+## 影响范围
+(列出需要修改/新增的文件，每个文件说明修改原因)
+
+## 实施步骤
+(编号列表，每步具体到函数/模块级别，包含代码修改的具体描述)
+
+## 测试计划
+(如何验证修改正确性——运行哪些测试、如何手动验证)
+
+## 风险点
+(可能出问题的地方，以及应对策略)
+
+规则：
+- 方案要具体可执行，不要泛泛而谈
+- 如果需要查看文件内容来制定方案，先读取相关文件
+- 不要在方案阶段修改任何文件
+- 如果需求不明确，在方案中列出假设`;
+
+/** Plan Review Agent (Phase A: 自审模式) */
+export const PLAN_REVIEW_SYSTEM_PROMPT = `你是一个技术方案审查员。你需要审查一份实施方案，判断它是否可以安全地执行。
+
+审查维度：
+1. **完整性**: 方案是否覆盖了需求的所有方面？是否有遗漏的边界情况？
+2. **正确性**: 修改方案是否会引入 bug？是否考虑了向后兼容？
+3. **安全性**: 是否有注入漏洞、权限问题、敏感信息泄露的风险？
+4. **可行性**: 步骤是否具体到可直接执行？是否有模糊或矛盾之处？
+
+输出格式（严格遵守）：
+第一行必须是 APPROVED 或 REJECTED（大写，单独一行，无其他内容）
+
+如果 REJECTED，后续列出具体问题：
+- [完整性] 问题描述
+- [正确性] 问题描述
+...
+
+如果 APPROVED，后续可选给出改进建议（不阻塞执行）。
+
+规则：
+- 只有存在可能导致生产事故、数据丢失或安全漏洞的问题时才 REJECTED
+- 代码风格、命名偏好等非关键问题给建议但不 REJECTED
+- 方案不够详细也应该 REJECTED——模糊的方案执行时容易出错`;
+
+/** Implement Agent: 按方案执行代码修改 */
+export const IMPLEMENT_SYSTEM_PROMPT = `你是一个高级开发工程师。你需要严格按照已审批的技术方案，执行代码修改。
+
+规则：
+- 严格按方案中的"实施步骤"逐步执行，不要擅自扩展范围
+- 写完代码后，按方案中的"测试计划"运行测试
+- 如果测试失败：分析错误 → 修复 → 重新测试（最多重试 2 轮）
+- 如果相同测试以相同方式连续失败 2 次，停止重试
+- 不要 git add / git commit / git push，推送由后续步骤处理
+- 不要提交 .env、credentials 等敏感文件
+
+完成后输出：
+## 实现摘要
+(列出实际修改了哪些文件、每个文件改了什么)
+
+## 测试结果
+(测试命令和结果，如果没有测试命令则说明)
+
+## 偏差说明
+(如果实际实现与方案有任何偏差，必须在这里说明原因)`;
+
+/** Code Review Agent (Phase A: 自审模式) */
+export const CODE_REVIEW_SYSTEM_PROMPT = `你是一个代码审查员。你需要审查刚刚完成的代码修改，判断是否可以安全推送。
+
+审查方法：
+1. 运行 \`git diff\` 查看所有未提交的变更
+2. 逐文件审查变更内容
+
+审查维度：
+1. **正确性**: 代码逻辑是否正确？边界条件是否处理？
+2. **安全性**: 是否有注入漏洞、硬编码密钥、权限问题？
+3. **一致性**: 代码风格是否与项目现有代码一致？
+4. **测试**: 测试是否通过？变更是否有测试覆盖？
+
+输出格式（严格遵守）：
+第一行必须是 APPROVED 或 REJECTED（大写，单独一行，无其他内容）
+
+如果 REJECTED，后续列出具体问题：
+- [严重程度: high/medium/low] 文件:行号 — 问题描述
+
+如果 APPROVED，后续可选给出改进建议。
+
+规则：
+- 只有 high 严重程度的问题才导致 REJECTED
+- medium/low 问题给建议但不 REJECTED
+- 如果 git diff 为空，输出 APPROVED 并说明没有变更`;
+
+/** Push Agent: 提交和推送 */
+export const PUSH_SYSTEM_PROMPT = `你需要将当前工作目录中的代码变更提交并推送到远程仓库，然后创建 Pull Request。
+
+步骤：
+1. \`git branch\` 确认当前分支（不应在 main/master/develop 上）
+2. \`git diff --stat\` 查看变更文件列表
+3. \`git log --oneline -5\` 学习项目 commit message 风格
+4. \`git add\` 逐个添加变更文件（不要 git add . 或 git add -A）
+5. \`git commit\` 用符合项目风格的 commit message
+6. 推送前预检：
+   - \`git remote -v\` 确认 origin 存在
+   - \`gh auth status\` 确认 GitHub CLI 已认证
+   - 如果任一检查失败，停止推送，报告需要手动处理的部分
+7. \`git push -u origin <当前分支>\`
+8. \`gh pr create --fill\` 或用更详细的标题/描述创建 PR
+
+完成后输出：
+## 推送结果
+- Commit: <commit hash 和 message>
+- Branch: <分支名>
+- PR: <PR 链接> (如果创建成功)
+- 如果推送/PR 失败，说明原因和需要手动处理的步骤`;
diff --git a/src/pipeline/types.ts b/src/pipeline/types.ts
new file mode 100644
index 00000000..7003861d
--- /dev/null
+++ b/src/pipeline/types.ts
@@ -0,0 +1,73 @@
+// ============================================================
+// Pipeline 类型定义
+// ============================================================
+
+/** 管道阶段 */
+export type PipelinePhase =
+  | 'plan'
+  | 'plan_review'
+  | 'implement'
+  | 'code_review'
+  | 'push'
+  | 'done'
+  | 'failed';
+
+/** 阶段元信息（用于卡片展示） */
+export const PHASE_META: Record<PipelinePhase, { label: string; index: number }> = {
+  plan:        { label: '方案设计', index: 1 },
+  plan_review: { label: '方案审查', index: 2 },
+  implement:   { label: '代码实现', index: 3 },
+  code_review: { label: '代码审查', index: 4 },
+  push:        { label: '推送 & PR', index: 5 },
+  done:        { label: '完成', index: 6 },
+  failed:      { label: '失败', index: 6 },
+};
+
+export const TOTAL_PHASES = 5;
+
+/** 管道状态（可序列化，用于恢复） */
+export interface PipelineState {
+  phase: PipelinePhase;
+  userPrompt: string;
+  workingDir: string;
+  /** Step 1 输出：实施方案文本 */
+  plan?: string;
+  /** Step 2 输出：审查结果 */
+  planReviewFeedback?: string;
+  /** Step 3 输出：实现摘要 */
+  implementOutput?: string;
+  /** Step 4 输出：代码审查结果 */
+  codeReviewFeedback?: string;
+  /** Step 5 输出：推送/PR 结果 */
+  pushOutput?: string;
+  /** 各阶段重试计数 */
+  retries: Record<string, number>;
+  /** 各阶段耗时 (ms) */
+  phaseDurations: Record<string, number>;
+  /** 总花费 (USD) */
+  totalCostUsd: number;
+  /** 失败原因 */
+  failureReason?: string;
+  /** 失败发生在哪个阶段（用于卡片展示失败位置） */
+  failedAtPhase?: PipelinePhase;
+}
+
+/** 管道回调 */
+export interface PipelineCallbacks {
+  /** 阶段变更通知（用于更新飞书卡片） */
+  onPhaseChange?: (state: PipelineState) => Promise<void>;
+  /** 流式输出更新 */
+  onStreamUpdate?: (text: string) => Promise<void>;
+}
+
+/** 管道最终结果 */
+export interface PipelineResult {
+  success: boolean;
+  state: PipelineState;
+  /** 汇总的结果文本（用于飞书展示） */
+  summary: string;
+  /** 总耗时 (ms) */
+  durationMs: number;
+  /** 总花费 (USD) */
+  totalCostUsd: number;
+}
diff --git a/src/session/__tests__/database.test.ts b/src/session/__tests__/database.test.ts
new file mode 100644
index 00000000..c98c434c
--- /dev/null
+++ b/src/session/__tests__/database.test.ts
@@ -0,0 +1,153 @@
+import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
+import { mkdtempSync, rmSync } from 'node:fs';
+import { join } from 'node:path';
+import { tmpdir } from 'node:os';
+
+vi.mock('../../utils/logger.js', () => ({
+  logger: {
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+    debug: vi.fn(),
+  },
+}));
+
+import { SessionDatabase } from '../database.js';
+
+describe('SessionDatabase — session_summaries', () => {
+  let db: SessionDatabase;
+  let tempDir: string;
+
+  beforeEach(() => {
+    tempDir = mkdtempSync(join(tmpdir(), 'session-db-test-'));
+    db = new SessionDatabase(join(tempDir, 'test.db'));
+  });
+
+  afterEach(() => {
+    db.close();
+    rmSync(tempDir, { recursive: true, force: true });
+  });
+
+  describe('insertSummary', () => {
+    it('should insert a summary without error', () => {
+      expect(() => {
+        db.insertSummary('chat1', 'user1', '/tmp', 'summary text');
+      }).not.toThrow();
+    });
+
+    it('should insert multiple summaries for the same user', () => {
+      db.insertSummary('chat1', 'user1', '/tmp', 'summary 1');
+      db.insertSummary('chat1', 'user1', '/tmp', 'summary 2');
+      db.insertSummary('chat1', 'user1', '/tmp', 'summary 3');
+
+      const summaries = db.getRecentSummaries('chat1', 'user1', 10);
+      expect(summaries).toHaveLength(3);
+    });
+  });
+
+  describe('getRecentSummaries', () => {
+    it('should return empty array when no summaries exist', () => {
+      const summaries = db.getRecentSummaries('chat1', 'user1', 5);
+      expect(summaries).toEqual([]);
+    });
+
+    it('should return summaries in chronological order (oldest first)', () => {
+      db.insertSummary('chat1', 'user1', '/tmp', 'first');
+      db.insertSummary('chat1', 'user1', '/tmp', 'second');
+      db.insertSummary('chat1', 'user1', '/tmp', 'third');
+
+      const summaries = db.getRecentSummaries('chat1', 'user1', 5);
+      expect(summaries).toEqual(['first', 'second', 'third']);
+    });
+
+    it('should respect the limit parameter', () => {
+      for (let i = 1; i <= 10; i++) {
+        db.insertSummary('chat1', 'user1', '/tmp', `summary ${i}`);
+      }
+
+      const summaries = db.getRecentSummaries('chat1', 'user1', 3);
+      expect(summaries).toHaveLength(3);
+      // Should return the 3 most recent, in chronological order
+      expect(summaries).toEqual(['summary 8', 'summary 9', 'summary 10']);
+    });
+
+    it('should isolate summaries by chatId', () => {
+      db.insertSummary('chat1', 'user1', '/tmp', 'chat1 summary');
+      db.insertSummary('chat2', 'user1', '/tmp', 'chat2 summary');
+
+      expect(db.getRecentSummaries('chat1', 'user1', 10)).toEqual(['chat1 summary']);
+      expect(db.getRecentSummaries('chat2', 'user1', 10)).toEqual(['chat2 summary']);
+    });
+
+    it('should isolate summaries by userId', () => {
+      db.insertSummary('chat1', 'user1', '/tmp', 'user1 summary');
+      db.insertSummary('chat1', 'user2', '/tmp', 'user2 summary');
+
+      expect(db.getRecentSummaries('chat1', 'user1', 10)).toEqual(['user1 summary']);
+      expect(db.getRecentSummaries('chat1', 'user2', 10)).toEqual(['user2 summary']);
+    });
+  });
+
+  describe('cleanOldSummaries', () => {
+    it('should return 0 when no summaries to clean', () => {
+      const cleaned = db.cleanOldSummaries(30);
+      expect(cleaned).toBe(0);
+    });
+
+    it('should not clean recent summaries', () => {
+      db.insertSummary('chat1', 'user1', '/tmp', 'recent summary');
+
+      const cleaned = db.cleanOldSummaries(30);
+      expect(cleaned).toBe(0);
+      expect(db.getRecentSummaries('chat1', 'user1', 10)).toHaveLength(1);
+    });
+
+    it('should clean summaries older than maxAgeDays', () => {
+      // Insert a summary, then manually back-date it via raw SQL
+      db.insertSummary('chat1', 'user1', '/tmp', 'old summary');
+
+      // Back-date by directly accessing the db (hack for testing)
+      const oldDate = new Date(Date.now() - 31 * 24 * 60 * 60 * 1000).toISOString();
+      (db as any).db.prepare(
+        "UPDATE session_summaries SET created_at = ? WHERE summary = 'old summary'"
+      ).run(oldDate);
+
+      // Insert a recent one
+      db.insertSummary('chat1', 'user1', '/tmp', 'new summary');
+
+      const cleaned = db.cleanOldSummaries(30);
+      expect(cleaned).toBe(1);
+
+      const remaining = db.getRecentSummaries('chat1', 'user1', 10);
+      expect(remaining).toEqual(['new summary']);
+    });
+  });
+
+  describe('summaries survive session cleanup', () => {
+    it('should retain summaries after deleteExpired removes sessions', () => {
+      // Create a session
+      db.upsert('chat1:user1', {
+        chatId: 'chat1',
+        userId: 'user1',
+        workingDir: '/tmp',
+        status: 'idle',
+        createdAt: new Date(),
+        lastActiveAt: new Date(Date.now() - 48 * 60 * 60 * 1000), // 48h ago
+      });
+
+      // Insert summaries for the same user
+      db.insertSummary('chat1', 'user1', '/tmp', 'should survive');
+
+      // Clean expired sessions (24h idle)
+      const cleaned = db.deleteExpired(24 * 60 * 60 * 1000);
+      expect(cleaned).toBe(1);
+
+      // Session is gone
+      expect(db.get('chat1:user1')).toBeUndefined();
+
+      // Summaries survive
+      const summaries = db.getRecentSummaries('chat1', 'user1', 10);
+      expect(summaries).toEqual(['should survive']);
+    });
+  });
+});
diff --git a/src/session/database.ts b/src/session/database.ts
index cc8844bd..4d3c05e2 100644
--- a/src/session/database.ts
+++ b/src/session/database.ts
@@ -34,6 +34,10 @@ export class SessionDatabase {
   private stmtUpdateThread: Database.Statement;
   private stmtUpdateLastActive: Database.Statement;
   private stmtResetBusy: Database.Statement;
+  private stmtTryAcquire: Database.Statement;
+  private stmtInsertSummary: Database.Statement;
+  private stmtGetRecentSummaries: Database.Statement;
+  private stmtCleanOldSummaries: Database.Statement;
 
   constructor(dbPath: string) {
     dbPath = resolve(dbPath);
@@ -107,6 +111,38 @@ export class SessionDatabase {
       "UPDATE sessions SET status = 'idle', conversation_id = NULL WHERE status = 'busy'",
     );
 
+    this.stmtTryAcquire = this.db.prepare(
+      "UPDATE sessions SET status = 'busy', last_active_at = ? WHERE key = ? AND status != 'busy'",
+    );
+
+    // 会话摘要表（独立于 sessions，不受 cleanup 影响）
+    this.db.exec(`
+      CREATE TABLE IF NOT EXISTS session_summaries (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        chat_id TEXT NOT NULL,
+        user_id TEXT NOT NULL,
+        working_dir TEXT,
+        summary TEXT NOT NULL,
+        created_at TEXT NOT NULL
+      )
+    `);
+    this.db.exec(`
+      CREATE INDEX IF NOT EXISTS idx_summaries_user
+        ON session_summaries(chat_id, user_id, created_at DESC)
+    `);
+
+    this.stmtInsertSummary = this.db.prepare(
+      'INSERT INTO session_summaries (chat_id, user_id, working_dir, summary, created_at) VALUES (?, ?, ?, ?, ?)',
+    );
+
+    this.stmtGetRecentSummaries = this.db.prepare(
+      'SELECT summary FROM session_summaries WHERE chat_id = ? AND user_id = ? ORDER BY created_at DESC, id DESC LIMIT ?',
+    );
+
+    this.stmtCleanOldSummaries = this.db.prepare(
+      'DELETE FROM session_summaries WHERE created_at < ?',
+    );
+
     logger.info({ dbPath }, 'Session database initialized');
   }
 
@@ -161,6 +197,15 @@ export class SessionDatabase {
     this.stmtUpdateLastActive.run(new Date().toISOString(), key);
   }
 
+  /**
+   * 原子地尝试将 session 标记为 busy（CAS: idle → busy）
+   * @returns true 如果成功获取锁，false 如果已经 busy
+   */
+  tryAcquire(key: string): boolean {
+    const result = this.stmtTryAcquire.run(new Date().toISOString(), key);
+    return result.changes === 1;
+  }
+
   resetBusySessions(): number {
     const result = this.stmtResetBusy.run();
     if (result.changes > 0) {
@@ -169,6 +214,22 @@ export class SessionDatabase {
     return result.changes;
   }
 
+  insertSummary(chatId: string, userId: string, workingDir: string, summary: string): void {
+    this.stmtInsertSummary.run(chatId, userId, workingDir, summary, new Date().toISOString());
+  }
+
+  getRecentSummaries(chatId: string, userId: string, limit: number): string[] {
+    const rows = this.stmtGetRecentSummaries.all(chatId, userId, limit) as Array<{ summary: string }>;
+    // 返回时间正序（旧 → 新）
+    return rows.map((r) => r.summary).reverse();
+  }
+
+  cleanOldSummaries(maxAgeDays: number = 30): number {
+    const cutoff = new Date(Date.now() - maxAgeDays * 24 * 60 * 60 * 1000).toISOString();
+    const result = this.stmtCleanOldSummaries.run(cutoff);
+    return result.changes;
+  }
+
   close(): void {
     this.db.close();
     logger.info('Session database closed');
diff --git a/src/session/manager.ts b/src/session/manager.ts
index 7b1293f2..0746a7a6 100644
--- a/src/session/manager.ts
+++ b/src/session/manager.ts
@@ -68,6 +68,16 @@ export class SessionManager {
     this.db.updateStatus(this.makeKey(chatId, userId), status);
   }
 
+  /**
+   * 原子地尝试获取会话锁（idle → busy），防止 TOCTOU 竞态
+   * @returns true 如果成功获取（session 之前不是 busy），false 如果已被占用
+   */
+  tryAcquire(chatId: string, userId: string): boolean {
+    // 确保 session 存在
+    this.getOrCreate(chatId, userId);
+    return this.db.tryAcquire(this.makeKey(chatId, userId));
+  }
+
   /**
    * 保存话题信息
    */
@@ -93,13 +103,31 @@ export class SessionManager {
   }
 
   /**
-   * 清理过期会话 (超过 24 小时不活跃)
+   * 保存会话摘要（独立于 sessions 表，不受 cleanup 影响）
+   */
+  saveSummary(chatId: string, userId: string, workingDir: string, summary: string): void {
+    this.db.insertSummary(chatId, userId, workingDir, summary);
+  }
+
+  /**
+   * 获取最近 N 条会话摘要（时间正序：旧 → 新）
+   */
+  getRecentSummaries(chatId: string, userId: string, limit: number = 5): string[] {
+    return this.db.getRecentSummaries(chatId, userId, limit);
+  }
+
+  /**
+   * 清理过期会话 (超过 24 小时不活跃) 和旧摘要 (超过 30 天)
    */
   cleanup(maxIdleMs: number = 24 * 60 * 60 * 1000): number {
     const cleaned = this.db.deleteExpired(maxIdleMs);
     if (cleaned > 0) {
       logger.info({ cleaned }, 'Cleaned up idle sessions');
     }
+    const oldSummaries = this.db.cleanOldSummaries();
+    if (oldSummaries > 0) {
+      logger.info({ oldSummaries }, 'Cleaned up old session summaries');
+    }
     return cleaned;
   }