feat(local-session): LOCAL session type, audio ingest task and session finalization

bots/management/commands/clean_up_bots_with_heartbeat_timeout_or_that_never_launched.py

@@ -8,7 +8,7 @@
 from django.utils import timezone
 from kubernetes import client, config
 
-from bots.models import Bot, BotEventManager, BotEventSubTypes, BotEventTypes
+from bots.models import Bot, BotEventManager, BotEventSubTypes, BotEventTypes, SessionTypes
 
 logger = logging.getLogger(__name__)
 
@@ -143,7 +143,9 @@ def terminate_bots_that_never_launched(self):
             # - created between 7 days and 1 hour ago AND join_at is null OR join_at is between 7 days and 1 hour ago
             # - first heartbeat is null (never launched)
             never_launched_q_filter = models.Q(created_at__gt=seven_days_ago, created_at__lt=one_hour_ago, first_heartbeat_timestamp__isnull=True, join_at__isnull=True) | models.Q(join_at__gt=seven_days_ago, join_at__lt=one_hour_ago, first_heartbeat_timestamp__isnull=True)
-            problem_bots = Bot.objects.filter(~BotEventManager.get_post_meeting_states_q_filter() & never_launched_q_filter)
+            # Local recordings launch no pod by design, so they never send a heartbeat and
+            # would otherwise all be reaped as "never launched" an hour after they start.
+            problem_bots = Bot.objects.filter(~BotEventManager.get_post_meeting_states_q_filter() & never_launched_q_filter).exclude(session_type=SessionTypes.LOCAL)
 
             logger.info(f"Found {problem_bots.count()} bots that never launched")
 

bots/migrations/0088_alter_bot_session_type.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.1.14 on 2026-07-16 09:10
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('bots', '0087_bot_meeting_dedup_key_and_more'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='bot',
+            name='session_type',
+            field=models.IntegerField(choices=[(1, 'Bot'), (2, 'App Session'), (3, 'Local Recording')], db_default=1, default=1),
+        ),
+    ]

bots/migrations/0089_alter_botevent_event_type.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.1.14 on 2026-07-19 11:18
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('bots', '0088_alter_bot_session_type'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='botevent',
+            name='event_type',
+            field=models.IntegerField(choices=[(1, 'Bot Put in Waiting Room'), (2, 'Bot Joined Meeting'), (3, 'Bot Recording Permission Granted'), (4, 'Meeting Ended'), (5, 'Bot Left Meeting'), (6, 'Bot requested to join meeting'), (7, 'Bot Encountered Fatal error'), (8, 'Bot requested to leave meeting'), (9, 'Bot could not join meeting'), (10, 'Post Processing Completed'), (11, 'Data Deleted'), (12, 'Bot staged'), (13, 'Recording Paused'), (14, 'Recording Resumed'), (15, 'Bot joined breakout room'), (16, 'Bot left breakout room'), (17, 'Bot began joining breakout room'), (18, 'Bot began leaving breakout room'), (19, 'Bot recording permission denied'), (100, 'App Session Connection Requested'), (101, 'App Session Connected'), (102, 'App Session Disconnect Requested'), (103, 'App Session Disconnected'), (104, 'Local Session Ended')]),
+        ),
+    ]

bots/models.py

@@ -686,6 +686,7 @@ class RecordingViews(models.TextChoices):
 class SessionTypes(models.IntegerChoices):
     BOT = 1, "Bot"
     APP_SESSION = 2, "App Session"
+    LOCAL = 3, "Local Recording"  # desktop local recording (mic + system audio uploaded as chunks)
 
 
 class TranscriptionSettings:
@@ -1191,6 +1192,8 @@ def last_bot_event(self):
     def object_id_prefix(self):
         if self.session_type == SessionTypes.BOT:
             return "bot_"
+        elif self.session_type == SessionTypes.LOCAL:
+            return "local_"
         else:
             return "app_"
 
@@ -1355,6 +1358,9 @@ class BotEventTypes(models.IntegerChoices):
     APP_SESSION_DISCONNECT_REQUESTED = 102, "App Session Disconnect Requested"
     APP_SESSION_DISCONNECTED = 103, "App Session Disconnected"
 
+    # Local recording events
+    LOCAL_SESSION_ENDED = 104, "Local Session Ended"
+
     @classmethod
     def type_to_api_code(cls, value):
         """Returns the API code for a given type value"""
@@ -1382,6 +1388,7 @@ def type_to_api_code(cls, value):
             cls.APP_SESSION_CONNECTED: "app_session_connected",
             cls.APP_SESSION_DISCONNECT_REQUESTED: "app_session_disconnect_requested",
             cls.APP_SESSION_DISCONNECTED: "app_session_disconnected",
+            cls.LOCAL_SESSION_ENDED: "local_session_ended",
         }
         return mapping.get(value)
 
@@ -1719,6 +1726,13 @@ class BotEventManager:
             "from": BotStates.DISCONNECTING,
             "to": BotStates.POST_PROCESSING,
         },
+        # Local recording: no pod/meeting, so stop ends the session directly. The recording
+        # must already be COMPLETE before this fires (finalize does that), otherwise the
+        # post-meeting transition would terminate the still-in-progress recording as FAILED.
+        BotEventTypes.LOCAL_SESSION_ENDED: {
+            "from": BotStates.READY,
+            "to": BotStates.ENDED,
+        },
     }
 
     @classmethod
@@ -2029,19 +2043,22 @@ def create_event(
                         metadata=event_metadata,
                     )
 
-                    # Trigger webhook for this event
-                    trigger_webhook(
-                        webhook_trigger_type=WebhookTriggerTypes.BOT_STATE_CHANGE,
-                        bot=bot,
-                        payload={
-                            "event_type": BotEventTypes.type_to_api_code(event_type),
-                            "event_sub_type": BotEventSubTypes.sub_type_to_api_code(event_sub_type),
-                            "event_metadata": event_metadata,
-                            "old_state": BotStates.state_to_api_code(old_state),
-                            "new_state": BotStates.state_to_api_code(bot.state),
-                            "created_at": event.created_at.isoformat(),
-                        },
-                    )
+                    # Trigger webhook for this event. Local recordings have no external
+                    # subscribers and would otherwise deliver "bot" state-change webhooks to
+                    # meeting-bot customers sharing the project, so they are skipped.
+                    if bot.session_type != SessionTypes.LOCAL:
+                        trigger_webhook(
+                            webhook_trigger_type=WebhookTriggerTypes.BOT_STATE_CHANGE,
+                            bot=bot,
+                            payload={
+                                "event_type": BotEventTypes.type_to_api_code(event_type),
+                                "event_sub_type": BotEventSubTypes.sub_type_to_api_code(event_sub_type),
+                                "event_metadata": event_metadata,
+                                "old_state": BotStates.state_to_api_code(old_state),
+                                "new_state": BotStates.state_to_api_code(bot.state),
+                                "created_at": event.created_at.isoformat(),
+                            },
+                        )
 
                     # If we are configured to log bot state changes, log it
                     if settings.LOG_BOT_STATE_CHANGES:

bots/tasks/deliver_webhook_task.py

@@ -84,6 +84,9 @@ def deliver_webhook(self, delivery_id):
         elif delivery.bot.session_type == SessionTypes.APP_SESSION:
             related_object_specific_webhook_data["app_session_id"] = delivery.bot.object_id
             related_object_specific_webhook_data["app_session_metadata"] = delivery.bot.metadata
+        elif delivery.bot.session_type == SessionTypes.LOCAL:
+            related_object_specific_webhook_data["local_session_id"] = delivery.bot.object_id
+            related_object_specific_webhook_data["local_session_metadata"] = delivery.bot.metadata
     elif delivery.calendar:
         related_object_specific_webhook_data["calendar_id"] = delivery.calendar.object_id
         related_object_specific_webhook_data["calendar_deduplication_key"] = delivery.calendar.deduplication_key

bots/tasks/process_local_audio_segment_task.py

@@ -0,0 +1,153 @@
+"""Celery tasks that segment a local recording's uploaded audio and queue transcription.
+
+The desktop uploads short PCM segments over HTTP; a sentence can straddle two of them, so
+audio still mid-utterance when a segment ends is kept as a Redis "tail" (see
+``bots.local_session_store``) and glued to the front of the next batch before the VAD (see
+``bots.local_audio_processing``) sees it. Ordering, exclusivity and a self-contained
+timeline are what keep that correct:
+
+* **Order/dedupe** -- segments drain from a per-source FIFO gated on a monotonic ``sequence``.
+* **Exclusivity** -- a per-source lock means exactly one drain owns the tail at a time.
+* **Timeline** -- times derive from ``offset_ms`` (ms since the session started), never the
+  desktop's wall clock, so clock skew cannot corrupt or drop audio.
+"""
+
+import logging
+
+from celery import shared_task
+
+from bots import local_session_store as store
+from bots.local_audio_processing import (
+    BYTES_PER_SAMPLE,
+    build_manager,
+    duration_ms,
+    feed,
+    flush_remaining,
+    offset_of_buffered,
+)
+from bots.models import Bot, BotEventManager, BotEventTypes, BotStates, Participant, Recording, RecordingManager
+
+logger = logging.getLogger(__name__)
+
+
+@shared_task(bind=True, soft_time_limit=300, autoretry_for=(Exception,), retry_backoff=True, max_retries=3)
+def drain_local_session_audio(self, bot_id, source, is_final=False):
+    client = store.redis_client()
+    lock_key = store.lock_key(bot_id, source)
+    # Exactly one drain may own the tail. A drain that loses the race simply exits: the
+    # holder will pick up whatever it queued, and stop re-queues a final drain anyway.
+    if not client.set(lock_key, "1", nx=True, ex=store.LOCK_TTL_SECONDS):
+        logger.info(f"Local session {bot_id}/{source}: drain already running, skipping")
+        return
+    try:
+        _drain(client, bot_id, source, is_final)
+    finally:
+        client.delete(lock_key)
+
+    # Anything that arrived while we held the lock still needs draining.
+    if client.llen(store.queue_key(bot_id, source)):
+        drain_local_session_audio.delay(bot_id, source)
+
+
+@shared_task(bind=True, soft_time_limit=300, autoretry_for=(Exception,), retry_backoff=True, max_retries=5)
+def finalize_local_session(self, bot_id):
+    """Close a stopped session: flush every source's last utterance, then complete once.
+
+    Running as a single task (rather than a final drain per source) is what makes marking
+    the recording complete safe -- every source's last utterance already exists by the time
+    we get there, so completing the recording can't race an as-yet-uncreated utterance.
+    """
+    client = store.redis_client()
+    for source in store.LOCAL_SESSION_SOURCES:
+        lock_key = store.lock_key(bot_id, source)
+        if not client.set(lock_key, "1", nx=True, ex=store.LOCK_TTL_SECONDS):
+            # A normal drain is still mid-flight for this source; come back once it releases.
+            raise self.retry(countdown=1)
+        try:
+            _drain(client, bot_id, source, is_final=True)
+        finally:
+            client.delete(lock_key)
+
+    bot = Bot.objects.get(id=bot_id)
+    recording = Recording.objects.filter(bot=bot, is_default_recording=True).first()
+    if recording is not None:
+        RecordingManager.set_recording_complete(recording)
+
+    # End the session -- but only from READY, and only after the recording is COMPLETE.
+    # The state guard makes a retried/duplicate stop a no-op instead of an invalid-transition
+    # error; completing the recording first stops the post-meeting transition from marking a
+    # local (file-less) recording FAILED.
+    bot.refresh_from_db()
+    if bot.state == BotStates.READY:
+        BotEventManager.create_event(bot=bot, event_type=BotEventTypes.LOCAL_SESSION_ENDED)
+
+
+def _drain(client, bot_id, source, is_final):
+    try:
+        bot = Bot.objects.get(id=bot_id)
+    except Bot.DoesNotExist:
+        # Session was deleted while audio was still queued -- drop it quietly, don't retry.
+        store.clear_session_state(bot_id)
+        return
+    recording = Recording.objects.filter(bot=bot, is_default_recording=True).first()
+    participant = Participant.objects.filter(bot=bot, uuid=source).first()
+    if recording is None or participant is None:
+        # Data was deleted (delete_data removes the recording contents + participants) -- the
+        # session is gone; clear any leftover Redis state and stop.
+        logger.info(f"Local session {bot_id}/{source}: recording/participant gone, dropping audio")
+        store.clear_session_state(bot_id)
+        return
+
+    tail = store.load_tail(client, bot_id, source)
+    segments = store.pop_segments(
+        client,
+        bot_id,
+        source,
+        tail["last_sequence"],
+        on_replay=lambda seq: logger.info(f"Local session {bot_id}/{source}: dropping replayed segment {seq}"),
+    )
+    if not segments and not (is_final and tail["audio"]):
+        return
+
+    audio = bytearray(tail["audio"])
+    start_offset_ms = tail["started_offset_ms"]
+    end_offset_ms = tail["end_offset_ms"]
+    sample_rate = segments[0]["sample_rate"] if segments else tail["sample_rate"]
+    last_sequence = tail["last_sequence"]
+
+    for segment in segments:
+        # A gap between what we hold and where this segment starts is real elapsed silence.
+        # Feeding it as zeros lets the VAD see the pause instead of splicing speech together.
+        if end_offset_ms is not None and segment["offset_ms"] > end_offset_ms:
+            gap_ms = segment["offset_ms"] - end_offset_ms
+            audio.extend(b"\x00" * int(gap_ms * sample_rate / 1000) * BYTES_PER_SAMPLE)
+        if start_offset_ms is None:
+            start_offset_ms = segment["offset_ms"]
+        audio.extend(segment["audio"])
+        end_offset_ms = segment["offset_ms"] + duration_ms(segment["audio"], sample_rate)
+        last_sequence = segment["sequence"]
+
+    epoch = bot.created_at.replace(tzinfo=None)
+    manager = build_manager(recording, participant, sample_rate)
+    consumed = feed(manager, source, bytes(audio), epoch, start_offset_ms or 0, sample_rate)
+
+    if is_final:
+        # Emit the last unfinished utterance for THIS source only. The recording is marked
+        # complete by finalize_local_session, after EVERY source has flushed -- doing it here
+        # would let one source close the recording before the other's last utterance exists.
+        flush_remaining(manager, source, epoch, end_offset_ms or 0)
+        store.save_tail(client, bot_id, source, b"", None, None, last_sequence, sample_rate)
+        return
+
+    remaining = manager.utterances.get(source, b"")
+    unconsumed = bytes(audio[consumed:])  # a partial frame we could not hand to the VAD yet
+    store.save_tail(
+        client,
+        bot_id,
+        source,
+        bytes(remaining) + unconsumed,
+        offset_of_buffered(manager, source, epoch),
+        end_offset_ms,
+        last_sequence,
+        sample_rate,
+    )