Write down the backend <-> ai-worker contract
`stream:camera_frames` and `channel:print_alerts` are the only interface between `continuum-backend` and `continuum-ai-worker`, and there was nothing specifying either. They had drifted completely: the backend wrote `farmId`, `printerId`, `r2ObjectKey` and `capturedAt`; the 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, because a stream entry is a flat string map and asking it for a field that is not there is not a failure. Same shape as the /ws/edge/v1 drift, found the same way. `schemas/continuum/v1/ai/` is now canonical, with a TypeBox mirror in ts-types and a Rust mirror in the worker. Two things the schemas have to say that JSON Schema does not naturally express, both of which caused real bugs: - A Valkey stream entry is a flat field/value map of *strings*. No nesting, no numbers, no null. So FrameTask has no numeric or nested properties, and a field with no value is omitted from the XADD rather than written as `""` - an empty string is a value, and the producer was writing one for `print_job_id`, which a consumer would carry into a lookup or a uuid column. - PrintAlert, by contrast, *is* a JSON document, because PUBLISH carries one opaque string. It is also fire-and-forget with no persistence, so it can only ever be a live-console feed and never the record of a detection. The README's layout section was stale (it still listed four DTOs that were removed in 2026-09) and did not mention `edge/` or `ai/` at all. It now names both, says why the inter-service protocols live here, and notes that their Rust mirrors sit in the single repo that consumes each rather than in `packages/rust-types`.
This commit is contained in:
@@ -0,0 +1,96 @@
|
||||
import { Type, type Static } from "@sinclair/typebox";
|
||||
|
||||
/**
|
||||
* The `stream:camera_frames` work queue and the `channel:print_alerts`
|
||||
* notification channel, mirroring `schemas/continuum/v1/ai/*.schema.json`
|
||||
* by hand - the JSON Schema files are canonical, and the Rust mirror lives
|
||||
* in `continuum-ai-worker/src/task.rs`.
|
||||
*
|
||||
* These two are the only interface between `continuum-backend` and
|
||||
* `continuum-ai-worker`. They are written down here because the previous
|
||||
* arrangement - a producer and a consumer each naming their own fields,
|
||||
* with nothing in between - produced exactly the drift it sounds like: the
|
||||
* backend wrote `printerId` and the worker read `printer_id`, so every
|
||||
* frame the worker ever handled was attributed to a printer called
|
||||
* "unknown", and nothing failed.
|
||||
*/
|
||||
|
||||
/**
|
||||
* One entry on `stream:camera_frames`.
|
||||
*
|
||||
* ## Every value is a string
|
||||
*
|
||||
* A Valkey stream entry is a flat field/value map of strings. There is no
|
||||
* nesting and no numeric type, which is why `FrameTaskSchema` has no
|
||||
* numbers and no nested objects - a producer that needs one has to encode
|
||||
* it, and then both sides have to agree on the encoding, which is the thing
|
||||
* this file exists to avoid.
|
||||
*
|
||||
* ## An absent field is absent
|
||||
*
|
||||
* `print_job_id` is omitted from the XADD when the frame belongs to no
|
||||
* tracked job - not written as `""`. An empty string is a value: a consumer
|
||||
* that passes it along ends up looking up job id `""`, and one that writes
|
||||
* it to Postgres violates a uuid column. So the producer omits the field and
|
||||
* the consumer treats absent and empty alike.
|
||||
*/
|
||||
export const FrameTaskSchema = Type.Object({
|
||||
/** `farms.id`. */
|
||||
farm_id: Type.String({ format: "uuid" }),
|
||||
/**
|
||||
* `printers.id` - the backend's UUID, already resolved from the gateway's
|
||||
* own printer id, which means nothing outside its farm.
|
||||
*/
|
||||
printer_id: Type.String({ format: "uuid" }),
|
||||
/** `print_jobs.id`, or omitted. Never `""`. */
|
||||
print_job_id: Type.Optional(Type.String({ format: "uuid" })),
|
||||
/** Where the frame is in R2. The bytes are never in the stream. */
|
||||
r2_object_key: Type.String({ minLength: 1 }),
|
||||
/**
|
||||
* The gateway's clock, at capture - not enqueue time. A gateway that
|
||||
* buffered frames through an uplink outage produces tasks whose
|
||||
* `captured_at` is minutes old.
|
||||
*/
|
||||
captured_at: Type.String({ format: "date-time" }),
|
||||
});
|
||||
export type FrameTask = Static<typeof FrameTaskSchema>;
|
||||
|
||||
/**
|
||||
* One message on `channel:print_alerts`, as JSON.
|
||||
*
|
||||
* Unlike a stream entry this one is a document, because PUBLISH carries a
|
||||
* single opaque string. It is also fire-and-forget: an alert published with
|
||||
* no subscriber is gone, so this channel is a live-console feed and not the
|
||||
* record of a detection.
|
||||
*/
|
||||
export const PrintAlertSchema = Type.Object({
|
||||
printer_id: Type.String({ format: "uuid" }),
|
||||
print_job_id: Type.Optional(Type.String({ format: "uuid" })),
|
||||
/**
|
||||
* Free string, not an enum: the label set belongs to the model file and
|
||||
* changes without this schema changing.
|
||||
*/
|
||||
label: Type.String(),
|
||||
/**
|
||||
* The model's own score. Thresholding is the consumer's policy - an
|
||||
* operator may well want to see near-misses.
|
||||
*/
|
||||
confidence: Type.Number({ minimum: 0, maximum: 1 }),
|
||||
/** From the frame, so a console shows when the printer looked like this. */
|
||||
captured_at: Type.String({ format: "date-time" }),
|
||||
/** So a console can show the picture, not just the label. */
|
||||
r2_object_key: Type.String(),
|
||||
});
|
||||
export type PrintAlert = Static<typeof PrintAlertSchema>;
|
||||
|
||||
/** The stream the tasks go on, and the group the worker reads as. */
|
||||
export const CAMERA_FRAMES_STREAM = "stream:camera_frames";
|
||||
export const AI_WORKER_CONSUMER_GROUP = "ai-worker";
|
||||
/**
|
||||
* Where entries go that cannot be processed and must not be retried - a
|
||||
* malformed task, or one that has been redelivered too many times. Separate
|
||||
* from the main stream so a poison message cannot block the queue and
|
||||
* cannot be lost silently either.
|
||||
*/
|
||||
export const CAMERA_FRAMES_DEAD_LETTER_STREAM = "stream:camera_frames:dead";
|
||||
export const PRINT_ALERTS_CHANNEL = "channel:print_alerts";
|
||||
@@ -9,3 +9,8 @@
|
||||
// a TypeBox mirror here, consumed by continuum-backend's edge WebSocket
|
||||
// route. Mirrors schemas/continuum/v1/edge/.
|
||||
export * from "./edge";
|
||||
|
||||
// The continuum-backend -> continuum-ai-worker interface: the
|
||||
// stream:camera_frames work queue and the channel:print_alerts notification
|
||||
// channel. Mirrors schemas/continuum/v1/ai/.
|
||||
export * from "./ai";
|
||||
|
||||
Reference in New Issue
Block a user