Make the frame consumer correct, and stop it inventing failures

Six bugs, found while writing down the `stream:camera_frames` contract
this worker is one end of.

**It never read the task.** It pulled one field, `printer_id`, from a
producer writing `printerId`, and substituted the literal string
"unknown" when that missed - which was always. Every other field,
including `r2_object_key`, the only thing saying *which picture*, was
never looked at. Nothing errored: a stream entry is a flat string map,
and asking it for an absent field is not a failure. `src/task.rs` is now
the Rust mirror of
`continuum-schemas/schemas/continuum/v1/ai/frame_task.schema.json` and a
missing field is an error, not a default. Its tests include the
regression directly: a camelCase entry must fail to parse rather than
yield defaults.

**It invented print failures.** `inference::detect` returned a fixed
`spaghetti` at 0.91 confidence for every input, including the empty
slice it was always handed, so a running worker declared a failure for
every frame of every print on evidence it had never looked at. Harmless
only because nothing subscribes to the alert channel - the moment
anything does, it is a false-alarm generator pointed at an operator. It
now abstains. An unimplemented model returning "nothing wrong" is a safe
default; one returning "failure" is not.

**It discarded the backlog.** The consumer group was created at `$`,
meaning "only entries added from now on", so a worker that was down
while a farm printed came back up and silently skipped every frame it
had missed. For a failure detector that is the opposite of the job. Now
`0`.

**It stranded entries forever.** An entry delivered to a consumer that
died before acking stays pending: it has been handed out, so no other
consumer is offered it, and nothing retried it. Now reclaimed via
`XPENDING`/`XCLAIM` once idle past `CONTINUUM_CLAIM_IDLE_MS`, and
dead-lettered onto `stream:camera_frames:dead` - with the reason and the
original fields, so it can be inspected and replayed - once it has been
delivered `CONTINUUM_MAX_DELIVERIES` times, so a poison entry can
neither be retried forever nor block the queue.

**One bad moment killed it.** `xread_options(...).await?` and
`reporter.report(...).await?` both propagated out of the loop, so a
single transient Valkey error ended the process. Reads now retry with a
growing delay and give up only after ten consecutive failures; a report
failure leaves the entry unacknowledged for redelivery rather than
taking the worker down with it. The `XACK` result is no longer discarded
with `let _: RedisResult<i64>`.

**`report` would panic** on `detections[0]` for an empty slice. The one
caller checks, which makes it a trap for the next one rather than a
live bug; it now picks the highest-confidence detection and returns
early on none.

Also corrects two comments that were simply false: `reporter` claimed
`channel:print_alerts` is "what continuum-backend's WebSocket fan-out
subscribes to" - nothing anywhere subscribes to it - and the README
advertised the fabricated detection as a deliberate aid to
understanding the pipeline. The README now states plainly what is real
(the loop, tested) and what is not (no frame fetch, no model, no
subscriber, no persistence), because "the plumbing works and detects
nothing" is the useful thing to know before enabling this.

Verified: 10 unit tests, zero clippy warnings, and
`continuum-backend/scripts/e2e-ai-frames.sh` - the real producer against
this real binary over real Valkey - asserting a well-formed task is
consumed and acked, a legacy camelCase entry is dead-lettered naming the
field it lacks, an entry stranded on a dead consumer is reclaimed, and
nothing is published to the alert channel.

.env.example

@@ -1,6 +1,26 @@
-CONTINUUM_REDIS_URL=redis://localhost:6379
+# Valkey, not Redis (licensing - see continuum-common/docs/LICENSES.md).
+# The `redis://` scheme is the RESP wire protocol; redis-rs has no
+# `valkey://` scheme to use instead.
+CONTINUUM_VALKEY_URL=redis://localhost:6379
+
 CONTINUUM_FRAME_STREAM=stream:camera_frames
+# Where entries go that must not be retried: a malformed task, or one
+# redelivered CONTINUUM_MAX_DELIVERIES times. A separate stream so a poison
+# entry can neither block the queue nor vanish.
+CONTINUUM_DEAD_LETTER_STREAM=stream:camera_frames:dead
+
 CONTINUUM_CONSUMER_GROUP=ai-worker
+# Must be unique per process: two consumers sharing a name share a pending
+# list, so each would reclaim entries the other is working on.
 CONTINUUM_CONSUMER_NAME=ai-worker-1
 
+# How long XREADGROUP blocks waiting for new entries.
+CONTINUUM_BLOCK_MS=5000
+# How long an unacknowledged entry must sit before another consumer may claim
+# it. Must comfortably exceed the time to process one frame - once inference
+# is real that is a model run, not the no-op it is today.
+CONTINUUM_CLAIM_IDLE_MS=60000
+# Deliveries before an entry is dead-lettered instead of retried again.
+CONTINUUM_MAX_DELIVERIES=5
+
 RUST_LOG=info,continuum_ai_worker=debug

Cargo.toml

@@ -3,7 +3,10 @@ name = "continuum-ai-worker"
 version = "0.1.0"
 edition = "2021"
 description = "Continuum AI worker - consumes camera frame batches from Redis Streams and runs print-failure detection"
-license = "UNLICENSED"
+# Proprietary: deliberately not an SPDX id, so cargo-deny treats it as
+# first-party. See docs/LICENSES.md in continuum-common.
+license = "LicenseRef-Proprietary"
+publish = false
 
 [[bin]]
 name = "continuum-ai-worker"

README.md

@@ -1,31 +1,77 @@
 # continuum-ai-worker
 
 Computer-vision failure detection worker for the Continuum print farm
-platform. Consumes camera frame batches pushed onto the Redis Stream
-`stream:camera_frames` (by `continuum-backend`'s `src/queue/`) and reports
-failures.
+platform. Consumes camera frame tasks from the Valkey Stream
+`stream:camera_frames` (written by `continuum-backend`'s `src/queue/`) and
+reports failures.
 
-## This is a learning-stage boilerplate
+## What is real and what is not
 
-Real ONNX Runtime inference (the `ort` crate) and the Postgres audit write
-(`sqlx`) are **not** implemented yet - `src/inference/mod.rs` always
-"detects" one fixed example failure, and `src/reporter/mod.rs` only prints
-where the Postgres write would go. The Redis Streams consumer-group loop in
-`main.rs` and the Redis pub/sub alert in `reporter/` are real. This lets the
-whole pipeline (read a frame, "detect", report, ack) run and be understood
-before adding a real model or a SQL database on top.
+The consumer-group loop is real and robust: it reads, parses, recovers
+entries stranded by a dead consumer, dead-letters what it cannot process,
+and acknowledges. The *detection* is not.
 
-## Pipeline
+- `src/inference/mod.rs` **abstains**. It has no model and it is handed no
+  frame, so it returns no detections. It used to return a fixed
+  `spaghetti` at 0.91 confidence for every call - including the empty slice
+  it was always given - which made a running worker a false-alarm generator
+  declaring a failure for every frame of every print, on evidence it had
+  never looked at. A stub that returns "nothing wrong" is a safe default; a
+  stub that returns "failure" is not.
+- **Nothing fetches the frame.** `FrameTask::r2_object_key` says where the
+  JPEG is in R2 and no code downloads it. This is the next piece of work and
+  the reason inference cannot do anything yet.
+- `src/reporter/mod.rs` publishes a real `PrintAlert` to
+  `channel:print_alerts`, but **nothing subscribes to that channel**, and
+  PUBLISH has no persistence - so an alert published today goes nowhere. The
+  fan-out is a `continuum-backend` route that does not exist yet.
+- **Nothing is persisted.** A detection's permanent record belongs in
+  Postgres; `frame_embeddings` exists for the vectors, with a pgvector
+  ivfflat index, and has never been written to.
 
-```
-continuum-proxy --(frames)--> continuum-backend --(XADD)--> stream:camera_frames
-                                                                    |
-                                                          XREADGROUP (this worker)
-                                                                    v
-                                                     inference::detect (stub for now)
-                                                                    v
-                                                 reporter::report (Redis pub/sub - real)
-```
+In short: the plumbing works and is tested; the worker does not yet detect
+anything. Do not enable it expecting alerts, and do not wire a console to
+`channel:print_alerts` expecting the alerts to mean something.
+
+## The contract
+
+`stream:camera_frames` and `channel:print_alerts` are the only interface
+between this worker and `continuum-backend`. Both are specified in
+`continuum-schemas/schemas/continuum/v1/ai/`, which is canonical:
+
+- `frame_task.schema.json` - one stream entry. A **flat map of strings**:
+  a stream entry has no nesting, no numbers and no null, which is why a
+  field with no value is *omitted* from the XADD rather than written as
+  `""`.
+- `print_alert.schema.json` - the JSON payload of a PUBLISH.
+
+`src/task.rs` is the Rust mirror; `ts-types/src/schemas/ai.ts` is the
+TypeScript one. Change the JSON Schema first, then both mirrors.
+
+This is written down because it drifted when it was not: the backend wrote
+`printerId`, `farmId`, `r2ObjectKey` and `capturedAt`, and this worker read
+a single field called `printer_id` and ignored the rest. Every task was
+delivered, acknowledged and attributed to a printer called `"unknown"`, and
+nothing errored - asking a string map for a field that is not there is not a
+failure. Both ends now have a contract test, and
+`continuum-backend/scripts/e2e-ai-frames.sh` runs the real producer against
+this real binary over real Valkey.
+
+## Delivery semantics
+
+| situation | what happens |
+| --- | --- |
+| a well-formed task | scored, then `XACK`ed |
+| a malformed task | copied to `stream:camera_frames:dead` with a reason, then `XACK`ed - it will never become valid, and retrying it would block the queue |
+| reporting fails | left unacknowledged, reclaimed after `CONTINUUM_CLAIM_IDLE_MS` |
+| a consumer dies mid-task | its pending entries are reclaimed by `XPENDING`/`XCLAIM` on the next pass |
+| a task fails `CONTINUUM_MAX_DELIVERIES` times | dead-lettered, so a poison entry cannot be redelivered forever |
+| Valkey blips | logged and retried with a growing delay; the worker exits only after `MAX_CONSECUTIVE_FAILURES` consecutive read failures |
+
+The consumer group is created at `0`, not `$`: `$` means "only entries added
+from now on", so a worker that was down while a farm printed would come back
+up and silently skip every frame it missed. For a failure detector, dropping
+the backlog is the opposite of the job.
 
 ## Getting started
 
@@ -34,12 +80,51 @@ cp .env.example .env
 cargo run
 ```
 
+```bash
+cargo test                                      # the task-parsing contract
+../continuum-backend/scripts/e2e-ai-frames.sh   # against real Valkey
+```
+
+## Configuration
+
+| variable | default | meaning |
+| --- | --- | --- |
+| `CONTINUUM_VALKEY_URL` | `redis://localhost:6379` | RESP server (Valkey, not Redis - see `continuum-common/docs/LICENSES.md`) |
+| `CONTINUUM_FRAME_STREAM` | `stream:camera_frames` | the work queue |
+| `CONTINUUM_DEAD_LETTER_STREAM` | `stream:camera_frames:dead` | where unprocessable entries go |
+| `CONTINUUM_CONSUMER_GROUP` | `ai-worker` | consumer group name |
+| `CONTINUUM_CONSUMER_NAME` | `ai-worker-1` | must be unique per process |
+| `CONTINUUM_BLOCK_MS` | `5000` | how long `XREADGROUP` blocks |
+| `CONTINUUM_CLAIM_IDLE_MS` | `60000` | how long before a pending entry may be reclaimed; must exceed the time to process one frame |
+| `CONTINUUM_MAX_DELIVERIES` | `5` | deliveries before dead-lettering |
+
+## Pipeline
+
+```
+continuum-proxy --(frame_batch)--> continuum-backend --(XADD)--> stream:camera_frames
+       |                                                                  |
+  no frame capture                                            XREADGROUP (this worker)
+  path yet either                                                         v
+                                                              FrameTask::from_stream_fields
+                                                                          v
+                                                       (missing: fetch r2_object_key from R2)
+                                                                          v
+                                                          inference::detect -- abstains
+                                                                          v
+                                       reporter::report -> channel:print_alerts (no subscriber)
+```
+
+Both ends of this pipeline are stubs: `continuum-proxy` restreams cameras
+through go2rtc and has no frame-capture path, so nothing sends `frame_batch`
+and the stream is empty in practice. The middle is what this repo tests.
+
 ## Structure
 
 ```
 src/
-  main.rs          Consumer-group loop: XREADGROUP -> detect -> report -> XACK
-  config.rs        Loads settings from environment variables
-  inference/       Detection stub - real ONNX Runtime inference goes here later
-  reporter/        Redis pub/sub alert (real) + a note for the Postgres write (later)
+  main.rs          Consumer-group loop: recover -> read -> parse -> score -> settle
+  task.rs          FrameTask: the Rust mirror of the frame_task schema
+  config.rs        Settings from environment variables
+  inference/       Detection - abstains; real ONNX Runtime inference goes here
+  reporter/        Valkey pub/sub alert (real, but unsubscribed)
 ```

deny.toml

@@ -0,0 +1,41 @@
+# Licence policy for a closed-source commercial product.
+# See docs/LICENSES.md in continuum-common for the rationale and the
+# non-cargo dependencies (external binaries, ML models) this file cannot see.
+#
+# Run: cargo deny check licenses
+
+[licenses]
+# Permissive only. Anything copyleft reaching a shipped binary would make
+# the product undistributable without releasing our source.
+allow = [
+    "MIT",
+    "Apache-2.0",
+    "Apache-2.0 WITH LLVM-exception",
+    "BSD-2-Clause",
+    "BSD-3-Clause",
+    "ISC",
+    "Zlib",
+    "Unicode-3.0",
+    "Unlicense",
+    "MPL-2.0",
+    "CC0-1.0",
+    "BlueOak-1.0.0",
+]
+confidence-threshold = 0.9
+
+# Our own crates declare `LicenseRef-Proprietary`, which is deliberately not
+# an SPDX identifier and must never appear in the allow list above - a
+# third-party crate claiming it would be a real finding. They are recognised
+# as first-party by `publish = false` in their manifests.
+private = { ignore = true }
+
+[bans]
+deny = []
+
+[advisories]
+yanked = "deny"
+
+# NOT VISIBLE TO THIS FILE: the ONNX model weights loaded at runtime carry
+# their own licence, independent of the `ort` crate (MIT). Ultralytics YOLO
+# and Obico's print-failure models are AGPL-3.0 and cannot ship in a
+# commercial product. See docs/LICENSES.md.

src/config.rs

@@ -2,19 +2,39 @@ use std::env;
 
 #[derive(Debug, Clone)]
 pub struct Config {
-    pub redis_url: String,
+    pub valkey_url: String,
     pub frame_stream: String,
+    /// Where entries go that must not be retried: a malformed task, or one
+    /// that has been redelivered too many times. A separate stream so a
+    /// poison message can neither block the queue nor vanish.
+    pub dead_letter_stream: String,
     pub consumer_group: String,
     pub consumer_name: String,
+    /// How long XREADGROUP blocks waiting for new entries.
+    pub block_ms: usize,
+    /// How long an unacknowledged entry must sit before another consumer may
+    /// claim it. Must comfortably exceed the time it takes to process one
+    /// frame, or two consumers will work the same entry.
+    pub claim_idle_ms: usize,
+    /// How many times an entry may be delivered before it is dead-lettered.
+    /// Without a cap, an entry that always fails is redelivered forever.
+    pub max_deliveries: usize,
 }
 
 impl Config {
     pub fn from_env() -> anyhow::Result<Self> {
         Ok(Self {
-            redis_url: env_or("CONTINUUM_REDIS_URL", "redis://localhost:6379"),
+            valkey_url: env_or("CONTINUUM_VALKEY_URL", "redis://localhost:6379"),
             frame_stream: env_or("CONTINUUM_FRAME_STREAM", "stream:camera_frames"),
+            dead_letter_stream: env_or(
+                "CONTINUUM_DEAD_LETTER_STREAM",
+                "stream:camera_frames:dead",
+            ),
             consumer_group: env_or("CONTINUUM_CONSUMER_GROUP", "ai-worker"),
             consumer_name: env_or("CONTINUUM_CONSUMER_NAME", "ai-worker-1"),
+            block_ms: env_num("CONTINUUM_BLOCK_MS", 5_000)?,
+            claim_idle_ms: env_num("CONTINUUM_CLAIM_IDLE_MS", 60_000)?,
+            max_deliveries: env_num("CONTINUUM_MAX_DELIVERIES", 5)?,
         })
     }
 }
@@ -22,3 +42,14 @@ impl Config {
 fn env_or(key: &str, default: &str) -> String {
     env::var(key).unwrap_or_else(|_| default.to_string())
 }
+
+fn env_num(key: &str, default: usize) -> anyhow::Result<usize> {
+    match env::var(key) {
+        Err(_) => Ok(default),
+        // Refuse rather than fall back to the default: a typo in a tuning
+        // value should not quietly run with different tuning.
+        Ok(raw) => raw
+            .parse()
+            .map_err(|_| anyhow::anyhow!("{key} must be a non-negative integer, got '{raw}'")),
+    }
+}

src/inference/mod.rs

@@ -7,13 +7,59 @@ pub struct Detection {
 }
 
 /// Stand-in for real ONNX Runtime inference (the `ort` crate, using a model
-/// trained on failure images). Always reports one fixed "failure" so the
-/// rest of the pipeline - reporting, acking the stream entry - has
-/// something to do and can be tested without a real model file.
+/// trained on failure images).
 ///
-/// Swap this out once you're comfortable with the pipeline around it:
-/// decode `frame_bytes` with the `image` crate, run it through an `ort`
-/// session, and turn the model's output into real `Detection`s.
-pub fn detect(_frame_bytes: &[u8]) -> Vec<Detection> {
-    vec![Detection { label: "spaghetti".to_string(), confidence: 0.91 }]
+/// **It abstains.** Given no frame bytes it returns no detections, and since
+/// nothing fetches frames yet, that is every call.
+///
+/// This used to return a fixed `spaghetti` at 0.91 confidence for every
+/// input, including the empty slice it was always handed, on the reasoning
+/// that the rest of the pipeline needed something to do. The effect was a
+/// worker that declared a print failure for every frame of every print, on
+/// evidence it had never looked at. A stub that fabricates findings is worse
+/// than one that does nothing: the moment anything subscribes to the alert
+/// channel, it is a false-alarm generator pointed at an operator, and the
+/// pipeline it was meant to exercise is exercised with data that proves
+/// nothing about it.
+///
+/// Tests that need a detection should construct one directly rather than
+/// making this lie.
+///
+/// To finish this: decode `frame_bytes` with the `image` crate, run it
+/// through an `ort` session, and turn the model's output into `Detection`s.
+/// The `r2_object_key` that says where the bytes are is already carried
+/// through `FrameTask`; what is missing is the fetch.
+pub fn detect(frame_bytes: &[u8]) -> Vec<Detection> {
+    if frame_bytes.is_empty() {
+        return Vec::new();
+    }
+
+    // Unreachable until something fetches frames, and deliberately not a
+    // fabricated detection when it becomes reachable - an unimplemented
+    // model that returns "nothing wrong" is a safe default; one that returns
+    // "failure" is not.
+    tracing::warn!(
+        bytes = frame_bytes.len(),
+        "inference is not implemented; frame scored as no-detection"
+    );
+    Vec::new()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn abstains_when_there_is_no_frame() {
+        assert!(detect(&[]).is_empty());
+    }
+
+    /// The regression test for the fabricated-alert bug: no input may ever
+    /// produce a detection from the stub.
+    #[test]
+    fn never_invents_a_detection() {
+        assert!(detect(&[]).is_empty());
+        assert!(detect(&[0xFF, 0xD8, 0xFF]).is_empty());
+        assert!(detect(&vec![0u8; 1024]).is_empty());
+    }
 }

src/main.rs

@@ -1,13 +1,48 @@
 mod config;
 mod inference;
 mod reporter;
+mod task;
 
-use redis::streams::{StreamKey, StreamReadOptions, StreamReadReply};
+use std::collections::HashMap;
+use std::time::Duration;
+
+use redis::streams::{
+    StreamClaimReply, StreamId, StreamKey, StreamPendingCountReply, StreamRangeReply,
+    StreamReadOptions, StreamReadReply,
+};
 use redis::AsyncCommands;
 use tracing_subscriber::EnvFilter;
 
 use crate::config::Config;
 use crate::reporter::Reporter;
+use crate::task::FrameTask;
+
+/// Entries already on the stream when the group is first created.
+///
+/// `0`, not `$`. `$` means "only entries added from now on", which silently
+/// discards whatever is already queued - so a worker that is down while a
+/// farm prints comes back up and never looks at the frames it missed. For a
+/// failure detector, dropping the backlog is the opposite of the job.
+const GROUP_START_ID: &str = "0";
+
+/// How many consecutive loop failures to tolerate before giving up. A
+/// transient Valkey blip should not kill the worker; a permanently wrong URL
+/// should not have it spin and log forever.
+const MAX_CONSECUTIVE_FAILURES: usize = 10;
+
+/// What to do with an entry once it has been looked at.
+enum Outcome {
+    /// Processed. Acknowledge it.
+    Done,
+    /// Will never succeed - a malformed task. Dead-letter it and acknowledge,
+    /// because redelivering it would block the queue behind something that
+    /// cannot be fixed by trying again.
+    Poison(String),
+    /// Failed for a reason that might not recur. Leave it unacknowledged so
+    /// `recover_pending` reclaims it later; it is dead-lettered once it has
+    /// been delivered `max_deliveries` times.
+    Retry(String),
+}
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -18,43 +53,272 @@ async fn main() -> anyhow::Result<()> {
     dotenvy::dotenv().ok();
     let config = Config::from_env()?;
 
-    let client = redis::Client::open(config.redis_url.clone())?;
+    // The `redis` crate is a RESP client; the server it talks to is Valkey.
+    // See continuum-common/docs/LICENSES.md for why.
+    let client = redis::Client::open(config.valkey_url.clone())?;
     let mut conn = client.get_multiplexed_tokio_connection().await?;
-    let mut reporter = Reporter::connect(&config.redis_url).await?;
+    let mut reporter = Reporter::connect(&config.valkey_url).await?;
 
-    // Consumer groups have to exist before you can read from them. This
-    // fails with "BUSYGROUP" if it already exists - that's fine, ignore it.
-    let _: redis::RedisResult<String> = conn
-        .xgroup_create_mkstream(&config.frame_stream, &config.consumer_group, "$")
-        .await;
+    // Consumer groups have to exist before you can read from them. BUSYGROUP
+    // means it already does, which is the normal case on every restart.
+    match conn
+        .xgroup_create_mkstream::<_, _, _, String>(
+            &config.frame_stream,
+            &config.consumer_group,
+            GROUP_START_ID,
+        )
+        .await
+    {
+        Ok(_) => tracing::info!(group = %config.consumer_group, "created consumer group"),
+        Err(err) if err.code() == Some("BUSYGROUP") => {
+            tracing::debug!("consumer group already exists")
+        }
+        // Anything else - a wrong URL, no permission, a key of the wrong
+        // type - is worth failing on rather than entering a read loop that
+        // cannot possibly work.
+        Err(err) => return Err(err).context_msg("could not create the consumer group"),
+    }
 
-    tracing::info!(stream = %config.frame_stream, group = %config.consumer_group, "waiting for camera frames");
+    tracing::info!(
+        stream = %config.frame_stream,
+        group = %config.consumer_group,
+        consumer = %config.consumer_name,
+        "waiting for camera frames"
+    );
 
     let read_opts = StreamReadOptions::default()
         .group(&config.consumer_group, &config.consumer_name)
-        .block(5000);
+        .block(config.block_ms);
+
+    let mut consecutive_failures = 0usize;
 
     loop {
-        let reply: StreamReadReply = conn.xread_options(&[&config.frame_stream], &[">"], &read_opts).await?;
+        // Anything a dead consumer left behind, first: an entry stuck pending
+        // is one a printer may have been failing during.
+        if let Err(err) = recover_pending(&mut conn, &mut reporter, &config).await {
+            tracing::warn!(error = %err, "could not recover pending entries this pass");
+        }
+
+        let reply: StreamReadReply = match conn
+            .xread_options(&[&config.frame_stream], &[">"], &read_opts)
+            .await
+        {
+            Ok(reply) => reply,
+            Err(err) => {
+                consecutive_failures += 1;
+                tracing::error!(
+                    error = %err,
+                    attempt = consecutive_failures,
+                    "reading the frame stream failed"
+                );
+                if consecutive_failures >= MAX_CONSECUTIVE_FAILURES {
+                    // Previously this was `?`, so the first transient error
+                    // ended the process.
+                    anyhow::bail!("gave up after {consecutive_failures} consecutive read failures");
+                }
+                tokio::time::sleep(Duration::from_millis(500 * consecutive_failures as u64)).await;
+                continue;
+            }
+        };
+        consecutive_failures = 0;
 
         for StreamKey { key, ids } in reply.keys {
             for entry in ids {
-                let printer_id = printer_id_field(&entry.map);
-
-                let detections = inference::detect(&[]);
-                if !detections.is_empty() {
-                    reporter.report(&printer_id, &detections).await?;
-                }
-
-                let _: redis::RedisResult<i64> = conn.xack(&key, &config.consumer_group, &[entry.id.as_str()]).await;
+                // One entry's failure must not skip the rest of the batch,
+                // and must not leave the loop.
+                handle_and_settle(&mut conn, &mut reporter, &config, &key, &entry).await;
             }
         }
     }
 }
 
-fn printer_id_field(fields: &std::collections::HashMap<String, redis::Value>) -> String {
-    match fields.get("printer_id") {
-        Some(redis::Value::BulkString(bytes)) => String::from_utf8_lossy(bytes).to_string(),
-        _ => "unknown".to_string(),
+/// Processes one entry and then acknowledges, dead-letters or leaves it
+/// pending according to the outcome.
+async fn handle_and_settle(
+    conn: &mut redis::aio::MultiplexedConnection,
+    reporter: &mut Reporter,
+    config: &Config,
+    stream: &str,
+    entry: &StreamId,
+) {
+    match process(reporter, &entry.map).await {
+        Outcome::Done => ack(conn, config, stream, &entry.id).await,
+        Outcome::Poison(reason) => {
+            tracing::error!(id = %entry.id, reason = %reason, "dead-lettering a malformed frame task");
+            dead_letter(conn, config, &entry.id, &entry.map, &reason).await;
+            ack(conn, config, stream, &entry.id).await;
+        }
+        Outcome::Retry(reason) => {
+            // Deliberately not acknowledged: it stays in the group's pending
+            // list and `recover_pending` will claim it after `claim_idle_ms`.
+            tracing::warn!(id = %entry.id, reason = %reason, "leaving a frame task for redelivery");
+        }
+    }
+}
+
+/// Scores one frame and reports anything found.
+async fn process(reporter: &mut Reporter, fields: &HashMap<String, redis::Value>) -> Outcome {
+    let task = match FrameTask::from_stream_fields(fields) {
+        Ok(task) => task,
+        Err(err) => return Outcome::Poison(err.to_string()),
+    };
+
+    // NOTE: the frame bytes are not fetched yet - `r2_object_key` says where
+    // they are and nothing downloads them. `inference::detect` is given an
+    // empty slice and, as of 2026-10-11, abstains rather than inventing a
+    // detection, so this is a no-op that acknowledges its input. See the
+    // module docs in src/inference/mod.rs.
+    let frame_bytes: Vec<u8> = Vec::new();
+
+    let detections = inference::detect(&frame_bytes);
+    if detections.is_empty() {
+        tracing::debug!(printer_id = %task.printer_id, "no detections");
+        return Outcome::Done;
+    }
+
+    match reporter.report(&task, &detections).await {
+        Ok(()) => Outcome::Done,
+        // A publish failure is transient by nature - the connection dropped,
+        // the server is loaded - so this is a retry and not a poison.
+        Err(err) => Outcome::Retry(format!("reporting failed: {err}")),
+    }
+}
+
+/// Reclaims entries that were delivered to a consumer which never
+/// acknowledged them - the usual cause being that it was killed mid-frame.
+///
+/// Without this, an entry delivered to a consumer that dies stays pending
+/// forever: it has been handed out, so no other consumer is offered it, and
+/// nothing retries it.
+async fn recover_pending(
+    conn: &mut redis::aio::MultiplexedConnection,
+    reporter: &mut Reporter,
+    config: &Config,
+) -> anyhow::Result<()> {
+    let pending: StreamPendingCountReply = conn
+        .xpending_count(&config.frame_stream, &config.consumer_group, "-", "+", 64usize)
+        .await?;
+
+    for entry in pending.ids {
+        // Only entries that have been idle long enough that whoever holds
+        // them is plainly not working on them any more.
+        if entry.last_delivered_ms < config.claim_idle_ms {
+            continue;
+        }
+
+        if entry.times_delivered >= config.max_deliveries {
+            let reason = format!(
+                "delivered {} times without being acknowledged (limit {})",
+                entry.times_delivered, config.max_deliveries
+            );
+            tracing::error!(id = %entry.id, reason = %reason, "dead-lettering a repeatedly failing frame task");
+            // Fetch the fields before acknowledging, so the dead-letter entry
+            // carries the original task rather than only its id.
+            let range: StreamRangeReply = conn
+                .xrange(&config.frame_stream, &entry.id, &entry.id)
+                .await
+                .unwrap_or(StreamRangeReply { ids: Vec::new() });
+            let fields = range.ids.first().map(|e| e.map.clone()).unwrap_or_default();
+            dead_letter(conn, config, &entry.id, &fields, &reason).await;
+            ack(conn, config, &config.frame_stream, &entry.id).await;
+            continue;
+        }
+
+        let claimed: StreamClaimReply = conn
+            .xclaim(
+                &config.frame_stream,
+                &config.consumer_group,
+                &config.consumer_name,
+                config.claim_idle_ms,
+                &[entry.id.as_str()],
+            )
+            .await?;
+
+        for reclaimed in &claimed.ids {
+            tracing::info!(
+                id = %reclaimed.id,
+                delivery = entry.times_delivered + 1,
+                "reclaimed a pending frame task"
+            );
+            handle_and_settle(conn, reporter, config, &config.frame_stream, reclaimed).await;
+        }
+    }
+
+    Ok(())
+}
+
+/// Acknowledges an entry, and says so if that fails.
+///
+/// The result used to be discarded with `let _: RedisResult<i64>`. A failed
+/// XACK means the entry stays pending and will be reclaimed and reprocessed
+/// later, which is survivable but is not nothing - it should not be invisible.
+async fn ack(
+    conn: &mut redis::aio::MultiplexedConnection,
+    config: &Config,
+    stream: &str,
+    id: &str,
+) {
+    match conn
+        .xack::<_, _, _, i64>(stream, &config.consumer_group, &[id])
+        .await
+    {
+        Ok(1) => {}
+        Ok(count) => tracing::warn!(id = %id, count, "XACK acknowledged an unexpected number of entries"),
+        Err(err) => tracing::error!(id = %id, error = %err, "XACK failed; the entry stays pending"),
+    }
+}
+
+/// Copies an entry onto the dead-letter stream, with why it went there.
+async fn dead_letter(
+    conn: &mut redis::aio::MultiplexedConnection,
+    config: &Config,
+    id: &str,
+    fields: &HashMap<String, redis::Value>,
+    reason: &str,
+) {
+    // The original fields verbatim, so a dead-lettered task can be inspected
+    // and replayed, plus provenance. Prefixed keys cannot collide with the
+    // task's own, which `additionalProperties: false` in the schema keeps to
+    // a known set.
+    let mut pairs: Vec<(String, String)> = vec![
+        ("dead_original_id".to_string(), id.to_string()),
+        ("dead_reason".to_string(), reason.to_string()),
+        ("dead_stream".to_string(), config.frame_stream.clone()),
+    ];
+    for (key, value) in fields {
+        let rendered = match value {
+            redis::Value::BulkString(bytes) => String::from_utf8_lossy(bytes).into_owned(),
+            redis::Value::SimpleString(text) => text.clone(),
+            redis::Value::Int(number) => number.to_string(),
+            other => format!("{other:?}"),
+        };
+        pairs.push((key.clone(), rendered));
+    }
+
+    let args: Vec<(&str, &str)> = pairs.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();
+    if let Err(err) = conn
+        .xadd::<_, _, _, _, String>(&config.dead_letter_stream, "*", &args)
+        .await
+    {
+        // Nothing left to do but be loud: the entry is about to be
+        // acknowledged, so this is the last chance to record it anywhere.
+        tracing::error!(
+            id = %id,
+            error = %err,
+            reason = %reason,
+            "could not write to the dead-letter stream; this task is being dropped"
+        );
+    }
+}
+
+/// `anyhow::Context` for `RedisError` without pulling the trait into scope at
+/// every call site.
+trait ContextMsg<T> {
+    fn context_msg(self, message: &'static str) -> anyhow::Result<T>;
+}
+
+impl<T> ContextMsg<T> for Result<T, redis::RedisError> {
+    fn context_msg(self, message: &'static str) -> anyhow::Result<T> {
+        self.map_err(|err| anyhow::anyhow!("{message}: {err}"))
     }
 }

src/reporter/mod.rs

@@ -2,39 +2,79 @@ use redis::AsyncCommands;
 use tracing::info;
 
 use crate::inference::Detection;
+use crate::task::FrameTask;
 
+/// Mirrors `x-channel` in
+/// `schemas/continuum/v1/ai/print_alert.schema.json`.
 const ALERT_CHANNEL: &str = "channel:print_alerts";
 
-/// Fast-paths a failure alert out to whatever's listening.
+/// Fast-paths a failure alert out to whatever is listening.
 pub struct Reporter {
-    redis: redis::aio::MultiplexedConnection,
+    valkey: redis::aio::MultiplexedConnection,
 }
 
 impl Reporter {
-    pub async fn connect(redis_url: &str) -> anyhow::Result<Self> {
-        let client = redis::Client::open(redis_url)?;
-        let redis = client.get_multiplexed_tokio_connection().await?;
-        Ok(Self { redis })
+    pub async fn connect(valkey_url: &str) -> anyhow::Result<Self> {
+        let client = redis::Client::open(valkey_url)?;
+        let valkey = client.get_multiplexed_tokio_connection().await?;
+        Ok(Self { valkey })
     }
 
-    /// Publishes to Redis (real - this is what `continuum-backend`'s
-    /// WebSocket fan-out subscribes to, for live operator-console alerts).
+    /// Publishes a `PrintAlert` onto `channel:print_alerts`, in the shape of
+    /// `schemas/continuum/v1/ai/print_alert.schema.json`.
     ///
-    /// A real version would also `INSERT` into Postgres here via `sqlx` for
-    /// the permanent audit trail - that's left as a `println!` for now;
-    /// come back to it once async SQL feels comfortable.
-    pub async fn report(&mut self, printer_id: &str, detections: &[Detection]) -> anyhow::Result<()> {
-        let best = &detections[0];
-        info!(printer_id, label = %best.label, confidence = best.confidence, "reporting print failure");
+    /// ## Nothing subscribes to this yet
+    ///
+    /// The previous version of this comment said the channel was "what
+    /// `continuum-backend`'s WebSocket fan-out subscribes to". It is not:
+    /// as of 2026-10-11 no process anywhere subscribes, and PUBLISH has no
+    /// persistence, so every alert published so far went nowhere. Finishing
+    /// that is a backend route, not a change here.
+    ///
+    /// ## And nothing persists it
+    ///
+    /// A detection's permanent record belongs in Postgres - `frame_embeddings`
+    /// exists for the vectors and has never been written to either. PUBLISH
+    /// is fire-and-forget: an alert published with no subscriber is gone, so
+    /// this channel can only ever be a live-console feed, never the record.
+    pub async fn report(&mut self, task: &FrameTask, detections: &[Detection]) -> anyhow::Result<()> {
+        // `detections[0]` would panic here. The caller does check, but a
+        // public method that panics on an empty slice is a trap for the next
+        // caller.
+        let Some(best) = detections
+            .iter()
+            .max_by(|a, b| a.confidence.total_cmp(&b.confidence))
+        else {
+            return Ok(());
+        };
 
-        let payload = serde_json::json!({
-            "printerId": printer_id,
+        info!(
+            printer_id = %task.printer_id,
+            label = %best.label,
+            confidence = best.confidence,
+            "reporting print failure"
+        );
+
+        // Built field by field rather than by deriving Serialize on a struct,
+        // because `print_job_id` must be *absent* when there is no job rather
+        // than `null` - same rule as the stream task, and the schema says so.
+        let mut payload = serde_json::json!({
+            "printer_id": task.printer_id,
             "label": best.label,
             "confidence": best.confidence,
+            // From the frame, not from now: an operator needs to know when
+            // the printer looked like this, not when a GPU got round to it.
+            "captured_at": task.captured_at,
+            // So a console can show the picture, not just a label.
+            "r2_object_key": task.r2_object_key,
         });
-        self.redis.publish::<_, _, ()>(ALERT_CHANNEL, payload.to_string()).await?;
+        if let Some(job_id) = &task.print_job_id {
+            payload["print_job_id"] = serde_json::Value::String(job_id.clone());
+        }
 
-        println!("(would also INSERT this alert into Postgres for the audit trail)");
+        self.valkey
+            .publish::<_, _, ()>(ALERT_CHANNEL, payload.to_string())
+            .await?;
 
         Ok(())
     }

src/task.rs

@@ -0,0 +1,249 @@
+//! The Rust mirror of `schemas/continuum/v1/ai/frame_task.schema.json`.
+//!
+//! The JSON Schema is canonical and the TypeScript mirror is in
+//! `continuum-schemas/packages/ts-types/src/schemas/ai.ts`. Change the schema
+//! first, then both mirrors.
+//!
+//! This module exists because the field names were previously nowhere: the
+//! backend wrote `printerId`, `farmId`, `r2ObjectKey` and `capturedAt`, and
+//! this worker read one field called `printer_id` and ignored the rest. The
+//! read never matched, so every frame was attributed to a printer called
+//! "unknown" and the object key - the only thing that says *which picture* -
+//! was never looked at. Nothing errored, because a stream entry is just a
+//! string map and asking it for a field that is not there is not a failure.
+
+use std::collections::HashMap;
+use std::fmt;
+
+/// One entry off `stream:camera_frames`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct FrameTask {
+    pub farm_id: String,
+    /// `printers.id` - the backend's UUID. Not the gateway's own printer id,
+    /// which is only meaningful inside its own farm.
+    pub printer_id: String,
+    /// `None` when the frame belongs to no tracked job: a printer may be
+    /// streaming while idle, or running something queued outside Continuum.
+    pub print_job_id: Option<String>,
+    pub r2_object_key: String,
+    /// The gateway's clock at capture, not enqueue time. A gateway that
+    /// buffered through an uplink outage produces tasks minutes older than
+    /// their stream id suggests.
+    pub captured_at: String,
+}
+
+/// Why a task could not be read. Every variant is terminal: a stream entry
+/// does not become well-formed by being retried, so these are dead-lettered
+/// rather than redelivered.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum TaskError {
+    MissingField(&'static str),
+    /// Present but empty. Called out separately from missing because it means
+    /// a producer wrote `""` where it meant "no value" - the exact bug the
+    /// schema's "omit the field" rule exists to prevent - and the message
+    /// should say so rather than claiming the field was absent.
+    EmptyField(&'static str),
+    NotAUuid { field: &'static str, value: String },
+}
+
+impl fmt::Display for TaskError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::MissingField(name) => write!(f, "missing required field '{name}'"),
+            Self::EmptyField(name) => {
+                write!(f, "field '{name}' is present but empty; a producer with no value must omit it")
+            }
+            Self::NotAUuid { field, value } => {
+                write!(f, "field '{field}' is not a uuid: '{value}'")
+            }
+        }
+    }
+}
+
+impl std::error::Error for TaskError {}
+
+/// A stream entry's values are opaque bulk strings; this is the only place
+/// that turns one into a `String`.
+fn read_field(fields: &HashMap<String, redis::Value>, name: &str) -> Option<String> {
+    match fields.get(name) {
+        Some(redis::Value::BulkString(bytes)) => Some(String::from_utf8_lossy(bytes).into_owned()),
+        Some(redis::Value::SimpleString(text)) => Some(text.clone()),
+        Some(redis::Value::Int(number)) => Some(number.to_string()),
+        _ => None,
+    }
+}
+
+fn required(fields: &HashMap<String, redis::Value>, name: &'static str) -> Result<String, TaskError> {
+    match read_field(fields, name) {
+        None => Err(TaskError::MissingField(name)),
+        Some(value) if value.is_empty() => Err(TaskError::EmptyField(name)),
+        Some(value) => Ok(value),
+    }
+}
+
+/// A deliberately cheap uuid check: the right length, hyphens in the right
+/// four places, hex everywhere else. Not a parse, and no `uuid` dependency
+/// for something that is never taken apart here.
+///
+/// It is worth doing at all because these ids are passed straight into an
+/// alert payload and, once the Postgres write exists, into uuid columns.
+/// Rejecting a malformed id here turns a bad row or a nonsense alert into one
+/// dead-lettered entry.
+fn looks_like_uuid(value: &str) -> bool {
+    if value.len() != 36 {
+        return false;
+    }
+    value.bytes().enumerate().all(|(index, byte)| match index {
+        8 | 13 | 18 | 23 => byte == b'-',
+        _ => byte.is_ascii_hexdigit(),
+    })
+}
+
+fn required_uuid(
+    fields: &HashMap<String, redis::Value>,
+    name: &'static str,
+) -> Result<String, TaskError> {
+    let value = required(fields, name)?;
+    if looks_like_uuid(&value) {
+        Ok(value)
+    } else {
+        Err(TaskError::NotAUuid { field: name, value })
+    }
+}
+
+impl FrameTask {
+    /// Reads a task out of a stream entry's field map.
+    pub fn from_stream_fields(fields: &HashMap<String, redis::Value>) -> Result<Self, TaskError> {
+        // An absent key and an empty value are treated alike, per the
+        // schema: the producer is supposed to omit what it has no value for,
+        // and a consumer that distinguishes the two would break the moment a
+        // producer got that wrong - which one did.
+        let print_job_id = match read_field(fields, "print_job_id") {
+            None => None,
+            Some(value) if value.is_empty() => None,
+            Some(value) if looks_like_uuid(&value) => Some(value),
+            Some(value) => {
+                return Err(TaskError::NotAUuid { field: "print_job_id", value });
+            }
+        };
+
+        Ok(Self {
+            farm_id: required_uuid(fields, "farm_id")?,
+            printer_id: required_uuid(fields, "printer_id")?,
+            print_job_id,
+            r2_object_key: required(fields, "r2_object_key")?,
+            captured_at: required(fields, "captured_at")?,
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const FARM: &str = "11111111-1111-1111-1111-111111111111";
+    const PRINTER: &str = "22222222-2222-2222-2222-222222222222";
+    const JOB: &str = "33333333-3333-3333-3333-333333333333";
+
+    fn fields(pairs: &[(&str, &str)]) -> HashMap<String, redis::Value> {
+        pairs
+            .iter()
+            .map(|(k, v)| ((*k).to_string(), redis::Value::BulkString(v.as_bytes().to_vec())))
+            .collect()
+    }
+
+    fn complete() -> Vec<(&'static str, &'static str)> {
+        vec![
+            ("farm_id", FARM),
+            ("printer_id", PRINTER),
+            ("r2_object_key", "farms/x/frames/1.jpg"),
+            ("captured_at", "2026-10-11T12:00:00Z"),
+        ]
+    }
+
+    #[test]
+    fn reads_a_complete_task() {
+        let task = FrameTask::from_stream_fields(&fields(&complete())).unwrap();
+        assert_eq!(task.farm_id, FARM);
+        assert_eq!(task.printer_id, PRINTER);
+        assert_eq!(task.print_job_id, None);
+        assert_eq!(task.r2_object_key, "farms/x/frames/1.jpg");
+    }
+
+    #[test]
+    fn reads_an_optional_job_id() {
+        let mut pairs = complete();
+        pairs.push(("print_job_id", JOB));
+        let task = FrameTask::from_stream_fields(&fields(&pairs)).unwrap();
+        assert_eq!(task.print_job_id, Some(JOB.to_string()));
+    }
+
+    #[test]
+    fn an_empty_job_id_is_no_job_id() {
+        // The backend used to write `""` for "no job". Readers have to accept
+        // that as absent rather than carrying an empty string forward.
+        let mut pairs = complete();
+        pairs.push(("print_job_id", ""));
+        let task = FrameTask::from_stream_fields(&fields(&pairs)).unwrap();
+        assert_eq!(task.print_job_id, None);
+    }
+
+    /// The regression test for the bug this module was written for. The old
+    /// consumer read `printer_id` from a producer writing `printerId` and
+    /// substituted the string "unknown" - so this has to be an error, not a
+    /// default.
+    #[test]
+    fn camel_case_field_names_are_an_error_not_a_default() {
+        let camel = fields(&[
+            ("farmId", FARM),
+            ("printerId", PRINTER),
+            ("r2ObjectKey", "farms/x/frames/1.jpg"),
+            ("capturedAt", "2026-10-11T12:00:00Z"),
+        ]);
+        let error = FrameTask::from_stream_fields(&camel).unwrap_err();
+        assert_eq!(error, TaskError::MissingField("farm_id"));
+    }
+
+    #[test]
+    fn every_required_field_is_actually_required() {
+        for missing in ["farm_id", "printer_id", "r2_object_key", "captured_at"] {
+            let pairs: Vec<_> = complete().into_iter().filter(|(k, _)| *k != missing).collect();
+            let error = FrameTask::from_stream_fields(&fields(&pairs)).unwrap_err();
+            assert_eq!(error, TaskError::MissingField(missing), "for {missing}");
+        }
+    }
+
+    #[test]
+    fn an_empty_required_field_says_so() {
+        let mut pairs = complete();
+        pairs.retain(|(k, _)| *k != "r2_object_key");
+        pairs.push(("r2_object_key", ""));
+        assert_eq!(
+            FrameTask::from_stream_fields(&fields(&pairs)).unwrap_err(),
+            TaskError::EmptyField("r2_object_key"),
+        );
+    }
+
+    #[test]
+    fn a_malformed_printer_id_is_refused() {
+        let mut pairs = complete();
+        pairs.retain(|(k, _)| *k != "printer_id");
+        pairs.push(("printer_id", "unknown"));
+        assert!(matches!(
+            FrameTask::from_stream_fields(&fields(&pairs)).unwrap_err(),
+            TaskError::NotAUuid { field: "printer_id", .. }
+        ));
+    }
+
+    #[test]
+    fn uuid_check_accepts_real_ids_and_rejects_near_misses() {
+        assert!(looks_like_uuid(FARM));
+        assert!(looks_like_uuid("0f8fad5b-d9cb-469f-a165-70867728950e"));
+        assert!(!looks_like_uuid(""));
+        assert!(!looks_like_uuid("unknown"));
+        // Right length, hyphens in the wrong places.
+        assert!(!looks_like_uuid("0f8fad5bd-9cb-469f-a165-70867728950"));
+        // Right shape, non-hex digit.
+        assert!(!looks_like_uuid("0f8fad5b-d9cb-469f-a165-70867728950g"));
+    }
+}