feat(pipeline): add rate limit handling and improve process termination

Author: Marina-L-Stoyanova

src/config.ts

@@ -55,6 +55,13 @@ export const TEST_TIMEOUT_MS = Number(process.env.TEST_TIMEOUT_MS || 5 * 60 * 10
 export const MATRIX_CONFIG = process.env.MATRIX_CONFIG || '';
 export const MATRIX_MAX_ENTRIES = Number(process.env.MATRIX_MAX_ENTRIES || 24);
 export const AGENT_TIMEOUT_MS = Number(process.env.AGENT_TIMEOUT_MS || 25 * 60 * 1000);
+// Text that flags a provider rate limit in opencode's own log
+// (<dataDir>/opencode/log/opencode.log) during the headless agent step; matched
+// case-insensitively against each fresh line so the entry aborts early instead of
+// burning the rest of AGENT_TIMEOUT_MS waiting on a provider that's already refusing
+// requests. opencode/provider-wording dependent (like the label regexes in
+// src/capture/usage.ts) — override if your provider phrases it differently.
+export const RATE_LIMIT_PATTERN = process.env.RATE_LIMIT_PATTERN || 'rate limit exceeded';
 // How long to wait for the (headless) post-edit dev-server build before giving up
 // and screenshotting anyway. Generous because the first build of an agent-edited
 // app (esp. Blazor) is slow across the bind mount.

src/matrix/matrix.ts

@@ -4,7 +4,7 @@ import * as path from 'path';
 import type { ChildProcess } from 'child_process';
 import * as history from '../history.ts';
 import { WORK, ARTIFACT_DIR } from '../config.ts';
-import { killTree } from '../proc/exec.ts';
+import { terminateTree } from '../proc/exec.ts';
 import { killWatcher } from '../proc/watcher.ts';
 import { createSSE } from '../stream/sse.ts';
 import { cleanupAppDir } from './cleanup.ts';
@@ -136,7 +136,7 @@ async function runMatrix(combos: Combo[], { prompt, matrixId, fixed, name }: { p
     // instead of running the step (e.g. don't start the agent after Cancel).
     const onChild = (child: ChildProcess) => {
       currentChild = child;
-      if (matrixCancelled || cancelledEntries.has(runId)) killTree(child, 'SIGTERM');
+      if (matrixCancelled || cancelledEntries.has(runId)) terminateTree(child);
     };
     try {
       const result = await runPipeline(cfg, { emit, headless: true, prompt, dataDir, artifactDir, onChild, appDir });
@@ -244,7 +244,7 @@ export function begin(combos: Combo[], { prompt, fixed, name = null }: { prompt:
 export function cancel(): { ok: boolean; error?: string } {
   if (!matrixRunning) return { ok: false, error: 'no matrix run in progress' };
   matrixCancelled = true;
-  killTree(currentChild, 'SIGTERM');
+  terminateTree(currentChild);
   killWatcher('app'); killWatcher('opencode');
   broadcast({ type: 'log', msg: 'cancellation requested — stopping the current step' });
   return { ok: true };
@@ -263,7 +263,7 @@ export function cancelEntry(runId: string): { ok: boolean; error?: string } {
   }
   cancelledEntries.add(runId);
   if (entry.status === 'running') {
-    killTree(currentChild, 'SIGTERM');
+    terminateTree(currentChild);
     killWatcher('app'); killWatcher('opencode');
     broadcast({ type: 'log', index: entry.index, msg: 'entry cancellation requested — stopping this run' });
   }

src/pipeline/pipeline.ts

@@ -11,9 +11,9 @@ import { parseOpencodeStats } from '../capture/usage.ts';
 import { collectToolUsage, installedSkills, summarizeToolUsage } from '../capture/tool-usage.ts';
 import {
   APP_DIR, LOG_DIR, OPENCODE_PORT, AGENT_TIMEOUT_MS, APP_READY_TIMEOUT_MS, MCP_COMMAND_BY_CLASS,
-  LOCAL_SKILLS_DIR, OPENCODE_DATA_DIR,
+  LOCAL_SKILLS_DIR, OPENCODE_DATA_DIR, RATE_LIMIT_PATTERN,
 } from '../config.ts';
-import { run, capture, type RunOpts } from '../proc/exec.ts';
+import { run, capture, terminateTree, type RunOpts } from '../proc/exec.ts';
 import { spawnWatcher, killWatcher } from '../proc/watcher.ts';
 import { waitForPort, waitForPortFree, waitForAppReady } from '../proc/ports.ts';
 import { ensureDirs, sleep, rmrf } from '../proc/fsutil.ts';
@@ -46,6 +46,13 @@ export interface PipelineOpts {
 const activeMcpServers = (block: Record<string, any>): string[] =>
   Object.entries(block).filter(([, def]) => !def || def.enabled !== false).map(([name]) => name);
 
+// RATE_LIMIT_PATTERN is env-overridable (see config.ts) so a bad override can't crash
+// the pipeline — fall back to a literal match of the configured text.
+const RATE_LIMIT_RE = (() => {
+  try { return new RegExp(RATE_LIMIT_PATTERN, 'i'); }
+  catch (_) { return new RegExp(RATE_LIMIT_PATTERN.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'), 'i'); }
+})();
+
 // Stages 1–4b are identical for an interactive session and a headless matrix entry.
 // Stage 5+ branches: interactive launches `opencode web` (long-lived); headless runs
 // `opencode run "<prompt>"` once, parses usage, then screenshots every route.
@@ -66,8 +73,14 @@ export async function runPipeline(
 
   // Report every spawned child to `onChild` (matrix cancel kills whatever is current)
   // so Cancel works during scaffold/npm-install too, not only the agent step.
-  const runStep = (cmd: string, argv: string[], cwd: string, e: Emit, opts: RunOpts = {}) =>
-    run(cmd, argv, cwd, e, { ...opts, onChild });
+  const runStep = (cmd: string, argv: string[], cwd: string, e: Emit, opts: RunOpts = {}) => {
+    const localOnChild = opts.onChild;
+    const mergedOnChild = (child: ChildProcess) => {
+      if (onChild) onChild(child);
+      if (localOnChild) localOnChild(child);
+    };
+    return run(cmd, argv, cwd, e, { ...opts, onChild: mergedOnChild });
+  };
 
   ensureDirs();
   // Clean any previous attempt. In matrix mode `appDir` is unique per entry, so
@@ -404,9 +417,72 @@ export async function runPipeline(
     'run', ...(process.env.OPENCODE_RUN_ARGS || '').split(' ').filter(Boolean), prompt || '',
     ...promptImageFiles.flatMap((f) => ['--file', f]),
   ];
-  await runStep('opencode', agentArgv, appDir, emit, {
-    env: ocEnv, timeoutMs: AGENT_TIMEOUT_MS, heartbeatMs: 20000,
-  });
+  let rateLimited = false;
+  let rateLimitLine = '';
+  let agentChild: ChildProcess | null = null;
+  let signaled = false;
+  const opencodeLogPath = path.join(toolDataDir, 'opencode', 'log', 'opencode.log');
+  let opencodeLogOffset = 0;
+  try {
+    if (fs.existsSync(opencodeLogPath)) opencodeLogOffset = fs.statSync(opencodeLogPath).size;
+  } catch (_) {}
+  const readFreshOpencodeLog = () => {
+    try {
+      const stat = fs.statSync(opencodeLogPath);
+      // log rotate/truncate: restart from the new beginning
+      if (stat.size < opencodeLogOffset) opencodeLogOffset = 0;
+      const bytes = stat.size - opencodeLogOffset;
+      if (bytes <= 0) return '';
+      const fd = fs.openSync(opencodeLogPath, 'r');
+      try {
+        const buf = Buffer.allocUnsafe(bytes);
+        const read = fs.readSync(fd, buf, 0, bytes, opencodeLogOffset);
+        opencodeLogOffset += read;
+        return read > 0 ? buf.subarray(0, read).toString() : '';
+      } finally {
+        fs.closeSync(fd);
+      }
+    } catch (_) {
+      return '';
+    }
+  };
+  const rateLimitWatch = setInterval(() => {
+    if (rateLimited || signaled) return;
+    const fresh = readFreshOpencodeLog();
+    if (!fresh) return;
+    const lines = fresh.split('\n').map((l) => l.trim()).filter(Boolean);
+    for (const line of lines) {
+      if (!RATE_LIMIT_RE.test(line)) continue;
+      rateLimited = true;
+      rateLimitLine = line;
+      emit('log', 'rate limit detected in opencode log; aborting agent early');
+      if (agentChild) {
+        signaled = true;
+        terminateTree(agentChild);
+      }
+      break;
+    }
+  }, 1000);
+  rateLimitWatch.unref && rateLimitWatch.unref();
+
+  try {
+    await runStep('opencode', agentArgv, appDir, emit, {
+      env: ocEnv,
+      timeoutMs: AGENT_TIMEOUT_MS,
+      heartbeatMs: 20000,
+      onChild: (child) => { agentChild = child; },
+    });
+    if (rateLimited) {
+      throw new Error(`opencode rate-limited${rateLimitLine ? `: ${rateLimitLine}` : ''}`);
+    }
+  } catch (e: any) {
+    if (rateLimited) {
+      throw new Error(`opencode rate-limited${rateLimitLine ? `: ${rateLimitLine}` : ''}`);
+    }
+    throw e;
+  } finally {
+    clearInterval(rateLimitWatch);
+  }
 
   // Parse token/cost usage from `opencode stats` (against this entry's data dir).
   let entryStats: Stats | null = null;

src/proc/exec.ts

@@ -18,6 +18,17 @@ export function killTree(child: ChildProcess | null | undefined, sig: NodeJS.Sig
   catch (_) { try { child.kill(sig); } catch (_) {} }
 }
 
+// Gracefully terminate a child's whole process group, then force-kill if it
+// ignores/outlives SIGTERM (e.g. stuck mid-request).
+export function terminateTree(child: ChildProcess | null | undefined, graceMs = 5000): void {
+  if (!child) return;
+  killTree(child, 'SIGTERM');
+  const t = setTimeout(() => {
+    if (child.exitCode === null && child.signalCode === null) killTree(child, 'SIGKILL');
+  }, graceMs);
+  t.unref && t.unref();
+}
+
 // Run a command to completion, streaming its output through `emit`. Optional
 // `opts.env` is merged over process.env; `opts.timeoutMs` kills + rejects on hang;
 // `opts.heartbeatMs` emits a liveness tick so a long-but-working run (e.g. the agent)
@@ -41,7 +52,7 @@ export function run(cmd: string, argv: string[], cwd: string, emit: Emit, opts:
     if (opts.timeoutMs) {
       timer = setTimeout(() => {
         cleanup();
-        killTree(child, 'SIGTERM');
+        terminateTree(child);
         reject(new Error(`${cmd} timed out after ${opts.timeoutMs}ms`));
       }, opts.timeoutMs);
     }