Process · 长进程与状态机

所有「不是即时响应」的代码：三个 Loop 的状态机详尽转换表、Agent 生命周期、SSE 心跳与 replay、retryWithReflection、清理与归档。

P-1 · L1 Unit Loop 状态机

· loop-controller.js → runUnit()

转换表（完整）

from	to	触发	副作用
pending	coding	LoopController 调度到 wave	markCoding + SSE unit.state
coding	testing	Coder 返回 success	markTesting + SSE
coding	failed	Coder 抛错（沙箱越权 / 超时）	fail · reason=coder_error
testing	passed	Tester result.ok = true	pass + 写 contract_hash + SSE
testing	failed	Tester result.ok = false	fail · 写 result.json
failed	coding	retries	retries++ · 调 Coder 带 history
failed	blocked	retries ≥ retry_budget	block + SSE human_intervention_needed
passed	pending	L2 触发 reset	retries 重置 · 旧代码归档
blocked	pending	人类显式 resume	retries 重置

伪代码

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 去重）
• retry 之间不复用 Agent session（每次新 ACP session，保证干净）

P-2 · L2 Integration Loop 状态机

· loop-controller.js → onL2Failure() · runProject() L2 段

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 类别的分支

category	是否改 SDD	是否 reset Unit	是否计入 l2_rounds
contract_mismatch	是	是	是
impl_bug	否	是	否（不烧预算）
spec_ambiguous	否（升 L3）	否	否
infra_flake	否	否	否

P-3 · L3 Human Feedback 队列

· loop-controller.js → onHumanFeedback()

// 每个项目维护一个反馈队列，串行消费
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 生命周期

· server/src/core/session-pool.js · packages/acp-core/protocol/ClaudeCliProtocol.js · 复用既有

// 复用 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

· server/services/sse-broker.js

// 每订阅一个 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 · 清理与归档

· 分布在 doc-store.js · unit-registry.js · sse-broker.js · scripts/devloop-cleanup-cron.js (NEW)

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

触发	动作	实现位置
DocStore.write*	旧版本 → history/	doc-store.js
UnitRegistry.reset	units/{id}/code 旧版 → history/removed/{id}-{ts}/	unit-registry.js
project.shipped 30 天后	项目目录置 .archived 标记	日 cron 脚本
project.json.compressOldRuns=true	runs/ 14 天前的 stdout.txt → .gz	同上
SSE close	清 subscriptions 项	sse-broker.js
Server SIGTERM	kill 所有 ACP 子进程	process-guard 既有

P-7 · 预算守护

· server/middleware/token-accountant.js (NEW) · 阈值嵌入 loop-controller.js

// 预算配置 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
};

预算	阈值	超出后
retry_budget（每 Unit）	默认 5	该 Unit → blocked
l2_budget（每 PRD rev）	默认 3	项目 → l2_exhausted
token_budget（每项目）	默认 5_000_000	项目 → blocked + SSE 告警
Agent 单次墙钟	默认 10 min	kill + 计 fail 一次
Integrator 总墙钟	默认 5 min	kill + 算 infra_flake

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

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

· loop-controller.js · agent-state/*.json · timeline.ndjson

调用前 / 中 / 后三阶段

// 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']
        });
      }
    }
  }
}

恢复决策三态

option	含义	动作
resume_from_step	从 current_step 继续	读 last_input_ref + todolist，重新 callAgent，input 携带 history
restart_fresh	整个 Unit 重写	UnitRegistry.reset 该 unit，清 todolist，回 L1 队列
abort	放弃	UnitRegistry.block 该 unit，等下一轮人类反馈

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
• 同一 (pid, agent) 任意时刻只允许一个 status=running

触发 SSE 事件

事件	何时
agent.started	callAgentWithPersistence 进入
agent.todolist	Agent write_todolist 后
agent.step	Agent update_todo_item 后（前端用于 「3/5」）
agent.done	成功结束
agent.error	抛错
resume_decision_needed	recoverOnBoot 检测到 paused

P-9 · Bug 状态机

· server/services/bug-tracker.js → transition()

状态机

合法转换

from	to	触发	副作用
(none)	open	Tester / Integrator 失败 → bugTracker.open	+ timeline · + SSE bug.opened
open	fixing	Coder 进入 coding 态（关联同 unit）	+ status_history · + SSE
open	wontfix	人类决策	需带 note
open	duplicate	BugTracker dedupKey 命中 + 人裁决	需填 duplicate_of
fixing	fixed	同 unit Tester 绿且 fix_refs 含新 run_id	+ SSE bug.fixed
fixing	open	Coder 退出但未通过	retries++
fixed	verified	L2 Integration 全部通过	+ SSE bug.verified
fixed	open	L2 失败且 BugTriager 归因到同一 root_cause	reopened
verified	open	新 L3 反馈引入回归	reopened

自动转换 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 事件

event	触发
bug.opened	bugTracker.open
bug.transitioned	任何状态转换
bug.reopened	verified → open 特化事件

P-10 · 自动 git commit 流程

· server/src/core/loop-controller.js · callAgentWithPersistence 的 success 分支扩展

// 完整设计见 arch-git-versioning.html · 此处只放骨架

async function commitAfterAgent(pid, name, result, input) {
  const spec = buildCommitSpec(name, result, input);
  if (!spec) return;       // 某些情况不 commit (Coder 等 Tester 绿,Tester 红不 commit)

  const { ok, sha, skipped } = await gitVault.commit(pid, spec);

  if (ok && !skipped) {
    timelineLogger.append(pid, { source:'system', kind:'git_committed',
      ref: sha.slice(0,7), summary: spec.summary });
    sseBroker.emit(pid, 'git.committed', { sha, kind: spec.kind, summary: spec.summary });
  }
}

// Commit 时机表
PM 完成 prd        → kind:'prd',    msg: 'prd: rev N · {summary}'
Architect generate → kind:'sdd',    msg: 'sdd: rev N · X units · ...'
Architect patch    → kind:'sdd',    msg: 'sdd: patch · {affected_units}'
Coder + Tester 绿  → kind:'code',   msg: 'code(unit-id): implement|retry · attempt N'
Coder 单独完成     → 不 commit (等 Tester 验证)
Tester 红          → 不 commit (避免噪音)
BugTriager         → kind:'triage', msg: 'triage(run-id): {category} · confidence {n}'
Integrator         → kind:'integ',  msg: 'integ(run-id): pass|fail · X/Y'
shipped            → tag:'shipped-v{prd_rev}-{sdd_rev}'
sign-off           → tag:'final-v{prd_rev}'
abandon            → tag:'abandoned'

• Coder commit 推迟到 Tester 绿（避免红色代码进 main）
• 所有 commit author 标 Agent 名
• nothing-to-commit 返 skipped=true，不抛错
• commit 失败 → 记 warning 但不阻塞 Agent 返回

• shipped/sign-off/abandon 用 tag 而非 commit，commit 是文件变化，tag 是项目里程碑

• tests: 项目创建后 git log 含 init commit + .gitignore

• tests: PM 完成后多 1 个 prd: rev 1 commit

• tests: Coder retry 3 次但只 Tester 绿一次 → 仅 1 个 code commit

• tests: shipped → tag shipped-v1-1 存在

recoverOnBoot 加 git 清理

async function recoverOnBoot() {
  for (const pid of listAllProjects()) {
    // P-8 既有:扫 agent-state running → paused (见 P-8)
    recoverAgentStates(pid);

    // P-10 新增:扫 git 工作区脏文件
    const status = await gitVault.status(pid);
    if (status.dirty) {
      sseBroker.emit(pid, 'recovery_dirty_tree', {
        files: status.files,
        options: ['commit_as_wip', 'reset_hard', 'manual_inspect'],
      });
    }
  }
}

新增 SSE 事件

event	何时
git.committed	commitAfterAgent 成功后
git.tagged	tag 操作后（shipped/sign-off/abandon）
recovery_dirty_tree	recoverOnBoot 检测到上次 crash 留下的脏文件

三层 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
}