Stages

Workflows let you reshape the Hub pipeline without code — wire a device through a few filters into a model on a visual canvas. Custom stages are the developer side of that same system: instead of choosing from the built-in nodes, you bring your own microservice in as a workflow stage. The engine is open — every built-in stage (classification, thumbnails, sprites) is just a service that consumes a message off a queue, does one job, and hands the result back — and your service plugs in the same way.

A workflow stage is a step in a workflow, implemented as a microservice the workflows engine triggers automatically for every recording: it receives a run from a queue, does the work, and returns the result — in whatever language suits the job, deployed and scaled on its own. Stages are asynchronous: they run alongside the built-in analysis and never block it.

This page is the contract your microservice codes against — the queue it listens on, the message it receives, how it returns a result, and how the engine tracks it to completion. It is capability-agnostic: it never assumes what your stage does, so the same mechanism serves a speech-to-text service, a custom detector, or any other step. For a concrete capability built on it, see the block types a stage can produce.

Status — rolling out. The queue, the WorkflowRun it dispatches and the completion mechanics here are already how the pipeline works internally. The config-driven stage registration (the kerberoshub.workflows.definitions values section — see Registering a stage) is the addition that lets a custom stage join without changing engine code — dispatched by the standalone workflows engine (hub-workflows), which runs alongside the analysis service and consumes the classify results it tees over. It is landing now for self-hosted deployments.

This page covers how a microservice delivers a result. For the complementary receiving side — one shared core that takes a result from either the API or the queue and routes each block to the right actions by its type — see Ingest.

When to add a stage

A stage is one of two transports for getting your data into the Hub. The other is an authenticated ingest API push. They deliver the same data to the same place; they differ in who triggers the work and where your code runs.

• Ingest API push — your service POSTs whenever it has data. Works on every deployment, needs no cluster access. The right starting point for most integrators. See Ingest → Over the API.
• Workflow stage (this page) — the pipeline triggers your service automatically on every ingest / re-analysis, with queue-level delivery guarantees. Available on self-hosted deployments that can run custom stages.

Reach for a stage only when you control the deployment and want the capability to run automatically as a built-in step of every recording’s analysis.

Anatomy of a stage

A stage has exactly two runtime dependencies: the message broker (to receive events and hand results back) and the database (to read and write event metadata). There is no service-to-service HTTP and no shared in-process state — every hand-off goes through the broker. That loose coupling is what lets any stage scale, restart or be replaced without touching the rest of the pipeline.

Registering a stage

A stage lives inside a workflow definition and pairs with a worker deployment — both declared in the chart’s values.yaml, no engine code changes. The definition itself (kerberoshub.workflows.definitions.<workflow>), its full field reference and running several workflows are covered in Defining a workflow in configuration. This page is about a stage’s own two parts:

• the stage entry — an item in a definition’s stages list, declaring the stage’s operation and how the engine routes to it;
• the worker deployment (kerberoshub.services.<operation>) — the microservice that consumes the stage’s queue.

The stage’s queue is not set on the stage — it is taken from the matching services.<operation>.queue, so the engine dispatches and the worker consumes the same name with no drift. Deploy the worker (services.<operation>.enabled) for every operation a workflow references; otherwise the engine dispatches to a queue nothing has declared and those messages are returned unroutable and dead-lettered (see Queue naming).

# values.yaml — a stage is one entry in a workflow definition + its worker
kerberoshub:
  workflows:
    enabled: true                    # master switch: the workflows engine
    definitions:
      speed-workflow:                # the workflow this stage belongs to
        enabled: true
        stages:
          - operation: speed         # ← the stage entry
            dispatch: conditional    # always | conditional
            needsMode: any           # any | all (conditional, multiple needs)
            needs:
              - operation: classify
                condition: { path: inputs.classify.details.*.classified, op: eq, value: car }

  services:
    speed:                           # ← the worker, keyed by the stage's operation id
      enabled: true                  # deploy the worker
      repository: ghcr.io/acme/speed
      tag: "v1.0.0"
      queue: "hub-workflows-speed"   # the queue the engine dispatches to and the worker consumes
      replicas: 1
      pullPolicy: IfNotPresent
      logLevel: info                 # trace | debug | info | warn | error
      resources: {}

Stage entry — an item in a definition’s stages list

Stage vs. operation. On this page, stage is the step and operation is its id — the operation field above, which is also the key everything else (result, queue, worker, completion) is bound by. See the glossary for the full vocabulary.

Worker deployment — kerberoshub.services.<operation>

A normal microservice Deployment, keyed to the stage’s operation id. (The workflows engine itself is deployed from this same section as services.workflows — the one services entry that is not a stage worker, and a chart default you don’t normally touch.)

Minimal stage. The smallest live stage is one stages entry (operation + dispatch: always) inside an enabled workflow definition, plus its worker under services.<operation> (enabled: true, image + queue). Everything else is additive — you can switch to conditional routing later without touching the rest.

Conditional routing

By default (dispatch: always) a stage runs on every recording. Set dispatch: conditional to run it only on recordings that match a rule you declare in the stage object — handy when the deciding signal (e.g. the classifier saw a car) isn’t known until the run is already underway. A conditional stage is never queued up front; the engine re-checks it as the run progresses and fires it the moment the rule holds. Recordings that never match never touch the stage.

The rule is a list of needs combined by needsMode:

stages:
  - operation: speed
    dispatch: conditional
    needsMode: any                 # any (default) | all
    needs:
      - operation: classify        # GATE — wait until this op is on the run
        condition:                 # PREDICATE — tested once the gate is ready
          path: inputs.classify.details.*.classified
          op: eq
          value: car

• operation — the gate: an upstream operation id whose result must be present on the run before this need is checked (e.g. classify, or another stage’s id). Leave it empty to check the run with no gate — for device / user / identity rules that are known the moment the run opens.
• condition — the predicate tested once the gate is ready. Omit it to fire on the gate alone (i.e. “as soon as classify is present”).
• needsMode — with more than one need: any (default) fires on the first satisfied need (OR); all fires only when every need is satisfied (AND — a join).

# only on one camera, and only when a car was seen — both must hold
needsMode: all
needs:
  - operation: classify
    condition: { path: inputs.classify.details.*.classified, op: eq, value: car }
  - operation:                     # no gate — checked the moment the run opens
    condition: { path: device.deviceKey, op: eq, value: device02 }

The condition object

A condition is { path, op, value }.

path — a dot-path relative to the WorkflowRun your stage receives (see The workflow run). It walks objects, and a * segment fans out across the elements of an array (see Matching inside arrays). Valid roots:

Credentials (storage, user.storage) are deliberately not matchable.

op and value — the comparison and its operand:

contains and in are mirror images: use contains when the run holds a list and you test for a member (inputs.classify.properties contains car); use in when the run holds a single value and you test it against a set (device.deviceKey in [device01, device02]).

matches is for partial / pattern text matching where contains (fixed substring) and eq (whole value) aren’t enough — e.g. results.anpr.plate matches ^[A-Z]{3}[0-9]{3}$ (plate shape), device.deviceName matches ^lobby- (name prefix), or a case-insensitive (?i)garage. The pattern is validated when the registry loads, so a bad regex fails fast.

Matching inside arrays

A path segment of * fans out across the elements of the array at that position and continues resolving from each element, so it reaches a field on every object in a list. This is how you branch on the result a delegated stage hands back: the engine mirrors your block envelope into results.<operation> grouped by block type, so a stage that emits detection blocks exposes results.<operation>.detections as an array.

Say your detector stage returns a list of boxes:

"results": { "detector": { "detections": [
  { "label": "car", "score": 0.92 },
  { "label": "car", "score": 0.41 }
] } }

A * matches across the list — the predicate holds when any element satisfies a positive operator (exists / eq / contains / in / matches / gt / gte / lt / lte):

# fire when the stage found ANY box scoring above 0.8
condition: { path: results.detector.detections.*.score, op: gt, value: 0.8 }

That is what makes per-element numbers work: each fanned-out value is a single score, so gt / lt compare it directly. ne is the inverse — it holds only when no element equals value (…detections.*.label, op: ne, value: car ⇒ not one box is a car). Nested lists compose: results.detector.detections.*.boxes.*.x reaches every box of every detection.

Two limits to keep in mind. * is the only index — there is no numeric [0], and a plain step onto an array (without *) matches nothing, so gate on the array itself with contains / exists or fan out with *. And separate needs don’t correlate: …*.label eq car and …*.score gt 0.8 can be satisfied by different elements — there is no “the same object has both” across two conditions. Built-in classify exposes details as a fan-out array (inputs.classify.details.*.classified) and properties as a flat list; objectCount stays a scalar. * applies to those and to your own operation’s result arrays.

The engine validates every condition path at boot and refuses to start on an unknown one — a typo fails fast instead of silently never firing.

Partial & pattern matching

eq tests a whole value and contains tests a fixed substring; when neither is enough — you want a shape, a prefix/suffix, or a case-insensitive test — reach for matches, an RE2 regular-expression test. It is a partial (unanchored) match: the pattern only has to be found somewhere in the value, so anchor it with ^…$ when you mean the whole string.

# fire only for plates shaped LLL-DDD (three letters, three digits)
needs:
  - operation: anpr
    condition: { path: results.anpr.plate, op: matches, value: "^[A-Z]{3}[0-9]{3}$" }

A few things to know:

• Partial by default. device.deviceName matches lobby fires for lobby-front, east-lobby and lobby. Anchor it — ^lobby- (prefix), -cam$ (suffix), ^lobby-front$ (exact) — to narrow it.
• Case-insensitive with the inline (?i) flag: device.deviceName matches (?i)garage fires for Garage, GARAGE, garage.
• Works on arrays too. Like contains, a bare array field matches when any string element matches — inputs.classify.properties matches ^car$ — or fan out explicitly with *: inputs.classify.details.*.classified matches ^(car|truck)$.
• Strings only. Numbers, booleans, maps and non-string array elements never match — use eq / gt / contains for those.
• Validated at boot. An invalid regular expression (or a non-string value) fails the registry fast, just like a bad path — it never silently no-ops at runtime.

How your microservice connects

The chart deploys your microservice from kerberoshub.services.<name> and injects a fixed set of environment variables — the connection contract. Whatever language your microservice is in, it reads these to reach the broker, find its queues, and fetch media; nothing else is wired for it.

Two things to note:

• No datastore by default. The microservice contract is broker + queues + media storage; the chart injects no database connection. A delegated stage hands its result back over WORKFLOWS_QUEUE for the platform to persist, while a stage that writes its own collection brings its own datastore access.
• Deploying outside the chart. To run the microservice yourself, leave services.<operation>.enabled off (so the chart deploys no pod) but keep the services.<operation> entry — its queue is what the engine dispatches to — and keep the stage in a workflows.definitions workflow so the engine still routes to it; then wire these same variables into your own deployment. The consume and return queue names are the only hard requirement.

The message you receive

Queue naming

Your microservice consumes from one queue, and you choose its name. The source of truth is the queue value on your stage’s deployment in the Helm chart — kerberoshub.services.<name>.queue:

kerberoshub:
  services:
    speed:
      queue: "hub-workflows-speed" # ← anything you want; your microservice consumes this exact name

The engine reads that same value from the matching services.<operation> entry and dispatches there, so the only rule is that the two agree — the queue is the one thing that binds the engine to your microservice. These examples use a consistent hub-workflows-<stage> form (hub-workflows-speed, hub-workflows-llm), but the name is an arbitrary string your broker accepts (vision.requests, team7-detector work just as well) and does not have to follow the platform’s kcloud-… convention — it only has to match on both sides.

If you omit queue, the engine falls back to a derived default, kcloud-<operation>-queue.fifo — so the convention is just that fallback, not the source of truth. Queue names are literal strings: the default deployment runs RabbitMQ, so a .fifo suffix is only part of a name, not an SQS feature.

Who creates the queue. The queue is declared by the consumer — your microservice — when it connects to the broker at pod startup, not on the first message and not by the engine. The engine only publishes to the name; it never creates it. This has a startup-ordering consequence: if the stage worker is not running (never deployed, or not yet started), its queue does not exist, so the engine’s dispatches are returned as unroutable and dead-lettered rather than buffered until a consumer appears. Deploy the microservice so its queue exists; once it has connected once, the durable queue persists across restarts and holds messages while the worker is briefly down.

The workflow run

Your microservice does not receive the pipeline’s internal PipelineEvent. The engine dispatches a single, self-contained models.WorkflowRun as JSON: the run’s identity, the read-only context your microservice needs, and the credentials to fetch the media. Model your microservice’s input type on this — every field below is present on the inbound dispatch:

{
  "operation": "speed",
  "runId": "665f1b2c3d4e5f6071829304",
  "key": "front-gate/2026/06/12/08-30-00.mp4",
  "signedUrl": "https://vault.example.com/kerberos-storage/front-gate/2026/06/12/08-30-00.mp4?X-Amz-Signature=…",
  "traceId": "8f3a1c2b4d5e6f70",
  "user": {
    "organisationId": "64f0a1b2c3d4e5f600112233",
    "storage": { "uri": "s3://kerberos-vault", "access_key": "AKIA…", "provider": "kerberos-vault", "secret_key": "…" }
  },
  "device": {
    "deviceKey": "front-gate",
    "deviceName": "Front Gate",
    "provider": "kerberos-vault",
    "storageSolution": "vault"
  },
  "inputs": {
    "classify": {
      "properties": ["car", "person"],
      "objectCount": 2,
      "details": [{ "classified": "car", "distance": 142.6, "isStatic": false }]
    }
  },
  "results": {},
  "storage": {
    "uri": "s3://kerberos-storage",
    "accessKey": "AKIA…",
    "secret": "…",
    "vaultOverrideUri": "s3://tenant-bucket",
    "vaultOverrideAccessKey": "AKIA…",
    "vaultOverrideSecret": "…",
    "vaultOverrideProvider": "kerberos-vault"
  }
}

Top-level fields

payload is not sent inbound — it is the channel you fill on the way back (delegated-ingest stages only). The dispatch also carries workflowId / workflowName identifying which workflow the run belongs to (several workflows can run over one recording, each opening its own run); a worker doesn’t act on them and echoes only runId on the way back.

user — account context

(user.storage uses snake_case keys — it is the account Storage object. The media-fetch storage below uses camelCase.)

device — recording context

inputs.classify — the trigger result

inputs is keyed by upstream operation. Every run opens from the classifier, so inputs.classify is always present — it is the classification result that triggered the run:

storage — media-fetch credentials

The credentials your microservice uses to fetch the recording. The base trio is always set; the vaultOverride* quartet appears when the recording lives on its own (per-tenant) backend — prefer the override when present, otherwise use the base.

Fetching the media

A dispatch gives you two ways to reach the recording, in order of preference:

1. signedUrl — when present, GET it directly; it needs no credentials. It is short-lived, so on a replayed or backlogged run it may have expired (the storage backend answers 403).
2. storage + key — the authoritative path. Re-sign a fresh URL for the recording from the storage credentials and download that. It always works, and is the fallback whenever signedUrl is absent or rejected.

To re-sign, POST {storage.uri}/storage/bulk with the X-Kerberos-Storage-AccessKey and X-Kerberos-Storage-SecretAccessKey headers and a [{ "filename": "<key>", "provider": "<provider>" }] body; the response is { "data": "<json map of filename → url>" } — read your key’s entry and GET it. Prefer the vaultOverride* credentials and vaultOverrideProvider when set, otherwise the base trio and device.provider. The first-party stages (object tracking, export, redaction) all fetch exactly this way: prefer the signed URL, re-sign from storage on absence or expiry.

inputs and results are your read-only upstream context — the same bags the condition matcher evaluates needs against. The engine routes purely by operation and the workflow definition; it never inspects your output to decide where the run goes.

Acknowledgement

The broker delivers at least once. Acknowledge a message only after the work is durably done (result written or routed back); on failure, let it nack so the broker can redeliver. Because redelivery is possible, make your stage idempotent — key your output by the recording (key) and the run (runId) so a replay replaces rather than duplicates.

Doing the work

Your microservice is a stateless consumer: pull a run, fetch the media (prefer signedUrl, else re-sign from storage — see Fetching the media), compute, route the result back. It can be written in any language that can speak the broker and the WorkflowRun JSON — the only contract is the queue it reads and the run it returns. Reuse the context already on the run (inputs / results) rather than re-fetching it. Keep it single-purpose; if you need a second capability, add a second stage.

Sending a result back

You return the same WorkflowRun you received — echo runId, key, traceId and user so the engine can locate and scope the run — with storage cleared and your result in exactly one channel. Publish it back to the engine’s queue (WORKFLOWS_QUEUE, default hub-workflows-queue); the engine marks the stage resolved and fires any conditional stage that was waiting on it.

There are two sinks. Default to letting the platform persist your result — hand it back and the ingest core stores it, so your microservice needs no datastore of its own. A stage that produces genuinely new data can instead own its storage and write its own collection.

Enrich in place

The default sink: hand the platform a block envelope in payload and it stores the result for you — your microservice needs no database. The envelope is an ordered list of typed blocks — { "type": "detection", "data": … }, { "type": "marker", "data": … } — and the engine routes each block by its type through the shared ingest core, persists it against the run’s own recording, and mirrors the blocks into results.<operation> grouped by type (results.<operation>.detections, …markers) so a downstream condition can branch on what the stage produced — element-wise with *. A stage no longer declares a result kind — each block self-describes via its type; see Blocks for each type’s contract. Set payload or results[operation], never both.

Own collection

For genuinely new, standalone data with no built-in block type — embeddings, descriptions, a custom search index — a stage can write its own collection, keyed by the recording, and set only its routing values under results.<operation> on the returned run (leave payload empty). The platform just records the resolution; your microservice owns the write, and so brings its own datastore access.

The difference between the sinks is only who writes the result — the platform, from the block envelope you hand back, or your microservice into its own collection. Either way the engine marks the stage resolved when your run comes back.

Completion and acknowledgement

Every custom stage is asynchronous: nothing blocks on it. The analysis service’s built-in pipeline continues independently, the workflows engine tracks the stage’s run on its own, and your stage’s result lands whenever the microservice finishes. (Blocking, “required” stages are intentionally out of scope in this design — there is no way for a custom stage to stall a run.)

Whichever sink you use, your microservice routes the run back to the workflows engine — its WORKFLOWS_QUEUE — once the work is durably done. The engine records the stage as resolved on the run ($addToSet resolvedoperations), which keeps the run’s provenance complete and stops a re-run from redoing the work. An own-collection stage’s returned run carries just its routing values under results.<operation>; a delegated stage carries the typed payload. A run that never hears back from a stage still completes on the engine’s own rules (with a safety timeout as a backstop), so a crashed microservice can’t wedge the pipeline.

Following a run. The engine logs and traces every run from open to finalize — including the timeout backstop above — under the recording’s mediaKey and traceId. See Observability for the log lines it emits and how to follow one recording end-to-end.

Failure modes & gotchas

• Routing without a worker (or vice-versa). A stage entry routes to its operation, and the worker is deployed from services.<operation> — they bind through that one name. A stage whose operation has no deployed worker dispatches to a queue nothing has declared — the consumer creates the queue, so with no consumer the engine’s messages are returned unroutable and dead-lettered rather than buffered (see Queue naming); a worker with no stage referencing its operation never receives any. Keep them in sync — they always address the same queue via services.<operation>.queue.
• No completion ack. A microservice that writes its result but never echoes back to the workflows engine (WORKFLOWS_QUEUE) leaves the stage absent from resolvedoperations. Harmless to the run (stages are async), but it breaks provenance and lets a re-run repeat the work. Always ack.
• Re-decode cost. A stage that re-fetches and re-decodes the video pays that cost per recording; reuse data already in the envelope or the database where you can.
• Non-idempotent writes. Redelivery will duplicate output unless you upsert on a stable key.

Checklist

• Pick a unique operation id — it’s the routing key, the result key (results.<id>) and the completion key (the queue is whatever you set in services.<id>.queue)
• Add the stage to a workflow under kerberoshub.workflows.definitions.<workflow>.stages (operation, dispatch: always) and its worker under kerberoshub.services.<operation> (enabled: true, image + queue)
• Make sure the workflows engine is enabled (kerberoshub.workflows.enabled)
• Consume the dispatched WorkflowRun, resolve the recording from key, fetch media — prefer signedUrl, fall back to re-signing from storage (see Fetching the media)
• Read upstream context from inputs / results instead of re-fetching it
• Pick a sink — enrich in place (set payload, the default) or your own collection (set results.<id>)
• Idempotent writes (upsert by key + runId)
• Route the run back to the workflows engine (WORKFLOWS_QUEUE) with storage cleared and your result in one channel
• (Optional) gate per-recording with dispatch: conditional + needs + needsMode