self-review

2025-12-23 05:39:05 +00:00 · 2025-12-22 18:35:33 -05:00 · 2025-12-22 18:18:03 -05:00 · 2025-12-22 18:06:59 -05:00 · 2025-12-22 17:19:00 -05:00 · 2025-12-22 15:59:52 -05:00
69 changed files with 6218 additions and 2301 deletions
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,34 @@
 # Changelog

+## [0.25.0](https://github.com/Monadical-SAS/reflector/compare/v0.24.0...v0.25.0) (2025-12-22)
+
+
+### Features
+
+* consent disable feature ([#799](https://github.com/Monadical-SAS/reflector/issues/799)) ([2257834](https://github.com/Monadical-SAS/reflector/commit/225783496f2e265d5cb58e3539a20bf6b55589b8))
+* durable ([#794](https://github.com/Monadical-SAS/reflector/issues/794)) ([1dac999](https://github.com/Monadical-SAS/reflector/commit/1dac999b56997582ce400e7d56e915adc1e4728d))
+* increase daily recording max duration ([#801](https://github.com/Monadical-SAS/reflector/issues/801)) ([f580b99](https://github.com/Monadical-SAS/reflector/commit/f580b996eef49cce16433c505abfc6454dd45de1))
+
+
+### Bug Fixes
+
+* logout redirect ([#802](https://github.com/Monadical-SAS/reflector/issues/802)) ([f0ee7b5](https://github.com/Monadical-SAS/reflector/commit/f0ee7b531a0911f214ccbb84d399e9a6c9b700c0))
+
+## [0.24.0](https://github.com/Monadical-SAS/reflector/compare/v0.23.2...v0.24.0) (2025-12-18)
+
+
+### Features
+
+* identify action items ([#790](https://github.com/Monadical-SAS/reflector/issues/790)) ([964cd78](https://github.com/Monadical-SAS/reflector/commit/964cd78bb699d83d012ae4b8c96565df25b90a5d))
+
+
+### Bug Fixes
+
+* automatically reprocess daily recordings ([#797](https://github.com/Monadical-SAS/reflector/issues/797)) ([5f458aa](https://github.com/Monadical-SAS/reflector/commit/5f458aa4a7ec3d00ca5ec49d62fcc8ad232b138e))
+* daily video optimisation ([#789](https://github.com/Monadical-SAS/reflector/issues/789)) ([16284e1](https://github.com/Monadical-SAS/reflector/commit/16284e1ac3faede2b74f0d91b50c0b5612af2c35))
+* main menu login ([#800](https://github.com/Monadical-SAS/reflector/issues/800)) ([0bc971b](https://github.com/Monadical-SAS/reflector/commit/0bc971ba966a52d719c8c240b47dc7b3bdea4391))
+* retry on workflow timeout ([#798](https://github.com/Monadical-SAS/reflector/issues/798)) ([5f7dfad](https://github.com/Monadical-SAS/reflector/commit/5f7dfadabd3e8017406ad3720ba495a59963ee34))
+
 ## [0.23.2](https://github.com/Monadical-SAS/reflector/compare/v0.23.1...v0.23.2) (2025-12-11)


--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -34,6 +34,20 @@ services:
    environment:
      ENTRYPOINT: beat

+  hatchet-worker:
+    build:
+      context: server
+    volumes:
+      - ./server/:/app/
+      - /app/.venv
+    env_file:
+      - ./server/.env
+    environment:
+      ENTRYPOINT: hatchet-worker
+    depends_on:
+      hatchet:
+        condition: service_healthy
+
  redis:
    image: redis:7.2
    ports:
@@ -55,6 +69,7 @@ services:

  postgres:
    image: postgres:17
+    command: postgres -c 'max_connections=200'
    ports:
      - 5432:5432
    environment:
@@ -63,6 +78,42 @@ services:
      POSTGRES_DB: reflector
    volumes:
      - ./data/postgres:/var/lib/postgresql/data
+      - ./server/docker/init-hatchet-db.sql:/docker-entrypoint-initdb.d/init-hatchet-db.sql:ro
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -d reflector -U reflector"]
+      interval: 10s
+      timeout: 10s
+      retries: 5
+      start_period: 10s
+
+  hatchet:
+    image: ghcr.io/hatchet-dev/hatchet/hatchet-lite:latest
+    ports:
+      - "8889:8888"
+      - "7078:7077"
+    depends_on:
+      postgres:
+        condition: service_healthy
+    environment:
+      DATABASE_URL: "postgresql://reflector:reflector@postgres:5432/hatchet?sslmode=disable"
+      SERVER_AUTH_COOKIE_DOMAIN: localhost
+      SERVER_AUTH_COOKIE_INSECURE: "t"
+      SERVER_GRPC_BIND_ADDRESS: "0.0.0.0"
+      SERVER_GRPC_INSECURE: "t"
+      SERVER_GRPC_BROADCAST_ADDRESS: hatchet:7077
+      SERVER_GRPC_PORT: "7077"
+      SERVER_URL: http://localhost:8889
+      SERVER_AUTH_SET_EMAIL_VERIFIED: "t"
+      # SERVER_DEFAULT_ENGINE_VERSION: "V1"  # default
+      SERVER_INTERNAL_CLIENT_INTERNAL_GRPC_BROADCAST_ADDRESS: hatchet:7077
+    volumes:
+      - ./data/hatchet-config:/config
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:8888/api/live"]
+      interval: 30s
+      timeout: 10s
+      retries: 5
+      start_period: 30s

 networks:
  default:
--- a/server/README.md
+++ b/server/README.md
@@ -53,6 +53,36 @@ response = sqs.receive_message(QueueUrl=queue_url, ...)
 uv run /app/requeue_uploaded_file.py TRANSCRIPT_ID
 ```

+## Hatchet Setup (Fresh DB)
+
+After resetting the Hatchet database:
+
+### Option A: Automatic (CLI)
+
+```bash
+# Get default tenant ID and create token in one command
+TENANT_ID=$(docker compose exec -T postgres psql -U reflector -d hatchet -t -c \
+  "SELECT id FROM \"Tenant\" WHERE slug = 'default';" | tr -d ' \n') && \
+TOKEN=$(docker compose exec -T hatchet /hatchet-admin token create \
+  --config /config --tenant-id "$TENANT_ID" 2>/dev/null | tr -d '\n') && \
+echo "HATCHET_CLIENT_TOKEN=$TOKEN"
+```
+
+Copy the output to `server/.env`.
+
+### Option B: Manual (UI)
+
+1. Create API token at http://localhost:8889 → Settings → API Tokens
+2. Update `server/.env`: `HATCHET_CLIENT_TOKEN=<new-token>`
+
+### Then restart workers
+
+```bash
+docker compose restart server hatchet-worker
+```
+
+Workflows register automatically when hatchet-worker starts.
+
 ## Pipeline Management

 ### Continue stuck pipeline from final summaries (identify_participants) step:
--- a/server/docker/init-hatchet-db.sql
+++ b/server/docker/init-hatchet-db.sql
@@ -0,0 +1,2 @@
+-- Create hatchet database for Hatchet workflow engine
+CREATE DATABASE hatchet;
--- a/server/migrations/versions/0f943fede0e0_add_workflow_run_id_to_transcript.py
+++ b/server/migrations/versions/0f943fede0e0_add_workflow_run_id_to_transcript.py
@@ -0,0 +1,28 @@
+"""add workflow_run_id to transcript
+
+Revision ID: 0f943fede0e0
+Revises: 20251217000000
+Create Date: 2025-12-16 01:54:13.855106
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "0f943fede0e0"
+down_revision: Union[str, None] = "20251217000000"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("transcript", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("workflow_run_id", sa.String(), nullable=True))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("transcript", schema=None) as batch_op:
+        batch_op.drop_column("workflow_run_id")
--- a/server/migrations/versions/20251217000000_add_skip_consent_to_room.py
+++ b/server/migrations/versions/20251217000000_add_skip_consent_to_room.py
@@ -0,0 +1,35 @@
+"""add skip_consent to room
+
+Revision ID: 20251217000000
+Revises: 05f8688d6895
+Create Date: 2025-12-17 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "20251217000000"
+down_revision: Union[str, None] = "05f8688d6895"
+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.add_column(
+            sa.Column(
+                "skip_consent",
+                sa.Boolean(),
+                nullable=False,
+                server_default=sa.text("false"),
+            )
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("room", schema=None) as batch_op:
+        batch_op.drop_column("skip_consent")
--- a/server/migrations/versions/bd3a729bb379_add_use_hatchet_to_room.py
+++ b/server/migrations/versions/bd3a729bb379_add_use_hatchet_to_room.py
@@ -0,0 +1,35 @@
+"""add use_hatchet to room
+
+Revision ID: bd3a729bb379
+Revises: 0f943fede0e0
+Create Date: 2025-12-16 16:34:03.594231
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "bd3a729bb379"
+down_revision: Union[str, None] = "0f943fede0e0"
+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.add_column(
+            sa.Column(
+                "use_hatchet",
+                sa.Boolean(),
+                server_default=sa.text("false"),
+                nullable=False,
+            )
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("room", schema=None) as batch_op:
+        batch_op.drop_column("use_hatchet")
--- a/server/pyproject.toml
+++ b/server/pyproject.toml
@@ -39,6 +39,7 @@ dependencies = [
    "pytest-env>=1.1.5",
    "webvtt-py>=0.5.0",
    "icalendar>=6.0.0",
+    "hatchet-sdk>=0.47.0",
 ]

 [dependency-groups]
--- a/server/reflector/dailyco_api/client.py
+++ b/server/reflector/dailyco_api/client.py
@@ -47,7 +47,7 @@ class DailyApiError(Exception):
        )

        super().__init__(
-            f"Daily.co API error: {operation} failed with status {self.status_code}"
+            f"Daily.co API error: {operation} failed with status {self.status_code}: {response.text}"
        )


--- a/server/reflector/dailyco_api/requests.py
+++ b/server/reflector/dailyco_api/requests.py
@@ -91,6 +91,10 @@ class MeetingTokenProperties(BaseModel):
    start_cloud_recording: bool = Field(
        default=False, description="Automatically start cloud recording on join"
    )
+    start_cloud_recording_opts: dict | None = Field(
+        default=None,
+        description="Options for startRecording when start_cloud_recording is true (e.g., maxDuration)",
+    )
    enable_recording_ui: bool = Field(
        default=True, description="Show recording controls in UI"
    )
--- a/server/reflector/db/rooms.py
+++ b/server/reflector/db/rooms.py
@@ -57,6 +57,18 @@ rooms = sqlalchemy.Table(
        sqlalchemy.String,
        nullable=False,
    ),
+    sqlalchemy.Column(
+        "use_hatchet",
+        sqlalchemy.Boolean,
+        nullable=False,
+        server_default=false(),
+    ),
+    sqlalchemy.Column(
+        "skip_consent",
+        sqlalchemy.Boolean,
+        nullable=False,
+        server_default=sqlalchemy.sql.false(),
+    ),
    sqlalchemy.Index("idx_room_is_shared", "is_shared"),
    sqlalchemy.Index("idx_room_ics_enabled", "ics_enabled"),
 )
@@ -85,6 +97,8 @@ 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_hatchet: bool = False
+    skip_consent: bool = False


 class RoomController:
@@ -139,6 +153,7 @@ class RoomController:
        ics_fetch_interval: int = 300,
        ics_enabled: bool = False,
        platform: Platform = settings.DEFAULT_VIDEO_PLATFORM,
+        skip_consent: bool = False,
    ):
        """
        Add a new room
@@ -163,6 +178,7 @@ class RoomController:
            "ics_fetch_interval": ics_fetch_interval,
            "ics_enabled": ics_enabled,
            "platform": platform,
+            "skip_consent": skip_consent,
        }

        room = Room(**room_data)
--- a/server/reflector/db/transcripts.py
+++ b/server/reflector/db/transcripts.py
@@ -84,6 +84,8 @@ transcripts = sqlalchemy.Table(
    sqlalchemy.Column("audio_deleted", sqlalchemy.Boolean),
    sqlalchemy.Column("room_id", sqlalchemy.String),
    sqlalchemy.Column("webvtt", sqlalchemy.Text),
+    # Hatchet workflow run ID for resumption of failed workflows
+    sqlalchemy.Column("workflow_run_id", sqlalchemy.String),
    sqlalchemy.Index("idx_transcript_recording_id", "recording_id"),
    sqlalchemy.Index("idx_transcript_user_id", "user_id"),
    sqlalchemy.Index("idx_transcript_created_at", "created_at"),
@@ -223,6 +225,7 @@ class Transcript(BaseModel):
    zulip_message_id: int | None = None
    audio_deleted: bool | None = None
    webvtt: str | None = None
+    workflow_run_id: str | None = None  # Hatchet workflow run ID for resumption

    @field_serializer("created_at", when_used="json")
    def serialize_datetime(self, dt: datetime) -> str:
--- a/server/reflector/hatchet/init.py
+++ b/server/reflector/hatchet/init.py
@@ -0,0 +1,5 @@
+"""Hatchet workflow orchestration for Reflector."""
+
+from reflector.hatchet.client import HatchetClientManager
+
+__all__ = ["HatchetClientManager"]
--- a/server/reflector/hatchet/broadcast.py
+++ b/server/reflector/hatchet/broadcast.py
@@ -0,0 +1,98 @@
+"""WebSocket broadcasting helpers for Hatchet workflows.
+
+DUPLICATION NOTE: To be kept when Celery is deprecated. Currently dupes Celery logic.
+
+Provides WebSocket broadcasting for Hatchet that matches Celery's @broadcast_to_sockets
+decorator behavior. Events are broadcast to transcript rooms and user rooms.
+"""
+
+from typing import Any
+
+import structlog
+
+from reflector.db.transcripts import Transcript, TranscriptEvent, transcripts_controller
+from reflector.utils.string import NonEmptyString
+from reflector.ws_manager import get_ws_manager
+
+# Events that should also be sent to user room (matches Celery behavior)
+USER_ROOM_EVENTS = {"STATUS", "FINAL_TITLE", "DURATION"}
+
+
+async def broadcast_event(
+    transcript_id: NonEmptyString,
+    event: TranscriptEvent,
+    logger: structlog.BoundLogger,
+) -> None:
+    """Broadcast a TranscriptEvent to WebSocket subscribers.
+
+    Fire-and-forget: errors are logged but don't interrupt workflow execution.
+    """
+    logger.info(
+        "Broadcasting event",
+        transcript_id=transcript_id,
+        event_type=event.event,
+    )
+    try:
+        ws_manager = get_ws_manager()
+
+        await ws_manager.send_json(
+            room_id=f"ts:{transcript_id}",
+            message=event.model_dump(mode="json"),
+        )
+        logger.info(
+            "Event sent to transcript room",
+            transcript_id=transcript_id,
+            event_type=event.event,
+        )
+
+        if event.event in USER_ROOM_EVENTS:
+            transcript = await transcripts_controller.get_by_id(transcript_id)
+            if transcript and transcript.user_id:
+                await ws_manager.send_json(
+                    room_id=f"user:{transcript.user_id}",
+                    message={
+                        "event": f"TRANSCRIPT_{event.event}",
+                        "data": {"id": transcript_id, **event.data},
+                    },
+                )
+    except Exception as e:
+        logger.warning(
+            "Failed to broadcast event",
+            error=str(e),
+            transcript_id=transcript_id,
+            event_type=event.event,
+        )
+
+
+async def set_status_and_broadcast(
+    transcript_id: NonEmptyString,
+    status: str,
+    logger: structlog.BoundLogger,
+) -> None:
+    """Set transcript status and broadcast to WebSocket.
+
+    Wrapper around transcripts_controller.set_status that adds WebSocket broadcasting.
+    """
+    event = await transcripts_controller.set_status(transcript_id, status)
+    if event:
+        await broadcast_event(transcript_id, event, logger=logger)
+
+
+async def append_event_and_broadcast(
+    transcript_id: NonEmptyString,
+    transcript: Transcript,
+    event_name: str,
+    data: Any,
+    logger: structlog.BoundLogger,
+) -> TranscriptEvent:
+    """Append event to transcript and broadcast to WebSocket.
+
+    Wrapper around transcripts_controller.append_event that adds WebSocket broadcasting.
+    """
+    event = await transcripts_controller.append_event(
+        transcript=transcript,
+        event=event_name,
+        data=data,
+    )
+    await broadcast_event(transcript_id, event, logger=logger)
+    return event
--- a/server/reflector/hatchet/client.py
+++ b/server/reflector/hatchet/client.py
@@ -0,0 +1,115 @@
+"""Hatchet Python client wrapper.
+
+Uses singleton pattern because:
+1. Hatchet client maintains persistent gRPC connections for workflow registration
+2. Creating multiple clients would cause registration conflicts and resource leaks
+3. The SDK is designed for a single client instance per process
+4. Tests use `HatchetClientManager.reset()` to isolate state between tests
+"""
+
+import logging
+import threading
+
+from hatchet_sdk import ClientConfig, Hatchet
+from hatchet_sdk.clients.rest.models import V1TaskStatus
+
+from reflector.logger import logger
+from reflector.settings import settings
+
+
+class HatchetClientManager:
+    """Singleton manager for Hatchet client connections.
+
+    See module docstring for rationale. For test isolation, use `reset()`.
+    """
+
+    _instance: Hatchet | None = None
+    _lock = threading.Lock()
+
+    @classmethod
+    def get_client(cls) -> Hatchet:
+        """Get or create the Hatchet client (thread-safe singleton)."""
+        if cls._instance is None:
+            with cls._lock:
+                if cls._instance is None:
+                    if not settings.HATCHET_CLIENT_TOKEN:
+                        raise ValueError("HATCHET_CLIENT_TOKEN must be set")
+
+                    # Pass root logger to Hatchet so workflow logs appear in dashboard
+                    root_logger = logging.getLogger()
+                    cls._instance = Hatchet(
+                        debug=settings.HATCHET_DEBUG,
+                        config=ClientConfig(logger=root_logger),
+                    )
+        return cls._instance
+
+    @classmethod
+    async def start_workflow(
+        cls,
+        workflow_name: str,
+        input_data: dict,
+        additional_metadata: dict | None = None,
+    ) -> str:
+        """Start a workflow and return the workflow run ID.
+
+        Args:
+            workflow_name: Name of the workflow to trigger.
+            input_data: Input data for the workflow run.
+            additional_metadata: Optional metadata for filtering in dashboard
+                (e.g., transcript_id, recording_id).
+        """
+        client = cls.get_client()
+        result = await client.runs.aio_create(
+            workflow_name,
+            input_data,
+            additional_metadata=additional_metadata,
+        )
+        return result.run.metadata.id
+
+    @classmethod
+    async def get_workflow_run_status(cls, workflow_run_id: str) -> V1TaskStatus:
+        client = cls.get_client()
+        return await client.runs.aio_get_status(workflow_run_id)
+
+    @classmethod
+    async def cancel_workflow(cls, workflow_run_id: str) -> None:
+        client = cls.get_client()
+        await client.runs.aio_cancel(workflow_run_id)
+        logger.info("[Hatchet] Cancelled workflow", workflow_run_id=workflow_run_id)
+
+    @classmethod
+    async def replay_workflow(cls, workflow_run_id: str) -> None:
+        client = cls.get_client()
+        await client.runs.aio_replay(workflow_run_id)
+        logger.info("[Hatchet] Replaying workflow", workflow_run_id=workflow_run_id)
+
+    @classmethod
+    async def can_replay(cls, workflow_run_id: str) -> bool:
+        """Check if workflow can be replayed (is FAILED only).
+
+        CANCELLED workflows should start fresh (new run ID) rather than replay,
+        since cancellation indicates user intent to abort.
+        """
+        try:
+            status = await cls.get_workflow_run_status(workflow_run_id)
+            return status == V1TaskStatus.FAILED
+        except Exception as e:
+            logger.warning(
+                "[Hatchet] Failed to check replay status",
+                workflow_run_id=workflow_run_id,
+                error=str(e),
+            )
+            return False
+
+    @classmethod
+    async def get_workflow_status(cls, workflow_run_id: str) -> dict:
+        """Get the full workflow run details as dict."""
+        client = cls.get_client()
+        run = await client.runs.aio_get(workflow_run_id)
+        return run.to_dict()
+
+    @classmethod
+    def reset(cls) -> None:
+        """Reset the client instance (for testing)."""
+        with cls._lock:
+            cls._instance = None
--- a/server/reflector/hatchet/run_workers.py
+++ b/server/reflector/hatchet/run_workers.py
@@ -0,0 +1,70 @@
+"""
+Run Hatchet workers for the diarization pipeline.
+Runs as a separate process, just like Celery workers.
+
+Usage:
+    uv run -m reflector.hatchet.run_workers
+
+    # Or via docker:
+    docker compose exec server uv run -m reflector.hatchet.run_workers
+"""
+
+import signal
+import sys
+
+from reflector.logger import logger
+from reflector.settings import settings
+
+
+def main() -> None:
+    """Start Hatchet worker polling."""
+    if not settings.HATCHET_ENABLED:
+        logger.error("HATCHET_ENABLED is False, not starting workers")
+        sys.exit(1)
+
+    if not settings.HATCHET_CLIENT_TOKEN:
+        logger.error("HATCHET_CLIENT_TOKEN is not set")
+        sys.exit(1)
+
+    logger.info(
+        "Starting Hatchet workers",
+        debug=settings.HATCHET_DEBUG,
+    )
+
+    # Import here (not top-level) - workflow modules call HatchetClientManager.get_client()
+    # at module level because Hatchet SDK decorators (@workflow.task) bind at import time.
+    # Can't use lazy init: decorators need the client object when function is defined.
+    from reflector.hatchet.client import HatchetClientManager  # noqa: PLC0415
+    from reflector.hatchet.workflows import (  # noqa: PLC0415
+        diarization_pipeline,
+        subject_workflow,
+        topic_chunk_workflow,
+        track_workflow,
+    )
+
+    hatchet = HatchetClientManager.get_client()
+
+    worker = hatchet.worker(
+        "reflector-diarization-worker",
+        workflows=[
+            diarization_pipeline,
+            subject_workflow,
+            topic_chunk_workflow,
+            track_workflow,
+        ],
+    )
+
+    def shutdown_handler(signum: int, frame) -> None:
+        logger.info("Received shutdown signal, stopping workers...")
+        # Worker cleanup happens automatically on exit
+        sys.exit(0)
+
+    signal.signal(signal.SIGINT, shutdown_handler)
+    signal.signal(signal.SIGTERM, shutdown_handler)
+
+    logger.info("Starting Hatchet worker polling...")
+    worker.start()
+
+
+if __name__ == "__main__":
+    main()
--- a/server/reflector/hatchet/workflows/init.py
+++ b/server/reflector/hatchet/workflows/init.py
@@ -0,0 +1,26 @@
+"""Hatchet workflow definitions."""
+
+from reflector.hatchet.workflows.diarization_pipeline import (
+    PipelineInput,
+    diarization_pipeline,
+)
+from reflector.hatchet.workflows.subject_processing import (
+    SubjectInput,
+    subject_workflow,
+)
+from reflector.hatchet.workflows.topic_chunk_processing import (
+    TopicChunkInput,
+    topic_chunk_workflow,
+)
+from reflector.hatchet.workflows.track_processing import TrackInput, track_workflow
+
+__all__ = [
+    "diarization_pipeline",
+    "subject_workflow",
+    "topic_chunk_workflow",
+    "track_workflow",
+    "PipelineInput",
+    "SubjectInput",
+    "TopicChunkInput",
+    "TrackInput",
+]
--- a/server/reflector/hatchet/workflows/diarization_pipeline.py
+++ b/server/reflector/hatchet/workflows/diarization_pipeline.py
--- a/server/reflector/hatchet/workflows/models.py
+++ b/server/reflector/hatchet/workflows/models.py
@@ -0,0 +1,171 @@
+"""
+Pydantic models for Hatchet workflow task return types.
+
+Provides static typing for all task outputs, enabling type checking
+and better IDE support.
+"""
+
+from pydantic import BaseModel
+
+from reflector.processors.types import TitleSummary, Word
+from reflector.utils.string import NonEmptyString
+
+
+class ParticipantInfo(BaseModel):
+    """Participant info with speaker index for workflow result."""
+
+    participant_id: NonEmptyString
+    user_name: NonEmptyString
+    speaker: int
+
+
+class PadTrackResult(BaseModel):
+    """Result from pad_track task."""
+
+    padded_key: NonEmptyString  # S3 key (not presigned URL) - presign on demand to avoid stale URLs on replay
+    bucket_name: (
+        NonEmptyString | None
+    )  # None means use default transcript storage bucket
+    size: int
+    track_index: int
+
+
+class TranscribeTrackResult(BaseModel):
+    """Result from transcribe_track task."""
+
+    words: list[Word]
+    track_index: int
+
+
+class RecordingResult(BaseModel):
+    """Result from get_recording task."""
+
+    id: NonEmptyString | None
+    mtg_session_id: NonEmptyString | None
+    duration: float
+
+
+class ParticipantsResult(BaseModel):
+    """Result from get_participants task."""
+
+    participants: list[ParticipantInfo]
+    num_tracks: int
+    source_language: NonEmptyString
+    target_language: NonEmptyString
+
+
+class PaddedTrackInfo(BaseModel):
+    """Info for a padded track - S3 key + bucket for on-demand presigning."""
+
+    key: NonEmptyString
+    bucket_name: NonEmptyString | None  # None = use default storage bucket
+
+
+class ProcessTracksResult(BaseModel):
+    """Result from process_tracks task."""
+
+    all_words: list[Word]
+    padded_tracks: list[PaddedTrackInfo]  # S3 keys, not presigned URLs
+    word_count: int
+    num_tracks: int
+    target_language: NonEmptyString
+    created_padded_files: list[NonEmptyString]
+
+
+class MixdownResult(BaseModel):
+    """Result from mixdown_tracks task."""
+
+    audio_key: NonEmptyString
+    duration: float
+    tracks_mixed: int
+
+
+class WaveformResult(BaseModel):
+    """Result from generate_waveform task."""
+
+    waveform_generated: bool
+
+
+class TopicChunkResult(BaseModel):
+    """Result from topic chunk child workflow."""
+
+    chunk_index: int
+    title: str
+    summary: str
+    timestamp: float
+    duration: float
+    words: list[Word]
+
+
+class TopicsResult(BaseModel):
+    """Result from detect_topics task."""
+
+    topics: list[TitleSummary]
+
+
+class TitleResult(BaseModel):
+    """Result from generate_title task."""
+
+    title: str | None
+
+
+class SubjectsResult(BaseModel):
+    """Result from extract_subjects task."""
+
+    subjects: list[str]
+    transcript_text: str  # Formatted transcript for LLM consumption
+    participant_names: list[str]
+    participant_name_to_id: dict[str, str]
+
+
+class SubjectSummaryResult(BaseModel):
+    """Result from subject summary child workflow."""
+
+    subject: str
+    subject_index: int
+    detailed_summary: str
+    paragraph_summary: str
+
+
+class ProcessSubjectsResult(BaseModel):
+    """Result from process_subjects fan-out task."""
+
+    subject_summaries: list[SubjectSummaryResult]
+
+
+class RecapResult(BaseModel):
+    """Result from generate_recap task."""
+
+    short_summary: str  # Recap paragraph
+    long_summary: str  # Full markdown summary
+
+
+class ActionItemsResult(BaseModel):
+    """Result from identify_action_items task."""
+
+    action_items: dict  # ActionItemsResponse as dict (may have empty lists)
+
+
+class FinalizeResult(BaseModel):
+    """Result from finalize task."""
+
+    status: NonEmptyString
+
+
+class ConsentResult(BaseModel):
+    """Result from cleanup_consent task."""
+
+
+class ZulipResult(BaseModel):
+    """Result from post_zulip task."""
+
+    zulip_message_id: int | None = None
+    skipped: bool = False
+
+
+class WebhookResult(BaseModel):
+    """Result from send_webhook task."""
+
+    webhook_sent: bool
+    skipped: bool = False
+    response_code: int | None = None
--- a/server/reflector/hatchet/workflows/subject_processing.py
+++ b/server/reflector/hatchet/workflows/subject_processing.py
@@ -0,0 +1,101 @@
+"""
+Hatchet child workflow: SubjectProcessing
+
+Handles individual subject/topic summary generation.
+Spawned dynamically by the main diarization pipeline for each extracted subject
+via aio_run_many() for parallel processing.
+"""
+
+from datetime import timedelta
+
+from hatchet_sdk import Context
+from pydantic import BaseModel
+
+from reflector.hatchet.client import HatchetClientManager
+from reflector.hatchet.workflows.models import SubjectSummaryResult
+from reflector.logger import logger
+from reflector.processors.summary.prompts import (
+    DETAILED_SUBJECT_PROMPT_TEMPLATE,
+    PARAGRAPH_SUMMARY_PROMPT,
+    build_participant_instructions,
+)
+
+
+class SubjectInput(BaseModel):
+    """Input for individual subject processing."""
+
+    subject: str
+    subject_index: int
+    transcript_text: str
+    participant_names: list[str]
+    participant_name_to_id: dict[str, str]
+
+
+hatchet = HatchetClientManager.get_client()
+
+subject_workflow = hatchet.workflow(
+    name="SubjectProcessing", input_validator=SubjectInput
+)
+
+
+@subject_workflow.task(execution_timeout=timedelta(seconds=120), retries=3)
+async def generate_detailed_summary(
+    input: SubjectInput, ctx: Context
+) -> SubjectSummaryResult:
+    """Generate detailed analysis for a single subject, then condense to paragraph."""
+    ctx.log(
+        f"generate_detailed_summary: subject '{input.subject}' (index {input.subject_index})"
+    )
+    logger.info(
+        "[Hatchet] generate_detailed_summary",
+        subject=input.subject,
+        subject_index=input.subject_index,
+    )
+
+    # Deferred imports: Hatchet workers fork processes, fresh imports ensure
+    # LLM HTTP connection pools aren't shared across forks
+    from reflector.llm import LLM  # noqa: PLC0415
+    from reflector.settings import settings  # noqa: PLC0415
+
+    llm = LLM(settings=settings)
+
+    participant_instructions = build_participant_instructions(input.participant_names)
+    detailed_prompt = DETAILED_SUBJECT_PROMPT_TEMPLATE.format(subject=input.subject)
+    if participant_instructions:
+        detailed_prompt = f"{detailed_prompt}\n\n{participant_instructions}"
+
+    ctx.log("generate_detailed_summary: calling LLM for detailed analysis")
+    detailed_response = await llm.get_response(
+        detailed_prompt,
+        [input.transcript_text],
+        tone_name="Topic assistant",
+    )
+    detailed_summary = str(detailed_response)
+
+    paragraph_prompt = PARAGRAPH_SUMMARY_PROMPT
+    if participant_instructions:
+        paragraph_prompt = f"{paragraph_prompt}\n\n{participant_instructions}"
+
+    ctx.log("generate_detailed_summary: calling LLM for paragraph summary")
+    paragraph_response = await llm.get_response(
+        paragraph_prompt,
+        [detailed_summary],
+        tone_name="Topic summarizer",
+    )
+    paragraph_summary = str(paragraph_response)
+
+    ctx.log(f"generate_detailed_summary complete: subject '{input.subject}'")
+    logger.info(
+        "[Hatchet] generate_detailed_summary complete",
+        subject=input.subject,
+        subject_index=input.subject_index,
+        detailed_len=len(detailed_summary),
+        paragraph_len=len(paragraph_summary),
+    )
+
+    return SubjectSummaryResult(
+        subject=input.subject,
+        subject_index=input.subject_index,
+        detailed_summary=detailed_summary,
+        paragraph_summary=paragraph_summary,
+    )
--- a/server/reflector/hatchet/workflows/topic_chunk_processing.py
+++ b/server/reflector/hatchet/workflows/topic_chunk_processing.py
@@ -0,0 +1,85 @@
+"""
+Hatchet child workflow: TopicChunkProcessing
+
+Handles topic detection for individual transcript chunks.
+Spawned dynamically by detect_topics via aio_run_many() for parallel processing.
+"""
+
+from datetime import timedelta
+
+from hatchet_sdk import Context
+from pydantic import BaseModel
+
+from reflector.hatchet.client import HatchetClientManager
+from reflector.hatchet.workflows.models import TopicChunkResult
+from reflector.logger import logger
+from reflector.processors.prompts import TOPIC_PROMPT
+from reflector.processors.types import Word
+
+
+class TopicChunkInput(BaseModel):
+    """Input for individual topic chunk processing."""
+
+    chunk_index: int
+    chunk_text: str
+    timestamp: float
+    duration: float
+    words: list[Word]
+
+
+hatchet = HatchetClientManager.get_client()
+
+topic_chunk_workflow = hatchet.workflow(
+    name="TopicChunkProcessing", input_validator=TopicChunkInput
+)
+
+
+@topic_chunk_workflow.task(execution_timeout=timedelta(seconds=60), retries=3)
+async def detect_chunk_topic(input: TopicChunkInput, ctx: Context) -> TopicChunkResult:
+    """Detect topic for a single transcript chunk."""
+    ctx.log(f"detect_chunk_topic: chunk {input.chunk_index}")
+    logger.info(
+        "[Hatchet] detect_chunk_topic",
+        chunk_index=input.chunk_index,
+        text_length=len(input.chunk_text),
+    )
+
+    # Deferred imports: Hatchet workers fork processes, fresh imports avoid
+    # sharing LLM HTTP connection pools across forks
+    from reflector.llm import LLM  # noqa: PLC0415
+    from reflector.processors.transcript_topic_detector import (  # noqa: PLC0415
+        TopicResponse,
+    )
+    from reflector.settings import settings  # noqa: PLC0415
+    from reflector.utils.text import clean_title  # noqa: PLC0415
+
+    llm = LLM(settings=settings, temperature=0.9, max_tokens=500)
+
+    prompt = TOPIC_PROMPT.format(text=input.chunk_text)
+    response = await llm.get_structured_response(
+        prompt,
+        [input.chunk_text],
+        TopicResponse,
+        tone_name="Topic analyzer",
+        timeout=settings.LLM_STRUCTURED_RESPONSE_TIMEOUT,
+    )
+
+    title = clean_title(response.title)
+
+    ctx.log(
+        f"detect_chunk_topic complete: chunk {input.chunk_index}, title='{title[:50]}'"
+    )
+    logger.info(
+        "[Hatchet] detect_chunk_topic complete",
+        chunk_index=input.chunk_index,
+        title=title[:50],
+    )
+
+    return TopicChunkResult(
+        chunk_index=input.chunk_index,
+        title=title,
+        summary=response.summary,
+        timestamp=input.timestamp,
+        duration=input.duration,
+        words=input.words,
+    )
--- a/server/reflector/hatchet/workflows/track_processing.py
+++ b/server/reflector/hatchet/workflows/track_processing.py
@@ -0,0 +1,219 @@
+"""
+Hatchet child workflow: TrackProcessing
+
+Handles individual audio track processing: padding and transcription.
+Spawned dynamically by the main diarization pipeline for each track.
+
+Architecture note: This is a separate workflow (not inline tasks in DiarizationPipeline)
+because Hatchet workflow DAGs are defined statically, but the number of tracks varies
+at runtime. Child workflow spawning via `aio_run()` + `asyncio.gather()` is the
+standard pattern for dynamic fan-out. See `process_tracks` in diarization_pipeline.py.
+
+Note: This file uses deferred imports (inside tasks) intentionally.
+Hatchet workers run in forked processes; fresh imports per task ensure
+storage/DB connections are not shared across forks.
+"""
+
+import tempfile
+from datetime import timedelta
+from pathlib import Path
+
+import av
+from hatchet_sdk import Context
+from pydantic import BaseModel
+
+from reflector.hatchet.client import HatchetClientManager
+from reflector.hatchet.workflows.models import PadTrackResult, TranscribeTrackResult
+from reflector.logger import logger
+from reflector.utils.audio_constants import PRESIGNED_URL_EXPIRATION_SECONDS
+from reflector.utils.audio_padding import (
+    apply_audio_padding_to_file,
+    extract_stream_start_time_from_container,
+)
+
+
+class TrackInput(BaseModel):
+    """Input for individual track processing."""
+
+    track_index: int
+    s3_key: str
+    bucket_name: str
+    transcript_id: str
+    language: str = "en"
+
+
+hatchet = HatchetClientManager.get_client()
+
+track_workflow = hatchet.workflow(name="TrackProcessing", input_validator=TrackInput)
+
+
+@track_workflow.task(execution_timeout=timedelta(seconds=300), retries=3)
+async def pad_track(input: TrackInput, ctx: Context) -> PadTrackResult:
+    """Pad single audio track with silence for alignment.
+
+    Extracts stream.start_time from WebM container metadata and applies
+    silence padding using PyAV filter graph (adelay).
+    """
+    ctx.log(f"pad_track: track {input.track_index}, s3_key={input.s3_key}")
+    logger.info(
+        "[Hatchet] pad_track",
+        track_index=input.track_index,
+        s3_key=input.s3_key,
+        transcript_id=input.transcript_id,
+    )
+
+    try:
+        # Create fresh storage instance to avoid aioboto3 fork issues
+        from reflector.settings import settings  # noqa: PLC0415
+        from reflector.storage.storage_aws import AwsStorage  # noqa: PLC0415
+
+        storage = AwsStorage(
+            aws_bucket_name=settings.TRANSCRIPT_STORAGE_AWS_BUCKET_NAME,
+            aws_region=settings.TRANSCRIPT_STORAGE_AWS_REGION,
+            aws_access_key_id=settings.TRANSCRIPT_STORAGE_AWS_ACCESS_KEY_ID,
+            aws_secret_access_key=settings.TRANSCRIPT_STORAGE_AWS_SECRET_ACCESS_KEY,
+        )
+
+        source_url = await storage.get_file_url(
+            input.s3_key,
+            operation="get_object",
+            expires_in=PRESIGNED_URL_EXPIRATION_SECONDS,
+            bucket=input.bucket_name,
+        )
+
+        with av.open(source_url) as in_container:
+            start_time_seconds = extract_stream_start_time_from_container(
+                in_container, input.track_index, logger=logger
+            )
+
+            # If no padding needed, return original S3 key
+            if start_time_seconds <= 0:
+                logger.info(
+                    f"Track {input.track_index} requires no padding",
+                    track_index=input.track_index,
+                )
+                return PadTrackResult(
+                    padded_key=input.s3_key,
+                    bucket_name=input.bucket_name,
+                    size=0,
+                    track_index=input.track_index,
+                )
+
+            with tempfile.NamedTemporaryFile(suffix=".webm", delete=False) as temp_file:
+                temp_path = temp_file.name
+
+            try:
+                apply_audio_padding_to_file(
+                    in_container,
+                    temp_path,
+                    start_time_seconds,
+                    input.track_index,
+                    logger=logger,
+                )
+
+                file_size = Path(temp_path).stat().st_size
+                storage_path = f"file_pipeline_hatchet/{input.transcript_id}/tracks/padded_{input.track_index}.webm"
+
+                logger.info(
+                    f"About to upload padded track",
+                    key=storage_path,
+                    size=file_size,
+                )
+
+                with open(temp_path, "rb") as padded_file:
+                    await storage.put_file(storage_path, padded_file)
+
+                logger.info(
+                    f"Uploaded padded track to S3",
+                    key=storage_path,
+                    size=file_size,
+                )
+            finally:
+                Path(temp_path).unlink(missing_ok=True)
+
+        ctx.log(f"pad_track complete: track {input.track_index} -> {storage_path}")
+        logger.info(
+            "[Hatchet] pad_track complete",
+            track_index=input.track_index,
+            padded_key=storage_path,
+        )
+
+        # Return S3 key (not presigned URL) - consumer tasks presign on demand
+        # This avoids stale URLs when workflow is replayed
+        return PadTrackResult(
+            padded_key=storage_path,
+            bucket_name=None,  # None = use default transcript storage bucket
+            size=file_size,
+            track_index=input.track_index,
+        )
+
+    except Exception as e:
+        logger.error("[Hatchet] pad_track failed", error=str(e), exc_info=True)
+        raise
+
+
+@track_workflow.task(
+    parents=[pad_track], execution_timeout=timedelta(seconds=600), retries=3
+)
+async def transcribe_track(input: TrackInput, ctx: Context) -> TranscribeTrackResult:
+    """Transcribe audio track using GPU (Modal.com) or local Whisper."""
+    ctx.log(f"transcribe_track: track {input.track_index}, language={input.language}")
+    logger.info(
+        "[Hatchet] transcribe_track",
+        track_index=input.track_index,
+        language=input.language,
+    )
+
+    try:
+        pad_result = ctx.task_output(pad_track)
+        padded_key = pad_result.padded_key
+        bucket_name = pad_result.bucket_name
+
+        if not padded_key:
+            raise ValueError("Missing padded_key from pad_track")
+
+        # Presign URL on demand (avoids stale URLs on workflow replay)
+        from reflector.settings import settings  # noqa: PLC0415
+        from reflector.storage.storage_aws import AwsStorage  # noqa: PLC0415
+
+        storage = AwsStorage(
+            aws_bucket_name=settings.TRANSCRIPT_STORAGE_AWS_BUCKET_NAME,
+            aws_region=settings.TRANSCRIPT_STORAGE_AWS_REGION,
+            aws_access_key_id=settings.TRANSCRIPT_STORAGE_AWS_ACCESS_KEY_ID,
+            aws_secret_access_key=settings.TRANSCRIPT_STORAGE_AWS_SECRET_ACCESS_KEY,
+        )
+
+        audio_url = await storage.get_file_url(
+            padded_key,
+            operation="get_object",
+            expires_in=PRESIGNED_URL_EXPIRATION_SECONDS,
+            bucket=bucket_name,
+        )
+
+        from reflector.pipelines.transcription_helpers import (  # noqa: PLC0415
+            transcribe_file_with_processor,
+        )
+
+        transcript = await transcribe_file_with_processor(audio_url, input.language)
+
+        # Tag all words with speaker index
+        for word in transcript.words:
+            word.speaker = input.track_index
+
+        ctx.log(
+            f"transcribe_track complete: track {input.track_index}, {len(transcript.words)} words"
+        )
+        logger.info(
+            "[Hatchet] transcribe_track complete",
+            track_index=input.track_index,
+            word_count=len(transcript.words),
+        )
+
+        return TranscribeTrackResult(
+            words=transcript.words,
+            track_index=input.track_index,
+        )
+
+    except Exception as e:
+        logger.error("[Hatchet] transcribe_track failed", error=str(e), exc_info=True)
+        raise
--- a/server/reflector/pipelines/main_file_pipeline.py
+++ b/server/reflector/pipelines/main_file_pipeline.py
@@ -97,13 +97,8 @@ class PipelineMainFile(PipelineMainBase):
                },
            )

-        # Extract audio and write to transcript location
        audio_path = await self.extract_and_write_audio(file_path, transcript)
-
-        # Upload for processing
        audio_url = await self.upload_audio(audio_path, transcript)
-
-        # Run parallel processing
        await self.run_parallel_processing(
            audio_path,
            audio_url,
@@ -197,7 +192,6 @@ class PipelineMainFile(PipelineMainBase):
        transcript_result = results[0]
        diarization_result = results[1]

-        # Handle errors - raise any exception that occurred
        self._handle_gather_exceptions(results, "parallel processing")
        for result in results:
            if isinstance(result, Exception):
@@ -212,7 +206,6 @@ class PipelineMainFile(PipelineMainBase):
            transcript=transcript_result, diarization=diarization_result or []
        )

-        # Store result for retrieval
        diarized_transcript: Transcript | None = None

        async def capture_result(transcript):
@@ -349,7 +342,6 @@ async def task_pipeline_file_process(*, transcript_id: str):
    try:
        await pipeline.set_status(transcript_id, "processing")

-        # Find the file to process
        audio_file = next(transcript.data_path.glob("upload.*"), None)
        if not audio_file:
            audio_file = next(transcript.data_path.glob("audio.*"), None)
--- a/server/reflector/pipelines/main_live_pipeline.py
+++ b/server/reflector/pipelines/main_live_pipeline.py
@@ -112,7 +112,7 @@ def get_transcript(func):
        transcript_id = kwargs.pop("transcript_id")
        transcript = await transcripts_controller.get_by_id(transcript_id=transcript_id)
        if not transcript:
-            raise Exception("Transcript {transcript_id} not found")
+            raise Exception(f"Transcript {transcript_id} not found")

        # Enhanced logger with Celery task context
        tlogger = logger.bind(transcript_id=transcript.id)
--- a/server/reflector/pipelines/main_multitrack_pipeline.py
+++ b/server/reflector/pipelines/main_multitrack_pipeline.py
@@ -1,11 +1,8 @@
 import asyncio
-import math
 import tempfile
-from fractions import Fraction
 from pathlib import Path

 import av
-from av.audio.resampler import AudioResampler
 from celery import chain, shared_task

 from reflector.asynctask import asynctask
@@ -32,6 +29,15 @@ from reflector.processors.audio_waveform_processor import AudioWaveformProcessor
 from reflector.processors.types import TitleSummary
 from reflector.processors.types import Transcript as TranscriptType
 from reflector.storage import Storage, get_transcripts_storage
+from reflector.utils.audio_constants import PRESIGNED_URL_EXPIRATION_SECONDS
+from reflector.utils.audio_mixdown import (
+    detect_sample_rate_from_tracks,
+    mixdown_tracks_pyav,
+)
+from reflector.utils.audio_padding import (
+    apply_audio_padding_to_file,
+    extract_stream_start_time_from_container,
+)
 from reflector.utils.daily import (
    filter_cam_audio_tracks,
    parse_daily_recording_filename,
@@ -39,13 +45,6 @@ from reflector.utils.daily import (
 from reflector.utils.string import NonEmptyString
 from reflector.video_platforms.factory import create_platform_client

-# Audio encoding constants
-OPUS_STANDARD_SAMPLE_RATE = 48000
-OPUS_DEFAULT_BIT_RATE = 128000
-
-# Storage operation constants
-PRESIGNED_URL_EXPIRATION_SECONDS = 7200  # 2 hours
-

 class PipelineMainMultitrack(PipelineMainBase):
    def __init__(self, transcript_id: str):
@@ -125,8 +124,8 @@ class PipelineMainMultitrack(PipelineMainBase):
        try:
            # PyAV streams input from S3 URL efficiently (2-5MB fixed overhead for codec/filters)
            with av.open(track_url) as in_container:
-                start_time_seconds = self._extract_stream_start_time_from_container(
-                    in_container, track_idx
+                start_time_seconds = extract_stream_start_time_from_container(
+                    in_container, track_idx, logger=self.logger
                )

                if start_time_seconds <= 0:
@@ -144,8 +143,12 @@ class PipelineMainMultitrack(PipelineMainBase):
                    temp_path = temp_file.name

                try:
-                    self._apply_audio_padding_to_file(
-                        in_container, temp_path, start_time_seconds, track_idx
+                    apply_audio_padding_to_file(
+                        in_container,
+                        temp_path,
+                        start_time_seconds,
+                        track_idx,
+                        logger=self.logger,
                    )

                    storage_path = (
@@ -156,7 +159,6 @@ class PipelineMainMultitrack(PipelineMainBase):
                    with open(temp_path, "rb") as padded_file:
                        await storage.put_file(storage_path, padded_file)
                finally:
-                    # Clean up temp file
                    Path(temp_path).unlink(missing_ok=True)

                padded_url = await storage.get_file_url(
@@ -186,316 +188,27 @@ class PipelineMainMultitrack(PipelineMainBase):
                f"Track {track_idx} padding failed - transcript would have incorrect timestamps"
            ) from e

-    def _extract_stream_start_time_from_container(
-        self, container, track_idx: int
-    ) -> float:
-        """
-        Extract meeting-relative start time from WebM stream metadata.
-        Uses PyAV to read stream.start_time from WebM container.
-        More accurate than filename timestamps by ~209ms due to network/encoding delays.
-        """
-        start_time_seconds = 0.0
-        try:
-            audio_streams = [s for s in container.streams if s.type == "audio"]
-            stream = audio_streams[0] if audio_streams else container.streams[0]
-
-            # 1) Try stream-level start_time (most reliable for Daily.co tracks)
-            if stream.start_time is not None and stream.time_base is not None:
-                start_time_seconds = float(stream.start_time * stream.time_base)
-
-            # 2) Fallback to container-level start_time (in av.time_base units)
-            if (start_time_seconds <= 0) and (container.start_time is not None):
-                start_time_seconds = float(container.start_time * av.time_base)
-
-            # 3) Fallback to first packet DTS in stream.time_base
-            if start_time_seconds <= 0:
-                for packet in container.demux(stream):
-                    if packet.dts is not None:
-                        start_time_seconds = float(packet.dts * stream.time_base)
-                        break
-        except Exception as e:
-            self.logger.warning(
-                "PyAV metadata read failed; assuming 0 start_time",
-                track_idx=track_idx,
-                error=str(e),
-            )
-            start_time_seconds = 0.0
-
-        self.logger.info(
-            f"Track {track_idx} stream metadata: start_time={start_time_seconds:.3f}s",
-            track_idx=track_idx,
-        )
-        return start_time_seconds
-
-    def _apply_audio_padding_to_file(
-        self,
-        in_container,
-        output_path: str,
-        start_time_seconds: float,
-        track_idx: int,
-    ) -> None:
-        """Apply silence padding to audio track using PyAV filter graph, writing to file"""
-        delay_ms = math.floor(start_time_seconds * 1000)
-
-        self.logger.info(
-            f"Padding track {track_idx} with {delay_ms}ms delay using PyAV",
-            track_idx=track_idx,
-            delay_ms=delay_ms,
-        )
-
-        try:
-            with av.open(output_path, "w", format="webm") as out_container:
-                in_stream = next(
-                    (s for s in in_container.streams if s.type == "audio"), None
-                )
-                if in_stream is None:
-                    raise Exception("No audio stream in input")
-
-                out_stream = out_container.add_stream(
-                    "libopus", rate=OPUS_STANDARD_SAMPLE_RATE
-                )
-                out_stream.bit_rate = OPUS_DEFAULT_BIT_RATE
-                graph = av.filter.Graph()
-
-                abuf_args = (
-                    f"time_base=1/{OPUS_STANDARD_SAMPLE_RATE}:"
-                    f"sample_rate={OPUS_STANDARD_SAMPLE_RATE}:"
-                    f"sample_fmt=s16:"
-                    f"channel_layout=stereo"
-                )
-                src = graph.add("abuffer", args=abuf_args, name="src")
-                aresample_f = graph.add("aresample", args="async=1", name="ares")
-                # adelay requires one delay value per channel separated by '|'
-                delays_arg = f"{delay_ms}|{delay_ms}"
-                adelay_f = graph.add(
-                    "adelay", args=f"delays={delays_arg}:all=1", name="delay"
-                )
-                sink = graph.add("abuffersink", name="sink")
-
-                src.link_to(aresample_f)
-                aresample_f.link_to(adelay_f)
-                adelay_f.link_to(sink)
-                graph.configure()
-
-                resampler = AudioResampler(
-                    format="s16", layout="stereo", rate=OPUS_STANDARD_SAMPLE_RATE
-                )
-                # Decode -> resample -> push through graph -> encode Opus
-                for frame in in_container.decode(in_stream):
-                    out_frames = resampler.resample(frame) or []
-                    for rframe in out_frames:
-                        rframe.sample_rate = OPUS_STANDARD_SAMPLE_RATE
-                        rframe.time_base = Fraction(1, OPUS_STANDARD_SAMPLE_RATE)
-                        src.push(rframe)
-
-                        while True:
-                            try:
-                                f_out = sink.pull()
-                            except Exception:
-                                break
-                            f_out.sample_rate = OPUS_STANDARD_SAMPLE_RATE
-                            f_out.time_base = Fraction(1, OPUS_STANDARD_SAMPLE_RATE)
-                            for packet in out_stream.encode(f_out):
-                                out_container.mux(packet)
-
-                src.push(None)
-                while True:
-                    try:
-                        f_out = sink.pull()
-                    except Exception:
-                        break
-                    f_out.sample_rate = OPUS_STANDARD_SAMPLE_RATE
-                    f_out.time_base = Fraction(1, OPUS_STANDARD_SAMPLE_RATE)
-                    for packet in out_stream.encode(f_out):
-                        out_container.mux(packet)
-
-                for packet in out_stream.encode(None):
-                    out_container.mux(packet)
-        except Exception as e:
-            self.logger.error(
-                "PyAV padding failed for track",
-                track_idx=track_idx,
-                delay_ms=delay_ms,
-                error=str(e),
-                exc_info=True,
-            )
-            raise
-
    async def mixdown_tracks(
        self,
        track_urls: list[str],
        writer: AudioFileWriterProcessor,
        offsets_seconds: list[float] | None = None,
    ) -> None:
-        """Multi-track mixdown using PyAV filter graph (amix), reading from S3 presigned URLs"""
-
-        target_sample_rate: int | None = None
-        for url in track_urls:
-            if not url:
-                continue
-            container = None
-            try:
-                container = av.open(url)
-                for frame in container.decode(audio=0):
-                    target_sample_rate = frame.sample_rate
-                    break
-            except Exception:
-                continue
-            finally:
-                if container is not None:
-                    container.close()
-            if target_sample_rate:
-                break
-
+        """Multi-track mixdown using PyAV filter graph (amix), reading from S3 presigned URLs."""
+        target_sample_rate = detect_sample_rate_from_tracks(
+            track_urls, logger=self.logger
+        )
        if not target_sample_rate:
            self.logger.error("Mixdown failed - no decodable audio frames found")
            raise Exception("Mixdown failed: No decodable audio frames in any track")
-        # Build PyAV filter graph:
-        # N abuffer (s32/stereo)
-        #   -> optional adelay per input (for alignment)
-        #   -> amix (s32)
-        #   -> aformat(s16)
-        #   -> sink
-        graph = av.filter.Graph()
-        inputs = []
-        valid_track_urls = [url for url in track_urls if url]
-        input_offsets_seconds = None
-        if offsets_seconds is not None:
-            input_offsets_seconds = [
-                offsets_seconds[i] for i, url in enumerate(track_urls) if url
-            ]
-        for idx, url in enumerate(valid_track_urls):
-            args = (
-                f"time_base=1/{target_sample_rate}:"
-                f"sample_rate={target_sample_rate}:"
-                f"sample_fmt=s32:"
-                f"channel_layout=stereo"
+
+        await mixdown_tracks_pyav(
+            track_urls,
+            writer,
+            target_sample_rate,
+            offsets_seconds=offsets_seconds,
+            logger=self.logger,
        )
-            in_ctx = graph.add("abuffer", args=args, name=f"in{idx}")
-            inputs.append(in_ctx)
-
-        if not inputs:
-            self.logger.error("Mixdown failed - no valid inputs for graph")
-            raise Exception("Mixdown failed: No valid inputs for filter graph")
-
-        mixer = graph.add("amix", args=f"inputs={len(inputs)}:normalize=0", name="mix")
-
-        fmt = graph.add(
-            "aformat",
-            args=(
-                f"sample_fmts=s32:channel_layouts=stereo:sample_rates={target_sample_rate}"
-            ),
-            name="fmt",
-        )
-
-        sink = graph.add("abuffersink", name="out")
-
-        # Optional per-input delay before mixing
-        delays_ms: list[int] = []
-        if input_offsets_seconds is not None:
-            base = min(input_offsets_seconds) if input_offsets_seconds else 0.0
-            delays_ms = [
-                max(0, int(round((o - base) * 1000))) for o in input_offsets_seconds
-            ]
-        else:
-            delays_ms = [0 for _ in inputs]
-
-        for idx, in_ctx in enumerate(inputs):
-            delay_ms = delays_ms[idx] if idx < len(delays_ms) else 0
-            if delay_ms > 0:
-                # adelay requires one value per channel; use same for stereo
-                adelay = graph.add(
-                    "adelay",
-                    args=f"delays={delay_ms}|{delay_ms}:all=1",
-                    name=f"delay{idx}",
-                )
-                in_ctx.link_to(adelay)
-                adelay.link_to(mixer, 0, idx)
-            else:
-                in_ctx.link_to(mixer, 0, idx)
-        mixer.link_to(fmt)
-        fmt.link_to(sink)
-        graph.configure()
-
-        containers = []
-        try:
-            # Open all containers with cleanup guaranteed
-            for i, url in enumerate(valid_track_urls):
-                try:
-                    c = av.open(
-                        url,
-                        options={
-                            # it's trying to stream from s3 by default
-                            "reconnect": "1",
-                            "reconnect_streamed": "1",
-                            "reconnect_delay_max": "5",
-                        },
-                    )
-                    containers.append(c)
-                except Exception as e:
-                    self.logger.warning(
-                        "Mixdown: failed to open container from URL",
-                        input=i,
-                        url=url,
-                        error=str(e),
-                    )
-
-            if not containers:
-                self.logger.error("Mixdown failed - no valid containers opened")
-                raise Exception("Mixdown failed: Could not open any track containers")
-
-            decoders = [c.decode(audio=0) for c in containers]
-            active = [True] * len(decoders)
-            resamplers = [
-                AudioResampler(format="s32", layout="stereo", rate=target_sample_rate)
-                for _ in decoders
-            ]
-
-            while any(active):
-                for i, (dec, is_active) in enumerate(zip(decoders, active)):
-                    if not is_active:
-                        continue
-                    try:
-                        frame = next(dec)
-                    except StopIteration:
-                        active[i] = False
-                        # causes stream to move on / unclogs memory
-                        inputs[i].push(None)
-                        continue
-
-                    if frame.sample_rate != target_sample_rate:
-                        continue
-                    out_frames = resamplers[i].resample(frame) or []
-                    for rf in out_frames:
-                        rf.sample_rate = target_sample_rate
-                        rf.time_base = Fraction(1, target_sample_rate)
-                        inputs[i].push(rf)
-
-                    while True:
-                        try:
-                            mixed = sink.pull()
-                        except Exception:
-                            break
-                        mixed.sample_rate = target_sample_rate
-                        mixed.time_base = Fraction(1, target_sample_rate)
-                        await writer.push(mixed)
-
-            while True:
-                try:
-                    mixed = sink.pull()
-                except Exception:
-                    break
-                mixed.sample_rate = target_sample_rate
-                mixed.time_base = Fraction(1, target_sample_rate)
-                await writer.push(mixed)
-        finally:
-            # Cleanup all containers, even if processing failed
-            for c in containers:
-                if c is not None:
-                    try:
-                        c.close()
-                    except Exception:
-                        pass  # Best effort cleanup

    @broadcast_to_sockets
    async def set_status(self, transcript_id: str, status: TranscriptStatus):
--- a/server/reflector/pipelines/topic_processing.py
+++ b/server/reflector/pipelines/topic_processing.py
@@ -18,6 +18,7 @@ from reflector.processors import (
 )
 from reflector.processors.types import TitleSummary
 from reflector.processors.types import Transcript as TranscriptType
+from reflector.utils.transcript_constants import TOPIC_CHUNK_WORD_COUNT


 class EmptyPipeline:
@@ -38,7 +39,7 @@ async def detect_topics(
    on_topic_callback: Callable,
    empty_pipeline: EmptyPipeline,
 ) -> list[TitleSummary]:
-    chunk_size = 300
+    chunk_size = TOPIC_CHUNK_WORD_COUNT
    topics: list[TitleSummary] = []

    async def on_topic(topic: TitleSummary):
@@ -74,13 +75,24 @@ async def generate_title(
        logger.warning("No topics for title generation")
        return

+    logger.info(
+        "generate_title: creating TranscriptFinalTitleProcessor",
+        topic_count=len(topics),
+    )
    processor = TranscriptFinalTitleProcessor(callback=on_title_callback)
    processor.set_pipeline(empty_pipeline)

-    for topic in topics:
+    for i, topic in enumerate(topics):
+        logger.info(
+            "generate_title: pushing topic to processor",
+            topic_index=i,
+            topic_title=topic.title[:50] if topic.title else None,
+        )
        await processor.push(topic)

+    logger.info("generate_title: calling processor.flush() - this triggers LLM call")
    await processor.flush()
+    logger.info("generate_title: processor.flush() completed")


 async def generate_summaries(
@@ -97,6 +109,10 @@ async def generate_summaries(
        logger.warning("No topics for summary generation")
        return

+    logger.info(
+        "generate_summaries: creating TranscriptFinalSummaryProcessor",
+        topic_count=len(topics),
+    )
    processor_kwargs = {
        "transcript": transcript,
        "callback": on_long_summary_callback,
@@ -107,7 +123,16 @@ async def generate_summaries(
    processor = TranscriptFinalSummaryProcessor(**processor_kwargs)
    processor.set_pipeline(empty_pipeline)

-    for topic in topics:
+    for i, topic in enumerate(topics):
+        logger.info(
+            "generate_summaries: pushing topic to processor",
+            topic_index=i,
+            topic_title=topic.title[:50] if topic.title else None,
+        )
        await processor.push(topic)

+    logger.info(
+        "generate_summaries: calling processor.flush() - this triggers LLM calls"
+    )
    await processor.flush()
+    logger.info("generate_summaries: processor.flush() completed")
--- a/server/reflector/processors/prompts.py
+++ b/server/reflector/processors/prompts.py
@@ -0,0 +1,30 @@
+"""
+LLM prompts for transcript processing.
+
+Extracted to a separate module to avoid circular imports when importing
+from processor modules (which import LLM/settings at module level).
+"""
+
+from textwrap import dedent
+
+TOPIC_PROMPT = dedent(
+    """
+    Analyze the following transcript segment and extract the main topic being discussed.
+    Focus on the substantive content and ignore small talk or administrative chatter.
+
+    Create a title that:
+    - Captures the specific subject matter being discussed
+    - Is descriptive and self-explanatory
+    - Uses professional language
+    - Is specific rather than generic
+
+    For the summary:
+    - Summarize the key points in maximum two sentences
+    - Focus on what was discussed, decided, or accomplished
+    - Be concise but informative
+
+    <transcript>
+    {text}
+    </transcript>
+    """
+).strip()
--- a/server/reflector/processors/summary/prompts.py
+++ b/server/reflector/processors/summary/prompts.py
@@ -0,0 +1,91 @@
+"""
+LLM prompts for summary generation.
+
+Extracted to a separate module to avoid circular imports when importing
+from summary_builder.py (which imports LLM/settings at module level).
+"""
+
+from textwrap import dedent
+
+
+def build_participant_instructions(participant_names: list[str]) -> str:
+    """Build participant context instructions for LLM prompts."""
+    if not participant_names:
+        return ""
+
+    participants_list = ", ".join(participant_names)
+    return dedent(
+        f"""
+        # IMPORTANT: Participant Names
+        The following participants are identified in this conversation: {participants_list}
+
+        You MUST use these specific participant names when referring to people in your response.
+        Do NOT use generic terms like "a participant", "someone", "attendee", "Speaker 1", "Speaker 2", etc.
+        Always refer to people by their actual names (e.g., "John suggested..." not "A participant suggested...").
+        """
+    ).strip()
+
+
+DETAILED_SUBJECT_PROMPT_TEMPLATE = dedent(
+    """
+    Get me information about the topic "{subject}"
+
+    # RESPONSE GUIDELINES
+    Follow this structured approach to create the topic summary:
+    - Highlight important arguments, insights, or data presented.
+    - Outline decisions made.
+    - Indicate any decisions reached, including any rationale or key factors
+      that influenced these decisions.
+    - Detail action items and responsibilities.
+    - For each decision or unresolved issue, list specific action items agreed
+      upon, along with assigned individuals or teams responsible for each task.
+    - Specify deadlines or timelines if mentioned. For each action item,
+      include any deadlines or timeframes discussed for completion or follow-up.
+    - Mention unresolved issues or topics needing further discussion, aiding in
+      planning future meetings or follow-up actions.
+    - Do not include topic unrelated to {subject}.
+
+    # OUTPUT
+    Your summary should be clear, concise, and structured, covering all major
+    points, decisions, and action items from the meeting. It should be easy to
+    understand for someone not present, providing a comprehensive understanding
+    of what transpired and what needs to be done next. The summary should not
+    exceed one page to ensure brevity and focus.
+    """
+).strip()
+
+PARAGRAPH_SUMMARY_PROMPT = dedent(
+    """
+    Summarize the mentioned topic in 1 paragraph.
+    It will be integrated into the final summary, so just for this topic.
+    """
+).strip()
+
+RECAP_PROMPT = dedent(
+    """
+    Provide a high-level quick recap of the following meeting, fitting in one paragraph.
+    Do not include decisions, action items or unresolved issue, just highlight the high moments.
+    Just dive into the meeting, be concise and do not include unnecessary details.
+    As we already know it is a meeting, do not start with 'During the meeting' or equivalent.
+    """
+).strip()
+
+
+def build_summary_markdown(recap: str, summaries: list[dict[str, str]]) -> str:
+    """Build markdown summary from recap and subject summaries."""
+    lines: list[str] = []
+    if recap:
+        lines.append("# Quick recap")
+        lines.append("")
+        lines.append(recap)
+        lines.append("")
+
+    if summaries:
+        lines.append("# Summary")
+        lines.append("")
+        for summary in summaries:
+            lines.append(f"**{summary['subject']}**")
+            lines.append(summary["summary"])
+            lines.append("")
+
+    return "\n".join(lines)
--- a/server/reflector/processors/summary/summary_builder.py
+++ b/server/reflector/processors/summary/summary_builder.py
@@ -15,6 +15,13 @@ import structlog
 from pydantic import BaseModel, Field

 from reflector.llm import LLM
+from reflector.processors.summary.prompts import (
+    DETAILED_SUBJECT_PROMPT_TEMPLATE,
+    PARAGRAPH_SUMMARY_PROMPT,
+    RECAP_PROMPT,
+    build_participant_instructions,
+    build_summary_markdown,
+)
 from reflector.settings import settings

 T = TypeVar("T", bound=BaseModel)
@@ -52,50 +59,6 @@ SUBJECTS_PROMPT = dedent(
    """
 ).strip()

-DETAILED_SUBJECT_PROMPT_TEMPLATE = dedent(
-    """
-    Get me information about the topic "{subject}"
-
-    # RESPONSE GUIDELINES
-    Follow this structured approach to create the topic summary:
-    - Highlight important arguments, insights, or data presented.
-    - Outline decisions made.
-    - Indicate any decisions reached, including any rationale or key factors
-      that influenced these decisions.
-    - Detail action items and responsibilities.
-    - For each decision or unresolved issue, list specific action items agreed
-      upon, along with assigned individuals or teams responsible for each task.
-    - Specify deadlines or timelines if mentioned. For each action item,
-      include any deadlines or timeframes discussed for completion or follow-up.
-    - Mention unresolved issues or topics needing further discussion, aiding in
-      planning future meetings or follow-up actions.
-    - Do not include topic unrelated to {subject}.
-
-    # OUTPUT
-    Your summary should be clear, concise, and structured, covering all major
-    points, decisions, and action items from the meeting. It should be easy to
-    understand for someone not present, providing a comprehensive understanding
-    of what transpired and what needs to be done next. The summary should not
-    exceed one page to ensure brevity and focus.
-    """
-).strip()
-
-PARAGRAPH_SUMMARY_PROMPT = dedent(
-    """
-    Summarize the mentioned topic in 1 paragraph.
-    It will be integrated into the final summary, so just for this topic.
-    """
-).strip()
-
-RECAP_PROMPT = dedent(
-    """
-    Provide a high-level quick recap of the following meeting, fitting in one paragraph.
-    Do not include decisions, action items or unresolved issue, just highlight the high moments.
-    Just dive into the meeting, be concise and do not include unnecessary details.
-    As we already know it is a meeting, do not start with 'During the meeting' or equivalent.
-    """
-).strip()
-
 ACTION_ITEMS_PROMPT = dedent(
    """
    Identify action items from this meeting transcript. Your goal is to identify what was decided and what needs to happen next.
@@ -331,17 +294,7 @@ class SummaryBuilder:
        participants_md = self.format_list_md(participants)
        self.transcript += f"\n\n# Participants\n\n{participants_md}"

-        participants_list = ", ".join(participants)
-        self.participant_instructions = dedent(
-            f"""
-            # IMPORTANT: Participant Names
-            The following participants are identified in this conversation: {participants_list}
-
-            You MUST use these specific participant names when referring to people in your response.
-            Do NOT use generic terms like "a participant", "someone", "attendee", "Speaker 1", "Speaker 2", etc.
-            Always refer to people by their actual names (e.g., "John suggested..." not "A participant suggested...").
-            """
-        ).strip()
+        self.participant_instructions = build_participant_instructions(participants)

    async def identify_participants(self) -> None:
        """
@@ -377,18 +330,9 @@ class SummaryBuilder:
                participants_md = self.format_list_md(unique_participants)
                self.transcript += f"\n\n# Participants\n\n{participants_md}"

-                # Set instructions that will be automatically added to all prompts
-                participants_list = ", ".join(unique_participants)
-                self.participant_instructions = dedent(
-                    f"""
-                    # IMPORTANT: Participant Names
-                    The following participants are identified in this conversation: {participants_list}
-
-                    You MUST use these specific participant names when referring to people in your response.
-                    Do NOT use generic terms like "a participant", "someone", "attendee", "Speaker 1", "Speaker 2", etc.
-                    Always refer to people by their actual names (e.g., "John suggested..." not "A participant suggested...").
-                    """
-                ).strip()
+                self.participant_instructions = build_participant_instructions(
+                    unique_participants
+                )
            else:
                self.logger.warning("No participants identified in the transcript")

@@ -613,22 +557,7 @@ class SummaryBuilder:
    # ----------------------------------------------------------------------------

    def as_markdown(self) -> str:
-        lines: list[str] = []
-        if self.recap:
-            lines.append("# Quick recap")
-            lines.append("")
-            lines.append(self.recap)
-            lines.append("")
-
-        if self.summaries:
-            lines.append("# Summary")
-            lines.append("")
-            for summary in self.summaries:
-                lines.append(f"**{summary['subject']}**")
-                lines.append(summary["summary"])
-                lines.append("")
-
-        return "\n".join(lines)
+        return build_summary_markdown(self.recap, self.summaries)

    def format_list_md(self, data: list[str]) -> str:
        return "\n".join([f"- {item}" for item in data])
--- a/server/reflector/processors/transcript_topic_detector.py
+++ b/server/reflector/processors/transcript_topic_detector.py
@@ -1,35 +1,12 @@
-from textwrap import dedent
-
 from pydantic import AliasChoices, BaseModel, Field

 from reflector.llm import LLM
 from reflector.processors.base import Processor
+from reflector.processors.prompts import TOPIC_PROMPT
 from reflector.processors.types import TitleSummary, Transcript
 from reflector.settings import settings
 from reflector.utils.text import clean_title

-TOPIC_PROMPT = dedent(
-    """
-    Analyze the following transcript segment and extract the main topic being discussed.
-    Focus on the substantive content and ignore small talk or administrative chatter.
-
-    Create a title that:
-    - Captures the specific subject matter being discussed
-    - Is descriptive and self-explanatory
-    - Uses professional language
-    - Is specific rather than generic
-
-    For the summary:
-    - Summarize the key points in maximum two sentences
-    - Focus on what was discussed, decided, or accomplished
-    - Be concise but informative
-
-    <transcript>
-    {text}
-    </transcript>
-    """
-).strip()
-

 class TopicResponse(BaseModel):
    """Structured response for topic detection"""
--- a/server/reflector/services/transcript_process.py
+++ b/server/reflector/services/transcript_process.py
@@ -11,13 +11,19 @@ from typing import Literal, Union, assert_never

 import celery
 from celery.result import AsyncResult
+from hatchet_sdk.clients.rest.exceptions import ApiException
+from hatchet_sdk.clients.rest.models import V1TaskStatus

 from reflector.db.recordings import recordings_controller
-from reflector.db.transcripts import Transcript
+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.settings import settings
 from reflector.utils.string import NonEmptyString


@@ -37,6 +43,8 @@ class MultitrackProcessingConfig:
    transcript_id: NonEmptyString
    bucket_name: NonEmptyString
    track_keys: list[str]
+    recording_id: NonEmptyString | None = None
+    room_id: NonEmptyString | None = None
    mode: Literal["multitrack"] = "multitrack"


@@ -49,6 +57,7 @@ class ValidationOk:
    # transcript currently doesnt always have recording_id
    recording_id: NonEmptyString | None
    transcript_id: NonEmptyString
+    room_id: NonEmptyString | None = None


@dataclass
@@ -93,9 +102,11 @@ async def validate_transcript_for_processing(
    if transcript.locked:
        return ValidationLocked(detail="Recording is locked")

-    if transcript.status == "idle":
+    # hatchet is idempotent anyways + if it wasn't dispatched successfully
+    if transcript.status == "idle" and not settings.HATCHET_ENABLED:
        return ValidationNotReady(detail="Recording is not ready for processing")

+    # Check Celery tasks
    if task_is_scheduled_or_active(
        "reflector.pipelines.main_file_pipeline.task_pipeline_file_process",
        transcript_id=transcript.id,
@@ -105,8 +116,23 @@ async def validate_transcript_for_processing(
    ):
        return ValidationAlreadyScheduled(detail="already running")

+    if settings.HATCHET_ENABLED and transcript.workflow_run_id:
+        try:
+            status = await HatchetClientManager.get_workflow_run_status(
+                transcript.workflow_run_id
+            )
+            if status in (V1TaskStatus.RUNNING, V1TaskStatus.QUEUED):
+                return ValidationAlreadyScheduled(
+                    detail="Hatchet workflow already running"
+                )
+        except ApiException:
+            # Workflow might be gone (404) or API issue - allow processing
+            pass
+
    return ValidationOk(
-        recording_id=transcript.recording_id, transcript_id=transcript.id
+        recording_id=transcript.recording_id,
+        transcript_id=transcript.id,
+        room_id=transcript.room_id,
    )


@@ -116,6 +142,7 @@ async def prepare_transcript_processing(validation: ValidationOk) -> PrepareResu
    """
    bucket_name: str | None = None
    track_keys: list[str] | None = None
+    recording_id: str | None = validation.recording_id

    if validation.recording_id:
        recording = await recordings_controller.get_by_id(validation.recording_id)
@@ -137,6 +164,8 @@ async def prepare_transcript_processing(validation: ValidationOk) -> PrepareResu
            bucket_name=bucket_name,  # type: ignore (validated above)
            track_keys=track_keys,
            transcript_id=validation.transcript_id,
+            recording_id=recording_id,
+            room_id=validation.room_id,
        )

    return FileProcessingConfig(
@@ -144,8 +173,115 @@ async def prepare_transcript_processing(validation: ValidationOk) -> PrepareResu
    )


-def dispatch_transcript_processing(config: ProcessingConfig) -> AsyncResult:
+async def dispatch_transcript_processing(
+    config: ProcessingConfig, force: bool = False
+) -> AsyncResult | None:
+    """Dispatch transcript processing to appropriate backend (Hatchet or Celery).
+
+    Returns AsyncResult for Celery tasks, None for Hatchet workflows.
+    """
    if isinstance(config, MultitrackProcessingConfig):
+        # Check if room has use_hatchet=True (overrides env vars)
+        room_forces_hatchet = False
+        if config.room_id:
+            room = await rooms_controller.get_by_id(config.room_id)
+            room_forces_hatchet = room.use_hatchet if room else False
+
+        # Start durable workflow if enabled (Hatchet)
+        # or if room has use_hatchet=True
+        use_hatchet = settings.HATCHET_ENABLED or room_forces_hatchet
+
+        if room_forces_hatchet:
+            logger.info(
+                "Room forces Hatchet workflow",
+                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:
+                can_replay = await HatchetClientManager.can_replay(
+                    transcript.workflow_run_id
+                )
+                if can_replay:
+                    await HatchetClientManager.replay_workflow(
+                        transcript.workflow_run_id
+                    )
+                    logger.info(
+                        "Replaying Hatchet workflow",
+                        workflow_id=transcript.workflow_run_id,
+                    )
+                    return None
+                else:
+                    # Workflow exists but can't replay (CANCELLED, COMPLETED, etc.)
+                    # Log and proceed to start new workflow
+                    status = await HatchetClientManager.get_workflow_run_status(
+                        transcript.workflow_run_id
+                    )
+                    logger.info(
+                        "Old workflow not replayable, starting new",
+                        old_workflow_id=transcript.workflow_run_id,
+                        old_status=status.value,
+                    )
+
+            # Force: cancel old workflow if exists
+            if force and transcript and transcript.workflow_run_id:
+                await HatchetClientManager.cancel_workflow(transcript.workflow_run_id)
+                logger.info(
+                    "Cancelled old workflow (--force)",
+                    workflow_id=transcript.workflow_run_id,
+                )
+                await transcripts_controller.update(
+                    transcript, {"workflow_run_id": None}
+                )
+
+            # Re-fetch and check for concurrent dispatch (optimistic approach).
+            # No database lock - worst case is duplicate dispatch, but Hatchet
+            # workflows are idempotent so this is acceptable.
+            transcript = await transcripts_controller.get_by_id(config.transcript_id)
+            if transcript and transcript.workflow_run_id:
+                # Another process started a workflow between validation and now
+                try:
+                    status = await HatchetClientManager.get_workflow_run_status(
+                        transcript.workflow_run_id
+                    )
+                    if status in (V1TaskStatus.RUNNING, V1TaskStatus.QUEUED):
+                        logger.info(
+                            "Concurrent workflow detected, skipping dispatch",
+                            workflow_id=transcript.workflow_run_id,
+                        )
+                        return None
+                except ApiException:
+                    # Workflow might be gone (404) or API issue - proceed with new workflow
+                    pass
+
+            workflow_id = await HatchetClientManager.start_workflow(
+                workflow_name="DiarizationPipeline",
+                input_data={
+                    "recording_id": config.recording_id,
+                    "tracks": [{"s3_key": k} for k in config.track_keys],
+                    "bucket_name": config.bucket_name,
+                    "transcript_id": config.transcript_id,
+                    "room_id": config.room_id,
+                },
+                additional_metadata={
+                    "transcript_id": config.transcript_id,
+                    "recording_id": config.recording_id,
+                    "daily_recording_id": config.recording_id,
+                },
+            )
+
+            if transcript:
+                await transcripts_controller.update(
+                    transcript, {"workflow_run_id": workflow_id}
+                )
+
+            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,
--- a/server/reflector/settings.py
+++ b/server/reflector/settings.py
@@ -153,5 +153,19 @@ class Settings(BaseSettings):
    ZULIP_API_KEY: str | None = None
    ZULIP_BOT_EMAIL: str | None = None

+    # Durable workflow orchestration
+    # Provider: "hatchet" (or "none" to disable)
+    DURABLE_WORKFLOW_PROVIDER: str = "none"
+
+    # Hatchet workflow orchestration
+    HATCHET_CLIENT_TOKEN: str | None = None
+    HATCHET_CLIENT_TLS_STRATEGY: str = "none"  # none, tls, mtls
+    HATCHET_DEBUG: bool = False
+
+    @property
+    def HATCHET_ENABLED(self) -> bool:
+        """True if Hatchet is the active provider."""
+        return self.DURABLE_WORKFLOW_PROVIDER == "hatchet"
+

 settings = Settings()
--- a/server/reflector/tools/process_transcript.py
+++ b/server/reflector/tools/process_transcript.py
@@ -15,8 +15,11 @@ import time
 from typing import Callable

 from celery.result import AsyncResult
+from hatchet_sdk.clients.rest.models import V1TaskStatus

+from reflector.db import get_database
 from reflector.db.transcripts import Transcript, transcripts_controller
+from reflector.hatchet.client import HatchetClientManager
 from reflector.services.transcript_process import (
    FileProcessingConfig,
    MultitrackProcessingConfig,
@@ -34,24 +37,26 @@ async def process_transcript_inner(
    transcript: Transcript,
    on_validation: Callable[[ValidationResult], None],
    on_preprocess: Callable[[PrepareResult], None],
-) -> AsyncResult:
+    force: bool = False,
+) -> AsyncResult | None:
    validation = await validate_transcript_for_processing(transcript)
    on_validation(validation)
    config = await prepare_transcript_processing(validation)
    on_preprocess(config)
-    return dispatch_transcript_processing(config)
+    return await dispatch_transcript_processing(config, force=force)


-async def process_transcript(transcript_id: str, sync: bool = False) -> None:
+async def process_transcript(
+    transcript_id: str, sync: bool = False, force: bool = False
+) -> None:
    """
    Process a transcript by ID, auto-detecting multitrack vs file pipeline.

    Args:
        transcript_id: The transcript UUID
        sync: If True, wait for task completion. If False, dispatch and exit.
+        force: If True, cancel old workflow and start new (latest code). If False, replay failed workflow.
    """
-    from reflector.db import get_database
-
    database = get_database()
    await database.connect()

@@ -82,10 +87,42 @@ async def process_transcript(transcript_id: str, sync: bool = False) -> None:
                print(f"Dispatching file pipeline", file=sys.stderr)

        result = await process_transcript_inner(
-            transcript, on_validation=on_validation, on_preprocess=on_preprocess
+            transcript,
+            on_validation=on_validation,
+            on_preprocess=on_preprocess,
+            force=force,
        )

+        if result is None:
+            # Hatchet workflow dispatched
            if sync:
+                # Re-fetch transcript to get workflow_run_id
+                transcript = await transcripts_controller.get_by_id(transcript_id)
+                if not transcript or not transcript.workflow_run_id:
+                    print("Error: workflow_run_id not found", file=sys.stderr)
+                    sys.exit(1)
+
+                print("Waiting for Hatchet workflow...", file=sys.stderr)
+                while True:
+                    status = await HatchetClientManager.get_workflow_run_status(
+                        transcript.workflow_run_id
+                    )
+                    print(f"  Status: {status.value}", file=sys.stderr)
+
+                    if status == V1TaskStatus.COMPLETED:
+                        print("Workflow completed successfully", file=sys.stderr)
+                        break
+                    elif status in (V1TaskStatus.FAILED, V1TaskStatus.CANCELLED):
+                        print(f"Workflow failed: {status}", file=sys.stderr)
+                        sys.exit(1)
+
+                    await asyncio.sleep(5)
+            else:
+                print(
+                    "Task dispatched (use --sync to wait for completion)",
+                    file=sys.stderr,
+                )
+        elif sync:
            print("Waiting for task completion...", file=sys.stderr)
            while not result.ready():
                print(f"  Status: {result.state}", file=sys.stderr)
@@ -118,9 +155,16 @@ def main():
        action="store_true",
        help="Wait for task completion instead of just dispatching",
    )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="Cancel old workflow and start new (uses latest code instead of replaying)",
+    )

    args = parser.parse_args()
-    asyncio.run(process_transcript(args.transcript_id, sync=args.sync))
+    asyncio.run(
+        process_transcript(args.transcript_id, sync=args.sync, force=args.force)
+    )


 if __name__ == "__main__":
--- a/server/reflector/utils/audio_constants.py
+++ b/server/reflector/utils/audio_constants.py
@@ -0,0 +1,15 @@
+"""
+Shared audio processing constants.
+
+Used by both Hatchet workflows and Celery pipelines for consistent audio encoding.
+"""
+
+# Opus codec settings
+OPUS_STANDARD_SAMPLE_RATE = 48000
+OPUS_DEFAULT_BIT_RATE = 128000  # 128kbps for good speech quality
+
+# S3 presigned URL expiration
+PRESIGNED_URL_EXPIRATION_SECONDS = 7200  # 2 hours
+
+# Waveform visualization
+WAVEFORM_SEGMENTS = 255
--- a/server/reflector/utils/audio_mixdown.py
+++ b/server/reflector/utils/audio_mixdown.py
@@ -0,0 +1,227 @@
+"""
+Audio track mixdown utilities.
+
+Shared PyAV-based functions for mixing multiple audio tracks into a single output.
+Used by both Hatchet workflows and Celery pipelines.
+"""
+
+from fractions import Fraction
+
+import av
+from av.audio.resampler import AudioResampler
+
+
+def detect_sample_rate_from_tracks(track_urls: list[str], logger=None) -> int | None:
+    """Detect sample rate from first decodable audio frame.
+
+    Args:
+        track_urls: List of URLs to audio files (S3 presigned or local)
+        logger: Optional logger instance
+
+    Returns:
+        Sample rate in Hz, or None if no decodable frames found
+    """
+    for url in track_urls:
+        if not url:
+            continue
+        container = None
+        try:
+            container = av.open(url)
+            for frame in container.decode(audio=0):
+                return frame.sample_rate
+        except Exception:
+            continue
+        finally:
+            if container is not None:
+                container.close()
+    return None
+
+
+async def mixdown_tracks_pyav(
+    track_urls: list[str],
+    writer,
+    target_sample_rate: int,
+    offsets_seconds: list[float] | None = None,
+    logger=None,
+) -> None:
+    """Multi-track mixdown using PyAV filter graph (amix).
+
+    Builds a filter graph: N abuffer -> optional adelay -> amix -> aformat -> sink
+    Reads from S3 presigned URLs or local files, pushes mixed frames to writer.
+
+    Args:
+        track_urls: List of URLs to audio tracks (S3 presigned or local)
+        writer: AudioFileWriterProcessor instance with async push() method
+        target_sample_rate: Sample rate for output (Hz)
+        offsets_seconds: Optional per-track delays in seconds for alignment.
+            If provided, must have same length as track_urls. Delays are relative
+            to the minimum offset (earliest track has delay=0).
+        logger: Optional logger instance
+
+    Raises:
+        ValueError: If offsets_seconds length doesn't match track_urls,
+            no valid tracks provided, or no containers can be opened
+    """
+    if offsets_seconds is not None and len(offsets_seconds) != len(track_urls):
+        raise ValueError(
+            f"offsets_seconds length ({len(offsets_seconds)}) must match track_urls ({len(track_urls)})"
+        )
+
+    valid_track_urls = [url for url in track_urls if url]
+    if not valid_track_urls:
+        if logger:
+            logger.error("Mixdown failed - no valid track URLs provided")
+        raise ValueError("Mixdown failed: No valid track URLs")
+
+    # Calculate per-input delays if offsets provided
+    input_offsets_seconds = None
+    if offsets_seconds is not None:
+        input_offsets_seconds = [
+            offsets_seconds[i] for i, url in enumerate(track_urls) if url
+        ]
+
+    # Build PyAV filter graph:
+    # N abuffer (s32/stereo)
+    #   -> optional adelay per input (for alignment)
+    #   -> amix (s32)
+    #   -> aformat(s16)
+    #   -> sink
+    graph = av.filter.Graph()
+    inputs = []
+
+    for idx, url in enumerate(valid_track_urls):
+        args = (
+            f"time_base=1/{target_sample_rate}:"
+            f"sample_rate={target_sample_rate}:"
+            f"sample_fmt=s32:"
+            f"channel_layout=stereo"
+        )
+        in_ctx = graph.add("abuffer", args=args, name=f"in{idx}")
+        inputs.append(in_ctx)
+
+    if not inputs:
+        if logger:
+            logger.error("Mixdown failed - no valid inputs for graph")
+        raise ValueError("Mixdown failed: No valid inputs for filter graph")
+
+    mixer = graph.add("amix", args=f"inputs={len(inputs)}:normalize=0", name="mix")
+
+    fmt = graph.add(
+        "aformat",
+        args=f"sample_fmts=s32:channel_layouts=stereo:sample_rates={target_sample_rate}",
+        name="fmt",
+    )
+
+    sink = graph.add("abuffersink", name="out")
+
+    # Optional per-input delay before mixing
+    delays_ms: list[int] = []
+    if input_offsets_seconds is not None:
+        base = min(input_offsets_seconds) if input_offsets_seconds else 0.0
+        delays_ms = [
+            max(0, int(round((o - base) * 1000))) for o in input_offsets_seconds
+        ]
+    else:
+        delays_ms = [0 for _ in inputs]
+
+    for idx, in_ctx in enumerate(inputs):
+        delay_ms = delays_ms[idx] if idx < len(delays_ms) else 0
+        if delay_ms > 0:
+            # adelay requires one value per channel; use same for stereo
+            adelay = graph.add(
+                "adelay",
+                args=f"delays={delay_ms}|{delay_ms}:all=1",
+                name=f"delay{idx}",
+            )
+            in_ctx.link_to(adelay)
+            adelay.link_to(mixer, 0, idx)
+        else:
+            in_ctx.link_to(mixer, 0, idx)
+
+    mixer.link_to(fmt)
+    fmt.link_to(sink)
+    graph.configure()
+
+    containers = []
+    try:
+        # Open all containers with cleanup guaranteed
+        for i, url in enumerate(valid_track_urls):
+            try:
+                c = av.open(
+                    url,
+                    options={
+                        # S3 streaming options
+                        "reconnect": "1",
+                        "reconnect_streamed": "1",
+                        "reconnect_delay_max": "5",
+                    },
+                )
+                containers.append(c)
+            except Exception as e:
+                if logger:
+                    logger.warning(
+                        "Mixdown: failed to open container from URL",
+                        input=i,
+                        url=url,
+                        error=str(e),
+                    )
+
+        if not containers:
+            if logger:
+                logger.error("Mixdown failed - no valid containers opened")
+            raise ValueError("Mixdown failed: Could not open any track containers")
+
+        decoders = [c.decode(audio=0) for c in containers]
+        active = [True] * len(decoders)
+        resamplers = [
+            AudioResampler(format="s32", layout="stereo", rate=target_sample_rate)
+            for _ in decoders
+        ]
+
+        while any(active):
+            for i, (dec, is_active) in enumerate(zip(decoders, active)):
+                if not is_active:
+                    continue
+                try:
+                    frame = next(dec)
+                except StopIteration:
+                    active[i] = False
+                    # Signal end of stream to filter graph
+                    inputs[i].push(None)
+                    continue
+
+                if frame.sample_rate != target_sample_rate:
+                    continue
+                out_frames = resamplers[i].resample(frame) or []
+                for rf in out_frames:
+                    rf.sample_rate = target_sample_rate
+                    rf.time_base = Fraction(1, target_sample_rate)
+                    inputs[i].push(rf)
+
+                while True:
+                    try:
+                        mixed = sink.pull()
+                    except Exception:
+                        break
+                    mixed.sample_rate = target_sample_rate
+                    mixed.time_base = Fraction(1, target_sample_rate)
+                    await writer.push(mixed)
+
+        # Flush remaining frames from filter graph
+        while True:
+            try:
+                mixed = sink.pull()
+            except Exception:
+                break
+            mixed.sample_rate = target_sample_rate
+            mixed.time_base = Fraction(1, target_sample_rate)
+            await writer.push(mixed)
+
+    finally:
+        # Cleanup all containers, even if processing failed
+        for c in containers:
+            if c is not None:
+                try:
+                    c.close()
+                except Exception:
+                    pass  # Best effort cleanup
--- a/server/reflector/utils/audio_padding.py
+++ b/server/reflector/utils/audio_padding.py
@@ -0,0 +1,186 @@
+"""
+Audio track padding utilities.
+
+Shared PyAV-based functions for extracting stream metadata and applying
+silence padding to audio tracks. Used by both Hatchet workflows and Celery pipelines.
+"""
+
+import math
+from fractions import Fraction
+
+import av
+from av.audio.resampler import AudioResampler
+
+from reflector.utils.audio_constants import (
+    OPUS_DEFAULT_BIT_RATE,
+    OPUS_STANDARD_SAMPLE_RATE,
+)
+
+
+def extract_stream_start_time_from_container(
+    container,
+    track_idx: int,
+    logger=None,
+) -> float:
+    """Extract meeting-relative start time from WebM stream metadata.
+
+    Uses PyAV to read stream.start_time from WebM container.
+    More accurate than filename timestamps by ~209ms due to network/encoding delays.
+
+    Args:
+        container: PyAV container opened from audio file/URL
+        track_idx: Track index for logging context
+        logger: Optional logger instance (structlog or stdlib compatible)
+
+    Returns:
+        Start time in seconds (0.0 if extraction fails)
+    """
+    start_time_seconds = 0.0
+    try:
+        audio_streams = [s for s in container.streams if s.type == "audio"]
+        stream = audio_streams[0] if audio_streams else container.streams[0]
+
+        # 1) Try stream-level start_time (most reliable for Daily.co tracks)
+        if stream.start_time is not None and stream.time_base is not None:
+            start_time_seconds = float(stream.start_time * stream.time_base)
+
+        # 2) Fallback to container-level start_time (in av.time_base units)
+        if (start_time_seconds <= 0) and (container.start_time is not None):
+            start_time_seconds = float(container.start_time * av.time_base)
+
+        # 3) Fallback to first packet DTS in stream.time_base
+        if start_time_seconds <= 0:
+            for packet in container.demux(stream):
+                if packet.dts is not None:
+                    start_time_seconds = float(packet.dts * stream.time_base)
+                    break
+    except Exception as e:
+        if logger:
+            logger.warning(
+                "PyAV metadata read failed; assuming 0 start_time",
+                track_idx=track_idx,
+                error=str(e),
+            )
+        start_time_seconds = 0.0
+
+    if logger:
+        logger.info(
+            f"Track {track_idx} stream metadata: start_time={start_time_seconds:.3f}s",
+            track_idx=track_idx,
+        )
+    return start_time_seconds
+
+
+def apply_audio_padding_to_file(
+    in_container,
+    output_path: str,
+    start_time_seconds: float,
+    track_idx: int,
+    logger=None,
+) -> None:
+    """Apply silence padding to audio track using PyAV filter graph.
+
+    Uses adelay filter to prepend silence, aligning track to meeting start time.
+    Output is WebM/Opus format.
+
+    Args:
+        in_container: PyAV container opened from source audio
+        output_path: Path for output WebM file
+        start_time_seconds: Amount of silence to prepend (in seconds)
+        track_idx: Track index for logging context
+        logger: Optional logger instance (structlog or stdlib compatible)
+
+    Raises:
+        Exception: If no audio stream found or PyAV processing fails
+    """
+    delay_ms = math.floor(start_time_seconds * 1000)
+
+    if logger:
+        logger.info(
+            f"Padding track {track_idx} with {delay_ms}ms delay using PyAV",
+            track_idx=track_idx,
+            delay_ms=delay_ms,
+        )
+
+    try:
+        with av.open(output_path, "w", format="webm") as out_container:
+            in_stream = next(
+                (s for s in in_container.streams if s.type == "audio"), None
+            )
+            if in_stream is None:
+                raise Exception("No audio stream in input")
+
+            out_stream = out_container.add_stream(
+                "libopus", rate=OPUS_STANDARD_SAMPLE_RATE
+            )
+            out_stream.bit_rate = OPUS_DEFAULT_BIT_RATE
+            graph = av.filter.Graph()
+
+            abuf_args = (
+                f"time_base=1/{OPUS_STANDARD_SAMPLE_RATE}:"
+                f"sample_rate={OPUS_STANDARD_SAMPLE_RATE}:"
+                f"sample_fmt=s16:"
+                f"channel_layout=stereo"
+            )
+            src = graph.add("abuffer", args=abuf_args, name="src")
+            aresample_f = graph.add("aresample", args="async=1", name="ares")
+            # adelay requires one delay value per channel separated by '|'
+            delays_arg = f"{delay_ms}|{delay_ms}"
+            adelay_f = graph.add(
+                "adelay", args=f"delays={delays_arg}:all=1", name="delay"
+            )
+            sink = graph.add("abuffersink", name="sink")
+
+            src.link_to(aresample_f)
+            aresample_f.link_to(adelay_f)
+            adelay_f.link_to(sink)
+            graph.configure()
+
+            resampler = AudioResampler(
+                format="s16", layout="stereo", rate=OPUS_STANDARD_SAMPLE_RATE
+            )
+
+            # Decode -> resample -> push through graph -> encode Opus
+            for frame in in_container.decode(in_stream):
+                out_frames = resampler.resample(frame) or []
+                for rframe in out_frames:
+                    rframe.sample_rate = OPUS_STANDARD_SAMPLE_RATE
+                    rframe.time_base = Fraction(1, OPUS_STANDARD_SAMPLE_RATE)
+                    src.push(rframe)
+
+                    while True:
+                        try:
+                            f_out = sink.pull()
+                        except Exception:
+                            break
+                        f_out.sample_rate = OPUS_STANDARD_SAMPLE_RATE
+                        f_out.time_base = Fraction(1, OPUS_STANDARD_SAMPLE_RATE)
+                        for packet in out_stream.encode(f_out):
+                            out_container.mux(packet)
+
+            # Flush remaining frames from filter graph
+            src.push(None)
+            while True:
+                try:
+                    f_out = sink.pull()
+                except Exception:
+                    break
+                f_out.sample_rate = OPUS_STANDARD_SAMPLE_RATE
+                f_out.time_base = Fraction(1, OPUS_STANDARD_SAMPLE_RATE)
+                for packet in out_stream.encode(f_out):
+                    out_container.mux(packet)
+
+            # Flush encoder
+            for packet in out_stream.encode(None):
+                out_container.mux(packet)
+
+    except Exception as e:
+        if logger:
+            logger.error(
+                "PyAV padding failed for track",
+                track_idx=track_idx,
+                delay_ms=delay_ms,
+                error=str(e),
+                exc_info=True,
+            )
+        raise
--- a/server/reflector/utils/common.py
+++ b/server/reflector/utils/common.py
@@ -0,0 +1,4 @@
+def assert_not_none[T](value: T | None, message: str = "Value is None") -> T:
+    if value is None:
+        raise ValueError(message)
+    return value
--- a/server/reflector/utils/string.py
+++ b/server/reflector/utils/string.py
@@ -2,6 +2,17 @@ from typing import Annotated, TypeVar

 from pydantic import Field, TypeAdapter, constr

+T_NotNone = TypeVar("T_NotNone")
+
+
+def assert_not_none(
+    value: T_NotNone | None, message: str = "Value is None"
+) -> T_NotNone:
+    if value is None:
+        raise ValueError(message)
+    return value
+
+
 NonEmptyStringBase = constr(min_length=1, strip_whitespace=False)
 NonEmptyString = Annotated[
    NonEmptyStringBase,
@@ -23,10 +34,18 @@ def try_parse_non_empty_string(s: str) -> NonEmptyString | None:
    return parse_non_empty_string(s)


-T = TypeVar("T", bound=str)
+T_Str = TypeVar("T_Str", bound=str)


-def assert_equal[T](s1: T, s2: T) -> T:
+def assert_equal(s1: T_Str, s2: T_Str) -> T_Str:
    if s1 != s2:
        raise ValueError(f"assert_equal: {s1} != {s2}")
    return s1
+
+
+def assert_non_none_and_non_empty(
+    value: str | None, error: str | None = None
+) -> NonEmptyString:
+    return parse_non_empty_string(
+        assert_not_none(value, error or "Value is None"), error
+    )
--- a/server/reflector/utils/transcript_constants.py
+++ b/server/reflector/utils/transcript_constants.py
@@ -0,0 +1,8 @@
+"""
+Shared transcript processing constants.
+
+Used by both Hatchet workflows and Celery pipelines for consistent processing.
+"""
+
+# Topic detection: number of words per chunk for topic extraction
+TOPIC_CHUNK_WORD_COUNT = 300
--- a/server/reflector/video_platforms/daily.py
+++ b/server/reflector/video_platforms/daily.py
@@ -181,11 +181,17 @@ class DailyClient(VideoPlatformClient):
        enable_recording_ui: bool,
        user_id: NonEmptyString | None = None,
        is_owner: bool = False,
+        max_recording_duration_seconds: int | None = None,
    ) -> NonEmptyString:
+        start_cloud_recording_opts = None
+        if start_cloud_recording and max_recording_duration_seconds:
+            start_cloud_recording_opts = {"maxDuration": max_recording_duration_seconds}
+
        properties = MeetingTokenProperties(
            room_name=room_name,
            user_id=user_id,
            start_cloud_recording=start_cloud_recording,
+            start_cloud_recording_opts=start_cloud_recording_opts,
            enable_recording_ui=enable_recording_ui,
            is_owner=is_owner,
        )
--- a/server/reflector/views/rooms.py
+++ b/server/reflector/views/rooms.py
@@ -44,6 +44,7 @@ class Room(BaseModel):
    ics_last_sync: Optional[datetime] = None
    ics_last_etag: Optional[str] = None
    platform: Platform
+    skip_consent: bool = False


 class RoomDetails(Room):
@@ -90,6 +91,7 @@ class CreateRoom(BaseModel):
    ics_fetch_interval: int = 300
    ics_enabled: bool = False
    platform: Platform
+    skip_consent: bool = False


 class UpdateRoom(BaseModel):
@@ -108,6 +110,7 @@ class UpdateRoom(BaseModel):
    ics_fetch_interval: Optional[int] = None
    ics_enabled: Optional[bool] = None
    platform: Optional[Platform] = None
+    skip_consent: Optional[bool] = None


 class CreateRoomMeeting(BaseModel):
@@ -249,6 +252,7 @@ async def rooms_create(
        ics_fetch_interval=room.ics_fetch_interval,
        ics_enabled=room.ics_enabled,
        platform=room.platform,
+        skip_consent=room.skip_consent,
    )


@@ -567,12 +571,26 @@ async def rooms_join_meeting(

    if meeting.platform == "daily" and user_id is not None:
        client = create_platform_client(meeting.platform)
+        # Show Daily's built-in recording UI when:
+        # - local recording (user controls when to record), OR
+        # - cloud recording with consent disabled (skip_consent=True)
+        # Hide it when cloud recording with consent enabled (we show custom consent UI)
+        enable_recording_ui = meeting.recording_type == "local" or (
+            meeting.recording_type == "cloud" and room.skip_consent
+        )
+        end_date = meeting.end_date
+        if end_date.tzinfo is None:
+            end_date = end_date.replace(tzinfo=timezone.utc)
+        remaining_seconds = min(
+            3 * 60 * 60, int((end_date - current_time).total_seconds())
+        )
        token = await client.create_meeting_token(
            meeting.room_name,
            start_cloud_recording=meeting.recording_type == "cloud",
-            enable_recording_ui=meeting.recording_type == "local",
+            enable_recording_ui=enable_recording_ui,
            user_id=user_id,
            is_owner=user_id == room.user_id,
+            max_recording_duration_seconds=remaining_seconds,
        )
        meeting = meeting.model_copy()
        meeting.room_url = add_query_param(meeting.room_url, "t", token)
--- a/server/reflector/views/transcripts_process.py
+++ b/server/reflector/views/transcripts_process.py
@@ -50,5 +50,5 @@ async def transcript_process(
    if isinstance(config, ProcessError):
        raise HTTPException(status_code=500, detail=config.detail)
    else:
-        dispatch_transcript_processing(config)
+        await dispatch_transcript_processing(config)
        return ProcessStatus(status="ok")
--- a/server/reflector/worker/healthcheck.py
+++ b/server/reflector/worker/healthcheck.py
@@ -7,6 +7,12 @@ from reflector.settings import settings
 logger = structlog.get_logger(__name__)


+@shared_task(name="celery.ping")
+def celery_ping():
+    """Compatibility task for Celery 5.x - celery.ping was removed but monitoring tools still call it."""
+    return "pong"
+
+
@shared_task
 def healthcheck_ping():
    url = settings.HEALTHCHECK_URL
--- a/server/reflector/worker/process.py
+++ b/server/reflector/worker/process.py
@@ -24,6 +24,7 @@ from reflector.db.transcripts import (
    SourceKind,
    transcripts_controller,
 )
+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 (
@@ -286,6 +287,45 @@ async def _process_multitrack_recording_inner(
            room_id=room.id,
        )

+    # Start durable workflow if enabled (Hatchet) or room overrides it
+    durable_started = False
+    use_hatchet = settings.HATCHET_ENABLED or (room and room.use_hatchet)
+
+    if room and room.use_hatchet and not settings.HATCHET_ENABLED:
+        logger.info(
+            "Room forces Hatchet workflow",
+            room_id=room.id,
+            transcript_id=transcript.id,
+        )
+
+    if use_hatchet:
+        workflow_id = await HatchetClientManager.start_workflow(
+            workflow_name="DiarizationPipeline",
+            input_data={
+                "recording_id": recording_id,
+                "tracks": [{"s3_key": k} for k in filter_cam_audio_tracks(track_keys)],
+                "bucket_name": bucket_name,
+                "transcript_id": transcript.id,
+                "room_id": room.id,
+            },
+            additional_metadata={
+                "transcript_id": transcript.id,
+                "recording_id": recording_id,
+                "daily_recording_id": recording_id,
+            },
+        )
+        logger.info(
+            "Started Hatchet workflow",
+            workflow_id=workflow_id,
+            transcript_id=transcript.id,
+        )
+
+        await transcripts_controller.update(
+            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,
@@ -570,12 +610,12 @@ async def process_meetings():
                client = create_platform_client(meeting.platform)
                room_sessions = await client.get_room_sessions(meeting.room_name)

-                has_active_sessions = room_sessions and any(
-                    s.ended_at is None for s in room_sessions
+                has_active_sessions = bool(
+                    room_sessions and any(s.ended_at is None for s in room_sessions)
                )
                has_had_sessions = bool(room_sessions)
                logger_.info(
-                    f"found {has_active_sessions} active sessions, had {has_had_sessions}"
+                    f"has_active_sessions={has_active_sessions}, has_had_sessions={has_had_sessions}"
                )

                if has_active_sessions:
@@ -772,6 +812,11 @@ async def reprocess_failed_daily_recordings():
                )
                continue

+            # Fetch room to check use_hatchet flag
+            room = None
+            if meeting.room_id:
+                room = await rooms_controller.get_by_id(meeting.room_id)
+
            transcript = None
            try:
                transcript = await transcripts_controller.get_by_recording_id(
@@ -791,8 +836,49 @@ async def reprocess_failed_daily_recordings():
                )
                continue

+            use_hatchet = settings.HATCHET_ENABLED or (room and room.use_hatchet)
+
+            if use_hatchet:
+                # Hatchet requires a transcript for workflow_run_id tracking
+                if not transcript:
+                    logger.warning(
+                        "No transcript for Hatchet reprocessing, skipping",
+                        recording_id=recording.id,
+                    )
+                    continue
+
+                workflow_id = await HatchetClientManager.start_workflow(
+                    workflow_name="DiarizationPipeline",
+                    input_data={
+                        "recording_id": recording.id,
+                        "tracks": [
+                            {"s3_key": k}
+                            for k in filter_cam_audio_tracks(recording.track_keys)
+                        ],
+                        "bucket_name": bucket_name,
+                        "transcript_id": transcript.id,
+                        "room_id": room.id if room else None,
+                    },
+                    additional_metadata={
+                        "transcript_id": transcript.id,
+                        "recording_id": recording.id,
+                        "reprocess": True,
+                    },
+                )
+                await transcripts_controller.update(
+                    transcript, {"workflow_run_id": workflow_id}
+                )
+
                logger.info(
-                "Queueing Daily recording for reprocessing",
+                    "Queued Daily recording for Hatchet reprocessing",
+                    recording_id=recording.id,
+                    workflow_id=workflow_id,
+                    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),
@@ -805,6 +891,7 @@ async def reprocess_failed_daily_recordings():
                    recording_id=recording.id,
                    track_keys=recording.track_keys,
                )
+
            reprocessed_count += 1

    except Exception as e:
--- a/server/reflector/zulip.py
+++ b/server/reflector/zulip.py
@@ -3,7 +3,8 @@ from urllib.parse import urlparse

 import httpx

-from reflector.db.transcripts import Transcript
+from reflector.db.rooms import rooms_controller
+from reflector.db.transcripts import Transcript, transcripts_controller
 from reflector.settings import settings


@@ -113,6 +114,49 @@ def get_zulip_message(transcript: Transcript, include_topics: bool):
    return message


+async def post_transcript_notification(transcript: Transcript) -> int | None:
+    """Post or update transcript notification in Zulip.
+
+    Uses transcript.room_id directly (Hatchet flow).
+    Celery's pipeline_post_to_zulip uses recording→meeting→room path instead.
+    DUPLICATION NOTE: This function will stay when we use Celery no more, and Celery one will be removed.
+    """
+    if not transcript.room_id:
+        return None
+
+    room = await rooms_controller.get_by_id(transcript.room_id)
+    if not room or not room.zulip_stream or not room.zulip_auto_post:
+        return None
+
+    message = get_zulip_message(transcript=transcript, include_topics=True)
+    message_updated = False
+
+    if transcript.zulip_message_id:
+        try:
+            await update_zulip_message(
+                transcript.zulip_message_id,
+                room.zulip_stream,
+                room.zulip_topic,
+                message,
+            )
+            message_updated = True
+        except Exception:
+            pass
+
+    if not message_updated:
+        response = await send_message_to_zulip(
+            room.zulip_stream, room.zulip_topic, message
+        )
+        message_id = response.get("id")
+        if message_id:
+            await transcripts_controller.update(
+                transcript, {"zulip_message_id": message_id}
+            )
+        return message_id
+
+    return transcript.zulip_message_id
+
+
 def extract_domain(url: str) -> str:
    return urlparse(url).netloc

--- a/server/runserver.sh
+++ b/server/runserver.sh
@@ -7,6 +7,8 @@ elif [ "${ENTRYPOINT}" = "worker" ]; then
    uv run celery -A reflector.worker.app worker --loglevel=info
 elif [ "${ENTRYPOINT}" = "beat" ]; then
    uv run celery -A reflector.worker.app beat --loglevel=info
+elif [ "${ENTRYPOINT}" = "hatchet-worker" ]; then
+    uv run python -m reflector.hatchet.run_workers
 else
    echo "Unknown command"
 fi
--- a/server/tests/conftest.py
+++ b/server/tests/conftest.py
@@ -527,6 +527,22 @@ def fake_mp3_upload():
        yield


+@pytest.fixture(autouse=True)
+def reset_hatchet_client():
+    """Reset HatchetClientManager singleton before and after each test.
+
+    This ensures test isolation - each test starts with a fresh client state.
+    The fixture is autouse=True so it applies to all tests automatically.
+    """
+    from reflector.hatchet.client import HatchetClientManager
+
+    # Reset before test
+    HatchetClientManager.reset()
+    yield
+    # Reset after test to clean up
+    HatchetClientManager.reset()
+
+
@pytest.fixture
 async def fake_transcript_with_topics(tmpdir, client):
    import shutil
--- a/server/tests/test_hatchet_client.py
+++ b/server/tests/test_hatchet_client.py
@@ -0,0 +1,54 @@
+"""
+Tests for HatchetClientManager error handling and validation.
+
+Only tests that catch real bugs - not mock verification tests.
+
+Note: The `reset_hatchet_client` fixture (autouse=True in conftest.py)
+automatically resets the singleton before and after each test.
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_hatchet_client_can_replay_handles_exception():
+    """Test can_replay returns False when status check fails.
+
+    Useful: Ensures network/API errors don't crash the system and
+    gracefully allow reprocessing when workflow state is unknown.
+    """
+    from reflector.hatchet.client import HatchetClientManager
+
+    with patch("reflector.hatchet.client.settings") as mock_settings:
+        mock_settings.HATCHET_CLIENT_TOKEN = "test-token"
+        mock_settings.HATCHET_DEBUG = False
+
+        with patch("reflector.hatchet.client.Hatchet") as mock_hatchet_class:
+            mock_client = MagicMock()
+            mock_hatchet_class.return_value = mock_client
+
+            mock_client.runs.aio_get_status = AsyncMock(
+                side_effect=Exception("Network error")
+            )
+
+            can_replay = await HatchetClientManager.can_replay("workflow-123")
+
+            # Should return False on error (workflow might be gone)
+            assert can_replay is False
+
+
+def test_hatchet_client_raises_without_token():
+    """Test that get_client raises ValueError without token.
+
+    Useful: Catches if someone removes the token validation,
+    which would cause cryptic errors later.
+    """
+    from reflector.hatchet.client import HatchetClientManager
+
+    with patch("reflector.hatchet.client.settings") as mock_settings:
+        mock_settings.HATCHET_CLIENT_TOKEN = None
+
+        with pytest.raises(ValueError, match="HATCHET_CLIENT_TOKEN must be set"):
+            HatchetClientManager.get_client()
--- a/server/tests/test_hatchet_dispatch.py
+++ b/server/tests/test_hatchet_dispatch.py
@@ -0,0 +1,398 @@
+"""
+Tests for Hatchet workflow dispatch and routing logic.
+
+These tests verify:
+1. Routing to Hatchet when HATCHET_ENABLED=True
+2. Replay logic for failed workflows
+3. Force flag to cancel and restart
+4. Validation prevents concurrent workflows
+"""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from hatchet_sdk.clients.rest.exceptions import ApiException
+from hatchet_sdk.clients.rest.models import V1TaskStatus
+
+from reflector.db.transcripts import Transcript
+
+
+@pytest.mark.usefixtures("setup_database")
+@pytest.mark.asyncio
+async def test_hatchet_validation_blocks_running_workflow():
+    """Test that validation blocks reprocessing when workflow is running."""
+    from reflector.services.transcript_process import (
+        ValidationAlreadyScheduled,
+        validate_transcript_for_processing,
+    )
+
+    mock_transcript = Transcript(
+        id="test-transcript-id",
+        name="Test",
+        status="processing",
+        source_kind="room",
+        workflow_run_id="running-workflow-123",
+    )
+
+    with patch("reflector.services.transcript_process.settings") as mock_settings:
+        mock_settings.HATCHET_ENABLED = True
+
+        with patch(
+            "reflector.services.transcript_process.HatchetClientManager"
+        ) as mock_hatchet:
+            mock_hatchet.get_workflow_run_status = AsyncMock(
+                return_value=V1TaskStatus.RUNNING
+            )
+
+            with patch(
+                "reflector.services.transcript_process.task_is_scheduled_or_active"
+            ) as mock_celery_check:
+                mock_celery_check.return_value = False
+
+                result = await validate_transcript_for_processing(mock_transcript)
+
+            assert isinstance(result, ValidationAlreadyScheduled)
+            assert "running" in result.detail.lower()
+
+
+@pytest.mark.usefixtures("setup_database")
+@pytest.mark.asyncio
+async def test_hatchet_validation_blocks_queued_workflow():
+    """Test that validation blocks reprocessing when workflow is queued."""
+    from reflector.services.transcript_process import (
+        ValidationAlreadyScheduled,
+        validate_transcript_for_processing,
+    )
+
+    mock_transcript = Transcript(
+        id="test-transcript-id",
+        name="Test",
+        status="processing",
+        source_kind="room",
+        workflow_run_id="queued-workflow-123",
+    )
+
+    with patch("reflector.services.transcript_process.settings") as mock_settings:
+        mock_settings.HATCHET_ENABLED = True
+
+        with patch(
+            "reflector.services.transcript_process.HatchetClientManager"
+        ) as mock_hatchet:
+            mock_hatchet.get_workflow_run_status = AsyncMock(
+                return_value=V1TaskStatus.QUEUED
+            )
+
+            with patch(
+                "reflector.services.transcript_process.task_is_scheduled_or_active"
+            ) as mock_celery_check:
+                mock_celery_check.return_value = False
+
+                result = await validate_transcript_for_processing(mock_transcript)
+
+            assert isinstance(result, ValidationAlreadyScheduled)
+
+
+@pytest.mark.usefixtures("setup_database")
+@pytest.mark.asyncio
+async def test_hatchet_validation_allows_failed_workflow():
+    """Test that validation allows reprocessing when workflow has failed."""
+    from reflector.services.transcript_process import (
+        ValidationOk,
+        validate_transcript_for_processing,
+    )
+
+    mock_transcript = Transcript(
+        id="test-transcript-id",
+        name="Test",
+        status="error",
+        source_kind="room",
+        workflow_run_id="failed-workflow-123",
+        recording_id="test-recording-id",
+    )
+
+    with patch("reflector.services.transcript_process.settings") as mock_settings:
+        mock_settings.HATCHET_ENABLED = True
+
+        with patch(
+            "reflector.services.transcript_process.HatchetClientManager"
+        ) as mock_hatchet:
+            mock_hatchet.get_workflow_run_status = AsyncMock(
+                return_value=V1TaskStatus.FAILED
+            )
+
+            with patch(
+                "reflector.services.transcript_process.task_is_scheduled_or_active"
+            ) as mock_celery_check:
+                mock_celery_check.return_value = False
+
+                result = await validate_transcript_for_processing(mock_transcript)
+
+            assert isinstance(result, ValidationOk)
+            assert result.transcript_id == "test-transcript-id"
+
+
+@pytest.mark.usefixtures("setup_database")
+@pytest.mark.asyncio
+async def test_hatchet_validation_allows_completed_workflow():
+    """Test that validation allows reprocessing when workflow has completed."""
+    from reflector.services.transcript_process import (
+        ValidationOk,
+        validate_transcript_for_processing,
+    )
+
+    mock_transcript = Transcript(
+        id="test-transcript-id",
+        name="Test",
+        status="ended",
+        source_kind="room",
+        workflow_run_id="completed-workflow-123",
+        recording_id="test-recording-id",
+    )
+
+    with patch("reflector.services.transcript_process.settings") as mock_settings:
+        mock_settings.HATCHET_ENABLED = True
+
+        with patch(
+            "reflector.services.transcript_process.HatchetClientManager"
+        ) as mock_hatchet:
+            mock_hatchet.get_workflow_run_status = AsyncMock(
+                return_value=V1TaskStatus.COMPLETED
+            )
+
+            with patch(
+                "reflector.services.transcript_process.task_is_scheduled_or_active"
+            ) as mock_celery_check:
+                mock_celery_check.return_value = False
+
+                result = await validate_transcript_for_processing(mock_transcript)
+
+            assert isinstance(result, ValidationOk)
+
+
+@pytest.mark.usefixtures("setup_database")
+@pytest.mark.asyncio
+async def test_hatchet_validation_allows_when_status_check_fails():
+    """Test that validation allows reprocessing when status check fails (workflow might be gone)."""
+    from reflector.services.transcript_process import (
+        ValidationOk,
+        validate_transcript_for_processing,
+    )
+
+    mock_transcript = Transcript(
+        id="test-transcript-id",
+        name="Test",
+        status="error",
+        source_kind="room",
+        workflow_run_id="old-workflow-123",
+        recording_id="test-recording-id",
+    )
+
+    with patch("reflector.services.transcript_process.settings") as mock_settings:
+        mock_settings.HATCHET_ENABLED = True
+
+        with patch(
+            "reflector.services.transcript_process.HatchetClientManager"
+        ) as mock_hatchet:
+            # Status check fails (workflow might be deleted)
+            mock_hatchet.get_workflow_run_status = AsyncMock(
+                side_effect=ApiException("Workflow not found")
+            )
+
+            with patch(
+                "reflector.services.transcript_process.task_is_scheduled_or_active"
+            ) as mock_celery_check:
+                mock_celery_check.return_value = False
+
+                result = await validate_transcript_for_processing(mock_transcript)
+
+            # Should allow processing when we can't get status
+            assert isinstance(result, ValidationOk)
+
+
+@pytest.mark.usefixtures("setup_database")
+@pytest.mark.asyncio
+async def test_hatchet_validation_skipped_when_no_workflow_id():
+    """Test that Hatchet validation is skipped when transcript has no workflow_run_id."""
+    from reflector.services.transcript_process import (
+        ValidationOk,
+        validate_transcript_for_processing,
+    )
+
+    mock_transcript = Transcript(
+        id="test-transcript-id",
+        name="Test",
+        status="uploaded",
+        source_kind="room",
+        workflow_run_id=None,  # No workflow yet
+        recording_id="test-recording-id",
+    )
+
+    with patch("reflector.services.transcript_process.settings") as mock_settings:
+        mock_settings.HATCHET_ENABLED = True
+
+        with patch(
+            "reflector.services.transcript_process.HatchetClientManager"
+        ) as mock_hatchet:
+            # Should not be called
+            mock_hatchet.get_workflow_run_status = AsyncMock()
+
+            with patch(
+                "reflector.services.transcript_process.task_is_scheduled_or_active"
+            ) as mock_celery_check:
+                mock_celery_check.return_value = False
+
+                result = await validate_transcript_for_processing(mock_transcript)
+
+            # Should not check Hatchet status
+            mock_hatchet.get_workflow_run_status.assert_not_called()
+            assert isinstance(result, ValidationOk)
+
+
+@pytest.mark.usefixtures("setup_database")
+@pytest.mark.asyncio
+async def test_hatchet_validation_skipped_when_disabled():
+    """Test that Hatchet validation is skipped when HATCHET_ENABLED is False."""
+    from reflector.services.transcript_process import (
+        ValidationOk,
+        validate_transcript_for_processing,
+    )
+
+    mock_transcript = Transcript(
+        id="test-transcript-id",
+        name="Test",
+        status="uploaded",
+        source_kind="room",
+        workflow_run_id="some-workflow-123",
+        recording_id="test-recording-id",
+    )
+
+    with patch("reflector.services.transcript_process.settings") as mock_settings:
+        mock_settings.HATCHET_ENABLED = False  # Hatchet disabled
+
+        with patch(
+            "reflector.services.transcript_process.task_is_scheduled_or_active"
+        ) as mock_celery_check:
+            mock_celery_check.return_value = False
+
+            result = await validate_transcript_for_processing(mock_transcript)
+
+        # Should not check Hatchet at all
+        assert isinstance(result, ValidationOk)
+
+
+@pytest.mark.usefixtures("setup_database")
+@pytest.mark.asyncio
+async def test_validation_locked_transcript():
+    """Test that validation rejects locked transcripts."""
+    from reflector.services.transcript_process import (
+        ValidationLocked,
+        validate_transcript_for_processing,
+    )
+
+    mock_transcript = Transcript(
+        id="test-transcript-id",
+        name="Test",
+        status="ended",
+        source_kind="room",
+        locked=True,
+    )
+
+    result = await validate_transcript_for_processing(mock_transcript)
+
+    assert isinstance(result, ValidationLocked)
+    assert "locked" in result.detail.lower()
+
+
+@pytest.mark.usefixtures("setup_database")
+@pytest.mark.asyncio
+async def test_validation_idle_transcript():
+    """Test that validation rejects idle transcripts (not ready)."""
+    from reflector.services.transcript_process import (
+        ValidationNotReady,
+        validate_transcript_for_processing,
+    )
+
+    mock_transcript = Transcript(
+        id="test-transcript-id",
+        name="Test",
+        status="idle",
+        source_kind="room",
+    )
+
+    result = await validate_transcript_for_processing(mock_transcript)
+
+    assert isinstance(result, ValidationNotReady)
+    assert "not ready" in result.detail.lower()
+
+
+@pytest.mark.usefixtures("setup_database")
+@pytest.mark.asyncio
+async def test_prepare_multitrack_config():
+    """Test preparing multitrack processing config."""
+    from reflector.db.recordings import Recording
+    from reflector.services.transcript_process import (
+        MultitrackProcessingConfig,
+        ValidationOk,
+        prepare_transcript_processing,
+    )
+
+    validation = ValidationOk(
+        recording_id="test-recording-id",
+        transcript_id="test-transcript-id",
+    )
+
+    mock_recording = Recording(
+        id="test-recording-id",
+        bucket_name="test-bucket",
+        object_key="recordings/test",
+        recorded_at="2024-01-01T00:00:00Z",
+        track_keys=["track1.webm", "track2.webm"],
+    )
+
+    with patch(
+        "reflector.services.transcript_process.recordings_controller"
+    ) as mock_rc:
+        mock_rc.get_by_id = AsyncMock(return_value=mock_recording)
+
+        result = await prepare_transcript_processing(validation)
+
+    assert isinstance(result, MultitrackProcessingConfig)
+    assert result.bucket_name == "test-bucket"
+    assert result.track_keys == ["track1.webm", "track2.webm"]
+    assert result.transcript_id == "test-transcript-id"
+    assert result.room_id is None  # ValidationOk didn't specify room_id
+
+
+@pytest.mark.usefixtures("setup_database")
+@pytest.mark.asyncio
+async def test_prepare_file_config():
+    """Test preparing file processing config (no track keys)."""
+    from reflector.db.recordings import Recording
+    from reflector.services.transcript_process import (
+        FileProcessingConfig,
+        ValidationOk,
+        prepare_transcript_processing,
+    )
+
+    validation = ValidationOk(
+        recording_id="test-recording-id",
+        transcript_id="test-transcript-id",
+    )
+
+    mock_recording = Recording(
+        id="test-recording-id",
+        bucket_name="test-bucket",
+        object_key="recordings/test.mp4",
+        recorded_at="2024-01-01T00:00:00Z",
+        track_keys=None,  # No track keys = file pipeline
+    )
+
+    with patch(
+        "reflector.services.transcript_process.recordings_controller"
+    ) as mock_rc:
+        mock_rc.get_by_id = AsyncMock(return_value=mock_recording)
+
+        result = await prepare_transcript_processing(validation)
+
+    assert isinstance(result, FileProcessingConfig)
+    assert result.transcript_id == "test-transcript-id"
--- a/server/uv.lock
+++ b/server/uv.lock
--- a/www/app/(app)/rooms/page.tsx
+++ b/www/app/(app)/rooms/page.tsx
@@ -91,6 +91,7 @@ const roomInitialState = {
  icsEnabled: false,
  icsFetchInterval: 5,
  platform: "whereby",
+  skipConsent: false,
 };

 export default function RoomsList() {
@@ -175,6 +176,7 @@ export default function RoomsList() {
            icsEnabled: detailedEditedRoom.ics_enabled || false,
            icsFetchInterval: detailedEditedRoom.ics_fetch_interval || 5,
            platform: detailedEditedRoom.platform,
+            skipConsent: detailedEditedRoom.skip_consent || false,
          }
        : null,
    [detailedEditedRoom],
@@ -326,6 +328,7 @@ export default function RoomsList() {
        ics_enabled: room.icsEnabled,
        ics_fetch_interval: room.icsFetchInterval,
        platform,
+        skip_consent: room.skipConsent,
      };

      if (isEditing) {
@@ -388,6 +391,7 @@ export default function RoomsList() {
      icsEnabled: roomData.ics_enabled || false,
      icsFetchInterval: roomData.ics_fetch_interval || 5,
      platform: roomData.platform,
+      skipConsent: roomData.skip_consent || false,
    });
    setEditRoomId(roomId);
    setIsEditing(true);
@@ -796,6 +800,34 @@ export default function RoomsList() {
                        <Checkbox.Label>Shared room</Checkbox.Label>
                      </Checkbox.Root>
                    </Field.Root>
+                    {room.recordingType === "cloud" && (
+                      <Field.Root mt={4}>
+                        <Checkbox.Root
+                          name="skipConsent"
+                          checked={room.skipConsent}
+                          onCheckedChange={(e) => {
+                            const syntheticEvent = {
+                              target: {
+                                name: "skipConsent",
+                                type: "checkbox",
+                                checked: e.checked,
+                              },
+                            };
+                            handleRoomChange(syntheticEvent);
+                          }}
+                        >
+                          <Checkbox.HiddenInput />
+                          <Checkbox.Control>
+                            <Checkbox.Indicator />
+                          </Checkbox.Control>
+                          <Checkbox.Label>Skip consent dialog</Checkbox.Label>
+                        </Checkbox.Root>
+                        <Field.HelperText>
+                          When enabled, participants won't be asked for
+                          recording consent. Audio will be stored automatically.
+                        </Field.HelperText>
+                      </Field.Root>
+                    )}
                  </Tabs.Content>

                  <Tabs.Content value="share" pt={6}>
--- a/www/app/(auth)/userInfo.tsx
+++ b/www/app/(auth)/userInfo.tsx
@@ -2,13 +2,19 @@

 import { Spinner, Link } from "@chakra-ui/react";
 import { useAuth } from "../lib/AuthProvider";
+import { usePathname } from "next/navigation";
+import { getLogoutRedirectUrl } from "../lib/auth";

 export default function UserInfo() {
  const auth = useAuth();
+  const pathname = usePathname();
  const status = auth.status;
  const isLoading = status === "loading";
  const isAuthenticated = status === "authenticated";
  const isRefreshing = status === "refreshing";
+
+  const callbackUrl = getLogoutRedirectUrl(pathname);
+
  return isLoading ? (
    <Spinner size="xs" className="mx-3" />
  ) : !isAuthenticated && !isRefreshing ? (
@@ -26,7 +32,7 @@ export default function UserInfo() {
    <Link
      href="#"
      className="font-light px-2"
-      onClick={() => auth.signOut({ callbackUrl: "/" })}
+      onClick={() => auth.signOut({ callbackUrl })}
    >
      Log out
    </Link>
--- a/www/app/[roomName]/MeetingSelection.tsx
+++ b/www/app/[roomName]/MeetingSelection.tsx
@@ -26,6 +26,7 @@ import { useRouter } from "next/navigation";
 import { formatDateTime, formatStartedAgo } from "../lib/timeUtils";
 import MeetingMinimalHeader from "../components/MeetingMinimalHeader";
 import { NonEmptyString } from "../lib/utils";
+import { MeetingId } from "../lib/types";

 type Meeting = components["schemas"]["Meeting"];

@@ -98,7 +99,7 @@ export default function MeetingSelection({
    onMeetingSelect(meeting);
  };

-  const handleEndMeeting = async (meetingId: string) => {
+  const handleEndMeeting = async (meetingId: MeetingId) => {
    try {
      await deactivateMeetingMutation.mutateAsync({
        params: {
--- a/www/app/[roomName]/components/DailyRoom.tsx
+++ b/www/app/[roomName]/components/DailyRoom.tsx
@@ -1,35 +1,194 @@
 "use client";

-import { useCallback, useEffect, useRef, useState } from "react";
+import {
+  RefObject,
+  useCallback,
+  useEffect,
+  useMemo,
+  useRef,
+  useState,
+} from "react";
 import { Box, Spinner, Center, Text } from "@chakra-ui/react";
 import { useRouter, useParams } from "next/navigation";
-import DailyIframe, { DailyCall } from "@daily-co/daily-js";
+import DailyIframe, {
+  DailyCall,
+  DailyCallOptions,
+  DailyCustomTrayButton,
+  DailyCustomTrayButtons,
+  DailyEventObjectCustomButtonClick,
+  DailyFactoryOptions,
+  DailyParticipantsObject,
+} from "@daily-co/daily-js";
 import type { components } from "../../reflector-api";
 import { useAuth } from "../../lib/AuthProvider";
-import {
-  ConsentDialogButton,
-  recordingTypeRequiresConsent,
-} from "../../lib/consent";
+import { useConsentDialog } from "../../lib/consent";
 import { useRoomJoinMeeting } from "../../lib/apiHooks";
+import { omit } from "remeda";
 import { assertExists } from "../../lib/utils";
+import { assertMeetingId } from "../../lib/types";
+
+const CONSENT_BUTTON_ID = "recording-consent";
+const RECORDING_INDICATOR_ID = "recording-indicator";

 type Meeting = components["schemas"]["Meeting"];
+type Room = components["schemas"]["RoomDetails"];

-interface DailyRoomProps {
+type DailyRoomProps = {
  meeting: Meeting;
-}
+  room: Room;
+};

-export default function DailyRoom({ meeting }: DailyRoomProps) {
+const useCustomTrayButtons = (
+  frame: {
+    updateCustomTrayButtons: (
+      customTrayButtons: DailyCustomTrayButtons,
+    ) => void;
+    joined: boolean;
+  } | null,
+) => {
+  const [, setCustomTrayButtons] = useState<DailyCustomTrayButtons>({});
+  return useCallback(
+    (id: string, button: DailyCustomTrayButton | null) => {
+      setCustomTrayButtons((prev) => {
+        // would blink state when frame blinks but it's ok here
+        const state =
+          button === null ? omit(prev, [id]) : { ...prev, [id]: button };
+        if (frame !== null && frame.joined)
+          frame.updateCustomTrayButtons(state);
+        return state;
+      });
+    },
+    [setCustomTrayButtons, frame],
+  );
+};
+
+const USE_FRAME_INIT_STATE = {
+  frame: null as DailyCall | null,
+  joined: false as boolean,
+} as const;
+
+// 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,
+  cbs: {
+    onLeftMeeting: () => void;
+    onCustomButtonClick: (ev: DailyEventObjectCustomButtonClick) => void;
+    onJoinMeeting: (
+      startRecording: (args: { type: "raw-tracks" }) => void,
+    ) => void;
+  },
+) => {
+  const [{ frame, joined }, setState] = useState(USE_FRAME_INIT_STATE);
+  const setJoined = useCallback(
+    (joined: boolean) => setState((prev) => ({ ...prev, joined })),
+    [setState],
+  );
+  const setFrame = useCallback(
+    (frame: DailyCall | null) => setState((prev) => ({ ...prev, frame })),
+    [setState],
+  );
+  useEffect(() => {
+    if (!container) return;
+    const init = async () => {
+      const existingFrame = DailyIframe.getCallInstance();
+      if (existingFrame) {
+        console.error("existing daily frame present");
+        await existingFrame.destroy();
+      }
+      const frameOptions: DailyFactoryOptions = {
+        iframeStyle: {
+          width: "100vw",
+          height: "100vh",
+          border: "none",
+        },
+        showLeaveButton: true,
+        showFullscreenButton: true,
+      };
+      const frame = DailyIframe.createFrame(container, frameOptions);
+      setFrame(frame);
+    };
+    init().catch(
+      console.error.bind(console, "Failed to initialize daily frame:"),
+    );
+    return () => {
+      frame
+        ?.destroy()
+        .catch(console.error.bind(console, "Failed to destroy daily frame:"));
+      setState(USE_FRAME_INIT_STATE);
+    };
+  }, [container]);
+  useEffect(() => {
+    if (!frame) return;
+    frame.on("left-meeting", cbs.onLeftMeeting);
+    frame.on("custom-button-click", cbs.onCustomButtonClick);
+    const joinCb = () => {
+      if (!frame) {
+        console.error("frame is null in joined-meeting callback");
+        return;
+      }
+      cbs.onJoinMeeting(frame.startRecording.bind(frame));
+    };
+    frame.on("joined-meeting", joinCb);
+    return () => {
+      frame.off("left-meeting", cbs.onLeftMeeting);
+      frame.off("custom-button-click", cbs.onCustomButtonClick);
+      frame.off("joined-meeting", joinCb);
+    };
+  }, [frame, cbs]);
+  const frame_ = useMemo(() => {
+    if (frame === null) return frame;
+    return {
+      join: async (
+        properties?: DailyCallOptions,
+      ): Promise<DailyParticipantsObject | void> => {
+        await frame.join(properties);
+        setJoined(!frame.isDestroyed());
+      },
+      updateCustomTrayButtons: (
+        customTrayButtons: DailyCustomTrayButtons,
+      ): DailyCall => frame.updateCustomTrayButtons(customTrayButtons),
+    };
+  }, [frame]);
+  const setCustomTrayButton = useCustomTrayButtons(
+    useMemo(() => {
+      if (frame_ === null) return null;
+      return {
+        updateCustomTrayButtons: frame_.updateCustomTrayButtons,
+        joined,
+      };
+    }, [frame_, joined]),
+  );
+  return [
+    frame_,
+    {
+      setCustomTrayButton,
+    },
+  ] as const;
+};
+
+export default function DailyRoom({ meeting, room }: DailyRoomProps) {
  const router = useRouter();
  const params = useParams();
  const auth = useAuth();
  const authLastUserId = auth.lastUserId;
-  const containerRef = useRef<HTMLDivElement>(null);
+  const [container, setContainer] = useState<HTMLDivElement | null>(null);
  const joinMutation = useRoomJoinMeeting();
  const [joinedMeeting, setJoinedMeeting] = useState<Meeting | null>(null);

  const roomName = params?.roomName as string;

+  const {
+    showConsentModal,
+    showRecordingIndicator: showRecordingInTray,
+    showConsentButton,
+  } = useConsentDialog({
+    meetingId: assertMeetingId(meeting.id),
+    recordingType: meeting.recording_type,
+    skipConsent: room.skip_consent,
+  });
+  const showConsentModalRef = useRef(showConsentModal);
+  showConsentModalRef.current = showConsentModal;
+
  useEffect(() => {
    if (authLastUserId === undefined || !meeting?.id || !roomName) return;

@@ -49,7 +208,7 @@ export default function DailyRoom({ meeting }: DailyRoomProps) {
      }
    };

-    join();
+    join().catch(console.error.bind(console, "Failed to join meeting:"));
  }, [meeting?.id, roomName, authLastUserId]);

  const roomUrl = joinedMeeting?.room_url;
@@ -58,54 +217,46 @@ export default function DailyRoom({ meeting }: DailyRoomProps) {
    router.push("/browse");
  }, [router]);

-  useEffect(() => {
-    if (authLastUserId === undefined || !roomUrl || !containerRef.current)
-      return;
-
-    let frame: DailyCall | null = null;
-    let destroyed = false;
-
-    const createAndJoin = async () => {
-      try {
-        const existingFrame = DailyIframe.getCallInstance();
-        if (existingFrame) {
-          await existingFrame.destroy();
+  const handleCustomButtonClick = useCallback(
+    (ev: DailyEventObjectCustomButtonClick) => {
+      if (ev.button_id === CONSENT_BUTTON_ID) {
+        showConsentModalRef.current();
      }
-
-        frame = DailyIframe.createFrame(containerRef.current!, {
-          iframeStyle: {
-            width: "100vw",
-            height: "100vh",
-            border: "none",
    },
-          showLeaveButton: true,
-          showFullscreenButton: true,
-        });
-
-        if (destroyed) {
-          await frame.destroy();
-          return;
-        }
-
-        frame.on("left-meeting", handleLeave);
-
-        frame.on("joined-meeting", async () => {
-          try {
-            const frameInstance = assertExists(
-              frame,
-              "frame object got lost somewhere after frame.on was called",
+    [
+      /*keep static; iframe recreation depends on it*/
+    ],
  );

+  const handleFrameJoinMeeting = useCallback(
+    (startRecording: (args: { type: "raw-tracks" }) => void) => {
+      try {
        if (meeting.recording_type === "cloud") {
          console.log("Starting cloud recording");
-              await frameInstance.startRecording({ type: "raw-tracks" });
+          startRecording({ type: "raw-tracks" });
        }
      } catch (error) {
        console.error("Failed to start recording:", error);
      }
+    },
+    [meeting.recording_type],
+  );
+
+  const recordingIconUrl = useMemo(
+    () => new URL("/recording-icon.svg", window.location.origin),
+    [],
+  );
+
+  const [frame, { setCustomTrayButton }] = useFrame(container, {
+    onLeftMeeting: handleLeave,
+    onCustomButtonClick: handleCustomButtonClick,
+    onJoinMeeting: handleFrameJoinMeeting,
  });

-        await frame.join({
+  useEffect(() => {
+    if (!frame || !roomUrl) return;
+    frame
+      .join({
        url: roomUrl,
        sendSettings: {
          video: {
@@ -117,25 +268,35 @@ export default function DailyRoom({ meeting }: DailyRoomProps) {
          },
          // Note: screenVideo intentionally not configured to preserve full quality for screen shares
        },
-        });
-      } catch (error) {
-        console.error("Error creating Daily frame:", error);
-      }
-    };
+      })
+      .catch(console.error.bind(console, "Failed to join daily room:"));
+  }, [frame, roomUrl]);

-    createAndJoin().catch((error) => {
-      console.error("Failed to create and join meeting:", error);
-    });
-
-    return () => {
-      destroyed = true;
-      if (frame) {
-        frame.destroy().catch((e) => {
-          console.error("Error destroying frame:", e);
-        });
+  useEffect(() => {
+    setCustomTrayButton(
+      RECORDING_INDICATOR_ID,
+      showRecordingInTray
+        ? {
+            iconPath: recordingIconUrl.href,
+            label: "Recording",
+            tooltip: "Recording in progress",
          }
-    };
-  }, [roomUrl, authLastUserId, handleLeave]);
+        : null,
+    );
+  }, [showRecordingInTray, recordingIconUrl, setCustomTrayButton]);
+
+  useEffect(() => {
+    setCustomTrayButton(
+      CONSENT_BUTTON_ID,
+      showConsentButton
+        ? {
+            iconPath: recordingIconUrl.href,
+            label: "Recording (click to consent)",
+            tooltip: "Recording (click to consent)",
+          }
+        : null,
+    );
+  }, [showConsentButton, recordingIconUrl, setCustomTrayButton]);

  if (authLastUserId === undefined) {
    return (
@@ -159,10 +320,7 @@ export default function DailyRoom({ meeting }: DailyRoomProps) {

  return (
    <Box position="relative" width="100vw" height="100vh">
-      <div ref={containerRef} style={{ width: "100%", height: "100%" }} />
-      {meeting.recording_type &&
-        recordingTypeRequiresConsent(meeting.recording_type) &&
-        meeting.id && <ConsentDialogButton meetingId={meeting.id} />}
+      <div ref={setContainer} style={{ width: "100%", height: "100%" }} />
    </Box>
  );
 }
--- a/www/app/[roomName]/components/RoomContainer.tsx
+++ b/www/app/[roomName]/components/RoomContainer.tsx
@@ -18,6 +18,7 @@ import { useAuth } from "../../lib/AuthProvider";
 import { useError } from "../../(errors)/errorContext";
 import { parseNonEmptyString } from "../../lib/utils";
 import { printApiError } from "../../api/_error";
+import { assertMeetingId } from "../../lib/types";

 type Meeting = components["schemas"]["Meeting"];

@@ -67,7 +68,10 @@ export default function RoomContainer(details: RoomDetails) {
    room && !room.ics_enabled && !pageMeetingId ? roomName : null,
  );

-  const explicitMeeting = useRoomGetMeeting(roomName, pageMeetingId || null);
+  const explicitMeeting = useRoomGetMeeting(
+    roomName,
+    pageMeetingId ? assertMeetingId(pageMeetingId) : null,
+  );

  const meeting = explicitMeeting.data || defaultMeeting.response;

@@ -192,9 +196,9 @@ export default function RoomContainer(details: RoomDetails) {

  switch (platform) {
    case "daily":
-      return <DailyRoom meeting={meeting} />;
+      return <DailyRoom meeting={meeting} room={room} />;
    case "whereby":
-      return <WherebyRoom meeting={meeting} />;
+      return <WherebyRoom meeting={meeting} room={room} />;
    default: {
      const _exhaustive: never = platform;
      return (
--- a/www/app/[roomName]/components/WherebyRoom.tsx
+++ b/www/app/[roomName]/components/WherebyRoom.tsx
@@ -5,24 +5,29 @@ import { useRouter } from "next/navigation";
 import type { components } from "../../reflector-api";
 import { useAuth } from "../../lib/AuthProvider";
 import { getWherebyUrl, useWhereby } from "../../lib/wherebyClient";
-import { assertExistsAndNonEmptyString, NonEmptyString } from "../../lib/utils";
 import {
  ConsentDialogButton as BaseConsentDialogButton,
  useConsentDialog,
-  recordingTypeRequiresConsent,
 } from "../../lib/consent";
+import { assertMeetingId, MeetingId } from "../../lib/types";

 type Meeting = components["schemas"]["Meeting"];
+type Room = components["schemas"]["RoomDetails"];

 interface WherebyRoomProps {
  meeting: Meeting;
+  room: Room;
 }

 function WherebyConsentDialogButton({
  meetingId,
+  recordingType,
+  skipConsent,
  wherebyRef,
 }: {
-  meetingId: NonEmptyString;
+  meetingId: MeetingId;
+  recordingType: Meeting["recording_type"];
+  skipConsent: boolean;
  wherebyRef: React.RefObject<HTMLElement>;
 }) {
  const previousFocusRef = useRef<HTMLElement | null>(null);
@@ -45,10 +50,16 @@ function WherebyConsentDialogButton({
    };
  }, [wherebyRef]);

-  return <BaseConsentDialogButton meetingId={meetingId} />;
+  return (
+    <BaseConsentDialogButton
+      meetingId={meetingId}
+      recordingType={recordingType}
+      skipConsent={skipConsent}
+    />
+  );
 }

-export default function WherebyRoom({ meeting }: WherebyRoomProps) {
+export default function WherebyRoom({ meeting, room }: WherebyRoomProps) {
  const wherebyLoaded = useWhereby();
  const wherebyRef = useRef<HTMLElement>(null);
  const router = useRouter();
@@ -57,9 +68,14 @@ export default function WherebyRoom({ meeting }: WherebyRoomProps) {
  const isAuthenticated = status === "authenticated";

  const wherebyRoomUrl = getWherebyUrl(meeting);
-  const recordingType = meeting.recording_type;
  const meetingId = meeting.id;

+  const { showConsentButton } = useConsentDialog({
+    meetingId: assertMeetingId(meetingId),
+    recordingType: meeting.recording_type,
+    skipConsent: room.skip_consent,
+  });
+
  const isLoading = status === "loading";

  const handleLeave = useCallback(() => {
@@ -88,11 +104,11 @@ export default function WherebyRoom({ meeting }: WherebyRoomProps) {
        room={wherebyRoomUrl}
        style={{ width: "100vw", height: "100vh" }}
      />
-      {recordingType &&
-        recordingTypeRequiresConsent(recordingType) &&
-        meetingId && (
+      {showConsentButton && (
        <WherebyConsentDialogButton
-            meetingId={assertExistsAndNonEmptyString(meetingId)}
+          meetingId={assertMeetingId(meetingId)}
+          recordingType={meeting.recording_type}
+          skipConsent={room.skip_consent}
          wherebyRef={wherebyRef}
        />
      )}
--- a/www/app/[roomName]/room.tsx
+++ b/www/app/[roomName]/room.tsx
@@ -6,7 +6,6 @@ import {
  useEffect,
  useRef,
  useState,
-  useContext,
  RefObject,
  use,
 } from "react";
@@ -25,8 +24,6 @@ import { useRecordingConsent } from "../recordingConsentContext";
 import {
  useMeetingAudioConsent,
  useRoomGetByName,
-  useRoomActiveMeetings,
-  useRoomUpcomingMeetings,
  useRoomsCreateMeeting,
  useRoomGetMeeting,
 } from "../lib/apiHooks";
@@ -39,12 +36,9 @@ import { FaBars } from "react-icons/fa6";
 import { useAuth } from "../lib/AuthProvider";
 import { getWherebyUrl, useWhereby } from "../lib/wherebyClient";
 import { useError } from "../(errors)/errorContext";
-import {
-  assertExistsAndNonEmptyString,
-  NonEmptyString,
-  parseNonEmptyString,
-} from "../lib/utils";
+import { parseNonEmptyString } from "../lib/utils";
 import { printApiError } from "../api/_error";
+import { assertMeetingId, MeetingId } from "../lib/types";

 export type RoomDetails = {
  params: Promise<{
@@ -92,16 +86,16 @@ const useConsentWherebyFocusManagement = (
 };

 const useConsentDialog = (
-  meetingId: string,
+  meetingId: MeetingId,
  wherebyRef: RefObject<HTMLElement> /*accessibility*/,
 ) => {
-  const { state: consentState, touch, hasConsent } = useRecordingConsent();
+  const { state: consentState, touch, hasAnswered } = useRecordingConsent();
  // toast would open duplicates, even with using "id=" prop
  const [modalOpen, setModalOpen] = useState(false);
  const audioConsentMutation = useMeetingAudioConsent();

  const handleConsent = useCallback(
-    async (meetingId: string, given: boolean) => {
+    async (meetingId: MeetingId, given: boolean) => {
      try {
        await audioConsentMutation.mutateAsync({
          params: {
@@ -114,7 +108,7 @@ const useConsentDialog = (
          },
        });

-        touch(meetingId);
+        touch(meetingId, given);
      } catch (error) {
        console.error("Error submitting consent:", error);
      }
@@ -216,7 +210,7 @@ const useConsentDialog = (
  return {
    showConsentModal,
    consentState,
-    hasConsent,
+    hasAnswered,
    consentLoading: audioConsentMutation.isPending,
  };
 };
@@ -225,13 +219,13 @@ function ConsentDialogButton({
  meetingId,
  wherebyRef,
 }: {
-  meetingId: NonEmptyString;
+  meetingId: MeetingId;
  wherebyRef: React.RefObject<HTMLElement>;
 }) {
-  const { showConsentModal, consentState, hasConsent, consentLoading } =
+  const { showConsentModal, consentState, hasAnswered, consentLoading } =
    useConsentDialog(meetingId, wherebyRef);

-  if (!consentState.ready || hasConsent(meetingId) || consentLoading) {
+  if (!consentState.ready || hasAnswered(meetingId) || consentLoading) {
    return null;
  }

@@ -284,7 +278,10 @@ export default function Room(details: RoomDetails) {
    room && !room.ics_enabled && !pageMeetingId ? roomName : null,
  );

-  const explicitMeeting = useRoomGetMeeting(roomName, pageMeetingId || null);
+  const explicitMeeting = useRoomGetMeeting(
+    roomName,
+    pageMeetingId ? assertMeetingId(pageMeetingId) : null,
+  );
  const wherebyRoomUrl = explicitMeeting.data
    ? getWherebyUrl(explicitMeeting.data)
    : defaultMeeting.response
@@ -437,7 +434,7 @@ export default function Room(details: RoomDetails) {
            recordingTypeRequiresConsent(recordingType) &&
            meetingId && (
              <ConsentDialogButton
-                meetingId={assertExistsAndNonEmptyString(meetingId)}
+                meetingId={assertMeetingId(meetingId)}
                wherebyRef={wherebyRef}
              />
            )}
--- a/www/app/lib/apiHooks.ts
+++ b/www/app/lib/apiHooks.ts
@@ -5,6 +5,7 @@ import { useError } from "../(errors)/errorContext";
 import { QueryClient, useQueryClient } from "@tanstack/react-query";
 import type { components } from "../reflector-api";
 import { useAuth } from "./AuthProvider";
+import { MeetingId } from "./types";

 /*
 * XXX error types returned from the hooks are not always correct; declared types are ValidationError but real type could be string or any other
@@ -718,7 +719,7 @@ export function useRoomActiveMeetings(roomName: string | null) {

 export function useRoomGetMeeting(
  roomName: string | null,
-  meetingId: string | null,
+  meetingId: MeetingId | null,
 ) {
  return $api.useQuery(
    "get",
--- a/www/app/lib/auth.ts
+++ b/www/app/lib/auth.ts
@@ -18,3 +18,8 @@ export const LOGIN_REQUIRED_PAGES = [
 export const PROTECTED_PAGES = new RegExp(
  LOGIN_REQUIRED_PAGES.map((page) => `^${page}$`).join("|"),
 );
+
+export function getLogoutRedirectUrl(pathname: string): string {
+  const transcriptPagePattern = /^\/transcripts\/[^/]+$/;
+  return transcriptPagePattern.test(pathname) ? pathname : "/";
+}
--- a/www/app/lib/consent/ConsentDialog.tsx
+++ b/www/app/lib/consent/ConsentDialog.tsx
@@ -1,5 +1,6 @@
 "use client";

+import { useState, useEffect } from "react";
 import { Box, Button, Text, VStack, HStack } from "@chakra-ui/react";
 import { CONSENT_DIALOG_TEXT } from "./constants";

@@ -9,6 +10,15 @@ interface ConsentDialogProps {
 }

 export function ConsentDialog({ onAccept, onReject }: ConsentDialogProps) {
+  const [acceptButton, setAcceptButton] = useState<HTMLButtonElement | null>(
+    null,
+  );
+
+  useEffect(() => {
+    // Auto-focus accept button so Escape key works (Daily iframe captures keyboard otherwise)
+    acceptButton?.focus();
+  }, [acceptButton]);
+
  return (
    <Box
      p={6}
@@ -26,7 +36,12 @@ export function ConsentDialog({ onAccept, onReject }: ConsentDialogProps) {
          <Button variant="ghost" size="sm" onClick={onReject}>
            {CONSENT_DIALOG_TEXT.rejectButton}
          </Button>
-          <Button colorPalette="primary" size="sm" onClick={onAccept}>
+          <Button
+            ref={setAcceptButton}
+            colorPalette="primary"
+            size="sm"
+            onClick={onAccept}
+          >
            {CONSENT_DIALOG_TEXT.acceptButton}
          </Button>
        </HStack>
--- a/www/app/lib/consent/ConsentDialogButton.tsx
+++ b/www/app/lib/consent/ConsentDialogButton.tsx
@@ -9,16 +9,26 @@ import {
  CONSENT_BUTTON_Z_INDEX,
  CONSENT_DIALOG_TEXT,
 } from "./constants";
+import { MeetingId } from "../types";
+import type { components } from "../../reflector-api";

-interface ConsentDialogButtonProps {
-  meetingId: string;
-}
+type Meeting = components["schemas"]["Meeting"];

-export function ConsentDialogButton({ meetingId }: ConsentDialogButtonProps) {
-  const { showConsentModal, consentState, hasConsent, consentLoading } =
-    useConsentDialog(meetingId);
+type ConsentDialogButtonProps = {
+  meetingId: MeetingId;
+  recordingType: Meeting["recording_type"];
+  skipConsent: boolean;
+};

-  if (!consentState.ready || hasConsent(meetingId) || consentLoading) {
+export function ConsentDialogButton({
+  meetingId,
+  recordingType,
+  skipConsent,
+}: ConsentDialogButtonProps) {
+  const { showConsentModal, consentState, showConsentButton, consentLoading } =
+    useConsentDialog({ meetingId, recordingType, skipConsent });
+
+  if (!consentState.ready || !showConsentButton || consentLoading) {
    return null;
  }

--- a/www/app/lib/consent/RecordingIndicator.tsx
+++ b/www/app/lib/consent/RecordingIndicator.tsx
@@ -0,0 +1,33 @@
+"use client";
+
+import { Box, Text } from "@chakra-ui/react";
+import { FaCircle } from "react-icons/fa6";
+import {
+  CONSENT_BUTTON_TOP_OFFSET,
+  CONSENT_BUTTON_LEFT_OFFSET,
+  CONSENT_BUTTON_Z_INDEX,
+} from "./constants";
+
+export function RecordingIndicator() {
+  return (
+    <Box
+      position="absolute"
+      top={CONSENT_BUTTON_TOP_OFFSET}
+      left={CONSENT_BUTTON_LEFT_OFFSET}
+      zIndex={CONSENT_BUTTON_Z_INDEX}
+      display="flex"
+      alignItems="center"
+      gap={2}
+      bg="red.500"
+      color="white"
+      px={3}
+      py={1.5}
+      borderRadius="md"
+      fontSize="sm"
+      fontWeight="medium"
+    >
+      <FaCircle size={8} />
+      <Text>Recording</Text>
+    </Box>
+  );
+}
--- a/www/app/lib/consent/index.ts
+++ b/www/app/lib/consent/index.ts
@@ -2,6 +2,7 @@

 export { ConsentDialogButton } from "./ConsentDialogButton";
 export { ConsentDialog } from "./ConsentDialog";
+export { RecordingIndicator } from "./RecordingIndicator";
 export { useConsentDialog } from "./useConsentDialog";
 export { recordingTypeRequiresConsent } from "./utils";
 export * from "./constants";
--- a/www/app/lib/consent/types.ts
+++ b/www/app/lib/consent/types.ts
@@ -1,9 +1,14 @@
-export interface ConsentDialogResult {
+import { MeetingId } from "../types";
+
+export type ConsentDialogResult = {
  showConsentModal: () => void;
  consentState: {
    ready: boolean;
-    consentAnsweredForMeetings?: Set<string>;
+    consentForMeetings?: Map<MeetingId, boolean>;
  };
-  hasConsent: (meetingId: string) => boolean;
+  hasAnswered: (meetingId: MeetingId) => boolean;
+  hasAccepted: (meetingId: MeetingId) => boolean;
  consentLoading: boolean;
-}
+  showRecordingIndicator: boolean;
+  showConsentButton: boolean;
+};
--- a/www/app/lib/consent/useConsentDialog.tsx
+++ b/www/app/lib/consent/useConsentDialog.tsx
@@ -7,9 +7,29 @@ import { useMeetingAudioConsent } from "../apiHooks";
 import { ConsentDialog } from "./ConsentDialog";
 import { TOAST_CHECK_INTERVAL_MS } from "./constants";
 import type { ConsentDialogResult } from "./types";
+import { MeetingId } from "../types";
+import { recordingTypeRequiresConsent } from "./utils";
+import type { components } from "../../reflector-api";

-export function useConsentDialog(meetingId: string): ConsentDialogResult {
-  const { state: consentState, touch, hasConsent } = useRecordingConsent();
+type Meeting = components["schemas"]["Meeting"];
+
+type UseConsentDialogParams = {
+  meetingId: MeetingId;
+  recordingType: Meeting["recording_type"];
+  skipConsent: boolean;
+};
+
+export function useConsentDialog({
+  meetingId,
+  recordingType,
+  skipConsent,
+}: UseConsentDialogParams): ConsentDialogResult {
+  const {
+    state: consentState,
+    touch,
+    hasAnswered,
+    hasAccepted,
+  } = useRecordingConsent();
  const [modalOpen, setModalOpen] = useState(false);
  const audioConsentMutation = useMeetingAudioConsent();
  const intervalRef = useRef<NodeJS.Timeout | null>(null);
@@ -42,7 +62,7 @@ export function useConsentDialog(meetingId: string): ConsentDialogResult {
          },
        });

-        touch(meetingId);
+        touch(meetingId, given);
      } catch (error) {
        console.error("Error submitting consent:", error);
      }
@@ -100,10 +120,23 @@ export function useConsentDialog(meetingId: string): ConsentDialogResult {
    });
  }, [handleConsent, modalOpen]);

+  const requiresConsent = Boolean(
+    recordingType && recordingTypeRequiresConsent(recordingType),
+  );
+
+  const showRecordingIndicator =
+    requiresConsent && (skipConsent || hasAccepted(meetingId));
+
+  const showConsentButton =
+    requiresConsent && !skipConsent && !hasAnswered(meetingId);
+
  return {
    showConsentModal,
    consentState,
-    hasConsent,
+    hasAnswered,
+    hasAccepted,
    consentLoading: audioConsentMutation.isPending,
+    showRecordingIndicator,
+    showConsentButton,
  };
 }
--- a/www/app/lib/types.ts
+++ b/www/app/lib/types.ts
@@ -1,6 +1,10 @@
 import type { Session } from "next-auth";
 import type { JWT } from "next-auth/jwt";
-import { parseMaybeNonEmptyString } from "./utils";
+import {
+  assertExistsAndNonEmptyString,
+  NonEmptyString,
+  parseMaybeNonEmptyString,
+} from "./utils";

 export interface JWTWithAccessToken extends JWT {
  accessToken: string;
@@ -78,3 +82,10 @@ export const assertCustomSession = <T extends Session>(
 export type Mutable<T> = {
  -readonly [P in keyof T]: T[P];
 };
+
+export type MeetingId = NonEmptyString & { __type: "MeetingId" };
+export const assertMeetingId = (s: string): MeetingId => {
+  const nes = assertExistsAndNonEmptyString(s);
+  // just cast for now
+  return nes as MeetingId;
+};
--- a/www/app/recordingConsentContext.tsx
+++ b/www/app/recordingConsentContext.tsx
@@ -1,18 +1,22 @@
 "use client";

 import React, { createContext, useContext, useEffect, useState } from "react";
+import { MeetingId } from "./lib/types";
+
+type ConsentMap = Map<MeetingId, boolean>;

 type ConsentContextState =
  | { ready: false }
  | {
      ready: true;
-      consentAnsweredForMeetings: Set<string>;
+      consentForMeetings: ConsentMap;
    };

 interface RecordingConsentContextValue {
  state: ConsentContextState;
-  touch: (meetingId: string) => void;
-  hasConsent: (meetingId: string) => boolean;
+  touch: (meetingId: MeetingId, accepted: boolean) => void;
+  hasAnswered: (meetingId: MeetingId) => boolean;
+  hasAccepted: (meetingId: MeetingId) => boolean;
 }

 const RecordingConsentContext = createContext<
@@ -35,81 +39,116 @@ interface RecordingConsentProviderProps {

 const LOCAL_STORAGE_KEY = "recording_consent_meetings";

+const ACCEPTED = "T" as const;
+type Accepted = typeof ACCEPTED;
+const REJECTED = "F" as const;
+type Rejected = typeof REJECTED;
+type Consent = Accepted | Rejected;
+const SEPARATOR = "|" as const;
+type Separator = typeof SEPARATOR;
+const DEFAULT_CONSENT = ACCEPTED;
+type Entry = `${MeetingId}${Separator}${Consent}`;
+type EntryAndDefault = Entry | MeetingId;
+
+// Format: "meetingId|T" or "meetingId|F", legacy format "meetingId" is treated as accepted
+const encodeEntry = (meetingId: MeetingId, accepted: boolean): Entry =>
+  `${meetingId}|${accepted ? ACCEPTED : REJECTED}`;
+
+const decodeEntry = (
+  entry: EntryAndDefault,
+): { meetingId: MeetingId; accepted: boolean } | null => {
+  const pipeIndex = entry.lastIndexOf(SEPARATOR);
+  if (pipeIndex === -1) {
+    // Legacy format: no pipe means accepted (backward compat)
+    return { meetingId: entry as MeetingId, accepted: true };
+  }
+  const suffix = entry.slice(pipeIndex + 1);
+  const meetingId = entry.slice(0, pipeIndex) as MeetingId;
+  // T = accepted, F = rejected, anything else = accepted (safe default)
+  const accepted = suffix !== REJECTED;
+  return { meetingId, accepted };
+};
+
 export const RecordingConsentProvider: React.FC<
  RecordingConsentProviderProps
 > = ({ children }) => {
  const [state, setState] = useState<ConsentContextState>({ ready: false });

-  const safeWriteToStorage = (meetingIds: string[]): void => {
+  const safeWriteToStorage = (consentMap: ConsentMap): void => {
    try {
      if (typeof window !== "undefined" && window.localStorage) {
-        localStorage.setItem(LOCAL_STORAGE_KEY, JSON.stringify(meetingIds));
+        const entries = Array.from(consentMap.entries())
+          .slice(-5)
+          .map(([id, accepted]) => encodeEntry(id, accepted));
+        localStorage.setItem(LOCAL_STORAGE_KEY, JSON.stringify(entries));
      }
    } catch (error) {
      console.error("Failed to save consent data to localStorage:", error);
    }
  };

-  // writes to local storage and to the state of context both
-  const touch = (meetingId: string): void => {
+  const touch = (meetingId: MeetingId, accepted: boolean): void => {
    if (!state.ready) {
      console.warn("Attempted to touch consent before context is ready");
      return;
    }

-    // has success regardless local storage write success: we don't handle that
-    // and don't want to crash anything with just consent functionality
-    const newSet = state.consentAnsweredForMeetings.has(meetingId)
-      ? state.consentAnsweredForMeetings
-      : new Set([...state.consentAnsweredForMeetings, meetingId]);
-    // note: preserves the set insertion order
-    const array = Array.from(newSet).slice(-5); // Keep latest 5
-    safeWriteToStorage(array);
-    setState({ ready: true, consentAnsweredForMeetings: newSet });
+    const newMap = new Map(state.consentForMeetings);
+    newMap.set(meetingId, accepted);
+    safeWriteToStorage(newMap);
+    setState({ ready: true, consentForMeetings: newMap });
  };

-  const hasConsent = (meetingId: string): boolean => {
+  const hasAnswered = (meetingId: MeetingId): boolean => {
    if (!state.ready) return false;
-    return state.consentAnsweredForMeetings.has(meetingId);
+    return state.consentForMeetings.has(meetingId);
+  };
+
+  const hasAccepted = (meetingId: MeetingId): boolean => {
+    if (!state.ready) return false;
+    return state.consentForMeetings.get(meetingId) === true;
  };

  // initialize on mount
  useEffect(() => {
    try {
      if (typeof window === "undefined" || !window.localStorage) {
-        setState({ ready: true, consentAnsweredForMeetings: new Set() });
+        setState({ ready: true, consentForMeetings: new Map() });
        return;
      }

      const stored = localStorage.getItem(LOCAL_STORAGE_KEY);
      if (!stored) {
-        setState({ ready: true, consentAnsweredForMeetings: new Set() });
+        setState({ ready: true, consentForMeetings: new Map() });
        return;
      }

      const parsed = JSON.parse(stored);
      if (!Array.isArray(parsed)) {
        console.warn("Invalid consent data format in localStorage, resetting");
-        setState({ ready: true, consentAnsweredForMeetings: new Set() });
+        setState({ ready: true, consentForMeetings: new Map() });
        return;
      }

-      // pre-historic way of parsing!
-      const consentAnsweredForMeetings = new Set(
-        parsed.filter((id) => !!id && typeof id === "string"),
-      );
-      setState({ ready: true, consentAnsweredForMeetings });
+      const consentForMeetings = new Map<MeetingId, boolean>();
+      for (const entry of parsed) {
+        const decoded = decodeEntry(entry);
+        if (decoded) {
+          consentForMeetings.set(decoded.meetingId, decoded.accepted);
+        }
+      }
+      setState({ ready: true, consentForMeetings });
    } catch (error) {
-      // we don't want to fail the page here; the component is not essential.
      console.error("Failed to parse consent data from localStorage:", error);
-      setState({ ready: true, consentAnsweredForMeetings: new Set() });
+      setState({ ready: true, consentForMeetings: new Map() });
    }
  }, []);

  const value: RecordingConsentContextValue = {
    state,
    touch,
-    hasConsent,
+    hasAnswered,
+    hasAccepted,
  };

  return (
--- a/www/app/reflector-api.d.ts
+++ b/www/app/reflector-api.d.ts
@@ -893,8 +893,16 @@ export interface components {
       * @default false
       */
      ics_enabled: boolean;
-      /** Platform */
-      platform?: ("whereby" | "daily") | null;
+      /**
+       * Platform
+       * @enum {string}
+       */
+      platform: "whereby" | "daily";
+      /**
+       * Skip Consent
+       * @default false
+       */
+      skip_consent: boolean;
    };
    /** CreateRoomMeeting */
    CreateRoomMeeting: {
@@ -1123,7 +1131,9 @@ export interface components {
      /** Audio Deleted */
      audio_deleted?: boolean | null;
      /** Participants */
-      participants: components["schemas"]["TranscriptParticipant"][] | null;
+      participants:
+        | components["schemas"]["TranscriptParticipantWithEmail"][]
+        | null;
      /**
       * @description discriminator enum property added by openapi-typescript
       * @enum {string}
@@ -1184,7 +1194,9 @@ export interface components {
      /** Audio Deleted */
      audio_deleted?: boolean | null;
      /** Participants */
-      participants: components["schemas"]["TranscriptParticipant"][] | null;
+      participants:
+        | components["schemas"]["TranscriptParticipantWithEmail"][]
+        | null;
    };
    /**
     * GetTranscriptWithText
@@ -1246,7 +1258,9 @@ export interface components {
      /** Audio Deleted */
      audio_deleted?: boolean | null;
      /** Participants */
-      participants: components["schemas"]["TranscriptParticipant"][] | null;
+      participants:
+        | components["schemas"]["TranscriptParticipantWithEmail"][]
+        | null;
      /**
       * @description discriminator enum property added by openapi-typescript
       * @enum {string}
@@ -1315,7 +1329,9 @@ export interface components {
      /** Audio Deleted */
      audio_deleted?: boolean | null;
      /** Participants */
-      participants: components["schemas"]["TranscriptParticipant"][] | null;
+      participants:
+        | components["schemas"]["TranscriptParticipantWithEmail"][]
+        | null;
      /**
       * @description discriminator enum property added by openapi-typescript
       * @enum {string}
@@ -1386,7 +1402,9 @@ export interface components {
      /** Audio Deleted */
      audio_deleted?: boolean | null;
      /** Participants */
-      participants: components["schemas"]["TranscriptParticipant"][] | null;
+      participants:
+        | components["schemas"]["TranscriptParticipantWithEmail"][]
+        | null;
      /**
       * @description discriminator enum property added by openapi-typescript
       * @enum {string}
@@ -1567,6 +1585,11 @@ export interface components {
      /** Name */
      name: string;
    };
+    /** ProcessStatus */
+    ProcessStatus: {
+      /** Status */
+      status: string;
+    };
    /** Room */
    Room: {
      /** Id */
@@ -1617,6 +1640,11 @@ export interface components {
       * @enum {string}
       */
      platform: "whereby" | "daily";
+      /**
+       * Skip Consent
+       * @default false
+       */
+      skip_consent: boolean;
    };
    /** RoomDetails */
    RoomDetails: {
@@ -1668,6 +1696,11 @@ export interface components {
       * @enum {string}
       */
      platform: "whereby" | "daily";
+      /**
+       * Skip Consent
+       * @default false
+       */
+      skip_consent: boolean;
      /** Webhook Url */
      webhook_url: string | null;
      /** Webhook Secret */
@@ -1813,6 +1846,19 @@ export interface components {
      /** User Id */
      user_id?: string | null;
    };
+    /** TranscriptParticipantWithEmail */
+    TranscriptParticipantWithEmail: {
+      /** Id */
+      id?: string;
+      /** Speaker */
+      speaker: number | null;
+      /** Name */
+      name: string;
+      /** User Id */
+      user_id?: string | null;
+      /** Email */
+      email?: string | null;
+    };
    /**
     * TranscriptSegment
     * @description A single transcript segment with speaker and timing information.
@@ -1868,6 +1914,8 @@ export interface components {
      ics_enabled?: boolean | null;
      /** Platform */
      platform?: ("whereby" | "daily") | null;
+      /** Skip Consent */
+      skip_consent?: boolean | null;
    };
    /** UpdateTranscript */
    UpdateTranscript: {
@@ -3362,7 +3410,7 @@ export interface operations {
          [name: string]: unknown;
        };
        content: {
-          "application/json": unknown;
+          "application/json": components["schemas"]["ProcessStatus"];
        };
      };
      /** @description Validation Error */
--- a/www/public/recording-icon.svg
+++ b/www/public/recording-icon.svg
@@ -0,0 +1,3 @@
+<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="#ef4444" stroke="none">
+  <circle cx="12" cy="12" r="8"/>
+</svg>
Author	SHA1	Message	Date
Igor Loskutov	2c1c48a219	self-review	2025-12-22 18:35:33 -05:00
Igor Loskutov	4b495417b5	self-review	2025-12-22 18:18:03 -05:00
Igor Loskutov	2cbc373cc3	self-review	2025-12-22 18:06:59 -05:00
Igor Loskutov	8665204ab1	self-review	2025-12-22 17:19:00 -05:00
Igor Loskutov	42bde5adb6	comments	2025-12-22 15:59:52 -05:00
Igor Loskutov	b9698c2aaf	dry (no-mistakes) (minimal)	2025-12-22 15:58:13 -05:00
Igor Loskutov	1f41f16928	parallelize hatchet (no-mistakes)	2025-12-22 15:23:54 -05:00
Mathieu Virbel	7c2d0698ed	chore(main): release 0.25.0 (#803 )	2025-12-22 12:18:35 -05:00
Igor Monadical	1dac999b56	feat: durable (#794 ) * durable (no-mistakes) * hatchet no-mistake * hatchet no-mistake * hatchet no-mistake, better logging * remove conductor and add hatchet tests (no-mistakes) * self-review (no-mistakes) * hatched logs * remove shadow mode for hatchet * and add hatchet processor setting to room * . * cleanup * hatchet init db * self-review (no-mistakes) * self-review (no-mistakes) * hatchet: restore zullip report * self-review round * self-review round * self-review round * dry hatchet with celery * dry hatched with celery - 2 * self-review round * more NES instead of str * self-review wip * self-review round * self-review round * self-review round * can_replay cancelled * add forgotten file * pr autoreviewer fixes * better log webhook events * durable_started return * migration sync * latest changes feature parity * migration merge * pr review --------- Co-authored-by: Igor Loskutov <igor.loskutoff@gmail.com>	2025-12-22 12:09:20 -05:00
Igor Monadical	f580b996ee	feat: increase daily recording max duration (#801 ) * increate daily recording max duration * recording end time: 3h min --------- Co-authored-by: Igor Loskutov <igor.loskutoff@gmail.com>	2025-12-22 09:02:14 -05:00
Igor Monadical	225783496f	feat: consent disable feature (#799 ) * consent disable feature (no-mistakes) * sync migration * consent disable refactor * daily backend code refactor * consent skip feature * consent skip feature * no forced whereby recording indicator * active meetings type precision * cleanup * cleanup --------- Co-authored-by: Igor Loskutov <igor.loskutoff@gmail.com>	2025-12-22 08:47:07 -05:00
Sergey Mankovsky	f0ee7b531a	fix: logout redirect (#802 )	2025-12-19 17:19:09 +01:00
Mathieu Virbel	37a454f283	chore(main): release 0.24.0 (#793 )	2025-12-19 15:00:43 +01:00