Merge branch 'main' into fix-daily-ux

When Cal.com events appear in an aggregated ICS feed, the same event
gets two different UIDs (one from Cal.com, one from Google Calendar).
This caused duplicate meetings to be created for the same time slot.

Add time-window dedup check in create_upcoming_meetings_for_event:
after verifying no meeting exists for the calendar_event_id, also check
if a meeting already exists for the same room + start_date + end_date.

server/migrations/versions/3aa20b96d963_drop_use_celery_column.py

@@ -0,0 +1,35 @@
+"""drop_use_celery_column
+
+Revision ID: 3aa20b96d963
+Revises: e69f08ead8ea
+Create Date: 2026-02-05 10:12:44.065279
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "3aa20b96d963"
+down_revision: Union[str, None] = "e69f08ead8ea"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("room", schema=None) as batch_op:
+        batch_op.drop_column("use_celery")
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("room", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "use_celery",
+                sa.Boolean(),
+                server_default=sa.text("false"),
+                nullable=False,
+            )
+        )

server/reflector/db/meetings.py

@@ -346,6 +346,27 @@ class MeetingController:
             return None
         return Meeting(**result)
 
+    async def get_by_room_and_time_window(
+        self, room: Room, start_date: datetime, end_date: datetime
+    ) -> Meeting | None:
+        """Check if a meeting already exists for this room with the same time window."""
+        query = (
+            meetings.select()
+            .where(
+                sa.and_(
+                    meetings.c.room_id == room.id,
+                    meetings.c.start_date == start_date,
+                    meetings.c.end_date == end_date,
+                    meetings.c.is_active,
+                )
+            )
+            .limit(1)
+        )
+        result = await get_database().fetch_one(query)
+        if not result:
+            return None
+        return Meeting(**result)
+
     async def update_meeting(self, meeting_id: str, **kwargs):
         query = meetings.update().where(meetings.c.id == meeting_id).values(**kwargs)
         await get_database().execute(query)

server/reflector/db/rooms.py

@@ -57,12 +57,6 @@ rooms = sqlalchemy.Table(
         sqlalchemy.String,
         nullable=False,
     ),
-    sqlalchemy.Column(
-        "use_celery",
-        sqlalchemy.Boolean,
-        nullable=False,
-        server_default=false(),
-    ),
     sqlalchemy.Column(
         "skip_consent",
         sqlalchemy.Boolean,
@@ -97,7 +91,6 @@ class Room(BaseModel):
     ics_last_sync: datetime | None = None
     ics_last_etag: str | None = None
     platform: Platform = Field(default_factory=lambda: settings.DEFAULT_VIDEO_PLATFORM)
-    use_celery: bool = False
     skip_consent: bool = False
 
 

server/reflector/presence/__init__.py

@@ -1,17 +0,0 @@
-"""Presence tracking for meetings."""
-
-from reflector.presence.pending_joins import (
-    PENDING_JOIN_PREFIX,
-    PENDING_JOIN_TTL,
-    create_pending_join,
-    delete_pending_join,
-    has_pending_joins,
-)
-
-__all__ = [
-    "PENDING_JOIN_PREFIX",
-    "PENDING_JOIN_TTL",
-    "create_pending_join",
-    "delete_pending_join",
-    "has_pending_joins",
-]

server/reflector/presence/pending_joins.py

@@ -1,59 +0,0 @@
-"""Track pending join intents in Redis.
-
-When a user signals intent to join a meeting (before WebRTC handshake completes),
-we store a pending join record. This prevents the meeting from being deactivated
-while users are still connecting.
-"""
-
-import time
-
-from redis.asyncio import Redis
-
-from reflector.logger import logger
-
-PENDING_JOIN_TTL = 30  # seconds
-PENDING_JOIN_PREFIX = "pending_join"
-# Max keys to scan per Redis SCAN iteration
-SCAN_BATCH_SIZE = 100
-
-
-async def create_pending_join(redis: Redis, meeting_id: str, user_id: str) -> None:
-    """Create a pending join record. Called before WebRTC handshake."""
-    key = f"{PENDING_JOIN_PREFIX}:{meeting_id}:{user_id}"
-    log = logger.bind(meeting_id=meeting_id, user_id=user_id, key=key)
-    await redis.setex(key, PENDING_JOIN_TTL, str(time.time()))
-    log.debug("Created pending join")
-
-
-async def delete_pending_join(redis: Redis, meeting_id: str, user_id: str) -> None:
-    """Delete pending join. Called after WebRTC connection established."""
-    key = f"{PENDING_JOIN_PREFIX}:{meeting_id}:{user_id}"
-    log = logger.bind(meeting_id=meeting_id, user_id=user_id, key=key)
-    await redis.delete(key)
-    log.debug("Deleted pending join")
-
-
-async def has_pending_joins(redis: Redis, meeting_id: str) -> bool:
-    """Check if meeting has any pending joins.
-
-    Uses Redis SCAN to iterate through all keys matching the pattern.
-    Properly iterates until cursor returns 0 to ensure all keys are checked.
-    """
-    pattern = f"{PENDING_JOIN_PREFIX}:{meeting_id}:*"
-    log = logger.bind(meeting_id=meeting_id, pattern=pattern)
-
-    cursor = 0
-    iterations = 0
-    while True:
-        cursor, keys = await redis.scan(
-            cursor=cursor, match=pattern, count=SCAN_BATCH_SIZE
-        )
-        iterations += 1
-        if keys:
-            log.debug("Found pending joins", count=len(keys), iterations=iterations)
-            return True
-        if cursor == 0:
-            break
-
-    log.debug("No pending joins found", iterations=iterations)
-    return False

server/reflector/services/transcript_process.py

@@ -15,14 +15,10 @@ from hatchet_sdk.clients.rest.exceptions import ApiException, NotFoundException
 from hatchet_sdk.clients.rest.models import V1TaskStatus
 
 from reflector.db.recordings import recordings_controller
-from reflector.db.rooms import rooms_controller
 from reflector.db.transcripts import Transcript, transcripts_controller
 from reflector.hatchet.client import HatchetClientManager
 from reflector.logger import logger
 from reflector.pipelines.main_file_pipeline import task_pipeline_file_process
-from reflector.pipelines.main_multitrack_pipeline import (
-    task_pipeline_multitrack_process,
-)
 from reflector.utils.string import NonEmptyString
 
 
@@ -181,21 +177,7 @@ async def dispatch_transcript_processing(
     Returns AsyncResult for Celery tasks, None for Hatchet workflows.
     """
     if isinstance(config, MultitrackProcessingConfig):
-        use_celery = False
+        # Multitrack processing always uses Hatchet (no Celery fallback)
-        if config.room_id:
-            room = await rooms_controller.get_by_id(config.room_id)
-            use_celery = room.use_celery if room else False
-
-        use_hatchet = not use_celery
-
-        if use_celery:
-            logger.info(
-                "Room uses legacy Celery processing",
-                room_id=config.room_id,
-                transcript_id=config.transcript_id,
-            )
-
-        if use_hatchet:
         # First check if we can replay (outside transaction since it's read-only)
         transcript = await transcripts_controller.get_by_id(config.transcript_id)
         if transcript and transcript.workflow_run_id and not force:
@@ -203,9 +185,7 @@ async def dispatch_transcript_processing(
                 transcript.workflow_run_id
             )
             if can_replay:
-                    await HatchetClientManager.replay_workflow(
+                await HatchetClientManager.replay_workflow(transcript.workflow_run_id)
-                        transcript.workflow_run_id
-                    )
                 logger.info(
                     "Replaying Hatchet workflow",
                     workflow_id=transcript.workflow_run_id,
@@ -233,9 +213,7 @@ async def dispatch_transcript_processing(
         # Force: cancel old workflow if exists
         if force and transcript and transcript.workflow_run_id:
             try:
-                    await HatchetClientManager.cancel_workflow(
+                await HatchetClientManager.cancel_workflow(transcript.workflow_run_id)
-                        transcript.workflow_run_id
-                    )
                 logger.info(
                     "Cancelled old workflow (--force)",
                     workflow_id=transcript.workflow_run_id,
@@ -245,9 +223,7 @@ async def dispatch_transcript_processing(
                     "Old workflow already deleted (--force)",
                     workflow_id=transcript.workflow_run_id,
                 )
-                await transcripts_controller.update(
+            await transcripts_controller.update(transcript, {"workflow_run_id": None})
-                    transcript, {"workflow_run_id": None}
-                )
 
         # Re-fetch and check for concurrent dispatch (optimistic approach).
         # No database lock - worst case is duplicate dispatch, but Hatchet
@@ -293,12 +269,6 @@ async def dispatch_transcript_processing(
         logger.info("Hatchet workflow dispatched", workflow_id=workflow_id)
         return None
 
-        # Celery pipeline (durable workflows disabled)
-        return task_pipeline_multitrack_process.delay(
-            transcript_id=config.transcript_id,
-            bucket_name=config.bucket_name,
-            track_keys=config.track_keys,
-        )
     elif isinstance(config, FileProcessingConfig):
         return task_pipeline_file_process.delay(transcript_id=config.transcript_id)
     else:

server/reflector/settings.py

@@ -1,7 +1,7 @@
 from pydantic.types import PositiveInt
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-from reflector.schemas.platform import WHEREBY_PLATFORM, Platform
+from reflector.schemas.platform import DAILY_PLATFORM, Platform
 from reflector.utils.string import NonEmptyString
 
 
@@ -155,7 +155,7 @@ class Settings(BaseSettings):
         None  # Webhook UUID for this environment. Not used by production code
     )
     # Platform Configuration
-    DEFAULT_VIDEO_PLATFORM: Platform = WHEREBY_PLATFORM
+    DEFAULT_VIDEO_PLATFORM: Platform = DAILY_PLATFORM
 
     # Zulip integration
     ZULIP_REALM: str | None = None

server/reflector/views/rooms.py

@@ -1,3 +1,4 @@
+import logging
 from datetime import datetime, timedelta, timezone
 from enum import Enum
 from typing import Annotated, Any, Literal, Optional
@@ -13,18 +14,16 @@ from reflector.db import get_database
 from reflector.db.calendar_events import calendar_events_controller
 from reflector.db.meetings import meetings_controller
 from reflector.db.rooms import rooms_controller
-from reflector.logger import logger
+from reflector.redis_cache import RedisAsyncLock
-from reflector.presence.pending_joins import create_pending_join, delete_pending_join
-from reflector.redis_cache import RedisAsyncLock, get_async_redis_client
 from reflector.schemas.platform import Platform
 from reflector.services.ics_sync import ics_sync_service
 from reflector.settings import settings
-from reflector.utils.string import NonEmptyString
 from reflector.utils.url import add_query_param
 from reflector.video_platforms.factory import create_platform_client
-from reflector.worker.process import poll_daily_room_presence_task
 from reflector.worker.webhook import test_webhook
 
+logger = logging.getLogger(__name__)
+
 
 class Room(BaseModel):
     id: str
@@ -598,158 +597,3 @@ async def rooms_join_meeting(
         meeting.room_url = add_query_param(meeting.room_url, "t", token)
 
     return meeting
-
-
-class JoiningRequest(BaseModel):
-    """Request body for /joining endpoint (before WebRTC handshake)."""
-
-    connection_id: NonEmptyString
-    """Unique identifier for this connection. Generated by client via crypto.randomUUID()."""
-
-
-class JoiningResponse(BaseModel):
-    status: Literal["ok"]
-
-
-class JoinedRequest(BaseModel):
-    """Request body for /joined endpoint (after WebRTC connection established)."""
-
-    connection_id: NonEmptyString
-    """Must match the connection_id sent to /joining."""
-
-
-class JoinedResponse(BaseModel):
-    status: Literal["ok"]
-
-
-def _get_pending_join_key(
-    user: Optional[auth.UserInfo], connection_id: NonEmptyString
-) -> str:
-    """Get a unique key for pending join tracking.
-
-    Uses user ID for authenticated users, connection_id for anonymous users.
-    This ensures each browser tab has its own unique pending join record.
-    """
-    if user:
-        return f"{user['sub']}:{connection_id}"
-    return f"anon:{connection_id}"
-
-
-@router.post(
-    "/rooms/{room_name}/meetings/{meeting_id}/joining", response_model=JoiningResponse
-)
-async def meeting_joining(
-    room_name: str,
-    meeting_id: str,
-    body: JoiningRequest,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
-) -> JoiningResponse:
-    """Signal intent to join meeting. Called before WebRTC handshake starts.
-
-    This creates a pending join record that prevents the meeting from being
-    deactivated while the WebRTC handshake is in progress. The record expires
-    automatically after 30 seconds if the connection is not established.
-    """
-    log = logger.bind(
-        room_name=room_name, meeting_id=meeting_id, connection_id=body.connection_id
-    )
-
-    room = await rooms_controller.get_by_name(room_name)
-    if not room:
-        raise HTTPException(status_code=404, detail="Room not found")
-
-    meeting = await meetings_controller.get_by_id(meeting_id, room=room)
-    if not meeting:
-        raise HTTPException(status_code=404, detail="Meeting not found")
-
-    if not meeting.is_active:
-        raise HTTPException(status_code=400, detail="Meeting is not active")
-
-    join_key = _get_pending_join_key(user, body.connection_id)
-
-    redis = await get_async_redis_client()
-    try:
-        await create_pending_join(redis, meeting_id, join_key)
-        log.debug("Created pending join intent", join_key=join_key)
-    finally:
-        await redis.aclose()
-
-    return JoiningResponse(status="ok")
-
-
-@router.post(
-    "/rooms/{room_name}/meetings/{meeting_id}/joined", response_model=JoinedResponse
-)
-async def meeting_joined(
-    room_name: str,
-    meeting_id: str,
-    body: JoinedRequest,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
-) -> JoinedResponse:
-    """Signal that WebRTC connection is established.
-
-    This clears the pending join record, confirming the user has successfully
-    connected to the meeting. Safe to call even if meeting was deactivated
-    during the handshake (idempotent cleanup).
-    """
-    log = logger.bind(
-        room_name=room_name, meeting_id=meeting_id, connection_id=body.connection_id
-    )
-
-    room = await rooms_controller.get_by_name(room_name)
-    if not room:
-        raise HTTPException(status_code=404, detail="Room not found")
-
-    meeting = await meetings_controller.get_by_id(meeting_id, room=room)
-    if not meeting:
-        raise HTTPException(status_code=404, detail="Meeting not found")
-
-    # Note: We don't check is_active here - the /joined call is a cleanup operation
-    # and should succeed even if the meeting was deactivated during the handshake
-
-    join_key = _get_pending_join_key(user, body.connection_id)
-
-    redis = await get_async_redis_client()
-    try:
-        await delete_pending_join(redis, meeting_id, join_key)
-        log.debug("Cleared pending join intent", join_key=join_key)
-    finally:
-        await redis.aclose()
-
-    # Trigger presence poll to detect the new participant faster than periodic poll
-    if meeting.platform == "daily":
-        poll_daily_room_presence_task.delay(meeting_id)
-
-    return JoinedResponse(status="ok")
-
-
-class LeaveResponse(BaseModel):
-    status: Literal["ok"]
-
-
-@router.post(
-    "/rooms/{room_name}/meetings/{meeting_id}/leave", response_model=LeaveResponse
-)
-async def meeting_leave(
-    room_name: str,
-    meeting_id: str,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
-) -> LeaveResponse:
-    """Trigger presence recheck when user leaves meeting.
-
-    Called on tab close/navigation via sendBeacon(). Immediately queues presence
-    poll to detect dirty disconnects faster than 30s periodic poll.
-    Daily.co webhooks handle clean disconnects, but tab close/crash need this.
-    """
-    room = await rooms_controller.get_by_name(room_name)
-    if not room:
-        raise HTTPException(status_code=404, detail="Room not found")
-
-    meeting = await meetings_controller.get_by_id(meeting_id, room=room)
-    if not meeting:
-        raise HTTPException(status_code=404, detail="Meeting not found")
-
-    if meeting.platform == "daily":
-        poll_daily_room_presence_task.delay(meeting_id)
-
-    return LeaveResponse(status="ok")

server/reflector/worker/ics_sync.py

@@ -5,7 +5,7 @@ from celery import shared_task
 from celery.utils.log import get_task_logger
 
 from reflector.asynctask import asynctask
-from reflector.db.calendar_events import calendar_events_controller
+from reflector.db.calendar_events import CalendarEvent, calendar_events_controller
 from reflector.db.meetings import meetings_controller
 from reflector.db.rooms import Room, rooms_controller
 from reflector.redis_cache import RedisAsyncLock
@@ -83,10 +83,9 @@ def _should_sync(room) -> bool:
     return time_since_sync.total_seconds() >= room.ics_fetch_interval
 
 
-MEETING_DEFAULT_DURATION = timedelta(hours=1)
+async def create_upcoming_meetings_for_event(
-
+    event: CalendarEvent, create_window: datetime, room: Room
-
+):
-async def create_upcoming_meetings_for_event(event, create_window, room: Room):
     if event.start_time <= create_window:
         return
     existing_meeting = await meetings_controller.get_by_calendar_event(event.id, room)
@@ -94,6 +93,21 @@ async def create_upcoming_meetings_for_event(event, create_window, room: Room):
     if existing_meeting:
         return
 
+    # Prevent duplicate meetings from aggregated calendar feeds
+    # (e.g. same event appears with different UIDs from Cal.com and Google Calendar)
+    end_date = event.end_time
+    existing_by_time = await meetings_controller.get_by_room_and_time_window(
+        room, event.start_time, end_date
+    )
+    if existing_by_time:
+        logger.info(
+            "Skipping duplicate calendar event - meeting already exists for this time window",
+            room_id=room.id,
+            event_id=event.id,
+            existing_meeting_id=existing_by_time.id,
+        )
+        return
+
     logger.info(
         "Pre-creating meeting for calendar event",
         room_id=room.id,
@@ -102,8 +116,6 @@ async def create_upcoming_meetings_for_event(event, create_window, room: Room):
     )
 
     try:
-        end_date = event.end_time or (event.start_time + MEETING_DEFAULT_DURATION)
-
         client = create_platform_client(room.platform)
 
         meeting_data = await client.create_meeting(

server/reflector/worker/process.py

@@ -27,14 +27,10 @@ from reflector.db.transcripts import (
 from reflector.hatchet.client import HatchetClientManager
 from reflector.pipelines.main_file_pipeline import task_pipeline_file_process
 from reflector.pipelines.main_live_pipeline import asynctask
-from reflector.pipelines.main_multitrack_pipeline import (
-    task_pipeline_multitrack_process,
-)
 from reflector.pipelines.topic_processing import EmptyPipeline
-from reflector.presence.pending_joins import has_pending_joins
 from reflector.processors import AudioFileWriterProcessor
 from reflector.processors.audio_waveform_processor import AudioWaveformProcessor
-from reflector.redis_cache import RedisAsyncLock, get_async_redis_client
+from reflector.redis_cache import RedisAsyncLock
 from reflector.settings import settings
 from reflector.storage import get_transcripts_storage
 from reflector.utils.daily import (
@@ -352,17 +348,7 @@ async def _process_multitrack_recording_inner(
             room_id=room.id,
         )
 
-    use_celery = room and room.use_celery
+    # Multitrack processing always uses Hatchet (no Celery fallback)
-    use_hatchet = not use_celery
-
-    if use_celery:
-        logger.info(
-            "Room uses legacy Celery processing",
-            room_id=room.id,
-            transcript_id=transcript.id,
-        )
-
-    if use_hatchet:
     workflow_id = await HatchetClientManager.start_workflow(
         workflow_name="DiarizationPipeline",
         input_data={
@@ -384,17 +370,7 @@ async def _process_multitrack_recording_inner(
         transcript_id=transcript.id,
     )
 
-        await transcripts_controller.update(
+    await transcripts_controller.update(transcript, {"workflow_run_id": workflow_id})
-            transcript, {"workflow_run_id": workflow_id}
-        )
-        return
-
-    # Celery pipeline (runs when durable workflows disabled)
-    task_pipeline_multitrack_process.delay(
-        transcript_id=transcript.id,
-        bucket_name=bucket_name,
-        track_keys=filter_cam_audio_tracks(track_keys),
-    )
 
 
 @shared_task
@@ -870,18 +846,6 @@ async def process_meetings():
                     logger_.debug("Meeting not yet started, keep it")
 
                 if should_deactivate:
-                    # Check for pending joins before deactivating
-                    # Users might be in the process of connecting via WebRTC
-                    redis = await get_async_redis_client()
-                    try:
-                        if await has_pending_joins(redis, meeting.id):
-                            logger_.info(
-                                "Meeting has pending joins, skipping deactivation"
-                            )
-                            continue
-                    finally:
-                        await redis.aclose()
-
                     await meetings_controller.update_meeting(
                         meeting.id, is_active=False
                     )
@@ -1085,10 +1049,7 @@ async def reprocess_failed_daily_recordings():
                 )
                 continue
 
-            use_celery = room and room.use_celery
+            # Multitrack reprocessing always uses Hatchet (no Celery fallback)
-            use_hatchet = not use_celery
-
-            if use_hatchet:
             if not transcript:
                 logger.warning(
                     "No transcript for Hatchet reprocessing, skipping",
@@ -1125,26 +1086,6 @@ async def reprocess_failed_daily_recordings():
                 room_name=meeting.room_name,
                 track_count=len(recording.track_keys),
             )
-            else:
-                logger.info(
-                    "Queueing Daily recording for Celery reprocessing",
-                    recording_id=recording.id,
-                    room_name=meeting.room_name,
-                    track_count=len(recording.track_keys),
-                    transcript_status=transcript.status if transcript else None,
-                )
-
-                # For reprocessing, pass actual recording time (though it's ignored - see _process_multitrack_recording_inner)
-                # Reprocessing uses recording.meeting_id directly instead of time-based matching
-                recording_start_ts = int(recording.recorded_at.timestamp())
-
-                process_multitrack_recording.delay(
-                    bucket_name=bucket_name,
-                    daily_room_name=meeting.room_name,
-                    recording_id=recording.id,
-                    track_keys=recording.track_keys,
-                    recording_start_ts=recording_start_ts,
-                )
 
             reprocessed_count += 1
 

server/tests/conftest.py

@@ -4,7 +4,7 @@ from unittest.mock import patch
 
 import pytest
 
-from reflector.schemas.platform import WHEREBY_PLATFORM
+from reflector.schemas.platform import DAILY_PLATFORM, WHEREBY_PLATFORM
 
 
 @pytest.fixture(scope="session", autouse=True)
@@ -14,6 +14,7 @@ def register_mock_platform():
     from reflector.video_platforms.registry import register_platform
 
     register_platform(WHEREBY_PLATFORM, MockPlatformClient)
+    register_platform(DAILY_PLATFORM, MockPlatformClient)
     yield
 
 

server/tests/test_ics_dedup.py

@@ -0,0 +1,190 @@
+from datetime import datetime, timedelta, timezone
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from reflector.db import get_database
+from reflector.db.calendar_events import CalendarEvent, calendar_events_controller
+from reflector.db.meetings import meetings
+from reflector.db.rooms import rooms_controller
+from reflector.worker.ics_sync import create_upcoming_meetings_for_event
+
+
+@pytest.mark.asyncio
+async def test_duplicate_calendar_event_does_not_create_duplicate_meeting():
+    """When an aggregated ICS feed contains the same event with different UIDs
+    (e.g. Cal.com UID + Google Calendar UUID), only one meeting should be created."""
+
+    room = await rooms_controller.add(
+        name="dedup-test-room",
+        user_id="test-user",
+        zulip_auto_post=False,
+        zulip_stream="",
+        zulip_topic="",
+        is_locked=False,
+        room_mode="normal",
+        recording_type="cloud",
+        recording_trigger="automatic-2nd-participant",
+        is_shared=False,
+        ics_url="https://calendar.example.com/dedup.ics",
+        ics_enabled=True,
+    )
+
+    now = datetime.now(timezone.utc)
+    start_time = now + timedelta(hours=1)
+    end_time = now + timedelta(hours=2)
+
+    # Create first calendar event (Cal.com UID)
+    event1 = await calendar_events_controller.upsert(
+        CalendarEvent(
+            room_id=room.id,
+            ics_uid="abc123@Cal.com",
+            title="Team Standup",
+            start_time=start_time,
+            end_time=end_time,
+        )
+    )
+
+    # create_window must be before start_time for the function to proceed
+    create_window = now - timedelta(minutes=6)
+
+    # Create meeting for event1
+    with patch(
+        "reflector.worker.ics_sync.create_platform_client"
+    ) as mock_platform_factory:
+        mock_client = AsyncMock()
+
+        async def mock_create_meeting_1(room_name_prefix, *, end_date, room):
+            return AsyncMock(
+                meeting_id="meeting-1",
+                room_name="dedup-test-room-abc",
+                room_url="https://mock.video/dedup-test-room-abc",
+                host_room_url="https://mock.video/dedup-test-room-abc?host=true",
+            )
+
+        mock_client.create_meeting = mock_create_meeting_1
+        mock_client.upload_logo = AsyncMock()
+        mock_platform_factory.return_value = mock_client
+
+        await create_upcoming_meetings_for_event(event1, create_window, room)
+
+    # Verify meeting was created
+    results = await get_database().fetch_all(
+        meetings.select().where(meetings.c.room_id == room.id)
+    )
+    assert len(results) == 1, f"Expected 1 meeting, got {len(results)}"
+
+    # Create second calendar event with different UID but same time window (Google Calendar UUID)
+    event2 = await calendar_events_controller.upsert(
+        CalendarEvent(
+            room_id=room.id,
+            ics_uid="550e8400-e29b-41d4-a716-446655440000",
+            title="Team Standup",
+            start_time=start_time,
+            end_time=end_time,
+        )
+    )
+
+    # Try to create meeting for event2 - should be skipped due to dedup
+    with patch(
+        "reflector.worker.ics_sync.create_platform_client"
+    ) as mock_platform_factory:
+        mock_client = AsyncMock()
+        create_meeting_called = False
+
+        async def mock_create_meeting_2(room_name_prefix, *, end_date, room):
+            nonlocal create_meeting_called
+            create_meeting_called = True
+
+        mock_client.create_meeting = mock_create_meeting_2
+        mock_client.upload_logo = AsyncMock()
+        mock_platform_factory.return_value = mock_client
+
+        await create_upcoming_meetings_for_event(event2, create_window, room)
+
+        # Platform client should NOT have been called for the duplicate
+        assert (
+            not create_meeting_called
+        ), "create_meeting should not be called for duplicate"
+
+    # Verify still only 1 meeting
+    results = await get_database().fetch_all(
+        meetings.select().where(meetings.c.room_id == room.id)
+    )
+    assert len(results) == 1, f"Expected 1 meeting after dedup, got {len(results)}"
+
+
+@pytest.mark.asyncio
+async def test_different_time_windows_create_separate_meetings():
+    """Events at different times should create separate meetings, even if titles match."""
+
+    room = await rooms_controller.add(
+        name="dedup-diff-time-room",
+        user_id="test-user",
+        zulip_auto_post=False,
+        zulip_stream="",
+        zulip_topic="",
+        is_locked=False,
+        room_mode="normal",
+        recording_type="cloud",
+        recording_trigger="automatic-2nd-participant",
+        is_shared=False,
+        ics_url="https://calendar.example.com/dedup2.ics",
+        ics_enabled=True,
+    )
+
+    now = datetime.now(timezone.utc)
+    create_window = now - timedelta(minutes=6)
+
+    # Event 1: 1-2pm
+    event1 = await calendar_events_controller.upsert(
+        CalendarEvent(
+            room_id=room.id,
+            ics_uid="event-morning@Cal.com",
+            title="Team Standup",
+            start_time=now + timedelta(hours=1),
+            end_time=now + timedelta(hours=2),
+        )
+    )
+
+    # Event 2: 3-4pm (different time)
+    event2 = await calendar_events_controller.upsert(
+        CalendarEvent(
+            room_id=room.id,
+            ics_uid="event-afternoon@Cal.com",
+            title="Team Standup",
+            start_time=now + timedelta(hours=3),
+            end_time=now + timedelta(hours=4),
+        )
+    )
+
+    with patch(
+        "reflector.worker.ics_sync.create_platform_client"
+    ) as mock_platform_factory:
+        mock_client = AsyncMock()
+
+        call_count = 0
+
+        async def mock_create_meeting(room_name_prefix, *, end_date, room):
+            nonlocal call_count
+            call_count += 1
+            return AsyncMock(
+                meeting_id=f"meeting-{call_count}",
+                room_name=f"dedup-diff-time-room-{call_count}",
+                room_url=f"https://mock.video/dedup-diff-time-room-{call_count}",
+                host_room_url=f"https://mock.video/dedup-diff-time-room-{call_count}?host=true",
+            )
+
+        mock_client.create_meeting = mock_create_meeting
+        mock_client.upload_logo = AsyncMock()
+        mock_platform_factory.return_value = mock_client
+
+        await create_upcoming_meetings_for_event(event1, create_window, room)
+        await create_upcoming_meetings_for_event(event2, create_window, room)
+
+    results = await get_database().fetch_all(
+        meetings.select().where(meetings.c.room_id == room.id)
+    )
+    assert (
+        len(results) == 2
+    ), f"Expected 2 meetings for different times, got {len(results)}"

server/tests/test_joining_endpoint.py

@@ -1,367 +0,0 @@
-"""Integration tests for /joining and /joined endpoints.
-
-Tests for the join intent tracking to prevent race conditions during
-WebRTC handshake when users join meetings.
-"""
-
-from datetime import datetime, timedelta, timezone
-from unittest.mock import AsyncMock, patch
-
-import pytest
-
-from reflector.db.meetings import Meeting
-from reflector.presence.pending_joins import PENDING_JOIN_PREFIX
-
-TEST_CONNECTION_ID = "test-connection-uuid-12345"
-
-
-@pytest.fixture
-def mock_room():
-    """Mock room object."""
-    from reflector.db.rooms import Room
-
-    return Room(
-        id="room-123",
-        name="test-room",
-        user_id="owner-user",
-        created_at=datetime.now(timezone.utc),
-        zulip_auto_post=False,
-        zulip_stream="",
-        zulip_topic="",
-        is_locked=False,
-        room_mode="normal",
-        recording_type="cloud",
-        recording_trigger="automatic",
-        is_shared=True,
-        platform="daily",
-        skip_consent=False,
-    )
-
-
-@pytest.fixture
-def mock_meeting():
-    """Mock meeting object."""
-    now = datetime.now(timezone.utc)
-    return Meeting(
-        id="meeting-456",
-        room_id="room-123",
-        room_name="test-room-20251118120000",
-        room_url="https://daily.co/test-room-20251118120000",
-        host_room_url="https://daily.co/test-room-20251118120000?t=host",
-        platform="daily",
-        num_clients=0,
-        is_active=True,
-        start_date=now,
-        end_date=now + timedelta(hours=1),
-    )
-
-
-@pytest.mark.asyncio
-@patch("reflector.views.rooms.rooms_controller.get_by_name")
-@patch("reflector.views.rooms.meetings_controller.get_by_id")
-@patch("reflector.views.rooms.get_async_redis_client")
-async def test_joining_endpoint_creates_pending_join(
-    mock_get_redis,
-    mock_get_meeting,
-    mock_get_room,
-    mock_room,
-    mock_meeting,
-    client,
-    authenticated_client,
-):
-    """Test that /joining endpoint creates pending join in Redis."""
-    mock_get_room.return_value = mock_room
-    mock_get_meeting.return_value = mock_meeting
-
-    mock_redis = AsyncMock()
-    mock_redis.setex = AsyncMock()
-    mock_redis.aclose = AsyncMock()
-    mock_get_redis.return_value = mock_redis
-
-    response = await client.post(
-        f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
-        json={"connection_id": TEST_CONNECTION_ID},
-    )
-
-    assert response.status_code == 200
-    assert response.json() == {"status": "ok"}
-
-    # Verify Redis setex was called with correct key pattern
-    mock_redis.setex.assert_called_once()
-    call_args = mock_redis.setex.call_args[0]
-    assert call_args[0].startswith(f"{PENDING_JOIN_PREFIX}:{mock_meeting.id}:")
-    assert TEST_CONNECTION_ID in call_args[0]
-
-
-@pytest.mark.asyncio
-@patch("reflector.views.rooms.poll_daily_room_presence_task")
-@patch("reflector.views.rooms.rooms_controller.get_by_name")
-@patch("reflector.views.rooms.meetings_controller.get_by_id")
-@patch("reflector.views.rooms.get_async_redis_client")
-async def test_joined_endpoint_deletes_pending_join(
-    mock_get_redis,
-    mock_get_meeting,
-    mock_get_room,
-    mock_poll_task,
-    mock_room,
-    mock_meeting,
-    client,
-    authenticated_client,
-):
-    """Test that /joined endpoint deletes pending join from Redis."""
-    mock_get_room.return_value = mock_room
-    mock_get_meeting.return_value = mock_meeting
-
-    mock_redis = AsyncMock()
-    mock_redis.delete = AsyncMock()
-    mock_redis.aclose = AsyncMock()
-    mock_get_redis.return_value = mock_redis
-
-    response = await client.post(
-        f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joined",
-        json={"connection_id": TEST_CONNECTION_ID},
-    )
-
-    assert response.status_code == 200
-    assert response.json() == {"status": "ok"}
-
-    # Verify Redis delete was called with correct key pattern
-    mock_redis.delete.assert_called_once()
-    call_args = mock_redis.delete.call_args[0]
-    assert call_args[0].startswith(f"{PENDING_JOIN_PREFIX}:{mock_meeting.id}:")
-    assert TEST_CONNECTION_ID in call_args[0]
-
-    # Verify presence poll was triggered for Daily meetings
-    mock_poll_task.delay.assert_called_once_with(mock_meeting.id)
-
-
-@pytest.mark.asyncio
-@patch("reflector.views.rooms.rooms_controller.get_by_name")
-async def test_joining_endpoint_room_not_found(
-    mock_get_room,
-    client,
-    authenticated_client,
-):
-    """Test that /joining returns 404 when room not found."""
-    mock_get_room.return_value = None
-
-    response = await client.post(
-        "/rooms/nonexistent-room/meetings/meeting-123/joining",
-        json={"connection_id": TEST_CONNECTION_ID},
-    )
-
-    assert response.status_code == 404
-    assert response.json()["detail"] == "Room not found"
-
-
-@pytest.mark.asyncio
-@patch("reflector.views.rooms.rooms_controller.get_by_name")
-@patch("reflector.views.rooms.meetings_controller.get_by_id")
-async def test_joining_endpoint_meeting_not_found(
-    mock_get_meeting,
-    mock_get_room,
-    mock_room,
-    client,
-    authenticated_client,
-):
-    """Test that /joining returns 404 when meeting not found."""
-    mock_get_room.return_value = mock_room
-    mock_get_meeting.return_value = None
-
-    response = await client.post(
-        f"/rooms/{mock_room.name}/meetings/nonexistent-meeting/joining",
-        json={"connection_id": TEST_CONNECTION_ID},
-    )
-
-    assert response.status_code == 404
-    assert response.json()["detail"] == "Meeting not found"
-
-
-@pytest.mark.asyncio
-@patch("reflector.views.rooms.rooms_controller.get_by_name")
-@patch("reflector.views.rooms.meetings_controller.get_by_id")
-async def test_joining_endpoint_meeting_not_active(
-    mock_get_meeting,
-    mock_get_room,
-    mock_room,
-    mock_meeting,
-    client,
-    authenticated_client,
-):
-    """Test that /joining returns 400 when meeting is not active."""
-    mock_get_room.return_value = mock_room
-    inactive_meeting = mock_meeting.model_copy(update={"is_active": False})
-    mock_get_meeting.return_value = inactive_meeting
-
-    response = await client.post(
-        f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
-        json={"connection_id": TEST_CONNECTION_ID},
-    )
-
-    assert response.status_code == 400
-    assert response.json()["detail"] == "Meeting is not active"
-
-
-@pytest.mark.asyncio
-@patch("reflector.views.rooms.rooms_controller.get_by_name")
-@patch("reflector.views.rooms.meetings_controller.get_by_id")
-@patch("reflector.views.rooms.get_async_redis_client")
-async def test_joining_endpoint_anonymous_user(
-    mock_get_redis,
-    mock_get_meeting,
-    mock_get_room,
-    mock_room,
-    mock_meeting,
-    client,
-):
-    """Test that /joining works for anonymous users with unique connection_id."""
-    mock_get_room.return_value = mock_room
-    mock_get_meeting.return_value = mock_meeting
-
-    mock_redis = AsyncMock()
-    mock_redis.setex = AsyncMock()
-    mock_redis.aclose = AsyncMock()
-    mock_get_redis.return_value = mock_redis
-
-    response = await client.post(
-        f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
-        json={"connection_id": TEST_CONNECTION_ID},
-    )
-
-    assert response.status_code == 200
-    assert response.json() == {"status": "ok"}
-
-    # Verify Redis setex was called with "anon:" prefix and connection_id
-    call_args = mock_redis.setex.call_args[0]
-    assert ":anon:" in call_args[0]
-    assert TEST_CONNECTION_ID in call_args[0]
-
-
-@pytest.mark.asyncio
-@patch("reflector.views.rooms.rooms_controller.get_by_name")
-@patch("reflector.views.rooms.meetings_controller.get_by_id")
-@patch("reflector.views.rooms.get_async_redis_client")
-async def test_joining_endpoint_redis_closed_on_success(
-    mock_get_redis,
-    mock_get_meeting,
-    mock_get_room,
-    mock_room,
-    mock_meeting,
-    client,
-    authenticated_client,
-):
-    """Test that Redis connection is closed after successful operation."""
-    mock_get_room.return_value = mock_room
-    mock_get_meeting.return_value = mock_meeting
-
-    mock_redis = AsyncMock()
-    mock_redis.setex = AsyncMock()
-    mock_redis.aclose = AsyncMock()
-    mock_get_redis.return_value = mock_redis
-
-    await client.post(
-        f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
-        json={"connection_id": TEST_CONNECTION_ID},
-    )
-
-    mock_redis.aclose.assert_called_once()
-
-
-@pytest.mark.asyncio
-@patch("reflector.views.rooms.rooms_controller.get_by_name")
-@patch("reflector.views.rooms.meetings_controller.get_by_id")
-@patch("reflector.views.rooms.get_async_redis_client")
-async def test_joining_endpoint_redis_closed_on_error(
-    mock_get_redis,
-    mock_get_meeting,
-    mock_get_room,
-    mock_room,
-    mock_meeting,
-    client,
-    authenticated_client,
-):
-    """Test that Redis connection is closed even when operation fails."""
-    mock_get_room.return_value = mock_room
-    mock_get_meeting.return_value = mock_meeting
-
-    mock_redis = AsyncMock()
-    mock_redis.setex = AsyncMock(side_effect=Exception("Redis error"))
-    mock_redis.aclose = AsyncMock()
-    mock_get_redis.return_value = mock_redis
-
-    with pytest.raises(Exception):
-        await client.post(
-            f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
-            json={"connection_id": TEST_CONNECTION_ID},
-        )
-
-    mock_redis.aclose.assert_called_once()
-
-
-@pytest.mark.asyncio
-async def test_joining_endpoint_requires_connection_id(
-    client,
-):
-    """Test that /joining returns 422 when connection_id is missing."""
-    response = await client.post(
-        "/rooms/test-room/meetings/meeting-123/joining",
-        json={},
-    )
-
-    assert response.status_code == 422  # Validation error
-
-
-@pytest.mark.asyncio
-async def test_joining_endpoint_rejects_empty_connection_id(
-    client,
-):
-    """Test that /joining returns 422 when connection_id is empty string."""
-    response = await client.post(
-        "/rooms/test-room/meetings/meeting-123/joining",
-        json={"connection_id": ""},
-    )
-
-    assert response.status_code == 422  # Validation error (NonEmptyString)
-
-
-@pytest.mark.asyncio
-@patch("reflector.views.rooms.rooms_controller.get_by_name")
-@patch("reflector.views.rooms.meetings_controller.get_by_id")
-@patch("reflector.views.rooms.get_async_redis_client")
-async def test_different_connection_ids_create_different_keys(
-    mock_get_redis,
-    mock_get_meeting,
-    mock_get_room,
-    mock_room,
-    mock_meeting,
-    client,
-):
-    """Test that different connection_ids create different Redis keys."""
-    mock_get_room.return_value = mock_room
-    mock_get_meeting.return_value = mock_meeting
-
-    mock_redis = AsyncMock()
-    mock_redis.setex = AsyncMock()
-    mock_redis.aclose = AsyncMock()
-    mock_get_redis.return_value = mock_redis
-
-    # First connection
-    await client.post(
-        f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
-        json={"connection_id": "connection-1"},
-    )
-    key1 = mock_redis.setex.call_args[0][0]
-
-    mock_redis.setex.reset_mock()
-
-    # Second connection (different tab)
-    await client.post(
-        f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
-        json={"connection_id": "connection-2"},
-    )
-    key2 = mock_redis.setex.call_args[0][0]
-
-    # Keys should be different
-    assert key1 != key2
-    assert "connection-1" in key1
-    assert "connection-2" in key2

server/tests/test_pending_joins.py

@@ -1,153 +0,0 @@
-"""Tests for pending joins Redis helper functions.
-
-TDD tests for tracking join intent to prevent race conditions during
-WebRTC handshake when users join meetings.
-"""
-
-from unittest.mock import AsyncMock
-
-import pytest
-
-from reflector.presence.pending_joins import (
-    PENDING_JOIN_PREFIX,
-    PENDING_JOIN_TTL,
-    create_pending_join,
-    delete_pending_join,
-    has_pending_joins,
-)
-
-
-@pytest.fixture
-def mock_redis():
-    """Mock async Redis client."""
-    redis = AsyncMock()
-    redis.setex = AsyncMock()
-    redis.delete = AsyncMock()
-    redis.scan = AsyncMock(return_value=(0, []))
-    return redis
-
-
-@pytest.mark.asyncio
-async def test_create_pending_join_sets_key_with_ttl(mock_redis):
-    """Test that create_pending_join stores key with correct TTL."""
-    meeting_id = "meeting-123"
-    user_id = "user-456"
-
-    await create_pending_join(mock_redis, meeting_id, user_id)
-
-    expected_key = f"{PENDING_JOIN_PREFIX}:{meeting_id}:{user_id}"
-    mock_redis.setex.assert_called_once()
-    call_args = mock_redis.setex.call_args
-    assert call_args[0][0] == expected_key
-    assert call_args[0][1] == PENDING_JOIN_TTL
-    # Value should be a timestamp string
-    assert call_args[0][2] is not None
-
-
-@pytest.mark.asyncio
-async def test_delete_pending_join_removes_key(mock_redis):
-    """Test that delete_pending_join removes the key."""
-    meeting_id = "meeting-123"
-    user_id = "user-456"
-
-    await delete_pending_join(mock_redis, meeting_id, user_id)
-
-    expected_key = f"{PENDING_JOIN_PREFIX}:{meeting_id}:{user_id}"
-    mock_redis.delete.assert_called_once_with(expected_key)
-
-
-@pytest.mark.asyncio
-async def test_has_pending_joins_returns_false_when_no_keys(mock_redis):
-    """Test has_pending_joins returns False when no matching keys."""
-    mock_redis.scan.return_value = (0, [])
-
-    result = await has_pending_joins(mock_redis, "meeting-123")
-
-    assert result is False
-    mock_redis.scan.assert_called_once()
-    call_kwargs = mock_redis.scan.call_args.kwargs
-    assert call_kwargs["match"] == f"{PENDING_JOIN_PREFIX}:meeting-123:*"
-
-
-@pytest.mark.asyncio
-async def test_has_pending_joins_returns_true_when_keys_exist(mock_redis):
-    """Test has_pending_joins returns True when matching keys found."""
-    mock_redis.scan.return_value = (0, [b"pending_join:meeting-123:user-1"])
-
-    result = await has_pending_joins(mock_redis, "meeting-123")
-
-    assert result is True
-
-
-@pytest.mark.asyncio
-async def test_has_pending_joins_scans_with_correct_pattern(mock_redis):
-    """Test has_pending_joins uses correct scan pattern."""
-    meeting_id = "meeting-abc-def"
-    mock_redis.scan.return_value = (0, [])
-
-    await has_pending_joins(mock_redis, meeting_id)
-
-    expected_pattern = f"{PENDING_JOIN_PREFIX}:{meeting_id}:*"
-    mock_redis.scan.assert_called_once()
-    call_kwargs = mock_redis.scan.call_args.kwargs
-    assert call_kwargs["match"] == expected_pattern
-    assert call_kwargs["count"] == 100
-
-
-@pytest.mark.asyncio
-async def test_multiple_users_pending_joins(mock_redis):
-    """Test that multiple users can have pending joins for same meeting."""
-    meeting_id = "meeting-123"
-    # Simulate two pending joins
-    mock_redis.scan.return_value = (
-        0,
-        [b"pending_join:meeting-123:user-1", b"pending_join:meeting-123:user-2"],
-    )
-
-    result = await has_pending_joins(mock_redis, meeting_id)
-
-    assert result is True
-
-
-@pytest.mark.asyncio
-async def test_pending_join_ttl_value():
-    """Test that PENDING_JOIN_TTL has expected value."""
-    # 30 seconds should be enough for WebRTC handshake but not too long
-    assert PENDING_JOIN_TTL == 30
-
-
-@pytest.mark.asyncio
-async def test_pending_join_prefix_value():
-    """Test that PENDING_JOIN_PREFIX has expected value."""
-    assert PENDING_JOIN_PREFIX == "pending_join"
-
-
-@pytest.mark.asyncio
-async def test_has_pending_joins_multi_iteration_scan_no_keys(mock_redis):
-    """Test has_pending_joins iterates until cursor returns 0."""
-    # Simulate multi-iteration scan: cursor 100 -> cursor 50 -> cursor 0
-    mock_redis.scan.side_effect = [
-        (100, []),  # First iteration, no keys, continue
-        (50, []),  # Second iteration, no keys, continue
-        (0, []),  # Third iteration, cursor 0, done
-    ]
-
-    result = await has_pending_joins(mock_redis, "meeting-123")
-
-    assert result is False
-    assert mock_redis.scan.call_count == 3
-
-
-@pytest.mark.asyncio
-async def test_has_pending_joins_multi_iteration_finds_key_later(mock_redis):
-    """Test has_pending_joins finds key on second iteration."""
-    # Simulate finding key on second scan iteration
-    mock_redis.scan.side_effect = [
-        (100, []),  # First iteration, no keys
-        (0, [b"pending_join:meeting-123:user-1"]),  # Second iteration, found key
-    ]
-
-    result = await has_pending_joins(mock_redis, "meeting-123")
-
-    assert result is True
-    assert mock_redis.scan.call_count == 2

server/tests/test_process_meetings_pending_joins.py

@@ -1,241 +0,0 @@
-"""Tests for process_meetings pending joins check.
-
-Tests that process_meetings correctly skips deactivation when
-pending joins exist for a meeting.
-"""
-
-from datetime import datetime, timedelta, timezone
-from unittest.mock import AsyncMock, patch
-
-import pytest
-
-from reflector.db.meetings import Meeting
-
-
-def _get_process_meetings_fn():
-    """Get the underlying async function without Celery/asynctask decorators."""
-    from reflector.worker import process
-
-    fn = process.process_meetings
-    # Get through both decorator layers (@shared_task and @asynctask)
-    if hasattr(fn, "__wrapped__"):
-        fn = fn.__wrapped__
-    if hasattr(fn, "__wrapped__"):
-        fn = fn.__wrapped__
-    return fn
-
-
-@pytest.fixture
-def mock_active_meeting():
-    """Mock an active meeting that should be considered for deactivation."""
-    now = datetime.now(timezone.utc)
-    return Meeting(
-        id="meeting-123",
-        room_id="room-456",
-        room_name="test-room-20251118120000",
-        room_url="https://daily.co/test-room-20251118120000",
-        host_room_url="https://daily.co/test-room-20251118120000?t=host",
-        platform="daily",
-        num_clients=0,
-        is_active=True,
-        start_date=now - timedelta(hours=1),
-        end_date=now - timedelta(minutes=30),  # Already ended
-    )
-
-
-@pytest.mark.asyncio
-@patch("reflector.worker.process.meetings_controller.get_all_active")
-@patch("reflector.worker.process.RedisAsyncLock")
-@patch("reflector.worker.process.create_platform_client")
-@patch("reflector.worker.process.get_async_redis_client")
-@patch("reflector.worker.process.has_pending_joins")
-@patch("reflector.worker.process.meetings_controller.update_meeting")
-async def test_process_meetings_skips_deactivation_with_pending_joins(
-    mock_update_meeting,
-    mock_has_pending_joins,
-    mock_get_redis,
-    mock_create_client,
-    mock_redis_lock_class,
-    mock_get_all_active,
-    mock_active_meeting,
-):
-    """Test that process_meetings skips deactivation when pending joins exist."""
-    process_meetings = _get_process_meetings_fn()
-
-    mock_get_all_active.return_value = [mock_active_meeting]
-
-    # Mock lock acquired
-    mock_lock_instance = AsyncMock()
-    mock_lock_instance.acquired = True
-    mock_lock_instance.__aenter__ = AsyncMock(return_value=mock_lock_instance)
-    mock_lock_instance.__aexit__ = AsyncMock()
-    mock_redis_lock_class.return_value = mock_lock_instance
-
-    # Mock platform client - no active sessions, but had sessions (triggers deactivation)
-    mock_daily_client = AsyncMock()
-    mock_session = AsyncMock()
-    mock_session.ended_at = datetime.now(timezone.utc)  # Session ended
-    mock_daily_client.get_room_sessions = AsyncMock(return_value=[mock_session])
-    mock_create_client.return_value = mock_daily_client
-
-    # Mock Redis client
-    mock_redis = AsyncMock()
-    mock_redis.aclose = AsyncMock()
-    mock_get_redis.return_value = mock_redis
-
-    # Mock pending joins exist
-    mock_has_pending_joins.return_value = True
-
-    await process_meetings()
-
-    # Verify has_pending_joins was called
-    mock_has_pending_joins.assert_called_once_with(mock_redis, mock_active_meeting.id)
-
-    # Verify meeting was NOT deactivated
-    mock_update_meeting.assert_not_called()
-
-    # Verify Redis was closed
-    mock_redis.aclose.assert_called_once()
-
-
-@pytest.mark.asyncio
-@patch("reflector.worker.process.meetings_controller.get_all_active")
-@patch("reflector.worker.process.RedisAsyncLock")
-@patch("reflector.worker.process.create_platform_client")
-@patch("reflector.worker.process.get_async_redis_client")
-@patch("reflector.worker.process.has_pending_joins")
-@patch("reflector.worker.process.meetings_controller.update_meeting")
-async def test_process_meetings_deactivates_without_pending_joins(
-    mock_update_meeting,
-    mock_has_pending_joins,
-    mock_get_redis,
-    mock_create_client,
-    mock_redis_lock_class,
-    mock_get_all_active,
-    mock_active_meeting,
-):
-    """Test that process_meetings deactivates when no pending joins."""
-    process_meetings = _get_process_meetings_fn()
-
-    mock_get_all_active.return_value = [mock_active_meeting]
-
-    # Mock lock acquired
-    mock_lock_instance = AsyncMock()
-    mock_lock_instance.acquired = True
-    mock_lock_instance.__aenter__ = AsyncMock(return_value=mock_lock_instance)
-    mock_lock_instance.__aexit__ = AsyncMock()
-    mock_redis_lock_class.return_value = mock_lock_instance
-
-    # Mock platform client - no active sessions, but had sessions
-    mock_daily_client = AsyncMock()
-    mock_session = AsyncMock()
-    mock_session.ended_at = datetime.now(timezone.utc)
-    mock_daily_client.get_room_sessions = AsyncMock(return_value=[mock_session])
-    mock_create_client.return_value = mock_daily_client
-
-    # Mock Redis client
-    mock_redis = AsyncMock()
-    mock_redis.aclose = AsyncMock()
-    mock_get_redis.return_value = mock_redis
-
-    # Mock no pending joins
-    mock_has_pending_joins.return_value = False
-
-    await process_meetings()
-
-    # Verify meeting was deactivated
-    mock_update_meeting.assert_called_once_with(mock_active_meeting.id, is_active=False)
-
-
-@pytest.mark.asyncio
-@patch("reflector.worker.process.meetings_controller.get_all_active")
-@patch("reflector.worker.process.RedisAsyncLock")
-@patch("reflector.worker.process.create_platform_client")
-async def test_process_meetings_no_check_when_active_sessions(
-    mock_create_client,
-    mock_redis_lock_class,
-    mock_get_all_active,
-    mock_active_meeting,
-):
-    """Test that pending joins check is skipped when there are active sessions."""
-    process_meetings = _get_process_meetings_fn()
-
-    mock_get_all_active.return_value = [mock_active_meeting]
-
-    # Mock lock acquired
-    mock_lock_instance = AsyncMock()
-    mock_lock_instance.acquired = True
-    mock_lock_instance.__aenter__ = AsyncMock(return_value=mock_lock_instance)
-    mock_lock_instance.__aexit__ = AsyncMock()
-    mock_redis_lock_class.return_value = mock_lock_instance
-
-    # Mock platform client - has active session
-    mock_daily_client = AsyncMock()
-    mock_session = AsyncMock()
-    mock_session.ended_at = None  # Still active
-    mock_daily_client.get_room_sessions = AsyncMock(return_value=[mock_session])
-    mock_create_client.return_value = mock_daily_client
-
-    with (
-        patch("reflector.worker.process.get_async_redis_client") as mock_get_redis,
-        patch("reflector.worker.process.has_pending_joins") as mock_has_pending_joins,
-        patch(
-            "reflector.worker.process.meetings_controller.update_meeting"
-        ) as mock_update_meeting,
-    ):
-        await process_meetings()
-
-        # Verify pending joins check was NOT called (no need - active sessions exist)
-        mock_has_pending_joins.assert_not_called()
-
-        # Verify meeting was NOT deactivated
-        mock_update_meeting.assert_not_called()
-
-
-@pytest.mark.asyncio
-@patch("reflector.worker.process.meetings_controller.get_all_active")
-@patch("reflector.worker.process.RedisAsyncLock")
-@patch("reflector.worker.process.create_platform_client")
-@patch("reflector.worker.process.get_async_redis_client")
-@patch("reflector.worker.process.has_pending_joins")
-@patch("reflector.worker.process.meetings_controller.update_meeting")
-async def test_process_meetings_closes_redis_even_on_continue(
-    mock_update_meeting,
-    mock_has_pending_joins,
-    mock_get_redis,
-    mock_create_client,
-    mock_redis_lock_class,
-    mock_get_all_active,
-    mock_active_meeting,
-):
-    """Test that Redis connection is always closed, even when skipping deactivation."""
-    process_meetings = _get_process_meetings_fn()
-
-    mock_get_all_active.return_value = [mock_active_meeting]
-
-    # Mock lock acquired
-    mock_lock_instance = AsyncMock()
-    mock_lock_instance.acquired = True
-    mock_lock_instance.__aenter__ = AsyncMock(return_value=mock_lock_instance)
-    mock_lock_instance.__aexit__ = AsyncMock()
-    mock_redis_lock_class.return_value = mock_lock_instance
-
-    # Mock platform client - no active sessions
-    mock_daily_client = AsyncMock()
-    mock_session = AsyncMock()
-    mock_session.ended_at = datetime.now(timezone.utc)
-    mock_daily_client.get_room_sessions = AsyncMock(return_value=[mock_session])
-    mock_create_client.return_value = mock_daily_client
-
-    # Mock Redis client
-    mock_redis = AsyncMock()
-    mock_redis.aclose = AsyncMock()
-    mock_get_redis.return_value = mock_redis
-
-    # Mock pending joins exist (will trigger continue)
-    mock_has_pending_joins.return_value = True
-
-    await process_meetings()
-
-    # Verify Redis was closed
-    mock_redis.aclose.assert_called_once()

server/tests/test_transcripts_process.py

@@ -1,6 +1,6 @@
 import asyncio
 import time
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
 from httpx import ASGITransport, AsyncClient
@@ -142,17 +142,17 @@ async def test_whereby_recording_uses_file_pipeline(client):
             "reflector.services.transcript_process.task_pipeline_file_process"
         ) as mock_file_pipeline,
         patch(
-            "reflector.services.transcript_process.task_pipeline_multitrack_process"
+            "reflector.services.transcript_process.HatchetClientManager"
-        ) as mock_multitrack_pipeline,
+        ) as mock_hatchet,
     ):
         response = await client.post(f"/transcripts/{transcript.id}/process")
 
         assert response.status_code == 200
         assert response.json()["status"] == "ok"
 
-        # Whereby recordings should use file pipeline
+        # Whereby recordings should use file pipeline, not Hatchet
         mock_file_pipeline.delay.assert_called_once_with(transcript_id=transcript.id)
-        mock_multitrack_pipeline.delay.assert_not_called()
+        mock_hatchet.start_workflow.assert_not_called()
 
 
 @pytest.mark.usefixtures("setup_database")
@@ -177,8 +177,6 @@ async def test_dailyco_recording_uses_multitrack_pipeline(client):
         recording_trigger="automatic-2nd-participant",
         is_shared=False,
     )
-    # Force Celery backend for test
-    await rooms_controller.update(room, {"use_celery": True})
 
     transcript = await transcripts_controller.add(
         "",
@@ -213,18 +211,23 @@ async def test_dailyco_recording_uses_multitrack_pipeline(client):
             "reflector.services.transcript_process.task_pipeline_file_process"
         ) as mock_file_pipeline,
         patch(
-            "reflector.services.transcript_process.task_pipeline_multitrack_process"
+            "reflector.services.transcript_process.HatchetClientManager"
-        ) as mock_multitrack_pipeline,
+        ) as mock_hatchet,
     ):
+        mock_hatchet.start_workflow = AsyncMock(return_value="test-workflow-id")
+
         response = await client.post(f"/transcripts/{transcript.id}/process")
 
         assert response.status_code == 200
         assert response.json()["status"] == "ok"
 
-        # Daily.co multitrack recordings should use multitrack pipeline
+        # Daily.co multitrack recordings should use Hatchet workflow
-        mock_multitrack_pipeline.delay.assert_called_once_with(
+        mock_hatchet.start_workflow.assert_called_once()
-            transcript_id=transcript.id,
+        call_kwargs = mock_hatchet.start_workflow.call_args.kwargs
-            bucket_name="daily-bucket",
+        assert call_kwargs["workflow_name"] == "DiarizationPipeline"
-            track_keys=track_keys,
+        assert call_kwargs["input_data"]["transcript_id"] == transcript.id
-        )
+        assert call_kwargs["input_data"]["bucket_name"] == "daily-bucket"
+        assert call_kwargs["input_data"]["tracks"] == [
+            {"s3_key": k} for k in track_keys
+        ]
         mock_file_pipeline.delay.assert_not_called()

www/app/[roomName]/components/DailyRoom.tsx

@@ -8,7 +8,7 @@ import {
   useRef,
   useState,
 } from "react";
-import { Box, Spinner, Center, Text } from "@chakra-ui/react";
+import { Box, Spinner, Center, Text, Button, VStack } from "@chakra-ui/react";
 import { useRouter, useParams } from "next/navigation";
 import DailyIframe, {
   DailyCall,
@@ -16,17 +16,18 @@ import DailyIframe, {
   DailyCustomTrayButton,
   DailyCustomTrayButtons,
   DailyEventObjectCustomButtonClick,
+  DailyEventObjectFatalError,
+  DailyFatalErrorType,
   DailyFactoryOptions,
   DailyParticipantsObject,
 } from "@daily-co/daily-js";
 import type { components } from "../../reflector-api";
+import { printApiError, ApiError } from "../../api/_error";
 import { useAuth } from "../../lib/AuthProvider";
 import { useConsentDialog } from "../../lib/consent";
 import {
   useRoomJoinMeeting,
   useMeetingStartRecording,
-  useMeetingJoining,
-  useMeetingJoined,
 } from "../../lib/apiHooks";
 import { omit } from "remeda";
 import {
@@ -47,6 +48,63 @@ const RAW_TRACKS_NAMESPACE = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
 const RECORDING_START_DELAY_MS = 2000;
 const RECORDING_START_MAX_RETRIES = 5;
 
+const FATAL_ERROR_MESSAGES: Partial<
+  Record<DailyFatalErrorType, { message: string; rejoinable?: boolean }>
+> = {
+  "connection-error": {
+    message: "Connection lost. Please check your network.",
+    rejoinable: true,
+  },
+  "exp-room": { message: "The meeting time has ended." },
+  "exp-token": { message: "Your session has expired.", rejoinable: true },
+  ejected: { message: "You were removed from this meeting." },
+  "meeting-full": { message: "This meeting is full." },
+  "not-allowed": { message: "You are not allowed to join this meeting." },
+  "nbf-room": { message: "This meeting hasn't started yet." },
+  "nbf-token": { message: "This meeting hasn't started yet." },
+  "no-room": { message: "This room does not exist." },
+  "end-of-life": { message: "This meeting room is no longer available." },
+};
+
+function FatalErrorScreen({
+  error,
+  roomName,
+}: {
+  error: FatalError;
+  roomName: string;
+}) {
+  const router = useRouter();
+  const info =
+    error.type !== "unknown" ? FATAL_ERROR_MESSAGES[error.type] : undefined;
+  const message = info?.message ?? `Something went wrong: ${error.message}`;
+  const rejoinable = info?.rejoinable ?? false;
+
+  return (
+    <Center width="100vw" height="100vh">
+      <VStack gap={4}>
+        <Text color="red.500">{message}</Text>
+        {rejoinable ? (
+          <>
+            <Button onClick={() => window.location.reload()}>
+              Try Rejoining
+            </Button>
+            <Button
+              variant="outline"
+              onClick={() => router.push(`/${roomName}`)}
+            >
+              Leave
+            </Button>
+          </>
+        ) : (
+          <Button onClick={() => router.push(`/${roomName}`)}>
+            Back to Room
+          </Button>
+        )}
+      </VStack>
+    </Center>
+  );
+}
+
 type Meeting = components["schemas"]["Meeting"];
 type Room = components["schemas"]["RoomDetails"];
 
@@ -84,6 +142,8 @@ const USE_FRAME_INIT_STATE = {
   joined: false as boolean,
 } as const;
 
+type FatalError = { type: DailyFatalErrorType | "unknown"; message: string };
+
 // Daily js and not Daily react used right now because daily-js allows for prebuild interface vs. -react is customizable but has no nice defaults
 const useFrame = (
   container: HTMLDivElement | null,
@@ -91,6 +151,7 @@ const useFrame = (
     onLeftMeeting: () => void;
     onCustomButtonClick: (ev: DailyEventObjectCustomButtonClick) => void;
     onJoinMeeting: () => void;
+    onError: (ev: DailyEventObjectFatalError) => void;
   },
 ) => {
   const [{ frame, joined }, setState] = useState(USE_FRAME_INIT_STATE);
@@ -136,6 +197,7 @@ const useFrame = (
     if (!frame) return;
     frame.on("left-meeting", cbs.onLeftMeeting);
     frame.on("custom-button-click", cbs.onCustomButtonClick);
+    frame.on("error", cbs.onError);
     const joinCb = () => {
       if (!frame) {
         console.error("frame is null in joined-meeting callback");
@@ -147,6 +209,7 @@ const useFrame = (
     return () => {
       frame.off("left-meeting", cbs.onLeftMeeting);
       frame.off("custom-button-click", cbs.onCustomButtonClick);
+      frame.off("error", cbs.onError);
       frame.off("joined-meeting", joinCb);
     };
   }, [frame, cbs]);
@@ -189,13 +252,8 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
   const [container, setContainer] = useState<HTMLDivElement | null>(null);
   const joinMutation = useRoomJoinMeeting();
   const startRecordingMutation = useMeetingStartRecording();
-  const joiningMutation = useMeetingJoining();
-  const joinedMutation = useMeetingJoined();
   const [joinedMeeting, setJoinedMeeting] = useState<Meeting | null>(null);
-
+  const [fatalError, setFatalError] = useState<FatalError | null>(null);
-  // Generate a stable connection ID for this component instance
-  // Used to track pending joins per browser tab (prevents key collision for anonymous users)
-  const connectionId = useMemo(() => crypto.randomUUID(), []);
 
   // Generate deterministic instanceIds so all participants use SAME IDs
   const cloudInstanceId = parseNonEmptyString(meeting.id);
@@ -242,22 +300,18 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
   const roomUrl = joinedMeeting?.room_url;
 
   const handleLeave = useCallback(() => {
+    // If a fatal error occurred, don't redirect — let the error UI show
+    if (fatalError) return;
     router.push("/browse");
-  }, [router]);
+  }, [router, fatalError]);
 
-  // Trigger presence recheck on dirty disconnects (tab close, navigation away)
+  const handleError = useCallback((ev: DailyEventObjectFatalError) => {
-  useEffect(() => {
+    const error: FatalError = {
-    if (!meeting?.id || !roomName) return;
+      type: ev.error?.type ?? "unknown",
-
+      message: ev.errorMsg,
-    const handleBeforeUnload = () => {
-      // sendBeacon guarantees delivery even if tab closes mid-request
-      const url = `/v1/rooms/${roomName}/meetings/${meeting.id}/leave`;
-      navigator.sendBeacon(url, JSON.stringify({}));
     };
-
+    setFatalError(error);
-    window.addEventListener("beforeunload", handleBeforeUnload);
+  }, []);
-    return () => window.removeEventListener("beforeunload", handleBeforeUnload);
-  }, [meeting?.id, roomName]);
 
   const handleCustomButtonClick = useCallback(
     (ev: DailyEventObjectCustomButtonClick) => {
@@ -271,28 +325,6 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
   );
 
   const handleFrameJoinMeeting = useCallback(() => {
-    // Signal that WebRTC connection is established
-    // This clears the pending join intent, confirming successful connection
-    joinedMutation.mutate(
-      {
-        params: {
-          path: {
-            room_name: roomName,
-            meeting_id: meeting.id,
-          },
-        },
-        body: {
-          connection_id: connectionId,
-        },
-      },
-      {
-        onError: (error: unknown) => {
-          // Non-blocking: log but don't fail - this is cleanup, not critical
-          console.warn("Failed to signal joined:", error);
-        },
-      },
-    );
-
     if (meeting.recording_type === "cloud") {
       console.log("Starting dual recording via REST API", {
         cloudInstanceId,
@@ -354,9 +386,6 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
   }, [
     meeting.recording_type,
     meeting.id,
-    roomName,
-    connectionId,
-    joinedMutation,
     startRecordingMutation,
     cloudInstanceId,
     rawTracksInstanceId,
@@ -371,32 +400,13 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
     onLeftMeeting: handleLeave,
     onCustomButtonClick: handleCustomButtonClick,
     onJoinMeeting: handleFrameJoinMeeting,
+    onError: handleError,
   });
 
   useEffect(() => {
     if (!frame || !roomUrl) return;
-
+    frame
-    const joinRoom = async () => {
+      .join({
-      // Signal intent to join before WebRTC handshake starts
-      // This prevents race condition where meeting is deactivated during handshake
-      try {
-        await joiningMutation.mutateAsync({
-          params: {
-            path: {
-              room_name: roomName,
-              meeting_id: meeting.id,
-            },
-          },
-          body: {
-            connection_id: connectionId,
-          },
-        });
-      } catch (error) {
-        // Non-blocking: log but continue with join
-        console.warn("Failed to signal joining intent:", error);
-      }
-
-      await frame.join({
         url: roomUrl,
         sendSettings: {
           video: {
@@ -408,13 +418,9 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
           },
           // Note: screenVideo intentionally not configured to preserve full quality for screen shares
         },
-      });
+      })
-    };
+      .catch(console.error.bind(console, "Failed to join daily room:"));
-
+  }, [frame, roomUrl]);
-    joinRoom().catch(console.error.bind(console, "Failed to join daily room:"));
-    // joiningMutation excluded from deps - it's a stable hook reference
-    // eslint-disable-next-line react-hooks/exhaustive-deps
-  }, [frame, roomUrl, roomName, meeting.id, connectionId]);
 
   useEffect(() => {
     setCustomTrayButton(
@@ -451,13 +457,27 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
   }
 
   if (joinMutation.isError) {
+    const apiDetail = printApiError(
+      joinMutation.error as /*ref 095959E6-01CC-4CF0-B3A9-F65F12F046D3*/ ApiError,
+    );
     return (
       <Center width="100vw" height="100vh">
-        <Text color="red.500">Failed to join meeting. Please try again.</Text>
+        <VStack gap={4}>
+          <Text color="red.500">
+            {apiDetail ?? "Failed to join meeting. Please try again."}
+          </Text>
+          <Button onClick={() => router.push(`/${roomName}`)}>
+            Back to Room
+          </Button>
+        </VStack>
       </Center>
     );
   }
 
+  if (fatalError) {
+    return <FatalErrorScreen error={fatalError} roomName={roomName} />;
+  }
+
   if (!roomUrl) {
     return null;
   }

www/app/lib/apiHooks.ts

@@ -9,6 +9,7 @@ import { MeetingId } from "./types";
 import { NonEmptyString } from "./utils";
 
 /*
+ * ref 095959E6-01CC-4CF0-B3A9-F65F12F046D3
  * XXX error types returned from the hooks are not always correct; declared types are ValidationError but real type could be string or any other
  * this is either a limitation or incorrect usage of Python json schema generator
  * or, limitation or incorrect usage of .d type generator from json schema
@@ -807,35 +808,6 @@ export function useRoomJoinMeeting() {
   );
 }
 
-// Presence race fix endpoints (not yet in OpenAPI spec)
-// These signal join intent to prevent race conditions during WebRTC handshake
-// eslint-disable-next-line @typescript-eslint/no-explicit-any
-export function useMeetingJoining(): any {
-  return ($api as any).useMutation(
-    "post",
-    "/v1/rooms/{room_name}/meetings/{meeting_id}/joining",
-    {},
-  );
-}
-
-// eslint-disable-next-line @typescript-eslint/no-explicit-any
-export function useMeetingJoined(): any {
-  return ($api as any).useMutation(
-    "post",
-    "/v1/rooms/{room_name}/meetings/{meeting_id}/joined",
-    {},
-  );
-}
-
-// eslint-disable-next-line @typescript-eslint/no-explicit-any
-export function useMeetingLeave(): any {
-  return ($api as any).useMutation(
-    "post",
-    "/v1/rooms/{room_name}/meetings/{meeting_id}/leave",
-    {},
-  );
-}
-
 export function useRoomIcsSync() {
   const { setError } = useError();