fix: keep streamed output delivery non-blocking (#186)

src/agent-output.ts

@@ -53,10 +53,3 @@ export function hasAgentOutputPayload(output: {
   }
   return output.result !== null && output.result !== undefined;
 }
-
-export function isSilentAgentOutput(_output: {
-  output?: StructuredAgentOutput;
-  result?: string | object | null;
-}): boolean {
-  return false;
-}

src/agent-runner-process.test.ts

@@ -0,0 +1,99 @@
+import type { ChildProcess } from 'child_process';
+import { EventEmitter } from 'events';
+import fs from 'fs';
+import os from 'os';
+import path from 'path';
+import { PassThrough } from 'stream';
+
+import { afterEach, describe, expect, it, vi } from 'vitest';
+
+import { OUTPUT_END_MARKER, OUTPUT_START_MARKER } from './agent-protocol.js';
+import { runSpawnedAgentProcess } from './agent-runner-process.js';
+import type { AgentInput, AgentOutput } from './agent-runner.js';
+import type { RegisteredGroup } from './types.js';
+
+function buildProcess(): ChildProcess {
+  const proc = new EventEmitter() as ChildProcess & {
+    stdout: PassThrough;
+    stderr: PassThrough;
+    kill: ReturnType<typeof vi.fn>;
+    killed: boolean;
+  };
+  proc.stdout = new PassThrough();
+  proc.stderr = new PassThrough();
+  proc.killed = false;
+  proc.kill = vi.fn(() => {
+    proc.killed = true;
+    return true;
+  });
+  return proc;
+}
+
+function timed<T>(promise: Promise<T>): Promise<T> {
+  return Promise.race([
+    promise,
+    new Promise<never>((_, reject) =>
+      setTimeout(() => reject(new Error('agent process did not resolve')), 100),
+    ),
+  ]);
+}
+
+const group: RegisteredGroup = {
+  name: 'Agent Runner Test',
+  folder: 'agent-runner-test',
+  trigger: '@Agent',
+  added_at: '2026-01-01T00:00:00.000Z',
+  agentType: 'codex',
+};
+
+const input: AgentInput = {
+  prompt: 'run',
+  groupFolder: 'agent-runner-test',
+  chatJid: 'dc:test',
+  runId: 'run-1',
+  isMain: false,
+};
+
+describe('runSpawnedAgentProcess', () => {
+  let logsDir: string | null = null;
+
+  afterEach(() => {
+    if (logsDir) {
+      fs.rmSync(logsDir, { recursive: true, force: true });
+      logsDir = null;
+    }
+  });
+
+  it('resolves even when streamed output delivery rejects', async () => {
+    logsDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ejclaw-agent-runner-'));
+    const proc = buildProcess();
+    const onOutput = vi.fn<(_: AgentOutput) => Promise<void>>(async () => {
+      throw new Error('delivery failed');
+    });
+
+    const resultPromise = runSpawnedAgentProcess({
+      proc,
+      group,
+      input,
+      processName: 'test-agent',
+      logsDir,
+      startTime: Date.now(),
+      onOutput,
+    });
+
+    (proc.stdout as PassThrough).write(
+      [
+        OUTPUT_START_MARKER,
+        JSON.stringify({ status: 'success', result: 'hello' }),
+        OUTPUT_END_MARKER,
+      ].join('\n'),
+    );
+    proc.emit('close', 0, null);
+
+    await expect(timed(resultPromise)).resolves.toMatchObject({
+      status: 'success',
+      result: null,
+    });
+    expect(onOutput).toHaveBeenCalledOnce();
+  });
+});

src/agent-runner-process.ts

@@ -39,6 +39,50 @@ function parseLegacyAgentOutput(stdout: string): AgentOutput {
   return JSON.parse(jsonLine) as AgentOutput;
 }
 
+function logStreamedOutputDeliveryError(
+  err: unknown,
+  group: RegisteredGroup,
+  input: AgentInput,
+): void {
+  logger.warn(
+    {
+      group: group.name,
+      chatJid: input.chatJid,
+      runId: input.runId,
+      error: getErrorMessage(err),
+    },
+    'Streamed agent output delivery failed',
+  );
+}
+
+function writeTimeoutLog(args: {
+  logsDir: string;
+  input: AgentInput;
+  group: RegisteredGroup;
+  processName: string;
+  duration: number;
+  code: number | null;
+  signal: NodeJS.Signals | null;
+  hadStreamingOutput: boolean;
+}): void {
+  const ts = new Date().toISOString().replace(/[:.]/g, '-');
+  fs.writeFileSync(
+    path.join(args.logsDir, `agent-${args.input.runId || 'adhoc'}-${ts}.log`),
+    [
+      `=== Agent Run Log (TIMEOUT) ===`,
+      `Timestamp: ${new Date().toISOString()}`,
+      `Group: ${args.group.name}`,
+      `ChatJid: ${args.input.chatJid}`,
+      `RunId: ${args.input.runId || 'n/a'}`,
+      `Process: ${args.processName}`,
+      `Duration: ${args.duration}ms`,
+      `Exit Code: ${args.code}`,
+      `Signal: ${args.signal}`,
+      `Had Streaming Output: ${args.hadStreamingOutput}`,
+    ].join('\n'),
+  );
+}
+
 export function runSpawnedAgentProcess(
   args: RunSpawnedAgentProcessArgs,
 ): Promise<AgentOutput> {
@@ -151,7 +195,9 @@ export function runSpawnedAgentProcess(
               'Streamed agent error output',
             );
           }
-          outputChain = outputChain.then(() => onOutput(parsed));
+          outputChain = outputChain
+            .then(() => onOutput(parsed))
+            .catch((err) => logStreamedOutputDeliveryError(err, group, input));
         } catch (err) {
           logger.warn(
             {
@@ -210,22 +256,16 @@ export function runSpawnedAgentProcess(
       const duration = Date.now() - startTime;
 
       if (timedOut) {
-        const ts = new Date().toISOString().replace(/[:.]/g, '-');
-        fs.writeFileSync(
-          path.join(logsDir, `agent-${input.runId || 'adhoc'}-${ts}.log`),
-          [
-            `=== Agent Run Log (TIMEOUT) ===`,
-            `Timestamp: ${new Date().toISOString()}`,
-            `Group: ${group.name}`,
-            `ChatJid: ${input.chatJid}`,
-            `RunId: ${input.runId || 'n/a'}`,
-            `Process: ${processName}`,
-            `Duration: ${duration}ms`,
-            `Exit Code: ${code}`,
-            `Signal: ${signal}`,
-            `Had Streaming Output: ${hadStreamingOutput}`,
-          ].join('\n'),
-        );
+        writeTimeoutLog({
+          logsDir,
+          input,
+          group,
+          processName,
+          duration,
+          code,
+          signal,
+          hadStreamingOutput,
+        });
 
         if (hadStreamingOutput) {
           logger.info(

src/streamed-output-evaluator.ts

@@ -1,8 +1,4 @@
-import {
-  getAgentOutputText,
-  hasAgentOutputPayload,
-  isSilentAgentOutput,
-} from './agent-output.js';
+import { getAgentOutputText, hasAgentOutputPayload } from './agent-output.js';
 import {
   classifyClaudeAuthError,
   classifyRotationTrigger,
@@ -62,7 +58,6 @@ export function evaluateStreamedOutput(
   const countsAsFinalOutput =
     output.phase === undefined || output.phase === 'final';
   const outputText = getAgentOutputText(output);
-  const silentOutput = isSilentAgentOutput(output);
 
   if (
     isPrimaryClaude &&
@@ -139,8 +134,7 @@ export function evaluateStreamedOutput(
     options.trackSuccessNullResult &&
     isPrimaryClaude &&
     output.status === 'success' &&
-    !state.sawOutput &&
-    !silentOutput
+    !state.sawOutput
   ) {
     nextState.sawSuccessNullResultWithoutOutput = true;
   }