baa-conductor

session.ts

  1import {
  2  type LocalRunLogContext,
  3  type LocalRunLogSession,
  4  type LocalRunLogSummary,
  5  type LocalRunPaths,
  6  type StreamChunkInput,
  7  type StreamChunkLogEntry,
  8  type StreamChunkLogStream,
  9  type StreamLogChannel,
 10  type WorkerLifecycleEventInput,
 11  type WorkerLifecycleLogEntry
 12} from "./contracts";
 13import { createLocalRunLogTargets } from "./paths";
 14
 15function allocateSeq(session: LocalRunLogSession): number {
 16  const seq = session.nextSeq;
 17  session.nextSeq += 1;
 18
 19  return seq;
 20}
 21
 22function getStream(session: LocalRunLogSession, channel: StreamLogChannel): StreamChunkLogStream {
 23  switch (channel) {
 24    case "stdout":
 25      return session.stdout;
 26    case "stderr":
 27      return session.stderr;
 28  }
 29}
 30
 31export function serializeLifecycleEvent(
 32  entry: Omit<WorkerLifecycleLogEntry, "renderedLine">
 33): string {
 34  return JSON.stringify({
 35    seq: entry.seq,
 36    eventId: entry.eventId,
 37    taskId: entry.taskId,
 38    stepId: entry.stepId,
 39    runId: entry.runId,
 40    type: entry.type,
 41    level: entry.level,
 42    createdAt: entry.createdAt,
 43    message: entry.message,
 44    data: entry.data
 45  });
 46}
 47
 48export function createLocalRunLogSession(
 49  paths: LocalRunPaths,
 50  context: LocalRunLogContext
 51): LocalRunLogSession {
 52  const targets = createLocalRunLogTargets(paths);
 53
 54  return {
 55    context,
 56    paths,
 57    nextSeq: 1,
 58    worker: {
 59      channel: "worker",
 60      filePath: targets.worker.filePath,
 61      entries: []
 62    },
 63    stdout: {
 64      channel: "stdout",
 65      filePath: targets.stdout.filePath,
 66      entries: []
 67    },
 68    stderr: {
 69      channel: "stderr",
 70      filePath: targets.stderr.filePath,
 71      entries: []
 72    }
 73  };
 74}
 75
 76export function recordLifecycleEvent(
 77  session: LocalRunLogSession,
 78  input: WorkerLifecycleEventInput
 79): WorkerLifecycleLogEntry {
 80  const seq = allocateSeq(session);
 81  const createdAt = input.createdAt ?? new Date().toISOString();
 82  const baseEntry = {
 83    ...session.context,
 84    seq,
 85    channel: "worker" as const,
 86    eventId: `${session.context.runId}:event:${String(seq).padStart(4, "0")}`,
 87    type: input.type,
 88    level: input.level,
 89    createdAt,
 90    message: input.message,
 91    data: input.data ?? {}
 92  };
 93  const entry: WorkerLifecycleLogEntry = {
 94    ...baseEntry,
 95    renderedLine: serializeLifecycleEvent(baseEntry)
 96  };
 97
 98  session.worker.entries.push(entry);
 99
100  return entry;
101}
102
103export function appendStreamChunk(
104  session: LocalRunLogSession,
105  channel: StreamLogChannel,
106  input: StreamChunkInput
107): StreamChunkLogEntry {
108  const seq = allocateSeq(session);
109  const entry: StreamChunkLogEntry = {
110    ...session.context,
111    seq,
112    channel,
113    createdAt: input.createdAt ?? new Date().toISOString(),
114    text: input.text
115  };
116
117  getStream(session, channel).entries.push(entry);
118
119  return entry;
120}
121
122export function summarizeLocalRunLogSession(session: LocalRunLogSession): LocalRunLogSummary {
123  const lifecycleEventCount = session.worker.entries.length;
124  const stdoutChunkCount = session.stdout.entries.length;
125  const stderrChunkCount = session.stderr.entries.length;
126
127  return {
128    totalEntries: lifecycleEventCount + stdoutChunkCount + stderrChunkCount,
129    lastSeq: session.nextSeq - 1,
130    lifecycleEventCount,
131    stdoutChunkCount,
132    stderrChunkCount
133  };
134}