Update comments for workflows integration in pipe-analysis and values.yaml

charts/hub/templates/kerberos-pipeline/_workflows-helpers.tpl

@@ -0,0 +1,32 @@
+{{/*
+Assemble the workflows engine stage registry (PIPELINE_STAGE_REGISTRY) as a
+JSON array from every *enabled* stage under kerberoshub.workflows.stages, so the
+engine's routing stays in lockstep with the deployed stage workers and the
+per-stage queue cannot drift. The `stages:` block is the routing source of truth;
+each stage's deployment (and its queue) lives under the matching
+kerberoshub.services.<name> entry.
+
+Each enabled stage contributes one descriptor:
+  operation  defaults to the stage's map key.
+  dispatch   defaults to "always".
+  queue      taken from the matching services.<key>.queue (authoritative; the
+             same value the worker consumes via <NAME>_QUEUE). Omitted when the
+             service or its queue is unset, so the engine derives
+             "kcloud-<operation>-queue.fifo".
+  needs      conditional stages only: the upstream dependencies (each
+             {operation, condition?}) carried through verbatim.
+*/}}
+{{- define "kerberoshub.workflows.stageRegistry" -}}
+{{- $entries := list -}}
+{{- $services := .Values.kerberoshub.services | default dict -}}
+{{- range $name, $stage := .Values.kerberoshub.workflows.stages -}}
+{{- if $stage.enabled -}}
+{{- $entry := dict "operation" (default $name $stage.operation) "dispatch" (default "always" $stage.dispatch) -}}
+{{- $service := index $services $name -}}
+{{- if $service }}{{- with $service.queue }}{{- $_ := set $entry "queue" . -}}{{- end }}{{- end }}
+{{- with $stage.needs }}{{- $_ := set $entry "needs" . -}}{{- end }}
+{{- $entries = append $entries $entry -}}
+{{- end -}}
+{{- end -}}
+{{- $entries | toJson -}}
+{{- end -}}

charts/hub/templates/kerberos-pipeline/pipe-analysis.yaml

@@ -82,9 +82,9 @@ spec:
         - name: RABBITMQ_PASSWORD
           value: "{{ .Values.rabbitmq.password }}"
 
-        # When true, analysis hands the completed result off to the hub-workflows
-        # service as the terminal pipeline stage (replacing the throttler /
-        # notification tail). Kept in sync with whether the workflows service runs.
+        # When true, analysis tees the classify result to the hub-workflows
+        # service in parallel with the throttler/notification tail (which still
+        # runs unchanged). Kept in sync with whether the workflows service runs.
         - name: WORKFLOWS_ENABLED
           value: "{{ .Values.kerberoshub.workflows.enabled }}"
           

charts/hub/templates/kerberos-pipeline/pipe-anpr.yaml

@@ -0,0 +1,100 @@
+{{- if and (or (eq .Values.mode "all") (eq .Values.mode "pipeline")) .Values.kerberoshub.workflows.enabled .Values.kerberoshub.services.anpr.enabled -}}
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: pipe-anpr
+  namespace: {{ .Release.Namespace }}
+spec:
+  replicas: {{ .Values.kerberoshub.services.anpr.replicas }}
+  selector:
+    matchLabels:
+      app: pipe-anpr
+  minReadySeconds: 10
+  strategy:
+    type: RollingUpdate
+    rollingUpdate:
+      maxUnavailable: 1
+      maxSurge: 1
+  template:
+    metadata:
+      labels:
+        app: pipe-anpr
+    spec:
+      {{- if .Values.kerberoshub.serviceAccount.create }}
+      serviceAccountName: {{ default (printf "%s-%s-sa" .Release.Name .Chart.Name | trunc 63 | trimSuffix "-") .Values.kerberoshub.serviceAccount.name }}
+      {{- end }}
+      {{- with .Values.imagePullSecrets }}
+      imagePullSecrets:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.kerberoshub.services.anpr.volumes }}
+      volumes:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.kerberoshub.services.anpr.topologySpreadConstraints }}
+      topologySpreadConstraints:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      containers:
+      - name: pipe-anpr
+        image: "{{ .Values.global.imageRegistry }}{{ .Values.kerberoshub.services.anpr.repository }}:{{ .Values.kerberoshub.services.anpr.tag }}"
+        imagePullPolicy: {{ .Values.kerberoshub.services.anpr.pullPolicy }}
+        {{- with .Values.kerberoshub.services.anpr.resources }}
+        resources:
+          {{- toYaml . | nindent 12 }}
+        {{- end }}
+        {{- with .Values.kerberoshub.services.anpr.volumeMounts }}
+        volumeMounts:
+          {{- toYaml . | nindent 12 }}
+        {{- end }}
+        env:
+        - name: LOG_LEVEL
+          value: "{{ .Values.kerberoshub.services.anpr.logLevel }}"
+        - name: QUEUE_SYSTEM
+          value: "{{ .Values.queueProvider }}"
+
+        # The queue this stage worker consumes dispatched "anpr" stages from
+        # (ANPR_QUEUE) and the workflows engine queue it routes its result back
+        # to (WORKFLOWS_QUEUE, so the run records the resolution and any stage
+        # that needs "anpr" can fire).
+        - name: ANPR_QUEUE
+          value: "{{ .Values.kerberoshub.services.anpr.queue }}"
+        - name: WORKFLOWS_QUEUE
+          value: "{{ .Values.kerberoshub.workflows.queue }}"
+
+        # RabbitMQ settings
+        - name: RABBITMQ_HOST
+          value: "{{ .Values.rabbitmq.host }}"
+        - name: RABBITMQ_EXCHANGE
+          value: "{{ .Values.rabbitmq.exchange }}"
+        - name: RABBITMQ_USERNAME
+          value: "{{ .Values.rabbitmq.username }}"
+        - name: RABBITMQ_PASSWORD
+          value: "{{ .Values.rabbitmq.password }}"
+
+        # Kerberos Vault — global storage credentials this stage uses to fetch
+        # the media it operates on.
+        - name: KERBEROS_STORAGE_URI
+          value: "{{ .Values.kerberosvault.uri }}"
+        - name: KERBEROS_STORAGE_ACCESS_KEY
+          value: "{{ .Values.kerberosvault.accesskey }}"
+        - name: KERBEROS_STORAGE_SECRET
+          value: "{{ .Values.kerberosvault.secretkey }}"
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: pipe-anpr
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app: pipe-anpr
+    service: pipe
+spec:
+  ports:
+    - name: hub-metrics
+      port: 8080
+      targetPort: 8080
+      protocol: TCP
+  selector:
+    app: pipe-anpr
+{{- end }}

charts/hub/templates/kerberos-pipeline/pipe-workflows.yaml

@@ -60,10 +60,12 @@ spec:
 
         # Queue this service consumes from (WORKFLOWS_QUEUE) and the custom
         # pipeline stage registry it may dispatch (PIPELINE_STAGE_REGISTRY).
+        # The registry is assembled from the enabled stages under
+        # kerberoshub.workflows.stages (see _workflows-helpers.tpl).
         - name: WORKFLOWS_QUEUE
           value: "{{ .Values.kerberoshub.workflows.queue }}"
         - name: PIPELINE_STAGE_REGISTRY
-          value: {{ .Values.kerberoshub.workflows.stageRegistry | quote }}
+          value: {{ include "kerberoshub.workflows.stageRegistry" . | quote }}
 
         # RabbitMQ settings
         - name: RABBITMQ_HOST

charts/hub/values.yaml

@@ -598,10 +598,10 @@ kerberoshub:
   # (not a document) with the analysis pipeline and is meant to grow into the
   # primary orchestrator. See https://github.com/uug-ai/hub-workflows.
   workflows:
-    # Disabled by default. When enabled, the analysis service hands each
-    # completed analysis off to this service as the terminal pipeline stage
-    # (replacing the throttler/notification tail) via WORKFLOWS_ENABLED on
-    # pipe-analysis. Flip to true to run the workflows engine.
+    # Disabled by default. When enabled, the analysis service tees each
+    # classify result to this service (via WORKFLOWS_ENABLED on pipe-analysis)
+    # in parallel with the normal throttler/notification tail, which still runs
+    # unchanged. Flip to true to run the workflows engine.
     enabled: false
     repository: ghcr.io/uug-ai/hub-workflows
     pullPolicy: IfNotPresent
@@ -619,33 +619,75 @@ kerberoshub:
     # Queue this service consumes ingest events and upstream results from
     # (WORKFLOWS_QUEUE). Must be fed the same messages the analysis service sees.
     queue: "kcloud-workflows-queue"
-    # Custom pipeline stage registry (PIPELINE_STAGE_REGISTRY): a JSON array of
-    # stage descriptors the engine may dispatch. Empty = no custom stages, so the
-    # engine runs as a safe no-op consumer (seeds runs, records results, dispatches
-    # nothing).
-    #
-    # Each descriptor carries the routing fields:
-    #   operation  (required) unique stage id; binds its queue and resolution.
-    #   dispatch   "always" (default) | "conditional".
-    #   needs      conditional stages only: an array of upstream dependencies,
-    #              each {"operation": <upstream>, "condition": <predicate?>}. The
-    #              stage fires when a listed upstream resolves and that
-    #              dependency's condition matches (omit condition = unconditional).
-    #   condition  predicate shape: {"path": <dot-path>, "op":
-    #              eq|ne|contains|exists|gt|lt, "value": <operand>}.
-    # Example (one always stage, one conditional fan-in on classify):
-    #   stageRegistry: |
-    #     [
-    #       {"operation":"anpr","dispatch":"always"},
-    #       {"operation":"nohelmet","dispatch":"conditional",
-    #        "needs":[{"operation":"classify",
-    #                  "condition":{"path":"label","op":"eq","value":"person"}}]}
-    #     ]
-    stageRegistry: "[]"
     resources:
       requests:
         memory: 10Mi
         cpu: 10m
+    # Stage routing — how each custom stage fits into the workflow. The engine's
+    # PIPELINE_STAGE_REGISTRY is assembled automatically from the *enabled*
+    # stages below: each contributes its operation and dispatch/needs routing,
+    # plus the queue from its matching `services.<name>` entry — so the engine's
+    # routing stays in lockstep with the deployed worker and the queue cannot
+    # drift. There is nothing to set for the registry directly.
+    #
+    # Each stage is keyed to a worker of the same name under kerberoshub.services.
+    # Routing (here) and deployment (services) are toggled independently:
+    #   workflows.stages.<name>.enabled  -> include the stage in the registry (route to it)
+    #   services.<name>.enabled          -> deploy the worker (pipe-<name>.yaml)
+    stages:
+      # hub-anpr — automatic number-plate recognition stage. Routing only; its
+      # worker deployment lives under kerberoshub.services.anpr.
+      anpr:
+        # Include this stage in the engine's registry (route to it).
+        enabled: false
+        # operation  unique stage id (defaults to the key "anpr" if omitted);
+        #            binds the queue (taken from services.anpr.queue) and how
+        #            the result is recorded.
+        # dispatch   "always" (run on every workflow) | "conditional".
+        # needs      conditional stages only: upstream dependencies, each
+        #            {operation, condition?}. condition shape:
+        #            {path: <dot-path>, op: eq|ne|contains|exists|gt|lt, value: <operand>}.
+        operation: anpr
+        dispatch: always
+        # Example conditional routing — only run anpr when classify reports a car:
+        #   dispatch: conditional
+        #   needs:
+        #     - operation: classify
+        #       condition: {path: "classify.details.0.classified", op: eq, value: car}
+  # Workflow stage worker deployments (siblings of the workflows engine). Each
+  # entry is a standalone service the engine dispatches to; it consumes its own
+  # queue and routes a result back to the workflows queue. Keyed to match a
+  # workflows.stages entry of the same name. Deploying a worker here is
+  # independent from routing to it (workflows.stages.<name>.enabled) — a worker
+  # only runs when its `enabled` is true AND the workflows engine is enabled.
+  services:
+    # hub-anpr — automatic number-plate recognition worker. Lives in the
+    # hub-workflows repository as its own module. See
+    # https://github.com/uug-ai/hub-workflows/tree/main/hub-anpr.
+    anpr:
+      # Deploy the pipe-anpr worker.
+      enabled: false
+      repository: ghcr.io/uug-ai/hub-anpr
+      pullPolicy: IfNotPresent
+      tag: "v1.0.0"
+      replicas: 1 # Number of pods for the worker.
+      topologySpreadConstraints: [] # Optional pod topology spread constraints (empty = none).
+      #volumes:
+      #  - name: extra
+      #    emptyDir: {}
+      #volumeMounts:
+      #  - name: extra
+      #    mountPath: /data
+      logLevel: "info" # possible values: trace, debug, info, warn, error
+      # Queue this worker consumes dispatched messages from (ANPR_QUEUE). This
+      # same value is read into the engine's generated registry entry for the
+      # matching stage, so the engine dispatches and the worker consumes the same
+      # queue with no drift. Convention: "kcloud-<operation>-queue.fifo".
+      queue: "kcloud-anpr-queue.fifo"
+      resources:
+        requests:
+          memory: 10Mi
+          cpu: 10m
   monitordevice:
     repository: ghcr.io/uug-ai/hub-monitor-device
     pullPolicy: IfNotPresent