diff --git a/scripts/e2e-ai-frames.sh b/scripts/e2e-ai-frames.sh
new file mode 100755
index 0000000..75770a3
--- /dev/null
+++ b/scripts/e2e-ai-frames.sh
@@ -0,0 +1,248 @@
+#!/usr/bin/env bash
+# End-to-end test of the stream:camera_frames contract: the real
+# continuum-backend producer, real Valkey, and the real continuum-ai-worker
+# binary.
+#
+# tests/frame_task_contract.test.ts and continuum-ai-worker's task.rs tests
+# each prove one side agrees with the schema. Only this proves the two agree
+# with each other, which is the thing that was wrong: the backend wrote
+# `printerId` and the worker read `printer_id`, so every task was delivered,
+# acknowledged, and attributed to a printer called "unknown". Nothing failed,
+# because a stream entry is a flat string map and asking it for a field that
+# is not there is not an error.
+#
+# What it asserts:
+#   1. a task the backend enqueues is consumed and acknowledged
+#   2. an old-format camelCase entry is dead-lettered, not silently mishandled
+#   3. an entry left pending by a dead consumer is reclaimed and acknowledged
+#   4. nothing is published to channel:print_alerts - the inference stub
+#      abstains, and used to fabricate a 0.91 "spaghetti" for every frame
+#
+# Usage:
+#   scripts/dev-env.sh up
+#   scripts/e2e-ai-frames.sh
+#
+#   WORKER_DIR=../continuum-ai-worker
+set -euo pipefail
+
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$REPO_ROOT"
+
+WORKER_DIR="${WORKER_DIR:-$REPO_ROOT/../continuum-ai-worker}"
+VALKEY_URL="${VALKEY_URL:-redis://localhost:6379}"
+VALKEY_PORT="${VALKEY_PORT:-6379}"
+STREAM="stream:camera_frames"
+DEAD="stream:camera_frames:dead"
+GROUP="ai-worker"
+ALERTS="channel:print_alerts"
+export VALKEY_URL
+
+WORK="$(mktemp -d)"
+WORKER_PID=""
+SUB_PID=""
+FAILURES=0
+
+cli() { redis-cli -p "$VALKEY_PORT" "$@"; }
+
+new_uuid() {
+  if [ -r /proc/sys/kernel/random/uuid ]; then
+    cat /proc/sys/kernel/random/uuid
+  else
+    python3 -c 'import uuid; print(uuid.uuid4())'
+  fi
+}
+
+cleanup() {
+  local rc=$?
+  # Never `kill $PID` with an unset PID: `kill 0` signals the whole process
+  # group, which includes this script and whatever launched it.
+  for pid in "$WORKER_PID" "$SUB_PID"; do
+    [ -n "$pid" ] && kill -9 "$pid" 2>/dev/null || true
+  done
+  if [ -z "${KEEP_DATA:-}" ]; then
+    cli DEL "$STREAM" "$DEAD" >/dev/null 2>&1 || true
+  fi
+  if [ $rc -ne 0 ] || [ "$FAILURES" -ne 0 ]; then
+    echo
+    echo "--- worker log ---"; cat "$WORK/worker.log" 2>/dev/null || true
+  fi
+  rm -rf "$WORK"
+  exit $rc
+}
+trap cleanup EXIT
+
+fail() { echo "FAIL: $*" >&2; exit 1; }
+ok()   { echo "  ok: $*"; }
+bad()  { echo "  FAILED: $*" >&2; FAILURES=$((FAILURES + 1)); }
+
+# --- 0. preconditions -----------------------------------------------------
+
+echo "==> checking Valkey"
+cli ping >/dev/null 2>&1 || fail "no Valkey at $VALKEY_URL - run scripts/dev-env.sh up"
+
+# This test owns and then deletes these two keys. Refuse to touch a Valkey
+# that has real frames queued on it.
+EXISTING="$(cli XLEN "$STREAM" 2>/dev/null || echo 0)"
+[ "$EXISTING" = "0" ] || fail "$STREAM already has $EXISTING entries; refusing to run against live data"
+cli DEL "$STREAM" "$DEAD" >/dev/null
+ok "Valkey up, $STREAM empty"
+
+echo "==> building the worker"
+( cd "$WORKER_DIR" && cargo build --quiet 2>&1 | tail -5 ) || fail "the worker did not build"
+WORKER_BIN="$WORKER_DIR/target/debug/continuum-ai-worker"
+[ -x "$WORKER_BIN" ] || fail "no worker binary at $WORKER_BIN"
+ok "worker built"
+
+FARM="$(new_uuid)"; PRINTER="$(new_uuid)"; JOB="$(new_uuid)"
+
+# --- 1. an entry stranded by a dead consumer -----------------------------
+
+# Done first, and with COUNT 1, so exactly one entry goes through recovery
+# and everything added afterwards goes through the ordinary `>` read. An
+# earlier version stranded every entry on the stream, which meant the normal
+# delivery path was never exercised at all.
+echo "==> stranding one entry on a consumer that never acknowledges it"
+cli XGROUP CREATE "$STREAM" "$GROUP" 0 MKSTREAM >/dev/null 2>&1 || true
+cli XADD "$STREAM" '*' \
+  farm_id "$FARM" printer_id "$PRINTER" \
+  r2_object_key "farms/$FARM/frames/stranded.jpg" captured_at "2026-10-11T12:00:03Z" >/dev/null
+# Read as a consumer that then "dies": the entry is delivered and pending, so
+# no other consumer is offered it. Without recovery it sits there forever.
+cli XREADGROUP GROUP "$GROUP" dead-consumer COUNT 1 STREAMS "$STREAM" '>' >/dev/null
+STRANDED="$(cli XPENDING "$STREAM" "$GROUP" | head -1)"
+[ "${STRANDED:-0}" = "1" ] && ok "1 entry pending on a consumer that is gone" \
+  || bad "expected exactly 1 pending entry, XPENDING says '$STRANDED'"
+
+# --- 2. enqueue through the real producer --------------------------------
+
+echo "==> enqueuing two tasks through the backend's own producer"
+# The real `enqueueFrameIngestionBatch`, not a hand-written XADD: the point is
+# to test what the backend actually writes.
+bun -e "
+const { enqueueFrameIngestionBatch, valkey } = await import('./src/queue/index.ts');
+const ids = await enqueueFrameIngestionBatch([
+  { farm_id: '$FARM', printer_id: '$PRINTER', r2_object_key: 'farms/$FARM/frames/a.jpg', captured_at: '2026-10-11T12:00:00Z' },
+  { farm_id: '$FARM', printer_id: '$PRINTER', print_job_id: '$JOB', r2_object_key: 'farms/$FARM/frames/b.jpg', captured_at: '2026-10-11T12:00:01Z' },
+]);
+console.log(ids.join(' '));
+valkey.disconnect();
+" > "$WORK/ids.txt" 2>"$WORK/producer.err" || fail "the producer failed: $(cat "$WORK/producer.err")"
+ok "2 tasks enqueued"
+
+# The field names actually on the wire, read back from Valkey rather than from
+# either side's source. This is the assertion the two repos could not make
+# about each other before.
+echo "==> the field names on the wire"
+cli XRANGE "$STREAM" - + > "$WORK/xrange.txt"
+for field in farm_id printer_id r2_object_key captured_at print_job_id; do
+  grep -qx "$field" "$WORK/xrange.txt" || bad "field '$field' is not on the wire"
+done
+for camel in farmId printerId r2ObjectKey capturedAt printJobId; do
+  grep -qx "$camel" "$WORK/xrange.txt" && bad "camelCase field '$camel' is on the wire" || true
+done
+# Exactly one entry carries a job id; the others omit the field entirely
+# rather than writing "".
+JOB_FIELDS="$(grep -cx print_job_id "$WORK/xrange.txt" || true)"
+[ "$JOB_FIELDS" = "1" ] && ok "print_job_id present on one entry, omitted on the others" \
+  || bad "print_job_id appears $JOB_FIELDS times, expected 1"
+ok "every field is snake_case, as the schema says"
+
+# --- 3. an old-format entry ----------------------------------------------
+
+echo "==> adding one entry in the old camelCase format"
+# Exactly what the previous producer wrote. The worker must dead-letter it
+# rather than process it with defaults.
+cli XADD "$STREAM" '*' \
+  farmId "$FARM" printerId "$PRINTER" printJobId "" \
+  r2ObjectKey "farms/$FARM/frames/legacy.jpg" capturedAt "2026-10-11T12:00:02Z" >/dev/null
+ok "legacy entry added"
+
+# --- 4. watch the alert channel ------------------------------------------
+
+cli SUBSCRIBE "$ALERTS" > "$WORK/alerts.txt" 2>/dev/null &
+SUB_PID=$!
+sleep 0.4
+
+# --- 5. run the real worker ----------------------------------------------
+
+echo "==> running the worker"
+CONTINUUM_VALKEY_URL="$VALKEY_URL" \
+CONTINUUM_FRAME_STREAM="$STREAM" \
+CONTINUUM_DEAD_LETTER_STREAM="$DEAD" \
+CONTINUUM_CONSUMER_GROUP="$GROUP" \
+CONTINUUM_CONSUMER_NAME="e2e-worker" \
+CONTINUUM_BLOCK_MS=500 \
+CONTINUUM_CLAIM_IDLE_MS=1000 \
+CONTINUUM_MAX_DELIVERIES=5 \
+RUST_LOG=continuum_ai_worker=debug \
+  "$WORKER_BIN" > "$WORK/worker.log" 2>&1 &
+WORKER_PID=$!
+
+# Long enough for the claim-idle window (1s) to elapse and a recovery pass to
+# run after it.
+for _ in $(seq 1 40); do
+  PENDING="$(cli XPENDING "$STREAM" "$GROUP" | head -1)"
+  DEADLEN="$(cli XLEN "$DEAD")"
+  [ "${PENDING:-1}" = "0" ] && [ "${DEADLEN:-0}" -ge 1 ] && break
+  sleep 0.5
+done
+
+kill "$WORKER_PID" 2>/dev/null || true
+sleep 0.5
+ok "worker ran"
+
+# --- 6. assertions -------------------------------------------------------
+
+echo "==> what the worker did"
+grep -qi "panic" "$WORK/worker.log" && bad "the worker panicked" || ok "no panic"
+
+PENDING="$(cli XPENDING "$STREAM" "$GROUP" | head -1)"
+[ "${PENDING:-x}" = "0" ] && ok "nothing left pending: every entry was acknowledged" \
+  || bad "$PENDING entries are still pending"
+
+DEADLEN="$(cli XLEN "$DEAD")"
+[ "$DEADLEN" = "1" ] && ok "exactly one entry was dead-lettered" \
+  || bad "the dead-letter stream holds $DEADLEN entries, expected 1"
+
+cli XRANGE "$DEAD" - + > "$WORK/dead.txt"
+grep -q "dead_reason" "$WORK/dead.txt" || bad "the dead-lettered entry carries no reason"
+grep -q "missing required field 'farm_id'" "$WORK/dead.txt" \
+  && ok "the legacy camelCase entry was dead-lettered, naming the field it lacked" \
+  || bad "wrong dead-letter reason: $(grep -A1 dead_reason "$WORK/dead.txt" | head -4 | tr '\n' ' ')"
+# The original fields are kept, so it can be inspected and replayed.
+grep -q "farmId" "$WORK/dead.txt" && ok "the original entry's fields were preserved" \
+  || bad "the dead-lettered entry did not keep the original fields"
+
+grep -q "reclaimed a pending frame task" "$WORK/worker.log" \
+  && ok "the stranded entry was reclaimed from the dead consumer" \
+  || bad "no evidence the stranded entry was reclaimed"
+
+# The three well-formed entries (one reclaimed, two read normally) must have
+# been *parsed*, not merely acknowledged - a parse failure would have
+# dead-lettered them instead, and an acknowledged-but-unparsed entry is
+# exactly the old bug. `no detections` is only reachable past the parse.
+PARSED="$(grep -c "no detections" "$WORK/worker.log" || true)"
+[ "${PARSED:-0}" -ge 3 ] && ok "all 3 well-formed tasks parsed and scored ($PARSED)" \
+  || bad "only $PARSED tasks got past parsing, expected 3"
+
+# The whole point of the inference fix. The stub used to return a fixed
+# 0.91 "spaghetti" for every frame, including the empty slice it was always
+# handed, so a running worker was a false-alarm generator.
+# `redis-cli SUBSCRIBE` prints its own confirmation first - the three lines
+# "subscribe", the channel name, "1" - so grepping the file for the channel
+# name matches that and not a message. A delivered message is a line that is
+# exactly "message".
+if grep -qx "message" "$WORK/alerts.txt" 2>/dev/null; then
+  bad "an alert was published: $(grep -A2 -x message "$WORK/alerts.txt" | tail -2 | tr '\n' ' ')"
+else
+  ok "no alert was published - the inference stub abstained instead of inventing one"
+fi
+
+echo
+if [ "$FAILURES" -eq 0 ]; then
+  echo "PASS - the backend's producer and the worker's consumer agree on"
+  echo "       stream:camera_frames, against real Valkey."
+else
+  echo "FAIL - $FAILURES assertion(s) failed." >&2
+  exit 1
+fi
diff --git a/src/queue/frame_task.ts b/src/queue/frame_task.ts
new file mode 100644
index 0000000..c7fc031
--- /dev/null
+++ b/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;
+}
+
diff --git a/src/queue/index.ts b/src/queue/index.ts
index a11ad02..a78b371 100644
--- a/src/queue/index.ts
+++ b/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;
+  });
 }
diff --git a/src/routes/ws/edge.ts b/src/routes/ws/edge.ts
index 4459a8c..21e8a25 100644
--- a/src/routes/ws/edge.ts
+++ b/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;
         }
       }
diff --git a/src/validation/formats.ts b/src/validation/formats.ts
new file mode 100644
index 0000000..dff5ad8
--- /dev/null
+++ b/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)));
+}
diff --git a/tests/frame_task_contract.test.ts b/tests/frame_task_contract.test.ts
new file mode 100644
index 0000000..7fbc44d
--- /dev/null
+++ b/tests/frame_task_contract.test.ts
@@ -0,0 +1,176 @@
+import { describe, expect, test } from "bun:test";
+import { Value } from "@sinclair/typebox/value";
+import { FrameTaskSchema, PrintAlertSchema, type FrameTask } from "@continuum/ts-types";
+import { checkFrameTask, frameTaskToStreamFields } from "../src/queue/frame_task";
+
+/**
+ * The mirror of `continuum-ai-worker/src/task.rs`'s test module. Both sides
+ * assert the same field names against the same canonical schema
+ * (`schemas/continuum/v1/ai/frame_task.schema.json`), because the bug this
+ * replaced was invisible from either side alone: the producer wrote
+ * `printerId` and `r2ObjectKey`, the consumer read `printer_id` and nothing
+ * else, every task was delivered and acknowledged, and every frame was
+ * attributed to a printer called "unknown".
+ *
+ * A stream entry is a flat string map, so there is nothing to parse and
+ * nothing to fail - which is exactly why the field names need a test rather
+ * than a type.
+ */
+
+const FARM = "11111111-1111-1111-1111-111111111111";
+const PRINTER = "22222222-2222-2222-2222-222222222222";
+const JOB = "33333333-3333-3333-3333-333333333333";
+
+const task: FrameTask = {
+  farm_id: FARM,
+  printer_id: PRINTER,
+  r2_object_key: "farms/x/frames/1.jpg",
+  captured_at: "2026-10-11T12:00:00Z",
+};
+
+/** The flat list XADD takes, as the `{field: value}` map a consumer sees. */
+function asMap(fields: string[]): Record<string, string> {
+  const map: Record<string, string> = {};
+  for (let i = 0; i < fields.length; i += 2) map[fields[i]!] = fields[i + 1]!;
+  return map;
+}
+
+describe("the stream:camera_frames wire form", () => {
+  test("is snake_case, and exactly the keys the schema names", () => {
+    // Written out rather than derived from the schema: a test that reads the
+    // field names from the same place the implementation does cannot catch
+    // them both being wrong.
+    expect(Object.keys(asMap(frameTaskToStreamFields(task))).sort()).toEqual([
+      "captured_at",
+      "farm_id",
+      "printer_id",
+      "r2_object_key",
+    ]);
+  });
+
+  test("carries the values through unchanged", () => {
+    expect(asMap(frameTaskToStreamFields(task))).toEqual({
+      farm_id: FARM,
+      printer_id: PRINTER,
+      r2_object_key: "farms/x/frames/1.jpg",
+      captured_at: "2026-10-11T12:00:00Z",
+    });
+  });
+
+  test("includes print_job_id when there is one", () => {
+    const fields = asMap(frameTaskToStreamFields({ ...task, print_job_id: JOB }));
+    expect(fields.print_job_id).toBe(JOB);
+  });
+
+  /**
+   * The counterpart of the Rust `an_empty_job_id_is_no_job_id` test. The old
+   * producer wrote `""`; a stream entry has no null, so the only way to say
+   * "no job" is to leave the field out.
+   */
+  test("omits print_job_id entirely rather than writing an empty string", () => {
+    const fields = frameTaskToStreamFields(task);
+    expect(fields).not.toContain("print_job_id");
+    expect(fields).not.toContain("");
+    expect("print_job_id" in asMap(fields)).toBe(false);
+  });
+
+  test("emits an even number of field/value entries", () => {
+    // XADD takes pairs; an odd count is a wrong-arity error at the server
+    // rather than at the type level.
+    expect(frameTaskToStreamFields(task).length % 2).toBe(0);
+    expect(frameTaskToStreamFields({ ...task, print_job_id: JOB }).length % 2).toBe(0);
+  });
+
+  test("never emits a camelCase key", () => {
+    // The actual regression. Both field name sets, side by side, so the
+    // failure message names the one that came back.
+    const keys = Object.keys(asMap(frameTaskToStreamFields({ ...task, print_job_id: JOB })));
+    for (const camel of ["farmId", "printerId", "printJobId", "r2ObjectKey", "capturedAt"]) {
+      expect(keys).not.toContain(camel);
+    }
+  });
+});
+
+describe("the producer's schema check", () => {
+  test("accepts a well-formed task", () => {
+    expect(checkFrameTask.Check(task)).toBe(true);
+  });
+
+  test("rejects a task with a camelCase field instead of the real one", () => {
+    const { printer_id, ...rest } = task;
+    expect(checkFrameTask.Check({ ...rest, printerId: printer_id })).toBe(false);
+  });
+
+  test("rejects a task missing any required field", () => {
+    for (const key of Object.keys(task) as Array<keyof FrameTask>) {
+      const partial = { ...task };
+      delete partial[key];
+      expect(checkFrameTask.Check(partial)).toBe(false);
+    }
+  });
+
+  test("rejects an empty object key", () => {
+    // `minLength: 1`: an empty key would send the worker looking for the
+    // object named "".
+    expect(checkFrameTask.Check({ ...task, r2_object_key: "" })).toBe(false);
+  });
+
+  /**
+   * Guards `src/validation/formats.ts`. TypeBox treats an unregistered
+   * format as a failure, not as "skip", so this check started life rejecting
+   * every task with `Unknown format 'uuid'`. These two assertions fail if
+   * the registration is lost (everything rejected) *or* if it became a no-op
+   * that accepts anything (nothing rejected).
+   */
+  test("the uuid format is actually enforced, not merely declared", () => {
+    expect(checkFrameTask.Check(task)).toBe(true);
+    expect(checkFrameTask.Check({ ...task, printer_id: "unknown" })).toBe(false);
+    expect(checkFrameTask.Check({ ...task, farm_id: "11111111-1111-1111-1111-11111111111" })).toBe(
+      false,
+    );
+  });
+
+  test("the date-time format is actually enforced", () => {
+    expect(checkFrameTask.Check({ ...task, captured_at: "yesterday" })).toBe(false);
+  });
+
+  test("rejects a print_job_id that is an empty string", () => {
+    // Belt and braces with the omit-don't-empty rule: if a caller does pass
+    // `""`, the producer refuses rather than writing it.
+    expect(checkFrameTask.Check({ ...task, print_job_id: "" })).toBe(false);
+  });
+});
+
+describe("the channel:print_alerts payload", () => {
+  // Asserted here as well as in the worker because the backend is the side
+  // that will have to parse these once the fan-out route exists, and this is
+  // where that expectation should be recorded.
+  const alert = {
+    printer_id: PRINTER,
+    label: "spaghetti",
+    confidence: 0.91,
+    captured_at: "2026-10-11T12:00:00Z",
+    r2_object_key: "farms/x/frames/1.jpg",
+  };
+
+  test("the shape the worker publishes validates", () => {
+    expect(Value.Check(PrintAlertSchema, alert)).toBe(true);
+  });
+
+  test("print_job_id is optional, and null is not the same as absent", () => {
+    expect(Value.Check(PrintAlertSchema, { ...alert, print_job_id: JOB })).toBe(true);
+    expect(Value.Check(PrintAlertSchema, { ...alert, print_job_id: null })).toBe(false);
+  });
+
+  test("confidence outside 0..1 is refused", () => {
+    expect(Value.Check(PrintAlertSchema, { ...alert, confidence: 1.5 })).toBe(false);
+    expect(Value.Check(PrintAlertSchema, { ...alert, confidence: -0.1 })).toBe(false);
+  });
+
+  test("an alert carries the frame it came from, not just a label", () => {
+    // Without `r2_object_key` an operator gets "printer X is failing" and no
+    // way to look at the evidence.
+    const { r2_object_key, ...withoutFrame } = alert;
+    expect(Value.Check(PrintAlertSchema, withoutFrame)).toBe(false);
+  });
+});