feat(observe): annotation queues

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,

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,

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';

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;
+}

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<ItemStatus, number>;
+}
+
+/** 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<string, unknown>): 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<string, unknown>): 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<string, unknown>)), 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<HumanScoreInput, 'trace_id' | 'obs_id'>[]; 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 };
+  });
+}