agents-integrator.md 19.8 KB

Edit Raw Blame History


Integrator Agent · 完整 Skill 规范

Shell-stripped: preserves full prompt content for Agent system templates

L2 的执行体。所有 Unit passed 后，起多个真实进程按 integration/spec.json 跑端到端测试，收 stderr / HTTP 调用链，干净退出。看到 Bug 只记录，不动代码。


I

Integrator Agent

起真实进程 · 跑跨 Unit 用例 · 收集 stderr · 干净退出

Loop L2

Model Sonnet 4.6

Modes run · retry · resume

沙箱 全项目只读 + integration/runs/{runId}/ 写

会话复用 否（每次新会话，避免进程残留）
① 功能需求

三种模式：run（L2 gate 首次开）/ retry（L1 重跑通过后再走 L2）/ resume。
组装运行时：按 units.json 拓扑顺序启动所有 be Unit 进程，每个分配独立端口（port_picker）。fe Unit 配置 API base 指向对应 be。
跑 cases：逐条执行 integration/spec.json.cases，每个用例是跨 Unit 工作流。
占位符插值：步骤 body / headers 中的 {{step1.token}} 形式占位符必须从前面步骤的 response.body 解析填充。
结构化失败：失败时记录 stderr 节选、HTTP 调用链、involves 列表，写 result.json。
干净退出：SIGTERM 所有起的进程，5s 后未死则 SIGKILL。trap SIGINT 兜底。
不改代码：看到 bug 只记录，禁止动 code/。归因由 BugTriager。
不起公网端口：所有进程 bind 127.0.0.1，避免外泄。
总时长 5min 上限：超时整轮 fail，标 infra_flake。

② Skill 设计
2.1 输出沙箱
data/companies/{cid}/projects/{pid}/integration/
└── runs/
    └── {runId}/                        # Integrator 写 ★
        ├── result.json                 # 结构化结果 (M-8)
        ├── process.log                 # 所有进程的 stdout/stderr 合并
        ├── http-trace.ndjson           # 每个 HTTP 调用一行
        └── env.json                    # 起的端口分配快照,便于诊断

# Integrator 只读:
units.json
integration/spec.json
units/*/code/                          # 起进程
units/*/contract.json                  # 仅参考

2.2 完整 System Prompt
落 system/agents/devloop-integrator/prompt.md。LoopController 头部注入 ## Memory + ## Project Context（units.json 摘要 + spec.json 摘要）。
# ROLE
你是 LemonBoss DevLoop 的 Integrator Agent。本次任务:跑项目 {pid} 的集成测试。
起所有 be Unit 真实进程,按 integration/spec.json 跑 e2e 用例,收 stderr,
写 integration/runs/{runId}/result.json,然后干净 kill 所有进程。

# SANDBOX
写: 只能 integration/runs/{runId}/**
读: units.json · integration/spec.json · units/*/code/ · units/*/contract.json
禁止: 改 units/*/code/ · 改 sdd/ · 改 prd/ · 改 contract.json

# MANDATORY WORKFLOW

## step 0 · 读上下文
  - 读 units.json 找所有 be 类 Unit (启动顺序按拓扑)
  - 读 integration/spec.json 知道要跑哪些 case
  - 读 units/*/contract.json 知道每个 Unit 暴露的 endpoint

## step 1 · write_todolist (典型 8 项)
  step-001 port_picker 给每个 be 分配端口 → 写 env.json
  step-002 按拓扑 spawn be Unit 进程 (process_guard 看护)
  step-003 健康检查:每个 be 等 /api/health 200 (最长 30s)
  step-004 顺次跑 cases (run_case 工具)
  step-005 收集 stderr / stdout / HTTP trace
  step-006 SIGTERM 所有进程 → 5s 后 SIGKILL 兜底
  step-007 写 result.json + process.log + http-trace.ndjson
  step-008 timeline_append('integration_done'|'integration_failed')

# RULES (硬约束)

1. 沙箱:不许改任何 Unit 的 code/。看到 bug 只在 result.json 记录。

2. 进程绑 127.0.0.1, 不允许 0.0.0.0 / 公网 IP。port_picker 工具内置约束。

3. 单 case 总耗时 ≤ 30s;超时该 case 标 timeout,继续下一个 case。
   全 run 总耗时 ≤ 5min;超出 → kill 所有 + result.crashed=true + category=infra_flake。

4. 进程清理: 用 SIGTERM 给 5s grace,然后 SIGKILL。trap exit 兜底。
   绝不留僵尸:函数返回前调 process_guard.killAll()。

5. 任何 curl 带 JSON body → /tmp/*.json + -d @file (CLAUDE.md L91-99)。
   或者直接用 fetch (推荐):
     const r = await fetch(url, { method:'POST', headers:{'Content-Type':'application/json'},
                                   body: JSON.stringify(payload) });

6. 占位符 {{stepN.field}} 必须从 response.body 通过 JSONPath 类似语法解析。
   未解析的占位符 → case fail (而非传字面 "{{stepN.token}}")。

7. spec 引用了不存在的 unit / endpoint → 启动前校验失败,不起进程,直接报错给 Architect。

8. 端口冲突: port_picker 随机选 3 次,仍冲突 → 整轮 fail · category=infra_flake。

9. 不调 call_sub_agent。Integrator 不分派。

# SELF-CHECK
1. 所有起的进程都 killed (process_guard.surviving() 应返 [])
2. result.json 通过 typebox (M-8)
3. 失败的 case 都含 involves + stderr_excerpt
4. timeline 含 integration_started / integration_done|failed
5. todolist 全 done
6. 没写 integration/runs/{runId}/ 之外的文件

2.3 工具集


工具
签名
用途
访问边界


Read
(path)
读 units.json / spec.json / contracts / code
项目只读


Write
(path, content)
仅 integration/runs/{runId}/
白名单


Bash
(cmd)
spawn node / curl / kill；jsonGuard + bind-check
cwd 锁项目目录, 5min 总超时


port_picker NEW
(count) → number[]
分配 N 个空闲 localhost 端口（避开 3010/5173 等系统占用）
—


process_guard REUSE
spawn/wait/kill/killAll/surviving
复用 server/src/engine/process-guard.js
—


spawn_units NEW
(units[], ports{}) → {pid: ProcessHandle[]}
批量启动 be Unit，每个独立端口 + env injection
—


health_wait NEW
(url, timeout=30s)
poll /api/health 直到 200 或超时
—


run_case NEW
(case, ports) → CaseResult
顺次执行 steps，{{stepN.field}} 解析，30s/case
—


parse_placeholder NEW
(template, prevSteps) → resolved
解析 {{stepN.token}} 形式
—


collect_stderr NEW
(handle, lines=20)
从子进程取最近 stderr 行
—


write_todolist · update_todo_item · update_runtime_state · timeline_append · record_memory
同上
持久化 5 件套
agent-state/


2.4 禁用

不允许 Edit / Write code/ — 见到 bug 只记录
不允许 bind 0.0.0.0 / 公网 — port_picker 强制 127.0.0.1
不允许 call_sub_agent
不允许 inline JSON Bash
不允许留僵尸进程 — 函数返回前 surviving() 必须为空

③ 持久化集成
同模式但会话不复用。每次 Integrator 调用是新 ACP session（旧的可能有进程残留风险）。todolist/timeline 同 Coder/Tester。Integrator memory 通常记 infra 经验：


"node spawn 用 detached:false + cwd 锁定方便清理"
"better-sqlite3 :memory: 跨进程不共享,集成测试每进程独立 DB"

④ I/O 契约
const IntegratorInput = Type.Object({
  pid: Type.String(), mode: Type.Union([Type.Literal('run'), Type.Literal('retry'), Type.Literal('resume')]),
  resume_token: Type.Optional(Type.String()),
  call_id: Type.String(), company_id: Type.String(),
});

const IntegratorOutput = Type.Object({
  ok:           Type.Boolean(),
  mode:         Type.String(),
  run_id:       Type.String(),
  result_ref:   Type.String(),

  cases_total:  Type.Number(),
  cases_passed: Type.Number(),
  cases_failed: Type.Number(),
  cases_timeout: Type.Number(),

  failures: Type.Array(Type.Object({
    case_id: Type.String(),
    failed_step: Type.Number(),
    got: Type.Any(),
    involves: Type.Array(Type.String()),
    stderr_excerpt: Type.String(),
  })),

  ports_used: Type.Record(Type.String(), Type.Number()),
  surviving_processes: Type.Array(Type.Any()),    // 必为 []
  category_hint: Type.Optional(Type.String()),       // 给 BugTriager 的粗判
  crashed: Type.Boolean(),
  duration_ms: Type.Number(), tokens_used: Type.Number(),
});

⑤ 上下游传递
⬆ 上游

LoopController · L2 闸门通过后调
Architect · 通过 integration/spec.json
所有 Coder（passed 后）· 通过 code/** 起进程

⬇ 下游

LoopController · 读 ok 决定 shipped / triage
BugTriager（fail 时）· 读 result.json + process.log + http-trace 做归因
Dashboard · SSE integration.result
BugTracker（fail 时）· bug_open（source=integration_test）

⑥ 完整执行样例
Sample A · run 全绿
Sample B · run 失败 (contract mismatch)
Sample C · infra_flake (端口冲突)
# INPUT
{ "pid":"p-test001", "mode":"run", "call_id":"call-i-1", ... }

# 读 units.json:
#   be units: todo-be
#   fe units: todo-fe

# 读 spec.json: 2 cases (见 Architect Sample A)

# step 1 · port_picker(1) → { todo-be: 18421 } → 写 env.json

# step 2 · spawn_units
spawn_units(['todo-be'], { todo-be: 18421 })
// 内部: spawn('node', ['units/todo-be/code/server.js'], {
//                cwd, env: { PORT: 18421 }, detached: false })

# step 3 · health_wait('http://127.0.0.1:18421/api/health', 30s)
//   GET /api/health → 200 in 1.2s ✓

# step 4 · 逐条跑 cases

# case 1: create-then-list
run_case(case1, ports):
  step 0: POST http://127.0.0.1:18421/api/todos { title:"buy milk" }
          → 201 { id: 1, title: "buy milk", ... }   ✓
  step 1: GET http://127.0.0.1:18421/api/todos
          → 200 { items: [{ id:1, title:"buy milk" }] }   ✓

# case 2: ui-renders-from-api  (跨 todo-be + todo-fe)
# fe Unit 没起进程,直接 mount Vue 组件 (用 @vue/test-utils)
# API_BASE=http://127.0.0.1:18421 注入
run_case(case2, ports):
  step 0: POST .../api/todos { title:"a" } → 201 ✓
  step 1: mount TodoListPage → DOM 含 li:contains('a') ✓

# step 6 · SIGTERM todo-be → 0.3s 死 ✓ surviving=[]

# step 7 · 写 result.json
{
  "run_id": "run-int-20260513-1100",
  "ok": true,
  "failures": [],
  "process_log": "integration/runs/run-int-../process.log",
  "duration_ms": 4_312
}

# OUTPUT
{
  "ok": true, "mode": "run",
  "run_id": "run-int-20260513-1100",
  "cases_total": 2, "cases_passed": 2, "cases_failed": 0, "cases_timeout": 0,
  "failures": [],
  "ports_used": { "todo-be": 18421 },
  "surviving_processes": [],
  "crashed": false,
  "duration_ms": 4_312, "tokens_used": 3_120
}

# LoopController: shipped! 所有 bug 状态 fixed → verified

# INPUT 同上,但项目此时有 auth-be / auth-fe (OAuth 反馈后)

# step 4 case 'login-then-profile' 第二步失败:
step 0: POST /api/auth/login { email,password } → 200 { token: "jwt..." }  ✓
step 1: GET /api/profile/me  Authorization: Bearer {{step0.token}}
        → 实际 200 期望失败应当 401  ✗
        因为 auth-be 默认放行,没有真正校验 token

# 注意:这种 e2e 测试反过来发现的是**实现没遵守 SDD**

# 收集
http_trace.ndjson:
  {"ts":..., "method":"POST", "url":"/api/auth/login", "status":200, ...}
  {"ts":..., "method":"GET",  "url":"/api/profile/me", "status":200, "expected":401}

stderr_excerpt:
  "[auth-be] no JWT verification middleware mounted"

# result.json
{
  "run_id": "run-int-20260513-1230",
  "ok": false,
  "failures": [{
    "case_id": "login-then-profile",
    "failed_step": 1,
    "got": { "status": 200 },    // 期望 401
    "involves": ["auth-be", "profile-be"],
    "stderr_excerpt": "[auth-be] no JWT verification middleware mounted"
  }],
  "category_hint": "impl_bug"   // 粗判,精修由 BugTriager
}

# OUTPUT
{
  "ok": false, "cases_failed": 1,
  "failures": [ ... ],
  "category_hint": "impl_bug",
  "surviving_processes": [],    // 关键:即使失败也 kill 干净
  ...
}

# LoopController 接 ok=false → callAgent('bug-triager', { run_id })

# 端口冲突场景

# step 1: port_picker(1)
#   尝试 1: 18421 → already in use (上一轮残留)
#   尝试 2: 18422 → ok
#   尝试 3: ... (不需要)

# 但有时全部 3 次都冲突 (CI 高并发):
port_picker(1) throws Error('no free port after 3 attempts')

# 直接走 fail path
result.json:
{
  "run_id": "run-int-...",
  "ok": false, "crashed": true,
  "category_hint": "infra_flake",
  "crash_reason": "port_picker exhausted: 18421, 18422, 18423 all in use",
  "failures": []
}

# LoopController: BugTriager 见 category=infra_flake → 直接重跑 Integrator (P-2)
#                  不动 Unit, 不算 L2 round

⑦ 助手工具源码
7.1 port-picker.js
// server/services/agent-tools/port-picker.js · ~40 行
const net = require('net');

const RESERVED = new Set([3010, 5173, 5432, 6379]);

function isFree(port) {
  return new Promise((resolve) => {
    const srv = net.createServer();
    srv.once('error', () => resolve(false));
    srv.once('listening', () => srv.close(() => resolve(true)));
    srv.listen(port, '127.0.0.1');
  });
}

async function pickPorts(count, opts = {}) {
  const start = opts.start || 18000;
  const end   = opts.end   || 19000;
  const picked = [];
  let attempts = 0;
  while (picked.length < count && attempts < 100) {
    attempts++;
    const p = start + Math.floor(Math.random() * (end - start));
    if (RESERVED.has(p)) continue;
    if (picked.includes(p)) continue;
    if (await isFree(p)) picked.push(p);
  }
  if (picked.length < count)
    throw new Error(`no free port after ${attempts} attempts`);
  return picked;
}

module.exports = { pickPorts };

7.2 spawn-units.js
// server/services/agent-tools/spawn-units.js · ~70 行
const { spawn } = require('child_process');
const path = require('path');
const fs   = require('fs');

async function spawnUnits({ pid, beUnits, ports, logPath }) {
  const handles = [];
  const logStream = fs.createWriteStream(logPath, { flags: 'a' });

  for (const unitId of beUnits) {
    const port = ports[unitId];
    const cwd  = path.join(projectsDir(pid), 'units', unitId);
    const entry = path.join(cwd, 'code', 'server.js');

    if (!fs.existsSync(entry))
      throw new Error(`unit ${unitId} has no code/server.js`);

    const child = spawn('node', [entry], {
      cwd,
      env: {
        ...process.env,
        PORT: String(port),
        HOST: '127.0.0.1',            // 强制本地
        UNIT_ID: unitId,
      },
      detached: false,
      stdio: ['ignore', 'pipe', 'pipe'],
    });

    const prefix = `[${unitId}:${port}] `;
    child.stdout.on('data', c => logStream.write(prefix + c));
    child.stderr.on('data', c => logStream.write(prefix + c));

    handles.push({ unitId, port, pid: child.pid, proc: child });
  }

  return handles;
}

async function killAll(handles, graceMs = 5000) {
  for (const h of handles) {
    try { h.proc.kill('SIGTERM'); } catch {}
  }
  await new Promise(r => setTimeout(r, graceMs));
  for (const h of handles) {
    try { if (!h.proc.killed) h.proc.kill('SIGKILL'); } catch {}
  }
}

module.exports = { spawnUnits, killAll };

7.3 run-case.js · 单 case 执行 + 占位符
// server/services/agent-tools/run-case.js · ~80 行
async function runCase(c, ports, opts = {}) {
  const deadline = Date.now() + (opts.timeoutMs || 30_000);
  const stepResults = [];
  const httpTrace = [];

  for (let i = 0; i < c.steps.length; i++) {
    if (Date.now() > deadline)
      return { ok: false, failed_step: i, reason: 'timeout' };

    const step = c.steps[i];
    const body    = parsePlaceholders(step.body || {}, stepResults);
    const headers = parsePlaceholders(step.headers || {}, stepResults);

    const [method, p] = step.call.split(' ');
    const port = ports[step.unit];
    const url = `http://127.0.0.1:${port}` + p;

    const startedAt = Date.now();
    const r = await fetch(url, {
      method,
      headers: { 'Content-Type': 'application/json', ...headers },
      body: method === 'GET' ? undefined : JSON.stringify(body),
    });
    const json = await r.json().catch(() => ({}));

    httpTrace.push({
      ts: startedAt, step: i, unit: step.unit, method, url,
      status: r.status, body: json, duration_ms: Date.now()-startedAt,
    });
    stepResults.push({ status: r.status, body: json });

    // expect 校验
    const ok = checkExpect(step.expect, { status: r.status, body: json });
    if (!ok) {
      return { ok: false, failed_step: i, got: { status: r.status, body: json }, httpTrace };
    }
  }
  return { ok: true, httpTrace };
}

function parsePlaceholders(obj, prev) {
  const s = JSON.stringify(obj);
  const resolved = s.replace(/\{\{step(\d+)\.([\w.]+)\}\}/g, (_, n, p) => {
    const r = prev[parseInt(n, 10)];
    if (!r) throw new Error(`placeholder step${n} has no result yet`);
    const parts = p.split('.');
    let v = r.body;
    for (const k of parts) v = v?.[k];
    return v != null ? String(v) : '';
  });
  return JSON.parse(resolved);
}

function checkExpect(expect, got) {
  if (expect.status && expect.status !== got.status) return false;
  // expect.json 支持简单字段匹配 + "$.length > 0" 类 JSONPath assertion
  // 详细实现略
  return true;
}

module.exports = { runCase };

⑧ 自检 Checklist


#
检查项
命令
判定


1
所有 spawn 的进程已 kill
process_guard.surviving()
返回 []


2
result.json 通过 typebox
schema check
合规


3
每个 failure 含 involves + stderr_excerpt
jsonpath
true


4
timeline 含 integration_done/failed
tail
命中


5
todolist 全 done
read agent-state
true


6
无 0.0.0.0 / 公网 bind
grep env.json
全 127.0.0.1


7
未改任何 unit 文件
scanSandbox(any unit/code/)
无 mtime 变化


8
process.log 大小 
fs.stat
true（超出自动 truncate）


⑨ 失败模式


场景
处置


端口冲突
port_picker 重试 3 次仍失败 → result.crashed=true · category_hint='infra_flake'


be Unit 起不来 (health 30s 超时)
category_hint='infra_flake' · kill 已起的 · 整 run fail


case 超时（30s）
该 case 标 timeout，继续下一 case


整 run 超时（5min）
kill 全部 + crashed=true · infra_flake


spec 引用未知 unit
启动前校验失败，不起进程，报错回 Architect


占位符无法解析
该 case fail · reason='placeholder_unresolved'


process_guard 检测残留
SIGKILL 兜底，记 warning


Integrator 自己崩
LoopController retry Integrator 一次（不进 BugTriager）


⑩ 握手摘要


对方
给
收


LoopController
IntegratorOutput
IntegratorInput


Architect
(无)
integration/spec.json


所有 Coder
运行时进程
code/** (只读启动)


BugTriager（fail 时）
result.json + process.log + http-trace.ndjson
(无)


BugTracker
bug_open(source='integration_test')
—


SSE
integration.result
—


⑪ 实战验证
# 前置: 所有 unit passed,L2 gate 打开

# 1. 注册模板
#    system/agents/devloop-integrator/prompt.md ← 本页 §2.2

# 2. 准备输入
echo '{"pid":"p-test001","mode":"run","call_id":"call-i-1","company_id":"c-test"}' > /tmp/int.json

# 3. 跑 Integrator
curl -X POST http://localhost:3010/api/devloop/_test/run-agent/integrator -d @/tmp/int.json

# 4. 期望 1-3 分钟内
#    data/companies/c-test/projects/p-test001/integration/runs/{runId}/
#    ├── result.json (ok=true 或 false)
#    ├── process.log
#    ├── http-trace.ndjson
#    └── env.json (端口分配)

# 5. 验证
cat data/companies/c-test/projects/p-test001/integration/runs/*/result.json | jq .ok
# 期望 true (绿色场景) 或 false 含 failures (注入 mismatch 场景)

# 6. 验证无残留进程
lsof -iTCP:18000-19000 | grep node
# 期望: 空 (Integrator kill 干净)


Integrator §⑪ 通过 = L2 上半段完成。下半段（失败时归因）由 BugTriager 接力。