fix: separate identities that shared a runtime representation

Seven confirmed defects from the typed-contracts audit, each one two
semantically different values wearing the same type and name.

Speaker tenant. speaker_recognition_client sent a literal user_id="1" on
every diarization, identification and wake-word gating call, so every
Chronicle user resolved to one shared gallery. The speaker service now
takes Chronicle's own user id as its tenant (users.id and the three FKs
become String) and the client passes it. Callers that had no tenant to
give — guided enrollment, fine-tuning, drift, enrollment health — thread
the owning user instead of defaulting to 1.

registered_clients. Stored as a {client_id: {...}} mapping while the
inverse lookup queried registered_clients.client_id, which MongoDB
resolves across array elements but not object keys, so it matched nothing
for every real device. Disconnects never stamped last_seen and admins
could not act on another user's device. Now a typed list[RegisteredClient].

ClientId. The upload controller had a second generate_client_id that
interpolated the raw device name, so one device got two identities
depending on how its audio arrived. One constructor now; server-generated
labels use synthetic_client_id, which does not sanitize a name that was
never user input.

End reason vs processing trigger. file_upload, reprocess_orphan,
reprocess_transcript and rebound were passed as end_reason while none is
an EndReason member, so each was stored as UNKNOWN while the emitted event
still reported the raw string. They travel in ProcessingTrigger now, and a
reprocess passes no end reason at all, so a recording keeps the reason it
really ended with.

Transcription provider vs mode. The streaming consumer wrote
provider=b"streaming" — a mode — which persisted as the transcript
version's provider AND model, losing which service produced it. provider,
mode and model are three fields now. The "deepgram" fallback is gone: a
guessed provider is indistinguishable from a real one afterwards.

Timeline audio claims. Split deep-copied every audio range onto both
halves, so both played the whole original; merge never unioned them, so
absorbed episodes' audio died with their documents. Ranges are now clipped
on split and unioned on merge. A chunk with no captured_at (3% here) can
be placed on neither side, so it stays with the head rather than being
dropped or double-claimed.

Durable spool identity. The phone's spool-file id travelled as
durable_session_id and came back as session_id, colliding with the backend
SessionId that means something else. Now spool_segment_id end to end.

Also fixed, same class, found while working:

- DerivedOperation was {split, merge} but maybe_trim_silence passes
  "silence_trim", so the remnant's lineage record raised at construction
  and silence trimming crashed outright.
- The wake-word service could never start: identities.py is committed but
  absent from both the .dockerignore allowlist and the Dockerfile COPY.
- A truncated day write was either mislabelled `written` or retried three
  times and settled `skipped`. DayWriteOutcome.PARTIAL is a third terminal
  state so it is neither.

Author: AnkushMalaker

app/src/hooks/useAudioStreamer.ts

@@ -222,16 +222,16 @@ export const useAudioStreamer = (options?: UseAudioStreamerOptions): UseAudioStr
   }, [setStateSafe]);
 
   const sendDurablePacket = useCallback((packet: SpoolPacket) => {
-    pendingPacketsRef.current.set(`${packet.sessionId}:${packet.sequence}`, packet);
+    pendingPacketsRef.current.set(`${packet.segmentId}:${packet.sequence}`, packet);
     outboundChainRef.current = outboundChainRef.current.then(async () => {
       if (websocketRef.current?.readyState !== WebSocket.OPEN) return;
       await sendWyomingEvent(
         {
           type: 'audio-chunk',
           data: {
             ...AUDIO_FORMAT,
-            durable_session_id: packet.sessionId,
-            durable_sequence: packet.sequence,
+            spool_segment_id: packet.segmentId,
+            spool_sequence: packet.sequence,
             captured_at_ms: packet.capturedAtMs,
           },
         },
@@ -426,7 +426,7 @@ export const useAudioStreamer = (options?: UseAudioStreamerOptions): UseAudioStr
               return;
             }
             if (msg.type === 'audio-ack') {
-              const key = `${msg.session_id}:${msg.sequence}`;
+              const key = `${msg.spool_segment_id}:${msg.sequence}`;
               const packet = pendingPacketsRef.current.get(key);
               if (packet) {
                 pendingPacketsRef.current.delete(key);

app/src/services/durableAudioSpool.ts

@@ -8,7 +8,13 @@ const ACK_PREFIX = 'chronicle.audioSpool.ack.';
 
 export interface SpoolPacket {
   fileName: string;
-  sessionId: string;
+  /**
+   * Identity of the spool *file* this packet was written to, not the backend audio
+   * session. It was called `sessionId` and sent as `durable_session_id`, which the
+   * backend echoed back as `session_id` — three names for a spool segment, all of
+   * them colliding with the real WebSocket SessionId that means something else.
+   */
+  segmentId: string;
   sequence: number;
   capturedAtMs: number;
   payload: Uint8Array;
@@ -17,12 +23,12 @@ export interface SpoolPacket {
 interface ActiveSegment {
   file: File;
   handle: FileHandle;
-  sessionId: string;
+  segmentId: string;
   startedAtMs: number;
   nextSequence: number;
 }
 
-const makeSessionId = (): string =>
+const makeSegmentId = (): string =>
   `${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 12)}`;
 
 /**
@@ -50,13 +56,13 @@ class DurableAudioSpool {
 
   private startSegment(capturedAtMs: number): ActiveSegment {
     this.ensureDirectory();
-    const sessionId = makeSessionId();
-    const file = new File(this.directory, `${sessionId}.spool`);
+    const segmentId = makeSegmentId();
+    const file = new File(this.directory, `${segmentId}.spool`);
     file.create({ overwrite: false, intermediates: true });
     const active = {
       file,
       handle: file.open(),
-      sessionId,
+      segmentId,
       startedAtMs: capturedAtMs,
       nextSequence: 0,
     };
@@ -82,7 +88,7 @@ class DurableAudioSpool {
 
     return {
       fileName: segment.file.name,
-      sessionId: segment.sessionId,
+      segmentId: segment.segmentId,
       sequence,
       capturedAtMs,
       payload,
@@ -97,7 +103,7 @@ class DurableAudioSpool {
       .filter((entry): entry is File => entry instanceof File && entry.name.endsWith('.spool'));
 
     for (const file of files) {
-      const sessionId = file.name.slice(0, -'.spool'.length);
+      const segmentId = file.name.slice(0, -'.spool'.length);
       const acknowledged = Number(await AsyncStorage.getItem(`${ACK_PREFIX}${file.name}`) ?? '-1');
       const bytes = file.bytesSync();
       let offset = 0;
@@ -113,7 +119,7 @@ class DurableAudioSpool {
         if (sequence > acknowledged) {
           packets.push({
             fileName: file.name,
-            sessionId,
+            segmentId,
             sequence,
             capturedAtMs,
             payload: bytes.slice(offset + HEADER_BYTES, end),

backends/advanced/src/advanced_omi_backend/client_manager.py

@@ -454,10 +454,48 @@ async def some_endpoint(client_manager: ClientManager = Depends(get_client_manag
     return client_manager
 
 
+def user_id_prefix(user: "User") -> str:
+    """The user-scoping prefix of every client id: last 6 of the ObjectId.
+
+    The single place this slice is taken. Constructing it inline is how the upload
+    path grew a second, differently-normalizing ``generate_client_id`` that gave the
+    same physical device a different identity than the WebSocket path did.
+    """
+    return str(user.id)[-6:]
+
+
+def owns_client_id(user: "User", client_id: str) -> bool:
+    """Whether ``client_id`` belongs to ``user`` by its prefix.
+
+    Prefix ownership is a *weak* check — it is derived from a 6-character slice, so it
+    is only safe where the caller has already established the user. Use the registry
+    (``User.has_client`` / ``get_user_by_client_id``) when the answer must be
+    authoritative.
+    """
+    return (client_id or "").startswith(user_id_prefix(user))
+
+
+def synthetic_client_id(user: "User", purpose: str) -> str:
+    """A client id for server-generated work that has no device behind it.
+
+    Kept apart from ``generate_client_id`` rather than folded into it: that function
+    sanitizes and truncates *user-supplied* device names to 10 characters, which would
+    silently rewrite a server-controlled label (``annotation-import`` →
+    ``annotation``) and split existing data across two ids. ``purpose`` is a literal in
+    Chronicle's own source, so it needs neither sanitizing nor bounding.
+    """
+    return f"{user_id_prefix(user)}-{purpose}"
+
+
 def generate_client_id(user: "User", device_name: Optional[str] = None) -> str:
     """
     Generate a STABLE client_id in the format: user_id_suffix-device_suffix
 
+    The one constructor for a *device's* identity, on every ingress path. The upload
+    controller used to have its own copy that interpolated the raw device name, so
+    ``MyPhone!`` became ``abc123-MyPhone!`` on upload and ``abc123-myphone`` over the
+    WebSocket — one physical device, two identities, two registry entries.
+
     The client_id is deterministic for a given (user, device_name): the same device
     reconnecting always maps to the same id. This is the device's stable identity —
     a reconnect reuses it (and evict-on-reconnect collapses any lingering connection
@@ -478,8 +516,7 @@ def generate_client_id(user: "User", device_name: Optional[str] = None) -> str:
         client_id as user_id_suffix-device_suffix (or user_id_suffix-<uuid> when no
         device name is supplied).
     """
-    # Use last 6 characters of MongoDB ObjectId as user identifier
-    user_id_suffix = str(user.id)[-6:]
+    user_id_suffix = user_id_prefix(user)
 
     if device_name:
         # Sanitize device name: lowercase, alphanumeric + hyphens only, max 10 chars

backends/advanced/src/advanced_omi_backend/controllers/audio_controller.py

@@ -16,6 +16,7 @@
 from fastapi.responses import JSONResponse
 from rq import Retry
 
+from advanced_omi_backend.client_manager import generate_client_id
 from advanced_omi_backend.controllers.queue_controller import (
     JOB_RESULT_TTL,
     start_post_conversation_jobs,
@@ -38,13 +39,6 @@
 audio_logger = logging.getLogger("audio_processing")
 
 
-def generate_client_id(user: User, device_name: str) -> str:
-    """Generate client ID for uploaded files."""
-    logger.debug(f"Generating client ID - user.id={user.id}, type={type(user.id)}")
-    user_id_suffix = str(user.id)[-6:]
-    return f"{user_id_suffix}-{device_name}"
-
-
 async def upload_and_process_audio_files(
     user: User,
     files: list[UploadFile],

backends/advanced/src/advanced_omi_backend/controllers/client_controller.py

@@ -14,6 +14,7 @@
 )
 from advanced_omi_backend.config import WS_IDLE_TIMEOUT_SECS
 from advanced_omi_backend.users import (
+    RegisteredClient,
     User,
     forget_client_for_user,
     get_user_by_client_id,
@@ -23,29 +24,30 @@
 logger = logging.getLogger(__name__)
 
 
-def _device_view(entry: dict, client_manager: ClientManager, now: float) -> dict:
+def _device_view(
+    entry: RegisteredClient, client_manager: ClientManager, now: float
+) -> dict:
     """Shape one registry device joined with its live connection state.
 
     `connected` and `last_seen` come from the in-memory ClientState when the device is
     live (authoritative), falling back to the persisted registry timestamp when offline.
     """
-    client_id = entry["client_id"]
+    client_id = entry.client_id
     state = client_manager.get_client(client_id)
     if state is not None:
         last_seen = max(0.0, now - state.last_activity)
         connected = last_seen < WS_IDLE_TIMEOUT_SECS
         has_active = bool(state.stream_session_id) or state.batch_started
     else:
-        last = entry.get("last_seen")
-        last_seen = max(0.0, now - last.timestamp()) if last is not None else None
+        last_seen = max(0.0, now - entry.last_seen.timestamp())
         connected = False
         has_active = False
     return {
         "client_id": client_id,
-        "device_name": entry.get("device_name"),
-        "name": entry.get("name") or entry.get("device_name") or client_id,
+        "device_name": entry.device_name,
+        "name": entry.name or entry.device_name or client_id,
         "connected": connected,
-        "last_seen": round(last_seen, 1) if last_seen is not None else None,
+        "last_seen": round(last_seen, 1),
         "has_active_conversation": has_active,
     }
 
@@ -61,7 +63,7 @@ async def list_devices(user: User, client_manager: ClientManager) -> dict:
     devices = []
     for u in users:
         owner_email = u.email
-        for entry in u.registered_clients.values():
+        for entry in u.registered_clients:
             view = _device_view(entry, client_manager, now)
             view["user_email"] = owner_email
             devices.append(view)
@@ -77,7 +79,7 @@ async def rename_device(user: User, client_id: str, name: str):
             status_code=400, content={"error": "name must not be empty"}
         )
 
-    owner = user if client_id in user.registered_clients else None
+    owner = user if user.has_client(client_id) else None
     if owner is None and user.is_superuser:
         owner = await get_user_by_client_id(client_id)
     if owner is None:
@@ -91,7 +93,7 @@ async def rename_device(user: User, client_id: str, name: str):
 async def forget_device(user: User, client_id: str, client_manager: ClientManager):
     """Remove a device from the registry. A currently-connected device is also evicted
     so it doesn't immediately re-appear from its live ClientState."""
-    owner = user if client_id in user.registered_clients else None
+    owner = user if user.has_client(client_id) else None
     if owner is None and user.is_superuser:
         owner = await get_user_by_client_id(client_id)
     if owner is None:

backends/advanced/src/advanced_omi_backend/controllers/conversation_controller.py

@@ -21,7 +21,7 @@
 from advanced_omi_backend.controllers.queue_controller import (
     JOB_RESULT_TTL,
     conversation_edit_chain_in_flight,
-    default_queue,
+    enqueue_summary_job_bundle,
     memory_queue,
     post_conv_enqueue_kwargs,
     start_post_conversation_jobs,
@@ -45,7 +45,6 @@
 )
 from advanced_omi_backend.services.plugin_service import get_plugin_router
 from advanced_omi_backend.users import User
-from advanced_omi_backend.workers.conversation_jobs import generate_title_summary_job
 from advanced_omi_backend.workers.memory_jobs import (
     enqueue_memory_processing,
     process_memory_job,
@@ -979,10 +978,13 @@ def _enqueue_transcript_reprocessing(
     user_id: str,
     source: str,
     job_id_prefix: str,
-    end_reason: str,
+    trigger: str,
 ) -> tuple:
     """Enqueue transcribe job + post-conversation chain.
 
+    ``end_reason`` is deliberately not passed: the conversation already ended for a
+    real reason, and re-transcribing it does not change how the recording ended.
+
     Returns (version_id, transcript_job, post_jobs dict).
     """
     version_id = str(uuid.uuid4())
@@ -1004,7 +1006,7 @@ def _enqueue_transcript_reprocessing(
         user_id=user_id,
         transcript_version_id=version_id,
         depends_on_job=transcript_job,
-        end_reason=end_reason,
+        trigger=trigger,
         memory_cause=MemoryCause.TRANSCRIPT_REPROCESS,
     )
 
@@ -1051,9 +1053,9 @@ def _enqueue_speaker_reprocessing_chain(
     source_version_id: str,
     diarization_source: str | None = None,
 ) -> dict:
-    """Enqueue speaker -> memory -> title_summary chain.
+    """Enqueue speaker -> memory -> ordered summary bundle.
 
-    Returns dict with keys: speaker, memory, title_summary (job IDs).
+    Returns speaker, memory, title, short-summary, and detailed-summary job IDs.
     """
     speaker_job = transcription_queue.enqueue(
         recognise_speakers_job,
@@ -1103,27 +1105,22 @@ def _enqueue_speaker_reprocessing_chain(
         f"Chained memory job {memory_job.id} after speaker job {speaker_job.id}"
     )
 
-    title_summary_job = default_queue.enqueue(
-        generate_title_summary_job,
+    summary_jobs = enqueue_summary_job_bundle(
         conversation_id,
-        job_timeout=300,
-        result_ttl=JOB_RESULT_TTL,
-        job_id=f"title_summary_{conversation_id[:12]}",
-        description=f"Regenerate title/summary for {conversation_id[:8]}",
-        **post_conv_enqueue_kwargs(
-            "title_summary",
-            {"conversation_id": conversation_id},
-            depends_on=memory_job,
-        ),
+        depends_on=memory_job,
+        meta={"conversation_id": conversation_id, "trigger": "speaker_reprocess"},
     )
     logger.info(
-        f"Chained title/summary job {title_summary_job.id} after memory job {memory_job.id}"
+        f"Chained summary bundle {[job.id for job in summary_jobs.values()]} "
+        f"after memory job {memory_job.id}"
     )
 
     return {
         "speaker": speaker_job.id,
         "memory": memory_job.id,
-        "title_summary": title_summary_job.id,
+        "title": summary_jobs["title"].id,
+        "short_summary": summary_jobs["short_summary"].id,
+        "detailed_summary": summary_jobs["detailed_summary"].id,
     }
 
 
@@ -1224,7 +1221,7 @@ async def reprocess_orphan(conversation_id: str, user: User):
             user_id=str(user.user_id),
             source="reprocess_orphan",
             job_id_prefix="orphan_transcribe",
-            end_reason="reprocess_orphan",
+            trigger=Conversation.ProcessingTrigger.REPROCESS_ORPHAN.value,
         )
 
         logger.info(
@@ -1236,7 +1233,9 @@ async def reprocess_orphan(conversation_id: str, user: User):
             content={
                 "message": f"Orphan reprocessing started for conversation {conversation_id}",
                 "job_id": transcript_job.id,
-                "title_summary_job_id": post_jobs.get("title_summary"),
+                "title_job_id": post_jobs.get("title"),
+                "short_summary_job_id": post_jobs.get("short_summary"),
+                "detailed_summary_job_id": post_jobs.get("detailed_summary"),
                 "version_id": version_id,
                 "status": "queued",
             }
@@ -1281,7 +1280,7 @@ async def reprocess_transcript(conversation_id: str, user: User):
             user_id=str(user.user_id),
             source="reprocess",
             job_id_prefix="reprocess",
-            end_reason="reprocess_transcript",
+            trigger=Conversation.ProcessingTrigger.REPROCESS_TRANSCRIPT.value,
         )
 
         logger.info(
@@ -1293,7 +1292,9 @@ async def reprocess_transcript(conversation_id: str, user: User):
             content={
                 "message": f"Transcript reprocessing started for conversation {conversation_id}",
                 "job_id": transcript_job.id,
-                "title_summary_job_id": post_jobs.get("title_summary"),
+                "title_job_id": post_jobs.get("title"),
+                "short_summary_job_id": post_jobs.get("short_summary"),
+                "detailed_summary_job_id": post_jobs.get("detailed_summary"),
                 "version_id": version_id,
                 "status": "queued",
             }
@@ -1519,7 +1520,9 @@ async def reprocess_speakers(
                 "message": "Speaker reprocessing started",
                 "job_id": job_ids["speaker"],
                 "memory_job_id": job_ids["memory"],
-                "title_summary_job_id": job_ids["title_summary"],
+                "title_job_id": job_ids["title"],
+                "short_summary_job_id": job_ids["short_summary"],
+                "detailed_summary_job_id": job_ids["detailed_summary"],
                 "version_id": new_version_id,
                 "source_version_id": source_version_id,
                 "diarization_source": diarization_source,