arch: Robust ProcessIO wrapper for stream handling

[bbopen]

Overview

Multiple issues relate to stream/IO handling in NodeBridge - EPIPE errors, backpressure, and recovery after write failures. This tracking issue proposes a robust ProcessIO abstraction.

Related Issues

• NodeBridge should reset process after stdin write failures #107 - NodeBridge should reset process after stdin write failures
• Handle stdio stream errors to avoid unhandled EPIPE crashes #91 - Handle stdio stream errors to avoid unhandled EPIPE crashes
• Handle stdin backpressure in NodeBridge and OptimizedNodeBridge #59 - Handle stdin backpressure in NodeBridge and OptimizedNodeBridge

Proposed Architecture

ProcessIO Wrapper

class ProcessIO {
  private writeQueue: Array<{ data: string; resolve: () => void; reject: (e: Error) => void }> = [];
  private draining = false;
  
  constructor(
    private stdin: Writable,
    private onError: (error: Error) => void
  ) {
    stdin.on('error', (err) => {
      this.rejectPending(err);
      this.onError(err);
    });
    
    stdin.on('drain', () => {
      this.draining = false;
      this.flush();
    });
  }

  async write(data: string): Promise<void> {
    if (!this.stdin.writable) {
      throw new BridgeProtocolError('stdin not writable');
    }
    
    return new Promise((resolve, reject) => {
      this.writeQueue.push({ data, resolve, reject });
      this.flush();
    });
  }

  private flush(): void {
    if (this.draining) return;
    
    while (this.writeQueue.length > 0) {
      const item = this.writeQueue[0];
      const ok = this.stdin.write(item.data);
      
      if (!ok) {
        this.draining = true;
        return; // Wait for drain
      }
      
      this.writeQueue.shift();
      item.resolve();
    }
  }

  private rejectPending(error: Error): void {
    for (const item of this.writeQueue) {
      item.reject(error);
    }
    this.writeQueue = [];
  }

  destroy(): void {
    this.rejectPending(new BridgeDisposedError('ProcessIO destroyed'));
    this.stdin.destroy();
  }
}

Integration with NodeBridge

// In spawnProcess()
worker.io = new ProcessIO(
  childProcess.stdin,
  (error) => this.quarantineWorker(worker, error)
);

// In BridgeCore write callback
write: async (data: string) => {
  await worker.io.write(data);
}

Acceptance Criteria

• EPIPE and stream errors are caught and trigger quarantine
• Backpressure is handled via write queue
• Write failures reject pending writes and trigger recovery
• No unhandled stream errors crash the process
• All 3 related issues can be closed

Scope

This fix touches:

• New: src/runtime/process-io.ts - ProcessIO wrapper
• src/runtime/node.ts - integrate ProcessIO
• src/runtime/bridge-core.ts - async write support

[bbopen]

Superseded by #149

This issue is addressed by the unified BoundedContext abstraction proposed in #149.

The ProcessIO/stream handling concerns here will be implemented via:

• trackResource(resource) - registers resources for automatic cleanup
• untrackResource(resource) - removes from tracking
• Automatic disposal of all tracked resources when context disposes
• Isolated failure handling (errors in one resource don't affect others)

Streams and child processes will be wrapped as tracked resources.