Skip to content

Commit 8e5de51

Browse files
marcusschiessermarcusschiesser
authored
Use UTC bucket sync for OTEL payloads and score manifests (#7)
1 parent fe4143a commit 8e5de51

25 files changed

Lines changed: 733 additions & 212 deletions

‎.changeset/utc-bucket-sync.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"agentpond": patch
3+
---
4+
5+
Use UTC time-bucket sync for OTEL payloads and non-OTEL score manifests.

‎.github/workflows/ci.yml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,5 +32,8 @@ jobs:
3232
- name: Run Biome
3333
run: pnpm run biome:ci
3434

35+
- name: Run typecheck
36+
run: pnpm typecheck
37+
3538
- name: Run tests
3639
run: pnpm test

‎README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ This gives you durable remote storage without requiring an always-on analytical
3131

3232
- Langfuse-compatible ingestion endpoints for SDK and OTLP traces
3333
- S3-compatible raw event storage
34-
- Manifest-based discovery and incremental synchronization
34+
- UTC bucket discovery and incremental synchronization
3535
- Local DuckDB cache containing:
3636
- `events_raw`
3737
- `traces`

‎apps/cli/src/index.ts‎

Lines changed: 16 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -1,19 +1,20 @@
11
#!/usr/bin/env node
2-
import { randomUUID } from "node:crypto";
2+
import { randomBytes, randomUUID } from "node:crypto";
33
import { realpathSync } from "node:fs";
44
import { fileURLToPath, pathToFileURL } from "node:url";
55
import {
6-
AcceptedEventWriter,
7-
type AgentPondConfig,
8-
type BatchManifest,
96
configFromEnv,
107
eventTypes,
118
type IngestionEvent,
129
loadEnvFile,
13-
type ObjectStore,
1410
S3ObjectStore,
1511
} from "@agentpond/core";
1612
import { AgentPondDuckDb } from "@agentpond/duckdb";
13+
import { manualTraceResourceSpans } from "./otel-trace.js";
14+
import {
15+
writeEventsAndSyncCache,
16+
writeOtelAndSyncCache,
17+
} from "./sync-write.js";
1718

1819
type ParsedArgs = {
1920
flags: Record<string, string | boolean>;
@@ -129,51 +130,14 @@ async function createTrace(
129130
json: boolean,
130131
): Promise<void> {
131132
const now = new Date().toISOString();
132-
const traceId = stringFlag(parsed, "id") ?? randomUUID();
133-
const event: IngestionEvent = {
134-
id: randomUUID(),
135-
timestamp: now,
136-
type: eventTypes.TRACE_CREATE,
137-
body: {
138-
id: traceId,
139-
name: stringFlag(parsed, "name") ?? "manual trace",
140-
userId: stringFlag(parsed, "userId"),
141-
sessionId: stringFlag(parsed, "sessionId"),
142-
metadata: jsonFlag(parsed, "metadata"),
143-
input: jsonOrStringFlag(parsed, "input"),
144-
output: jsonOrStringFlag(parsed, "output"),
145-
startTime: now,
146-
environment: "default",
147-
},
148-
};
149-
133+
const traceId = stringFlag(parsed, "id") ?? createOtelTraceId();
150134
const store = new S3ObjectStore(config.s3);
151-
const manifest = await writeEventsAndSyncCache(config, store, [event]);
152-
print({ eventId: event.id, traceId, objects: manifest.objects }, json);
153-
}
154-
155-
export async function writeEventsAndSyncCache(
156-
config: Pick<AgentPondConfig, "dbPath" | "projectId" | "s3">,
157-
store: ObjectStore,
158-
events: IngestionEvent[],
159-
): Promise<BatchManifest> {
160-
const writer = new AcceptedEventWriter({
135+
const object = await writeOtelAndSyncCache(
136+
config,
161137
store,
162-
projectId: config.projectId,
163-
prefix: config.s3.prefix,
164-
});
165-
const manifest = await writer.writeAcceptedEvents(events);
166-
const db = new AgentPondDuckDb(config.dbPath);
167-
try {
168-
await db.syncFromStore({
169-
store,
170-
projectId: config.projectId,
171-
prefix: config.s3.prefix,
172-
});
173-
} finally {
174-
await db.close();
175-
}
176-
return manifest;
138+
manualTraceResourceSpans(parsed, traceId, now),
139+
);
140+
print({ traceId, object }, json);
177141
}
178142

179143
async function runReadCommand(
@@ -265,29 +229,6 @@ function requiredFlag(parsed: ParsedArgs, name: string): string {
265229
return value;
266230
}
267231

268-
function jsonFlag(
269-
parsed: ParsedArgs,
270-
name: string,
271-
): Record<string, unknown> | undefined {
272-
const raw = stringFlag(parsed, name);
273-
if (!raw) return undefined;
274-
const value = JSON.parse(raw) as unknown;
275-
if (!value || typeof value !== "object" || Array.isArray(value)) {
276-
throw new CliError(`--${name} must be a JSON object`);
277-
}
278-
return value as Record<string, unknown>;
279-
}
280-
281-
function jsonOrStringFlag(parsed: ParsedArgs, name: string): unknown {
282-
const raw = stringFlag(parsed, name);
283-
if (!raw) return undefined;
284-
try {
285-
return JSON.parse(raw) as unknown;
286-
} catch {
287-
return raw;
288-
}
289-
}
290-
291232
function limit(parsed: ParsedArgs): number {
292233
const raw = stringFlag(parsed, "limit");
293234
if (!raw) return 100;
@@ -304,6 +245,10 @@ function parseScoreValue(value: string): string | number | boolean {
304245
return Number.isNaN(numeric) ? value : numeric;
305246
}
306247

248+
export function createOtelTraceId(): string {
249+
return randomBytes(16).toString("hex");
250+
}
251+
307252
function sql(value: string): string {
308253
return `'${value.replaceAll("'", "''")}'`;
309254
}

‎apps/cli/src/otel-trace.ts‎

Lines changed: 117 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,117 @@
1+
import { randomBytes } from "node:crypto";
2+
3+
type ParsedArgs = {
4+
flags: Record<string, string | boolean>;
5+
positionals: string[];
6+
};
7+
8+
export function manualTraceResourceSpans(
9+
parsed: ParsedArgs,
10+
traceId: string,
11+
timestamp: string,
12+
): unknown[] {
13+
const name = stringFlag(parsed, "name") ?? "manual trace";
14+
return [
15+
{
16+
scopeSpans: [
17+
{
18+
spans: [
19+
{
20+
traceId,
21+
spanId: randomBytes(8).toString("hex"),
22+
name,
23+
startTimeUnixNano: isoToUnixNanos(timestamp),
24+
endTimeUnixNano: isoToUnixNanos(timestamp),
25+
attributes: traceCreateAttributes(parsed, name),
26+
},
27+
],
28+
},
29+
],
30+
},
31+
];
32+
}
33+
34+
function traceCreateAttributes(
35+
parsed: ParsedArgs,
36+
name: string,
37+
): Array<Record<string, unknown>> {
38+
const attributes: Array<Record<string, unknown>> = [
39+
otelAttr("langfuse.observation.type", "span"),
40+
otelAttr("langfuse.trace.name", name),
41+
otelAttr("langfuse.environment", "default"),
42+
];
43+
const userId = stringFlag(parsed, "userId");
44+
if (userId) attributes.push(otelAttr("user.id", userId));
45+
const sessionId = stringFlag(parsed, "sessionId");
46+
if (sessionId) attributes.push(otelAttr("session.id", sessionId));
47+
const input = jsonOrStringFlag(parsed, "input");
48+
if (input !== undefined) {
49+
attributes.push(otelAttr("langfuse.trace.input", input));
50+
attributes.push(otelAttr("langfuse.observation.input", input));
51+
}
52+
const output = jsonOrStringFlag(parsed, "output");
53+
if (output !== undefined) {
54+
attributes.push(otelAttr("langfuse.trace.output", output));
55+
attributes.push(otelAttr("langfuse.observation.output", output));
56+
}
57+
const metadata = jsonFlag(parsed, "metadata");
58+
if (metadata) {
59+
for (const [key, value] of Object.entries(metadata)) {
60+
attributes.push(otelAttr(`langfuse.trace.metadata.${key}`, value));
61+
}
62+
}
63+
return attributes;
64+
}
65+
66+
function otelAttr(key: string, value: unknown): Record<string, unknown> {
67+
if (typeof value === "boolean") return { key, value: { boolValue: value } };
68+
if (typeof value === "number") return { key, value: { doubleValue: value } };
69+
if (Array.isArray(value)) {
70+
return {
71+
key,
72+
value: {
73+
arrayValue: {
74+
values: value.map((item) => ({ stringValue: String(item) })),
75+
},
76+
},
77+
};
78+
}
79+
return {
80+
key,
81+
value: {
82+
stringValue: typeof value === "string" ? value : JSON.stringify(value),
83+
},
84+
};
85+
}
86+
87+
function stringFlag(parsed: ParsedArgs, name: string): string | undefined {
88+
const value = parsed.flags[name];
89+
return typeof value === "string" ? value : undefined;
90+
}
91+
92+
function jsonFlag(
93+
parsed: ParsedArgs,
94+
name: string,
95+
): Record<string, unknown> | undefined {
96+
const raw = stringFlag(parsed, name);
97+
if (!raw) return undefined;
98+
const value = JSON.parse(raw) as unknown;
99+
if (!value || typeof value !== "object" || Array.isArray(value)) {
100+
throw new Error(`--${name} must be a JSON object`);
101+
}
102+
return value as Record<string, unknown>;
103+
}
104+
105+
function jsonOrStringFlag(parsed: ParsedArgs, name: string): unknown {
106+
const raw = stringFlag(parsed, name);
107+
if (!raw) return undefined;
108+
try {
109+
return JSON.parse(raw) as unknown;
110+
} catch {
111+
return raw;
112+
}
113+
}
114+
115+
function isoToUnixNanos(value: string): string {
116+
return `${BigInt(Date.parse(value)) * 1_000_000n}`;
117+
}

‎apps/cli/src/sync-write.ts‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
import {
2+
AcceptedEventWriter,
3+
type AgentPondConfig,
4+
type BatchManifest,
5+
type IngestionEvent,
6+
type ObjectStore,
7+
} from "@agentpond/core";
8+
import { AgentPondDuckDb } from "@agentpond/duckdb";
9+
10+
export async function writeOtelAndSyncCache(
11+
config: Pick<AgentPondConfig, "dbPath" | "projectId" | "s3">,
12+
store: ObjectStore,
13+
resourceSpans: unknown[],
14+
) {
15+
const writer = new AcceptedEventWriter({
16+
store,
17+
projectId: config.projectId,
18+
prefix: config.s3.prefix,
19+
});
20+
const object = await writer.writeOtelResourceSpans(resourceSpans);
21+
const db = new AgentPondDuckDb(config.dbPath);
22+
try {
23+
await db.syncFromStore({
24+
store,
25+
projectId: config.projectId,
26+
prefix: config.s3.prefix,
27+
});
28+
} finally {
29+
await db.close();
30+
}
31+
return object;
32+
}
33+
34+
export async function writeEventsAndSyncCache(
35+
config: Pick<AgentPondConfig, "dbPath" | "projectId" | "s3">,
36+
store: ObjectStore,
37+
events: IngestionEvent[],
38+
): Promise<BatchManifest> {
39+
const writer = new AcceptedEventWriter({
40+
store,
41+
projectId: config.projectId,
42+
prefix: config.s3.prefix,
43+
});
44+
const manifest = await writer.writeAcceptedEvents(events);
45+
const db = new AgentPondDuckDb(config.dbPath);
46+
try {
47+
await db.syncFromStore({
48+
store,
49+
projectId: config.projectId,
50+
prefix: config.s3.prefix,
51+
});
52+
} finally {
53+
await db.close();
54+
}
55+
return manifest;
56+
}

‎docs/agentpond-sync.md‎

Lines changed: 18 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2,29 +2,39 @@
22

33
## Goal
44

5-
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.
5+
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.
66

77
## Write Path
88

9-
Ingestion validates incoming events before accepting them. Accepted events are grouped by entity, such as trace, observation, score, or event.
9+
OTEL ingestion validates and decodes trace export requests, then stores raw `resourceSpans` under UTC minute buckets:
1010

11-
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`.
11+
```txt
12+
<prefix>otel/<project-id>/<yyyy>/<mm>/<dd>/<hh>/<min>/<batch-id>.json
13+
```
1214

13-
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.
15+
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:
16+
17+
```txt
18+
<prefix><project-id>/score/<score-id>/<event-id>.json
19+
<prefix><project-id>/manifests/<yyyy>/<mm>/<dd>/<hh>/<min>/<batch-id>.json
20+
```
1421

1522
## Sync Flow
1623

17-
`agentpond sync` lists manifests for the configured object-store prefix and project ID.
24+
`agentpond sync` scans UTC bucket windows for both sources:
25+
26+
- OTEL objects are read directly from `otel/<project-id>/...` and normalized during sync.
27+
- Non-OTEL manifests are read from `<project-id>/manifests/...`; sync then reads their referenced event objects.
1828

19-
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.
29+
Every normalized event is written to `events_raw` and projected into `traces`, `observations`, and `scores`.
2030

2131
The `sessions` relation is a DuckDB view derived from traces with session IDs.
2232

2333
## Idempotency
2434

25-
DuckDB tracks imported manifests in `processed_manifests` and imported event objects in `processed_objects`.
35+
DuckDB tracks imported OTEL objects and non-OTEL event objects in `processed_objects`. Non-OTEL manifests are tracked in `processed_manifests`.
2636

27-
Running sync repeatedly is safe: already processed manifests and objects are skipped.
37+
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.
2838

2939
## Query Model
3040

‎docs/cli.md‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ By default, AgentPond stores its DuckDB cache at `./.agentpond/cache.duckdb` in
5050

5151
## Sync
5252

53-
Sync reads accepted event manifests from object storage and projects them into the local DuckDB cache:
53+
Sync scans UTC object-storage buckets for OTEL trace payloads and non-OTEL score manifests, then projects new data into the local DuckDB cache:
5454

5555
```sh
5656
agentpond sync
@@ -64,7 +64,7 @@ agentpond sync --json
6464

6565
## Traces
6666

67-
Create a manual trace:
67+
Create a manual trace. The CLI writes this as a Langfuse-compatible OTEL root span:
6868

6969
```sh
7070
agentpond traces create \
@@ -113,7 +113,7 @@ agentpond sessions get <session-id>
113113

114114
## Scores
115115

116-
Create a score for a trace:
116+
Create a score for a trace. Scores use the same non-OTEL `score-create` event shape as the Langfuse SDK:
117117

118118
```sh
119119
agentpond scores create \

0 commit comments

Comments
 (0)