feat: Brady bunch sync

.gitleaksignore

@@ -4,3 +4,4 @@ docs/docs/installation/daily-setup.md:curl-auth-header:277
 gpu/self_hosted/DEV_SETUP.md:curl-auth-header:74
 gpu/self_hosted/DEV_SETUP.md:curl-auth-header:83
 server/reflector/worker/process.py:generic-api-key:465
+server/tests/test_recording_request_flow.py:generic-api-key:121

server/docs/DAILY_REFLECTOR_DATA_MODEL.md

@@ -86,7 +86,7 @@ Daily.co Room: "daily-private-igor-20260110042117"
 | **Purpose** | Tracks active session state | Links recordings, transcripts, participants |
 | **Scope** | Per room instance | Per Reflector room + timestamp |
 
-**Critical Limitation:** Daily.co's recordings API often does NOT return `mtgSessionId`, requiring time-based matching (see [Time-Based Matching](#time-based-matching)).
+**Critical Limitation:** Daily.co's recordings API often does NOT return `mtgSessionId` (can be null), requiring time-based matching (see [Time-Based Matching](#time-based-matching)).
 
 ### Recording
 
@@ -101,6 +101,30 @@ Daily.co Room: "daily-private-igor-20260110042117"
 
 **Critical Behavior:** Recording **stops/restarts** create **separate recording objects** with unique IDs.
 
+### instanceId (Reflector-Generated)
+
+**Definition:** UUID we generate and send when starting recording via REST API.
+
+**Generation:** Deterministic from meeting_id
+- Cloud: `instanceId = meeting_id` directly
+- Raw-tracks: `instanceId = UUIDv5(meeting_id, namespace)`
+
+**Key behaviors:**
+- ✅ **Reuse allowed:** Same instanceId can be used after stop (validated 2026-01-20)
+- ❌ **Not returned:** Daily.co does NOT echo instanceId back in GET /recordings response
+- ✅ **Present in error webhooks:** `recording.error` webhook includes instanceId
+- **Purpose:** Allows multiple concurrent recordings (cloud + raw-tracks) in same room
+
+**Stop/restart example:**
+```
+Recording 1: POST /start with instanceId="779e6376..." → recording_id="ee00c4e8..."
+Stop recording
+Recording 2: POST /start with instanceId="779e6376..." (SAME) → recording_id="b702f509..." (DIFFERENT)
+✅ Both succeed, different recording_ids returned
+```
+
+**Implication:** Cannot match recordings by instanceId (not in response) - must use recording_id.
+
 ---
 
 ## Entity Relationships
@@ -196,6 +220,19 @@ Daily.co Room: "daily-private-igor-20260110042117"
 
 `mtgSessionId` identifies a **Daily.co meeting session** (not individual participants, not a room).
 
+**Reliability:** Can be null or present in GET /recordings response (unreliable).
+
+**When present:** Multiple recordings from same session (stop/restart with participants connected) share same mtgSessionId.
+
+**Example (validated 2026-01-20):**
+```json
+Recording 1: {"id": "ee00c4e8...", "mtgSessionId": "92c4136a-a8da-41c5-9c45-e9a2baae6bd6"}
+Recording 2: {"id": "b702f509...", "mtgSessionId": "92c4136a-a8da-41c5-9c45-e9a2baae6bd6"}
+// Same mtgSessionId (stop/restart in same session)
+```
+
+**When null:** Common - Daily.co API does not reliably populate this field.
+
 ### session_id (Per-Participant)
 
 **Different concept:** Per-participant connection identifier from webhooks.
@@ -220,16 +257,24 @@ TABLE daily_participant_session (
 
 Daily.co's recordings API does not reliably return `mtgSessionId`, making it impossible to directly link recordings to meetings via Daily.co's identifiers.
 
-**Example API response:**
+**Example API response (mtgSessionId can be null OR present):**
 ```json
 {
   "id": "recording-uuid",
   "room_name": "daily-private-igor-20260110042117",
   "start_ts": 1768018896,
-  "mtgSessionId": null  ← Missing!
+  "mtgSessionId": null  // ← Often null (unreliable)
+}
+
+// OR (when present):
+{
+  "id": "recording-uuid",
+  "mtgSessionId": "92c4136a-a8da-41c5-9c45-e9a2baae6bd6"  // ← Sometimes present
 }
 ```
 
+**Key insight:** Cannot rely on mtgSessionId for matching (unreliable). instanceId also not returned. Only reliable identifier is recording.id.
+
 ### Solution: Time-Based Matching
 
 **Implementation:** `reflector/db/meetings.py:get_by_room_name_and_time()`
@@ -491,6 +536,10 @@ UI: User sees 3 separate transcripts
 
 
 ---
-**Document Version:** 1.0
-**Last Verified:** 2026-01-15
-**Data Source:** Production database + Daily.co API inspection
+**Document Version:** 1.1
+**Last Updated:** 2026-01-20
+**Data Source:** Production database + Daily.co API inspection + empirical testing
+**Changes in 1.1:**
+- Added instanceId behavior documentation (reuse allowed, not returned in API)
+- Clarified mtgSessionId reliability (can be null or present)
+- Added empirical validation of stop/restart behavior

server/migrations/versions/f5b008fa8a14_add_daily_recording_requests.py

@@ -0,0 +1,67 @@
+"""add_daily_recording_requests
+
+Revision ID: f5b008fa8a14
+Revises: 1b1e6a6fc465
+Create Date: 2026-01-20 22:32:06.697144
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "f5b008fa8a14"
+down_revision: Union[str, None] = "1b1e6a6fc465"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "daily_recording_request",
+        sa.Column("recording_id", sa.String(), nullable=False),
+        sa.Column("meeting_id", sa.String(), nullable=False),
+        sa.Column("instance_id", sa.String(), nullable=False),
+        sa.Column("type", sa.String(), nullable=False),
+        sa.Column("requested_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["meeting_id"], ["meeting.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("recording_id"),
+    )
+    op.create_index("idx_meeting_id", "daily_recording_request", ["meeting_id"])
+    op.create_index("idx_instance_id", "daily_recording_request", ["instance_id"])
+
+    # Clean up orphaned recordings before adding FK constraint
+    op.execute("""
+        UPDATE recording SET status = 'orphan', meeting_id = NULL
+        WHERE meeting_id IS NOT NULL
+        AND meeting_id NOT IN (SELECT id FROM meeting)
+    """)
+
+    # Add FK constraint to recording table (cascade delete recordings when meeting deleted)
+    op.execute("""
+        ALTER TABLE recording ADD CONSTRAINT fk_recording_meeting
+          FOREIGN KEY (meeting_id) REFERENCES meeting(id) ON DELETE CASCADE
+    """)
+
+    # Add CHECK constraints to enforce orphan invariants
+    op.execute("""
+        ALTER TABLE recording ADD CONSTRAINT chk_orphan_no_meeting
+          CHECK (status != 'orphan' OR meeting_id IS NULL)
+    """)
+    op.execute("""
+        ALTER TABLE recording ADD CONSTRAINT chk_non_orphan_has_meeting
+          CHECK (status = 'orphan' OR meeting_id IS NOT NULL)
+    """)
+
+
+def downgrade() -> None:
+    op.execute("ALTER TABLE recording DROP CONSTRAINT IF EXISTS chk_orphan_no_meeting")
+    op.execute(
+        "ALTER TABLE recording DROP CONSTRAINT IF EXISTS chk_non_orphan_has_meeting"
+    )
+    op.execute("ALTER TABLE recording DROP CONSTRAINT IF EXISTS fk_recording_meeting")
+    op.drop_index("idx_instance_id", table_name="daily_recording_request")
+    op.drop_index("idx_meeting_id", table_name="daily_recording_request")
+    op.drop_table("daily_recording_request")

server/reflector/dailyco_api/recording_orphans.py

@@ -0,0 +1,56 @@
+"""Utility for creating orphan recordings."""
+
+import os
+from datetime import datetime, timezone
+
+from reflector.db.recordings import Recording, recordings_controller
+from reflector.logger import logger
+from reflector.utils.string import NonEmptyString
+
+
+async def create_and_log_orphan(
+    recording_id: NonEmptyString,
+    bucket_name: str,
+    room_name: str,
+    start_ts: int,
+    track_keys: list[str] | None,
+    source: str,
+) -> bool:
+    """Create orphan recording and log if first occurrence.
+
+    Args:
+        recording_id: Daily.co recording ID
+        bucket_name: S3 bucket (empty string for cloud recordings)
+        room_name: Daily.co room name
+        start_ts: Unix timestamp
+        track_keys: Track keys for raw-tracks, None for cloud
+        source: "webhook" or "polling" for logging
+
+    Returns:
+        True if created (first poller), False if already exists
+    """
+    if track_keys:
+        object_key = os.path.dirname(track_keys[0]) if track_keys else room_name
+    else:
+        object_key = room_name
+
+    created = await recordings_controller.create_orphan(
+        Recording(
+            id=recording_id,
+            bucket_name=bucket_name,
+            object_key=object_key,
+            recorded_at=datetime.fromtimestamp(start_ts, tz=timezone.utc),
+            track_keys=track_keys,
+            meeting_id=None,
+            status="orphan",
+        )
+    )
+
+    if created:
+        logger.error(
+            f"Orphan recording ({source})",
+            recording_id=recording_id,
+            room_name=room_name,
+        )
+
+    return created

server/reflector/db/__init__.py

@@ -26,6 +26,7 @@ def get_database() -> databases.Database:
 # import models
 import reflector.db.calendar_events  # noqa
 import reflector.db.daily_participant_sessions  # noqa
+import reflector.db.daily_recording_requests  # noqa
 import reflector.db.meetings  # noqa
 import reflector.db.recordings  # noqa
 import reflector.db.rooms  # noqa

server/reflector/db/daily_recording_requests.py

@@ -0,0 +1,111 @@
+from datetime import datetime
+from typing import Literal
+from uuid import UUID
+
+import sqlalchemy as sa
+from pydantic import BaseModel
+from sqlalchemy.dialects.postgresql import insert
+
+from reflector.db import get_database, metadata
+from reflector.utils.string import NonEmptyString
+
+daily_recording_requests = sa.Table(
+    "daily_recording_request",
+    metadata,
+    sa.Column("recording_id", sa.String, primary_key=True),
+    sa.Column(
+        "meeting_id",
+        sa.String,
+        sa.ForeignKey("meeting.id", ondelete="CASCADE"),
+        nullable=False,
+    ),
+    sa.Column("instance_id", sa.String, nullable=False),
+    sa.Column("type", sa.String, nullable=False),
+    sa.Column("requested_at", sa.DateTime(timezone=True), nullable=False),
+    sa.Index("idx_meeting_id", "meeting_id"),
+    sa.Index("idx_instance_id", "instance_id"),
+)
+
+
+class DailyRecordingRequest(BaseModel):
+    recording_id: NonEmptyString
+    meeting_id: NonEmptyString
+    instance_id: UUID
+    type: Literal["cloud", "raw-tracks"]
+    requested_at: datetime
+
+
+class DailyRecordingRequestsController:
+    async def create(self, request: DailyRecordingRequest) -> None:
+        stmt = insert(daily_recording_requests).values(
+            recording_id=request.recording_id,
+            meeting_id=request.meeting_id,
+            instance_id=str(request.instance_id),
+            type=request.type,
+            requested_at=request.requested_at,
+        )
+        stmt = stmt.on_conflict_do_nothing(index_elements=["recording_id"])
+        await get_database().execute(stmt)
+
+    async def find_by_recording_id(
+        self,
+        recording_id: NonEmptyString,
+    ) -> tuple[NonEmptyString, Literal["cloud", "raw-tracks"]] | None:
+        query = daily_recording_requests.select().where(
+            daily_recording_requests.c.recording_id == recording_id
+        )
+        result = await get_database().fetch_one(query)
+
+        if not result:
+            return None
+
+        req = DailyRecordingRequest(
+            recording_id=result["recording_id"],
+            meeting_id=result["meeting_id"],
+            instance_id=UUID(result["instance_id"]),
+            type=result["type"],
+            requested_at=result["requested_at"],
+        )
+        return (req.meeting_id, req.type)
+
+    async def find_by_instance_id(
+        self,
+        instance_id: UUID,
+    ) -> list[DailyRecordingRequest]:
+        """Multiple recordings can have same instance_id (stop/restart)."""
+        query = daily_recording_requests.select().where(
+            daily_recording_requests.c.instance_id == str(instance_id)
+        )
+        results = await get_database().fetch_all(query)
+        return [
+            DailyRecordingRequest(
+                recording_id=r["recording_id"],
+                meeting_id=r["meeting_id"],
+                instance_id=UUID(r["instance_id"]),
+                type=r["type"],
+                requested_at=r["requested_at"],
+            )
+            for r in results
+        ]
+
+    async def get_by_meeting_id(
+        self,
+        meeting_id: NonEmptyString,
+    ) -> list[DailyRecordingRequest]:
+        query = daily_recording_requests.select().where(
+            daily_recording_requests.c.meeting_id == meeting_id
+        )
+        results = await get_database().fetch_all(query)
+        return [
+            DailyRecordingRequest(
+                recording_id=r["recording_id"],
+                meeting_id=r["meeting_id"],
+                instance_id=UUID(r["instance_id"]),
+                type=r["type"],
+                requested_at=r["requested_at"],
+            )
+            for r in results
+        ]
+
+
+daily_recording_requests_controller = DailyRecordingRequestsController()