Process · 长进程与状态机

P-1 · L1 Unit Loop 状态机

转换表（完整）

伪代码

async function runUnit(pid, unitId) {
  const unit = unitRegistry.list(pid).find(u => u.id === unitId);
  let failures = [];

  for (let i = unit.retries; i < unit.retry_budget; i++) {
    unitRegistry.markCoding(pid, unitId);
    const coderRes = await agentManager.callAgent('coder', {
      pid, unit_id: unitId, history: failures
    });
    if (!coderRes.ok) { unitRegistry.fail(pid, unitId, 'coder_error'); continue; }

    unitRegistry.markTesting(pid, unitId);
    const testerRes = await agentManager.callAgent('tester', { pid, unit_id: unitId });
    if (testerRes.ok) {
      const hash = docStore.hashContract(await readContract(pid, unitId));
      unitRegistry.pass(pid, unitId, hash);
      return { ok: true, retries: i };
    }
    failures.push(testerRes.failures);
    unitRegistry.fail(pid, unitId, 'test_red');
  }

  unitRegistry.block(pid, unitId, 'budget_exceeded');
  return { ok: false };
}

• 同一 Unit 任何时刻只允许一个 runUnit 实例（用 Map<unitId, Promise> 去重）
• retry 之间不复用 Agent session（每次新 ACP session，保证干净）

P-2 · L2 Integration Loop 状态机

L2 闸门判定

function l2GateOpen(pid) {
  const units = unitRegistry.list(pid);
  const sdd   = docStore.readSDD(pid);
  return units.every(u => {
    if (u.status !== 'passed') return false;
    const expectedHash = contractHashFromSDD(sdd, u.id);
    return u.contract_hash === expectedHash;
  });
}

L2 主循环

async function runL2(pid) {
  const intRes = await agentManager.callAgent('integrator', { pid });
  if (intRes.ok) {
    sseBroker.emit(pid, 'project.shipped', {}); return 'shipped';
  }

  const tri = await agentManager.callAgent('bug-triager', {
    pid, run_id: intRes.run_id
  });

  if (tri.confidence < 0.5 || tri.category === 'spec_ambiguous') {
    sseBroker.emit(pid, 'human_intervention_needed', tri);
    return 'l2_blocked';
  }

  if (tri.category === 'infra_flake') {
    return runL2(pid);                           // 直接重跑，不动 Unit
  }

  if (tri.category === 'impl_bug') {
    // 不改 SDD，只 reset 受影响 Unit
    unitRegistry.reset(pid, tri.affected_units);
    return 'retry_l1';
  }

  // contract_mismatch → patch SDD + reset
  await agentManager.callAgent('architect', {
    pid, mode: 'patch', patch: tri.proposed_sdd_patch
  });
  unitRegistry.reset(pid, tri.affected_units);

  const project = getProjectMeta(pid);
  if (++project.l2_rounds >= project.l2_budget) {
    sseBroker.emit(pid, 'project.exhausted', { last_triage: tri });
    return 'l2_exhausted';
  }
  return 'retry_l1';
}

四种 triage 类别的分支

P-3 · L3 Human Feedback 队列

// 每个项目维护一个反馈队列，串行消费
const feedbackQueues = new Map<string, Feedback[]>();

function onHumanFeedback(pid, feedback) {
  if (!feedbackQueues.has(pid)) feedbackQueues.set(pid, []);
  feedbackQueues.get(pid).push(feedback);

  // 单 worker per pid
  if (!isWorking(pid)) drainFeedback(pid);
}

async function drainFeedback(pid) {
  while (feedbackQueues.get(pid).length) {
    const fb = feedbackQueues.get(pid).shift();

    // 1. 等当前 L1 / L2 跑完（不打断）
    await waitForCurrentLoop(pid);

    // 2. PM 增量补 PRD
    await agentManager.callAgent('pm', { pid, feedback: fb });

    // 3. Architect 重算 SDD + units.json (mode=generate)
    await agentManager.callAgent('architect', { pid, mode: 'generate' });

    // 4. UnitRegistry.diffWithSDD → reset changed/removed
    const diff = unitRegistry.diffWithSDD(pid);
    unitRegistry.reset(pid, [...diff.changed, ...diff.removed]);

    // 5. 触发 runProject（增量 L1 + L2）
    runProject(pid);
  }
}

• 同一 pid 永远只有一个 worker 在 drain（保证文档一致性）
• 多 pid 之间互不阻塞
• worker 崩了，重启 server 后会自动 resume（从 stream.log 反推队列状态）

P-4 · Agent 生命周期

// 复用 LemonBoss SessionPool
// key = (agentName, cid, pid)

// 1. 首次 callAgent
//    ↓ SessionPool 检查无该 key
//    ↓ 通过 ProcessLauncher 起一个 claude CLI 子进程
//    ↓ 走 ClaudeCliProtocol.initialize() + newSession()
//    ↓ 注入 system prompt (system/templates/devloop-*.md)
//    ↓ sendPrompt(input)
//    ↓ 流式累积 → AgentResult

// 2. 第二次同 key 的 callAgent
//    ↓ SessionPool 命中，复用 session
//    ↓ sendPrompt(input)  ← 上下文延续

// 3. 项目结束（shipped/abandoned）
//    ↓ SessionPool.closeProject(pid)
//    ↓ 杀掉所有该 pid 的 claude 进程

// 4. 项目空闲 30 分钟
//    ↓ 自动 sweep 该 pid 的 session（省内存）
//    ↓ 下次 callAgent 重启冷启 OK

• 每个 Agent 进程绑死自己的 cwd（沙箱）
• Bash 工具受 jsonGuard 中间件保护（防 inline JSON）
• process-guard 看护：5s SIGTERM → 10s SIGKILL

P-5 · SSE Broker 心跳与 Replay

// 每订阅一个 pid，open 一份连接列表
// subscriptions: Map<pid, Set<ServerResponse>>

function subscribe(pid, res, opts) {
  subscriptions.get(pid)?.add(res) || subscriptions.set(pid, new Set([res]));

  // Replay：从 stream.log 重放给当前订阅者
  if (opts?.replayFromOffset) replayTo(res, pid, opts.replayFromOffset);

  // 心跳
  const heartbeat = setInterval(() => res.write(':keepalive\n\n'), 25_000);
  res.on('close', () => { clearInterval(heartbeat); subscriptions.get(pid).delete(res); });
}

function emit(pid, event, data) {
  const id = ++eventCounter;
  const line = `id: ${id}\nevent: ${event}\ndata: ${JSON.stringify(data)}\n\n`;

  // 1. 落盘 stream.log
  fs.appendFileSync(streamLogPath(pid), line);
  // 2. 广播给所有订阅者
  for (const res of subscriptions.get(pid) || []) res.write(line);
}

• stream.log 保留 7 天，期间任何时刻 replay 都能精确
• 同事件不重复广播（去重以 event id 为准）

P-6 · 清理与归档

// 清理钩子标准签名（被各服务调用 + 一个日 cron）
type CleanupHook = (pid: string) => Promise<void>;
// cron: scripts/devloop-cleanup-cron.js · 每日 04:00 跑

P-7 · 预算守护

// 预算配置 schema（落在 project.json）
type ProjectBudgets = {
  retry_budget:      number;   // 默认 5 · 每 Unit
  l2_budget:         number;   // 默认 3 · 每 PRD rev
  token_budget:      number;   // 默认 5_000_000 · 项目
  agent_wallclock_ms:number;   // 默认 600_000 · 单 Agent 调用
  integrator_wallclock_ms: number; // 默认 300_000 · 单 L2 run
};

所有阈值在 project.json 可覆盖。生产建议保持默认。

P-8 · Agent State 持久化 + 崩溃恢复

调用前 / 中 / 后三阶段

// LoopController.callAgent 的完整包裹流程
async function callAgentWithPersistence(name, input) {
  const call_id  = genCallId();
  const statePath = `${pidDir}/agent-state/${name}.json`;

  // ── 调用前 ──
  const inputSnap = await snapshot(input);   // → agent-state/snapshots/{call_id}-in.json
  fs.writeFileSync(statePath, JSON.stringify({
    agent: name, call_id, status: 'running',
    current_step: 0, total_steps: 0,   // Agent 自己稍后会写
    last_input_ref: inputSnap,
    started_at: now(), updated_at: now(),
  }));
  timelineLogger.append(pid, {
    source: 'system', kind: 'agent_started',
    ref: `agent/${name}/${call_id}`,
    summary: `${name} started (unit=${input.unit_id || '-'})`
  });

  try {
    // ── 调用中 ── (Agent 自己通过工具更新)
    //   - record_memory(topic, content, relevance)
    //   - write_todolist(items)              → agent-state/{agent}-todo.json
    //   - update_todo_item(id, status, ref)  → 更新单条 item + 同步 progress_pct
    //   - update_runtime_state(current_step) → 修 agent-state/{agent}.json
    //   - timeline_append(kind, ref, summary)→ 写 timeline.ndjson
    const result = await agentManager.callAgent(name, input);

    // ── 调用后 (成功) ──
    const outSnap = await snapshot(result);
    fs.writeFileSync(statePath, JSON.stringify({
      ...readState(statePath),
      status: 'done',
      last_output_ref: outSnap,
      updated_at: now(),
    }));
    timelineLogger.append(pid, {
      source: name, kind: 'agent_done',
      ref: `agent/${name}/${call_id}`,
      summary: `${name} done in ${result.duration_ms}ms`,
      payload: { tokens: result.tokens_used }
    });
    return result;
  } catch (e) {
    // ── 调用后 (失败) ──
    fs.writeFileSync(statePath, JSON.stringify({
      ...readState(statePath),
      status: 'error',
      error: { code: e.code, message: e.message, stack: e.stack },
      updated_at: now(),
    }));
    timelineLogger.append(pid, {
      source: 'system', kind: 'agent_error',
      ref: `agent/${name}/${call_id}`,
      summary: e.message
    });
    throw e;
  }
}

崩溃恢复流程

// server 启动时调用一次
async function recoverOnBoot() {
  const projects = listAllProjects();   // 扫 data/companies/*/projects/*/
  for (const pid of projects) {
    const states = readAgentStates(pid);  // 读全部 agent-state/*.json
    for (const s of states) {
      if (s.status === 'running') {
        // 之前的进程被杀,改 paused 等待人决策
        writeState(s.agent, {
          ...s,
          status: 'paused',
          resume_token: genResumeToken(),
          updated_at: now()
        });
        timelineLogger.append(pid, {
          source: 'system', kind: 'agent_paused_by_recovery',
          ref: `agent/${s.agent}/${s.call_id}`,
          summary: 'Server 重启,Agent 已挂起等待决策'
        });
        sseBroker.emit(pid, 'resume_decision_needed', {
          agent: s.agent, call_id: s.call_id,
          last_step: s.current_step, total: s.total_steps,
          options: ['resume_from_step', 'restart_fresh', 'abort']
        });
      }
    }
  }
}

恢复决策三态

Memory 注入流程

// 每次 Agent 启动前,把相关 memory 注入 system prompt
function buildPromptWithMemory(agent, basePrompt) {
  const memDir = `${companyDir}/memory/devloop/${agent}/`;
  if (!fs.existsSync(memDir)) return basePrompt;

  const memories = fs.readdirSync(memDir)
    .map(f => parseMemory(path.join(memDir, f)))
    .filter(m => m.relevance_score >= 0.5)
    .sort((a, b) => b.relevance_score - a.relevance_score)
    .slice(0, 5);

  if (memories.length === 0) return basePrompt;

  const memSection = `## Memory (from prior projects)\n` +
    memories.map(m => `### ${m.topic}\n${m.content}`).join('\n\n');

  return basePrompt + '\n\n' + memSection;
}

• statePath 写入是原子的（fs.writeFileSync + sync flag）
• resume_token 有效期 24 小时，过期视同 restart_fresh
• memory 注入不超过 5 条，避免 prompt 膨胀
• relevance < 0.5 的 memory 不注入但保留文件
• 同一 (pid, agent) 任意时刻只允许一个 status=running

触发 SSE 事件

P-9 · Bug 状态机

状态机

合法转换

自动转换 vs 人为转换

• 自动：open → fixing → fixed → verified 主链由 LoopController + Tester + Integrator 触发
• 人为：wontfix / duplicate 必须人类显式调 API + 带 note
• 重开：fixed/verified → open 由系统自动触发但必须关联到具体 run_id + triage_id 证据

与 L1/L2 的耦合

// L1 内：Tester 失败 → 开 bug + 设 unit failed
testerResult.failures.forEach(f => {
  bugTracker.open(pid, {
    source: 'unit_test', unit_id: unit.id, run_id: runId,
    test_name: f.test, category: 'unknown',
    severity: 'high', title: f.test + ' ' + brief(f),
    detail: JSON.stringify(f), affected_units: [unit.id]
  });
});

// 下一轮 Coder 进入 → 关联 bug 状态转 fixing
bugTracker.list(pid, { unit_id: unit.id, status: ['open'] })
  .forEach(b => bugTracker.transition(pid, b.id, { to:'fixing', by:'coder' }));

// L2 通过 → 所有 fixed → verified
if (integratorResult.ok) {
  bugTracker.list(pid, { status: ['fixed'] })
    .forEach(b => bugTracker.transition(pid, b.id, { to:'verified', by:'system' }));
}

新增 SSE 事件

三层 Loop 嵌套伪代码（最终版）

// 完整调度循环（runProject 核心）
async function runProject(pid) {
  while (true) {
    // ===== L1 =====
    const waves = scheduler.getExecutionGroups(unitRegistry.pending(pid));
    for (const wave of waves) {
      await Promise.all(wave.map(u => runUnit(pid, u.id)));
    }
    if (unitRegistry.anyBlocked(pid)) return 'blocked';

    // ===== L2 =====
    if (!unitRegistry.l2GateOpen(pid)) continue;
    const r = await runL2(pid);
    if (r === 'shipped')        return 'shipped';
    if (r === 'l2_blocked')     return 'blocked';
    if (r === 'l2_exhausted')   return 'l2_exhausted';
    // r === 'retry_l1' → 继续外 while
  }
  // L3 入口 onHumanFeedback() 异步入队，由 drainFeedback 串行调用 runProject
}