@@ -691,9 +691,9 @@ def start_streaming_jobs(
     2. Audio persistence job - writes audio chunks to WAV file (file rotation per conversation)
 
     Args:
-        session_id: Stream session ID (equals client_id for streaming)
+        session_id: Stream/audio session ID
         user_id: User identifier
-        client_id: Client identifier
+        client_id: Connected device identifier
 
     Returns:
         Dict with job IDs: {'speech_detection': job_id, 'audio_persistence': job_id}
 
@@ -42,6 +42,7 @@
 from advanced_omi_backend.models.conversation import create_conversation
 from advanced_omi_backend.plugins.events import BUTTON_STATE_TO_EVENT, ButtonState
 from advanced_omi_backend.redis_factory import create_async_redis
+from advanced_omi_backend.redis_keys import ClientId, device_downlink_channel
 from advanced_omi_backend.services.audio_stream.conversation_lifecycle import (
     ensure_active_session_placeholder,
 )
@@ -199,7 +200,9 @@ async def subscribe_to_interim_results(websocket: WebSocket, session_id: str) ->
             )
 
 
-async def subscribe_to_device_downlink(websocket: WebSocket, client_id: str) -> None:
+async def subscribe_to_device_downlink(
+    websocket: WebSocket, client_id: ClientId
+) -> None:
     """Forward backend→device control messages from Redis Pub/Sub to the device WebSocket.
 
     Any backend component (wake-word service, plugins) can push a message to a
@@ -209,10 +212,13 @@ async def subscribe_to_device_downlink(websocket: WebSocket, client_id: str) ->
 
     Runs as a background task for the lifetime of the WebSocket connection.
     """
-    channel = f"device:downlink:{client_id}"
+    if not isinstance(client_id, ClientId):
+        raise TypeError("subscribe_to_device_downlink requires ClientId")
+    client_id_value = str(client_id)
+    channel = str(device_downlink_channel(client_id))
     redis_client = None
     pubsub = None
-    opus_stream = is_opus_streaming_client(client_id)
+    opus_stream = is_opus_streaming_client(client_id_value)
 
     try:
         redis_client = create_async_redis(decode_responses=True)
@@ -253,14 +259,14 @@ async def subscribe_to_device_downlink(websocket: WebSocket, client_id: str) ->
                         # device; for others there's nothing streaming to cancel, so
                         # just forward the control frame best-effort.
                         if opus_stream:
-                            await stop_play_audio(websocket, client_id)
+                            await stop_play_audio(websocket, client_id_value)
                         else:
                             await websocket.send_json(payload)
                     elif opus_stream and msg_type == "play-audio":
                         # RAM-limited devices can't take a big base64 WAV frame;
                         # transcode + stream it as small Opus packets instead.
                         streamed = await stream_play_audio_as_opus(
-                            websocket, payload.get("data") or {}, client_id
+                            websocket, payload.get("data") or {}, client_id_value
                         )
                         if not streamed:
                             await websocket.send_json(payload)  # fallback
@@ -1545,7 +1551,9 @@ async def _websocket_session(ws, token, device_name, connection_type):
         audio_stream_producer = get_audio_stream_producer()
 
         # Forward backend→device control messages (tones, TTS) for this device.
-        downlink_task = asyncio.create_task(subscribe_to_device_downlink(ws, client_id))
+        downlink_task = asyncio.create_task(
+            subscribe_to_device_downlink(ws, ClientId.from_value(client_id))
+        )
 
         yield (client_id, client_state, user, audio_stream_producer, interim_holder)
 
 
@@ -11,6 +11,7 @@
 
 from advanced_omi_backend.models.conversation import Conversation
 from advanced_omi_backend.redis_factory import create_async_redis
+from advanced_omi_backend.redis_keys import ClientId, device_downlink_channel
 from advanced_omi_backend.services.audio_stream.session_store import SessionStore
 from advanced_omi_backend.users import User
 
@@ -135,7 +136,10 @@ async def stop_playback(self, client_id: str) -> bool:
             logger.warning("stop_playback called with no client_id")
             return False
         message = json.dumps({"type": "stop-audio", "data": {}})
-        await self._async_redis.publish(f"device:downlink:{client_id}", message)
+        client_ref = ClientId.from_value(client_id)
+        await self._async_redis.publish(
+            str(device_downlink_channel(client_ref)), message
+        )
         logger.info(f"⏹ Requested stop-audio for {client_id}")
         return True
 
 
@@ -10,47 +10,143 @@
 (including ``utils.conversation_utils``, which the RQ workers import, and
 ``SessionStore``) without risk of an import cycle.
 
-**Scope (Stage 1):** the session-scoped pointer/lock keys and the
-``speech_detection_job`` family. Other key families (``transcription:*``,
-``audio:stream:*``, ``sse:*``, …) are still built inline and may migrate here later.
+**Current scope:** session-scoped pointer/lock keys, ``speech_detection_job`` keys,
+raw audio stream names, final transcription result stream names, and device
+downlink channels. Other key families (``sse:*``, some admin/debug keys, …) are
+still built inline and may migrate here later.
 
 **Invariant:** every WebSocket recording has a unique ``session_id``. A client may
 have many historical sessions, and each session owns an immutable raw-audio WAL.
 """
 
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class _Identity:
+    """Runtime-distinct string identity.
+
+    These are intentionally not ``NewType`` aliases. ``NewType`` helps static
+    checkers, but at runtime it is still just ``str``. These value objects let key
+    builders reject a session identity where a device identity is required.
+    """
+
+    value: str
+
+    @classmethod
+    def from_value(cls, value: object, field_name: str | None = None):
+        label = field_name or cls.__name__
+        if value is None:
+            raise ValueError(f"{label} is required")
+        if isinstance(value, bytes):
+            value = value.decode()
+        if not isinstance(value, str):
+            raise TypeError(f"{label} must be a string")
+        value = value.strip()
+        if not value:
+            raise ValueError(f"{label} cannot be empty")
+        return cls(value)
+
+    def __str__(self) -> str:
+        return self.value
+
+
+class ClientId(_Identity):
+    """Stable connected device identity, e.g. ``a421c9-elato``."""
+
+
+class SessionId(_Identity):
+    """Per-WebSocket/audio-stream identity."""
+
+
+class UserId(_Identity):
+    """Chronicle user identity carried across process boundaries."""
+
+
+class AudioStreamName(_Identity):
+    """Redis stream name for raw audio WAL entries."""
+
+
+class DeviceDownlinkChannel(_Identity):
+    """Redis Pub/Sub channel consumed by a connected device WebSocket."""
+
+
+class TranscriptionResultsStream(_Identity):
+    """Redis stream name for final transcription results."""
+
+
+def _identity_value(identity: _Identity | str) -> str:
+    return identity.value if isinstance(identity, _Identity) else identity
+
+
+def _require_identity(value: object, expected_type: type[_Identity]) -> _Identity:
+    if not isinstance(value, expected_type):
+        raise TypeError(
+            f"expected {expected_type.__name__}, got {type(value).__name__}"
+        )
+    return value
+
+
 # --- session-scoped ---
 
 
-def audio_session(session_id: str) -> str:
+def audio_session(session_id: str | SessionId) -> str:
     """Hash holding the cross-process session state (owned by ``SessionStore``)."""
-    return f"audio:session:{session_id}"
+    return f"audio:session:{_identity_value(session_id)}"
 
 
-def session_signal(session_id: str) -> str:
+def session_signal(session_id: str | SessionId) -> str:
     """Pub/sub channel for session lifecycle signals."""
-    return f"session:signal:{session_id}"
+    return f"session:signal:{_identity_value(session_id)}"
 
 
-def session_conversation_count(session_id: str) -> str:
+def session_conversation_count(session_id: str | SessionId) -> str:
     """Counter of conversations opened during the session."""
-    return f"session:conversation_count:{session_id}"
+    return f"session:conversation_count:{_identity_value(session_id)}"
 
 
-def conversation_current(session_id: str) -> str:
+def conversation_current(session_id: str | SessionId) -> str:
     """Pointer to the session's current conversation (drives WAV rotation)."""
-    return f"conversation:current:{session_id}"
+    return f"conversation:current:{_identity_value(session_id)}"
 
 
-def conversation_create_lock(session_id: str) -> str:
+def conversation_create_lock(session_id: str | SessionId) -> str:
     """Per-session lock serializing first-conversation creation across workers."""
-    return f"conversation:create_lock:{session_id}"
+    return f"conversation:create_lock:{_identity_value(session_id)}"
 
 
-def speech_detection_job(session_id: str) -> str:
+def speech_detection_job(session_id: str | SessionId) -> str:
     """Pointer to the live speech-detection job id for the session."""
-    return f"speech_detection_job:{session_id}"
+    return f"speech_detection_job:{_identity_value(session_id)}"
 
 
-def speech_detection_enqueue_lock(session_id: str) -> str:
+def speech_detection_enqueue_lock(session_id: str | SessionId) -> str:
     """Single-flight lock guarding speech-detection job enqueue bursts."""
-    return f"speech_detection_enqueue_lock:{session_id}"
+    return f"speech_detection_enqueue_lock:{_identity_value(session_id)}"
+
+
+def audio_stream(session_id: SessionId) -> AudioStreamName:
+    """Redis stream for the session's raw audio WAL."""
+    session_id = _require_identity(session_id, SessionId)
+    return AudioStreamName.from_value(f"audio:stream:{session_id}")
+
+
+def parse_audio_stream_name(stream_name: str | AudioStreamName) -> SessionId:
+    """Extract a ``SessionId`` from ``audio:stream:{session_id}``."""
+    value = _identity_value(stream_name)
+    prefix = "audio:stream:"
+    if not value.startswith(prefix):
+        raise ValueError(f"not an audio stream name: {value!r}")
+    return SessionId.from_value(value.removeprefix(prefix), "session_id")
+
+
+def transcription_results_stream(session_id: SessionId) -> TranscriptionResultsStream:
+    """Redis stream for the session's final transcription results."""
+    session_id = _require_identity(session_id, SessionId)
+    return TranscriptionResultsStream.from_value(f"transcription:results:{session_id}")
+
+
+def device_downlink_channel(client_id: ClientId) -> DeviceDownlinkChannel:
+    """Redis Pub/Sub channel for backend-to-device output."""
+    client_id = _require_identity(client_id, ClientId)
+    return DeviceDownlinkChannel.from_value(f"device:downlink:{client_id}")
@@ -287,14 +287,15 @@ async def init_session(
     ) -> None:
         """Create the session hash (status=active). Single atomic write.
 
-        ``session_id`` is stable across reconnects (it equals ``client_id``), so a
-        reconnecting device re-initializes the SAME hash. Connection-scoped fields
-        must therefore be RESET here, not left to carry over from the previous
-        connection: a stale ``transcription_error`` would make the next speech
-        detection job break on iteration 1, a stale ``transcription_seconds_sent``
-        would shift the new connection's transcript timestamps by a bogus offset,
-        and a stale ``completion_reason`` would misclassify the new connection's
-        diagnostics. They are explicitly cleared in the mapping below.
+        ``session_id`` is the audio-session identity; ``client_id`` is the connected
+        device identity. They may differ, and the session hash is the authoritative
+        join point between them. Re-initializing a session hash must RESET
+        connection-scoped fields instead of carrying them over: a stale
+        ``transcription_error`` would make the next speech detection job break on
+        iteration 1, a stale ``transcription_seconds_sent`` would shift the
+        connection's transcript timestamps by a bogus offset, and a stale
+        ``completion_reason`` would misclassify diagnostics. They are explicitly
+        cleared in the mapping below.
         """
         now = str(time.time())
         await self._redis.hset(
 
@@ -679,7 +679,7 @@ async def transcribe(self, audio_data: bytes, sample_rate: int, **kwargs) -> dic
         )
 
     async def start_stream(
-        self, client_id: str, sample_rate: int = 16000, diarize: bool = False
+        self, stream_id: str, sample_rate: int = 16000, diarize: bool = False
     ):
         base_url = self.model.resolved_url()
         ops = self.model.operations or {}
@@ -752,7 +752,7 @@ async def start_stream(
         if start_msg:
             # Inject session_id if placeholder present
             start_msg = json.loads(json.dumps(start_msg))  # deep copy
-            start_msg.setdefault("session_id", client_id)
+            start_msg.setdefault("session_id", stream_id)
             # Apply sample rate and diarization if present
             if "config" in start_msg and isinstance(start_msg["config"], dict):
                 start_msg["config"].setdefault("sample_rate", sample_rate)
@@ -766,28 +766,28 @@ async def start_stream(
             except Exception:
                 pass
 
-        self._streams[client_id] = {
+        self._streams[stream_id] = {
             "ws": ws,
             "sample_rate": sample_rate,
             "final": None,
             "interim": [],
         }
 
     async def process_audio_chunk(
-        self, client_id: str, audio_chunk: bytes
+        self, stream_id: str, audio_chunk: bytes
     ) -> dict | None:
-        if client_id not in self._streams:
+        if stream_id not in self._streams:
             return None
-        ws = self._streams[client_id]["ws"]
+        ws = self._streams[stream_id]["ws"]
         ops = self.model.operations or {}
 
         # Send chunk header if required (for providers like Parakeet)
         chunk_hdr = (ops.get("chunk_header", {}) or {}).get("message", {})
         if chunk_hdr:
             hdr = json.loads(json.dumps(chunk_hdr))
             hdr.setdefault("type", "audio_chunk")
-            hdr.setdefault("session_id", client_id)
-            hdr.setdefault("rate", self._streams[client_id]["sample_rate"])
+            hdr.setdefault("session_id", stream_id)
+            hdr.setdefault("rate", self._streams[stream_id]["sample_rate"])
             await ws.send(json.dumps(hdr))
 
         # Send audio chunk (raw bytes for Deepgram, or after header for others)
@@ -862,13 +862,13 @@ async def process_audio_chunk(
             # healthy socket that simply has no transcript available yet.
             raise
         except Exception as e:
-            logger.error(f"Error processing audio chunk result for {client_id}: {e}")
+            logger.error(f"Error processing audio chunk result for {stream_id}: {e}")
             return None
 
-    async def end_stream(self, client_id: str) -> dict:
-        if client_id not in self._streams:
+    async def end_stream(self, stream_id: str) -> dict:
+        if stream_id not in self._streams:
             return {"text": "", "words": [], "segments": []}
-        ws = self._streams[client_id]["ws"]
+        ws = self._streams[stream_id]["ws"]
         ops = self.model.operations or {}
         end_msg = (ops.get("end", {}) or {}).get("message", {"type": "stop"})
         await ws.send(json.dumps(end_msg))
@@ -896,7 +896,7 @@ async def end_stream(self, client_id: str) -> dict:
         except Exception:
             pass
 
-        self._streams.pop(client_id, None)
+        self._streams.pop(stream_id, None)
 
         if not isinstance(final, dict):
             return {"text": "", "words": [], "segments": []}
 
@@ -103,20 +103,20 @@ def capabilities(self) -> set:
 
     @abc.abstractmethod
     async def start_stream(
-        self, client_id: str, sample_rate: int = 16000, diarize: bool = False
+        self, stream_id: str, sample_rate: int = 16000, diarize: bool = False
     ):
-        """Start a transcription stream for a client.
+        """Start a transcription stream.
 
         Args:
-            client_id: Unique client identifier
+            stream_id: Unique provider stream/session identifier
             sample_rate: Audio sample rate
             diarize: Whether to enable speaker diarization (provider-dependent)
         """
         pass
 
     @abc.abstractmethod
     async def process_audio_chunk(
-        self, client_id: str, audio_chunk: bytes
+        self, stream_id: str, audio_chunk: bytes
     ) -> Optional[dict]:
         """
         Process audio chunk and return partial/final transcription.
@@ -127,7 +127,7 @@ async def process_audio_chunk(
         pass
 
     @abc.abstractmethod
-    async def end_stream(self, client_id: str) -> dict:
+    async def end_stream(self, stream_id: str) -> dict:
         """End stream and return final transcription with word-level timestamps."""
         pass