fix(opencode): reuse task runtime for output recovery

Author: tomdps

src/agent-wrapper.js

@@ -476,8 +476,8 @@ class AgentWrapper {
    * Spawn claude-zeroshots process and stream output via message bus
    * @private
    */
-  _spawnClaudeTask(context) {
-    return spawnClaudeTask(this, context);
+  _spawnClaudeTask(context, options = {}) {
+    return spawnClaudeTask(this, context, options);
   }
 
   /**

src/agent/agent-task-executor.js

@@ -589,9 +589,10 @@ function ensureDangerousGitHook(targetClaudeDir = null) {
  * Spawn claude-zeroshots process and stream output via message bus
  * @param {Object} agent - Agent instance
  * @param {String} context - Context to pass to Claude
+ * @param {{skipStructuredResultCheck?: boolean}} [options] - Internal nested-task controls
  * @returns {Promise<Object>} Result object { success, output, error }
  */
-async function spawnClaudeTask(agent, context) {
+async function spawnClaudeTask(agent, context, options = {}) {
   const providerName = agent._resolveProvider ? agent._resolveProvider() : 'claude';
   const modelSpec = resolveAgentModelSpec(agent);
 
@@ -628,7 +629,7 @@ async function spawnClaudeTask(agent, context) {
 
   // MOCK SUPPORT: Use injected mock function if provided
   if (agent.mockSpawnFn) {
-    return agent.mockSpawnFn(args, { context });
+    return agent.mockSpawnFn(args, { context, options });
   }
 
   // SAFETY: Fail hard if testMode=true but no mock (should be caught in constructor)
@@ -641,7 +642,7 @@ async function spawnClaudeTask(agent, context) {
 
   // ISOLATION MODE: Run inside Docker container
   if (agent.isolation?.enabled) {
-    return spawnClaudeTaskIsolated(agent, context);
+    return spawnClaudeTaskIsolated(agent, context, options);
   }
 
   // NON-ISOLATION MODE: For Claude, use user's existing Claude config
@@ -703,7 +704,7 @@ async function spawnClaudeTask(agent, context) {
   }
 
   // Now follow the logs and stream output
-  return followClaudeTaskLogs(agent, taskId);
+  return followClaudeTaskLogs(agent, taskId, options);
 }
 
 function resolveAgentModelSpec(agent) {
@@ -1212,7 +1213,9 @@ function buildFailureContext({ agent, taskId, providerName, state, stdout }) {
 }
 
 async function buildCompletionResult({ agent, taskId, providerName, state, stdout, success }) {
-  const classified = await evaluateStructuredSuccess({ agent, taskId, state, success });
+  const classified = state.skipStructuredResultCheck
+    ? { success, error: null }
+    : await evaluateStructuredSuccess({ agent, taskId, state, success });
   let errorContext = classified.error;
   if (!errorContext && !classified.success) {
     errorContext = buildFailureContext({ agent, taskId, providerName, state, stdout });
@@ -1401,9 +1404,17 @@ function buildKillHandler({ agent, taskId, state, providerName, resolve }) {
   };
 }
 
-function createLogFollower({ agent, taskId, fsModule, ctPath, providerName }) {
+function createLogFollower({
+  agent,
+  taskId,
+  fsModule,
+  ctPath,
+  providerName,
+  skipStructuredResultCheck = false,
+}) {
   return new Promise((resolve) => {
     const state = createLogFollowState();
+    state.skipStructuredResultCheck = skipStructuredResultCheck;
 
     state.logFilePath = lookupLogFilePath(ctPath, taskId);
     if (state.logFilePath) {
@@ -1463,12 +1474,19 @@ function createLogFollower({ agent, taskId, fsModule, ctPath, providerName }) {
  * @param {String} taskId - Task ID to follow
  * @returns {Promise<Object>} Result object { success, output, error }
  */
-function followClaudeTaskLogs(agent, taskId) {
+function followClaudeTaskLogs(agent, taskId, options = {}) {
   const fsModule = require('fs');
   const ctPath = getClaudeTasksPath();
   const providerName = agent._resolveProvider ? agent._resolveProvider() : 'claude';
 
-  return createLogFollower({ agent, taskId, fsModule, ctPath, providerName });
+  return createLogFollower({
+    agent,
+    taskId,
+    fsModule,
+    ctPath,
+    providerName,
+    skipStructuredResultCheck: options.skipStructuredResultCheck === true,
+  });
 }
 
 // Cache zeroshot path at module load time (when PATH is correct)
@@ -1506,9 +1524,10 @@ function getClaudeTasksPath() {
  * Runs Claude CLI inside the container for full isolation
  * @param {Object} agent - Agent instance
  * @param {String} context - Context to pass to Claude
+ * @param {{skipStructuredResultCheck?: boolean}} [options] - Internal nested-task controls
  * @returns {Promise<Object>} Result object { success, output, error }
  */
-async function spawnClaudeTaskIsolated(agent, context) {
+async function spawnClaudeTaskIsolated(agent, context, options = {}) {
   const { manager, clusterId } = agent.isolation;
   const providerName = agent._resolveProvider ? agent._resolveProvider() : 'claude';
   const modelSpec = resolveAgentModelSpec(agent);
@@ -1626,7 +1645,7 @@ async function spawnClaudeTaskIsolated(agent, context) {
 
   // STEP 2: Install the lifecycle-owned handle before liveness monitoring can
   // observe the task, then follow the task's log file inside the container.
-  const execution = followClaudeTaskLogsIsolated(agent, taskId);
+  const execution = followClaudeTaskLogsIsolated(agent, taskId, options);
   if (agent.enableLivenessCheck) {
     agent.taskStartedAt = Date.now();
     agent.lastOutputTime = agent.taskStartedAt;
@@ -1658,7 +1677,7 @@ async function spawnClaudeTaskIsolated(agent, context) {
  * - Status checks reduced to every 2 seconds (not every poll)
  * - Result: 10-20% overall latency reduction
  */
-function createIsolatedLogState() {
+function createIsolatedLogState(skipStructuredResultCheck = false) {
   return {
     taskExited: false,
     resolved: false,
@@ -1670,6 +1689,7 @@ function createIsolatedLogState() {
     statusCheckInterval: null,
     timeoutTimer: null,
     lineBuffer: '',
+    skipStructuredResultCheck,
   };
 }
 
@@ -1816,7 +1836,9 @@ function settleIsolatedTerminalStatus({
           },
         })
       : null;
-    const parsedResult = await agent._parseResultOutput(state.fullOutput);
+    const parsedResult = state.skipStructuredResultCheck
+      ? null
+      : await agent._parseResultOutput(state.fullOutput);
 
     settleIsolatedFollower({
       agent,
@@ -2056,7 +2078,7 @@ function startIsolatedStatusChecks({
   }, 2000);
 }
 
-function followClaudeTaskLogsIsolated(agent, taskId) {
+function followClaudeTaskLogsIsolated(agent, taskId, options = {}) {
   const { isolation } = agent;
   if (!isolation?.manager) {
     throw new Error('followClaudeTaskLogsIsolated: isolation manager not found');
@@ -2067,7 +2089,7 @@ function followClaudeTaskLogsIsolated(agent, taskId) {
   const providerName = agent._resolveProvider ? agent._resolveProvider() : 'claude';
 
   return new Promise((resolve, reject) => {
-    const state = createIsolatedLogState();
+    const state = createIsolatedLogState(options.skipStructuredResultCheck === true);
     const cleanup = buildIsolatedCleanup(state);
     const onLine = (line) => broadcastIsolatedLine({ agent, providerName, taskId, line });
     state.lifecycleHandle = buildIsolatedLifecycleHandle({
@@ -2203,6 +2225,8 @@ async function parseResultOutput(agent, output) {
         schema: agent.config.jsonSchema,
         providerName,
         isCancelled: () => agent.running === false || agent.state === 'stopped',
+        runReformat: (prompt) =>
+          agent._spawnClaudeTask(prompt, { skipStructuredResultCheck: true }),
         onAttempt: (attempt, lastError) => {
           if (lastError) {
             console.warn(`[Agent ${agent.id}] Reformat attempt ${attempt}: ${lastError}`);
@@ -2214,6 +2238,7 @@ async function parseResultOutput(agent, output) {
         },
       });
     } catch (reformatError) {
+      if (reformatError.code === 'REFORMAT_CANCELLED') throw reformatError;
       // Reformatting failed - fall through to error below
       console.error(`[Agent ${agent.id}] Reformatting failed: ${reformatError.message}`);
     }

src/agent/output-reformatter.js

@@ -10,96 +10,13 @@
 
 const DEFAULT_MAX_ATTEMPTS = 3;
 
-const childProcess = require('child_process');
-
-// Provider used to reformat non-JSON agent output into the target schema.
-// opencode is used as the reformatting backend because it reliably emits clean
-// JSON via `--format json`. This makes the reformat fallback work for agents
-// whose own output isn't directly parseable (e.g. an opencode planner that
-// spends its turn on tool-calls instead of emitting the final JSON block).
-const REFORMAT_PROVIDER_BIN = 'opencode';
-const REFORMAT_TIMEOUT_MS = 180000;
 
 function createCancellationError() {
   const error = new Error('Output reformatting cancelled');
   error.code = 'REFORMAT_CANCELLED';
   return error;
 }
 
-/**
- * Call the opencode CLI with a prompt.
- * Stdin is ignored because opencode otherwise waits for EOF.
- *
- * @param {string} prompt
- * @param {{isCancelled?: () => boolean}} [options]
- * @returns {Promise<string|null>}
- */
-function callReformatModel(prompt, { isCancelled = () => false } = {}) {
-  if (isCancelled()) return Promise.reject(createCancellationError());
-
-  return new Promise((resolve, reject) => {
-    const child = childProcess.spawn(
-      REFORMAT_PROVIDER_BIN,
-      ['run', '--format', 'json', prompt],
-      {
-        stdio: ['ignore', 'pipe', 'pipe'],
-      }
-    );
-    let stdout = '';
-    let stderr = '';
-    let settled = false;
-    let timeoutTimer;
-    let cancellationTimer;
-
-    const finish = (error, value = null) => {
-      if (settled) return;
-      settled = true;
-      clearTimeout(timeoutTimer);
-      clearInterval(cancellationTimer);
-      if (error) {
-        reject(error);
-      } else {
-        resolve(value);
-      }
-    };
-
-    timeoutTimer = setTimeout(() => {
-      child.kill('SIGKILL');
-      finish(new Error(`opencode reformat timed out after ${REFORMAT_TIMEOUT_MS}ms`));
-    }, REFORMAT_TIMEOUT_MS);
-
-    cancellationTimer = setInterval(() => {
-      if (!isCancelled()) return;
-      child.kill('SIGKILL');
-      finish(createCancellationError());
-    }, 50);
-
-    child.stdout.on('data', (data) => {
-      stdout += data.toString();
-    });
-    child.stderr.on('data', (data) => {
-      stderr += data.toString();
-    });
-    child.on('close', (code, signal) => {
-      if (isCancelled()) {
-        finish(createCancellationError());
-      } else if (code !== 0) {
-        const detail = stderr.trim().slice(-500);
-        finish(
-          new Error(
-            `opencode reformat exited with code ${code}${signal ? ` (${signal})` : ''}` +
-              (detail ? `: ${detail}` : '')
-          )
-        );
-      } else {
-        finish(null, stdout || null);
-      }
-    });
-    child.on('error', (error) => {
-      finish(error);
-    });
-  });
-}
 
 /**
  * Build the reformatting prompt
@@ -157,6 +74,7 @@ Fix this issue in your response.`;
  * @param {number} [options.maxAttempts=3] - Maximum reformatting attempts
  * @param {Function} [options.onAttempt] - Callback for each attempt (attempt, error)
  * @param {Function} [options.isCancelled] - Returns true after agent cancellation
+ * @param {Function} options.runReformat - Runs the prompt in the active agent execution context
  * @returns {Promise<Object>} The reformatted JSON object
  * @throws {Error} If the provider is unsupported, cancellation occurs, or attempts fail
  */
@@ -167,13 +85,18 @@ async function reformatOutput({
   maxAttempts = DEFAULT_MAX_ATTEMPTS,
   onAttempt,
   isCancelled = () => false,
+  runReformat,
 }) {
   if (providerName !== 'opencode') {
     throw new Error(
       `Output reformatting not available for provider "${providerName}". ` +
         `Agent output must be valid JSON. Raw output (last 200 chars): ${(rawOutput || '').slice(-200)}`
     );
   }
+  if (typeof runReformat !== 'function') {
+    throw new Error('Output reformatting requires the active agent execution context');
+  }
+
 
   const { extractJsonFromOutput } = require('./output-extraction');
   let lastError = null;
@@ -187,9 +110,15 @@ async function reformatOutput({
     const prompt = buildReformatPrompt(rawOutput, schema, lastError);
 
     try {
-      const output = await callReformatModel(prompt, { isCancelled });
+      const result = await runReformat(prompt);
+      if (isCancelled()) throw createCancellationError();
+      if (!result?.success) {
+        lastError = result?.error || 'reformat task failed';
+        continue;
+      }
+      const output = result.output;
       if (!output) {
-        lastError = 'reformat model returned no output';
+        lastError = 'reformat task returned no output';
         continue;
       }