Use UTC bucket sync for OTEL payloads

Author: marcusschiesser

.changeset/utc-bucket-sync.md

@@ -0,0 +1,5 @@
+---
+"agentpond": patch
+---
+
+Use UTC time-bucket sync for OTEL payloads and non-OTEL score manifests.

README.md

@@ -31,7 +31,7 @@ This gives you durable remote storage without requiring an always-on analytical
 
 - Langfuse-compatible ingestion endpoints for SDK and OTLP traces
 - S3-compatible raw event storage
-- Manifest-based discovery and incremental synchronization
+- UTC bucket discovery and incremental synchronization
 - Local DuckDB cache containing:
   - `events_raw`
   - `traces`

apps/cli/src/index.ts

@@ -1,19 +1,20 @@
 #!/usr/bin/env node
-import { randomUUID } from "node:crypto";
+import { randomBytes, randomUUID } from "node:crypto";
 import { realpathSync } from "node:fs";
 import { fileURLToPath, pathToFileURL } from "node:url";
 import {
-	AcceptedEventWriter,
-	type AgentPondConfig,
-	type BatchManifest,
 	configFromEnv,
 	eventTypes,
 	type IngestionEvent,
 	loadEnvFile,
-	type ObjectStore,
 	S3ObjectStore,
 } from "@agentpond/core";
 import { AgentPondDuckDb } from "@agentpond/duckdb";
+import { manualTraceResourceSpans } from "./otel-trace.js";
+import {
+	writeEventsAndSyncCache,
+	writeOtelAndSyncCache,
+} from "./sync-write.js";
 
 type ParsedArgs = {
 	flags: Record<string, string | boolean>;
@@ -129,51 +130,14 @@ async function createTrace(
 	json: boolean,
 ): Promise<void> {
 	const now = new Date().toISOString();
-	const traceId = stringFlag(parsed, "id") ?? randomUUID();
-	const event: IngestionEvent = {
-		id: randomUUID(),
-		timestamp: now,
-		type: eventTypes.TRACE_CREATE,
-		body: {
-			id: traceId,
-			name: stringFlag(parsed, "name") ?? "manual trace",
-			userId: stringFlag(parsed, "userId"),
-			sessionId: stringFlag(parsed, "sessionId"),
-			metadata: jsonFlag(parsed, "metadata"),
-			input: jsonOrStringFlag(parsed, "input"),
-			output: jsonOrStringFlag(parsed, "output"),
-			startTime: now,
-			environment: "default",
-		},
-	};
-
+	const traceId = stringFlag(parsed, "id") ?? createOtelTraceId();
 	const store = new S3ObjectStore(config.s3);
-	const manifest = await writeEventsAndSyncCache(config, store, [event]);
-	print({ eventId: event.id, traceId, objects: manifest.objects }, json);
-}
-
-export async function writeEventsAndSyncCache(
-	config: Pick<AgentPondConfig, "dbPath" | "projectId" | "s3">,
-	store: ObjectStore,
-	events: IngestionEvent[],
-): Promise<BatchManifest> {
-	const writer = new AcceptedEventWriter({
+	const object = await writeOtelAndSyncCache(
+		config,
 		store,
-		projectId: config.projectId,
-		prefix: config.s3.prefix,
-	});
-	const manifest = await writer.writeAcceptedEvents(events);
-	const db = new AgentPondDuckDb(config.dbPath);
-	try {
-		await db.syncFromStore({
-			store,
-			projectId: config.projectId,
-			prefix: config.s3.prefix,
-		});
-	} finally {
-		await db.close();
-	}
-	return manifest;
+		manualTraceResourceSpans(parsed, traceId, now),
+	);
+	print({ traceId, object }, json);
 }
 
 async function runReadCommand(
@@ -265,29 +229,6 @@ function requiredFlag(parsed: ParsedArgs, name: string): string {
 	return value;
 }
 
-function jsonFlag(
-	parsed: ParsedArgs,
-	name: string,
-): Record<string, unknown> | undefined {
-	const raw = stringFlag(parsed, name);
-	if (!raw) return undefined;
-	const value = JSON.parse(raw) as unknown;
-	if (!value || typeof value !== "object" || Array.isArray(value)) {
-		throw new CliError(`--${name} must be a JSON object`);
-	}
-	return value as Record<string, unknown>;
-}
-
-function jsonOrStringFlag(parsed: ParsedArgs, name: string): unknown {
-	const raw = stringFlag(parsed, name);
-	if (!raw) return undefined;
-	try {
-		return JSON.parse(raw) as unknown;
-	} catch {
-		return raw;
-	}
-}
-
 function limit(parsed: ParsedArgs): number {
 	const raw = stringFlag(parsed, "limit");
 	if (!raw) return 100;
@@ -304,6 +245,10 @@ function parseScoreValue(value: string): string | number | boolean {
 	return Number.isNaN(numeric) ? value : numeric;
 }
 
+export function createOtelTraceId(): string {
+	return randomBytes(16).toString("hex");
+}
+
 function sql(value: string): string {
 	return `'${value.replaceAll("'", "''")}'`;
 }

apps/cli/src/otel-trace.ts

@@ -0,0 +1,117 @@
+import { randomBytes } from "node:crypto";
+
+type ParsedArgs = {
+	flags: Record<string, string | boolean>;
+	positionals: string[];
+};
+
+export function manualTraceResourceSpans(
+	parsed: ParsedArgs,
+	traceId: string,
+	timestamp: string,
+): unknown[] {
+	const name = stringFlag(parsed, "name") ?? "manual trace";
+	return [
+		{
+			scopeSpans: [
+				{
+					spans: [
+						{
+							traceId,
+							spanId: randomBytes(8).toString("hex"),
+							name,
+							startTimeUnixNano: isoToUnixNanos(timestamp),
+							endTimeUnixNano: isoToUnixNanos(timestamp),
+							attributes: traceCreateAttributes(parsed, name),
+						},
+					],
+				},
+			],
+		},
+	];
+}
+
+function traceCreateAttributes(
+	parsed: ParsedArgs,
+	name: string,
+): Array<Record<string, unknown>> {
+	const attributes: Array<Record<string, unknown>> = [
+		otelAttr("langfuse.observation.type", "span"),
+		otelAttr("langfuse.trace.name", name),
+		otelAttr("langfuse.environment", "default"),
+	];
+	const userId = stringFlag(parsed, "userId");
+	if (userId) attributes.push(otelAttr("user.id", userId));
+	const sessionId = stringFlag(parsed, "sessionId");
+	if (sessionId) attributes.push(otelAttr("session.id", sessionId));
+	const input = jsonOrStringFlag(parsed, "input");
+	if (input !== undefined) {
+		attributes.push(otelAttr("langfuse.trace.input", input));
+		attributes.push(otelAttr("langfuse.observation.input", input));
+	}
+	const output = jsonOrStringFlag(parsed, "output");
+	if (output !== undefined) {
+		attributes.push(otelAttr("langfuse.trace.output", output));
+		attributes.push(otelAttr("langfuse.observation.output", output));
+	}
+	const metadata = jsonFlag(parsed, "metadata");
+	if (metadata) {
+		for (const [key, value] of Object.entries(metadata)) {
+			attributes.push(otelAttr(`langfuse.trace.metadata.${key}`, value));
+		}
+	}
+	return attributes;
+}
+
+function otelAttr(key: string, value: unknown): Record<string, unknown> {
+	if (typeof value === "boolean") return { key, value: { boolValue: value } };
+	if (typeof value === "number") return { key, value: { doubleValue: value } };
+	if (Array.isArray(value)) {
+		return {
+			key,
+			value: {
+				arrayValue: {
+					values: value.map((item) => ({ stringValue: String(item) })),
+				},
+			},
+		};
+	}
+	return {
+		key,
+		value: {
+			stringValue: typeof value === "string" ? value : JSON.stringify(value),
+		},
+	};
+}
+
+function stringFlag(parsed: ParsedArgs, name: string): string | undefined {
+	const value = parsed.flags[name];
+	return typeof value === "string" ? value : undefined;
+}
+
+function jsonFlag(
+	parsed: ParsedArgs,
+	name: string,
+): Record<string, unknown> | undefined {
+	const raw = stringFlag(parsed, name);
+	if (!raw) return undefined;
+	const value = JSON.parse(raw) as unknown;
+	if (!value || typeof value !== "object" || Array.isArray(value)) {
+		throw new Error(`--${name} must be a JSON object`);
+	}
+	return value as Record<string, unknown>;
+}
+
+function jsonOrStringFlag(parsed: ParsedArgs, name: string): unknown {
+	const raw = stringFlag(parsed, name);
+	if (!raw) return undefined;
+	try {
+		return JSON.parse(raw) as unknown;
+	} catch {
+		return raw;
+	}
+}
+
+function isoToUnixNanos(value: string): string {
+	return `${BigInt(Date.parse(value)) * 1_000_000n}`;
+}

apps/cli/src/sync-write.ts

@@ -0,0 +1,56 @@
+import {
+	AcceptedEventWriter,
+	type AgentPondConfig,
+	type BatchManifest,
+	type IngestionEvent,
+	type ObjectStore,
+} from "@agentpond/core";
+import { AgentPondDuckDb } from "@agentpond/duckdb";
+
+export async function writeOtelAndSyncCache(
+	config: Pick<AgentPondConfig, "dbPath" | "projectId" | "s3">,
+	store: ObjectStore,
+	resourceSpans: unknown[],
+) {
+	const writer = new AcceptedEventWriter({
+		store,
+		projectId: config.projectId,
+		prefix: config.s3.prefix,
+	});
+	const object = await writer.writeOtelResourceSpans(resourceSpans);
+	const db = new AgentPondDuckDb(config.dbPath);
+	try {
+		await db.syncFromStore({
+			store,
+			projectId: config.projectId,
+			prefix: config.s3.prefix,
+		});
+	} finally {
+		await db.close();
+	}
+	return object;
+}
+
+export async function writeEventsAndSyncCache(
+	config: Pick<AgentPondConfig, "dbPath" | "projectId" | "s3">,
+	store: ObjectStore,
+	events: IngestionEvent[],
+): Promise<BatchManifest> {
+	const writer = new AcceptedEventWriter({
+		store,
+		projectId: config.projectId,
+		prefix: config.s3.prefix,
+	});
+	const manifest = await writer.writeAcceptedEvents(events);
+	const db = new AgentPondDuckDb(config.dbPath);
+	try {
+		await db.syncFromStore({
+			store,
+			projectId: config.projectId,
+			prefix: config.s3.prefix,
+		});
+	} finally {
+		await db.close();
+	}
+	return manifest;
+}

docs/agentpond-sync.md

@@ -2,29 +2,39 @@
 
 ## Goal
 
-AgentPond stores raw ingestion events in a remote object storage (e.g. S3) and uses DuckDB as a local query cache. Object storage is the durable source of truth; DuckDB is the materialized analysis layer.
+AgentPond stores raw OTEL payloads and Langfuse-compatible score ingestion events in remote object storage (e.g. S3) and uses DuckDB as a local query cache. Object storage is the durable source of truth; DuckDB is the materialized analysis layer.
 
 ## Write Path
 
-Ingestion validates incoming events before accepting them. Accepted events are grouped by entity, such as trace, observation, score, or event.
+OTEL ingestion validates and decodes trace export requests, then stores raw `resourceSpans` under UTC minute buckets:
 
-Each group is written as a JSON object under the configured object-store prefix and project ID. With `AGENTPOND_S3_PREFIX=archive/` and `AGENTPOND_PROJECT_ID=project-a`, trace objects are written under paths like `archive/project-a/trace/<trace-id>/<event-id>.json`.
+```txt
+<prefix>otel/<project-id>/<yyyy>/<mm>/<dd>/<hh>/<min>/<batch-id>.json
+```
 
-A batch manifest is then written under the same prefix and project ID, in `archive/project-a/manifests/...`, and references all entity objects in the accepted batch.
+Non-OTEL ingestion remains for Langfuse SDK scores. Accepted non-OTEL events are grouped by entity, written under the configured project prefix, and referenced by a UTC minute-bucketed manifest:
+
+```txt
+<prefix><project-id>/score/<score-id>/<event-id>.json
+<prefix><project-id>/manifests/<yyyy>/<mm>/<dd>/<hh>/<min>/<batch-id>.json
+```
 
 ## Sync Flow
 
-`agentpond sync` lists manifests for the configured object-store prefix and project ID.
+`agentpond sync` scans UTC bucket windows for both sources:
+
+- OTEL objects are read directly from `otel/<project-id>/...` and normalized during sync.
+- Non-OTEL manifests are read from `<project-id>/manifests/...`; sync then reads their referenced event objects.
 
-For each new manifest, sync reads the referenced event objects, writes every event to `events_raw` table in DuckDB, and projects typed rows into `traces`, `observations`, and `scores` tables.
+Every normalized event is written to `events_raw` and projected into `traces`, `observations`, and `scores`.
 
 The `sessions` relation is a DuckDB view derived from traces with session IDs.
 
 ## Idempotency
 
-DuckDB tracks imported manifests in `processed_manifests` and imported event objects in `processed_objects`.
+DuckDB tracks imported OTEL objects and non-OTEL event objects in `processed_objects`. Non-OTEL manifests are tracked in `processed_manifests`.
 
-Running sync repeatedly is safe: already processed manifests and objects are skipped.
+DuckDB also stores per-source UTC bucket watermarks. The first sync scans all current-layout source keys; later syncs rescan recent buckets for late writes and skip already processed object or manifest keys.
 
 ## Query Model
 

docs/cli.md

@@ -50,7 +50,7 @@ By default, AgentPond stores its DuckDB cache at `./.agentpond/cache.duckdb` in
 
 ## Sync
 
-Sync reads accepted event manifests from object storage and projects them into the local DuckDB cache:
+Sync scans UTC object-storage buckets for OTEL trace payloads and non-OTEL score manifests, then projects new data into the local DuckDB cache:
 
 ```sh
 agentpond sync
@@ -64,7 +64,7 @@ agentpond sync --json
 
 ## Traces
 
-Create a manual trace:
+Create a manual trace. The CLI writes this as a Langfuse-compatible OTEL root span:
 
 ```sh
 agentpond traces create \
@@ -113,7 +113,7 @@ agentpond sessions get <session-id>
 
 ## Scores
 
-Create a score for a trace:
+Create a score for a trace. Scores use the same non-OTEL `score-create` event shape as the Langfuse SDK:
 
 ```sh
 agentpond scores create \

docs/langfuse-ingestion-differences.md

@@ -14,7 +14,7 @@ This design supports a continuously running production service with asynchronous
 
 ## Main Differences
 
-- AgentPond uses manifest-based pull sync instead of queue-driven workers.
+- AgentPond uses UTC bucket pull sync instead of queue-driven workers. OTEL trace payloads are discovered directly from UTC object buckets; non-OTEL events use time-bucketed manifests for discovery.
 - AgentPond uses local DuckDB instead of requiring a hosted ClickHouse service for analysis.
 - AgentPond keeps the raw event model in object storage, so the analysis layer can evolve independently.