# continuum-ai-worker

Computer-vision failure detection worker for the Continuum print farm
platform. Consumes camera frame tasks from the Valkey Stream
`stream:camera_frames` (written by `continuum-backend`'s `src/queue/`) and
reports failures.

## What is real and what is not

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.

- `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.

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

```bash
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: 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)
```