81 lines
5.5 KiB
JavaScript
81 lines
5.5 KiB
JavaScript
// 云端 Agent Loop Worker：无状态、用完即退。
|
||
// 每次被调度器拉起时：读 Session 日志 → 判断下一步 → 调模型或把工具调用发给设备 → 结果写回 Session。
|
||
// 进程内存里没有任何必须保存的东西：随时 kill -9，下一个 Worker 读日志就能接上。
|
||
const { tracer, pendingWork, rebuildMessages } = require('../shared');
|
||
|
||
const SESSION = process.env.SESSION_ID;
|
||
const CLOUD = process.env.CLOUD_URL;
|
||
const GATEWAY = process.env.GATEWAY_URL; // 私有网关，Anthropic Messages 协议
|
||
const MODEL = process.env.GATEWAY_MODEL;
|
||
const trace = tracer('cloud-worker');
|
||
|
||
const SYSTEM = [
|
||
'你是运行在云端的编码 Agent。用户的代码仓库在用户自己的电脑上，你只能通过工具在那台电脑上读写文件、执行命令。',
|
||
'完整测试 `npm test` 含约 12 秒的集成测试，必须用 start_background 在后台运行，启动后直接结束本轮回复、简短告诉用户已启动即可，不要等待；结果会以“[后台任务完成]”消息送回给你。',
|
||
'单元测试可以用 run_command 执行 `node --test test/token.test.js`。',
|
||
'修改要最小。不要提交 git。回复用中文，简短。',
|
||
].join('\n');
|
||
|
||
const TOOLS = [
|
||
{ name: 'read_file', description: '读取用户电脑上仓库里的文件（带行号）', input_schema: { type: 'object', properties: { path: { type: 'string' } }, required: ['path'] } },
|
||
{ name: 'edit_file', description: '把文件中唯一出现的 old_text 替换为 new_text', input_schema: { type: 'object', properties: { path: { type: 'string' }, old_text: { type: 'string' }, new_text: { type: 'string' } }, required: ['path', 'old_text', 'new_text'] } },
|
||
{ name: 'write_file', description: '创建或覆盖文件', input_schema: { type: 'object', properties: { path: { type: 'string' }, content: { type: 'string' } }, required: ['path', 'content'] } },
|
||
{ name: 'run_command', description: '在仓库根目录同步执行一条 shell 命令（15 秒超时），返回退出码和输出', input_schema: { type: 'object', properties: { command: { type: 'string' } }, required: ['command'] } },
|
||
{ name: 'start_background', description: '在仓库根目录后台启动一条命令，立即返回 job_id；完成后结果会作为新消息送回', input_schema: { type: 'object', properties: { command: { type: 'string' } }, required: ['command'] } },
|
||
];
|
||
|
||
const get = async (p) => (await fetch(CLOUD + p)).json();
|
||
const post = async (p, b) => (await fetch(CLOUD + p, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(b) })).json();
|
||
const append = (ev) => post(`/sessions/${SESSION}/events`, { ...ev, by: `worker:${process.pid}` });
|
||
|
||
async function callModel(events) {
|
||
const messages = rebuildMessages(events);
|
||
const reqBody = { model: MODEL, max_tokens: 2000, system: SYSTEM, tools: TOOLS, messages };
|
||
const started = Date.now();
|
||
trace('model_request', { session: SESSION, url: GATEWAY + '/v1/messages', model: MODEL, message_count: messages.length, request_bytes: JSON.stringify(reqBody).length, request: reqBody });
|
||
const r = await fetch(GATEWAY + '/v1/messages', {
|
||
method: 'POST',
|
||
headers: { 'content-type': 'application/json', 'anthropic-version': '2023-06-01', Authorization: `Bearer ${process.env.GATEWAY_KEY}` },
|
||
body: JSON.stringify(reqBody),
|
||
});
|
||
const text = await r.text();
|
||
if (!r.ok) throw new Error(`gateway ${r.status}: ${text.slice(0, 300)}`);
|
||
const d = JSON.parse(text);
|
||
trace('model_response', { session: SESSION, ms: Date.now() - started, status: r.status, stop_reason: d.stop_reason, usage: d.usage, response_id: d.id, content: d.content });
|
||
return d;
|
||
}
|
||
|
||
async function main() {
|
||
trace('worker_boot', { session: SESSION, holds_repo: false, cwd: process.cwd() });
|
||
for (let step = 0; step < 12; step++) {
|
||
const events = await get(`/sessions/${SESSION}/events`);
|
||
const work = pendingWork(events);
|
||
trace('worker_decide', { session: SESSION, event_count: events.length, decision: work.kind, open_calls: work.calls?.map((c) => c.id) });
|
||
if (work.kind === 'idle') break;
|
||
|
||
if (work.kind === 'dispatch') {
|
||
const { device } = await get(`/sessions/${SESSION}/device`);
|
||
for (const call of work.calls) {
|
||
const started = Date.now();
|
||
const r = await post('/dispatch', { device, session: SESSION, call_id: call.id, tool: call.name, input: call.input });
|
||
if (r.offline) {
|
||
await append({ type: 'waiting_device', device, call_id: call.id, tool: call.name });
|
||
trace('worker_park', { session: SESSION, reason: 'device offline', call_id: call.id });
|
||
return; // 不空等：直接退出，设备回来时调度器会再拉起一个 Worker
|
||
}
|
||
trace('tool_roundtrip', { session: SESSION, call_id: call.id, tool: call.name, ms: Date.now() - started, ran_on: r.host, ran_pid: r.pid });
|
||
await append({ type: 'tool_result', call_id: call.id, tool: call.name, output: r.output, ran_on: r.host, ran_pid: r.pid, repo_after: r.repo_after });
|
||
}
|
||
continue;
|
||
}
|
||
|
||
const d = await callModel(events);
|
||
await append({ type: 'model_response', content: d.content, stop_reason: d.stop_reason, usage: d.usage, model: MODEL });
|
||
const textOut = d.content.filter((b) => b.type === 'text').map((b) => b.text).join('\n').trim();
|
||
if (d.stop_reason !== 'tool_use' && textOut) await append({ type: 'reply_to_user', text: textOut });
|
||
}
|
||
trace('worker_done', { session: SESSION });
|
||
}
|
||
|
||
main().catch((e) => { trace('worker_error', { session: SESSION, error: String(e) }); process.exit(1); });
|