my-calm-game/development/_reference/ai/scripts/run-pi-worker.mjs

#!/usr/bin/env node

import { createWriteStream, existsSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs';
import { join, resolve } from 'node:path';
import { execFileSync, spawn } from 'node:child_process';

function parseArgs(argv) {
  const out = {};
  for (let i = 0; i < argv.length; i += 1) {
    const arg = argv[i];
    if (!arg.startsWith('--')) {
      out._ = out._ || [];
      out._.push(arg);
      continue;
    }
    const key = arg.slice(2);
    if (key === 'dry-run') {
      out[key] = true;
      continue;
    }
    const value = argv[i + 1];
    if (!value || value.startsWith('--')) {
      throw new Error(`Missing value for --${key}`);
    }
    out[key] = value;
    i += 1;
  }
  return out;
}

function usage() {
  return `Usage:
  node development/_reference/ai/scripts/run-pi-worker.mjs \\
    --version v0.1.0 \\
    --name "Add calm reset flow" \\
    --task "Implement only the concrete worker task described by Codex."

Options:
  --version       Required development version, for example v0.6.47.
  --name          Optional run label. Defaults to a slug from --task.
  --task          Worker assignment text. Required unless --prompt-file is used.
  --prompt-file   Optional file with a longer Codex-authored assignment.
  --session-id    Optional Pi session id. Defaults to a stable version/name id.
  --model         Optional Pi model. Defaults to zai/glm-5.2.
  --thinking      Optional Pi thinking level. Defaults to xhigh, which Z.ai maps to reasoning_effort=max.
  --pi            Optional pi binary path. Defaults to PI_BIN or pi.
  --pi-mode       Optional Pi mode: json or text. Defaults to json.
  --runs-dir      Optional base directory for run artifacts. Defaults to .codex/logs/agent-runs.
  --event-idle-timeout-ms
                  Optional watchdog timeout without worker output. Defaults to 600000 (10 minutes). Use 0 to disable.
  --dry-run       Write prompt/meta, print command, but do not launch Pi.
`;
}

function requireArg(args, name) {
  if (!args[name]) throw new Error(`Missing required --${name}\n\n${usage()}`);
  return args[name];
}

function slugify(value) {
  return String(value || 'worker')
    .toLowerCase()
    .replace(/[^a-z0-9]+/g, '-')
    .replace(/^-+|-+$/g, '')
    .slice(0, 80) || 'worker';
}

function timestamp() {
  return new Date().toISOString().replace(/[:.]/g, '-');
}

function readOptionalFile(path) {
  if (!path) return '';
  return readFileSync(resolve(path), 'utf8').trim();
}

function parseNonNegativeInt(value, fallback) {
  if (value === undefined) return fallback;
  const parsed = Number.parseInt(value, 10);
  if (!Number.isFinite(parsed) || parsed < 0) {
    throw new Error(`Expected a non-negative integer, got: ${value}`);
  }
  return parsed;
}

function getGitStatus(cwd) {
  try {
    return execFileSync('git', ['status', '--short'], {
      cwd,
      encoding: 'utf8',
      stdio: ['ignore', 'pipe', 'pipe'],
    }).trim();
  } catch (error) {
    return `GIT_STATUS_ERROR: ${error.message || error}`;
  }
}

function textFromContent(value) {
  if (!value) return '';
  if (typeof value === 'string') return value;
  if (Array.isArray(value)) return value.map(textFromContent).filter(Boolean).join('');
  if (typeof value !== 'object') return '';

  if (typeof value.text === 'string') return value.text;
  if (typeof value.delta === 'string') return value.delta;
  if (typeof value.content === 'string') return value.content;
  if (Array.isArray(value.content)) return textFromContent(value.content);
  if (value.type === 'text' && typeof value.value === 'string') return value.value;

  return '';
}

function textFromMessage(message) {
  if (!message || message.role !== 'assistant') return '';
  return textFromContent(message.content).trim();
}

function extractTextDelta(event) {
  const update = event?.assistantMessageEvent;
  if (!update || typeof update !== 'object') return '';
  if (typeof update.delta === 'string') return update.delta;
  if (typeof update.text === 'string') return update.text;
  return '';
}

function findLastAssistantTextFromMessages(messages) {
  if (!Array.isArray(messages)) return '';
  for (let i = messages.length - 1; i >= 0; i -= 1) {
    const text = textFromMessage(messages[i]);
    if (text) return text;
  }
  return '';
}

function findFinalStatus(text) {
  const matches = [...String(text || '').matchAll(/(?:^|\n)\s*WORKER_STATUS:\s*(WORKER_DONE|BLOCKED|NEEDS_DISCUSSION)\b/g)];
  return matches.at(-1)?.[1] || null;
}

function summarizeOrchestratorStatus({ finalStatus, exitCode, signal, killedByIdle, gitStatusAfter }) {
  if (killedByIdle) {
    return gitStatusAfter ? 'WORKER_STALLED_WITH_DIFF' : 'WORKER_STALLED_NO_DIFF';
  }
  if (signal) return `PROCESS_SIGNAL_${signal}`;
  if (exitCode !== null && exitCode !== undefined && exitCode !== 0) return 'PROCESS_FAILED';
  if (finalStatus === 'WORKER_DONE') return 'WORKER_DONE_PENDING_CODEX_GATE';
  if (finalStatus === 'BLOCKED') return 'WORKER_BLOCKED';
  if (finalStatus === 'NEEDS_DISCUSSION') return 'WORKER_NEEDS_DISCUSSION';
  if (exitCode === null || exitCode === undefined) return 'RUNNING_AWAITING_STATUS';
  return 'NO_STRUCTURED_STATUS';
}

function renderEventForConsole(event) {
  if (!event || typeof event !== 'object') return null;
  if (event.type === 'message_update') return extractTextDelta(event) || null;
  if (event.type === 'tool_execution_start') return `\n[pi] tool start: ${event.toolName || 'unknown'}\n`;
  if (event.type === 'tool_execution_end') {
    const status = event.isError ? 'error' : 'ok';
    return `\n[pi] tool end: ${event.toolName || 'unknown'} (${status})\n`;
  }
  if (
    event.type === 'agent_start' ||
    event.type === 'agent_end' ||
    event.type === 'turn_start' ||
    event.type === 'turn_end' ||
    event.type === 'compaction_start' ||
    event.type === 'compaction_end' ||
    event.type === 'auto_retry_start' ||
    event.type === 'auto_retry_end' ||
    event.type === 'extension_error'
  ) {
    return `\n[pi] ${event.type}\n`;
  }
  return null;
}

function assertRequiredFiles(cwd, version) {
  const required = [
    'AGENTS.md',
    'development/_reference/ai/autonomous-agent-harness.md',
  ];

  const missing = required.filter((file) => !existsSync(resolve(cwd, file)));
  if (missing.length > 0) {
    throw new Error(`Missing required orchestration files:\n${missing.map((file) => `- ${file}`).join('\n')}`);
  }
}

function existingContextFiles(cwd, version) {
  return [
    'README.md',
    `development/${version}/prd.md`,
    `development/${version}/plan.md`,
    `development/${version}/execution-log.md`,
    'development/_reference/ai/autonomous-agent-harness.md',
  ].filter((file) => existsSync(resolve(cwd, file)));
}

function buildPrompt({ contextFiles, name, taskText, version }) {
  const contextLine = contextFiles.length > 0
    ? `Read these context files when relevant: ${contextFiles.join(', ')}.`
    : 'No version planning files were found; use the direct Codex assignment as the scoped contract.';

  return `You are the Pi worker launched by Codex-Orchestrator for this project.

Codex remains the orchestrator, verifier, reviewer, and committer. You are only the bounded implementation worker.

Task label:
${name}

Worker assignment:
${taskText}

Repository contract:
- Read AGENTS.md before editing.
- ${contextLine}
- The direct Codex assignment is the binding scope. Do not broaden it into adjacent cleanup, dependent tasks, or a full release pass.
- If the assignment is too large to finish safely, implement a coherent safe subset and report the recommended next worker task.
- Do not edit planning or reference docs unless the assignment explicitly asks.
- Do not create or update execution logs unless the assignment explicitly asks. Durable progress is captured by Codex through reviewed git checkpoint commits.
- Do not commit, push, merge, reset, clean, or rewrite git history.
- Do not deploy, publish, or touch external services unless the assignment explicitly says so.
- Do not run destructive cleanup or wipe commands.
- Do not edit .pi/**, .agents/**, AGENTS.md, or development/_reference/ai/**.
- If unrelated uncommitted changes exist, work around them. Stop with NEEDS_DISCUSSION if they block the task.
- If credentials, external access, clean local stand setup, or human business choice is required, stop with BLOCKED instead of asking a question.
- The ask_question tool is disabled. Do not request interactive confirmation.

Implementation rules:
- Prefer existing repo patterns and narrow diffs.
- This is a SvelteKit + TypeScript app. Keep UI behavior calm, readable, and consistent with the existing design.
- Keep game data, game state, and UI components separated according to the current structure.
- Keep user-facing copy consistent with the app language and tone already present in the touched files.
- Prefer existing npm scripts from package.json. Do not invent scripts.
- For UI behavior changes, describe the local browser smoke path Codex should verify.

Worker output contract:
- Run only relevant local checks that exist in package.json / repo docs.
- If UI behavior changed and local browser smoke is required, use only a local dev server.
- Finish with one of: WORKER_DONE, BLOCKED, or NEEDS_DISCUSSION.
- End your final response with a standalone final status line in this exact shape:
  WORKER_STATUS: WORKER_DONE
  or WORKER_STATUS: BLOCKED
  or WORKER_STATUS: NEEDS_DISCUSSION
- List changed files and observed command exit codes.
- If there is meaningful follow-up work, propose the next concrete worker task.
- Do not claim review approval or human acceptance. Codex will verify, review, and commit if appropriate.
`;
}

async function main() {
  const args = parseArgs(process.argv.slice(2));
  const cwd = process.cwd();
  const version = requireArg(args, 'version');
  const model = args.model || 'zai/glm-5.2';
  const thinking = args.thinking || 'xhigh';
  const piBin = args.pi || process.env.PI_BIN || 'pi';
  const piMode = args['pi-mode'] || 'json';
  const eventIdleTimeoutMs = parseNonNegativeInt(args['event-idle-timeout-ms'], 600000);
  const taskText = [args.task || '', readOptionalFile(args['prompt-file'])]
    .filter(Boolean)
    .join('\n\nAdditional Codex assignment details:\n')
    .trim();

  if (!taskText) throw new Error(`Missing --task or --prompt-file\n\n${usage()}`);
  if (!['json', 'text'].includes(piMode)) {
    throw new Error(`Unsupported --pi-mode ${piMode}. Expected json or text.`);
  }

  assertRequiredFiles(cwd, version);
  const name = args.name || slugify(taskText);
  const contextFiles = existingContextFiles(cwd, version);
  const runsBaseDir = resolve(cwd, args['runs-dir'] || '.codex/logs/agent-runs');

  const runSlug = slugify(`${name}`);
  const runDir = resolve(runsBaseDir, version, `${timestamp()}-pi-worker-${runSlug}`);
  mkdirSync(runDir, { recursive: true });

  const prompt = buildPrompt({ contextFiles, name, taskText, version });
  const promptPath = join(runDir, 'prompt.txt');
  const stdoutPath = join(runDir, 'stdout.txt');
  const stderrPath = join(runDir, 'stderr.txt');
  const metaPath = join(runDir, 'meta.json');
  const eventsPath = join(runDir, 'events.jsonl');
  const summaryPath = join(runDir, 'summary.json');
  const lastMessagePath = join(runDir, 'last-message.txt');

  const sessionId = args['session-id'] || `calm-${version.replace(/[^a-zA-Z0-9]+/g, '-')}-${runSlug}`;
  const piArgs = [
    ...(piMode === 'json' ? ['--mode', 'json'] : ['-p']),
    '--model',
    model,
    '--thinking',
    thinking,
    '--approve',
    '--exclude-tools',
    'ask_question',
    '--session-id',
    sessionId,
    prompt,
  ];

  writeFileSync(promptPath, prompt);
  const startedAt = new Date().toISOString();
  const baseMeta = {
    tool: 'pi',
    piBin,
    args: piArgs.slice(0, -1).concat(['<prompt>']),
    cwd,
    version,
    name,
    model,
    thinking,
    piMode,
    eventIdleTimeoutMs,
    sessionId,
    runDir,
    promptPath,
    stdoutPath,
    stderrPath,
    eventsPath,
    summaryPath,
    lastMessagePath,
    startedAt,
    dryRun: Boolean(args['dry-run']),
  };

  writeFileSync(metaPath, `${JSON.stringify(baseMeta, null, 2)}\n`);
  writeFileSync(summaryPath, `${JSON.stringify({
    ...baseMeta,
    status: 'NOT_STARTED',
    finalStatus: null,
    orchestratorStatus: 'NOT_STARTED',
    eventCounts: {},
    eventCount: 0,
    parseErrors: [],
    gitStatusAfter: null,
  }, null, 2)}\n`);

  console.log(`Run directory: ${runDir}`);
  console.log(`Prompt: ${promptPath}`);
  console.log(`Events: ${eventsPath}`);
  console.log(`Summary: ${summaryPath}`);
  console.log(`Command: ${piBin} ${piArgs.slice(0, -1).join(' ')} <prompt>`);

  if (args['dry-run']) {
    console.log('Dry run: Pi was not launched.');
    return;
  }

  const stdout = createWriteStream(stdoutPath);
  const stderr = createWriteStream(stderrPath);
  const events = createWriteStream(eventsPath);
  let stdoutRemainder = '';
  let stdoutTail = '';
  let lastAssistantText = '';
  let lastEventAt = null;
  let lastEventType = null;
  let parentSignal = null;
  let killedByIdle = false;
  let idleTimer = null;
  const parseErrors = [];
  const eventCounts = {};
  let eventCount = 0;

  const child = spawn(piBin, piArgs, {
    cwd,
    env: process.env,
    stdio: ['ignore', 'pipe', 'pipe'],
  });

  function writeRunningSummary(extra = {}) {
    const gitStatusAfter = extra.gitStatusAfter ?? null;
    const finalStatus = findFinalStatus(lastAssistantText);
    const orchestratorStatus = summarizeOrchestratorStatus({
      finalStatus,
      exitCode: extra.exitCode ?? null,
      signal: extra.signal ?? null,
      killedByIdle,
      gitStatusAfter,
    });

    writeFileSync(summaryPath, `${JSON.stringify({
      ...baseMeta,
      status: extra.status || 'RUNNING',
      finalStatus,
      orchestratorStatus,
      eventCounts,
      eventCount,
      parseErrors,
      lastEventAt,
      lastEventType,
      parentSignal,
      killedByIdle,
      gitStatusAfter,
      stdoutTail,
      lastAssistantTextPath: lastMessagePath,
      ...extra,
    }, null, 2)}\n`);
  }

  function resetIdleTimer() {
    if (eventIdleTimeoutMs === 0) return;
    clearTimeout(idleTimer);
    idleTimer = setTimeout(() => {
      killedByIdle = true;
      console.error(`\nPi worker produced no output for ${eventIdleTimeoutMs}ms; sending SIGTERM.`);
      child.kill('SIGTERM');
      setTimeout(() => {
        if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL');
      }, 10000).unref();
    }, eventIdleTimeoutMs);
    idleTimer.unref();
  }

  function handleEvent(event) {
    eventCount += 1;
    eventCounts[event.type || 'unknown'] = (eventCounts[event.type || 'unknown'] || 0) + 1;
    lastEventAt = new Date().toISOString();
    lastEventType = event.type || 'unknown';
    events.write(`${JSON.stringify(event)}\n`);

    if (event.type === 'message_end') {
      const text = textFromMessage(event.message);
      if (text) lastAssistantText = text;
    } else if (event.type === 'agent_end') {
      const text = findLastAssistantTextFromMessages(event.messages);
      if (text) lastAssistantText = text;
    }

    if (lastAssistantText) writeFileSync(lastMessagePath, `${lastAssistantText}\n`);

    const rendered = renderEventForConsole(event);
    if (rendered) process.stdout.write(rendered);
  }

  function handleStdoutChunk(chunk) {
    const text = chunk.toString('utf8');
    stdoutTail = `${stdoutTail}${text}`.slice(-12000);
    stdout.write(chunk);
    resetIdleTimer();

    if (piMode !== 'json') {
      process.stdout.write(chunk);
      return;
    }

    stdoutRemainder += text;
    const lines = stdoutRemainder.split('\n');
    stdoutRemainder = lines.pop() || '';

    for (const line of lines) {
      const trimmed = line.trim();
      if (!trimmed) continue;
      try {
        handleEvent(JSON.parse(trimmed));
      } catch (error) {
        parseErrors.push({
          at: new Date().toISOString(),
          error: error.message || String(error),
          line: trimmed.slice(0, 500),
        });
        process.stdout.write(`${line}\n`);
      }
    }

    writeRunningSummary();
  }

  function handleParentSignal(signal) {
    parentSignal = signal;
    console.error(`\nReceived ${signal}; forwarding to Pi worker.`);
    child.kill(signal);
  }

  process.once('SIGINT', handleParentSignal);
  process.once('SIGTERM', handleParentSignal);

  child.on('error', (error) => {
    parseErrors.push({
      at: new Date().toISOString(),
      error: error.message || String(error),
      line: 'spawn',
    });
  });

  child.stdout.on('data', (chunk) => {
    handleStdoutChunk(chunk);
  });

  child.stderr.on('data', (chunk) => {
    resetIdleTimer();
    process.stderr.write(chunk);
    stderr.write(chunk);
  });

  resetIdleTimer();
  writeRunningSummary();

  const { exitCode, signal } = await new Promise((resolveExit) => {
    child.on('close', (code, closeSignal) => resolveExit({ exitCode: code, signal: closeSignal }));
  });
  clearTimeout(idleTimer);

  if (piMode === 'json' && stdoutRemainder.trim()) {
    try {
      handleEvent(JSON.parse(stdoutRemainder.trim()));
    } catch (error) {
      parseErrors.push({
        at: new Date().toISOString(),
        error: error.message || String(error),
        line: stdoutRemainder.trim().slice(0, 500),
      });
    }
  }

  stdout.end();
  stderr.end();
  events.end();

  const finishedAt = new Date().toISOString();
  const gitStatusAfter = getGitStatus(cwd);
  const finalStatus = piMode === 'json' ? findFinalStatus(lastAssistantText) : findFinalStatus(stdoutTail);
  const orchestratorStatus = summarizeOrchestratorStatus({
    finalStatus,
    exitCode,
    signal,
    killedByIdle,
    gitStatusAfter,
  });

  writeFileSync(
    metaPath,
    `${JSON.stringify({ ...baseMeta, finishedAt, exitCode, signal }, null, 2)}\n`,
  );
  writeFileSync(
    summaryPath,
    `${JSON.stringify({
      ...baseMeta,
      status: 'FINISHED',
      finalStatus,
      orchestratorStatus,
      eventCounts,
      eventCount,
      parseErrors,
      lastEventAt,
      lastEventType,
      parentSignal,
      killedByIdle,
      gitStatusAfter,
      stdoutTail,
      lastAssistantTextPath: lastMessagePath,
      finishedAt,
      exitCode,
      signal,
    }, null, 2)}\n`,
  );

  console.log(`\nPi worker finished with ${orchestratorStatus}. See ${runDir}`);

  if (parentSignal === 'SIGINT') process.exit(130);
  if (parentSignal === 'SIGTERM') process.exit(143);
  if (killedByIdle) process.exit(124);
  if (exitCode !== 0 || signal) {
    console.error(`Pi worker exited with code ${exitCode ?? 'null'} signal ${signal ?? 'null'}. See ${runDir}`);
    process.exit(exitCode || 1);
  }
  if (!finalStatus) {
    console.error(`Pi worker exited without WORKER_STATUS marker. See ${summaryPath}`);
    process.exit(3);
  }

  console.log(`Pi worker completed. Structured summary: ${summaryPath}`);
}

main().catch((error) => {
  console.error(error.message || error);
  process.exit(1);
});