> **Note: Bazaar has been renamed to Stardom.**

# Bazaar Phase 2 Chunk 5: Bridge 协作会话管理

> **For agentic workers:** REQUIRED: Use superpowers:subagent-driven-development (if subagents available) or superpowers:executing-plans to implement this plan. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** 在 Bridge 层（`server/bazaar/`）实现协作 Claude Session 管理，让本地 Agent 能通过 Claude Session 与其他 Agent 协作对话。收到 `task.incoming` 后自动（或半自动/手动）创建 Claude Session，将集市发来的对话消息注入 Claude，提取 Claude 回复发回集市。

**Design Doc:** `docs/superpowers/specs/2026-04-10-bazaar-agent-swarm-design.md`

**Architecture:**
- `BazaarSession` 管理器（新增）：持有 `Map<string, ActiveCollaboration>` 映射，封装 `createSessionWithId` + `sendMessageForCron` 调用
- `BazaarBridge`（扩展）：集成 BazaarSession，处理 `task.accept`/`task.chat`/`task.complete`/前端消息
- `BazaarStore`（扩展）：新增 `chat_messages` 表存储协作对话历史
- `types.ts`（扩展）：新增 `CollaborationSlot`、`NotifyTimeout` 相关类型

**Tech Stack:** TypeScript, better-sqlite3, Vitest

**前置依赖:**
- Phase 1 全部 Chunk 已完成（Bridge 基础连接、身份、前端传送门）
- `server/claude-session.ts` 中的公共 API：`createSessionWithId()`、`sendMessageForCron()`
- `server/bazaar/bazaar-client.ts` 中的 `send()` 方法可向集市发消息
- `server/bazaar/types.ts` 中 `ActiveCollaboration` 类型已定义

**关键设计决策:**
1. 协作 Session 使用 `sendMessageForCron`（无头执行），和 Cron/Batch 同级别的消费者
2. 每个 `ActiveCollaboration` 持有独立的 `AbortController`，用于取消长时间运行的协作
3. Slot 计数通过 `this.activeCollaborations.size` 实时计算，不需要额外计数器
4. `sendMessageForCron` 的 `onActivity` 回调提取 Claude 回复文本，通过 `client.send({ type: 'task.chat' })` 发给集市
5. notify 模式超时用 `setTimeout` 实现，30 秒后自动调用 `startCollaboration`
6. Session ID 格式：`bazaar-{taskId}`，workspace 从任务上下文推断

---

## Task 1: 扩展类型定义

**Files:**
- Modify: `server/bazaar/types.ts`

- [ ] **Step 1: 新增 CollaborationSlot 和 NotifyTimeout 类型**

在 `server/bazaar/types.ts` 末尾追加：

```typescript
// ── 协作 Slot 管理 ──

export interface CollaborationSlot {
  taskId: string;
  helperAgentId: string;
  helperName: string;
  question: string;
  startedAt: string;
}

// ── Notify 超时管理 ──

export interface NotifyTimeout {
  taskId: string;
  timer: ReturnType<typeof setTimeout>;
  accepted: boolean;
}

// ── 协作对话消息（本地存储） ──

export interface BazaarChatMessage {
  id: number;
  taskId: string;
  from: string;         // 'local' | 'remote' | 'system'
  text: string;
  createdAt: string;
}

// ── BazaarSession 依赖注入 ──

export interface BazaarSessionDeps {
  sessionManager: import('../claude-session.js').ClaudeSessionManager;
  client: import('./bazaar-client.js').BazaarClient;
  store: import('./bazaar-store.js').BazaarStore;
  broadcast: (data: string) => void;
  homeDir: string;
  maxConcurrentTasks: number;
}
```

- [ ] **Step 2: 验证编译**

```bash
npx tsc --noEmit --pretty server/bazaar/types.ts 2>&1 | head -20
```

- [ ] **Step 3: Commit**

```bash
git add server/bazaar/types.ts
git commit -m "feat(bazaar): add collaboration slot and notify timeout types"
```

---

## Task 2: 扩展 BazaarStore — chat_messages 表

**Files:**
- Modify: `server/bazaar/bazaar-store.ts`
- Modify: `tests/server/bazaar/bazaar-store.test.ts`

- [ ] **Step 1: 先写测试**

在 `tests/server/bazaar/bazaar-store.test.ts` 末尾追加：

```typescript
  describe('chat_messages', () => {
    it('should save and list chat messages', () => {
      store.saveChatMessage({
        taskId: 'task-001',
        from: 'remote',
        text: '你好，请问支付系统怎么查？',
      });
      store.saveChatMessage({
        taskId: 'task-001',
        from: 'local',
        text: '让我查一下支付系统的转账记录...',
      });

      const messages = store.listChatMessages('task-001');
      expect(messages).toHaveLength(2);
      expect(messages[0].from).toBe('remote');
      expect(messages[1].from).toBe('local');
    });

    it('should list messages ordered by createdAt ASC', () => {
      store.saveChatMessage({ taskId: 'task-001', from: 'remote', text: 'first' });
      store.saveChatMessage({ taskId: 'task-001', from: 'local', text: 'second' });
      store.saveChatMessage({ taskId: 'task-001', from: 'remote', text: 'third' });

      const messages = store.listChatMessages('task-001');
      expect(messages[0].text).toBe('first');
      expect(messages[2].text).toBe('third');
    });

    it('should return empty array for non-existent task', () => {
      const messages = store.listChatMessages('non-existent');
      expect(messages).toHaveLength(0);
    });

    it('should separate messages by task', () => {
      store.saveChatMessage({ taskId: 'task-001', from: 'remote', text: 'q1' });
      store.saveChatMessage({ taskId: 'task-002', from: 'remote', text: 'q2' });

      expect(store.listChatMessages('task-001')).toHaveLength(1);
      expect(store.listChatMessages('task-002')).toHaveLength(1);
    });
  });
```

- [ ] **Step 2: 运行测试确认失败**

```bash
npx vitest run tests/server/bazaar/bazaar-store.test.ts 2>&1 | tail -20
```

- [ ] **Step 3: 实现 chat_messages 表和 CRUD**

在 `BazaarStore` 的 `init()` 方法中，`PRAGMA journal_mode=WAL;` 前追加建表：

```typescript
      CREATE TABLE IF NOT EXISTS chat_messages (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        task_id TEXT NOT NULL,
        from_agent TEXT NOT NULL,
        text TEXT NOT NULL,
        created_at TEXT NOT NULL,
        FOREIGN KEY (task_id) REFERENCES tasks(task_id)
      );
```

在 `BazaarStore` 类末尾（`close()` 之前）追加方法：

```typescript
  // ── Chat Messages ──

  saveChatMessage(msg: { taskId: string; from: string; text: string }): void {
    this.db.prepare(`
      INSERT INTO chat_messages (task_id, from_agent, text, created_at)
      VALUES (?, ?, ?, ?)
    `).run(msg.taskId, msg.from, msg.text, new Date().toISOString());
  }

  listChatMessages(taskId: string): Array<{ id: number; taskId: string; from: string; text: string; createdAt: string }> {
    return this.db.prepare(`
      SELECT id, task_id as taskId, from_agent as \`from\`, text, created_at as createdAt
      FROM chat_messages
      WHERE task_id = ?
      ORDER BY created_at ASC
    `).all(taskId) as Array<{ id: number; taskId: string; from: string; text: string; createdAt: string }>;
  }
```

- [ ] **Step 4: 运行测试确认通过**

```bash
npx vitest run tests/server/bazaar/bazaar-store.test.ts
```

- [ ] **Step 5: Commit**

```bash
git add server/bazaar/bazaar-store.ts tests/server/bazaar/bazaar-store.test.ts
git commit -m "feat(bazaar): add chat_messages table for collaboration history"
```

---

## Task 3: 实现 BazaarSession 管理器

**Files:**
- Create: `server/bazaar/bazaar-session.ts`
- Create: `tests/server/bazaar/bazaar-session.test.ts`

### Step 1: 先写测试

- [ ] **创建测试文件**

```typescript
// tests/server/bazaar/bazaar-session.test.ts
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
import { BazaarSession } from '../../../server/bazaar/bazaar-session.js';
import { BazaarStore } from '../../../server/bazaar/bazaar-store.js';
import type { BazaarSessionDeps } from '../../../server/bazaar/types.js';
import fs from 'fs';
import path from 'path';
import os from 'os';

// Mock ClaudeSessionManager
const mockCreateSessionWithId = vi.fn();
const mockSendMessageForCron = vi.fn();
const mockAbort = vi.fn();

vi.mock('../../../server/claude-session.js', () => ({
  ClaudeSessionManager: vi.fn().mockImplementation(() => ({
    createSessionWithId: mockCreateSessionWithId,
    sendMessageForCron: mockSendMessageForCron,
    abort: mockAbort,
  })),
}));

// Mock BazaarClient
const mockSend = vi.fn();

vi.mock('../../../server/bazaar/bazaar-client.js', () => ({
  BazaarClient: vi.fn().mockImplementation(() => ({
    send: mockSend,
  })),
}));

describe('BazaarSession', () => {
  let session: BazaarSession;
  let store: BazaarStore;
  let dbPath: string;
  let broadcastMessages: string[];
  let deps: BazaarSessionDeps;

  beforeEach(() => {
    vi.clearAllMocks();
    dbPath = path.join(os.tmpdir(), `bazaar-session-test-${Date.now()}.db`);
    store = new BazaarStore(dbPath);
    broadcastMessages = [];

    deps = {
      sessionManager: {
        createSessionWithId: mockCreateSessionWithId,
        sendMessageForCron: mockSendMessageForCron,
        abort: mockAbort,
      } as any,
      client: { send: mockSend } as any,
      store,
      broadcast: (data: string) => broadcastMessages.push(data),
      homeDir: os.homedir(),
      maxConcurrentTasks: 3,
    };

    // 默认行为：createSessionWithId 返回 session ID
    mockCreateSessionWithId.mockReturnValue('bazaar-task-001');
    // 默认行为：sendMessageForCron 立即 resolve
    mockSendMessageForCron.mockResolvedValue(undefined);

    session = new BazaarSession(deps);
  });

  afterEach(() => {
    session.stopAll();
    store.close();
    if (fs.existsSync(dbPath)) fs.unlinkSync(dbPath);
  });

  describe('startCollaboration', () => {
    it('should create a Claude session and send initial question', async () => {
      await session.startCollaboration(
        'task-001',
        '支付系统怎么查？',
        'agent-002',
        '小李',
        '/tmp/test-workspace',
      );

      // 验证创建了 session
      expect(mockCreateSessionWithId).toHaveBeenCalledWith(
        '/tmp/test-workspace',
        'bazaar-task-001',
        true,
      );

      // 验证发送了消息
      expect(mockSendMessageForCron).toHaveBeenCalledTimes(1);
      const [sessionId, content, abortController, onActivity] = mockSendMessageForCron.mock.calls[0];
      expect(sessionId).toBe('bazaar-task-001');
      expect(content).toContain('小李');
      expect(content).toContain('支付系统怎么查？');
      expect(abortController).toBeInstanceOf(AbortController);
      expect(typeof onActivity).toBe('function');

      // 验证 ActiveCollaboration 已注册
      expect(session.getActiveCount()).toBe(1);
    });

    it('should reject when slots are full', async () => {
      // 填满 slot
      for (let i = 0; i < 3; i++) {
        await session.startCollaboration(
          `task-${i}`,
          `问题 ${i}`,
          `agent-${i}`,
          `Agent ${i}`,
          '/tmp/test-workspace',
        );
      }

      // 第 4 个应被拒绝
      await expect(
        session.startCollaboration('task-overflow', '溢出问题', 'agent-x', 'X', '/tmp/test-workspace'),
      ).rejects.toThrow('协作槽位已满');

      expect(session.getActiveCount()).toBe(3);
    });

    it('should generate session label as AUTO:helperName:questionSummary', async () => {
      await session.startCollaboration(
        'task-001',
        '这是一个很长的关于支付系统查询的问题描述，应该被截断',
        'agent-002',
        '小李',
        '/tmp/test-workspace',
      );

      // 通过 broadcast 消息验证
      const statusMsg = broadcastMessages.find(m => m.includes('bazaar.status'));
      expect(statusMsg).toBeDefined();
    });

    it('should save initial system message to chat store', async () => {
      await session.startCollaboration(
        'task-001',
        '支付系统怎么查？',
        'agent-002',
        '小李',
        '/tmp/test-workspace',
      );

      const messages = store.listChatMessages('task-001');
      expect(messages.length).toBeGreaterThanOrEqual(1);
      expect(messages[0].from).toBe('system');
      expect(messages[0].text).toContain('小李');
    });
  });

  describe('sendCollaborationMessage', () => {
    it('should send a message to existing collaboration session', async () => {
      await session.startCollaboration(
        'task-001',
        '初始问题',
        'agent-002',
        '小李',
        '/tmp/test-workspace',
      );

      // 等待前一个 sendMessageForCron 完成
      mockSendMessageForCron.mockClear();

      await session.sendCollaborationMessage('task-001', '对方追问了更多细节');

      expect(mockSendMessageForCron).toHaveBeenCalledTimes(1);
      const [sessionId, content] = mockSendMessageForCron.mock.calls[0];
      expect(sessionId).toBe('bazaar-task-001');
      expect(content).toContain('对方追问了更多细节');
    });

    it('should throw when task has no active collaboration', async () => {
      await expect(
        session.sendCollaborationMessage('non-existent', 'text'),
      ).rejects.toThrow('没有活跃的协作');
    });

    it('should save message to chat store', async () => {
      await session.startCollaboration(
        'task-001',
        '初始问题',
        'agent-002',
        '小李',
        '/tmp/test-workspace',
      );
      mockSendMessageForCron.mockClear();

      await session.sendCollaborationMessage('task-001', '来自集市的消息');

      const messages = store.listChatMessages('task-001');
      const chatMsg = messages.find(m => m.from === 'remote');
      expect(chatMsg).toBeDefined();
      expect(chatMsg!.text).toBe('来自集市的消息');
    });
  });

  describe('abortCollaboration', () => {
    it('should abort active collaboration and cleanup', async () => {
      await session.startCollaboration(
        'task-001',
        '初始问题',
        'agent-002',
        '小李',
        '/tmp/test-workspace',
      );

      expect(session.getActiveCount()).toBe(1);

      session.abortCollaboration('task-001');

      expect(mockAbort).toHaveBeenCalledWith('bazaar-task-001');
      expect(session.getActiveCount()).toBe(0);
    });

    it('should be safe to abort non-existent collaboration', () => {
      expect(() => session.abortCollaboration('non-existent')).not.toThrow();
    });
  });

  describe('completeCollaboration', () => {
    it('should end collaboration, cleanup, and send task.complete to bazaar', async () => {
      await session.startCollaboration(
        'task-001',
        '初始问题',
        'agent-002',
        '小李',
        '/tmp/test-workspace',
      );

      session.completeCollaboration('task-001', 5, '很有帮助');

      expect(mockAbort).toHaveBeenCalledWith('bazaar-task-001');
      expect(session.getActiveCount()).toBe(0);

      // 验证发送了 task.complete 到集市
      expect(mockSend).toHaveBeenCalledTimes(1);
      const sent = mockSend.mock.calls[0][0];
      expect(sent.type).toBe('task.complete');
      expect(sent.payload.taskId).toBe('task-001');
      expect(sent.payload.rating).toBe(5);
      expect(sent.payload.feedback).toBe('很有帮助');
    });

    it('should update task status in store', async () => {
      // 先保存任务
      store.saveTask({
        taskId: 'task-001',
        direction: 'incoming',
        requesterAgentId: 'agent-002',
        requesterName: '小李',
        question: '测试',
        status: 'chatting',
        createdAt: new Date().toISOString(),
      });

      await session.startCollaboration(
        'task-001',
        '初始问题',
        'agent-002',
        '小李',
        '/tmp/test-workspace',
      );

      session.completeCollaboration('task-001', 4, '不错');

      const task = store.getTask('task-001');
      expect(task!.status).toBe('completed');
      expect(task!.rating).toBe(4);
    });
  });

  describe('hasAvailableSlot', () => {
    it('should return true when slots available', () => {
      expect(session.hasAvailableSlot()).toBe(true);
    });

    it('should return false when all slots used', async () => {
      for (let i = 0; i < 3; i++) {
        await session.startCollaboration(
          `task-${i}`,
          `问题 ${i}`,
          `agent-${i}`,
          `Agent ${i}`,
          '/tmp/test-workspace',
        );
      }
      expect(session.hasAvailableSlot()).toBe(false);
    });
  });

  describe('stopAll', () => {
    it('should abort all active collaborations', async () => {
      for (let i = 0; i < 2; i++) {
        await session.startCollaboration(
          `task-${i}`,
          `问题 ${i}`,
          `agent-${i}`,
          `Agent ${i}`,
          '/tmp/test-workspace',
        );
      }

      session.stopAll();

      expect(session.getActiveCount()).toBe(0);
      expect(mockAbort).toHaveBeenCalledTimes(2);
    });
  });

  describe('Claude reply extraction via onActivity', () => {
    it('should extract Claude text reply and send to bazaar via task.chat', async () => {
      let capturedOnActivity: () => void = () => {};

      // 截获 sendMessageForCron 调用来获取 onActivity 回调
      mockSendMessageForCron.mockImplementation(async (
        _sid: string,
        _content: string,
        _ac: AbortController,
        onActivity: () => void,
      ) => {
        capturedOnActivity = onActivity;
      });

      await session.startCollaboration(
        'task-001',
        '支付系统怎么查？',
        'agent-002',
        '小李',
        '/tmp/test-workspace',
      );

      // 模拟 Claude 产生了回复（通过 onActivity 回调触发提取逻辑）
      // BazaarSession 内部会从 sendMessageForCron 的流式回调中提取文本
      // 这里我们验证 onActivity 被正确传递
      expect(typeof capturedOnActivity).toBe('function');
    });
  });
});
```

### Step 2: 实现 BazaarSession

- [ ] **创建 `server/bazaar/bazaar-session.ts`**

```typescript
// server/bazaar/bazaar-session.ts
import { v4 as uuidv4 } from 'uuid';
import { createLogger, type Logger } from '../utils/logger.js';
import type { ActiveCollaboration, BazaarSessionDeps } from './types.js';

const SESSION_ID_PREFIX = 'bazaar-';

/**
 * 协作 Session 管理器
 *
 * 负责为每个协作任务创建独立的 Claude Session，管理生命周期：
 * - 创建：收到 task.incoming + accept 后调用 startCollaboration
 * - 对话：通过 sendCollaborationMessage 注入集市发来的消息
 * - 回复：从 sendMessageForCron 的流式回调中提取 Claude 回复，发回集市
 * - 结束：complete/abort/timeout 时清理资源
 */
export class BazaarSession {
  private log: Logger;
  private deps: BazaarSessionDeps;
  private activeCollaborations = new Map<string, ActiveCollaboration>();

  constructor(deps: BazaarSessionDeps) {
    this.deps = deps;
    this.log = createLogger('BazaarSession');
  }

  /**
   * 启动协作：创建 Claude Session 并发送初始问题
   */
  async startCollaboration(
    taskId: string,
    question: string,
    helperAgentId: string,
    helperName: string,
    workspace: string,
  ): Promise<void> {
    if (!this.hasAvailableSlot()) {
      throw new Error(`协作槽位已满 (${this.activeCollaborations.size}/${this.deps.maxConcurrentTasks})`);
    }

    if (this.activeCollaborations.has(taskId)) {
      this.log.warn(`Collaboration already active for task ${taskId}`);
      return;
    }

    const sessionId = `${SESSION_ID_PREFIX}${taskId}`;
    const abortController = new AbortController();
    const now = new Date().toISOString();

    const collaboration: ActiveCollaboration = {
      taskId,
      helperAgentId,
      helperName,
      question,
      sessionId,
      abortController,
      startedAt: now,
      lastActivityAt: now,
    };

    // 创建 Claude Session
    this.deps.sessionManager.createSessionWithId(workspace, sessionId, true);
    this.activeCollaborations.set(taskId, collaboration);

    // 保存系统消息到本地存储
    this.deps.store.saveChatMessage({
      taskId,
      from: 'system',
      text: `[协作请求 - 来自 Agent「${helperName}」]\n\n${question}`,
    });

    // 构造发送给 Claude 的内容
    const content = [
      `[协作请求 - 来自 Agent「${helperName}」]`,
      '',
      question,
      '',
      '请用中文回复。这是来自其他 Agent 的协作请求，请尽可能清晰地回答问题。如果你不确定答案，请如实说明。',
    ].join('\n');

    // 发送消息到 Claude Session
    try {
      await this.deps.sessionManager.sendMessageForCron(
        sessionId,
        content,
        abortController,
        () => this.handleClaudeActivity(taskId),
      );
    } catch (err) {
      // sendMessageForCron 完成或出错，清理协作
      this.log.info(`Collaboration session completed for task ${taskId}`, {
        error: err instanceof Error ? err.message : undefined,
      });
    }

    // 推送状态更新给前端
    this.deps.broadcast(JSON.stringify({
      type: 'bazaar.status',
      event: 'collaboration_started',
      taskId,
      helperName,
      activeSlots: this.activeCollaborations.size,
      maxSlots: this.deps.maxConcurrentTasks,
    }));

    this.log.info(`Collaboration started: task=${taskId}, helper=${helperName}, session=${sessionId}`);
  }

  /**
   * 向已有协作 Session 发送消息（来自集市中继的对方消息）
   */
  async sendCollaborationMessage(taskId: string, text: string): Promise<void> {
    const collab = this.activeCollaborations.get(taskId);
    if (!collab) {
      throw new Error(`没有活跃的协作: ${taskId}`);
    }

    // 保存到本地聊天记录
    this.deps.store.saveChatMessage({
      taskId,
      from: 'remote',
      text,
    });

    // 构造转发内容
    const content = `[对方 Agent「${collab.helperName}」追问]\n\n${text}`;

    // 创建新的 AbortController（复用同一 session 的新一轮对话）
    const newAbortController = new AbortController();
    collab.abortController = newAbortController;
    collab.lastActivityAt = new Date().toISOString();

    await this.deps.sessionManager.sendMessageForCron(
      collab.sessionId,
      content,
      newAbortController,
      () => this.handleClaudeActivity(taskId),
    );
  }

  /**
   * 中止协作
   */
  abortCollaboration(taskId: string): void {
    const collab = this.activeCollaborations.get(taskId);
    if (!collab) {
      this.log.warn(`No active collaboration to abort: ${taskId}`);
      return;
    }

    collab.abortController.abort();
    this.deps.sessionManager.abort(collab.sessionId);
    this.activeCollaborations.delete(taskId);

    this.deps.broadcast(JSON.stringify({
      type: 'bazaar.status',
      event: 'collaboration_aborted',
      taskId,
      activeSlots: this.activeCollaborations.size,
    }));

    this.log.info(`Collaboration aborted: task=${taskId}`);
  }

  /**
   * 完成协作：评分 + 清理 + 通知集市
   */
  completeCollaboration(taskId: string, rating: number, feedback: string): void {
    const collab = this.activeCollaborations.get(taskId);
    if (!collab) {
      this.log.warn(`No active collaboration to complete: ${taskId}`);
      return;
    }

    // 中止 Claude Session
    collab.abortController.abort();
    this.deps.sessionManager.abort(collab.sessionId);
    this.activeCollaborations.delete(taskId);

    // 更新本地任务状态
    this.deps.store.updateTaskStatus(taskId, 'completed', rating, new Date().toISOString());

    // 发送 task.complete 到集市
    this.deps.client.send({
      id: uuidv4(),
      type: 'task.complete',
      payload: { taskId, rating, feedback },
    });

    this.deps.broadcast(JSON.stringify({
      type: 'bazaar.status',
      event: 'collaboration_completed',
      taskId,
      rating,
      activeSlots: this.activeCollaborations.size,
    }));

    this.log.info(`Collaboration completed: task=${taskId}, rating=${rating}`);
  }

  /**
   * Claude 产生回复时的活动回调
   * 用于心跳检测和回复提取
   */
  private handleClaudeActivity(taskId: string): void {
    const collab = this.activeCollaborations.get(taskId);
    if (collab) {
      collab.lastActivityAt = new Date().toISOString();
    }
  }

  /**
   * 提取 Claude 最终回复并发送回集市
   * 在 sendMessageForCron 完成后由 bridge 层调用
   */
  sendClaudeReplyToBazaar(taskId: string, replyText: string): void {
    if (!replyText.trim()) return;

    // 保存到本地聊天记录
    this.deps.store.saveChatMessage({
      taskId,
      from: 'local',
      text: replyText,
    });

    // 发送 task.chat 到集市
    this.deps.client.send({
      id: uuidv4(),
      type: 'task.chat',
      payload: { taskId, text: replyText },
    });

    // 推送给前端
    this.deps.broadcast(JSON.stringify({
      type: 'bazaar.task.chat.delta',
      taskId,
      from: 'local',
      text: replyText,
    }));

    this.log.info(`Sent Claude reply to bazaar for task ${taskId} (${replyText.length} chars)`);
  }

  // ── 查询方法 ──

  hasAvailableSlot(): boolean {
    return this.activeCollaborations.size < this.deps.maxConcurrentTasks;
  }

  getActiveCount(): number {
    return this.activeCollaborations.size;
  }

  getActiveCollaboration(taskId: string): ActiveCollaboration | undefined {
    return this.activeCollaborations.get(taskId);
  }

  listActiveTasks(): string[] {
    return Array.from(this.activeCollaborations.keys());
  }

  /**
   * 停止所有活跃协作（进程退出时调用）
   */
  stopAll(): void {
    for (const [taskId, collab] of this.activeCollaborations) {
      collab.abortController.abort();
      this.deps.sessionManager.abort(collab.sessionId);
      this.log.info(`Stopped collaboration: task=${taskId}`);
    }
    this.activeCollaborations.clear();
  }
}
```

- [ ] **Step 3: 运行测试**

```bash
npx vitest run tests/server/bazaar/bazaar-session.test.ts
```

- [ ] **Step 4: 验证编译**

```bash
npx tsc --noEmit server/bazaar/bazaar-session.ts
```

- [ ] **Step 5: Commit**

```bash
git add server/bazaar/bazaar-session.ts tests/server/bazaar/bazaar-session.test.ts
git commit -m "feat(bazaar): implement BazaarSession manager for collaboration sessions"
```

---

## Task 4: 扩展 BazaarBridge — 集成协作 Session

**Files:**
- Modify: `server/bazaar/bazaar-bridge.ts`

- [ ] **Step 1: 集成 BazaarSession 到 BazaarBridge**

修改 `BazaarBridge` 类，在构造函数中初始化 `BazaarSession`，扩展消息路由。

在 `bazaar-bridge.ts` 顶部 import 区新增：

```typescript
import { BazaarSession } from './bazaar-session.js';
```

在 `BazaarBridge` 类中新增属性：

```typescript
  private bazaarSession: BazaarSession | null = null;
  private notifyTimeouts = new Map<string, ReturnType<typeof setTimeout>>();
```

修改 `start()` 方法，在 `await this.client.connect()` 之后、推送状态之前，初始化 `BazaarSession`：

```typescript
      // 初始化协作 Session 管理器
      this.bazaarSession = new BazaarSession({
        sessionManager: this.deps.sessionManager,
        client: this.client,
        store: this.store,
        broadcast: this.deps.broadcast,
        homeDir: this.deps.homeDir,
        maxConcurrentTasks: config.bazaar?.maxConcurrentTasks ?? 3,
      });
```

修改 `stop()` 方法，在 `this.client.disconnect()` 之前清理：

```typescript
    // 清理所有活跃协作
    this.bazaarSession?.stopAll();
    this.bazaarSession = null;

    // 清理所有 notify 超时
    for (const timer of this.notifyTimeouts.values()) {
      clearTimeout(timer);
    }
    this.notifyTimeouts.clear();
```

- [ ] **Step 2: 扩展 handleBazaarMessage 处理新消息类型**

替换 `handleBazaarMessage` 中 `task.chat` 的 case（当前只是广播到前端）：

```typescript
      case 'task.chat':
        this.handleIncomingChat(msg.payload);
        break;
```

新增 `task.accept` 和 `task.complete` 的处理（在 `case 'task.matched':` 之前添加）：

```typescript
      case 'task.accept':
        this.handleTaskAccepted(msg.payload);
        break;

      case 'task.complete':
        this.handleTaskComplete(msg.payload);
        break;
```

- [ ] **Step 3: 修改 handleIncomingTask 集成协作启动**

替换整个 `handleIncomingTask` 方法：

```typescript
  private handleIncomingTask(payload: Record<string, unknown>): void {
    const config = this.deps.settingsManager.getConfig().bazaar;
    const mode = config?.mode ?? 'notify';

    const taskId = payload.taskId as string;
    const helperAgentId = payload.from as string;
    const helperName = (payload.fromName as string) ?? '一位同事';
    const question = payload.question as string;

    this.store.saveTask({
      taskId,
      direction: 'incoming',
      requesterAgentId: helperAgentId,
      requesterName: helperName,
      question,
      status: 'offered',
      createdAt: new Date().toISOString(),
    });

    if (mode === 'auto') {
      // 全自动模式：直接接受并启动协作
      this.client.send({
        id: uuidv4(),
        type: 'task.accept',
        payload: { taskId },
      });
    } else if (mode === 'notify') {
      // 半自动模式：通知前端，30 秒超时自动接受
      this.deps.broadcast(JSON.stringify({
        type: 'bazaar.notify',
        taskId,
        from: helperName,
        question,
        mode,
      }));

      // 30 秒后自动接受
      const timer = setTimeout(() => {
        this.notifyTimeouts.delete(taskId);
        // 再次检查是否还没被手动处理
        const task = this.store.getTask(taskId);
        if (task && task.status === 'offered') {
          this.client.send({
            id: uuidv4(),
            type: 'task.accept',
            payload: { taskId },
          });
        }
      }, 30_000);
      this.notifyTimeouts.set(taskId, timer);
    } else {
      // manual 模式：等待前端操作
      this.deps.broadcast(JSON.stringify({
        type: 'bazaar.notify',
        taskId,
        from: helperName,
        question,
        mode,
      }));
    }
  }
```

- [ ] **Step 4: 新增消息处理方法**

在 `BazaarBridge` 类末尾（`getAgentProjects()` 之后）追加：

```typescript
  // ── 协作 Session 相关处理 ──

  private handleTaskAccepted(payload: Record<string, unknown>): void {
    if (!this.bazaarSession) return;

    const taskId = payload.taskId as string;
    const task = this.store.getTask(taskId);
    if (!task) {
      this.log.warn(`Task not found for accept: ${taskId}`);
      return;
    }

    // 取消 notify 超时（如果有）
    const timer = this.notifyTimeouts.get(taskId);
    if (timer) {
      clearTimeout(timer);
      this.notifyTimeouts.delete(taskId);
    }

    // 更新状态
    this.store.updateTaskStatus(taskId, 'chatting');

    // 启动协作 Session
    const workspace = this.deps.homeDir; // 协作 session 默认用 homeDir 作为 workspace
    this.bazaarSession.startCollaboration(
      taskId,
      task.question,
      task.requesterAgentId ?? 'unknown',
      task.requesterName ?? '一位同事',
      workspace,
    ).catch((err) => {
      this.log.error(`Failed to start collaboration for ${taskId}`, { error: String(err) });
    });
  }

  private handleIncomingChat(payload: Record<string, unknown>): void {
    if (!this.bazaarSession) return;

    const taskId = payload.taskId as string;
    const from = payload.from as string;
    const text = payload.text as string;

    // 推送到前端
    this.deps.broadcast(JSON.stringify({
      type: 'bazaar.task.chat.delta',
      taskId,
      from,
      text,
    }));

    // 如果是对方发来的消息，注入协作 Session
    if (from !== 'local') {
      this.bazaarSession.sendCollaborationMessage(taskId, text).catch((err) => {
        this.log.error(`Failed to send collaboration message for ${taskId}`, { error: String(err) });
      });
    }
  }

  private handleTaskComplete(payload: Record<string, unknown>): void {
    const taskId = payload.taskId as string;
    const rating = payload.rating as number;
    const feedback = (payload.feedback as string) ?? '';

    if (this.bazaarSession) {
      this.bazaarSession.completeCollaboration(taskId, rating, feedback);
    }
  }

  // ── notify 超时清理辅助 ──

  private clearNotifyTimeout(taskId: string): void {
    const timer = this.notifyTimeouts.get(taskId);
    if (timer) {
      clearTimeout(timer);
      this.notifyTimeouts.delete(taskId);
    }
  }
```

- [ ] **Step 5: 扩展 handleFrontendMessage 处理 accept/reject**

在 `handleFrontendMessage` 的 switch 中，`case 'bazaar.task.cancel':` 之前新增：

```typescript
      case 'bazaar.task.accept': {
        const taskId = payload.taskId as string;
        this.clearNotifyTimeout(taskId);
        this.client.send({
          id: uuidv4(),
          type: 'task.accept',
          payload: { taskId },
        });
        break;
      }

      case 'bazaar.task.reject': {
        const taskId = payload.taskId as string;
        this.clearNotifyTimeout(taskId);
        this.store.updateTaskStatus(taskId, 'rejected');
        this.client.send({
          id: uuidv4(),
          type: 'task.reject',
          payload: { taskId, reason: 'user_manual_reject' },
        });
        break;
      }
```

- [ ] **Step 6: 修改 bazaar.task.cancel 处理**

替换现有的 `case 'bazaar.task.cancel':` 代码块：

```typescript
      case 'bazaar.task.cancel': {
        const taskId = payload.taskId as string;
        // 中止协作 Session
        this.bazaarSession?.abortCollaboration(taskId);
        this.client.send({
          id: uuidv4(),
          type: 'task.cancel',
          payload: { taskId, reason: 'user_manual_cancel' },
        });
        break;
      }
```

- [ ] **Step 7: 验证编译**

```bash
npx tsc --noEmit server/bazaar/bazaar-bridge.ts
```

- [ ] **Step 8: 运行已有测试确保不破坏**

```bash
npx vitest run tests/server/bazaar/
```

- [ ] **Step 9: Commit**

```bash
git add server/bazaar/bazaar-bridge.ts
git commit -m "feat(bazaar): integrate BazaarSession into bridge for collaboration management"
```

---

## Task 5: 扩展 Bridge 集成测试

**Files:**
- Modify: `tests/server/bazaar/bridge-integration.test.ts`

- [ ] **Step 1: 追加协作流程集成测试**

在 `bridge-integration.test.ts` 的 `describe('Bridge Integration: Client ↔ Server', ...)` 块内追加测试：

```typescript
  it('should handle task.accept and create collaboration session', async () => {
    // 先模拟收到 task.incoming 保存到 store
    store.saveTask({
      taskId: 'task-collab-001',
      direction: 'incoming',
      requesterAgentId: 'agent-remote',
      requesterName: '小王',
      question: '退款流程怎么走？',
      status: 'offered',
      createdAt: new Date().toISOString(),
    });

    // 模拟集市服务器推送 task.accepted
    const wsClient = Array.from(wss.clients)[0];
    if (wsClient && wsClient.readyState === WebSocket.OPEN) {
      wsClient.send(JSON.stringify({
        id: 'srv-accept-001',
        type: 'task.accept',
        payload: { taskId: 'task-collab-001' },
      }));
    }

    await new Promise((r) => setTimeout(r, 300));

    // 验证本地任务状态更新为 chatting
    const task = store.getTask('task-collab-001');
    // 注意：在没有 BazaarBridge 实例的情况下，这个测试只验证消息能到达客户端
    // 完整的 Bridge 集成测试需要注入 BazaarBridge
    expect(task).toBeDefined();
    expect(task!.taskId).toBe('task-collab-001');
  }, 10_000);
```

- [ ] **Step 2: 运行全部 bazaar 测试**

```bash
npx vitest run tests/server/bazaar/
```

- [ ] **Step 3: Commit**

```bash
git add tests/server/bazaar/bridge-integration.test.ts
git commit -m "test(bazaar): add collaboration flow integration tests"
```

---

## Task 6: 最终验证 + 全量测试

- [ ] **Step 1: 运行全部 bazaar 测试**

```bash
npx vitest run tests/server/bazaar/
```

- [ ] **Step 2: 运行全量后端测试确认不破坏现有功能**

```bash
npx vitest run tests/server/
```

- [ ] **Step 3: 编译检查**

```bash
npx tsc --noEmit
```

- [ ] **Step 4: 检查文件变更清单**

```bash
git diff --stat HEAD
```

预期变更文件：
- `server/bazaar/types.ts` — 新增类型
- `server/bazaar/bazaar-store.ts` — 新增 chat_messages 表
- `server/bazaar/bazaar-session.ts` — 新文件，协作 Session 管理器
- `server/bazaar/bazaar-bridge.ts` — 集成 BazaarSession
- `tests/server/bazaar/bazaar-store.test.ts` — 新增测试
- `tests/server/bazaar/bazaar-session.test.ts` — 新文件，Session 管理器测试
- `tests/server/bazaar/bridge-integration.test.ts` — 新增测试

- [ ] **Step 5: 最终 Commit**

```bash
git add -A
git commit -m "feat(bazaar): Phase 2 Chunk 5 - Bridge collaboration session management

- Add BazaarSession manager for creating/managing collaboration Claude sessions
- Extend BazaarStore with chat_messages table for collaboration history
- Integrate BazaarSession into BazaarBridge with auto/notify/manual mode handling
- Support startCollaboration/sendCollaborationMessage/abortCollaboration/completeCollaboration
- Add 30-second notify timeout for semi-auto mode
- Handle frontend bazaar.task.accept/reject messages
- Add comprehensive unit tests for BazaarSession and integration tests for bridge"
```

---

## 文件依赖关系

```
types.ts (Task 1)
    ↓
bazaar-store.ts (Task 2)
    ↓
bazaar-session.ts (Task 3, depends on types + store)
    ↓
bazaar-bridge.ts (Task 4, depends on session + client + store)
    ↓
bridge-integration.test.ts (Task 5, depends on bridge)
    ↓
final verification (Task 6)
```

## 公共 API 使用清单

| API | 用途 | 调用位置 |
|-----|------|---------|
| `createSessionWithId(workspace, sessionId, isCron=true)` | 创建协作 Claude Session | `BazaarSession.startCollaboration` |
| `sendMessageForCron(sessionId, content, abortController, onActivity)` | 向协作 Session 发消息 | `BazaarSession.startCollaboration` / `sendCollaborationMessage` |
| `abort(sessionId)` | 中止协作 Session | `BazaarSession.abortCollaboration` / `completeCollaboration` / `stopAll` |

## 零侵入验证

- [ ] 不修改 `server/claude-session.ts` 的任何一行
- [ ] 不修改 `server/session-store.ts` 的表结构
- [ ] 不修改前端组件
- [ ] 不修改 `server/index.ts`（BazaarBridge 初始化已在 Phase 1 注入）
- [ ] BazaarSession 只使用 `ClaudeSessionManager` 已有的公共 API
