AirCoding/packages/runtime/src/workers/WorkerProcess.ts

/**
 * WorkerProcess - Owns NDJSON pipe for a Bun child process
 *
 * Implements DD §8.1. stdout=protocol, stderr=fatal/log; exit-code table 0–5.
 *
 * @module packages/runtime/src/workers/WorkerProcess
 */

import type { ChildProcess } from 'child_process'
import { WorkerProtocol, type WorkerMessage, type WorkerMessageType, type WorkerMessageDirection } from './WorkerProtocol.js'

export type WorkerExitCode =
  | 0   // Normal exit
  | 1   // Error (unrecoverable)
  | 2   // Protocol error
  | 3   // Permission denied
  | 4   // Parent cancelled
  | 5   // Timeout

interface ExitCodeInfo {
  semantic: string
  description: string
}

const EXIT_CODE_TABLE: Record<WorkerExitCode, ExitCodeInfo> = {
  0: { semantic: 'normal', description: 'Worker completed successfully' },
  1: { semantic: 'error', description: 'Unrecoverable error occurred' },
  2: { semantic: 'protocol_error', description: 'Protocol violation or deserialization failure' },
  3: { semantic: 'permission_denied', description: 'Worker denied permission for operation' },
  4: { semantic: 'parent_cancelled', description: 'Parent process cancelled this worker' },
  5: { semantic: 'timeout', description: 'Worker exceeded time limit' }
}

export class WorkerProcess {
  private proc: ChildProcess | null = null
  private protocol: WorkerProtocol
  private message_handlers: Map<string, (msg: WorkerMessage) => void> = new Map()
  private buffer: string = ''

  constructor() {
    this.protocol = new WorkerProtocol()
  }

  /**
   * Set the child process.
   */
  set_process(proc: ChildProcess): void {
    this.proc = proc
    this.setup_streams()
  }

  /**
   * Send a message to the worker process.
   */
  send(message: WorkerMessage): void {
    if (!this.proc?.stdin?.writable) {
      throw new Error('Worker process stdin is not writable')
    }

    const line = this.protocol.encode(message)
    this.proc.stdin.write(line)
  }

  /**
   * Register a message handler.
   */
  on_message(type: WorkerMessageType, handler: (msg: WorkerMessage) => void): void {
    this.message_handlers.set(type, handler)
  }

  /**
   * Get exit code info.
   */
  get_exit_code_info(code: number): ExitCodeInfo | undefined {
    return EXIT_CODE_TABLE[code as WorkerExitCode]
  }

  /**
   * Check if process is alive.
   */
  is_alive(): boolean {
    if (!this.proc) return false
    return this.proc.exitCode === null
  }

  /**
   * Kill the worker process.
   */
  kill(signal: NodeJS.Signals = 'SIGTERM'): boolean {
    return this.proc?.kill(signal) || false
  }

  /**
   * Get the process ID.
   */
  get_pid(): number | undefined {
    return this.proc?.pid
  }

  // ============================================================================
  // Private
  // ============================================================================

  private setup_streams(): void {
    if (!this.proc) return

    // stdout = protocol channel
    if (this.proc.stdout) {
      this.proc.stdout.on('data', (data: Buffer) => {
        this.buffer += data.toString()
        this.process_buffer()
      })
    }

    // stderr = log/fatal
    if (this.proc.stderr) {
      this.proc.stderr.on('data', (data: Buffer) => {
        const message = data.toString().trim()
        if (message) {
          console.error('[Worker stderr]', message)
        }
      })
    }

    // Exit handler
    this.proc.on('exit', (code, signal) => {
      const info = this.get_exit_code_info(code || 1)
      console.log(`[Worker] exited with code ${code} (${info?.semantic || 'unknown'}): ${info?.description || ''}`)
    })
  }

  private process_buffer(): void {
    const lines = this.buffer.split('\n')
    this.buffer = lines.pop() || ''

    for (const line of lines) {
      const message = this.protocol.decode(line)
      if (!message) continue

      // Route to handler
      const handler = this.message_handlers.get(message.type)
      if (handler) {
        handler(message)
      }
    }
  }
}