@@ -16,6 +16,7 @@ RUSTSEC
 SARIF
 Standardised
 TMDL
+unkeyed
 WIQL
 Workback
 XMLA
 
@@ -35,6 +35,8 @@ jobs:
             crate: src/500-application/507-ai-inference/services/ai-edge-inference
           - name: 507-ai-edge-inference-crate
             crate: src/500-application/507-ai-inference/services/ai-edge-inference-crate
+          - name: 507-snapshot-normalizer
+            crate: src/500-application/507-ai-inference/services/snapshot-normalizer
           - name: 507-snapshot-normalizer-core
             crate: src/500-application/507-ai-inference/services/snapshot-normalizer-core
           - name: 901-video-to-gif
@@ -200,6 +202,7 @@ jobs:
           - name: 504-mqtt-otel-trace-exporter
           - name: 507-ai-edge-inference
           - name: 507-ai-edge-inference-crate
+          - name: 507-snapshot-normalizer
           - name: 507-snapshot-normalizer-core
           - name: 901-video-to-gif
     steps:
 
@@ -65,9 +65,11 @@ This component implements a scalable AI inference service designed for industria
 ├── services/                    # Service implementations
 │   ├── ai-edge-inference/       # Main inference service (Rust)
 │   ├── ai-edge-inference-crate/ # Shared Rust crate
+│   ├── snapshot-normalizer/     # MQTT snapshot adapter (Rust)
 │   └── snapshot-normalizer-core/ # Snapshot normalization library (Rust)
 ├── charts/                      # Kubernetes deployment manifests
 │   ├── base/                    # Base Kubernetes resources
+│   ├── snapshot-normalizer/     # Helm chart for the snapshot adapter
 │   └── model-downloader-job.yaml
 └── resources/                   # Configuration and model files
     ├── model_configs/           # Model configuration files
@@ -130,6 +132,24 @@ defaults, not schema-enforced limits.
 See [`services/snapshot-normalizer-core/README.md`](services/snapshot-normalizer-core/README.md)
 for the full public surface.
 
+## Snapshot Normalizer
+
+`services/snapshot-normalizer/` is the MQTT adapter built on the core library.
+It subscribes to a binary JPEG snapshot topic, such as a media connector
+`snapshot-to-mqtt` stream, and publishes each accepted snapshot as an
+`image_snapshot` v1 request to
+`edge-ai/v1/snapshot-normalizer/camera/{camera-id}/snapshots`, which the
+inference service receives through its pinned `edge-ai/v1/+/camera/+/snapshots`
+input filter.
+
+- Carries CloudEvents attributes as MQTTv5 user properties
+- Deduplicates on the producer-supplied CloudEvents `id`, never on content
+- Refuses to start when its output topic matches its own input filter
+- Deploys with the [`charts/snapshot-normalizer`](charts/snapshot-normalizer/) Helm chart as a single-replica StatefulSet per camera, with a stable, unique client ID
+
+See [`services/snapshot-normalizer/README.md`](services/snapshot-normalizer/README.md)
+for configuration, delivery semantics, and an authorization example.
+
 ## Quick Start
 
 ### Local Development
 
@@ -0,0 +1,11 @@
+apiVersion: v2
+name: snapshot-normalizer
+description: Publishes binary JPEG MQTT snapshots as image_snapshot v1 requests for edge inference
+type: application
+version: 0.1.0
+appVersion: "0.1.0"
+home: https://github.com/microsoft/edge-ai
+sources:
+  - https://github.com/microsoft/edge-ai
+maintainers:
+  - name: Edge AI Team
@@ -0,0 +1,51 @@
+{{- define "snapshot-normalizer.name" -}}
+{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{- define "snapshot-normalizer.fullname" -}}
+{{- if .Values.fullnameOverride }}
+{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- $name := default .Chart.Name .Values.nameOverride }}
+{{- if contains $name .Release.Name }}
+{{- .Release.Name | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }}
+{{- end }}
+{{- end }}
+{{- end }}
+
+{{- define "snapshot-normalizer.chart" -}}
+{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{- define "snapshot-normalizer.labels" -}}
+helm.sh/chart: {{ include "snapshot-normalizer.chart" . }}
+{{ include "snapshot-normalizer.selectorLabels" . }}
+{{- if .Chart.AppVersion }}
+app.kubernetes.io/version: {{ .Chart.AppVersion | quote }}
+{{- end }}
+app.kubernetes.io/managed-by: {{ .Release.Service }}
+{{- end }}
+
+{{- define "snapshot-normalizer.selectorLabels" -}}
+app.kubernetes.io/name: {{ include "snapshot-normalizer.name" . }}
+app.kubernetes.io/instance: {{ .Release.Name }}
+{{- end }}
+
+{{- define "snapshot-normalizer.namespace" -}}
+{{- default "azure-iot-operations" .Values.namespace }}
+{{- end }}
+
+{{- define "snapshot-normalizer.serviceAccountName" -}}
+{{- if .Values.serviceAccount.create }}
+{{- default (include "snapshot-normalizer.fullname" .) .Values.serviceAccount.name }}
+{{- else }}
+{{- required "serviceAccount.name is required when serviceAccount.create is false" .Values.serviceAccount.name }}
+{{- end }}
+{{- end }}
+
+{{- define "snapshot-normalizer.image" -}}
+{{- $repository := required "image.repository is required" .Values.image.repository }}
+{{- printf "%s:%s" $repository (.Values.image.tag | default .Chart.AppVersion) }}
+{{- end }}
@@ -0,0 +1,14 @@
+{{- if .Values.serviceAccount.create }}
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: {{ include "snapshot-normalizer.serviceAccountName" . }}
+  namespace: {{ include "snapshot-normalizer.namespace" . }}
+  labels:
+    {{- include "snapshot-normalizer.labels" . | nindent 4 }}
+  {{- with .Values.serviceAccount.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+automountServiceAccountToken: false
+{{- end }}
@@ -0,0 +1,126 @@
+{{- $n := .Values.normalizer }}
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: {{ include "snapshot-normalizer.fullname" . }}
+  namespace: {{ include "snapshot-normalizer.namespace" . }}
+  labels:
+    {{- include "snapshot-normalizer.labels" . | nindent 4 }}
+spec:
+  # In-memory deduplication and a non-shared subscription require one replica.
+  # The StatefulSet keeps the pod name, and so the MQTT client ID, stable at
+  # <fullname>-0 and never runs two copies, so the persistent session resumes.
+  replicas: 1
+  serviceName: {{ include "snapshot-normalizer.fullname" . }}
+  updateStrategy:
+    type: RollingUpdate
+  selector:
+    matchLabels:
+      {{- include "snapshot-normalizer.selectorLabels" . | nindent 6 }}
+  template:
+    metadata:
+      {{- with .Values.podAnnotations }}
+      annotations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      labels:
+        {{- include "snapshot-normalizer.selectorLabels" . | nindent 8 }}
+        {{- with .Values.podLabels }}
+        {{- toYaml . | nindent 8 }}
+        {{- end }}
+    spec:
+      {{- with .Values.imagePullSecrets }}
+      imagePullSecrets:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      serviceAccountName: {{ include "snapshot-normalizer.serviceAccountName" . }}
+      automountServiceAccountToken: false
+      securityContext:
+        {{- toYaml .Values.podSecurityContext | nindent 8 }}
+      containers:
+        - name: {{ .Chart.Name }}
+          securityContext:
+            {{- toYaml .Values.securityContext | nindent 12 }}
+          image: {{ include "snapshot-normalizer.image" . }}
+          imagePullPolicy: {{ .Values.image.pullPolicy }}
+          env:
+            - name: POD_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.name
+            - name: AIO_MQTT_CLIENT_ID
+              value: {{ printf "%s-$(POD_NAME)" (required "mqtt.clientIdPrefix is required" .Values.mqtt.clientIdPrefix) | quote }}
+            - name: AIO_BROKER_HOSTNAME
+              value: {{ .Values.mqtt.brokerHostname | quote }}
+            - name: AIO_BROKER_TCP_PORT
+              value: {{ .Values.mqtt.brokerTcpPort | quote }}
+            - name: AIO_MQTT_USE_TLS
+              value: "true"
+            - name: AIO_MQTT_SESSION_EXPIRY
+              value: {{ required "mqtt.sessionExpirySeconds is required" .Values.mqtt.sessionExpirySeconds | int64 | quote }}
+            - name: AIO_TLS_CA_FILE
+              value: {{ .Values.mqtt.tlsCaFile | quote }}
+            - name: AIO_SAT_FILE
+              value: {{ .Values.mqtt.satFile | quote }}
+            - name: CAMERA_ID
+              value: {{ required "normalizer.cameraId is required" $n.cameraId | quote }}
+            - name: DEVICE_NAME
+              value: {{ required "normalizer.deviceName is required" $n.deviceName | quote }}
+            - name: INPUT_TOPIC
+              value: {{ required "normalizer.inputTopic is required" $n.inputTopic | quote }}
+            {{- with $n.outputTopic }}
+            - name: OUTPUT_TOPIC
+              value: {{ . | quote }}
+            {{- end }}
+            - name: EVENT_SOURCE
+              value: {{ $n.eventSource | quote }}
+            {{- with $n.dataSchema }}
+            - name: DATA_SCHEMA
+              value: {{ . | quote }}
+            {{- end }}
+            - name: MAX_JPEG_BYTES
+              value: {{ $n.maxJpegBytes | int64 | quote }}
+            - name: MAX_ENVELOPE_BYTES
+              value: {{ $n.maxEnvelopeBytes | int64 | quote }}
+            - name: DEDUP_CAPACITY
+              value: {{ $n.dedupCapacity | int64 | quote }}
+            - name: DEDUP_WINDOW_SECONDS
+              value: {{ $n.dedupWindowSeconds | int64 | quote }}
+            - name: PUBLISH_ATTEMPTS
+              value: {{ $n.publishAttempts | int64 | quote }}
+            - name: COUNTERS_INTERVAL_SECONDS
+              value: {{ $n.countersIntervalSeconds | int64 | quote }}
+            - name: RUST_LOG
+              value: {{ $n.logLevel | quote }}
+          volumeMounts:
+            - name: mq-sat
+              mountPath: {{ dir .Values.mqtt.satFile }}
+              readOnly: true
+            - name: trust-bundle
+              mountPath: {{ dir .Values.mqtt.tlsCaFile }}
+              readOnly: true
+          resources:
+            {{- toYaml .Values.resources | nindent 12 }}
+      volumes:
+        - name: mq-sat
+          projected:
+            sources:
+              - serviceAccountToken:
+                  path: {{ base .Values.mqtt.satFile }}
+                  audience: {{ .Values.volumes.mqSat.audience | quote }}
+                  expirationSeconds: {{ .Values.volumes.mqSat.expirationSeconds | int64 }}
+        - name: trust-bundle
+          configMap:
+            name: {{ .Values.volumes.trustBundle.configMapName }}
+      {{- with .Values.nodeSelector }}
+      nodeSelector:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.affinity }}
+      affinity:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.tolerations }}
+      tolerations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
@@ -0,0 +1,98 @@
+# The adapter keeps deduplication state in memory and subscribes without a
+# shared subscription, so the chart always runs a single-replica StatefulSet.
+# Run one release per input topic to scale out.
+
+image:
+  # Placeholder. Build the image from services/snapshot-normalizer and push it
+  # to a registry you own.
+  repository: your-registry.azurecr.io/snapshot-normalizer
+  pullPolicy: IfNotPresent
+  tag: ""
+
+imagePullSecrets: []
+nameOverride: ""
+fullnameOverride: ""
+namespace: azure-iot-operations
+
+serviceAccount:
+  # Use a distinct service account per workload trust boundary so broker
+  # authorization rules can scope this workload alone.
+  create: true
+  name: ""
+  # Annotations prefixed with aio-broker-auth/ become broker authorization
+  # attributes for clients that authenticate with this service account token.
+  annotations:
+    aio-broker-auth/workload: snapshot-normalizer
+
+normalizer:
+  # Opaque, lowercase camera identifier. It becomes the subject attribute and a
+  # segment of the default output topic.
+  cameraId: camera-01
+  deviceName: camera-device-01
+  # Topic the media connector snapshot-to-mqtt stream publishes to.
+  inputTopic: media/v1/media-connector/camera-01/snapshots
+  # Leave empty for edge-ai/v1/snapshot-normalizer/camera/<cameraId>/snapshots,
+  # which the component 507 inference service receives through its pinned
+  # edge-ai/v1/+/camera/+/snapshots input filter.
+  outputTopic: ""
+  eventSource: snapshot-normalizer
+  # Optional dataschema reference. Keep it consistent with any data flow source
+  # schema that consumes the output topic.
+  dataSchema: ""
+  # Keep maxEnvelopeBytes below the broker maximum packet size, which the AIO
+  # memory profile bounds (4 MB on Tiny). Base64 adds about one third.
+  maxJpegBytes: 4194304
+  maxEnvelopeBytes: 8388608
+  dedupCapacity: 1024
+  dedupWindowSeconds: 300
+  publishAttempts: 3
+  countersIntervalSeconds: 60
+  logLevel: info
+
+mqtt:
+  brokerHostname: aio-broker.azure-iot-operations
+  brokerTcpPort: "18883"
+  # The pod connects as <clientIdPrefix>-<release fullname>-0, which is stable
+  # across restarts so the broker resumes the persistent session.
+  clientIdPrefix: snapshot-normalizer
+  # Seconds the broker keeps the session, and its queued QoS 1 snapshots, after
+  # the pod disconnects. Bounds orphaned sessions; 0 discards them at once.
+  sessionExpirySeconds: 3600
+  tlsCaFile: /var/run/certs/ca.crt
+  satFile: /var/run/secrets/tokens/mq-sat
+
+volumes:
+  mqSat:
+    audience: aio-internal
+    expirationSeconds: 86400
+  trustBundle:
+    configMapName: azure-iot-operations-aio-ca-trust-bundle
+
+resources:
+  limits:
+    cpu: 500m
+    memory: 256Mi
+  requests:
+    cpu: 50m
+    memory: 64Mi
+
+podSecurityContext:
+  runAsNonRoot: true
+  runAsUser: 1000
+  runAsGroup: 1000
+  fsGroup: 1000
+  seccompProfile:
+    type: RuntimeDefault
+
+securityContext:
+  allowPrivilegeEscalation: false
+  capabilities:
+    drop:
+      - ALL
+  readOnlyRootFilesystem: true
+
+nodeSelector: {}
+tolerations: []
+affinity: {}
+podAnnotations: {}
+podLabels: {}
@@ -40,6 +40,23 @@ services:
     networks:
       - edge-ai-inference
 
+  # Snapshot normalizer: republishes binary JPEG snapshots as image_snapshot v1
+  # requests on edge-ai/v1/snapshot-normalizer/camera/{CAMERA_ID}/snapshots,
+  # which the inference service receives through edge-ai/v1/+/camera/+/snapshots
+  snapshot-normalizer:
+    build:
+      context: ./services
+      dockerfile: snapshot-normalizer/Dockerfile
+    image: ${REGISTRY:-local}/${APPLICATION:-ai-inference}/snapshot-normalizer:${BUILD_ID:-local}
+    # Defaults from the template; an optional .env beside it overrides them
+    env_file:
+      - path: ./services/snapshot-normalizer/.env.template
+      - path: ./services/snapshot-normalizer/.env
+        required: false
+    read_only: true
+    networks:
+      - edge-ai-inference
+
   # Optional: MQTT client for testing
   mqtt-client:
     image: eclipse-mosquitto:latest