Codex App Server通信を追って、エージェントの UI をつくる。

import { spawn } from 'node:child_process';
import { fileURLToPath } from 'node:url';
import { RpcClient } from './rpc.mjs';

const args = process.argv.slice(2);
const live = args.includes('--live');
const handshake = args.includes('--handshake');
if (live && handshake) throw new Error('--live と --handshake は同時に使えません');
const resumeAt = args.indexOf('--resume');
const resumeId = resumeAt >= 0 ? args[resumeAt + 1] : null;
if (resumeAt >= 0 && (!resumeId || resumeId.startsWith('--'))) throw new Error('--resume に threadId が必要です');
const proc = live || handshake
  ? spawn('codex', ['app-server'], { stdio: ['pipe', 'pipe', 'inherit'] })
  : spawn(process.execPath, [fileURLToPath(new URL('./mock-server.mjs', import.meta.url))], {
      stdio: ['pipe', 'pipe', 'inherit'],
      env: { ...process.env, GUIDE_MOCK_APPROVAL: args.includes('--approval') ? '1' : '0' }
    });
const completed = new Map();
const messages = new Map();
const key = (threadId, turnId) => JSON.stringify([threadId, turnId]);
let rpc;
let deadline;
try {
  rpc = new RpcClient(proc, {
    onServerRequest: m => {
      if (['item/commandExecution/requestApproval', 'item/fileChange/requestApproval'].includes(m.method)) {
        console.log(`[approval] ${m.method} → decline（教材の既定値）`);
        return { decision: 'decline' };
      }
      // 未対応要求を許可したことにしない。RpcClient が -32601 を返す。
    },
    onNotification: m => {
      const p = m.params;
      if (m.method === 'item/agentMessage/delta') {
        const k = JSON.stringify([p.threadId, p.turnId, p.itemId]);
        messages.set(k, (messages.get(k) ?? '') + p.delta);
        process.stdout.write(p.delta);
      }
      if (m.method === 'item/completed' && p.item?.type === 'agentMessage') {
        const k = JSON.stringify([p.threadId, p.turnId, p.item.id]);
        messages.set(k, p.item.text); // 確定内容で置換。delta に追記しない。
      }
      if (m.method === 'turn/completed') completed.set(key(p.threadId, p.turn.id), p.turn);
    }
  });
  await rpc.request('initialize', { clientInfo: { name: 'app_server_handson', title: 'App Server Hands-on', version: '1.0.0' } });
  rpc.notify('initialized');
  console.log('[ready] initialize → initialized');
  if (!handshake) {
    const params = { cwd: process.cwd(), sandbox: 'read-only', approvalPolicy: 'never' };
    const result = resumeId
      ? await rpc.request('thread/resume', { ...params, threadId: resumeId })
      : await rpc.request('thread/start', params);
    const threadId = result.thread.id;
    console.log(`[thread] ${threadId}`);
    const { turn } = await rpc.request('turn/start', {
      threadId,
      input: [{ type: 'text', text: 'ツールを使わず、日本語で「接続できました」とだけ返してください。', text_elements: [] }]
    });
    const until = Date.now() + 120000;
    while (!completed.has(key(threadId, turn.id))) {
      if (rpc.closed) throw new Error('完了通知の前に接続が終了しました');
      if (Date.now() > until) {
        // タイムアウトは完了ではない。停止を要求し、成功と表示しない。
        await rpc.request('turn/interrupt', { threadId, turnId: turn.id }).catch(() => {});
        throw new Error('turn timeout: 停止要求を送信しました。状態を確認してください');
      }
      await new Promise(resolve => { deadline = setTimeout(resolve, 20); });
    }
    const final = completed.get(key(threadId, turn.id));
    console.log(`\n[turn] ${final.id}: ${final.status}`);
    console.log('[final]', [...messages.values()].join('\n'));
    if (final.status !== 'completed') throw new Error(`turn ${final.status}: ${final.error?.message ?? ''}`);
  }
} catch (error) {
  console.error(error.message);
  process.exitCode = 1;
} finally {
  clearTimeout(deadline);
  rpc?.close();
  if (!rpc) proc.kill();
}

import readline from 'node:readline';

// 受信順に依存せず、応答・通知・サーバー要求を振り分ける。
export class RpcClient {
  constructor(proc, { onNotification = () => {}, onServerRequest, timeoutMs = 15000 } = {}) {
    this.proc = proc;
    this.pending = new Map();
    this.nextId = 1;
    this.timeoutMs = timeoutMs;
    this.closed = false;
    this.rl = readline.createInterface({ input: proc.stdout });
    this.rl.on('line', line => {
      try {
        const m = JSON.parse(line);
        const hasId = Object.hasOwn(m, 'id');
        if (m.method && hasId) {
          // 両方向の id は別物。同じ値でも pending の応答として扱わない。
          Promise.resolve().then(() => onServerRequest?.(m)).then(result => {
            this.send(result === undefined
              ? { id: m.id, error: { code: -32601, message: 'Unsupported server request' } }
              : { id: m.id, result });
          }).catch(() => {
            if (!this.closed) this.send({ id: m.id, error: { code: -32603, message: 'Client handler failed' } });
          });
        } else if (m.method) {
          onNotification(m);
        } else if (hasId) {
          const p = this.pending.get(m.id);
          if (!p) return;
          clearTimeout(p.timer);
          this.pending.delete(m.id);
          m.error ? p.reject(new Error(`${m.error.code}: ${m.error.message}`)) : p.resolve(m.result);
        }
      } catch (error) { this.fail(error); }
    });
    proc.on('error', error => this.fail(error));
    proc.on('exit', (code, signal) => this.fail(new Error(`server exit: ${code ?? signal}`)));
    proc.stdin.on('error', error => this.fail(error));
    proc.stdout.on('error', error => this.fail(error));
  }
  send(message) {
    if (this.closed) throw new Error('connection closed');
    this.proc.stdin.write(JSON.stringify(message) + '\n');
  }
  request(method, params = {}) {
    if (this.closed) return Promise.reject(new Error('connection closed'));
    const id = this.nextId++;
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        this.pending.delete(id);
        reject(new Error(`timeout: ${method}; 結果不明のため自動再送しません`));
      }, this.timeoutMs);
      this.pending.set(id, { resolve, reject, timer });
      try { this.send({ id, method, params }); }
      catch (error) { clearTimeout(timer); this.pending.delete(id); reject(error); }
    });
  }
  notify(method, params = {}) { this.send({ method, params }); }
  fail(error) {
    this.closed = true;
    for (const p of this.pending.values()) { clearTimeout(p.timer); p.reject(error); }
    this.pending.clear();
  }
  close() {
    this.fail(new Error('client closed'));
    this.rl.close();
    this.proc.stdin.end();
    this.proc.kill();
  }
}

モックは初期化、開始・再開、テキスト通知、承認の一部だけを再現します。実サーバーの完全なスキーマや実行環境を代替するものではありません。

import readline from 'node:readline';
const send = m => process.stdout.write(JSON.stringify(m) + '\n');
let initialized = false;
let acknowledged = false;
let approvalId;
let turn = 0;
const finish = () => {
  const ids = { threadId: 'thr_demo', turnId: `turn_${turn}`, itemId: `msg_${turn}` };
  send({ method: 'item/started', params: { ...ids, item: { id: ids.itemId, type: 'agentMessage', text: '' } } });
  for (const delta of ['接続できました。', 'これはモック応答です。']) {
    send({ method: 'item/agentMessage/delta', params: { ...ids, delta } });
  }
  send({ method: 'item/completed', params: { ...ids, item: { id: ids.itemId, type: 'agentMessage', text: '接続できました。これはモック応答です。' } } });
  send({ method: 'turn/completed', params: { threadId: ids.threadId, turn: { id: ids.turnId, status: 'completed', items: [], error: null } } });
};
readline.createInterface({ input: process.stdin }).on('line', line => {
  const m = JSON.parse(line);
  if (!m.method) {
    if (m.id === approvalId) {
      send({ method: 'serverRequest/resolved', params: { threadId: 'thr_demo', requestId: m.id } });
      finish();
    }
    return;
  }
  if (m.method === 'initialize') {
    if (initialized) return send({ id: m.id, error: { code: -32600, message: 'Already initialized' } });
    initialized = true;
    return send({ id: m.id, result: { userAgent: 'guide-mock/1.0', platformFamily: 'unix', platformOs: 'mock' } });
  }
  if (m.method === 'initialized') { acknowledged = initialized; return; }
  if (!acknowledged) return send({ id: m.id, error: { code: -32600, message: 'Not initialized' } });
  if (m.method === 'thread/start' || m.method === 'thread/resume') {
    return send({ id: m.id, result: { thread: { id: 'thr_demo' } } });
  }
  if (m.method === 'turn/start') {
    turn++;
    send({ method: 'turn/started', params: { threadId: 'thr_demo', turn: { id: `turn_${turn}`, status: 'inProgress', items: [], error: null } } });
    if (process.env.GUIDE_MOCK_APPROVAL === '1') {
      approvalId = m.id; // 同じ id の逆向き要求でも衝突しないことを試す。
      send({ method: 'item/commandExecution/requestApproval', id: approvalId, params: { threadId: 'thr_demo', turnId: `turn_${turn}`, itemId: 'cmd_demo', command: 'echo demo', cwd: '/tmp', reason: '教材の模擬承認' } });
    } else finish(); // 応答より先に完了通知を送り、取りこぼしを試す。
    send({ id: m.id, result: { turn: { id: `turn_${turn}`, status: 'inProgress', items: [], error: null } } });
    return;
  }
  send({ id: m.id, error: { code: -32601, message: 'Method not found' } });
});

1. 複数の Item を個別の行に表示し、進行中・完了・失敗を区別する。
2. 承認要求をキューで管理し、対象と理由を表示してから判断を返す。
3. 切断時は「状態未確認」と表示し、再接続後に履歴と照合する。
4. UI から停止要求を送り、対象ターンの終了通知まで観察する。

完成条件は、途中経過が見えること、拒否や失敗も表示できること、同じ作業を勝手に重複実行しないことです。