baa-conductor


baa-conductor / apps / claude-coded / src
im_wower  ·  2026-03-29

daemon.ts

  1import { spawn } from "node:child_process";
  2import { randomUUID } from "node:crypto";
  3import { appendFile, mkdir, readFile, writeFile } from "node:fs/promises";
  4
  5import type {
  6  ClaudeCodedAskResult,
  7  ClaudeCodedDaemonIdentity,
  8  ClaudeCodedDaemonState,
  9  ClaudeCodedEnvironment,
 10  ClaudeCodedEventLevel,
 11  ClaudeCodedManagedChildState,
 12  ClaudeCodedRecentEvent,
 13  ClaudeCodedRecentEventCache,
 14  ClaudeCodedResolvedConfig,
 15  ClaudeCodedStatusSnapshot,
 16  ClaudeCodedStreamEvent
 17} from "./contracts.js";
 18import {
 19  createStreamJsonTransport,
 20  type StreamJsonTransport
 21} from "./stream-json-transport.js";
 22
 23export interface ClaudeCodedChildProcessLike {
 24  pid?: number;
 25  stdin?: { end(chunk?: string | Uint8Array): unknown; write(chunk: string | Uint8Array): boolean };
 26  stderr?: {
 27    on(event: "data", listener: (chunk: string | Uint8Array) => void): unknown;
 28    on(event: "end", listener: () => void): unknown;
 29    on(event: "error", listener: (error: Error) => void): unknown;
 30    off?(event: "data", listener: (chunk: string | Uint8Array) => void): unknown;
 31    off?(event: "end", listener: () => void): unknown;
 32    off?(event: "error", listener: (error: Error) => void): unknown;
 33    setEncoding?(encoding: string): unknown;
 34  };
 35  stdout?: {
 36    on(event: "data", listener: (chunk: string | Uint8Array) => void): unknown;
 37    on(event: "end", listener: () => void): unknown;
 38    on(event: "error", listener: (error: Error) => void): unknown;
 39    off?(event: "data", listener: (chunk: string | Uint8Array) => void): unknown;
 40    off?(event: "end", listener: () => void): unknown;
 41    off?(event: "error", listener: (error: Error) => void): unknown;
 42    setEncoding?(encoding: string): unknown;
 43  };
 44  kill(signal?: string): boolean;
 45  on(event: "error", listener: (error: Error) => void): unknown;
 46  on(event: "exit", listener: (code: number | null, signal: string | null) => void): unknown;
 47  on(event: "spawn", listener: () => void): unknown;
 48  off?(event: "error", listener: (error: Error) => void): unknown;
 49  off?(event: "exit", listener: (code: number | null, signal: string | null) => void): unknown;
 50  off?(event: "spawn", listener: () => void): unknown;
 51  once(event: "error", listener: (error: Error) => void): unknown;
 52  once(event: "exit", listener: (code: number | null, signal: string | null) => void): unknown;
 53  once(event: "spawn", listener: () => void): unknown;
 54}
 55
 56export type ClaudeCodedRuntimeEventListener = (event: ClaudeCodedRecentEvent) => void;
 57
 58export interface ClaudeCodedDaemonOptions {
 59  env?: ClaudeCodedEnvironment;
 60  spawner?: {
 61    spawn(
 62      command: string,
 63      args: readonly string[],
 64      options: { cwd: string; env: ClaudeCodedEnvironment }
 65    ): ClaudeCodedChildProcessLike;
 66  };
 67}
 68
 69type AskWaiter = {
 70  events: ClaudeCodedStreamEvent[];
 71  onEvent: ((event: ClaudeCodedStreamEvent) => void) | null;
 72  resolve: (result: ClaudeCodedAskResult) => void;
 73  reject: (error: Error) => void;
 74  timer: unknown;
 75};
 76
 77const STOP_TIMEOUT_MS = 5_000;
 78const RESTART_BASE_DELAY_MS = 1_000;
 79const RESTART_MAX_DELAY_MS = 60_000;
 80
 81export class ClaudeCodedDaemon {
 82  private child: ClaudeCodedChildProcessLike | null = null;
 83  private transport: StreamJsonTransport | null = null;
 84  private identity: ClaudeCodedDaemonIdentity | null = null;
 85  private daemonState: ClaudeCodedDaemonState | null = null;
 86  private recentEvents: ClaudeCodedRecentEventCache;
 87  private readonly env: ClaudeCodedEnvironment;
 88  private readonly eventListeners = new Set<ClaudeCodedRuntimeEventListener>();
 89  private readonly spawner: ClaudeCodedDaemonOptions["spawner"];
 90  private started = false;
 91  private initialized = false;
 92  private restartCount = 0;
 93  private restartTimer: unknown = null;
 94  private stopping = false;
 95  private pendingAsk: AskWaiter | null = null;
 96  private nextEventSeq = 1;
 97  private stderrBuffer = "";
 98
 99  constructor(
100    private readonly config: ClaudeCodedResolvedConfig,
101    options: ClaudeCodedDaemonOptions = {}
102  ) {
103    this.env = options.env ?? (typeof process !== "undefined" ? process.env : {});
104    this.spawner = options.spawner ?? {
105      spawn(command, args, spawnOptions) {
106        return spawn(command, [...args], {
107          cwd: spawnOptions.cwd,
108          env: spawnOptions.env,
109          stdio: ["pipe", "pipe", "pipe"]
110        }) as unknown as ClaudeCodedChildProcessLike;
111      }
112    };
113    this.recentEvents = {
114      maxEntries: config.eventCacheSize,
115      updatedAt: null,
116      events: []
117    };
118  }
119
120  async start(): Promise<ClaudeCodedStatusSnapshot> {
121    await this.initialize();
122
123    if (this.started) {
124      return this.getStatusSnapshot();
125    }
126
127    this.started = true;
128    const now = new Date().toISOString();
129    this.daemonState = {
130      started: true,
131      startedAt: now,
132      stoppedAt: null,
133      updatedAt: now,
134      pid: typeof process !== "undefined" ? process.pid ?? null : null,
135      child: createIdleChildState(this.config)
136    };
137    await this.persistDaemonState();
138
139    this.recordEvent({
140      level: "info",
141      type: "daemon.started",
142      message: "claude-coded started."
143    });
144
145    await this.spawnChild();
146    return this.getStatusSnapshot();
147  }
148
149  async stop(): Promise<ClaudeCodedStatusSnapshot> {
150    this.stopping = true;
151
152    if (this.restartTimer != null) {
153      clearTimeout(this.restartTimer as ReturnType<typeof setTimeout>);
154      this.restartTimer = null;
155    }
156
157    this.rejectPendingAsk(new Error("Daemon is stopping."));
158
159    if (this.transport != null) {
160      this.transport.close();
161      this.transport = null;
162    }
163
164    if (this.child != null) {
165      const child = this.child;
166      this.child = null;
167
168      const exited = waitForChildExit(child, STOP_TIMEOUT_MS);
169
170      try {
171        child.kill("SIGTERM");
172      } catch {
173        // ignore
174      }
175
176      const didExit = await exited;
177
178      if (!didExit) {
179        try {
180          child.kill("SIGKILL");
181        } catch {
182          // ignore
183        }
184      }
185    }
186
187    const now = new Date().toISOString();
188
189    if (this.daemonState != null) {
190      this.daemonState.started = false;
191      this.daemonState.stoppedAt = now;
192      this.daemonState.updatedAt = now;
193      this.daemonState.child.status = "stopped";
194      await this.persistDaemonState();
195    }
196
197    this.recordEvent({
198      level: "info",
199      type: "daemon.stopped",
200      message: "claude-coded stopped."
201    });
202
203    this.started = false;
204    this.stopping = false;
205    return this.getStatusSnapshot();
206  }
207
208  getStatusSnapshot(): ClaudeCodedStatusSnapshot {
209    return {
210      config: this.config,
211      identity: this.identity ?? createDefaultIdentity(this.config),
212      daemon: this.daemonState ?? createDefaultDaemonState(this.config),
213      recentEvents: { ...this.recentEvents }
214    };
215  }
216
217  subscribe(listener: ClaudeCodedRuntimeEventListener): { unsubscribe: () => void } {
218    this.eventListeners.add(listener);
219    return {
220      unsubscribe: () => {
221        this.eventListeners.delete(listener);
222      }
223    };
224  }
225
226  async ask(prompt: string): Promise<ClaudeCodedAskResult> {
227    if (!this.started || this.transport == null || this.transport.closed) {
228      throw new Error("claude-coded child is not running.");
229    }
230
231    if (this.pendingAsk != null) {
232      throw new Error("A request is already in progress. claude-coded processes one request at a time.");
233    }
234
235    return new Promise<ClaudeCodedAskResult>((resolve, reject) => {
236      const timer = setTimeout(() => {
237        this.rejectPendingAsk(new Error(`Ask timed out after ${this.config.turnTimeoutMs}ms.`));
238      }, this.config.turnTimeoutMs);
239
240      this.pendingAsk = {
241        events: [],
242        onEvent: null,
243        resolve,
244        reject,
245        timer
246      };
247
248      try {
249        this.transport!.send({
250          type: "user",
251          message: { role: "user", content: prompt },
252          parent_tool_use_id: null,
253          session_id: null
254        });
255      } catch (error) {
256        this.rejectPendingAsk(error instanceof Error ? error : new Error(String(error)));
257      }
258    });
259  }
260
261  askStream(prompt: string): {
262    events: AsyncIterable<ClaudeCodedStreamEvent>;
263    result: Promise<ClaudeCodedAskResult>;
264  } {
265    if (!this.started || this.transport == null || this.transport.closed) {
266      throw new Error("claude-coded child is not running.");
267    }
268
269    if (this.pendingAsk != null) {
270      throw new Error("A request is already in progress. claude-coded processes one request at a time.");
271    }
272
273    let eventResolve: ((value: IteratorResult<ClaudeCodedStreamEvent>) => void) | null = null;
274    const eventQueue: ClaudeCodedStreamEvent[] = [];
275    let done = false;
276
277    const events: AsyncIterable<ClaudeCodedStreamEvent> = {
278      [Symbol.asyncIterator]() {
279        return {
280          next(): Promise<IteratorResult<ClaudeCodedStreamEvent>> {
281            if (eventQueue.length > 0) {
282              return Promise.resolve({ value: eventQueue.shift()!, done: false });
283            }
284            if (done) {
285              return Promise.resolve({ value: undefined as unknown as ClaudeCodedStreamEvent, done: true });
286            }
287            return new Promise((resolve) => {
288              eventResolve = resolve;
289            });
290          }
291        };
292      }
293    };
294
295    const result = new Promise<ClaudeCodedAskResult>((resolve, reject) => {
296      const timer = setTimeout(() => {
297        done = true;
298        if (eventResolve) {
299          eventResolve({ value: undefined as unknown as ClaudeCodedStreamEvent, done: true });
300          eventResolve = null;
301        }
302        this.rejectPendingAsk(new Error(`Ask timed out after ${this.config.turnTimeoutMs}ms.`));
303      }, this.config.turnTimeoutMs);
304
305      this.pendingAsk = {
306        events: [],
307        onEvent: (event) => {
308          if (eventResolve) {
309            const r = eventResolve;
310            eventResolve = null;
311            r({ value: event, done: false });
312          } else {
313            eventQueue.push(event);
314          }
315        },
316        resolve: (askResult) => {
317          clearTimeout(timer as ReturnType<typeof setTimeout>);
318          done = true;
319          if (eventResolve) {
320            eventResolve({ value: undefined as unknown as ClaudeCodedStreamEvent, done: true });
321            eventResolve = null;
322          }
323          resolve(askResult);
324        },
325        reject: (error) => {
326          clearTimeout(timer as ReturnType<typeof setTimeout>);
327          done = true;
328          if (eventResolve) {
329            eventResolve({ value: undefined as unknown as ClaudeCodedStreamEvent, done: true });
330            eventResolve = null;
331          }
332          reject(error);
333        },
334        timer
335      };
336
337      try {
338        this.transport!.send({
339          type: "user",
340          message: { role: "user", content: prompt },
341          parent_tool_use_id: null,
342          session_id: null
343        });
344      } catch (error) {
345        done = true;
346        if (eventResolve) {
347          eventResolve({ value: undefined as unknown as ClaudeCodedStreamEvent, done: true });
348          eventResolve = null;
349        }
350        this.rejectPendingAsk(error instanceof Error ? error : new Error(String(error)));
351      }
352    });
353
354    return { events, result };
355  }
356
357  private async initialize(): Promise<void> {
358    if (this.initialized) {
359      return;
360    }
361
362    await mkdir(this.config.paths.logsDir, { recursive: true });
363    await mkdir(this.config.paths.stateDir, { recursive: true });
364
365    this.identity = await readJsonOrDefault<ClaudeCodedDaemonIdentity | null>(
366      this.config.paths.identityPath,
367      null
368    );
369
370    if (this.identity == null) {
371      this.identity = {
372        daemonId: randomUUID(),
373        nodeId: this.config.nodeId,
374        repoRoot: this.config.paths.repoRoot,
375        createdAt: new Date().toISOString(),
376        version: this.config.version
377      };
378      await writeFile(
379        this.config.paths.identityPath,
380        JSON.stringify(this.identity, null, 2),
381        "utf8"
382      );
383    }
384
385    this.daemonState = await readJsonOrDefault<ClaudeCodedDaemonState | null>(
386      this.config.paths.daemonStatePath,
387      null
388    );
389
390    this.initialized = true;
391  }
392
393  private async spawnChild(): Promise<void> {
394    if (this.stopping) {
395      return;
396    }
397
398    const now = new Date().toISOString();
399    this.updateChildState({
400      status: "starting",
401      pid: null,
402      startedAt: null,
403      exitedAt: null,
404      exitCode: null,
405      signal: null,
406      lastError: null,
407      restartCount: this.restartCount
408    });
409
410    let child: ClaudeCodedChildProcessLike;
411
412    try {
413      child = this.spawner!.spawn(
414        this.config.child.command,
415        this.config.child.args,
416        {
417          cwd: this.config.child.cwd,
418          env: {
419            ...this.env,
420            BAA_CLAUDE_CODED_DAEMON_ID: this.identity?.daemonId ?? ""
421          }
422        }
423      );
424    } catch (error) {
425      this.updateChildState({
426        status: "failed",
427        lastError: formatErrorMessage(error)
428      });
429      this.recordEvent({
430        level: "error",
431        type: "child.spawn.failed",
432        message: formatErrorMessage(error)
433      });
434      this.scheduleRestart();
435      return;
436    }
437
438    this.child = child;
439
440    try {
441      await waitForChildSpawn(child);
442    } catch (error) {
443      this.updateChildState({
444        status: "failed",
445        exitedAt: new Date().toISOString(),
446        lastError: formatErrorMessage(error)
447      });
448      this.recordEvent({
449        level: "error",
450        type: "child.spawn.failed",
451        message: formatErrorMessage(error)
452      });
453      this.child = null;
454      this.scheduleRestart();
455      return;
456    }
457
458    this.updateChildState({
459      status: "running",
460      pid: child.pid ?? null,
461      startedAt: new Date().toISOString()
462    });
463    this.recordEvent({
464      level: "info",
465      type: "child.started",
466      message: `Started Claude Code child process ${child.pid ?? "unknown"}.`,
467      detail: {
468        args: this.config.child.args,
469        command: this.config.child.command,
470        cwd: this.config.child.cwd
471      }
472    });
473
474    this.restartCount = 0;
475
476    const transport = createStreamJsonTransport({
477      process: child,
478      onMessage: (message) => {
479        this.handleChildMessage(message);
480      },
481      onClose: (error) => {
482        this.handleChildClose(error);
483      },
484      onStderr: (text) => {
485        this.stderrBuffer += text;
486        if (this.stderrBuffer.length > 4096) {
487          this.stderrBuffer = this.stderrBuffer.slice(-2048);
488        }
489      },
490      onCloseDiagnostic: (diagnostic) => {
491        this.recordEvent({
492          level: "warn",
493          type: diagnostic.source === "stderr.error" ? "transport.stderr.error" : "transport.closed",
494          message: diagnostic.message,
495          detail: {
496            source: diagnostic.source,
497            exitCode: diagnostic.exitCode,
498            signal: diagnostic.signal
499          }
500        });
501      }
502    });
503
504    transport.connect();
505    this.transport = transport;
506
507    await this.persistDaemonState();
508  }
509
510  private handleChildMessage(message: Record<string, unknown>): void {
511    const event: ClaudeCodedStreamEvent = message as ClaudeCodedStreamEvent;
512    const messageType = typeof message.type === "string" ? message.type : "unknown";
513
514    if (this.pendingAsk != null) {
515      this.pendingAsk.events.push(event);
516      this.pendingAsk.onEvent?.(event);
517
518      if (messageType === "result") {
519        const waiter = this.pendingAsk;
520        this.pendingAsk = null;
521        clearTimeout(waiter.timer as ReturnType<typeof setTimeout>);
522
523        const result: ClaudeCodedAskResult = {
524          ok: message.is_error !== true,
525          result: typeof message.result === "string" ? message.result : null,
526          sessionId: typeof message.session_id === "string" ? message.session_id : null,
527          costUsd: typeof message.total_cost_usd === "number" ? message.total_cost_usd : (typeof message.cost_usd === "number" ? message.cost_usd : null),
528          durationMs: typeof message.duration_ms === "number" ? message.duration_ms : null,
529          isError: message.is_error === true,
530          events: waiter.events
531        };
532        waiter.resolve(result);
533      }
534    }
535
536    this.emitRuntimeEvent({
537      level: "info",
538      type: `stream.${messageType}`,
539      message: `Claude Code stream event: ${messageType}`,
540      detail: message as Record<string, unknown>
541    });
542  }
543
544  private handleChildClose(error: Error): void {
545    const exitedAt = new Date().toISOString();
546
547    this.updateChildState({
548      status: "failed",
549      pid: null,
550      exitedAt,
551      lastError: error.message
552    });
553
554    this.recordEvent({
555      level: "error",
556      type: "child.exited",
557      message: error.message,
558      detail: {
559        stderrTail: this.stderrBuffer.slice(-512) || null
560      }
561    });
562
563    this.child = null;
564    this.transport = null;
565    this.stderrBuffer = "";
566
567    this.rejectPendingAsk(new Error(`Claude Code child exited: ${error.message}`));
568
569    if (this.started && !this.stopping) {
570      this.scheduleRestart();
571    }
572
573    void this.persistDaemonState();
574  }
575
576  private scheduleRestart(): void {
577    if (this.stopping || !this.started) {
578      return;
579    }
580
581    this.restartCount += 1;
582    const delay = Math.min(
583      RESTART_BASE_DELAY_MS * Math.pow(2, this.restartCount - 1),
584      RESTART_MAX_DELAY_MS
585    );
586
587    this.recordEvent({
588      level: "info",
589      type: "child.restart.scheduled",
590      message: `Scheduling restart #${this.restartCount} in ${delay}ms.`
591    });
592
593    this.restartTimer = setTimeout(() => {
594      this.restartTimer = null;
595      void this.spawnChild();
596    }, delay);
597  }
598
599  private rejectPendingAsk(error: Error): void {
600    if (this.pendingAsk != null) {
601      const waiter = this.pendingAsk;
602      this.pendingAsk = null;
603      clearTimeout(waiter.timer as ReturnType<typeof setTimeout>);
604      waiter.reject(error);
605    }
606  }
607
608  private updateChildState(partial: Partial<ClaudeCodedManagedChildState>): void {
609    if (this.daemonState == null) {
610      return;
611    }
612
613    Object.assign(this.daemonState.child, partial);
614    this.daemonState.updatedAt = new Date().toISOString();
615  }
616
617  private recordEvent(input: {
618    detail?: Record<string, unknown> | null;
619    level: ClaudeCodedEventLevel;
620    message: string;
621    type: string;
622  }): void {
623    const event: ClaudeCodedRecentEvent = {
624      seq: this.nextEventSeq++,
625      createdAt: new Date().toISOString(),
626      level: input.level,
627      type: input.type,
628      message: input.message,
629      detail: input.detail ?? null
630    };
631
632    this.recentEvents.events.push(event);
633    this.recentEvents.updatedAt = event.createdAt;
634
635    while (this.recentEvents.events.length > this.recentEvents.maxEntries) {
636      this.recentEvents.events.shift();
637    }
638
639    this.emitRuntimeEvent(event);
640
641    void appendFile(
642      this.config.paths.structuredEventLogPath,
643      `${JSON.stringify(event)}\n`,
644      "utf8"
645    ).catch(() => {});
646  }
647
648  private emitRuntimeEvent(event: ClaudeCodedRecentEvent | {
649    level: ClaudeCodedEventLevel;
650    type: string;
651    message: string;
652    detail?: Record<string, unknown> | null;
653  }): void {
654    const normalized: ClaudeCodedRecentEvent = "seq" in event
655      ? event
656      : {
657          seq: this.nextEventSeq++,
658          createdAt: new Date().toISOString(),
659          level: event.level,
660          type: event.type,
661          message: event.message,
662          detail: event.detail ?? null
663        };
664
665    for (const listener of this.eventListeners) {
666      try {
667        listener(normalized);
668      } catch {
669        // ignore listener errors
670      }
671    }
672  }
673
674  private async persistDaemonState(): Promise<void> {
675    if (this.daemonState == null) {
676      return;
677    }
678
679    try {
680      await writeFile(
681        this.config.paths.daemonStatePath,
682        JSON.stringify(this.daemonState, null, 2),
683        "utf8"
684      );
685    } catch {
686      // ignore persistence errors
687    }
688  }
689}
690
691function createIdleChildState(config: ClaudeCodedResolvedConfig): ClaudeCodedManagedChildState {
692  return {
693    status: "idle",
694    command: config.child.command,
695    args: [...config.child.args],
696    cwd: config.child.cwd,
697    pid: null,
698    startedAt: null,
699    exitedAt: null,
700    exitCode: null,
701    signal: null,
702    lastError: null,
703    restartCount: 0
704  };
705}
706
707function createDefaultIdentity(config: ClaudeCodedResolvedConfig): ClaudeCodedDaemonIdentity {
708  return {
709    daemonId: "uninitialized",
710    nodeId: config.nodeId,
711    repoRoot: config.paths.repoRoot,
712    createdAt: new Date().toISOString(),
713    version: config.version
714  };
715}
716
717function createDefaultDaemonState(config: ClaudeCodedResolvedConfig): ClaudeCodedDaemonState {
718  return {
719    started: false,
720    startedAt: null,
721    stoppedAt: null,
722    updatedAt: new Date().toISOString(),
723    pid: null,
724    child: createIdleChildState(config)
725  };
726}
727
728function formatErrorMessage(error: unknown): string {
729  if (error instanceof Error) {
730    return error.message;
731  }
732
733  return String(error);
734}
735
736function waitForChildSpawn(child: ClaudeCodedChildProcessLike): Promise<void> {
737  return new Promise((resolve, reject) => {
738    const onSpawn = () => {
739      child.once("error", () => {});
740      resolve();
741    };
742    const onError = (error: Error) => {
743      reject(error);
744    };
745
746    child.once("spawn", onSpawn);
747    child.once("error", onError);
748  });
749}
750
751function waitForChildExit(child: ClaudeCodedChildProcessLike, timeoutMs: number): Promise<boolean> {
752  return new Promise((resolve) => {
753    let resolved = false;
754    const timer = setTimeout(() => {
755      if (!resolved) {
756        resolved = true;
757        resolve(false);
758      }
759    }, timeoutMs);
760
761    child.once("exit", () => {
762      if (!resolved) {
763        resolved = true;
764        clearTimeout(timer as ReturnType<typeof setTimeout>);
765        resolve(true);
766      }
767    });
768  });
769}
770
771async function readJsonOrDefault<T>(path: string, fallback: T): Promise<T> {
772  try {
773    const text = await readFile(path, "utf8");
774    return JSON.parse(text) as T;
775  } catch {
776    return fallback;
777  }
778}