baa-conductor

persistence.ts

 1import { appendFile, mkdir, writeFile } from "node:fs/promises";
 2import type {
 3  LocalRunLogSession,
 4  LocalRunPaths,
 5  RunMetadata,
 6  RunStateSnapshot,
 7  StreamChunkLogEntry,
 8  WorkerLifecycleLogEntry
 9} from "./contracts";
10
11function renderJsonFile(value: unknown): string {
12  return `${JSON.stringify(value, null, 2)}\n`;
13}
14
15function ensureTrailingNewline(value: string): string {
16  if (value === "" || value.endsWith("\n")) {
17    return value;
18  }
19
20  return `${value}\n`;
21}
22
23export async function ensureLocalRunLayout(paths: LocalRunPaths): Promise<void> {
24  await Promise.all([
25    mkdir(paths.taskRunsDir, { recursive: true }),
26    mkdir(paths.runDir, { recursive: true }),
27    mkdir(paths.checkpointsDir, { recursive: true }),
28    mkdir(paths.artifactsDir, { recursive: true })
29  ]);
30}
31
32export async function writeRunMetadata(
33  paths: LocalRunPaths,
34  metadata: RunMetadata
35): Promise<void> {
36  await writeFile(paths.metaPath, renderJsonFile(metadata), "utf8");
37}
38
39export async function writeRunState(
40  paths: LocalRunPaths,
41  state: RunStateSnapshot
42): Promise<void> {
43  await writeFile(paths.statePath, renderJsonFile(state), "utf8");
44}
45
46export async function initializeLocalRunFiles(
47  paths: LocalRunPaths,
48  metadata: RunMetadata,
49  state: RunStateSnapshot
50): Promise<void> {
51  await ensureLocalRunLayout(paths);
52  await Promise.all([
53    writeRunMetadata(paths, metadata),
54    writeRunState(paths, state),
55    writeFile(paths.workerLogPath, "", "utf8"),
56    writeFile(paths.stdoutLogPath, "", "utf8"),
57    writeFile(paths.stderrLogPath, "", "utf8")
58  ]);
59}
60
61export async function appendLifecycleEntry(
62  session: LocalRunLogSession,
63  entry: WorkerLifecycleLogEntry
64): Promise<void> {
65  await appendFile(session.worker.filePath, ensureTrailingNewline(entry.renderedLine), "utf8");
66}
67
68export async function appendStreamEntry(
69  session: LocalRunLogSession,
70  entry: StreamChunkLogEntry
71): Promise<void> {
72  const filePath =
73    entry.channel === "stdout" ? session.stdout.filePath : session.stderr.filePath;
74
75  await appendFile(filePath, ensureTrailingNewline(entry.text), "utf8");
76}