Merge branch 'main' into fix-daily-ux

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

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

CHANGELOG.md

@@ -1,17 +1,5 @@
 # Changelog
 
-## [0.33.0](https://github.com/Monadical-SAS/reflector/compare/v0.32.2...v0.33.0) (2026-02-05)
-
-
-### Features
-
-* Daily+hatchet default ([#846](https://github.com/Monadical-SAS/reflector/issues/846)) ([15ab2e3](https://github.com/Monadical-SAS/reflector/commit/15ab2e306eacf575494b4b5d2b2ad779d44a1c7f))
-
-
-### Bug Fixes
-
-* websocket tests ([#825](https://github.com/Monadical-SAS/reflector/issues/825)) ([1ce1c7a](https://github.com/Monadical-SAS/reflector/commit/1ce1c7a910b6c374115d2437b17f9d288ef094dc))
-
 ## [0.32.2](https://github.com/Monadical-SAS/reflector/compare/v0.32.1...v0.32.2) (2026-02-03)
 
 

server/reflector/db/meetings.py

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

server/reflector/hatchet/dag_zulip.py

@@ -1,144 +0,0 @@
-"""
-Hatchet DAG Status -> Zulip Live Updates.
-
-Posts/updates/deletes a Zulip message showing the Hatchet workflow DAG status.
-All functions are fire-and-forget (catch + warning log on failure).
-
-Note: Uses deferred imports throughout for fork-safety,
-consistent with the pipeline pattern in daily_multitrack_pipeline.py.
-"""
-
-from reflector.logger import logger
-from reflector.settings import settings
-
-
-def _dag_zulip_enabled() -> bool:
-    return bool(
-        settings.ZULIP_REALM and settings.ZULIP_DAG_STREAM and settings.ZULIP_DAG_TOPIC
-    )
-
-
-async def create_dag_zulip_message(transcript_id: str, workflow_run_id: str) -> None:
-    """Post initial DAG status to Zulip. Called at dispatch time (normal DB context)."""
-    if not _dag_zulip_enabled():
-        return
-
-    try:
-        from reflector.db.transcripts import transcripts_controller  # noqa: PLC0415
-        from reflector.hatchet.client import HatchetClientManager  # noqa: PLC0415
-        from reflector.tools.render_hatchet_run import (  # noqa: PLC0415
-            render_run_detail,
-        )
-        from reflector.zulip import send_message_to_zulip  # noqa: PLC0415
-
-        client = HatchetClientManager.get_client()
-        details = await client.runs.aio_get(workflow_run_id)
-        content = render_run_detail(details)
-
-        response = await send_message_to_zulip(
-            settings.ZULIP_DAG_STREAM, settings.ZULIP_DAG_TOPIC, content
-        )
-        message_id = response.get("id")
-
-        if message_id:
-            transcript = await transcripts_controller.get_by_id(transcript_id)
-            if transcript:
-                await transcripts_controller.update(
-                    transcript, {"zulip_message_id": message_id}
-                )
-    except Exception:
-        logger.warning(
-            "[DAG Zulip] Failed to create DAG message",
-            transcript_id=transcript_id,
-            workflow_run_id=workflow_run_id,
-            exc_info=True,
-        )
-
-
-async def update_dag_zulip_message(
-    transcript_id: str,
-    workflow_run_id: str,
-    error_message: str | None = None,
-) -> None:
-    """Update existing DAG status in Zulip. Called from Hatchet worker (forked).
-
-    Args:
-        error_message: If set, appended as an error banner to the rendered DAG.
-    """
-    if not _dag_zulip_enabled():
-        return
-
-    try:
-        from reflector.db.transcripts import transcripts_controller  # noqa: PLC0415
-        from reflector.hatchet.client import HatchetClientManager  # noqa: PLC0415
-        from reflector.hatchet.workflows.daily_multitrack_pipeline import (  # noqa: PLC0415
-            fresh_db_connection,
-        )
-        from reflector.tools.render_hatchet_run import (  # noqa: PLC0415
-            render_run_detail,
-        )
-        from reflector.zulip import update_zulip_message  # noqa: PLC0415
-
-        async with fresh_db_connection():
-            transcript = await transcripts_controller.get_by_id(transcript_id)
-            if not transcript or not transcript.zulip_message_id:
-                return
-
-            client = HatchetClientManager.get_client()
-            details = await client.runs.aio_get(workflow_run_id)
-            content = render_run_detail(details)
-
-            if error_message:
-                content += f"\n\n:cross_mark: **{error_message}**"
-
-            await update_zulip_message(
-                transcript.zulip_message_id,
-                settings.ZULIP_DAG_STREAM,
-                settings.ZULIP_DAG_TOPIC,
-                content,
-            )
-    except Exception:
-        logger.warning(
-            "[DAG Zulip] Failed to update DAG message",
-            transcript_id=transcript_id,
-            workflow_run_id=workflow_run_id,
-            exc_info=True,
-        )
-
-
-async def delete_dag_zulip_message(transcript_id: str) -> None:
-    """Delete DAG Zulip message and clear zulip_message_id.
-
-    Called from post_zulip task (already inside fresh_db_connection).
-    Swallows InvalidMessageError (message already deleted).
-    """
-    if not _dag_zulip_enabled():
-        return
-
-    try:
-        from reflector.db.transcripts import transcripts_controller  # noqa: PLC0415
-        from reflector.zulip import (  # noqa: PLC0415
-            InvalidMessageError,
-            delete_zulip_message,
-        )
-
-        transcript = await transcripts_controller.get_by_id(transcript_id)
-        if not transcript or not transcript.zulip_message_id:
-            return
-
-        try:
-            await delete_zulip_message(transcript.zulip_message_id)
-        except InvalidMessageError:
-            logger.warning(
-                "[DAG Zulip] Message already deleted",
-                transcript_id=transcript_id,
-                zulip_message_id=transcript.zulip_message_id,
-            )
-
-        await transcripts_controller.update(transcript, {"zulip_message_id": None})
-    except Exception:
-        logger.warning(
-            "[DAG Zulip] Failed to delete DAG message",
-            transcript_id=transcript_id,
-            exc_info=True,
-        )

server/reflector/hatchet/workflows/daily_multitrack_pipeline.py

@@ -45,7 +45,6 @@ from reflector.hatchet.constants import (
     TIMEOUT_SHORT,
     TaskName,
 )
-from reflector.hatchet.dag_zulip import update_dag_zulip_message
 from reflector.hatchet.workflows.models import (
     ActionItemsResult,
     ConsentResult,
@@ -239,14 +238,7 @@ def with_error_handling(
         @functools.wraps(func)
         async def wrapper(input: PipelineInput, ctx: Context) -> R:
             try:
-                result = await func(input, ctx)
-                try:
-                    await update_dag_zulip_message(
-                        input.transcript_id, ctx.workflow_run_id
-                    )
-                except Exception:
-                    pass
-                return result
+                return await func(input, ctx)
             except Exception as e:
                 logger.error(
                     f"[Hatchet] {step_name} failed",
@@ -254,14 +246,6 @@ def with_error_handling(
                     error=str(e),
                     exc_info=True,
                 )
-                try:
-                    await update_dag_zulip_message(
-                        input.transcript_id,
-                        ctx.workflow_run_id,
-                        error_message=f"{step_name} failed: {e}",
-                    )
-                except Exception:
-                    pass
                 if set_error_status:
                     await set_workflow_error_status(input.transcript_id)
                 raise
@@ -1310,11 +1294,6 @@ async def post_zulip(input: PipelineInput, ctx: Context) -> ZulipResult:
 
     async with fresh_db_connection():
         from reflector.db.transcripts import transcripts_controller  # noqa: PLC0415
-        from reflector.hatchet.dag_zulip import (  # noqa: PLC0415
-            delete_dag_zulip_message,
-        )
-
-        await delete_dag_zulip_message(input.transcript_id)
 
         transcript = await transcripts_controller.get_by_id(input.transcript_id)
         if transcript:

server/reflector/services/transcript_process.py

@@ -17,7 +17,6 @@ from hatchet_sdk.clients.rest.models import V1TaskStatus
 from reflector.db.recordings import recordings_controller
 from reflector.db.transcripts import Transcript, transcripts_controller
 from reflector.hatchet.client import HatchetClientManager
-from reflector.hatchet.dag_zulip import create_dag_zulip_message
 from reflector.logger import logger
 from reflector.pipelines.main_file_pipeline import task_pipeline_file_process
 from reflector.utils.string import NonEmptyString
@@ -267,16 +266,6 @@ async def dispatch_transcript_processing(
                 transcript, {"workflow_run_id": workflow_id}
             )
 
-        try:
-            await create_dag_zulip_message(config.transcript_id, workflow_id)
-        except Exception:
-            logger.warning(
-                "[DAG Zulip] Failed to create DAG message at dispatch",
-                transcript_id=config.transcript_id,
-                workflow_id=workflow_id,
-                exc_info=True,
-            )
-
         logger.info("Hatchet workflow dispatched", workflow_id=workflow_id)
         return None
 

server/reflector/settings.py

@@ -161,9 +161,6 @@ class Settings(BaseSettings):
     ZULIP_REALM: str | None = None
     ZULIP_API_KEY: str | None = None
     ZULIP_BOT_EMAIL: str | None = None
-    ZULIP_DAG_STREAM: str | None = None
-    ZULIP_DAG_TOPIC: str | None = None
-    ZULIP_HOST_HEADER: str | None = None
 
     # Hatchet workflow orchestration (always enabled for multitrack processing)
     HATCHET_CLIENT_TOKEN: str | None = None

server/reflector/tools/render_hatchet_run.py

@@ -1,412 +0,0 @@
-"""
-Render Hatchet workflow runs as text DAG.
-
-Usage:
-    # Show latest 5 runs (summary table)
-    uv run -m reflector.tools.render_hatchet_run
-
-    # Show specific run with full DAG + task details
-    uv run -m reflector.tools.render_hatchet_run <workflow_run_id>
-
-    # Drill into Nth run from the list (1-indexed)
-    uv run -m reflector.tools.render_hatchet_run --show 1
-
-    # Show latest N runs
-    uv run -m reflector.tools.render_hatchet_run --last 10
-
-    # Filter by status
-    uv run -m reflector.tools.render_hatchet_run --status FAILED
-    uv run -m reflector.tools.render_hatchet_run --status RUNNING
-"""
-
-import argparse
-import asyncio
-from collections import defaultdict
-from datetime import datetime, timedelta, timezone
-
-from hatchet_sdk.clients.rest.models import (
-    V1TaskEvent,
-    V1TaskStatus,
-    V1TaskSummary,
-    V1WorkflowRunDetails,
-    WorkflowRunShapeItemForWorkflowRunDetails,
-)
-
-from reflector.hatchet.client import HatchetClientManager
-
-STATUS_ICON = {
-    V1TaskStatus.COMPLETED: "\u2705",
-    V1TaskStatus.RUNNING: "\u23f3",
-    V1TaskStatus.FAILED: "\u274c",
-    V1TaskStatus.QUEUED: "\u23f8\ufe0f",
-    V1TaskStatus.CANCELLED: "\u26a0\ufe0f",
-}
-
-STATUS_LABEL = {
-    V1TaskStatus.COMPLETED: "Complete",
-    V1TaskStatus.RUNNING: "Running",
-    V1TaskStatus.FAILED: "FAILED",
-    V1TaskStatus.QUEUED: "Queued",
-    V1TaskStatus.CANCELLED: "Cancelled",
-}
-
-
-def _fmt_time(dt: datetime | None) -> str:
-    if dt is None:
-        return "-"
-    return dt.strftime("%H:%M:%S")
-
-
-def _fmt_duration(ms: int | None) -> str:
-    if ms is None:
-        return "-"
-    secs = ms / 1000
-    if secs < 60:
-        return f"{secs:.1f}s"
-    mins = secs / 60
-    return f"{mins:.1f}m"
-
-
-def _fmt_status_line(task: V1TaskSummary) -> str:
-    """Format a status line like: Complete (finished 20:31:44)"""
-    label = STATUS_LABEL.get(task.status, task.status.value)
-    icon = STATUS_ICON.get(task.status, "?")
-
-    if task.status == V1TaskStatus.COMPLETED and task.finished_at:
-        return f"{icon} {label} (finished {_fmt_time(task.finished_at)})"
-    elif task.status == V1TaskStatus.RUNNING and task.started_at:
-        parts = [f"started {_fmt_time(task.started_at)}"]
-        if task.duration:
-            parts.append(f"{_fmt_duration(task.duration)} elapsed")
-        return f"{icon} {label} ({', '.join(parts)})"
-    elif task.status == V1TaskStatus.FAILED and task.finished_at:
-        return f"{icon} {label} (failed {_fmt_time(task.finished_at)})"
-    elif task.status == V1TaskStatus.CANCELLED:
-        return f"{icon} {label}"
-    elif task.status == V1TaskStatus.QUEUED:
-        return f"{icon} {label}"
-    return f"{icon} {label}"
-
-
-def _topo_sort(
-    shape: list[WorkflowRunShapeItemForWorkflowRunDetails],
-) -> list[str]:
-    """Topological sort of step_ids from shape DAG."""
-    step_ids = {s.step_id for s in shape}
-    children_map: dict[str, list[str]] = {}
-    in_degree: dict[str, int] = {sid: 0 for sid in step_ids}
-
-    for s in shape:
-        children = [c for c in (s.children_step_ids or []) if c in step_ids]
-        children_map[s.step_id] = children
-        for c in children:
-            in_degree[c] += 1
-
-    queue = sorted(sid for sid, deg in in_degree.items() if deg == 0)
-    result: list[str] = []
-    while queue:
-        node = queue.pop(0)
-        result.append(node)
-        for c in children_map.get(node, []):
-            in_degree[c] -= 1
-            if in_degree[c] == 0:
-                queue.append(c)
-                queue.sort()
-
-    return result
-
-
-def render_run_detail(details: V1WorkflowRunDetails) -> str:
-    """Render a single workflow run as markdown DAG with task details."""
-    shape = details.shape or []
-    tasks = details.tasks or []
-    events = details.task_events or []
-    run = details.run
-
-    if not shape:
-        return f"Run {run.metadata.id}: {run.status.value} (no shape data)"
-
-    # Build lookups
-    step_to_shape: dict[str, WorkflowRunShapeItemForWorkflowRunDetails] = {
-        s.step_id: s for s in shape
-    }
-    step_to_name: dict[str, str] = {s.step_id: s.task_name for s in shape}
-
-    # Reverse edges (parents)
-    parents: dict[str, list[str]] = {s.step_id: [] for s in shape}
-    for s in shape:
-        for child_id in s.children_step_ids or []:
-            if child_id in parents:
-                parents[child_id].append(s.step_id)
-
-    # Join tasks by step_id
-    task_by_step: dict[str, V1TaskSummary] = {}
-    for t in tasks:
-        if t.step_id and t.step_id in step_to_name:
-            task_by_step[t.step_id] = t
-
-    # Events indexed by task_external_id
-    events_by_task: dict[str, list[V1TaskEvent]] = defaultdict(list)
-    for ev in events:
-        events_by_task[ev.task_id].append(ev)
-
-    ordered = _topo_sort(shape)
-
-    lines: list[str] = []
-
-    # Run header
-    run_icon = STATUS_ICON.get(run.status, "?")
-    run_name = run.display_name or run.workflow_id
-    dur = _fmt_duration(run.duration)
-    lines.append(f"**{run_name}** {run_icon} {dur}")
-    lines.append(f"ID: `{run.metadata.id}`")
-    if run.additional_metadata:
-        meta_parts = [f"{k}=`{v}`" for k, v in run.additional_metadata.items()]
-        lines.append(f"Meta: {', '.join(meta_parts)}")
-    if run.error_message:
-        # Take first line of error only for header
-        first_line = run.error_message.split("\n")[0]
-        lines.append(f"Error: {first_line}")
-    lines.append("")
-
-    # DAG Status Overview table
-    lines.append("**DAG Status Overview**")
-    lines.append("")
-    lines.append("| Node | Status | Duration | Dependencies |")
-    lines.append("|------|--------|----------|--------------|")
-
-    for step_id in ordered:
-        s = step_to_shape[step_id]
-        t = task_by_step.get(step_id)
-        name = step_to_name[step_id]
-        icon = STATUS_ICON.get(t.status, "?") if t else "?"
-        dur = _fmt_duration(t.duration) if t else "-"
-
-        parent_names = [step_to_name[p] for p in parents[step_id]]
-        child_names = [
-            step_to_name[c] for c in (s.children_step_ids or []) if c in step_to_name
-        ]
-        deps_left = ", ".join(parent_names) if parent_names else ""
-        deps_right = ", ".join(child_names) if child_names else ""
-        if deps_left and deps_right:
-            deps = f"{deps_left} \u2192 {deps_right}"
-        elif deps_right:
-            deps = f"\u2192 {deps_right}"
-        elif deps_left:
-            deps = f"{deps_left} \u2192"
-        else:
-            deps = "-"
-
-        lines.append(f"| {name} | {icon} | {dur} | {deps} |")
-
-    lines.append("")
-    lines.append("---")
-    lines.append("")
-
-    # Node details
-    for step_id in ordered:
-        t = task_by_step.get(step_id)
-        name = step_to_name[step_id]
-
-        if not t:
-            lines.append(f"**\U0001f4e6 {name}**")
-            lines.append("Status: no task data")
-            lines.append("")
-            continue
-
-        lines.append(f"**\U0001f4e6 {name}**")
-        lines.append(f"Status: {_fmt_status_line(t)}")
-
-        if t.duration:
-            lines.append(f"Duration: {_fmt_duration(t.duration)}")
-        if t.retry_count and t.retry_count > 0:
-            lines.append(f"Retries: {t.retry_count}")
-
-        # Fan-out children
-        if t.num_spawned_children and t.num_spawned_children > 0:
-            children = t.children or []
-            completed = sum(1 for c in children if c.status == V1TaskStatus.COMPLETED)
-            failed = sum(1 for c in children if c.status == V1TaskStatus.FAILED)
-            running = sum(1 for c in children if c.status == V1TaskStatus.RUNNING)
-            lines.append(
-                f"Spawned children: {completed}/{t.num_spawned_children} done"
-                f"{f', {running} running' if running else ''}"
-                f"{f', {failed} failed' if failed else ''}"
-            )
-
-        # Error message (first meaningful line only, full trace in events)
-        if t.error_message:
-            err_lines = t.error_message.strip().split("\n")
-            # Find first non-empty, non-traceback line
-            err_summary = err_lines[0]
-            for line in err_lines:
-                stripped = line.strip()
-                if stripped and not stripped.startswith(
-                    ("Traceback", "File ", "{", ")")
-                ):
-                    err_summary = stripped
-                    break
-            lines.append(f"Error: `{err_summary}`")
-
-        # Events log
-        task_events = sorted(
-            events_by_task.get(t.task_external_id, []),
-            key=lambda e: e.timestamp,
-        )
-        if task_events:
-            lines.append("Events:")
-            for ev in task_events:
-                ts = ev.timestamp.strftime("%H:%M:%S")
-                ev_icon = ""
-                if ev.event_type.value == "FINISHED":
-                    ev_icon = "\u2705 "
-                elif ev.event_type.value in ("FAILED", "TIMED_OUT"):
-                    ev_icon = "\u274c "
-                elif ev.event_type.value == "STARTED":
-                    ev_icon = "\u25b6\ufe0f "
-                elif ev.event_type.value == "RETRYING":
-                    ev_icon = "\U0001f504 "
-                elif ev.event_type.value == "CANCELLED":
-                    ev_icon = "\u26a0\ufe0f "
-
-                msg = ev.message.strip()
-                if ev.error_message:
-                    # Just first line of error in event log
-                    err_first = ev.error_message.strip().split("\n")[0]
-                    if msg:
-                        msg += f" | {err_first}"
-                    else:
-                        msg = err_first
-
-                if msg:
-                    lines.append(f"  `{ts}` {ev_icon}{ev.event_type.value}: {msg}")
-                else:
-                    lines.append(f"  `{ts}` {ev_icon}{ev.event_type.value}")
-
-        lines.append("")
-
-    return "\n".join(lines)
-
-
-def render_run_summary(idx: int, run: V1TaskSummary) -> str:
-    """One-line summary for a run in the list view."""
-    icon = STATUS_ICON.get(run.status, "?")
-    name = run.display_name or run.workflow_name or "?"
-    run_id = run.workflow_run_external_id or "?"
-    dur = _fmt_duration(run.duration)
-    started = _fmt_time(run.started_at)
-    meta = ""
-    if run.additional_metadata:
-        meta_parts = [f"{k}=`{v}`" for k, v in run.additional_metadata.items()]
-        meta = f"  ({', '.join(meta_parts)})"
-    return (
-        f"  {idx}. {icon} **{name}** started={started} dur={dur}{meta}\n"
-        f"     `{run_id}`"
-    )
-
-
-async def _fetch_run_list(
-    count: int = 5,
-    statuses: list[V1TaskStatus] | None = None,
-) -> list[V1TaskSummary]:
-    client = HatchetClientManager.get_client()
-    since = datetime.now(timezone.utc) - timedelta(days=7)
-    runs = await client.runs.aio_list(
-        since=since,
-        statuses=statuses,
-        limit=count,
-    )
-    return runs.rows or []
-
-
-async def list_recent_runs(
-    count: int = 5,
-    statuses: list[V1TaskStatus] | None = None,
-) -> str:
-    """List recent workflow runs as text."""
-    rows = await _fetch_run_list(count, statuses)
-
-    if not rows:
-        return "No runs found in the last 7 days."
-
-    lines = [f"Recent runs ({len(rows)}):", ""]
-    for i, run in enumerate(rows, 1):
-        lines.append(render_run_summary(i, run))
-
-    lines.append("")
-    lines.append("Use `--show N` to see full DAG for run N")
-    return "\n".join(lines)
-
-
-async def show_run(workflow_run_id: str) -> str:
-    """Fetch and render a single run."""
-    client = HatchetClientManager.get_client()
-    details = await client.runs.aio_get(workflow_run_id)
-    return render_run_detail(details)
-
-
-async def show_nth_run(
-    n: int,
-    count: int = 5,
-    statuses: list[V1TaskStatus] | None = None,
-) -> str:
-    """Fetch list, then drill into Nth run."""
-    rows = await _fetch_run_list(count, statuses)
-
-    if not rows:
-        return "No runs found in the last 7 days."
-    if n < 1 or n > len(rows):
-        return f"Invalid index {n}. Have {len(rows)} runs (1-{len(rows)})."
-
-    run = rows[n - 1]
-    return await show_run(run.workflow_run_external_id)
-
-
-async def main_async(args: argparse.Namespace) -> None:
-    statuses = [V1TaskStatus(args.status)] if args.status else None
-
-    if args.run_id:
-        output = await show_run(args.run_id)
-    elif args.show is not None:
-        output = await show_nth_run(args.show, count=args.last, statuses=statuses)
-    else:
-        output = await list_recent_runs(count=args.last, statuses=statuses)
-
-    print(output)
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(
-        description="Render Hatchet workflow runs as text DAG"
-    )
-    parser.add_argument(
-        "run_id",
-        nargs="?",
-        default=None,
-        help="Workflow run ID to show in detail. If omitted, lists recent runs.",
-    )
-    parser.add_argument(
-        "--show",
-        type=int,
-        default=None,
-        metavar="N",
-        help="Show full DAG for the Nth run in the list (1-indexed)",
-    )
-    parser.add_argument(
-        "--last",
-        type=int,
-        default=5,
-        help="Number of recent runs to list (default: 5)",
-    )
-    parser.add_argument(
-        "--status",
-        choices=["QUEUED", "RUNNING", "COMPLETED", "FAILED", "CANCELLED"],
-        help="Filter by status",
-    )
-
-    args = parser.parse_args()
-    asyncio.run(main_async(args))
-
-
-if __name__ == "__main__":
-    main()

server/reflector/worker/ics_sync.py

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

server/reflector/worker/process.py

@@ -25,7 +25,6 @@ from reflector.db.transcripts import (
     transcripts_controller,
 )
 from reflector.hatchet.client import HatchetClientManager
-from reflector.hatchet.dag_zulip import create_dag_zulip_message
 from reflector.pipelines.main_file_pipeline import task_pipeline_file_process
 from reflector.pipelines.main_live_pipeline import asynctask
 from reflector.pipelines.topic_processing import EmptyPipeline
@@ -373,16 +372,6 @@ async def _process_multitrack_recording_inner(
 
     await transcripts_controller.update(transcript, {"workflow_run_id": workflow_id})
 
-    try:
-        await create_dag_zulip_message(transcript.id, workflow_id)
-    except Exception:
-        logger.warning(
-            "[DAG Zulip] Failed to create DAG message at dispatch",
-            transcript_id=transcript.id,
-            workflow_id=workflow_id,
-            exc_info=True,
-        )
-
 
 @shared_task
 @asynctask
@@ -1090,16 +1079,6 @@ async def reprocess_failed_daily_recordings():
                 transcript, {"workflow_run_id": workflow_id}
             )
 
-            try:
-                await create_dag_zulip_message(transcript.id, workflow_id)
-            except Exception:
-                logger.warning(
-                    "[DAG Zulip] Failed to create DAG message at reprocess dispatch",
-                    transcript_id=transcript.id,
-                    workflow_id=workflow_id,
-                    exc_info=True,
-                )
-
             logger.info(
                 "Queued Daily recording for Hatchet reprocessing",
                 recording_id=recording.id,

server/reflector/zulip.py

@@ -12,16 +12,9 @@ class InvalidMessageError(Exception):
     pass
 
 
-def _zulip_client() -> httpx.AsyncClient:
-    headers = {}
-    if settings.ZULIP_HOST_HEADER:
-        headers["Host"] = settings.ZULIP_HOST_HEADER
-    return httpx.AsyncClient(verify=False, headers=headers)
-
-
 async def get_zulip_topics(stream_id: int) -> list[dict]:
     try:
-        async with _zulip_client() as client:
+        async with httpx.AsyncClient() as client:
             response = await client.get(
                 f"https://{settings.ZULIP_REALM}/api/v1/users/me/{stream_id}/topics",
                 auth=(settings.ZULIP_BOT_EMAIL, settings.ZULIP_API_KEY),
@@ -36,7 +29,7 @@ async def get_zulip_topics(stream_id: int) -> list[dict]:
 
 async def get_zulip_streams() -> list[dict]:
     try:
-        async with _zulip_client() as client:
+        async with httpx.AsyncClient() as client:
             response = await client.get(
                 f"https://{settings.ZULIP_REALM}/api/v1/streams",
                 auth=(settings.ZULIP_BOT_EMAIL, settings.ZULIP_API_KEY),
@@ -51,7 +44,7 @@ async def get_zulip_streams() -> list[dict]:
 
 async def send_message_to_zulip(stream: str, topic: str, content: str):
     try:
-        async with _zulip_client() as client:
+        async with httpx.AsyncClient() as client:
             response = await client.post(
                 f"https://{settings.ZULIP_REALM}/api/v1/messages",
                 data={
@@ -73,7 +66,7 @@ async def send_message_to_zulip(stream: str, topic: str, content: str):
 
 async def update_zulip_message(message_id: int, stream: str, topic: str, content: str):
     try:
-        async with _zulip_client() as client:
+        async with httpx.AsyncClient() as client:
             response = await client.patch(
                 f"https://{settings.ZULIP_REALM}/api/v1/messages/{message_id}",
                 data={
@@ -97,27 +90,6 @@ async def update_zulip_message(message_id: int, stream: str, topic: str, content
         raise Exception(f"Failed to update Zulip message: {error}")
 
 
-async def delete_zulip_message(message_id: int):
-    try:
-        async with _zulip_client() as client:
-            response = await client.delete(
-                f"https://{settings.ZULIP_REALM}/api/v1/messages/{message_id}",
-                auth=(settings.ZULIP_BOT_EMAIL, settings.ZULIP_API_KEY),
-            )
-
-            if (
-                response.status_code == 400
-                and response.json()["msg"] == "Invalid message(s)"
-            ):
-                raise InvalidMessageError(f"There is no message with id: {message_id}")
-
-            response.raise_for_status()
-
-            return response.json()
-    except httpx.RequestError as error:
-        raise Exception(f"Failed to delete Zulip message: {error}")
-
-
 def get_zulip_message(transcript: Transcript, include_topics: bool):
     transcript_url = f"{settings.UI_BASE_URL}/transcripts/{transcript.id}"
 

server/tests/test_dag_zulip.py

@@ -1,536 +0,0 @@
-"""
-Tests for Hatchet DAG Status -> Zulip Live Updates.
-
-Tests cover:
-- _dag_zulip_enabled() guard logic
-- create_dag_zulip_message: sends + stores message ID
-- update_dag_zulip_message: updates existing; noop when no message_id
-- delete_dag_zulip_message: deletes + clears; handles InvalidMessageError
-- delete_zulip_message: sends HTTP DELETE; raises on 400
-- with_error_handling integration: calls update after success + failure
-"""
-
-from unittest.mock import AsyncMock, MagicMock, patch
-
-import pytest
-
-from reflector.db.transcripts import Transcript
-
-
-@pytest.fixture
-def dag_settings():
-    """Patch settings for DAG Zulip tests."""
-    with patch("reflector.hatchet.dag_zulip.settings") as mock_settings:
-        mock_settings.ZULIP_REALM = "zulip.example.com"
-        mock_settings.ZULIP_DAG_STREAM = "dag-stream"
-        mock_settings.ZULIP_DAG_TOPIC = "dag-topic"
-        mock_settings.ZULIP_BOT_EMAIL = "bot@example.com"
-        mock_settings.ZULIP_API_KEY = "fake-key"
-        yield mock_settings
-
-
-@pytest.fixture
-def dag_settings_disabled():
-    """Patch settings with DAG Zulip disabled."""
-    with patch("reflector.hatchet.dag_zulip.settings") as mock_settings:
-        mock_settings.ZULIP_REALM = "zulip.example.com"
-        mock_settings.ZULIP_DAG_STREAM = None
-        mock_settings.ZULIP_DAG_TOPIC = None
-        yield mock_settings
-
-
-@pytest.fixture
-def mock_transcript():
-    return Transcript(
-        id="test-transcript-id",
-        name="Test",
-        status="processing",
-        source_kind="room",
-        zulip_message_id=None,
-    )
-
-
-@pytest.fixture
-def mock_transcript_with_zulip_id():
-    return Transcript(
-        id="test-transcript-id",
-        name="Test",
-        status="processing",
-        source_kind="room",
-        zulip_message_id=42,
-    )
-
-
-class TestDagZulipEnabled:
-    def test_enabled_when_all_set(self, dag_settings):
-        from reflector.hatchet.dag_zulip import _dag_zulip_enabled
-
-        assert _dag_zulip_enabled() is True
-
-    def test_disabled_when_realm_missing(self, dag_settings):
-        dag_settings.ZULIP_REALM = None
-        from reflector.hatchet.dag_zulip import _dag_zulip_enabled
-
-        assert _dag_zulip_enabled() is False
-
-    def test_disabled_when_stream_missing(self, dag_settings):
-        dag_settings.ZULIP_DAG_STREAM = None
-        from reflector.hatchet.dag_zulip import _dag_zulip_enabled
-
-        assert _dag_zulip_enabled() is False
-
-    def test_disabled_when_topic_missing(self, dag_settings):
-        dag_settings.ZULIP_DAG_TOPIC = None
-        from reflector.hatchet.dag_zulip import _dag_zulip_enabled
-
-        assert _dag_zulip_enabled() is False
-
-
-@pytest.mark.usefixtures("setup_database")
-@pytest.mark.asyncio
-class TestCreateDagZulipMessage:
-    async def test_sends_and_stores_message_id(self, dag_settings, mock_transcript):
-        mock_run_details = MagicMock()
-        rendered_md = "**DAG** rendered"
-
-        with (
-            patch(
-                "reflector.hatchet.client.HatchetClientManager.get_client"
-            ) as mock_get_client,
-            patch(
-                "reflector.tools.render_hatchet_run.render_run_detail",
-                return_value=rendered_md,
-            ),
-            patch(
-                "reflector.zulip.send_message_to_zulip",
-                new_callable=AsyncMock,
-                return_value={"id": 99},
-            ) as mock_send,
-            patch(
-                "reflector.db.transcripts.transcripts_controller.get_by_id",
-                new_callable=AsyncMock,
-                return_value=mock_transcript,
-            ),
-            patch(
-                "reflector.db.transcripts.transcripts_controller.update",
-                new_callable=AsyncMock,
-            ) as mock_update,
-        ):
-            mock_client = MagicMock()
-            mock_client.runs.aio_get = AsyncMock(return_value=mock_run_details)
-            mock_get_client.return_value = mock_client
-
-            from reflector.hatchet.dag_zulip import create_dag_zulip_message
-
-            await create_dag_zulip_message("test-transcript-id", "workflow-run-123")
-
-            mock_send.assert_called_once_with("dag-stream", "dag-topic", rendered_md)
-            mock_update.assert_called_once_with(
-                mock_transcript, {"zulip_message_id": 99}
-            )
-
-    async def test_noop_when_disabled(self, dag_settings_disabled):
-        with patch(
-            "reflector.zulip.send_message_to_zulip",
-            new_callable=AsyncMock,
-        ) as mock_send:
-            from reflector.hatchet.dag_zulip import create_dag_zulip_message
-
-            await create_dag_zulip_message("test-transcript-id", "workflow-run-123")
-            mock_send.assert_not_called()
-
-    async def test_logs_warning_on_failure(self, dag_settings, mock_transcript):
-        with (
-            patch(
-                "reflector.hatchet.client.HatchetClientManager.get_client"
-            ) as mock_get_client,
-            patch(
-                "reflector.tools.render_hatchet_run.render_run_detail",
-                return_value="rendered",
-            ),
-            patch(
-                "reflector.zulip.send_message_to_zulip",
-                new_callable=AsyncMock,
-                side_effect=Exception("Zulip down"),
-            ),
-            patch(
-                "reflector.db.transcripts.transcripts_controller.get_by_id",
-                new_callable=AsyncMock,
-                return_value=mock_transcript,
-            ),
-            patch("reflector.hatchet.dag_zulip.logger") as mock_logger,
-        ):
-            mock_client = MagicMock()
-            mock_client.runs.aio_get = AsyncMock(return_value=MagicMock())
-            mock_get_client.return_value = mock_client
-
-            from reflector.hatchet.dag_zulip import create_dag_zulip_message
-
-            # Should not raise
-            await create_dag_zulip_message("test-transcript-id", "workflow-run-123")
-            mock_logger.warning.assert_called()
-
-
-@pytest.mark.usefixtures("setup_database")
-@pytest.mark.asyncio
-class TestUpdateDagZulipMessage:
-    async def test_updates_existing_message(
-        self, dag_settings, mock_transcript_with_zulip_id
-    ):
-        mock_run_details = MagicMock()
-        rendered_md = "**DAG** updated"
-
-        with (
-            patch(
-                "reflector.hatchet.client.HatchetClientManager.get_client"
-            ) as mock_get_client,
-            patch(
-                "reflector.tools.render_hatchet_run.render_run_detail",
-                return_value=rendered_md,
-            ),
-            patch(
-                "reflector.zulip.update_zulip_message",
-                new_callable=AsyncMock,
-            ) as mock_update,
-            patch(
-                "reflector.db.transcripts.transcripts_controller.get_by_id",
-                new_callable=AsyncMock,
-                return_value=mock_transcript_with_zulip_id,
-            ),
-            patch(
-                "reflector.hatchet.workflows.daily_multitrack_pipeline.fresh_db_connection"
-            ) as mock_fresh_db,
-        ):
-            mock_client = MagicMock()
-            mock_client.runs.aio_get = AsyncMock(return_value=mock_run_details)
-            mock_get_client.return_value = mock_client
-            mock_fresh_db.return_value.__aenter__ = AsyncMock()
-            mock_fresh_db.return_value.__aexit__ = AsyncMock(return_value=False)
-
-            from reflector.hatchet.dag_zulip import update_dag_zulip_message
-
-            await update_dag_zulip_message("test-transcript-id", "workflow-run-123")
-
-            mock_update.assert_called_once_with(
-                42, "dag-stream", "dag-topic", rendered_md
-            )
-
-    async def test_appends_error_banner(
-        self, dag_settings, mock_transcript_with_zulip_id
-    ):
-        mock_run_details = MagicMock()
-        rendered_md = "**DAG** updated"
-
-        with (
-            patch(
-                "reflector.hatchet.client.HatchetClientManager.get_client"
-            ) as mock_get_client,
-            patch(
-                "reflector.tools.render_hatchet_run.render_run_detail",
-                return_value=rendered_md,
-            ),
-            patch(
-                "reflector.zulip.update_zulip_message",
-                new_callable=AsyncMock,
-            ) as mock_update,
-            patch(
-                "reflector.db.transcripts.transcripts_controller.get_by_id",
-                new_callable=AsyncMock,
-                return_value=mock_transcript_with_zulip_id,
-            ),
-            patch(
-                "reflector.hatchet.workflows.daily_multitrack_pipeline.fresh_db_connection"
-            ) as mock_fresh_db,
-        ):
-            mock_client = MagicMock()
-            mock_client.runs.aio_get = AsyncMock(return_value=mock_run_details)
-            mock_get_client.return_value = mock_client
-            mock_fresh_db.return_value.__aenter__ = AsyncMock()
-            mock_fresh_db.return_value.__aexit__ = AsyncMock(return_value=False)
-
-            from reflector.hatchet.dag_zulip import update_dag_zulip_message
-
-            await update_dag_zulip_message(
-                "test-transcript-id",
-                "workflow-run-123",
-                error_message="get_recording failed: connection timeout",
-            )
-
-            call_args = mock_update.call_args
-            content = call_args[0][3]
-            assert rendered_md in content
-            assert "get_recording failed: connection timeout" in content
-
-    async def test_noop_when_no_message_id(self, dag_settings, mock_transcript):
-        with (
-            patch(
-                "reflector.zulip.update_zulip_message",
-                new_callable=AsyncMock,
-            ) as mock_update,
-            patch(
-                "reflector.db.transcripts.transcripts_controller.get_by_id",
-                new_callable=AsyncMock,
-                return_value=mock_transcript,
-            ),
-            patch(
-                "reflector.hatchet.workflows.daily_multitrack_pipeline.fresh_db_connection"
-            ) as mock_fresh_db,
-        ):
-            mock_fresh_db.return_value.__aenter__ = AsyncMock()
-            mock_fresh_db.return_value.__aexit__ = AsyncMock(return_value=False)
-
-            from reflector.hatchet.dag_zulip import update_dag_zulip_message
-
-            await update_dag_zulip_message("test-transcript-id", "workflow-run-123")
-            mock_update.assert_not_called()
-
-    async def test_noop_when_disabled(self, dag_settings_disabled):
-        with patch(
-            "reflector.zulip.update_zulip_message",
-            new_callable=AsyncMock,
-        ) as mock_update:
-            from reflector.hatchet.dag_zulip import update_dag_zulip_message
-
-            await update_dag_zulip_message("test-transcript-id", "workflow-run-123")
-            mock_update.assert_not_called()
-
-
-@pytest.mark.usefixtures("setup_database")
-@pytest.mark.asyncio
-class TestDeleteDagZulipMessage:
-    async def test_deletes_and_clears(
-        self, dag_settings, mock_transcript_with_zulip_id
-    ):
-        with (
-            patch(
-                "reflector.zulip.delete_zulip_message",
-                new_callable=AsyncMock,
-            ) as mock_delete,
-            patch(
-                "reflector.db.transcripts.transcripts_controller.get_by_id",
-                new_callable=AsyncMock,
-                return_value=mock_transcript_with_zulip_id,
-            ),
-            patch(
-                "reflector.db.transcripts.transcripts_controller.update",
-                new_callable=AsyncMock,
-            ) as mock_tc_update,
-        ):
-            from reflector.hatchet.dag_zulip import delete_dag_zulip_message
-
-            await delete_dag_zulip_message("test-transcript-id")
-
-            mock_delete.assert_called_once_with(42)
-            mock_tc_update.assert_called_once_with(
-                mock_transcript_with_zulip_id, {"zulip_message_id": None}
-            )
-
-    async def test_noop_when_no_message_id(self, dag_settings, mock_transcript):
-        with (
-            patch(
-                "reflector.zulip.delete_zulip_message",
-                new_callable=AsyncMock,
-            ) as mock_delete,
-            patch(
-                "reflector.db.transcripts.transcripts_controller.get_by_id",
-                new_callable=AsyncMock,
-                return_value=mock_transcript,
-            ),
-        ):
-            from reflector.hatchet.dag_zulip import delete_dag_zulip_message
-
-            await delete_dag_zulip_message("test-transcript-id")
-            mock_delete.assert_not_called()
-
-    async def test_handles_invalid_message_error(
-        self, dag_settings, mock_transcript_with_zulip_id
-    ):
-        from reflector.zulip import InvalidMessageError
-
-        with (
-            patch(
-                "reflector.zulip.delete_zulip_message",
-                new_callable=AsyncMock,
-                side_effect=InvalidMessageError("gone"),
-            ),
-            patch(
-                "reflector.db.transcripts.transcripts_controller.get_by_id",
-                new_callable=AsyncMock,
-                return_value=mock_transcript_with_zulip_id,
-            ),
-            patch(
-                "reflector.db.transcripts.transcripts_controller.update",
-                new_callable=AsyncMock,
-            ) as mock_tc_update,
-            patch("reflector.hatchet.dag_zulip.logger"),
-        ):
-            from reflector.hatchet.dag_zulip import delete_dag_zulip_message
-
-            # Should not raise; should still clear the message_id
-            await delete_dag_zulip_message("test-transcript-id")
-            mock_tc_update.assert_called_once_with(
-                mock_transcript_with_zulip_id, {"zulip_message_id": None}
-            )
-
-    async def test_noop_when_disabled(self, dag_settings_disabled):
-        with patch(
-            "reflector.zulip.delete_zulip_message",
-            new_callable=AsyncMock,
-        ) as mock_delete:
-            from reflector.hatchet.dag_zulip import delete_dag_zulip_message
-
-            await delete_dag_zulip_message("test-transcript-id")
-            mock_delete.assert_not_called()
-
-
-@pytest.mark.asyncio
-class TestDeleteZulipMessage:
-    async def test_sends_delete_request(self):
-        mock_response = MagicMock()
-        mock_response.status_code = 200
-        mock_response.raise_for_status = MagicMock()
-        mock_response.json.return_value = {"result": "success"}
-
-        mock_client = AsyncMock()
-        mock_client.delete = AsyncMock(return_value=mock_response)
-        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
-        mock_client.__aexit__ = AsyncMock(return_value=False)
-
-        with patch("reflector.zulip.httpx.AsyncClient", return_value=mock_client):
-            with patch("reflector.zulip.settings") as mock_settings:
-                mock_settings.ZULIP_REALM = "zulip.example.com"
-                mock_settings.ZULIP_BOT_EMAIL = "bot@example.com"
-                mock_settings.ZULIP_API_KEY = "fake-key"
-
-                from reflector.zulip import delete_zulip_message
-
-                result = await delete_zulip_message(123)
-                assert result == {"result": "success"}
-
-                mock_client.delete.assert_called_once()
-                call_args = mock_client.delete.call_args
-                assert "123" in call_args.args[0]
-
-    async def test_raises_invalid_message_on_400(self):
-        from reflector.zulip import InvalidMessageError
-
-        mock_response = MagicMock()
-        mock_response.status_code = 400
-        mock_response.json.return_value = {"msg": "Invalid message(s)"}
-
-        mock_client = AsyncMock()
-        mock_client.delete = AsyncMock(return_value=mock_response)
-        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
-        mock_client.__aexit__ = AsyncMock(return_value=False)
-
-        with patch("reflector.zulip.httpx.AsyncClient", return_value=mock_client):
-            with patch("reflector.zulip.settings") as mock_settings:
-                mock_settings.ZULIP_REALM = "zulip.example.com"
-                mock_settings.ZULIP_BOT_EMAIL = "bot@example.com"
-                mock_settings.ZULIP_API_KEY = "fake-key"
-
-                from reflector.zulip import delete_zulip_message
-
-                with pytest.raises(InvalidMessageError):
-                    await delete_zulip_message(999)
-
-
-@pytest.mark.asyncio
-class TestWithErrorHandlingDagUpdate:
-    """Test that with_error_handling calls update_dag_zulip_message."""
-
-    async def test_calls_update_on_success(self):
-        from reflector.hatchet.constants import TaskName
-        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
-            PipelineInput,
-            with_error_handling,
-        )
-
-        mock_ctx = MagicMock()
-        mock_ctx.workflow_run_id = "wfr-123"
-
-        input_data = PipelineInput(
-            recording_id="rec-1",
-            tracks=[{"s3_key": "k"}],
-            bucket_name="bucket",
-            transcript_id="tid-1",
-        )
-
-        @with_error_handling(TaskName.GET_RECORDING)
-        async def fake_task(input: PipelineInput, ctx) -> str:
-            return "ok"
-
-        with patch(
-            "reflector.hatchet.workflows.daily_multitrack_pipeline.update_dag_zulip_message",
-            new_callable=AsyncMock,
-        ) as mock_update:
-            result = await fake_task(input_data, mock_ctx)
-            assert result == "ok"
-            mock_update.assert_called_once_with("tid-1", "wfr-123")
-
-    async def test_calls_update_on_failure_with_error_message(self):
-        from reflector.hatchet.constants import TaskName
-        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
-            PipelineInput,
-            with_error_handling,
-        )
-
-        mock_ctx = MagicMock()
-        mock_ctx.workflow_run_id = "wfr-123"
-
-        input_data = PipelineInput(
-            recording_id="rec-1",
-            tracks=[{"s3_key": "k"}],
-            bucket_name="bucket",
-            transcript_id="tid-1",
-        )
-
-        @with_error_handling(TaskName.GET_RECORDING)
-        async def failing_task(input: PipelineInput, ctx) -> str:
-            raise ValueError("boom")
-
-        with (
-            patch(
-                "reflector.hatchet.workflows.daily_multitrack_pipeline.update_dag_zulip_message",
-                new_callable=AsyncMock,
-            ) as mock_update,
-            patch(
-                "reflector.hatchet.workflows.daily_multitrack_pipeline.set_workflow_error_status",
-                new_callable=AsyncMock,
-            ),
-        ):
-            with pytest.raises(ValueError, match="boom"):
-                await failing_task(input_data, mock_ctx)
-            mock_update.assert_called_once_with(
-                "tid-1", "wfr-123", error_message="get_recording failed: boom"
-            )
-
-    async def test_dag_failure_doesnt_affect_task(self):
-        """DAG update failure should not prevent task from succeeding."""
-        from reflector.hatchet.constants import TaskName
-        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
-            PipelineInput,
-            with_error_handling,
-        )
-
-        mock_ctx = MagicMock()
-        mock_ctx.workflow_run_id = "wfr-123"
-
-        input_data = PipelineInput(
-            recording_id="rec-1",
-            tracks=[{"s3_key": "k"}],
-            bucket_name="bucket",
-            transcript_id="tid-1",
-        )
-
-        @with_error_handling(TaskName.GET_RECORDING)
-        async def ok_task(input: PipelineInput, ctx) -> str:
-            return "ok"
-
-        with patch(
-            "reflector.hatchet.workflows.daily_multitrack_pipeline.update_dag_zulip_message",
-            new_callable=AsyncMock,
-            side_effect=Exception("zulip exploded"),
-        ):
-            result = await ok_task(input_data, mock_ctx)
-            assert result == "ok"

server/tests/test_ics_dedup.py

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

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

@@ -8,7 +8,7 @@ import {
   useRef,
   useState,
 } from "react";
-import { Box, Spinner, Center, Text } from "@chakra-ui/react";
+import { Box, Spinner, Center, Text, Button, VStack } from "@chakra-ui/react";
 import { useRouter, useParams } from "next/navigation";
 import DailyIframe, {
   DailyCall,
@@ -16,10 +16,13 @@ import DailyIframe, {
   DailyCustomTrayButton,
   DailyCustomTrayButtons,
   DailyEventObjectCustomButtonClick,
+  DailyEventObjectFatalError,
+  DailyFatalErrorType,
   DailyFactoryOptions,
   DailyParticipantsObject,
 } from "@daily-co/daily-js";
 import type { components } from "../../reflector-api";
+import { printApiError, ApiError } from "../../api/_error";
 import { useAuth } from "../../lib/AuthProvider";
 import { useConsentDialog } from "../../lib/consent";
 import {
@@ -45,6 +48,63 @@ const RAW_TRACKS_NAMESPACE = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
 const RECORDING_START_DELAY_MS = 2000;
 const RECORDING_START_MAX_RETRIES = 5;
 
+const FATAL_ERROR_MESSAGES: Partial<
+  Record<DailyFatalErrorType, { message: string; rejoinable?: boolean }>
+> = {
+  "connection-error": {
+    message: "Connection lost. Please check your network.",
+    rejoinable: true,
+  },
+  "exp-room": { message: "The meeting time has ended." },
+  "exp-token": { message: "Your session has expired.", rejoinable: true },
+  ejected: { message: "You were removed from this meeting." },
+  "meeting-full": { message: "This meeting is full." },
+  "not-allowed": { message: "You are not allowed to join this meeting." },
+  "nbf-room": { message: "This meeting hasn't started yet." },
+  "nbf-token": { message: "This meeting hasn't started yet." },
+  "no-room": { message: "This room does not exist." },
+  "end-of-life": { message: "This meeting room is no longer available." },
+};
+
+function FatalErrorScreen({
+  error,
+  roomName,
+}: {
+  error: FatalError;
+  roomName: string;
+}) {
+  const router = useRouter();
+  const info =
+    error.type !== "unknown" ? FATAL_ERROR_MESSAGES[error.type] : undefined;
+  const message = info?.message ?? `Something went wrong: ${error.message}`;
+  const rejoinable = info?.rejoinable ?? false;
+
+  return (
+    <Center width="100vw" height="100vh">
+      <VStack gap={4}>
+        <Text color="red.500">{message}</Text>
+        {rejoinable ? (
+          <>
+            <Button onClick={() => window.location.reload()}>
+              Try Rejoining
+            </Button>
+            <Button
+              variant="outline"
+              onClick={() => router.push(`/${roomName}`)}
+            >
+              Leave
+            </Button>
+          </>
+        ) : (
+          <Button onClick={() => router.push(`/${roomName}`)}>
+            Back to Room
+          </Button>
+        )}
+      </VStack>
+    </Center>
+  );
+}
+
 type Meeting = components["schemas"]["Meeting"];
 type Room = components["schemas"]["RoomDetails"];
 
@@ -82,6 +142,8 @@ const USE_FRAME_INIT_STATE = {
   joined: false as boolean,
 } as const;
 
+type FatalError = { type: DailyFatalErrorType | "unknown"; message: string };
+
 // Daily js and not Daily react used right now because daily-js allows for prebuild interface vs. -react is customizable but has no nice defaults
 const useFrame = (
   container: HTMLDivElement | null,
@@ -89,6 +151,7 @@ const useFrame = (
     onLeftMeeting: () => void;
     onCustomButtonClick: (ev: DailyEventObjectCustomButtonClick) => void;
     onJoinMeeting: () => void;
+    onError: (ev: DailyEventObjectFatalError) => void;
   },
 ) => {
   const [{ frame, joined }, setState] = useState(USE_FRAME_INIT_STATE);
@@ -134,6 +197,7 @@ const useFrame = (
     if (!frame) return;
     frame.on("left-meeting", cbs.onLeftMeeting);
     frame.on("custom-button-click", cbs.onCustomButtonClick);
+    frame.on("error", cbs.onError);
     const joinCb = () => {
       if (!frame) {
         console.error("frame is null in joined-meeting callback");
@@ -145,6 +209,7 @@ const useFrame = (
     return () => {
       frame.off("left-meeting", cbs.onLeftMeeting);
       frame.off("custom-button-click", cbs.onCustomButtonClick);
+      frame.off("error", cbs.onError);
       frame.off("joined-meeting", joinCb);
     };
   }, [frame, cbs]);
@@ -188,6 +253,7 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
   const joinMutation = useRoomJoinMeeting();
   const startRecordingMutation = useMeetingStartRecording();
   const [joinedMeeting, setJoinedMeeting] = useState<Meeting | null>(null);
+  const [fatalError, setFatalError] = useState<FatalError | null>(null);
 
   // Generate deterministic instanceIds so all participants use SAME IDs
   const cloudInstanceId = parseNonEmptyString(meeting.id);
@@ -234,8 +300,18 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
   const roomUrl = joinedMeeting?.room_url;
 
   const handleLeave = useCallback(() => {
+    // If a fatal error occurred, don't redirect — let the error UI show
+    if (fatalError) return;
     router.push("/browse");
-  }, [router]);
+  }, [router, fatalError]);
+
+  const handleError = useCallback((ev: DailyEventObjectFatalError) => {
+    const error: FatalError = {
+      type: ev.error?.type ?? "unknown",
+      message: ev.errorMsg,
+    };
+    setFatalError(error);
+  }, []);
 
   const handleCustomButtonClick = useCallback(
     (ev: DailyEventObjectCustomButtonClick) => {
@@ -324,6 +400,7 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
     onLeftMeeting: handleLeave,
     onCustomButtonClick: handleCustomButtonClick,
     onJoinMeeting: handleFrameJoinMeeting,
+    onError: handleError,
   });
 
   useEffect(() => {
@@ -380,13 +457,27 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
   }
 
   if (joinMutation.isError) {
+    const apiDetail = printApiError(
+      joinMutation.error as /*ref 095959E6-01CC-4CF0-B3A9-F65F12F046D3*/ ApiError,
+    );
     return (
       <Center width="100vw" height="100vh">
-        <Text color="red.500">Failed to join meeting. Please try again.</Text>
+        <VStack gap={4}>
+          <Text color="red.500">
+            {apiDetail ?? "Failed to join meeting. Please try again."}
+          </Text>
+          <Button onClick={() => router.push(`/${roomName}`)}>
+            Back to Room
+          </Button>
+        </VStack>
       </Center>
     );
   }
 
+  if (fatalError) {
+    return <FatalErrorScreen error={fatalError} roomName={roomName} />;
+  }
+
   if (!roomUrl) {
     return null;
   }

www/app/lib/apiHooks.ts

@@ -9,6 +9,7 @@ import { MeetingId } from "./types";
 import { NonEmptyString } from "./utils";
 
 /*
+ * ref 095959E6-01CC-4CF0-B3A9-F65F12F046D3
  * XXX error types returned from the hooks are not always correct; declared types are ValidationError but real type could be string or any other
  * this is either a limitation or incorrect usage of Python json schema generator
  * or, limitation or incorrect usage of .d type generator from json schema