Worker Service — 异步处理引擎

Hook（同步，必须快）          Worker（异步，可以慢）
┌──────────────────┐         ┌──────────────────────────┐
│ 读 stdin         │  HTTP   │ 接收观察数据              │
│ POST 到 Worker   │────────→│ 入队 pending_messages     │
│ 返回 success     │         │ 取出队列逐条处理          │
│ 耗时 < 30ms      │         │ 调用 Claude SDK 压缩      │
└──────────────────┘         │ 存入 observations 表      │
                             │ 同步 ChromaDB embedding   │
                             │ 耗时 5-30s / 条           │
                             └──────────────────────────┘

// src/services/worker-service.ts（简化）
// 端口计算：37700 + (uid % 100)
const port = getWorkerPort(); // 默认 37700
 
const server = new Server(app);
// 注册路由组
server.registerRoutes(new SessionRoutes(sessionManager, dbManager));
server.registerRoutes(new SearchRoutes(searchManager));
server.registerRoutes(new DataRoutes(dbManager));
server.registerRoutes(new ViewerRoutes());

// 简化的入队逻辑
// Worker 接收 POST /api/sessions/observations
async handleObservation(req, res) {
  const { contentSessionId, tool_name, tool_input, tool_response } = req.body;
 
  // 找到对应的数据库 session
  const session = this.dbManager.findSessionByContentId(contentSessionId);
 
  // 入队（写入 SQLite pending_messages 表）
  this.pendingStore.enqueue(session.id, {
    type: 'observation',
    tool_name,
    tool_input,
    tool_response,
  });
 
  // 立即返回（不等待处理完成）
  res.json({ status: 'queued' });
}

enqueue() → INSERT INTO pending_messages (status='pending')
    ↓
Worker 取出 → UPDATE status='processing'
    ↓
AI 压缩完成 → Parser 解析成功 → clearPendingForSession()
    ↓
如果解析失败 → 保留在队列中，等待下次尝试

// 简化的 Generator 循环
async function* processSession(session: ActiveSession) {
  while (true) {
    // 等待新的 pending messages
    const messages = await waitForPending(session.id);
    if (!messages.length) break; // session 结束
 
    // 构造 prompt，发送给 Claude SDK
    const prompt = buildCompressionPrompt(messages);
    const response = await claudeAgent.send(prompt);
 
    // 解析 AI 返回的结构化 Observation
    const parsed = parseResponse(response);
    if (parsed.valid) {
      yield parsed.observations; // 产出结构化 Observation
      clearPending(session.id);
    }
    // 无效响应：不清除队列，下次重试
  }
}

Generator 崩溃 → 等 1s → 重试 1
                → 等 2s → 重试 2
                → 等 4s → 重试 3
                → consecutiveRestarts > 3 → 放弃，让 Generator 自然结束

-- sdk_sessions 表
CREATE TABLE sdk_sessions (
  id INTEGER PRIMARY KEY,
  content_session_id TEXT NOT NULL,  -- 来自 Claude Code
  memory_session_id TEXT,            -- 来自 SDK Agent（可能变化）
  project TEXT,
  status TEXT DEFAULT 'active',
  created_at_epoch INTEGER
);

// 简化逻辑
async function ensureWorkerStarted(): Promise<void> {
  const pid = readPidFile();
  if (pid && isProcessAlive(pid)) {
    return; // Worker 已在运行
  }
  // 清理 stale PID file
  if (pid) removePidFile();
  // 启动新 Worker
  spawnDaemon();
  await waitForHealth(); // 等待 /health 端点可用
}

{
  "status": "healthy",
  "version": "12.6.2",
  "uptime": 3600,
  "sessions": { "active": 2, "total": 15 },
  "observations": { "total": 342, "pending": 3 }
}

// 每 5 分钟执行
function reapOrphans() {
  for (const [pid, process] of processRegistry) {
    if (!hasActiveSession(process.sessionId)) {
      kill(pid);
      processRegistry.delete(pid);
    }
  }
}

// src/services/infrastructure/GracefulShutdown.ts（简化）
export async function performGracefulShutdown(config: GracefulShutdownConfig): Promise<void> {
  // Step 1: 关闭 HTTP server（强制断开现有连接 + 停止监听）
  if (config.server) {
    await closeHttpServer(config.server);
  }
 
  // Step 2: 关闭所有活跃 Session（及其 SDK Agent）
  await config.sessionManager.shutdownAll();
 
  // Step 3: 关闭 MCP client（若存在）
  if (config.mcpClient) {
    await config.mcpClient.close();
  }
 
  // Step 4: 停止 ChromaDB 的 MCP 连接（若存在）
  if (config.chromaMcpManager) {
    await config.chromaMcpManager.stop();
  }
 
  // Step 5: 关闭 SQLite 数据库（若存在）
  if (config.dbManager) {
    await config.dbManager.close();
  }
 
  // Step 6: 停止子进程 supervisor
  await getSupervisor().stop();
}