agent-loop/demo/cloud/server.js

// 云端控制面：Session 存储 + 设备通道 + 调度器。
// 它不调模型、不碰代码，只做三件事：
//   1. 把事件追加进 Session（写到 cloud/sessions/<id>.jsonl，相当于数据库）
//   2. 让本地守护进程长轮询领取工具调用、回传结果
//   3. 看到 Session 有活要干、又没有 Worker 在跑时，拉起一个新的 Worker 进程
const http = require('node:http');
const fs = require('node:fs');
const path = require('node:path');
const { spawn } = require('node:child_process');
const { tracer, pendingWork } = require('../shared');

const PORT = Number(process.env.PORT || 7700);
const DATA = process.env.CLOUD_DATA;
const trace = tracer('cloud-server');
fs.mkdirSync(path.join(DATA, 'sessions'), { recursive: true });

// ---------- Session 存储 ----------
const sessFile = (id) => path.join(DATA, 'sessions', `${id}.jsonl`);
function readEvents(id) {
  if (!fs.existsSync(sessFile(id))) return [];
  return fs.readFileSync(sessFile(id), 'utf8').trim().split('\n').filter(Boolean).map((l) => JSON.parse(l));
}
function appendEvent(id, ev) {
  const events = readEvents(id);
  const rec = { seq: events.length + 1, at: Date.now(), ...ev };
  fs.appendFileSync(sessFile(id), JSON.stringify(rec) + '\n');
  trace('session_append', { session: id, seq: rec.seq, type: rec.type, bytes: JSON.stringify(rec).length, by: ev.by });
  setTimeout(() => schedule(id), 0);
  return rec;
}

// ---------- 设备通道 ----------
const devices = {}; // id -> { waiting: res|null, queue: [], pending: Map(reqId -> res), lastSeen }
const dev = (id) => (devices[id] ||= { waiting: null, queue: [], pending: new Map(), lastSeen: 0, online: false });
function isOnline(id) {
  const d = devices[id];
  return !!(d && d.online && (d.waiting || Date.now() - d.lastSeen < 20000));
}
function flush(d) {
  if (d.waiting && d.queue.length) {
    const res = d.waiting; d.waiting = null; d.lastSeen = Date.now();
    res.end(JSON.stringify(d.queue.shift()));
  }
}

// ---------- 调度器 ----------
const running = {}; // session -> pid
const sessionDevice = {}; // session -> device
function schedule(id) {
  if (running[id]) return;
  const work = pendingWork(readEvents(id));
  if (work.kind === 'idle') return;
  if (work.kind === 'dispatch' && !isOnline(sessionDevice[id])) return; // 设备不在线：不拉起 Worker，等它回来
  const child = spawn(process.execPath, [path.join(__dirname, 'loop-worker.js')], {
    cwd: path.join(DATA, 'worker-scratch'),
    env: { ...process.env, SESSION_ID: id, CLOUD_URL: `http://127.0.0.1:${PORT}`, HOST_LABEL: 'cloud-worker' },
    stdio: ['ignore', 'inherit', 'inherit'],
  });
  running[id] = child.pid;
  trace('worker_spawn', { session: id, worker_pid: child.pid, reason: work.kind });
  child.on('exit', (code, signal) => {
    trace('worker_exit', { session: id, worker_pid: child.pid, code, signal });
    delete running[id];
    setTimeout(() => schedule(id), signal ? 800 : 0); // 被杀掉的话稍等一下再看，模拟“故障检测”
  });
}
fs.mkdirSync(path.join(DATA, 'worker-scratch'), { recursive: true });

// ---------- HTTP ----------
const body = (req) => new Promise((ok) => { let s = ''; req.on('data', (c) => (s += c)); req.on('end', () => ok(s ? JSON.parse(s) : {})); });
const json = (res, code, obj) => { res.writeHead(code, { 'content-type': 'application/json' }); res.end(JSON.stringify(obj)); };

http.createServer(async (req, res) => {
  const url = new URL(req.url, 'http://x');
  const p = url.pathname.split('/').filter(Boolean);
  const handled = (req.method === 'POST' && p[0] === 'dispatch') || (req.method === 'GET' && p[0] === 'devices' && p[2] === 'poll');
  try {
    // 用户 / 外部事件 / Worker 往 Session 里追加事件
    if (req.method === 'POST' && p[0] === 'sessions' && p[2] === 'events') {
      const ev = await body(req);
      if (ev.device) sessionDevice[p[1]] = ev.device;
      return json(res, 200, appendEvent(p[1], ev));
    }
    if (req.method === 'GET' && p[0] === 'sessions' && p[2] === 'events') return json(res, 200, readEvents(p[1]));
    if (req.method === 'GET' && p[0] === 'sessions' && p[2] === 'device') return json(res, 200, { device: sessionDevice[p[1]], online: isOnline(sessionDevice[p[1]]) });

    // Worker 把一次工具调用交给设备，同步等结果；设备不在线立刻返回 offline
    if (req.method === 'POST' && p[0] === 'dispatch') {
      const cmd = await body(req);
      const d = dev(cmd.device);
      if (!isOnline(cmd.device)) {
        trace('dispatch_offline', { device: cmd.device, call_id: cmd.call_id, tool: cmd.tool });
        return json(res, 200, { offline: true });
      }
      cmd.req_id = `${cmd.call_id}`;
      d.pending.set(cmd.req_id, res);
      d.queue.push(cmd);
      trace('dispatch', { device: cmd.device, call_id: cmd.call_id, tool: cmd.tool, input: cmd.input });
      flush(d);
      return void 0;
    }

    // 守护进程长轮询领取命令（设备在 NAT 后面，只能它主动连云端）
    if (req.method === 'GET' && p[0] === 'devices' && p[2] === 'poll') {
      const d = dev(p[1]);
      const wasOnline = isOnline(p[1]);
      d.online = true; d.lastSeen = Date.now(); d.waiting = res;
      if (!wasOnline) {
        trace('device_online', { device: p[1] });
        for (const s of Object.keys(sessionDevice)) if (sessionDevice[s] === p[1]) setTimeout(() => schedule(s), 50);
      }
      req.on('close', () => { if (d.waiting === res) { d.waiting = null; d.lastSeen = Date.now(); } });
      flush(d);
      setTimeout(() => { if (d.waiting === res) { d.waiting = null; d.lastSeen = Date.now(); res.end('null'); } }, 15000);
      return;
    }
    if (req.method === 'POST' && p[0] === 'devices' && p[2] === 'result') {
      const r = await body(req);
      const d = dev(p[1]); d.lastSeen = Date.now();
      const waiter = d.pending.get(r.req_id);
      d.pending.delete(r.req_id);
      if (waiter) waiter.end(JSON.stringify(r));
      return json(res, 200, { ok: true });
    }
    if (req.method === 'POST' && p[0] === 'devices' && p[2] === 'bye') {
      const d = dev(p[1]); d.online = false;
      trace('device_offline', { device: p[1] });
      return json(res, 200, { ok: true });
    }
    if (req.method === 'GET' && p[0] === 'status') return json(res, 200, { running, devices: Object.fromEntries(Object.keys(devices).map((k) => [k, isOnline(k)])) });
    if (!res.headersSent && !res.writableEnded && !handled) json(res, 404, { error: 'not found' });
  } catch (e) {
    console.error('[cloud-server]', e);
    if (!res.headersSent) json(res, 500, { error: String(e) });
  }
}).listen(PORT, () => trace('server_start', { port: PORT, data_dir: DATA }));