Write dev ingestion directly to DuckDB cache

Author: marcusschiesser

.changeset/direct-duckdb-dev.md

@@ -0,0 +1,5 @@
+---
+"agentpond": minor
+---
+
+Write `agentpond dev` ingestion directly to the dev DuckDB cache and make dev sync a no-op while blocking competing dev writes.

apps/cli/src/commands/dev.ts

@@ -1,9 +1,10 @@
 import {
+	acquireDevServerLock,
 	configFromEnv,
-	FileSystemObjectStore,
 	initAgentPondEnvironment,
 	selectAgentPondEnvironment,
 } from "@agentpond/core";
+import { AgentPondCache, DuckDbIngestionWriter } from "@agentpond/duckdb";
 import { buildServer } from "@agentpond/ingest";
 import {
 	CliError,
@@ -19,18 +20,37 @@ export async function startDevServer(parsed: ParsedArgs): Promise<void> {
 	selectAgentPondEnvironment(environment.name);
 	const devConfig = configFromEnv({
 		envName: environment.name,
-		eventStorePath: stringFlag(parsed, "event-store"),
 		storeType: "local",
 	});
 	const devEnvironment = devConfig.environment;
 	if (!devEnvironment)
 		throw new CliError("Missing dev environment configuration");
+	const lock = acquireDevServerLock(devEnvironment);
+	const db = new AgentPondCache(devConfig.dbPath);
 	const server = buildServer({
 		config: devConfig,
-		store: new FileSystemObjectStore(devEnvironment.eventStorePath),
+		handlers: new DuckDbIngestionWriter(db.directIngestion()),
 		authMode: "disabled",
 	});
-	await server.listen({ host, port });
+	const shutdown = async () => {
+		await server.close();
+	};
+	server.addHook("onClose", async () => {
+		await db.close();
+		lock.release();
+	});
+	process.once("SIGINT", shutdown);
+	process.once("SIGTERM", shutdown);
+	try {
+		await db.init();
+		await server.listen({ host, port });
+	} catch (error) {
+		process.off("SIGINT", shutdown);
+		process.off("SIGTERM", shutdown);
+		await db.close();
+		lock.release();
+		throw error;
+	}
 	const baseUrl = `http://${host}:${port}`;
 	console.log(`AgentPond dev server listening at ${baseUrl}`);
 	console.log("");
@@ -40,6 +60,5 @@ export async function startDevServer(parsed: ParsedArgs): Promise<void> {
 	console.log("LANGFUSE_SECRET_KEY=sk-agentpond-dev");
 	console.log("");
 	console.log(`Environment: ${devEnvironment.name}`);
-	console.log(`Event store: ${devEnvironment.eventStorePath}`);
 	console.log(`DuckDB cache: ${devConfig.dbPath}`);
 }

apps/cli/src/commands/environment.ts

@@ -1,9 +1,14 @@
+import { existsSync } from "node:fs";
 import {
+	configFromEnv,
+	DEV_SERVER_RUNNING_MESSAGE,
 	initAgentPondEnvironment,
+	isDevServerRunning,
 	listAgentPondEnvironments,
 	resolveAgentPondEnvironment,
 	selectAgentPondEnvironment,
 } from "@agentpond/core";
+import { AgentPondCache } from "@agentpond/duckdb";
 import {
 	CliError,
 	type ParsedArgs,
@@ -58,6 +63,34 @@ export async function handleEnvironmentCommand(
 	throw new CliError(`Unknown command: env ${action}`);
 }
 
+export function isDevEnvironment(
+	config: ReturnType<typeof configFromEnv>,
+): boolean {
+	return config.environment?.name === "dev";
+}
+
+export function cacheForRead(
+	config: ReturnType<typeof configFromEnv>,
+): AgentPondCache {
+	if (config.environment && isDevServerRunning(config.environment)) {
+		if (!existsSync(config.dbPath)) {
+			throw new CliError(
+				"dev cache is not initialized yet; ingest a trace or stop the dev server",
+			);
+		}
+		return new AgentPondCache(config.dbPath, { accessMode: "readonly" });
+	}
+	return new AgentPondCache(config.dbPath);
+}
+
+export function assertDevServerNotRunning(
+	config: ReturnType<typeof configFromEnv>,
+): void {
+	if (config.environment && isDevServerRunning(config.environment)) {
+		throw new CliError(DEV_SERVER_RUNNING_MESSAGE);
+	}
+}
+
 function printEnvironmentHelp(): void {
 	console.log(`agentpond env - manage local AgentPond environments
 

apps/cli/src/index.ts

@@ -18,7 +18,12 @@ import {
 	stringFlag,
 } from "./cli-support.js";
 import { startDevServer } from "./commands/dev.js";
-import { handleEnvironmentCommand } from "./commands/environment.js";
+import {
+	assertDevServerNotRunning,
+	cacheForRead,
+	handleEnvironmentCommand,
+	isDevEnvironment,
+} from "./commands/environment.js";
 import { objectStoreForConfig } from "./object-store.js";
 import { manualTraceResourceSpans } from "./otel-trace.js";
 import {
@@ -36,7 +41,7 @@ export async function main(argv = process.argv): Promise<void> {
 			return handleEnvironmentCommand(action, rest, parsed);
 		if (parsed.flags.help || parsed.flags.h) return printHelp();
 		if (resource === "dev") {
-			return startDevServer(parsed);
+			return await startDevServer(parsed);
 		}
 		const config = configFromEnv({
 			envName: stringFlag(parsed, "env"),
@@ -49,6 +54,15 @@ export async function main(argv = process.argv): Promise<void> {
 		const json = Boolean(parsed.flags.json);
 		logImplicitEnvironment(parsed, config, json);
 		if (resource === "sync") {
+			if (isDevEnvironment(config)) {
+				return print(
+					{
+						skipped: true,
+						reason: "dev environment is written directly by agentpond dev",
+					},
+					json,
+				);
+			}
 			const db = new AgentPondCache(config.dbPath);
 			const result = await db.syncFromStore({
 				store: objectStoreForConfig(config),
@@ -61,19 +75,19 @@ export async function main(argv = process.argv): Promise<void> {
 		if (resource === "sql") {
 			const query = rest.length > 0 ? [action, ...rest].join(" ") : action;
 			if (!query) throw new CliError("Missing SQL query");
-			const db = new AgentPondCache(config.dbPath);
+			const db = cacheForRead(config);
 			const rows = await db.query(query);
 			await db.close();
 			return print(rows, json);
 		}
 		if (resource === "scores" && action === "create") {
-			return createScore(parsed, config, json);
+			return await createScore(parsed, config, json);
 		}
 		if (resource === "traces" && action === "create") {
-			return createTrace(parsed, config, json);
+			return await createTrace(parsed, config, json);
 		}
 
-		const db = new AgentPondCache(config.dbPath);
+		const db = cacheForRead(config);
 		const rows = await runReadCommand(db, resource, action, rest, parsed);
 		await db.close();
 		return print(rows, json);
@@ -98,6 +112,7 @@ async function createScore(
 	config: ReturnType<typeof configFromEnv>,
 	json: boolean,
 ): Promise<void> {
+	assertDevServerNotRunning(config);
 	const name = requiredFlag(parsed, "name");
 	const value = requiredFlag(parsed, "value");
 	const source = stringFlag(parsed, "source") ?? "API";
@@ -148,6 +163,7 @@ async function createTrace(
 	config: ReturnType<typeof configFromEnv>,
 	json: boolean,
 ): Promise<void> {
+	assertDevServerNotRunning(config);
 	const now = new Date().toISOString();
 	const traceId = stringFlag(parsed, "id") ?? createOtelTraceId();
 	const store = objectStoreForConfig(config);

apps/cli/tsconfig.json

@@ -7,7 +7,8 @@
 	},
 	"references": [
 		{ "path": "../../packages/core" },
-		{ "path": "../../packages/duckdb" }
+		{ "path": "../../packages/duckdb" },
+		{ "path": "../../packages/ingest" }
 	],
 	"include": ["src/**/*.ts"]
 }

packages/core/src/dev-lock.ts

@@ -0,0 +1,116 @@
+import {
+	closeSync,
+	existsSync,
+	mkdirSync,
+	openSync,
+	readFileSync,
+	unlinkSync,
+	writeFileSync,
+} from "node:fs";
+import { dirname, join } from "node:path";
+import type { AgentPondEnvironment } from "./environment.js";
+
+export const DEV_SERVER_RUNNING_MESSAGE =
+	"dev server is running; stop it or use the dev ingestion endpoint";
+
+export type DevServerLock = {
+	pid: number;
+	startedAt: string;
+	dbPath: string;
+	command: string;
+};
+
+export type AcquiredDevServerLock = {
+	path: string;
+	release: () => void;
+};
+
+export function devServerLockPath(environment: AgentPondEnvironment): string {
+	return join(environment.envDir, "dev-server.lock");
+}
+
+export function acquireDevServerLock(
+	environment: AgentPondEnvironment,
+): AcquiredDevServerLock {
+	const path = devServerLockPath(environment);
+	removeStaleDevServerLock(path);
+	mkdirSync(dirname(path), { recursive: true });
+	const fd = openSync(path, "wx");
+	closeSync(fd);
+	const lock: DevServerLock = {
+		pid: process.pid,
+		startedAt: new Date().toISOString(),
+		dbPath: environment.dbPath,
+		command: process.argv.join(" "),
+	};
+	writeFileSync(path, `${JSON.stringify(lock, null, 2)}\n`, "utf8");
+	let released = false;
+	return {
+		path,
+		release: () => {
+			if (released) return;
+			released = true;
+			try {
+				const current = readDevServerLock(path);
+				if (current?.pid === process.pid) unlinkSync(path);
+			} catch {
+				// Best-effort cleanup during process shutdown.
+			}
+		},
+	};
+}
+
+export function isDevServerRunning(environment: AgentPondEnvironment): boolean {
+	const path = devServerLockPath(environment);
+	removeStaleDevServerLock(path);
+	return readDevServerLock(path) !== undefined;
+}
+
+function removeStaleDevServerLock(path: string): void {
+	const lock = readDevServerLock(path);
+	if (!lock) {
+		if (existsSync(path)) {
+			try {
+				unlinkSync(path);
+			} catch {
+				// Another process may have removed or replaced it.
+			}
+		}
+		return;
+	}
+	if (isProcessAlive(lock.pid)) return;
+	try {
+		unlinkSync(path);
+	} catch {
+		// Another process may have removed or replaced it.
+	}
+}
+
+function readDevServerLock(path: string): DevServerLock | undefined {
+	if (!existsSync(path)) return undefined;
+	try {
+		const parsed = JSON.parse(
+			readFileSync(path, "utf8"),
+		) as Partial<DevServerLock>;
+		if (typeof parsed.pid !== "number") return undefined;
+		return {
+			pid: parsed.pid,
+			startedAt: String(parsed.startedAt ?? ""),
+			dbPath: String(parsed.dbPath ?? ""),
+			command: String(parsed.command ?? ""),
+		};
+	} catch {
+		return undefined;
+	}
+}
+
+function isProcessAlive(pid: number): boolean {
+	if (!Number.isInteger(pid) || pid <= 0) return false;
+	try {
+		process.kill(pid, 0);
+		return true;
+	} catch (error) {
+		const code = (error as NodeJS.ErrnoException).code;
+		return code === "EPERM";
+	}
+}

packages/core/src/index.ts

@@ -1,5 +1,6 @@
 export * from "./auth.js";
 export * from "./config.js";
+export * from "./dev-lock.js";
 export * from "./environment.js";
 export * from "./object-store.js";
 export * from "./otel.js";

packages/core/src/object-store/index.ts

@@ -1,4 +1,5 @@
 export * from "./filesystem.js";
+export * from "./ingestion-handler.js";
 export * from "./memory.js";
 export * from "./s3.js";
 export * from "./types.js";

packages/core/src/object-store/ingestion-handler.ts

@@ -0,0 +1,30 @@
+import { AcceptedEventWriter, type BatchResult } from "../writer.js";
+import type { ObjectStore } from "./types.js";
+
+export class ObjectStoreIngestionHandler {
+	constructor(private readonly store: ObjectStore) {}
+
+	async processBatch(params: {
+		projectId: string;
+		prefix: string;
+		batch: unknown[];
+	}): Promise<BatchResult> {
+		return new AcceptedEventWriter({
+			store: this.store,
+			projectId: params.projectId,
+			prefix: params.prefix,
+		}).processBatch(params.batch);
+	}
+
+	async writeOtelResourceSpans(params: {
+		projectId: string;
+		prefix: string;
+		resourceSpans: unknown[];
+	}): Promise<void> {
+		await new AcceptedEventWriter({
+			store: this.store,
+			projectId: params.projectId,
+			prefix: params.prefix,
+		}).writeOtelResourceSpans(params.resourceSpans);
+	}
+}