spb/khaelor Public

1/**2 * KHAELOR3 * File: src/workspace/processes.ts4 * Description: ProcessManager — background processes with ring buffers, spill logs, and group kill (TOOL_PROTOCOL §8).5 *6 * Author: Simon-Pierre Boucher7 * Contact: contact@spboucher.ai8 */910import { spawn } from "node:child_process";11import type { ChildProcess } from "node:child_process";12import { appendFileSync, mkdirSync, readFileSync } from "node:fs";13import * as path from "node:path";14import { WorkspaceError } from "./workspace.js";1516export type ManagedProcessStatus = "running" | "exited" | "stopped" | "failed";1718/** A background process managed for the session (TOOL_PROTOCOL §8.2). */19export interface ManagedProcess {20  /** "p1", "p2" … session-scoped handle — NOT the OS pid. */21  id: string;22  /** OS pid, with start time recorded (PID-reuse guard, Hermes §6.2). */23  pid: number;24  command: string;25  status: ManagedProcessStatus;26  exitCode: number | null;27  startedAt: number;28  cwd: string;29  /** Full output spill — never truncated. */30  logPath: string;31}3233/** One page of process output. */34export interface ProcessRead {35  id: string;36  lines: string[];37  /** 1-based absolute line number of lines[0]; 0 when lines is empty. */38  startLine: number;39  totalLines: number;40  /** True when the page cap cut this read short — continue with offset, or read logPath. */41  truncated: boolean;42  status: ManagedProcessStatus;43  exitCode: number | null;44  logPath: string;45}4647/** An already-spawned child handed over by the `bash` timeout redirect (TOOL_PROTOCOL §7.2). */48export interface SpawnedChild {49  child: ChildProcess;50  command: string;51  cwd: string;52  startedAt: number;53  /** Output captured before adoption — replayed into the ring + log. */54  outputSoFar?: string;55}5657/**58 * Injected lifecycle listener. The kernel wires this to the event bus59 * (ProcessStarted / ProcessExited durable events) — the workspace never60 * imports `session`; the dependency is inverted here.61 */62export interface ProcessEventSink {63  onProcessStarted?(process: ManagedProcess): void;64  onProcessExited?(process: ManagedProcess): void;65}6667/** Model-facing background process manager (TOOL_PROTOCOL §8.2). */68export interface ProcessManager {69  start(command: string, cwd: string): Promise<ManagedProcess>;70  list(): ManagedProcess[];71  read(id: string, opts?: { offset?: number }): ProcessRead;72  write(id: string, input: string): Promise<void>;73  /** SIGTERM → grace → SIGKILL, whole process group. */74  stop(id: string): Promise<{ exitCode: number | null }>;75  /** bash timeout redirect (§7.2). */76  adopt(spawned: SpawnedChild): ManagedProcess;77  /** Session end only — NOT on interrupt (ADR-11). */78  stopAll(): Promise<void>;79}8081/** Per-process in-memory ring: 10,000 lines / 2 MB (overflow loses nothing — the log has it). */82const DEFAULT_MAX_RING_LINES = 10_000;83const DEFAULT_MAX_RING_BYTES = 2 * 1024 * 1024;84/** Read page cap: 300 lines / 20 KB (TOOL_PROTOCOL §8.2). */85const DEFAULT_READ_PAGE_LINES = 300;86const DEFAULT_READ_PAGE_BYTES = 20 * 1024;87/** SIGTERM → grace → SIGKILL. */88const DEFAULT_GRACE_MS = 3000;8990export interface LocalProcessManagerOptions {91  /** Directory for spill logs, e.g. ~/.khaelor/process-logs/<session>/. */92  logDir: string;93  eventSink?: ProcessEventSink;94  /** Shell used for `start`. Default: $SHELL, falling back to /bin/sh. */95  shell?: string;96  graceMs?: number;97  maxRingLines?: number;98  maxRingBytes?: number;99  readPageLines?: number;100  readPageBytes?: number;101}102103interface ProcRecord {104  info: ManagedProcess;105  child: ChildProcess;106  /** Complete output lines currently held in memory. */107  ring: string[];108  ringBytes: number;109  /** Total complete lines ever produced (ring + evicted-to-log). */110  totalLines: number;111  /** Trailing output not yet terminated by a newline. */112  partial: string;113  /** Absolute line number last returned by a cursor read (0 initially). */114  cursor: number;115  exited: boolean;116  failed: boolean;117  stopRequested: boolean;118  exitWaiters: Array<() => void>;119}120121export class LocalProcessManager implements ProcessManager {122  private readonly logDir: string;123  private readonly sink: ProcessEventSink | undefined;124  private readonly shell: string;125  private readonly graceMs: number;126  private readonly maxRingLines: number;127  private readonly maxRingBytes: number;128  private readonly readPageLines: number;129  private readonly readPageBytes: number;130  private readonly procs = new Map<string, ProcRecord>();131  private counter = 0;132133  constructor(options: LocalProcessManagerOptions) {134    this.logDir = path.resolve(options.logDir);135    this.sink = options.eventSink;136    this.shell = options.shell ?? process.env["SHELL"] ?? "/bin/sh";137    this.graceMs = options.graceMs ?? DEFAULT_GRACE_MS;138    this.maxRingLines = options.maxRingLines ?? DEFAULT_MAX_RING_LINES;139    this.maxRingBytes = options.maxRingBytes ?? DEFAULT_MAX_RING_BYTES;140    this.readPageLines = options.readPageLines ?? DEFAULT_READ_PAGE_LINES;141    this.readPageBytes = options.readPageBytes ?? DEFAULT_READ_PAGE_BYTES;142    mkdirSync(this.logDir, { recursive: true });143  }144145  async start(command: string, cwd: string): Promise<ManagedProcess> {146    const child = spawn(this.shell, ["-c", command], {147      cwd,148      env: process.env,149      detached: true,150      stdio: ["pipe", "pipe", "pipe"],151    });152    return this.register(child, command, cwd, Date.now(), undefined);153  }154155  adopt(spawned: SpawnedChild): ManagedProcess {156    return this.register(157      spawned.child,158      spawned.command,159      spawned.cwd,160      spawned.startedAt,161      spawned.outputSoFar,162    );163  }164165  list(): ManagedProcess[] {166    return [...this.procs.values()].map((rec) => ({ ...rec.info }));167  }168169  read(id: string, opts?: { offset?: number }): ProcessRead {170    const rec = this.require(id);171    const cursorRead = opts?.offset === undefined;172    const start = opts?.offset ?? rec.cursor + 1;173    const total = rec.totalLines;174175    if (start < 1 || start > total) {176      return {177        id,178        lines: [],179        startLine: 0,180        totalLines: total,181        truncated: false,182        status: rec.info.status,183        exitCode: rec.info.exitCode,184        logPath: rec.info.logPath,185      };186    }187188    const ringFirst = total - rec.ring.length + 1;189    let lines: string[];190    if (start >= ringFirst) {191      lines = rec.ring.slice(start - ringFirst);192    } else {193      // Evicted from the ring — nothing is lost; re-read from the spill log.194      const raw = readFileSync(rec.info.logPath, "utf8");195      const all = raw.split("\n");196      if (all.length > 0 && all[all.length - 1] === "") all.pop();197      lines = all.slice(start - 1, total);198    }199200    // Page cap: N lines / M bytes with continuation via offset or logPath.201    let truncated = false;202    if (lines.length > this.readPageLines) {203      lines = lines.slice(0, this.readPageLines);204      truncated = true;205    }206    let bytes = 0;207    for (let i = 0; i < lines.length; i++) {208      bytes += (lines[i] as string).length + 1;209      if (bytes > this.readPageBytes && i > 0) {210        lines = lines.slice(0, i);211        truncated = true;212        break;213      }214    }215216    if (cursorRead) {217      rec.cursor = start - 1 + lines.length;218    }219220    return {221      id,222      lines,223      startLine: lines.length > 0 ? start : 0,224      totalLines: total,225      truncated,226      status: rec.info.status,227      exitCode: rec.info.exitCode,228      logPath: rec.info.logPath,229    };230  }231232  async write(id: string, input: string): Promise<void> {233    const rec = this.require(id);234    if (rec.info.status !== "running") {235      throw new WorkspaceError(236        "process-not-running",237        `Process ${id} is not running (status ${rec.info.status}, exit code ${String(rec.info.exitCode)}). Full log: ${rec.info.logPath}`,238        { id, status: rec.info.status },239      );240    }241    const stdin = rec.child.stdin;242    if (stdin === null || !stdin.writable) {243      throw new WorkspaceError("process-not-running", `Process ${id} has no writable stdin.`, { id });244    }245    await new Promise<void>((resolvePromise, rejectPromise) => {246      stdin.write(input, (cause) => {247        if (cause) {248          rejectPromise(249            new WorkspaceError("process-not-running", `Failed to write to ${id} stdin: ${cause.message}`, {250              id,251            }),252          );253        } else {254          resolvePromise();255        }256      });257    });258  }259260  async stop(id: string): Promise<{ exitCode: number | null }> {261    const rec = this.require(id);262    if (rec.exited) {263      return { exitCode: rec.info.exitCode };264    }265    rec.stopRequested = true;266    this.signalGroup(rec, "SIGTERM");267    const graceful = await this.waitExit(rec, this.graceMs);268    if (!graceful) {269      this.signalGroup(rec, "SIGKILL");270      await this.waitExit(rec, this.graceMs);271    }272    return { exitCode: rec.info.exitCode };273  }274275  async stopAll(): Promise<void> {276    const running = [...this.procs.values()].filter((rec) => !rec.exited);277    await Promise.all(running.map((rec) => this.stop(rec.info.id)));278  }279280  private register(281    child: ChildProcess,282    command: string,283    cwd: string,284    startedAt: number,285    outputSoFar: string | undefined,286  ): ManagedProcess {287    this.counter += 1;288    const id = `p${this.counter}`;289    const logPath = path.join(this.logDir, `${id}.log`);290    const info: ManagedProcess = {291      id,292      pid: child.pid ?? -1,293      command,294      status: "running",295      exitCode: null,296      startedAt,297      cwd,298      logPath,299    };300    const rec: ProcRecord = {301      info,302      child,303      ring: [],304      ringBytes: 0,305      totalLines: 0,306      partial: "",307      cursor: 0,308      exited: false,309      failed: false,310      stopRequested: false,311      exitWaiters: [],312    };313    this.procs.set(id, rec);314315    // Create the log file immediately so it exists even before any output.316    appendFileSync(logPath, "");317    if (outputSoFar !== undefined && outputSoFar.length > 0) {318      this.ingest(rec, outputSoFar);319    }320321    const onData = (chunk: Buffer): void => {322      this.ingest(rec, chunk.toString("utf8"));323    };324    child.stdout?.on("data", onData);325    child.stderr?.on("data", onData);326327    child.on("error", () => {328      rec.failed = true;329      this.finalize(rec, null);330    });331    child.on("close", (code) => {332      this.finalize(rec, code);333    });334335    this.sink?.onProcessStarted?.({ ...info });336    return info;337  }338339  private ingest(rec: ProcRecord, text: string): void {340    // Everything goes to the spill log first — the ring may evict, the log never does.341    appendFileSync(rec.info.logPath, text);342    const combined = rec.partial + text;343    const parts = combined.split("\n");344    rec.partial = parts.pop() ?? "";345    for (const line of parts) {346      this.pushLine(rec, line);347    }348  }349350  private pushLine(rec: ProcRecord, line: string): void {351    rec.ring.push(line);352    rec.ringBytes += line.length + 1;353    rec.totalLines += 1;354    while (355      rec.ring.length > 0 &&356      (rec.ring.length > this.maxRingLines || rec.ringBytes > this.maxRingBytes)357    ) {358      const evicted = rec.ring.shift() as string;359      rec.ringBytes -= evicted.length + 1;360    }361  }362363  private finalize(rec: ProcRecord, exitCode: number | null): void {364    if (rec.exited) return;365    rec.exited = true;366    if (rec.partial.length > 0) {367      // Flush the trailing unterminated line so reads see the final output.368      this.pushLine(rec, rec.partial);369      rec.partial = "";370    }371    rec.info.exitCode = exitCode;372    rec.info.status = rec.failed ? "failed" : rec.stopRequested ? "stopped" : "exited";373    const waiters = rec.exitWaiters.splice(0);374    for (const waiter of waiters) waiter();375    this.sink?.onProcessExited?.({ ...rec.info });376  }377378  /**379   * Signal the whole process group. PID-reuse guard (Hermes §6.2): never380   * signal after we have observed the exit, and only signal the exact pid381   * recorded at spawn time.382   */383  private signalGroup(rec: ProcRecord, signal: NodeJS.Signals): void {384    if (rec.exited) return;385    const pid = rec.child.pid;386    if (pid === undefined || pid !== rec.info.pid) return;387    try {388      process.kill(-pid, signal);389    } catch {390      try {391        rec.child.kill(signal);392      } catch {393        // Already gone.394      }395    }396  }397398  private waitExit(rec: ProcRecord, ms: number): Promise<boolean> {399    if (rec.exited) return Promise.resolve(true);400    return new Promise<boolean>((resolvePromise) => {401      const timer = setTimeout(() => {402        const index = rec.exitWaiters.indexOf(waiter);403        if (index !== -1) rec.exitWaiters.splice(index, 1);404        resolvePromise(false);405      }, ms);406      const waiter = (): void => {407        clearTimeout(timer);408        resolvePromise(true);409      };410      rec.exitWaiters.push(waiter);411    });412  }413414  private require(id: string): ProcRecord {415    const rec = this.procs.get(id);416    if (rec === undefined) {417      const active = [...this.procs.values()]418        .map((r) => `${r.info.id} (${r.info.command}, ${r.info.status})`)419        .join(", ");420      throw new WorkspaceError(421        "process-unknown",422        `No process "${id}".${active.length > 0 ? ` Known: ${active}.` : " No processes have been started."}`,423        { id },424      );425    }426    return rec;427  }428}429