diff --git a/migrations/0004_llm_trace_observability.sql b/migrations/0004_llm_trace_observability.sql new file mode 100644 index 0000000..4f34740 --- /dev/null +++ b/migrations/0004_llm_trace_observability.sql @@ -0,0 +1,48 @@ +CREATE TABLE IF NOT EXISTS llm_trace_runs ( + job_id TEXT PRIMARY KEY, + case_id TEXT, + status TEXT NOT NULL, + current_stage TEXT NOT NULL, + trace_completeness TEXT NOT NULL, + error_stage TEXT, + error_summary TEXT, + started_at TEXT NOT NULL, + finished_at TEXT, + updated_at TEXT NOT NULL, + FOREIGN KEY (job_id) REFERENCES article_jobs(id) ON DELETE CASCADE, + FOREIGN KEY (case_id) REFERENCES optimization_cases(id) ON DELETE SET NULL +); + +CREATE TABLE IF NOT EXISTS llm_trace_calls ( + call_id TEXT PRIMARY KEY, + job_id TEXT NOT NULL, + sequence INTEGER NOT NULL, + task TEXT NOT NULL, + workflow_stage TEXT NOT NULL, + rewrite_round INTEGER, + provider TEXT NOT NULL, + model TEXT NOT NULL, + status TEXT NOT NULL, + request_object_key TEXT, + response_object_key TEXT, + token_usage TEXT, + schema_name TEXT, + schema_valid INTEGER, + validation_issues TEXT NOT NULL DEFAULT '[]', + business_status TEXT, + duration_ms INTEGER, + started_at TEXT NOT NULL, + responded_at TEXT, + validated_at TEXT, + failed_at TEXT, + error_type TEXT, + error_summary TEXT, + UNIQUE (job_id, sequence), + FOREIGN KEY (job_id) REFERENCES llm_trace_runs(job_id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_llm_trace_runs_status_updated + ON llm_trace_runs(status, updated_at DESC); + +CREATE INDEX IF NOT EXISTS idx_llm_trace_calls_job_sequence + ON llm_trace_calls(job_id, sequence); diff --git a/src/lib/db/schema.ts b/src/lib/db/schema.ts index 565fd4b..4cf5085 100644 --- a/src/lib/db/schema.ts +++ b/src/lib/db/schema.ts @@ -115,6 +115,49 @@ export function initializeSchema(db: Database.Database) { foreign key (article_job_id) references article_jobs(id) on delete set null ); + create table if not exists llm_trace_runs ( + job_id text primary key, + case_id text, + status text not null, + current_stage text not null, + trace_completeness text not null, + error_stage text, + error_summary text, + started_at text not null, + finished_at text, + updated_at text not null, + foreign key (job_id) references article_jobs(id) on delete cascade, + foreign key (case_id) references optimization_cases(id) on delete set null + ); + + create table if not exists llm_trace_calls ( + call_id text primary key, + job_id text not null, + sequence integer not null, + task text not null, + workflow_stage text not null, + rewrite_round integer, + provider text not null, + model text not null, + status text not null, + request_object_key text, + response_object_key text, + token_usage text, + schema_name text, + schema_valid integer, + validation_issues text not null default '[]', + business_status text, + duration_ms integer, + started_at text not null, + responded_at text, + validated_at text, + failed_at text, + error_type text, + error_summary text, + unique (job_id, sequence), + foreign key (job_id) references llm_trace_runs(job_id) on delete cascade + ); + create table if not exists rubric_versions ( id text primary key, version text not null, @@ -204,6 +247,12 @@ export function initializeSchema(db: Database.Database) { create index if not exists idx_performance_snapshots_publication on performance_snapshots(publication_id); + + create index if not exists idx_llm_trace_runs_status_updated + on llm_trace_runs(status, updated_at desc); + + create index if not exists idx_llm_trace_calls_job_sequence + on llm_trace_calls(job_id, sequence); `); ensureColumn(db, "article_jobs", "case_id", "text"); diff --git a/src/lib/llm/__tests__/d1-trace-repository.test.ts b/src/lib/llm/__tests__/d1-trace-repository.test.ts new file mode 100644 index 0000000..13aeb0c --- /dev/null +++ b/src/lib/llm/__tests__/d1-trace-repository.test.ts @@ -0,0 +1,114 @@ +import { describe, expect, it, vi } from "vitest"; + +import { createD1TraceRepository } from "../d1-trace-repository"; +import type { LlmTraceCall, LlmTraceRun } from "../trace-types"; + +const run: LlmTraceRun = { + job_id: "job_1", + case_id: "case_1", + status: "running", + current_stage: "draft", + trace_completeness: "complete", + error_stage: null, + error_summary: null, + started_at: "2026-07-16T00:00:00.000Z", + finished_at: null, + updated_at: "2026-07-16T00:00:01.000Z", +}; + +const call: LlmTraceCall = { + call_id: "llmcall_1", + job_id: "job_1", + sequence: 1, + task: "article_optimizer", + workflow_stage: "draft", + rewrite_round: null, + provider: "deepseek", + model: "deepseek-v4-pro", + status: "validated", + request_object_key: "llm-traces/job_1/llmcall_1/request.json", + response_object_key: "llm-traces/job_1/llmcall_1/response.json", + token_usage: { prompt_tokens: 10, completion_tokens: 4, total_tokens: 14 }, + schema_name: "optimizedArticleSchema", + schema_valid: true, + validation_issues: ["title: required"], + business_status: null, + duration_ms: 1200, + started_at: "2026-07-16T00:00:00.000Z", + responded_at: "2026-07-16T00:00:01.000Z", + validated_at: "2026-07-16T00:00:01.200Z", + failed_at: null, + error_type: null, + error_summary: null, +}; + +describe("createD1TraceRepository", () => { + it("binds run and call snapshots with JSON fields serialized", async () => { + const runStatement = vi.fn().mockResolvedValue({ success: true }); + const bind = vi.fn().mockReturnValue({ run: runStatement }); + const prepare = vi.fn().mockReturnValue({ bind }); + const repository = createD1TraceRepository( + { prepare } as unknown as D1Database, + ); + + await repository.putRun(run); + await repository.putCall(call); + + expect(prepare).toHaveBeenNthCalledWith( + 1, + expect.stringContaining("insert into llm_trace_runs"), + ); + expect(prepare).toHaveBeenNthCalledWith( + 2, + expect.stringContaining("insert into llm_trace_calls"), + ); + expect(bind).toHaveBeenNthCalledWith( + 2, + call.call_id, + call.job_id, + call.sequence, + call.task, + call.workflow_stage, + null, + call.provider, + call.model, + call.status, + call.request_object_key, + call.response_object_key, + JSON.stringify(call.token_usage), + call.schema_name, + 1, + JSON.stringify(call.validation_issues), + null, + call.duration_ms, + call.started_at, + call.responded_at, + call.validated_at, + null, + null, + null, + ); + }); + + it("reads calls in sequence order and parses JSON fields", async () => { + const all = vi.fn().mockResolvedValue({ + results: [{ + ...call, + token_usage: JSON.stringify(call.token_usage), + schema_valid: 1, + validation_issues: JSON.stringify(call.validation_issues), + }], + }); + const bind = vi.fn().mockReturnValue({ all }); + const prepare = vi.fn().mockReturnValue({ bind }); + const repository = createD1TraceRepository( + { prepare } as unknown as D1Database, + ); + + await expect(repository.listCalls("job_1")).resolves.toEqual([call]); + expect(prepare).toHaveBeenCalledWith( + expect.stringContaining("order by sequence"), + ); + expect(bind).toHaveBeenCalledWith("job_1"); + }); +}); diff --git a/src/lib/llm/__tests__/trace-repository.test.ts b/src/lib/llm/__tests__/trace-repository.test.ts new file mode 100644 index 0000000..5c855cb --- /dev/null +++ b/src/lib/llm/__tests__/trace-repository.test.ts @@ -0,0 +1,99 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +import { createSqliteRepository } from "../../db/sqlite-repository"; +import type { LlmTraceCall, LlmTraceRun } from "../trace-types"; +import { createSqliteTraceRepository } from "../sqlite-trace-repository"; + +function runFixture(jobId: string): LlmTraceRun { + return { + job_id: jobId, + case_id: null, + status: "running", + current_stage: "fact_card", + trace_completeness: "complete", + error_stage: null, + error_summary: null, + started_at: "2026-07-16T00:00:00.000Z", + finished_at: null, + updated_at: "2026-07-16T00:00:00.000Z", + }; +} + +function callFixture( + jobId: string, + callId: string, + sequence: number, +): LlmTraceCall { + return { + call_id: callId, + job_id: jobId, + sequence, + task: sequence === 1 ? "fact_extractor" : "article_optimizer", + workflow_stage: sequence === 1 ? "fact_card" : "draft", + rewrite_round: null, + provider: "deepseek", + model: "deepseek-v4-pro", + status: "started", + request_object_key: `llm-traces/${jobId}/${callId}/request.json`, + response_object_key: null, + token_usage: null, + schema_name: null, + schema_valid: null, + validation_issues: [], + business_status: null, + duration_ms: null, + started_at: `2026-07-16T00:00:0${sequence}.000Z`, + responded_at: null, + validated_at: null, + failed_at: null, + error_type: null, + error_summary: null, + }; +} + +describe("createSqliteTraceRepository", () => { + let tempDir: string; + let dbPath: string; + + beforeEach(() => { + tempDir = mkdtempSync(join(tmpdir(), "geo-llm-traces-")); + dbPath = join(tempDir, "app.db"); + }); + + afterEach(() => { + rmSync(tempDir, { recursive: true, force: true }); + }); + + it("stores a run and ordered calls, then deletes only trace rows", async () => { + const appRepository = createSqliteRepository(dbPath); + const job = await appRepository.createArticleJob({ + source_title: "Title", + source_body: "Body", + image_inputs: [], + publish_platform: "official_site", + user_instructions: "", + }); + const repository = createSqliteTraceRepository(dbPath); + + await repository.putRun(runFixture(job.id)); + await repository.putCall(callFixture(job.id, "llmcall_1", 1)); + await repository.putCall(callFixture(job.id, "llmcall_2", 2)); + + await expect(repository.getLatestRun()).resolves.toMatchObject({ + job_id: job.id, + status: "running", + }); + await expect(repository.listCalls(job.id)).resolves.toEqual([ + expect.objectContaining({ call_id: "llmcall_1", sequence: 1 }), + expect.objectContaining({ call_id: "llmcall_2", sequence: 2 }), + ]); + + await repository.deleteRun(job.id); + await expect(appRepository.getArticleJob(job.id)).resolves.not.toBeNull(); + await expect(repository.getRun(job.id)).resolves.toBeNull(); + }); +}); diff --git a/src/lib/llm/d1-trace-repository.ts b/src/lib/llm/d1-trace-repository.ts new file mode 100644 index 0000000..b6739ce --- /dev/null +++ b/src/lib/llm/d1-trace-repository.ts @@ -0,0 +1,192 @@ +import type { LlmTraceRepository } from "./trace-repository"; +import type { + LlmBusinessStatus, + LlmProviderName, + LlmTaskName, + LlmTraceCall, + LlmTraceCallStatus, + LlmTraceCompleteness, + LlmTraceErrorType, + LlmTraceRun, + LlmTraceRunStatus, + LlmTraceWorkflowStage, +} from "./trace-types"; + +interface TraceRunRow { + job_id: string; + case_id: string | null; + status: LlmTraceRunStatus; + current_stage: LlmTraceWorkflowStage; + trace_completeness: LlmTraceCompleteness; + error_stage: string | null; + error_summary: string | null; + started_at: string; + finished_at: string | null; + updated_at: string; +} + +interface TraceCallRow { + call_id: string; + job_id: string; + sequence: number; + task: LlmTaskName; + workflow_stage: LlmTraceWorkflowStage; + rewrite_round: number | null; + provider: LlmProviderName; + model: string; + status: LlmTraceCallStatus; + request_object_key: string | null; + response_object_key: string | null; + token_usage: string | null; + schema_name: string | null; + schema_valid: 0 | 1 | null; + validation_issues: string; + business_status: LlmBusinessStatus | null; + duration_ms: number | null; + started_at: string; + responded_at: string | null; + validated_at: string | null; + failed_at: string | null; + error_type: LlmTraceErrorType | null; + error_summary: string | null; +} + +const putRunSql = ` + insert into llm_trace_runs ( + job_id, case_id, status, current_stage, trace_completeness, + error_stage, error_summary, started_at, finished_at, updated_at + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + on conflict(job_id) do update set + case_id = excluded.case_id, + status = excluded.status, + current_stage = excluded.current_stage, + trace_completeness = excluded.trace_completeness, + error_stage = excluded.error_stage, + error_summary = excluded.error_summary, + started_at = excluded.started_at, + finished_at = excluded.finished_at, + updated_at = excluded.updated_at +`; + +const putCallSql = ` + insert into llm_trace_calls ( + call_id, job_id, sequence, task, workflow_stage, rewrite_round, + provider, model, status, request_object_key, response_object_key, + token_usage, schema_name, schema_valid, validation_issues, + business_status, duration_ms, started_at, responded_at, validated_at, + failed_at, error_type, error_summary + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + on conflict(call_id) do update set + sequence = excluded.sequence, + task = excluded.task, + workflow_stage = excluded.workflow_stage, + rewrite_round = excluded.rewrite_round, + provider = excluded.provider, + model = excluded.model, + status = excluded.status, + request_object_key = excluded.request_object_key, + response_object_key = excluded.response_object_key, + token_usage = excluded.token_usage, + schema_name = excluded.schema_name, + schema_valid = excluded.schema_valid, + validation_issues = excluded.validation_issues, + business_status = excluded.business_status, + duration_ms = excluded.duration_ms, + started_at = excluded.started_at, + responded_at = excluded.responded_at, + validated_at = excluded.validated_at, + failed_at = excluded.failed_at, + error_type = excluded.error_type, + error_summary = excluded.error_summary +`; + +function toTraceCall(row: TraceCallRow): LlmTraceCall { + return { + ...row, + token_usage: row.token_usage == null + ? null + : JSON.parse(row.token_usage) as Record, + schema_valid: row.schema_valid == null ? null : row.schema_valid === 1, + validation_issues: JSON.parse(row.validation_issues) as string[], + }; +} + +export function createD1TraceRepository(db: D1Database): LlmTraceRepository { + return { + async putRun(run) { + await db.prepare(putRunSql).bind( + run.job_id, + run.case_id, + run.status, + run.current_stage, + run.trace_completeness, + run.error_stage, + run.error_summary, + run.started_at, + run.finished_at, + run.updated_at, + ).run(); + }, + async putCall(call) { + await db.prepare(putCallSql).bind( + call.call_id, + call.job_id, + call.sequence, + call.task, + call.workflow_stage, + call.rewrite_round, + call.provider, + call.model, + call.status, + call.request_object_key, + call.response_object_key, + call.token_usage == null ? null : JSON.stringify(call.token_usage), + call.schema_name, + call.schema_valid == null ? null : Number(call.schema_valid), + JSON.stringify(call.validation_issues), + call.business_status, + call.duration_ms, + call.started_at, + call.responded_at, + call.validated_at, + call.failed_at, + call.error_type, + call.error_summary, + ).run(); + }, + async getRun(jobId) { + return await db + .prepare("select * from llm_trace_runs where job_id = ?") + .bind(jobId) + .first() ?? null; + }, + async getLatestRun() { + return await db.prepare(` + select * from llm_trace_runs + order by case when status = 'running' then 0 else 1 end, updated_at desc + limit 1 + `).first() ?? null; + }, + async listCalls(jobId) { + const result = await db + .prepare("select * from llm_trace_calls where job_id = ? order by sequence") + .bind(jobId) + .all(); + return result.results.map(toTraceCall); + }, + async listTerminalRunsExcept(jobId) { + const result = await db.prepare(` + select * from llm_trace_runs + where status <> 'running' and job_id <> ? + order by finished_at desc, updated_at desc + `).bind(jobId).all(); + return result.results; + }, + async deleteRun(jobId) { + await db + .prepare("delete from llm_trace_runs where job_id = ?") + .bind(jobId) + .run(); + }, + }; +} diff --git a/src/lib/llm/sqlite-trace-repository.ts b/src/lib/llm/sqlite-trace-repository.ts new file mode 100644 index 0000000..6938faa --- /dev/null +++ b/src/lib/llm/sqlite-trace-repository.ts @@ -0,0 +1,195 @@ +import { createDatabase, getDefaultDatabasePath } from "../db/connection"; +import { initializeSchema } from "../db/schema"; +import type { LlmTraceRepository } from "./trace-repository"; +import type { + LlmBusinessStatus, + LlmProviderName, + LlmTaskName, + LlmTraceCall, + LlmTraceCallStatus, + LlmTraceCompleteness, + LlmTraceErrorType, + LlmTraceRun, + LlmTraceRunStatus, + LlmTraceWorkflowStage, +} from "./trace-types"; + +interface TraceRunRow { + job_id: string; + case_id: string | null; + status: LlmTraceRunStatus; + current_stage: LlmTraceWorkflowStage; + trace_completeness: LlmTraceCompleteness; + error_stage: string | null; + error_summary: string | null; + started_at: string; + finished_at: string | null; + updated_at: string; +} + +interface TraceCallRow { + call_id: string; + job_id: string; + sequence: number; + task: LlmTaskName; + workflow_stage: LlmTraceWorkflowStage; + rewrite_round: number | null; + provider: LlmProviderName; + model: string; + status: LlmTraceCallStatus; + request_object_key: string | null; + response_object_key: string | null; + token_usage: string | null; + schema_name: string | null; + schema_valid: 0 | 1 | null; + validation_issues: string; + business_status: LlmBusinessStatus | null; + duration_ms: number | null; + started_at: string; + responded_at: string | null; + validated_at: string | null; + failed_at: string | null; + error_type: LlmTraceErrorType | null; + error_summary: string | null; +} + +const putRunSql = ` + insert into llm_trace_runs ( + job_id, case_id, status, current_stage, trace_completeness, + error_stage, error_summary, started_at, finished_at, updated_at + ) values ( + @job_id, @case_id, @status, @current_stage, @trace_completeness, + @error_stage, @error_summary, @started_at, @finished_at, @updated_at + ) + on conflict(job_id) do update set + case_id = excluded.case_id, + status = excluded.status, + current_stage = excluded.current_stage, + trace_completeness = excluded.trace_completeness, + error_stage = excluded.error_stage, + error_summary = excluded.error_summary, + started_at = excluded.started_at, + finished_at = excluded.finished_at, + updated_at = excluded.updated_at +`; + +const putCallSql = ` + insert into llm_trace_calls ( + call_id, job_id, sequence, task, workflow_stage, rewrite_round, + provider, model, status, request_object_key, response_object_key, + token_usage, schema_name, schema_valid, validation_issues, + business_status, duration_ms, started_at, responded_at, validated_at, + failed_at, error_type, error_summary + ) values ( + @call_id, @job_id, @sequence, @task, @workflow_stage, @rewrite_round, + @provider, @model, @status, @request_object_key, @response_object_key, + @token_usage, @schema_name, @schema_valid, @validation_issues, + @business_status, @duration_ms, @started_at, @responded_at, @validated_at, + @failed_at, @error_type, @error_summary + ) + on conflict(call_id) do update set + sequence = excluded.sequence, + task = excluded.task, + workflow_stage = excluded.workflow_stage, + rewrite_round = excluded.rewrite_round, + provider = excluded.provider, + model = excluded.model, + status = excluded.status, + request_object_key = excluded.request_object_key, + response_object_key = excluded.response_object_key, + token_usage = excluded.token_usage, + schema_name = excluded.schema_name, + schema_valid = excluded.schema_valid, + validation_issues = excluded.validation_issues, + business_status = excluded.business_status, + duration_ms = excluded.duration_ms, + started_at = excluded.started_at, + responded_at = excluded.responded_at, + validated_at = excluded.validated_at, + failed_at = excluded.failed_at, + error_type = excluded.error_type, + error_summary = excluded.error_summary +`; + +function toCallParams(call: LlmTraceCall) { + return { + ...call, + token_usage: call.token_usage == null ? null : JSON.stringify(call.token_usage), + schema_valid: call.schema_valid == null ? null : Number(call.schema_valid), + validation_issues: JSON.stringify(call.validation_issues), + }; +} + +function toTraceCall(row: TraceCallRow): LlmTraceCall { + return { + ...row, + token_usage: row.token_usage == null + ? null + : JSON.parse(row.token_usage) as Record, + schema_valid: row.schema_valid == null ? null : row.schema_valid === 1, + validation_issues: JSON.parse(row.validation_issues) as string[], + }; +} + +export function createSqliteTraceRepository( + dbPath = getDefaultDatabasePath(), +): LlmTraceRepository { + function withDb(action: (db: ReturnType) => T) { + const db = createDatabase(dbPath); + initializeSchema(db); + try { + return action(db); + } finally { + db.close(); + } + } + + return { + async putRun(run) { + withDb((db) => db.prepare(putRunSql).run(run)); + }, + async putCall(call) { + withDb((db) => db.prepare(putCallSql).run(toCallParams(call))); + }, + async getRun(jobId) { + return withDb((db) => { + const row = db + .prepare("select * from llm_trace_runs where job_id = ?") + .get(jobId) as TraceRunRow | undefined; + return row ?? null; + }); + }, + async getLatestRun() { + return withDb((db) => { + const row = db.prepare(` + select * from llm_trace_runs + order by case when status = 'running' then 0 else 1 end, updated_at desc + limit 1 + `).get() as TraceRunRow | undefined; + return row ?? null; + }); + }, + async listCalls(jobId) { + return withDb((db) => { + const rows = db + .prepare("select * from llm_trace_calls where job_id = ? order by sequence") + .all(jobId) as TraceCallRow[]; + return rows.map(toTraceCall); + }); + }, + async listTerminalRunsExcept(jobId) { + return withDb((db) => db + .prepare(` + select * from llm_trace_runs + where status <> 'running' and job_id <> ? + order by finished_at desc, updated_at desc + `) + .all(jobId) as TraceRunRow[]); + }, + async deleteRun(jobId) { + withDb((db) => db + .prepare("delete from llm_trace_runs where job_id = ?") + .run(jobId)); + }, + }; +} diff --git a/src/lib/llm/trace-repository.ts b/src/lib/llm/trace-repository.ts new file mode 100644 index 0000000..1986735 --- /dev/null +++ b/src/lib/llm/trace-repository.ts @@ -0,0 +1,24 @@ +import { getDefaultDatabasePath } from "../db/connection"; +import { getAppCloudflareEnv } from "../runtime/cloudflare"; +import { createD1TraceRepository } from "./d1-trace-repository"; +import { createSqliteTraceRepository } from "./sqlite-trace-repository"; +import type { LlmTraceCall, LlmTraceRun } from "./trace-types"; + +export interface LlmTraceRepository { + putRun(run: LlmTraceRun): Promise; + putCall(call: LlmTraceCall): Promise; + getRun(jobId: string): Promise; + getLatestRun(): Promise; + listCalls(jobId: string): Promise; + listTerminalRunsExcept(jobId: string): Promise; + deleteRun(jobId: string): Promise; +} + +export function getLlmTraceRepositoryFromRuntime(): LlmTraceRepository { + if (process.env.APP_RUNTIME === "cloudflare") { + const env = getAppCloudflareEnv(); + if (!env?.DB) throw new Error("Cloudflare D1 binding DB is required"); + return createD1TraceRepository(env.DB); + } + return createSqliteTraceRepository(getDefaultDatabasePath()); +}