111 lines
3.2 KiB
TypeScript
111 lines
3.2 KiB
TypeScript
import { mkdir, readFile, readdir, rm } from "node:fs/promises";
|
|
import { join } from "node:path";
|
|
import type {
|
|
AgentRun,
|
|
RunEvent,
|
|
RunRepository,
|
|
} from "@great-agent/agent-core";
|
|
import { appendNdjson } from "../atomic-writes/append-ndjson";
|
|
import { writeJsonAtomically } from "../atomic-writes/write-json-atomically";
|
|
import type { DataLayout } from "../layout/data-layout";
|
|
|
|
export class FileRunRepository implements RunRepository {
|
|
constructor(private readonly layout: DataLayout) {}
|
|
|
|
async create(run: AgentRun): Promise<void> {
|
|
await this.update(run);
|
|
}
|
|
|
|
async update(run: AgentRun): Promise<void> {
|
|
await mkdir(this.runDirectory(run.id), { recursive: true });
|
|
await writeJsonAtomically(this.summaryPath(run.id), run);
|
|
}
|
|
|
|
async getById(id: string): Promise<AgentRun | null> {
|
|
try {
|
|
return JSON.parse(
|
|
await readFile(this.summaryPath(id), "utf8"),
|
|
) as AgentRun;
|
|
} catch (error) {
|
|
if (error instanceof Error && "code" in error && error.code === "ENOENT")
|
|
return null;
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async findActive(): Promise<AgentRun | null> {
|
|
return (
|
|
(await this.readAll()).find(
|
|
(run) => run.status === "running" || run.status === "waiting_user",
|
|
) ?? null
|
|
);
|
|
}
|
|
|
|
async appendEvent(event: RunEvent): Promise<void> {
|
|
await appendNdjson(this.eventsPath(event.runId), event);
|
|
}
|
|
|
|
async listEvents(runId: string): Promise<readonly RunEvent[]> {
|
|
try {
|
|
const text = await readFile(this.eventsPath(runId), "utf8");
|
|
return text
|
|
.trim()
|
|
.split("\n")
|
|
.filter(Boolean)
|
|
.map((line) => JSON.parse(line) as RunEvent);
|
|
} catch (error) {
|
|
if (error instanceof Error && "code" in error && error.code === "ENOENT")
|
|
return [];
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async hasActiveForConversations(
|
|
conversationIds: readonly string[],
|
|
): Promise<boolean> {
|
|
if (conversationIds.length === 0) return false;
|
|
const targets = new Set(conversationIds);
|
|
return (await this.readAll()).some(
|
|
(run) =>
|
|
targets.has(run.conversationId) &&
|
|
(run.status === "running" || run.status === "waiting_user"),
|
|
);
|
|
}
|
|
|
|
async deleteByConversations(
|
|
conversationIds: readonly string[],
|
|
): Promise<void> {
|
|
if (conversationIds.length === 0) return;
|
|
const targets = new Set(conversationIds);
|
|
const runs = await this.readAll();
|
|
await Promise.all(
|
|
runs
|
|
.filter((run) => targets.has(run.conversationId))
|
|
.map((run) =>
|
|
rm(this.runDirectory(run.id), { recursive: true, force: true }),
|
|
),
|
|
);
|
|
}
|
|
|
|
private async readAll(): Promise<AgentRun[]> {
|
|
await mkdir(this.layout.runs, { recursive: true });
|
|
const names = await readdir(this.layout.runs, { withFileTypes: true });
|
|
const values = await Promise.all(
|
|
names
|
|
.filter((entry) => entry.isDirectory())
|
|
.map((entry) => this.getById(entry.name)),
|
|
);
|
|
return values.filter((run): run is AgentRun => run !== null);
|
|
}
|
|
|
|
private runDirectory(id: string) {
|
|
return join(this.layout.runs, id);
|
|
}
|
|
private summaryPath(id: string) {
|
|
return join(this.runDirectory(id), "run.json");
|
|
}
|
|
private eventsPath(id: string) {
|
|
return join(this.runDirectory(id), "events.ndjson");
|
|
}
|
|
}
|