From c2d707506cc17b209faae46ccdac3994f0bf9aef Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 29 Oct 2026 00:01:44 +0300 Subject: [PATCH 1/2] feat(observe): annotation queues Adds annotation_queues/annotation_items (migration 6) and the /api/v1/queues routes: create/list/get/delete, add trace ids, next (oldest pending, 204 when none, no lock) and complete, which writes human scores through a shared writeHumanScore helper (also used by POST /api/v1/scores) in one transaction with the status update. transaction() now joins an already-open transaction on the same db. Co-Authored-By: Claude Opus 5.5 --- docs/observability.md | 50 ++++++ llms-full.txt | 50 ++++++ src/observe/humanScores.ts | 64 ++++++++ src/observe/queues.ts | 205 +++++++++++++++++++++++ src/observe/routes/index.ts | 4 +- src/observe/routes/queues.test.ts | 265 ++++++++++++++++++++++++++++++ src/observe/routes/queues.ts | 216 ++++++++++++++++++++++++ src/observe/routes/scores.ts | 37 ++--- src/observe/schema.ts | 25 +++ src/observe/store.test.ts | 18 ++ src/observe/store.ts | 13 +- 11 files changed, 918 insertions(+), 29 deletions(-) create mode 100644 src/observe/humanScores.ts create mode 100644 src/observe/queues.ts create mode 100644 src/observe/routes/queues.test.ts create mode 100644 src/observe/routes/queues.ts diff --git a/docs/observability.md b/docs/observability.md index b8d3b3cb..747e3722 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -618,6 +618,56 @@ An unknown field, an unknown `kind` or a `config` that does not match the `404` with the same code. `updated_at` moves on every `PUT`. The body may be up to 128 KiB. Retention never deletes rules. +### Annotation queues + +An annotation queue is a named list of traces for a person to review one by +one. Completing an item writes its scores as human scores, the same +`source: "human"` rows `POST /api/v1/scores` writes, so they show up in +`GET /api/v1/scores` and on the trace. The routes take the same token as +ingest: + +| Route | Answer | +| --- | --- | +| `POST /api/v1/queues` `{ "name", "rubric"? }` | `201 { queue }`. A taken name is a `409`. | +| `GET /api/v1/queues?limit&offset` | `{ queues, total, limit, offset }`, newest first. `limit` defaults to 100 (max 1000). | +| `GET /api/v1/queues/:id` | `{ queue }`. | +| `DELETE /api/v1/queues/:id` | `204`. The queue's items go with it; scores already written stay. | +| `POST /api/v1/queues/:id/items` `{ "trace_ids": [...] }` | `201 { added, skipped_existing }`. Takes 1-500 ids, queued in the order given. Every id must name a stored trace, or the answer is a `404` naming the first unknown one and nothing is added. Ids already in the queue are listed in `skipped_existing`, not treated as errors. | +| `GET /api/v1/queues/:id/items?status&limit&offset` | `{ items, total, limit, offset }`, oldest first. `status` is `pending`, `done` or `skipped`; any other value is a `400`. | +| `GET /api/v1/queues/:id/next` | `{ item }`, the oldest pending item, or `204` with no body when none is pending. | +| `POST /api/v1/queues/:id/items/:item/complete` `{ "scores", "reviewer"?, "status"? }` | `{ item, score_ids }`. | + +A queue row is `{ queue_id, name, rubric, created_at, counts: { pending, done, +skipped } }`. An item row is `{ item_id, queue_id, trace_id, status, reviewer, +created_at, done_at }`, where `reviewer` and `done_at` are `null` until the +item is finished. `:id` is the `queue_id`. + +- `name`: 1-200 characters after trimming, unique. +- `rubric`: up to 50 entries of `{ "name", "description"?, "min"?, "max"? }`, + `[]` by default. Names are 1-200 characters and unique, a description is up + to 2000 characters, and `min` may not exceed `max`. + +The complete body takes `scores`, up to 50 entries of `{ "name", "value", +"comment"? }` where `value` is a number or `null`, an optional `reviewer` (up +to 200 characters) and `status`, which is `done` (the default) or `skipped`. +Each score is written on the item's trace, and the status, `reviewer` and +`done_at` are set in the same transaction. When the queue has a rubric, every +score name must be one of its names and the value must sit within that +entry's `min` and `max`, or the answer is a `400`. A skip with scores is a +`400`. Completing an item that is already done or skipped is a `409` and +writes nothing. An item that is not in the queue is a `404`. + +`next` does not claim or lock the item it returns. Asking twice returns the +same item until someone completes it. The queue is meant for one reviewer +working locally, so two reviewers on one queue can score the same trace. + +Retention never touches queues or items. If an item's trace is pruned, the +item keeps its `trace_id` and `next` still returns it. Completing it with +scores is then a `404` naming the trace, and the item stays pending, so skip +it instead. Bad bodies and unknown fields are a `400` with code +`LOUSHO_REQUEST_INVALID`, and an unknown queue is a `404` with the same code. +Bodies may be up to 256 KiB. + ### Retention The daemon keeps 30 days of runs by default. On its first sweep after start, diff --git a/llms-full.txt b/llms-full.txt index ec48126a..44088a45 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -14512,6 +14512,56 @@ An unknown field, an unknown `kind` or a `config` that does not match the `404` with the same code. `updated_at` moves on every `PUT`. The body may be up to 128 KiB. Retention never deletes rules. +### Annotation queues + +An annotation queue is a named list of traces for a person to review one by +one. Completing an item writes its scores as human scores, the same +`source: "human"` rows `POST /api/v1/scores` writes, so they show up in +`GET /api/v1/scores` and on the trace. The routes take the same token as +ingest: + +| Route | Answer | +| --- | --- | +| `POST /api/v1/queues` `{ "name", "rubric"? }` | `201 { queue }`. A taken name is a `409`. | +| `GET /api/v1/queues?limit&offset` | `{ queues, total, limit, offset }`, newest first. `limit` defaults to 100 (max 1000). | +| `GET /api/v1/queues/:id` | `{ queue }`. | +| `DELETE /api/v1/queues/:id` | `204`. The queue's items go with it; scores already written stay. | +| `POST /api/v1/queues/:id/items` `{ "trace_ids": [...] }` | `201 { added, skipped_existing }`. Takes 1-500 ids, queued in the order given. Every id must name a stored trace, or the answer is a `404` naming the first unknown one and nothing is added. Ids already in the queue are listed in `skipped_existing`, not treated as errors. | +| `GET /api/v1/queues/:id/items?status&limit&offset` | `{ items, total, limit, offset }`, oldest first. `status` is `pending`, `done` or `skipped`; any other value is a `400`. | +| `GET /api/v1/queues/:id/next` | `{ item }`, the oldest pending item, or `204` with no body when none is pending. | +| `POST /api/v1/queues/:id/items/:item/complete` `{ "scores", "reviewer"?, "status"? }` | `{ item, score_ids }`. | + +A queue row is `{ queue_id, name, rubric, created_at, counts: { pending, done, +skipped } }`. An item row is `{ item_id, queue_id, trace_id, status, reviewer, +created_at, done_at }`, where `reviewer` and `done_at` are `null` until the +item is finished. `:id` is the `queue_id`. + +- `name`: 1-200 characters after trimming, unique. +- `rubric`: up to 50 entries of `{ "name", "description"?, "min"?, "max"? }`, + `[]` by default. Names are 1-200 characters and unique, a description is up + to 2000 characters, and `min` may not exceed `max`. + +The complete body takes `scores`, up to 50 entries of `{ "name", "value", +"comment"? }` where `value` is a number or `null`, an optional `reviewer` (up +to 200 characters) and `status`, which is `done` (the default) or `skipped`. +Each score is written on the item's trace, and the status, `reviewer` and +`done_at` are set in the same transaction. When the queue has a rubric, every +score name must be one of its names and the value must sit within that +entry's `min` and `max`, or the answer is a `400`. A skip with scores is a +`400`. Completing an item that is already done or skipped is a `409` and +writes nothing. An item that is not in the queue is a `404`. + +`next` does not claim or lock the item it returns. Asking twice returns the +same item until someone completes it. The queue is meant for one reviewer +working locally, so two reviewers on one queue can score the same trace. + +Retention never touches queues or items. If an item's trace is pruned, the +item keeps its `trace_id` and `next` still returns it. Completing it with +scores is then a `404` naming the trace, and the item stays pending, so skip +it instead. Bad bodies and unknown fields are a `400` with code +`LOUSHO_REQUEST_INVALID`, and an unknown queue is a `404` with the same code. +Bodies may be up to 256 KiB. + ### Retention The daemon keeps 30 days of runs by default. On its first sweep after start, diff --git a/src/observe/humanScores.ts b/src/observe/humanScores.ts new file mode 100644 index 00000000..a05c5af8 --- /dev/null +++ b/src/observe/humanScores.ts @@ -0,0 +1,64 @@ +/** + * The one human-score write (LangSmith "Add feedback"): one `source: 'human'` + * score row on a trace, through ingest so rollups and the scores table see it + * like any other score. `POST /api/v1/scores` and the annotation-queue + * complete route both write through here. + */ + +import { randomUUID } from 'node:crypto'; + +import type { SqlDatabase } from '../storage/sqlite/driver'; +import { ingestBatch } from './store'; + +/** What a human score says about a trace (or one observation in it). */ +export interface HumanScoreInput { + trace_id: string; + obs_id?: string; + name: string; + value: number | null; + comment?: string; +} + +/** Thrown by {@link writeHumanScore} when the trace (or the observation in it) does not exist; routes answer 404. */ +export class ScoreTargetMissingError extends Error { + override readonly name = 'ScoreTargetMissingError'; +} + +/** + * Append one human score and return its `score_id`. The row's `agent_id` is + * the trace root's. Throws {@link ScoreTargetMissingError} for an unknown + * trace or an `obs_id` outside it. Joins an open `transaction()`. + */ +export function writeHumanScore(db: SqlDatabase, input: HumanScoreInput, now: number = Date.now()): string { + const root = db + .prepare('SELECT agent_id FROM observations WHERE trace_id = ? ORDER BY parent_obs_id IS NULL DESC LIMIT 1') + .get(input.trace_id); + if (root === undefined) throw new ScoreTargetMissingError(`No trace '${input.trace_id}'.`); + if (input.obs_id !== undefined) { + const inTrace = db.prepare('SELECT 1 AS ok FROM observations WHERE obs_id = ? AND trace_id = ?').get(input.obs_id, input.trace_id); + if (inTrace === undefined) { + throw new ScoreTargetMissingError(`No observation '${input.obs_id}' in trace '${input.trace_id}'.`); + } + } + const score_id = randomUUID(); + ingestBatch( + db, + [ + { + type: 'score', + data: { + score_id, + trace_id: input.trace_id, + obs_id: input.obs_id, + name: input.name, + value: input.value ?? undefined, + source: 'human', + comment: input.comment, + agent_id: typeof root.agent_id === 'string' ? root.agent_id : undefined, + }, + }, + ], + now + ); + return score_id; +} diff --git a/src/observe/queues.ts b/src/observe/queues.ts new file mode 100644 index 00000000..bb7a0a7b --- /dev/null +++ b/src/observe/queues.ts @@ -0,0 +1,205 @@ +/** + * Annotation queues data access: named lists of traces for a human to + * review, each with an optional rubric. Completing an item writes its scores + * as human score rows (the same rows `POST /api/v1/scores` writes). Kept out + * of store.ts so retention (`pruneBefore`) never learns about them; an item's + * `trace_id` may outlive its trace. + */ + +import { randomUUID } from 'node:crypto'; + +import type { SqlDatabase } from '../storage/sqlite/driver'; +import { writeHumanScore, type HumanScoreInput } from './humanScores'; +import { transaction } from './store'; + +/** One rubric entry: a score name the queue expects, optionally bounded. */ +export interface RubricEntry { + name: string; + description?: string; + min?: number; + max?: number; +} + +/** Where an item stands. */ +export type ItemStatus = 'pending' | 'done' | 'skipped'; + +/** The item statuses, in display order. */ +export const ITEM_STATUSES: readonly ItemStatus[] = ['pending', 'done', 'skipped']; + +/** One queue, with how many items it holds per status. */ +export interface QueueRow { + queue_id: string; + name: string; + rubric: RubricEntry[]; + created_at: number; + counts: Record; +} + +/** One queued trace. */ +export interface ItemRow { + item_id: string; + queue_id: string; + trace_id: string; + status: ItemStatus; + reviewer: string | null; + created_at: number; + done_at: number | null; +} + +const QUEUE_SELECT = ` + SELECT q.queue_id, q.name, q.rubric, q.created_at, + (SELECT COUNT(*) FROM annotation_items i WHERE i.queue_id = q.queue_id AND i.status = 'pending') AS pending, + (SELECT COUNT(*) FROM annotation_items i WHERE i.queue_id = q.queue_id AND i.status = 'done') AS done, + (SELECT COUNT(*) FROM annotation_items i WHERE i.queue_id = q.queue_id AND i.status = 'skipped') AS skipped + FROM annotation_queues q`; + +function toQueue(row: Record): QueueRow { + return { + queue_id: String(row.queue_id), + name: String(row.name), + rubric: JSON.parse(String(row.rubric)) as RubricEntry[], + created_at: Number(row.created_at), + counts: { pending: Number(row.pending), done: Number(row.done), skipped: Number(row.skipped) }, + }; +} + +function toItem(row: Record): ItemRow { + return { + item_id: String(row.item_id), + queue_id: String(row.queue_id), + trace_id: String(row.trace_id), + status: String(row.status) as ItemStatus, + reviewer: row.reviewer === null ? null : String(row.reviewer), + created_at: Number(row.created_at), + done_at: row.done_at === null ? null : Number(row.done_at), + }; +} + +/** The queue `queueId`, or `undefined`. */ +export function getQueue(db: SqlDatabase, queueId: string): QueueRow | undefined { + const row = db.prepare(`${QUEUE_SELECT} WHERE q.queue_id = ?`).get(queueId); + return row === undefined ? undefined : toQueue(row); +} + +/** Create a queue; `undefined` when the name is taken. */ +export function createQueue( + db: SqlDatabase, + fields: { name: string; rubric: RubricEntry[] }, + now: number = Date.now() +): QueueRow | undefined { + if (db.prepare('SELECT 1 AS ok FROM annotation_queues WHERE name = ?').get(fields.name) !== undefined) return undefined; + const queueId = randomUUID(); + db.prepare('INSERT INTO annotation_queues (queue_id, name, rubric, created_at) VALUES (?, ?, ?, ?)').run( + queueId, + fields.name, + JSON.stringify(fields.rubric), + now + ); + return getQueue(db, queueId); +} + +/** Delete a queue and its items; `false` when it is unknown. */ +export function deleteQueue(db: SqlDatabase, queueId: string): boolean { + return transaction(db, () => { + db.prepare('DELETE FROM annotation_items WHERE queue_id = ?').run(queueId); + return Number(db.prepare('DELETE FROM annotation_queues WHERE queue_id = ?').run(queueId).changes) > 0; + }); +} + +/** A page of queues, newest first, plus the total. */ +export function listQueues(db: SqlDatabase, options: { limit: number; offset: number }): { queues: QueueRow[]; total: number } { + const rows = db.prepare(`${QUEUE_SELECT} ORDER BY q.created_at DESC, q.rowid DESC LIMIT ? OFFSET ?`).all(options.limit, options.offset); + const total = db.prepare('SELECT COUNT(*) AS n FROM annotation_queues').get(); + return { queues: rows.map(toQueue), total: Number(total?.n ?? 0) }; +} + +/** A page of a queue's items, oldest first (insertion order breaks ties), plus the total matching. */ +export function listItems( + db: SqlDatabase, + queueId: string, + options: { limit: number; offset: number; status?: ItemStatus } +): { items: ItemRow[]; total: number } { + const clause = options.status === undefined ? '' : ' AND status = ?'; + const args: string[] = options.status === undefined ? [queueId] : [queueId, options.status]; + const rows = db + .prepare(`SELECT * FROM annotation_items WHERE queue_id = ?${clause} ORDER BY created_at ASC, rowid ASC LIMIT ? OFFSET ?`) + .all(...args, options.limit, options.offset); + const total = db.prepare(`SELECT COUNT(*) AS n FROM annotation_items WHERE queue_id = ?${clause}`).get(...args); + return { items: rows.map(toItem), total: Number(total?.n ?? 0) }; +} + +/** + * Queue `traceIds` in the order given, in one transaction. Every id must + * name a stored trace, else nothing is added and the first unknown id comes + * back as `{ unknown }`. Ids already in the queue (or repeated in the call) + * are skipped, not errors. + */ +export function addItems( + db: SqlDatabase, + queueId: string, + traceIds: readonly string[], + now: number = Date.now() +): { added: ItemRow[]; skipped_existing: string[] } | { unknown: string } { + return transaction(db, () => { + const traceExists = db.prepare('SELECT 1 AS ok FROM observations WHERE trace_id = ? LIMIT 1'); + const unknown = traceIds.find((traceId) => traceExists.get(traceId) === undefined); + if (unknown !== undefined) return { unknown }; + const queued = db.prepare('SELECT 1 AS ok FROM annotation_items WHERE queue_id = ? AND trace_id = ?'); + const insert = db.prepare( + "INSERT INTO annotation_items (item_id, queue_id, trace_id, status, created_at) VALUES (?, ?, ?, 'pending', ?)" + ); + const addedIds: string[] = []; + const skipped_existing: string[] = []; + for (const traceId of traceIds) { + if (queued.get(queueId, traceId) !== undefined) { + skipped_existing.push(traceId); + continue; + } + const itemId = randomUUID(); + insert.run(itemId, queueId, traceId, now); + addedIds.push(itemId); + } + const byId = db.prepare('SELECT * FROM annotation_items WHERE item_id = ?'); + return { added: addedIds.map((id) => toItem(byId.get(id) as Record)), skipped_existing }; + }); +} + +/** The queue's oldest pending item, or `undefined`. Does not claim it. */ +export function nextItem(db: SqlDatabase, queueId: string): ItemRow | undefined { + const row = db + .prepare("SELECT * FROM annotation_items WHERE queue_id = ? AND status = 'pending' ORDER BY created_at ASC, rowid ASC LIMIT 1") + .get(queueId); + return row === undefined ? undefined : toItem(row); +} + +/** The item `itemId` of queue `queueId`, or `undefined`. */ +export function getItem(db: SqlDatabase, queueId: string, itemId: string): ItemRow | undefined { + const row = db.prepare('SELECT * FROM annotation_items WHERE item_id = ? AND queue_id = ?').get(itemId, queueId); + return row === undefined ? undefined : toItem(row); +} + +/** + * Finish a pending item: write each score as a human score on its trace and + * set `status`, `reviewer` and `done_at`, all in one transaction. `'finished'` + * when the item is already done or skipped. A pruned trace throws + * `ScoreTargetMissingError` (from the score write) and the item stays pending. + */ +export function completeItem( + db: SqlDatabase, + item: ItemRow, + outcome: { scores: Omit[]; reviewer?: string; status: 'done' | 'skipped' }, + now: number = Date.now() +): { item: ItemRow; score_ids: string[] } | 'finished' { + return transaction(db, () => { + const current = getItem(db, item.queue_id, item.item_id); + if (current === undefined || current.status !== 'pending') return 'finished'; + const score_ids = outcome.scores.map((score) => writeHumanScore(db, { ...score, trace_id: item.trace_id }, now)); + db.prepare('UPDATE annotation_items SET status = ?, reviewer = ?, done_at = ? WHERE item_id = ?').run( + outcome.status, + outcome.reviewer ?? null, + now, + item.item_id + ); + return { item: getItem(db, item.queue_id, item.item_id) as ItemRow, score_ids }; + }); +} diff --git a/src/observe/routes/index.ts b/src/observe/routes/index.ts index de27bd5d..0eff1002 100644 --- a/src/observe/routes/index.ts +++ b/src/observe/routes/index.ts @@ -1,6 +1,6 @@ /** * The daemon's route registry. One file per feature area (ingest, traces, - * insights, agents, scores, status, prompts, hub, datasets, evaluators) exports a `routes` array; this index + * insights, agents, scores, status, prompts, hub, datasets, evaluators, queues) exports a `routes` array; this index * concatenates them - so a later child spec (prompts, hub, alerts) adds a * file here and never touches the daemon's dispatch internals. * @@ -17,6 +17,7 @@ import { hubRoutes } from './hub'; import { ingestRoutes } from './ingest'; import { insightsRoutes } from './insights'; import { promptsRoutes } from './prompts'; +import { queuesRoutes } from './queues'; import { scoresRoutes } from './scores'; import { sessionsRoutes } from './sessions'; import { statusRoutes } from './status'; @@ -42,6 +43,7 @@ const routes: readonly Route[] = [ ...hubRoutes, ...datasetsRoutes, ...evaluatorsRoutes, + ...queuesRoutes, ]; interface CompiledRoute { diff --git a/src/observe/routes/queues.test.ts b/src/observe/routes/queues.test.ts new file mode 100644 index 00000000..5dd93242 --- /dev/null +++ b/src/observe/routes/queues.test.ts @@ -0,0 +1,265 @@ +/** + * Annotation queues API: create/list/get/delete queues, add trace ids, walk + * them with `next`, and complete items into human score rows. + */ +import { describe, expect, it } from 'vitest'; +import type * as http from 'node:http'; +import type { IngestEvent } from '../envelope'; +import { ingestBatch, openDb, pruneBefore, type ScoreRow } from '../store'; +import type { ItemRow, QueueRow } from '../queues'; +import type { SqlDatabase } from '../../storage/sqlite/driver'; +// Through the registry, production's load order (see scores.test.ts). +import { matchRoute } from './index'; + +const T0 = 1_700_000_000_000; + +interface Captured { + status: number; + body: unknown; +} + +/** Calls the matched route with a recording ServerResponse stand-in. */ +async function call( + database: SqlDatabase, + method: 'GET' | 'POST' | 'PUT' | 'DELETE', + path: string, + body?: unknown +): Promise { + const captured: Captured = { status: 0, body: undefined }; + const res = { + writeHead(status: number) { + captured.status = status; + return res; + }, + end(chunk?: string) { + captured.body = chunk === undefined || chunk === '' ? undefined : JSON.parse(chunk); + }, + } as unknown as http.ServerResponse; + const [pathname, search = ''] = path.split('?'); + const matched = matchRoute(method, pathname); + if (!matched) throw new Error(`no route for ${method} ${pathname}`); + const text = typeof body === 'string' ? body : JSON.stringify(body ?? {}); + await matched.route.handle({ + db: database, + dbPath: ':memory:', + req: {} as http.IncomingMessage, + res, + params: matched.params, + query: new URLSearchParams(search), + readBody: () => Promise.resolve(text), + }); + return captured; +} + +const root = (traceId: string): IngestEvent => + ({ + type: 'observation', + data: { obs_id: `${traceId}-root`, trace_id: traceId, agent_id: 'ag', kind: 'span', name: 'run', start: T0, end: T0 + 10 }, + }) as IngestEvent; + +async function withDb(work: (database: SqlDatabase) => Promise): Promise { + const database = openDb(':memory:'); + ingestBatch(database, [root('t1'), root('t2'), root('t3')]); + try { + await work(database); + } finally { + database.close(); + } +} + +const invalid = { code: 'LOUSHO_REQUEST_INVALID' }; +const queueOf = (c: Captured) => (c.body as { queue: QueueRow }).queue; +const itemOf = (c: Captured) => (c.body as { item: ItemRow }).item; + +async function makeQueue(database: SqlDatabase, body: unknown = { name: 'review' }): Promise { + const created = await call(database, 'POST', '/api/v1/queues', body); + expect(created.status).toBe(201); + return queueOf(created); +} + +const add = (database: SqlDatabase, id: string, trace_ids: string[]) => + call(database, 'POST', `/api/v1/queues/${id}/items`, { trace_ids }); +const next = (database: SqlDatabase, id: string) => call(database, 'GET', `/api/v1/queues/${id}/next`); +const complete = (database: SqlDatabase, id: string, item: string, body: unknown) => + call(database, 'POST', `/api/v1/queues/${id}/items/${item}/complete`, body); +const humanScores = async (database: SqlDatabase, traceId: string) => + ((await call(database, 'GET', `/api/v1/scores?trace_id=${traceId}&source=human`)).body as { scores: ScoreRow[] }).scores; + +describe('annotation queues API', () => { + it('creates, lists, gets and deletes queues', async () => { + await withDb(async (database) => { + const first = await makeQueue(database, { name: 'a' }); + expect(first).toMatchObject({ name: 'a', rubric: [], counts: { pending: 0, done: 0, skipped: 0 } }); + expect(typeof first.queue_id).toBe('string'); + const second = await makeQueue(database, { name: 'b', rubric: [{ name: 'quality', min: 0, max: 5 }] }); + expect(second.rubric).toEqual([{ name: 'quality', min: 0, max: 5 }]); + + const list = (await call(database, 'GET', '/api/v1/queues')).body as { queues: QueueRow[]; total: number; limit: number; offset: number }; + expect(list.total).toBe(2); + expect(list.limit).toBe(100); + expect(list.offset).toBe(0); + expect(list.queues.map((q) => q.name)).toEqual(['b', 'a']); + + const got = await call(database, 'GET', `/api/v1/queues/${first.queue_id}`); + expect(got.status).toBe(200); + expect(queueOf(got).name).toBe('a'); + + await add(database, first.queue_id, ['t1']); + const deleted = await call(database, 'DELETE', `/api/v1/queues/${first.queue_id}`); + expect(deleted.status).toBe(204); + expect((await call(database, 'GET', `/api/v1/queues/${first.queue_id}`)).status).toBe(404); + expect((await call(database, 'DELETE', `/api/v1/queues/${first.queue_id}`)).status).toBe(404); + const orphans = database.prepare('SELECT COUNT(*) AS n FROM annotation_items WHERE queue_id = ?').get(first.queue_id); + expect(Number(orphans?.n)).toBe(0); + }); + }); + + it('409s a duplicate name and 400s bad bodies and rubrics', async () => { + await withDb(async (database) => { + await makeQueue(database, { name: 'a' }); + const dup = await call(database, 'POST', '/api/v1/queues', { name: 'a' }); + expect(dup.status).toBe(409); + expect(dup.body).toMatchObject(invalid); + const bad = [ + {}, + { name: '' }, + { name: 'x', extra: 1 }, + { name: 'x', rubric: [{ name: 'q', min: 5, max: 1 }] }, + { name: 'x', rubric: [{ name: 'q' }, { name: 'q' }] }, + { name: 'x', rubric: Array.from({ length: 51 }, (_, i) => ({ name: `r${i}` })) }, + { name: 'x', rubric: [{ name: 'q', extra: true }] }, + ]; + for (const body of bad) await expect(call(database, 'POST', '/api/v1/queues', body)).rejects.toMatchObject(invalid); + }); + }); + + it('adds trace ids in order, skips already-queued ones, and lists items oldest first', async () => { + await withDb(async (database) => { + const queue = await makeQueue(database); + const first = await add(database, queue.queue_id, ['t2', 't1']); + expect(first.status).toBe(201); + const added = (first.body as { added: ItemRow[]; skipped_existing: string[] }).added; + expect(added.map((i) => i.trace_id)).toEqual(['t2', 't1']); + expect(added[0]).toMatchObject({ queue_id: queue.queue_id, status: 'pending', reviewer: null, done_at: null }); + + const again = (await add(database, queue.queue_id, ['t1', 't3'])).body as { added: ItemRow[]; skipped_existing: string[] }; + expect(again.added.map((i) => i.trace_id)).toEqual(['t3']); + expect(again.skipped_existing).toEqual(['t1']); + + const items = (await call(database, 'GET', `/api/v1/queues/${queue.queue_id}/items`)).body as { items: ItemRow[]; total: number }; + expect(items.total).toBe(3); + expect(items.items.map((i) => i.trace_id)).toEqual(['t2', 't1', 't3']); + expect(queueOf(await call(database, 'GET', `/api/v1/queues/${queue.queue_id}`)).counts).toEqual({ pending: 3, done: 0, skipped: 0 }); + await expect(call(database, 'GET', `/api/v1/queues/${queue.queue_id}/items?status=nope`)).rejects.toMatchObject(invalid); + }); + }); + + it('404s an unknown trace id and adds nothing', async () => { + await withDb(async (database) => { + const queue = await makeQueue(database); + const out = await add(database, queue.queue_id, ['t1', 'ghost', 't2']); + expect(out.status).toBe(404); + expect(out.body).toMatchObject({ ...invalid, error: expect.stringContaining('ghost') }); + const items = (await call(database, 'GET', `/api/v1/queues/${queue.queue_id}/items`)).body as { total: number }; + expect(items.total).toBe(0); + expect((await add(database, 'no-such-queue', ['t1'])).status).toBe(404); + await expect(add(database, queue.queue_id, [])).rejects.toMatchObject(invalid); + }); + }); + + it('next returns the oldest pending item, 204 when none, 404 for an unknown queue', async () => { + await withDb(async (database) => { + const queue = await makeQueue(database); + const empty = await next(database, queue.queue_id); + expect(empty.status).toBe(204); + expect(empty.body).toBeUndefined(); + expect((await next(database, 'nope')).status).toBe(404); + + await add(database, queue.queue_id, ['t3', 't1']); + const got = await next(database, queue.queue_id); + expect(got.status).toBe(200); + expect(itemOf(got).trace_id).toBe('t3'); + // next does not claim: asking again returns the same item. + expect(itemOf(await next(database, queue.queue_id)).item_id).toBe(itemOf(got).item_id); + }); + }); + + it('complete writes human scores, marks done, and next moves on', async () => { + await withDb(async (database) => { + const queue = await makeQueue(database); + await add(database, queue.queue_id, ['t1', 't2']); + const item = itemOf(await next(database, queue.queue_id)); + const done = await complete(database, queue.queue_id, item.item_id, { + scores: [ + { name: 'quality', value: 4, comment: 'good' }, + { name: 'flagged', value: null }, + ], + reviewer: 'ali', + }); + expect(done.status).toBe(200); + const body = done.body as { item: ItemRow; score_ids: string[] }; + expect(body.score_ids).toHaveLength(2); + expect(body.item).toMatchObject({ status: 'done', reviewer: 'ali' }); + expect(typeof body.item.done_at).toBe('number'); + + const scores = await humanScores(database, 't1'); + expect(scores.map((s) => s.name).sort()).toEqual(['flagged', 'quality']); + expect(scores.find((s) => s.name === 'quality')).toMatchObject({ value: 4, comment: 'good', source: 'human', agent_id: 'ag' }); + + expect(itemOf(await next(database, queue.queue_id)).trace_id).toBe('t2'); + expect(queueOf(await call(database, 'GET', `/api/v1/queues/${queue.queue_id}`)).counts).toEqual({ pending: 1, done: 1, skipped: 0 }); + + const twice = await complete(database, queue.queue_id, item.item_id, { scores: [{ name: 'quality', value: 1 }] }); + expect(twice.status).toBe(409); + expect(await humanScores(database, 't1')).toHaveLength(2); + }); + }); + + it('skips an item, and 400s a skip carrying scores', async () => { + await withDb(async (database) => { + const queue = await makeQueue(database); + await add(database, queue.queue_id, ['t1', 't2']); + const item = itemOf(await next(database, queue.queue_id)); + await expect( + complete(database, queue.queue_id, item.item_id, { status: 'skipped', scores: [{ name: 'q', value: 1 }] }) + ).rejects.toMatchObject(invalid); + const skipped = await complete(database, queue.queue_id, item.item_id, { status: 'skipped', scores: [] }); + expect(skipped.status).toBe(200); + expect(itemOf(skipped).status).toBe('skipped'); + expect(itemOf(await next(database, queue.queue_id)).trace_id).toBe('t2'); + expect((await complete(database, queue.queue_id, 'nope', { scores: [] })).status).toBe(404); + }); + }); + + it('checks scores against a non-empty rubric', async () => { + await withDb(async (database) => { + const queue = await makeQueue(database, { name: 'r', rubric: [{ name: 'quality', min: 0, max: 5 }, { name: 'tone' }] }); + await add(database, queue.queue_id, ['t1']); + const item = itemOf(await next(database, queue.queue_id)); + await expect(complete(database, queue.queue_id, item.item_id, { scores: [{ name: 'other', value: 1 }] })).rejects.toMatchObject(invalid); + await expect(complete(database, queue.queue_id, item.item_id, { scores: [{ name: 'quality', value: 9 }] })).rejects.toMatchObject(invalid); + expect(await humanScores(database, 't1')).toHaveLength(0); + const ok = await complete(database, queue.queue_id, item.item_id, { + scores: [ + { name: 'quality', value: 5 }, + { name: 'tone', value: -3 }, + ], + }); + expect(ok.status).toBe(200); + }); + }); + + it('keeps items of a pruned trace; completing one 404s and leaves it pending', async () => { + await withDb(async (database) => { + const queue = await makeQueue(database); + await add(database, queue.queue_id, ['t1']); + pruneBefore(database, T0 + 1_000_000); + const item = itemOf(await next(database, queue.queue_id)); + expect(item.trace_id).toBe('t1'); + const out = await complete(database, queue.queue_id, item.item_id, { scores: [{ name: 'q', value: 1 }] }); + expect(out.status).toBe(404); + expect(out.body).toMatchObject({ ...invalid, error: expect.stringContaining('t1') }); + expect(itemOf(await next(database, queue.queue_id)).status).toBe('pending'); + }); + }); +}); diff --git a/src/observe/routes/queues.ts b/src/observe/routes/queues.ts new file mode 100644 index 00000000..aeaed195 --- /dev/null +++ b/src/observe/routes/queues.ts @@ -0,0 +1,216 @@ +/** + * Annotation queues (LangSmith "annotation queues"): named lists of traces a + * human walks through, scoring each one. Completing an item writes its scores + * as human score rows, the same rows `POST /api/v1/scores` writes. + * + * - `GET /api/v1/queues?limit&offset` -> `{ queues, total, limit, offset }`, newest first. + * - `POST /api/v1/queues` `{ name, rubric? }` -> `201 { queue }` (a taken name is a 409). + * - `GET /api/v1/queues/:id` -> `{ queue }`; `DELETE` -> `204`. + * - `GET /api/v1/queues/:id/items?status&limit&offset` -> `{ items, total, limit, offset }`, oldest first. + * - `POST /api/v1/queues/:id/items` `{ trace_ids }` -> `201 { added, skipped_existing }`. + * - `GET /api/v1/queues/:id/next` -> `{ item }`, the oldest pending; `204` when none. + * It does not claim the item: this is a single-reviewer local tool. + * - `POST /api/v1/queues/:id/items/:item/complete` `{ scores, reviewer?, status? }` + * -> `{ item, score_ids }`. + * + * Auth is the daemon's global API-token check; nothing per-route. + */ + +import type * as http from 'node:http'; +import { z } from 'zod'; + +import { SDKError } from '../../execution/errors'; +import { ScoreTargetMissingError } from '../humanScores'; +import { + addItems, + completeItem, + createQueue, + deleteQueue, + getItem, + getQueue, + ITEM_STATUSES, + listItems, + listQueues, + nextItem, + type ItemStatus, + type QueueRow, +} from '../queues'; +import { intParam, requireJsonBody, sendJson, type Route } from './httpKit'; + +const QUEUE_BODY_LIMIT = 256 * 1024; + +const rubricEntry = z + .object({ + name: z.string().trim().min(1).max(200), + description: z.string().max(2000).optional(), + min: z.number().finite().optional(), + max: z.number().finite().optional(), + }) + .strict() + .refine((entry) => entry.min === undefined || entry.max === undefined || entry.min <= entry.max, { + message: 'min must not exceed max', + }); + +const rubricSchema = z + .array(rubricEntry) + .max(50) + .refine((entries) => new Set(entries.map((e) => e.name)).size === entries.length, { message: 'rubric names must be unique' }); + +const createBody = z.object({ name: z.string().trim().min(1).max(200), rubric: rubricSchema.default([]) }).strict(); + +const addBody = z.object({ trace_ids: z.array(z.string().min(1)).min(1).max(500) }).strict(); + +const completeBody = z + .object({ + scores: z + .array( + z + .object({ + name: z.string().trim().min(1).max(200), + value: z.number().finite().nullable(), + comment: z.string().max(10_000).optional(), + }) + .strict() + ) + .max(50), + reviewer: z.string().max(200).optional(), + status: z.enum(['done', 'skipped']).default('done'), + }) + .strict(); + +const invalid = (message: string): SDKError => new SDKError(message, 'LOUSHO_REQUEST_INVALID', { appendHelp: false }); + +const answer = (res: http.ServerResponse, status: number, error: string): void => + sendJson(res, status, { error, code: 'LOUSHO_REQUEST_INVALID' }); + +const queueMissing = (res: http.ServerResponse, id: string): void => answer(res, 404, `No annotation queue '${id}'.`); + +/** The list's `status` filter; a bad value throws `LOUSHO_REQUEST_INVALID`. */ +function statusFilter(query: URLSearchParams): ItemStatus | undefined { + const status = query.get('status'); + if (status === null) return undefined; + if (!(ITEM_STATUSES as readonly string[]).includes(status)) { + throw invalid(`Invalid queue items query: status must be one of ${ITEM_STATUSES.map((s) => `'${s}'`).join(', ')}.`); + } + return status as ItemStatus; +} + +/** A complete body checked against the queue: no scores on a skip, and rubric names/bounds when it has one. */ +function checkOutcome(queue: QueueRow, body: z.infer): void { + if (body.status === 'skipped' && body.scores.length > 0) throw invalid('Invalid complete body: a skipped item takes no scores.'); + if (queue.rubric.length === 0) return; + for (const score of body.scores) { + const entry = queue.rubric.find((r) => r.name === score.name); + if (entry === undefined) { + throw invalid(`Invalid complete body: score '${score.name}' is not in the queue's rubric (${queue.rubric.map((r) => r.name).join(', ')}).`); + } + const value = score.value; + if (value === null) continue; + if ((entry.min !== undefined && value < entry.min) || (entry.max !== undefined && value > entry.max)) { + throw invalid(`Invalid complete body: score '${score.name}' value ${value} is outside [${entry.min ?? '-inf'}, ${entry.max ?? 'inf'}].`); + } + } +} + +/** Pagination shared by both list routes. */ +function page(query: URLSearchParams): { limit: number; offset: number } { + return { + limit: intParam(query, 'limit', 100, { min: 1, max: 1000 }), + offset: intParam(query, 'offset', 0, { min: 0, max: Number.MAX_SAFE_INTEGER }), + }; +} + +export const queuesRoutes: readonly Route[] = [ + { + method: 'GET', + path: '/api/v1/queues', + handle: ({ db, res, query }) => { + const paging = page(query); + sendJson(res, 200, { ...listQueues(db, paging), ...paging }); + }, + }, + { + method: 'POST', + path: '/api/v1/queues', + handle: async ({ db, res, readBody }) => { + const body = requireJsonBody(createBody, await readBody(QUEUE_BODY_LIMIT), 'annotation queue'); + const queue = createQueue(db, body); + if (queue === undefined) return answer(res, 409, `An annotation queue named '${body.name}' already exists.`); + sendJson(res, 201, { queue }); + }, + }, + { + method: 'GET', + path: '/api/v1/queues/:id', + handle: ({ db, res, params }) => { + const queue = getQueue(db, params.id); + if (queue === undefined) return queueMissing(res, params.id); + sendJson(res, 200, { queue }); + }, + }, + { + method: 'DELETE', + path: '/api/v1/queues/:id', + handle: ({ db, res, params }) => { + if (!deleteQueue(db, params.id)) return queueMissing(res, params.id); + res.writeHead(204); + res.end(); + }, + }, + { + method: 'GET', + path: '/api/v1/queues/:id/items', + handle: ({ db, res, params, query }) => { + const status = statusFilter(query); + if (getQueue(db, params.id) === undefined) return queueMissing(res, params.id); + const paging = page(query); + sendJson(res, 200, { ...listItems(db, params.id, { ...paging, status }), ...paging }); + }, + }, + { + method: 'POST', + path: '/api/v1/queues/:id/items', + handle: async ({ db, res, params, readBody }) => { + const body = requireJsonBody(addBody, await readBody(QUEUE_BODY_LIMIT), 'queue items'); + if (getQueue(db, params.id) === undefined) return queueMissing(res, params.id); + const result = addItems(db, params.id, body.trace_ids); + if ('unknown' in result) return answer(res, 404, `No trace '${result.unknown}'; nothing was added.`); + sendJson(res, 201, result); + }, + }, + { + method: 'GET', + path: '/api/v1/queues/:id/next', + handle: ({ db, res, params }) => { + if (getQueue(db, params.id) === undefined) return queueMissing(res, params.id); + const item = nextItem(db, params.id); + if (item === undefined) { + res.writeHead(204); + res.end(); + return; + } + sendJson(res, 200, { item }); + }, + }, + { + method: 'POST', + path: '/api/v1/queues/:id/items/:item/complete', + handle: async ({ db, res, params, readBody }) => { + const body = requireJsonBody(completeBody, await readBody(QUEUE_BODY_LIMIT), 'complete'); + const queue = getQueue(db, params.id); + if (queue === undefined) return queueMissing(res, params.id); + const item = getItem(db, params.id, params.item); + if (item === undefined) return answer(res, 404, `No item '${params.item}' in annotation queue '${params.id}'.`); + checkOutcome(queue, body); + let result: ReturnType; + try { + result = completeItem(db, item, body); + } catch (error) { + if (error instanceof ScoreTargetMissingError) return answer(res, 404, `${error.message} It may have been pruned.`); + throw error; + } + if (result === 'finished') return answer(res, 409, `Item '${item.item_id}' is already ${item.status === 'pending' ? 'finished' : item.status}.`); + sendJson(res, 200, result); + }, + }, +]; diff --git a/src/observe/routes/scores.ts b/src/observe/routes/scores.ts index 9d426ab3..6eaf2e9d 100644 --- a/src/observe/routes/scores.ts +++ b/src/observe/routes/scores.ts @@ -15,11 +15,11 @@ * global API-token check; nothing per-route. */ -import { randomUUID } from 'node:crypto'; import { z } from 'zod'; import { SCORE_SOURCES } from '../envelope'; -import { ingestBatch, listScores, scorePassRates, type ScoreFilter } from '../store'; +import { ScoreTargetMissingError, writeHumanScore } from '../humanScores'; +import { listScores, scorePassRates, type ScoreFilter } from '../store'; import { badRequest, intParam, requireJsonBody, sendJson, timeParam, type Route } from './httpKit'; const FEEDBACK_BODY_LIMIT = 64 * 1024; @@ -32,38 +32,21 @@ const feedbackBody = z.object({ comment: z.string().max(10_000).optional(), }); - export const scoresRoutes: readonly Route[] = [ { method: 'POST', path: '/api/v1/scores', handle: async ({ db, res, readBody }) => { const body = requireJsonBody(feedbackBody, await readBody(FEEDBACK_BODY_LIMIT), 'feedback'); - const notFound = (error: string) => sendJson(res, 404, { error, code: 'LOUSHO_REQUEST_INVALID' }); - const root = db - .prepare('SELECT agent_id FROM observations WHERE trace_id = ? ORDER BY parent_obs_id IS NULL DESC LIMIT 1') - .get(body.trace_id); - if (root === undefined) return notFound(`No trace '${body.trace_id}'.`); - if (body.obs_id !== undefined) { - const inTrace = db.prepare('SELECT 1 AS ok FROM observations WHERE obs_id = ? AND trace_id = ?').get(body.obs_id, body.trace_id); - if (inTrace === undefined) return notFound(`No observation '${body.obs_id}' in trace '${body.trace_id}'.`); + let score_id: string; + try { + score_id = writeHumanScore(db, body); + } catch (error) { + if (error instanceof ScoreTargetMissingError) { + return sendJson(res, 404, { error: error.message, code: 'LOUSHO_REQUEST_INVALID' }); + } + throw error; } - const score_id = randomUUID(); - ingestBatch(db, [ - { - type: 'score', - data: { - score_id, - trace_id: body.trace_id, - obs_id: body.obs_id, - name: body.name, - value: body.value ?? undefined, - source: 'human', - comment: body.comment, - agent_id: typeof root.agent_id === 'string' ? root.agent_id : undefined, - }, - }, - ]); sendJson(res, 201, { score_id }); }, }, diff --git a/src/observe/schema.ts b/src/observe/schema.ts index 7b197e9d..c012307b 100644 --- a/src/observe/schema.ts +++ b/src/observe/schema.ts @@ -249,6 +249,31 @@ export const OBSERVE_MIGRATIONS: readonly string[] = [ updated_at INTEGER NOT NULL ); `, + ` + -- Annotation queues: named lists of traces for a human to review, with an + -- optional rubric (JSON array of { name, description?, min?, max? }). + -- annotation_items.trace_id is deliberately not a foreign key - retention + -- may prune the trace, the item stays. Retention never touches these + -- tables. foreign_keys is off, so deleting a queue deletes its items + -- explicitly; the CASCADE only documents the intent. + CREATE TABLE annotation_queues ( + queue_id TEXT PRIMARY KEY, + name TEXT NOT NULL UNIQUE, + rubric TEXT NOT NULL DEFAULT '[]', + created_at INTEGER NOT NULL + ); + CREATE TABLE annotation_items ( + item_id TEXT PRIMARY KEY, + queue_id TEXT NOT NULL REFERENCES annotation_queues(queue_id) ON DELETE CASCADE, + trace_id TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending','done','skipped')), + reviewer TEXT, + created_at INTEGER NOT NULL, + done_at INTEGER, + UNIQUE (queue_id, trace_id) + ); + CREATE INDEX annotation_items_by_status ON annotation_items (queue_id, status, created_at); + `, ]; /** diff --git a/src/observe/store.test.ts b/src/observe/store.test.ts index 6cad4b2e..9f2025f9 100644 --- a/src/observe/store.test.ts +++ b/src/observe/store.test.ts @@ -403,6 +403,24 @@ describe('transaction', () => { database.close(); } }); + + it('a nested call joins the outer transaction, so an outer throw undoes the inner writes', () => { + const database = db(); + try { + expect(() => + transaction(database, () => { + ingestBatch(database, [obs()]); + throw new Error('boom'); + }) + ).toThrow('boom'); + expect(observeCounts(database).observations).toBe(0); + // and the db is usable for a fresh transaction afterwards + ingestBatch(database, [obs()]); + expect(observeCounts(database).observations).toBe(1); + } finally { + database.close(); + } + }); }); function ingestBatchInner(database: SqlDatabase): void { diff --git a/src/observe/store.ts b/src/observe/store.ts index 89f6f6f3..1a6da484 100644 --- a/src/observe/store.ts +++ b/src/observe/store.ts @@ -62,9 +62,18 @@ export function openDb(path: string): SqlDatabase { return db; } -/** `work` in one write transaction (`BEGIN IMMEDIATE`), rolled back when it throws. */ +/** Databases with a `transaction()` open; a nested call joins the outer one. */ +const openTransactions = new WeakSet(); + +/** + * `work` in one write transaction (`BEGIN IMMEDIATE`), rolled back when it + * throws. Called inside another `transaction()` on the same db it just runs + * `work` there, so a throw rolls back the outer transaction as a whole. + */ export function transaction(db: SqlDatabase, work: () => T): T { + if (openTransactions.has(db)) return work(); db.exec('BEGIN IMMEDIATE'); + openTransactions.add(db); try { const result = work(); db.exec('COMMIT'); @@ -76,6 +85,8 @@ export function transaction(db: SqlDatabase, work: () => T): T { // the transaction is already gone; keep the original error } throw error; + } finally { + openTransactions.delete(db); } } From 1827a2f0975b1145fb294e052d715a0d9559a620 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 29 Oct 2026 00:51:51 +0300 Subject: [PATCH 2/2] ci(pack-smoke): unpacked budget 18 MiB after the Studio revamp Co-Authored-By: Claude Opus 5.5 --- scripts/pack-smoke.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/scripts/pack-smoke.ts b/scripts/pack-smoke.ts index 7db00dc8..65cd4d8a 100644 --- a/scripts/pack-smoke.ts +++ b/scripts/pack-smoke.ts @@ -65,9 +65,11 @@ const REPO_ROOT = path.resolve(__dirname, '..'); * scripts/size-check.ts holds the tighter, committed budget. * Measured 4,721,107 bytes (~4.5 MiB) packed after the observe parity work * (2026-10-24); packed budget raised to 5 MiB (~10% headroom). + * CI measured 16,796,541 bytes (~16.0 MiB) unpacked on 2026-10-29 (Studio + * revamp dist plus docs); unpacked budget raised to 18 MiB (~10% headroom). */ const MAX_ENTRIES = 850; -const MAX_UNPACKED_BYTES = 16 * 1024 * 1024; +const MAX_UNPACKED_BYTES = 18 * 1024 * 1024; const MAX_PACKED_BYTES = 5 * 1024 * 1024; const DEFAULT_PEERS = 'ai@7 zod@4 @ai-sdk/openai@4 react@19 vue@3 @opentelemetry/api@1';