Fix the frame producer's field names, and pipeline the batch

The other end of the `stream:camera_frames` drift. This side wrote
`farmId`, `printerId`, `printJobId`, `r2ObjectKey` and `capturedAt`;
`continuum-ai-worker` read `printer_id` and nothing else. Every task was
delivered and acknowledged, and every frame was attributed to a printer
called "unknown", with no error on either side - a stream entry is a
flat string map, so there is nothing to parse and nothing to fail.

The field names are now snake_case, from
`schemas/continuum/v1/ai/frame_task.schema.json`. That is the convention
this repo already follows for anything shared with a Rust consumer (see
`/ws/edge/v1`), and it is not an inconsistency with the camelCase HTTP
routes: this is not an HTTP route.

`print_job_id` is omitted from the XADD when there is no job rather than
written as `""`. An empty string is a value: a consumer that believes it
looks up job id `""`, or puts it in a uuid column. A stream entry has no
null, so "no value" has to mean "no field".

The producer checks each task against the shared schema before writing
it. The check is not for the database's sake - a stream accepts any
field name - it is to make this exact class of drift loud. It earned its
keep immediately by failing on every task with `Unknown format 'uuid'`:
TypeBox ships no format validators and treats an *unregistered* format
as a failure rather than skipping it, so a schema using `format: "uuid"`
checked outside Elysia rejects everything. `src/validation/formats.ts`
registers the two this codebase uses, guarded by `Has` so Elysia's own
registrations still win for the route validators. (The HTTP routes were
never affected - importing Elysia registers ~24 formats including
`uuid`, which I confirmed by asserting a non-uuid body gets a 422.)

A `frame_batch` now goes onto the stream in one pipelined round trip
instead of one per frame, which was a round trip each while holding up
the WebSocket message handler - and a farm produces roughly a frame per
printer per second. ioredis reports per-command pipeline errors in the
result tuple rather than throwing, so those are checked explicitly; a
pipeline where one XADD failed otherwise looks like success.

`frameTaskToStreamFields` and the compiled check live in their own
module because `src/queue/index.ts` opens a Valkey connection on import.
Keeping the wire format pure means `tests/frame_task_contract.test.ts`
asserts it with nothing running.

Tests, mirroring the worker's own so neither side can agree only with
itself: the exact key set, values carried through unchanged,
`print_job_id` present when there is a job and the key absent when there
is not, no camelCase key ever emitted, and the uuid and date-time
formats actually enforced - that last one fails if the format
registration is lost (everything rejected) or becomes a no-op
(nothing rejected). Plus `PrintAlertSchema`, which this repo will have
to parse once the fan-out route exists.

`scripts/e2e-ai-frames.sh` runs this producer against the real worker
binary over real Valkey and asserts the field names by reading them back
out of Valkey - rather than from either side's source, which is what
made the original bug invisible.

Two faults in the first draft of that script, both worth naming because
a test that passes for the wrong reason is worse than no test: it
stranded *every* entry on a dead consumer, so the ordinary delivery path
was never exercised at all; and it grepped the subscriber output for
"print_alert", which matched `redis-cli`'s own subscription
confirmation line and reported a published alert that did not exist.

src/queue/frame_task.ts

@@ -0,0 +1,54 @@
+import { TypeCompiler } from "@sinclair/typebox/compiler";
+import { FrameTaskSchema, type FrameTask } from "@continuum/ts-types";
+// Side-effect import, and it must come before the Compile() below: an
+// unregistered format fails rather than being skipped, so without this every
+// task is rejected with `Unknown format 'uuid'`.
+import "../validation/formats";
+
+/**
+ * The `stream:camera_frames` wire form, kept apart from `./index.ts` because
+ * that module opens a Valkey connection on import and this needs nothing but
+ * the schema. `tests/frame_task_contract.test.ts` asserts the exact field
+ * names against this module, with no infrastructure running.
+ */
+
+/**
+ * Compiled once at module load. The check is not for the sake of the
+ * database - a stream will accept any field name - it is to catch this end
+ * of the contract drifting from `schemas/continuum/v1/ai/frame_task.schema.json`.
+ *
+ * It is worth the line because the failure it catches is silent: this
+ * producer previously wrote `printerId`/`r2ObjectKey`/`capturedAt` while the
+ * worker read `printer_id` and nothing else, so every task was accepted,
+ * delivered, and attributed to a printer named "unknown". No error anywhere.
+ */
+export const checkFrameTask = TypeCompiler.Compile(FrameTaskSchema);
+
+/**
+ * Flattens a task into the flat field/value list XADD takes.
+ *
+ * Exported because this - the exact field names, and which of them are
+ * written at all - *is* the contract with continuum-ai-worker. Keeping it a
+ * pure function means `tests/frame_task_contract.test.ts` can assert the
+ * wire form without a Valkey connection, and means there is one place to
+ * look when the Rust mirror in `continuum-ai-worker/src/task.rs` disagrees.
+ */
+export function frameTaskToStreamFields(task: FrameTask): string[] {
+  // Field order is not significant to a consumer, which reads a map - but
+  // keeping it stable keeps `XRANGE` output readable by hand.
+  const fields = [
+    "farm_id", task.farm_id,
+    "printer_id", task.printer_id,
+    "r2_object_key", task.r2_object_key,
+    "captured_at", task.captured_at,
+  ];
+  // Omitted, not empty. `print_job_id` used to be written as `""` when there
+  // was no job; an empty string is a value, and a consumer that believed it
+  // would look up job id `""` or try to put it in a uuid column. A stream
+  // entry has no null, so "no value" means "no field".
+  if (task.print_job_id !== undefined) {
+    fields.push("print_job_id", task.print_job_id);
+  }
+  return fields;
+}
+

src/queue/index.ts

@@ -2,35 +2,55 @@
 // continuum-common for why the server is Valkey and not Redis. The `redis://`
 // scheme names the wire protocol, not the product.
 import Redis from "ioredis";
+import { CAMERA_FRAMES_STREAM, type FrameTask } from "@continuum/ts-types";
+import { checkFrameTask, frameTaskToStreamFields } from "./frame_task";
 
 export const valkey = new Redis(process.env.VALKEY_URL ?? "redis://localhost:6379", {
   maxRetriesPerRequest: 3,
 });
 
-export const CAMERA_FRAMES_STREAM = "stream:camera_frames";
+export { CAMERA_FRAMES_STREAM };
+export type { FrameTask };
+export { frameTaskToStreamFields } from "./frame_task";
 
-export interface FrameIngestionTask {
-  farmId: string;
-  printerId: string;
-  printJobId: string | null;
-  r2ObjectKey: string;
-  capturedAt: string;
+/** Pushes one frame-ingestion task for continuum-ai-worker to pick up. */
+export async function enqueueFrameIngestion(task: FrameTask): Promise<string> {
+  const [id] = await enqueueFrameIngestionBatch([task]);
+  return id!;
 }
 
-/** Pushes a frame-ingestion task for continuum-ai-worker to pick up via XREADGROUP. */
-export async function enqueueFrameIngestion(task: FrameIngestionTask): Promise<string> {
-  return valkey.xadd(
-    CAMERA_FRAMES_STREAM,
-    "*",
-    "farmId",
-    task.farmId,
-    "printerId",
-    task.printerId,
-    "printJobId",
-    task.printJobId ?? "",
-    "r2ObjectKey",
-    task.r2ObjectKey,
-    "capturedAt",
-    task.capturedAt,
-  ) as Promise<string>;
+/**
+ * Pushes a whole batch in one round trip.
+ *
+ * Frames arrive from a gateway as a `frame_batch` - one message covering
+ * several stills - and a farm produces one per printer per second or so.
+ * Enqueuing them one at a time, as this used to, costs a round trip each
+ * while holding up the WebSocket message handler.
+ */
+export async function enqueueFrameIngestionBatch(tasks: FrameTask[]): Promise<string[]> {
+  if (tasks.length === 0) return [];
+
+  const pipeline = valkey.pipeline();
+  for (const task of tasks) {
+    if (!checkFrameTask.Check(task)) {
+      const first = [...checkFrameTask.Errors(task)][0];
+      throw new Error(
+        `frame task does not match continuum/v1/ai/frame_task: ${first?.path} ${first?.message}`,
+      );
+    }
+    pipeline.xadd(CAMERA_FRAMES_STREAM, "*", ...frameTaskToStreamFields(task));
+  }
+
+  const results = await pipeline.exec();
+  if (!results) throw new Error("valkey pipeline returned no results");
+
+  return results.map(([err, id], index) => {
+    // ioredis reports per-command errors in the result tuple rather than
+    // throwing, so a pipeline where one XADD failed looks like success.
+    if (err) throw err;
+    if (typeof id !== "string") {
+      throw new Error(`XADD for frame ${index} returned ${typeof id}, expected an id`);
+    }
+    return id;
+  });
 }

src/routes/ws/edge.ts

@@ -9,7 +9,7 @@ import {
 } from "@continuum/ts-types";
 import { db } from "../../db";
 import { printers } from "../../db/schema";
-import { enqueueFrameIngestion } from "../../queue";
+import { enqueueFrameIngestionBatch } from "../../queue";
 
 /**
  * Edge-gateway uplink. A continuum-proxy gateway connects here with a
@@ -326,15 +326,23 @@ export const edgeWsRoute = new Elysia()
             return;
           }
 
-          for (const frame of message.frames) {
-            await enqueueFrameIngestion({
-              farmId: conn.farmId,
-              printerId,
-              printJobId: message.print_job_id ?? null,
-              r2ObjectKey: frame.object_key,
-              capturedAt: frame.captured_at,
-            });
-          }
+          // One round trip for the whole batch, and snake_case field names:
+          // this is the `stream:camera_frames` contract in
+          // `schemas/continuum/v1/ai/`, not an HTTP response, and the
+          // consumer at the other end is serde-derived Rust.
+          await enqueueFrameIngestionBatch(
+            message.frames.map((frame) => ({
+              farm_id: conn.farmId,
+              printer_id: printerId,
+              // Omitted rather than null when there is no job: a stream
+              // entry has no null, and `""` is a value a consumer would
+              // believe. `?? undefined` collapses the wire's "absent or
+              // null" into the one the schema describes.
+              ...(message.print_job_id != null ? { print_job_id: message.print_job_id } : {}),
+              r2_object_key: frame.object_key,
+              captured_at: frame.captured_at,
+            })),
+          );
           break;
         }
       }

src/validation/formats.ts

@@ -0,0 +1,37 @@
+import { FormatRegistry } from "@sinclair/typebox";
+
+/**
+ * Registers the string formats this codebase uses with TypeBox.
+ *
+ * TypeBox ships no format validators: `format` is a registry lookup, and an
+ * *unregistered* format does not pass - it fails. So a schema written with
+ * `format: "uuid"` and checked outside Elysia rejects every value, including
+ * valid ones. That is not theoretical; it is how
+ * `src/queue/frame_task.ts`'s compiled check came to refuse every frame task
+ * with `Unknown format 'uuid'`.
+ *
+ * Elysia registers a full set (`uuid`, `date-time` and ~20 more) as a side
+ * effect of importing it, which is why the HTTP routes validate correctly.
+ * Modules that deliberately do not import Elysia - anything that must stay
+ * free of its Valkey/HTTP weight for testing - have to import this instead.
+ *
+ * `Has` guards each one so that whichever of the two loads first wins, rather
+ * than this overriding Elysia's implementation for the route validators. The
+ * two agree on what these formats accept; `tests/frame_task_contract.test.ts`
+ * asserts that the active implementation actually rejects a malformed uuid,
+ * so this cannot quietly become a no-op.
+ */
+
+const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
+
+if (!FormatRegistry.Has("uuid")) {
+  FormatRegistry.Set("uuid", (value) => UUID.test(value));
+}
+
+if (!FormatRegistry.Has("date-time")) {
+  // Deliberately lenient: anything `Date.parse` understands. The producers
+  // are `new Date().toISOString()` in TypeScript and chrono's RFC 3339 in
+  // Rust, and rejecting a timestamp a printer actually sent would cost a
+  // frame for no gain.
+  FormatRegistry.Set("date-time", (value) => !Number.isNaN(Date.parse(value)));
+}