fix: send last DAG_STATUS on WebSocket connect instead of skipping all

Previously all DAG_STATUS events were skipped during historical replay
on WS connect, so reconnecting clients (React strict mode remount,
page navigation) lost current DAG state. Now sends only the most
recent DAG_STATUS event on connect.

server/reflector/db/search.py

@@ -1,6 +1,7 @@
 """Search functionality for transcripts and other entities."""
 
 import itertools
+import json
 from dataclasses import dataclass
 from datetime import datetime
 from io import StringIO
@@ -172,6 +173,9 @@ class SearchResult(BaseModel):
     total_match_count: NonNegativeInt = Field(
         default=0, description="Total number of matches found in the transcript"
     )
+    dag_status: list[dict] | None = Field(
+        default=None, description="Latest DAG task status for processing transcripts"
+    )
 
     @field_serializer("created_at", when_used="json")
     def serialize_datetime(self, dt: datetime) -> str:
@@ -328,6 +332,42 @@ class SnippetGenerator:
         return summary_snippets + webvtt_snippets, total_matches
 
 
+async def _fetch_dag_statuses(transcript_ids: list[str]) -> dict[str, list[dict]]:
+    """Fetch latest DAG_STATUS event data for given transcript IDs.
+
+    Returns dict mapping transcript_id -> tasks list from the last DAG_STATUS event.
+    """
+    if not transcript_ids:
+        return {}
+
+    db = get_database()
+    query = sqlalchemy.select(
+        [
+            transcripts.c.id,
+            transcripts.c.events,
+        ]
+    ).where(transcripts.c.id.in_(transcript_ids))
+
+    rows = await db.fetch_all(query)
+    result: dict[str, list[dict]] = {}
+
+    for row in rows:
+        events_raw = row["events"]
+        if not events_raw:
+            continue
+        # events is stored as JSON list
+        events = events_raw if isinstance(events_raw, list) else json.loads(events_raw)
+        # Find last DAG_STATUS event
+        for ev in reversed(events):
+            if isinstance(ev, dict) and ev.get("event") == "DAG_STATUS":
+                tasks = ev.get("data", {}).get("tasks")
+                if tasks:
+                    result[row["id"]] = tasks
+                break
+
+    return result
+
+
 class SearchController:
     """Controller for search operations across different entities."""
 
@@ -470,6 +510,14 @@ class SearchController:
             logger.error(f"Error processing search results: {e}", exc_info=True)
             raise
 
+        # Enrich processing transcripts with DAG status
+        processing_ids = [r.id for r in results if r.status == "processing"]
+        if processing_ids:
+            dag_statuses = await _fetch_dag_statuses(processing_ids)
+            for r in results:
+                if r.id in dag_statuses:
+                    r.dag_status = dag_statuses[r.id]
+
         return results, total
 
 

server/reflector/db/transcripts.py

@@ -234,7 +234,7 @@ class Transcript(BaseModel):
         return dt.isoformat()
 
     def add_event(self, event: str, data: BaseModel) -> TranscriptEvent:
-        ev = TranscriptEvent(event=event, data=data.model_dump())
+        ev = TranscriptEvent(event=event, data=data.model_dump(mode="json"))
         self.events.append(ev)
         return ev
 

server/reflector/hatchet/broadcast.py

@@ -15,7 +15,7 @@ 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"}
+USER_ROOM_EVENTS = {"STATUS", "FINAL_TITLE", "DURATION", "DAG_STATUS"}
 
 
 async def broadcast_event(

server/reflector/hatchet/dag_progress.py

@@ -0,0 +1,230 @@
+"""
+DAG Progress Reporting — models and transform.
+
+Converts Hatchet V1WorkflowRunDetails into structured DagTask list
+for frontend WebSocket/REST consumption.
+
+Ported from render_hatchet_run.py (feat-dag-zulip) which renders markdown;
+this module produces structured Pydantic models instead.
+"""
+
+from datetime import datetime
+from enum import StrEnum
+
+from hatchet_sdk.clients.rest.models import (
+    V1TaskStatus,
+    V1WorkflowRunDetails,
+    WorkflowRunShapeItemForWorkflowRunDetails,
+)
+from pydantic import BaseModel
+
+
+class DagTaskStatus(StrEnum):
+    QUEUED = "queued"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+_HATCHET_TO_DAG_STATUS: dict[V1TaskStatus, DagTaskStatus] = {
+    V1TaskStatus.QUEUED: DagTaskStatus.QUEUED,
+    V1TaskStatus.RUNNING: DagTaskStatus.RUNNING,
+    V1TaskStatus.COMPLETED: DagTaskStatus.COMPLETED,
+    V1TaskStatus.FAILED: DagTaskStatus.FAILED,
+    V1TaskStatus.CANCELLED: DagTaskStatus.CANCELLED,
+}
+
+
+class DagTask(BaseModel):
+    name: str
+    status: DagTaskStatus
+    started_at: datetime | None
+    finished_at: datetime | None
+    duration_seconds: float | None
+    parents: list[str]
+    error: str | None
+    children_total: int | None
+    children_completed: int | None
+    progress_pct: float | None
+
+
+class DagStatusData(BaseModel):
+    workflow_run_id: str
+    tasks: list[DagTask]
+
+
+def _topo_sort(
+    shape: list[WorkflowRunShapeItemForWorkflowRunDetails],
+) -> list[str]:
+    """Topological sort of step_ids from shape DAG (Kahn's algorithm).
+
+    Ported from render_hatchet_run.py.
+    """
+    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 _extract_error_summary(error_message: str | None) -> str | None:
+    """Extract first meaningful line from error message, skipping traceback frames."""
+    if not error_message or not error_message.strip():
+        return None
+
+    err_lines = error_message.strip().split("\n")
+    err_summary = err_lines[0]
+    for line in err_lines:
+        stripped = line.strip()
+        if stripped and not stripped.startswith(("Traceback", "File ", "{", ")")):
+            err_summary = stripped
+    return err_summary
+
+
+def extract_dag_tasks(details: V1WorkflowRunDetails) -> list[DagTask]:
+    """Extract structured DagTask list from Hatchet workflow run details.
+
+    Returns tasks in topological order with status, timestamps, parents,
+    error summaries, and fan-out children counts.
+    """
+    shape = details.shape or []
+    tasks = details.tasks or []
+
+    if not shape:
+        return []
+
+    # 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: child -> parent names
+    parents_by_step: 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_by_step:
+                parents_by_step[child_id].append(step_to_name[s.step_id])
+
+    # Join tasks by step_id
+    from hatchet_sdk.clients.rest.models import V1TaskSummary  # noqa: PLC0415
+
+    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
+
+    ordered = _topo_sort(shape)
+
+    result: list[DagTask] = []
+    for step_id in ordered:
+        name = step_to_name[step_id]
+        t = task_by_step.get(step_id)
+
+        if not t:
+            result.append(
+                DagTask(
+                    name=name,
+                    status=DagTaskStatus.QUEUED,
+                    started_at=None,
+                    finished_at=None,
+                    duration_seconds=None,
+                    parents=parents_by_step.get(step_id, []),
+                    error=None,
+                    children_total=None,
+                    children_completed=None,
+                    progress_pct=None,
+                )
+            )
+            continue
+
+        status = _HATCHET_TO_DAG_STATUS.get(t.status, DagTaskStatus.QUEUED)
+
+        duration_seconds: float | None = None
+        if t.duration is not None:
+            duration_seconds = t.duration / 1000.0
+
+        # Fan-out children
+        children_total: int | None = None
+        children_completed: int | None = None
+        if t.num_spawned_children and t.num_spawned_children > 0:
+            children_total = t.num_spawned_children
+            children_completed = sum(
+                1 for c in (t.children or []) if c.status == V1TaskStatus.COMPLETED
+            )
+
+        result.append(
+            DagTask(
+                name=name,
+                status=status,
+                started_at=t.started_at,
+                finished_at=t.finished_at,
+                duration_seconds=duration_seconds,
+                parents=parents_by_step.get(step_id, []),
+                error=_extract_error_summary(t.error_message),
+                children_total=children_total,
+                children_completed=children_completed,
+                progress_pct=None,
+            )
+        )
+
+    return result
+
+
+async def broadcast_dag_status(transcript_id: str, workflow_run_id: str) -> None:
+    """Fetch current DAG state from Hatchet and broadcast via WebSocket.
+
+    Fire-and-forget: exceptions are logged but never raised.
+    All imports are deferred for fork-safety (Hatchet workers fork processes).
+    """
+    try:
+        from reflector.db.transcripts import transcripts_controller  # noqa: I001, PLC0415
+        from reflector.hatchet.broadcast import append_event_and_broadcast  # 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.logger import logger  # noqa: PLC0415
+
+        async with fresh_db_connection():
+            client = HatchetClientManager.get_client()
+            details = await client.runs.aio_get(workflow_run_id)
+            dag_tasks = extract_dag_tasks(details)
+            dag_status = DagStatusData(workflow_run_id=workflow_run_id, tasks=dag_tasks)
+
+            transcript = await transcripts_controller.get_by_id(transcript_id)
+            if transcript:
+                await append_event_and_broadcast(
+                    transcript_id,
+                    transcript,
+                    "DAG_STATUS",
+                    dag_status,
+                    logger,
+                )
+    except Exception:
+        from reflector.logger import logger  # noqa: PLC0415
+
+        logger.warning(
+            "[DAG Progress] Failed to broadcast DAG status",
+            transcript_id=transcript_id,
+            workflow_run_id=workflow_run_id,
+            exc_info=True,
+        )

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,
@@ -185,7 +184,10 @@ class Loggable(Protocol):
 
 
 def make_audio_progress_logger(
-    ctx: Loggable, task_name: TaskName, interval: float = 5.0
+    ctx: Loggable,
+    task_name: TaskName,
+    interval: float = 5.0,
+    transcript_id: str | None = None,
 ) -> Callable[[float | None, float], None]:
     """Create a throttled progress logger callback for audio processing.
 
@@ -193,6 +195,7 @@ def make_audio_progress_logger(
         ctx: Object with .log() method (e.g., Hatchet Context).
         task_name: Name to prefix in log messages.
         interval: Minimum seconds between log messages.
+        transcript_id: If provided, broadcasts transient DAG_TASK_PROGRESS events.
 
     Returns:
         Callback(progress_pct, audio_position) that logs at most every `interval` seconds.
@@ -214,6 +217,27 @@ def make_audio_progress_logger(
                 )
             last_log_time[0] = now
 
+        if transcript_id and progress_pct is not None:
+            try:
+                import asyncio  # noqa: PLC0415
+
+                from reflector.db.transcripts import TranscriptEvent  # noqa: PLC0415
+                from reflector.hatchet.broadcast import broadcast_event  # noqa: PLC0415
+
+                loop = asyncio.get_event_loop()
+                loop.create_task(
+                    broadcast_event(
+                        transcript_id,
+                        TranscriptEvent(
+                            event="DAG_TASK_PROGRESS",
+                            data={"task_name": task_name, "progress_pct": progress_pct},
+                        ),
+                        logger=logger,
+                    )
+                )
+            except Exception:
+                pass  # transient, never fail the callback
+
     return callback
 
 
@@ -238,12 +262,12 @@ def with_error_handling(
     ) -> Callable[[PipelineInput, Context], Coroutine[Any, Any, R]]:
         @functools.wraps(func)
         async def wrapper(input: PipelineInput, ctx: Context) -> R:
+            from reflector.hatchet.dag_progress import broadcast_dag_status  # noqa: I001, PLC0415
+
             try:
                 result = await func(input, ctx)
                 try:
-                    await update_dag_zulip_message(
-                        input.transcript_id, ctx.workflow_run_id
-                    )
+                    await broadcast_dag_status(input.transcript_id, ctx.workflow_run_id)
                 except Exception:
                     pass
                 return result
@@ -255,11 +279,7 @@ def with_error_handling(
                     exc_info=True,
                 )
                 try:
-                    await update_dag_zulip_message(
-                        input.transcript_id,
-                        ctx.workflow_run_id,
-                        error_message=f"{step_name} failed: {e}",
-                    )
+                    await broadcast_dag_status(input.transcript_id, ctx.workflow_run_id)
                 except Exception:
                     pass
                 if set_error_status:
@@ -576,7 +596,9 @@ async def mixdown_tracks(input: PipelineInput, ctx: Context) -> MixdownResult:
         target_sample_rate,
         offsets_seconds=None,
         logger=logger,
-        progress_callback=make_audio_progress_logger(ctx, TaskName.MIXDOWN_TRACKS),
+        progress_callback=make_audio_progress_logger(
+            ctx, TaskName.MIXDOWN_TRACKS, transcript_id=input.transcript_id
+        ),
         expected_duration_sec=recording_duration if recording_duration > 0 else None,
     )
     await writer.flush()
@@ -1310,11 +1332,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,17 +266,20 @@ async def dispatch_transcript_processing(
                 transcript, {"workflow_run_id": workflow_id}
             )
 
+        logger.info("Hatchet workflow dispatched", workflow_id=workflow_id)
+
         try:
-            await create_dag_zulip_message(config.transcript_id, workflow_id)
+            from reflector.hatchet.dag_progress import broadcast_dag_status  # noqa: I001, PLC0415
+
+            await broadcast_dag_status(config.transcript_id, workflow_id)
         except Exception:
             logger.warning(
-                "[DAG Zulip] Failed to create DAG message at dispatch",
+                "[DAG Progress] Failed initial broadcast",
                 transcript_id=config.transcript_id,
                 workflow_id=workflow_id,
                 exc_info=True,
             )
 
-        logger.info("Hatchet workflow dispatched", workflow_id=workflow_id)
         return None
 
     elif isinstance(config, FileProcessingConfig):

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/views/transcripts.py

@@ -111,6 +111,7 @@ class GetTranscriptMinimal(BaseModel):
     room_id: str | None = None
     room_name: str | None = None
     audio_deleted: bool | None = None
+    dag_status: list[dict] | None = None
 
 
 class TranscriptParticipantWithEmail(TranscriptParticipant):
@@ -491,6 +492,13 @@ async def transcript_get(
                 )
             )
 
+    dag_status = None
+    if transcript.status == "processing" and transcript.events:
+        for ev in reversed(transcript.events):
+            if ev.event == "DAG_STATUS":
+                dag_status = ev.data.get("tasks") if isinstance(ev.data, dict) else None
+                break
+
     base_data = {
         "id": transcript.id,
         "user_id": transcript.user_id,
@@ -512,6 +520,7 @@ async def transcript_get(
         "room_id": transcript.room_id,
         "room_name": room_name,
         "audio_deleted": transcript.audio_deleted,
+        "dag_status": dag_status,
         "participants": participants,
     }
 

server/reflector/views/transcripts_websocket.py

@@ -41,13 +41,19 @@ async def transcript_events_websocket(
 
     try:
         # on first connection, send all events only to the current user
+        # Find the last DAG_STATUS to send after other historical events
+        last_dag_status = None
         for event in transcript.events:
-            # for now, do not send TRANSCRIPT or STATUS options - theses are live event
-            # not necessary to be sent to the client; but keep the rest
             name = event.event
             if name in ("TRANSCRIPT", "STATUS"):
                 continue
+            if name == "DAG_STATUS":
+                last_dag_status = event
+                continue
             await websocket.send_json(event.model_dump(mode="json"))
+        # Send only the most recent DAG_STATUS so reconnecting clients get current state
+        if last_dag_status is not None:
+            await websocket.send_json(last_dag_status.model_dump(mode="json"))
 
         # XXX if transcript is final (locked=True and status=ended)
         # XXX send a final event to the client and close the connection

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_progress.py

@@ -0,0 +1,959 @@
+"""Tests for DAG progress models and transform function.
+
+Tests the extract_dag_tasks function that converts Hatchet V1WorkflowRunDetails
+into structured DagTask list for frontend consumption.
+"""
+
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from reflector.hatchet.constants import TaskName
+from reflector.hatchet.dag_progress import (
+    DagStatusData,
+    DagTask,
+    DagTaskStatus,
+    extract_dag_tasks,
+)
+
+
+def _make_shape_item(
+    step_id: str,
+    task_name: str,
+    children_step_ids: list[str] | None = None,
+) -> MagicMock:
+    """Create a mock WorkflowRunShapeItemForWorkflowRunDetails."""
+    item = MagicMock()
+    item.step_id = step_id
+    item.task_name = task_name
+    item.children_step_ids = children_step_ids or []
+    return item
+
+
+def _make_task_summary(
+    step_id: str,
+    status: str = "QUEUED",
+    started_at: datetime | None = None,
+    finished_at: datetime | None = None,
+    duration: int | None = None,
+    error_message: str | None = None,
+    task_external_id: str | None = None,
+    num_spawned_children: int | None = None,
+    children: list | None = None,
+) -> MagicMock:
+    """Create a mock V1TaskSummary."""
+    from hatchet_sdk.clients.rest.models import V1TaskStatus
+
+    task = MagicMock()
+    task.step_id = step_id
+    task.status = V1TaskStatus(status)
+    task.started_at = started_at
+    task.finished_at = finished_at
+    task.duration = duration
+    task.error_message = error_message
+    task.task_external_id = task_external_id or f"ext-{step_id}"
+    task.num_spawned_children = num_spawned_children
+    task.children = children or []
+    return task
+
+
+def _make_details(
+    shape: list,
+    tasks: list,
+    run_id: str = "test-run-id",
+) -> MagicMock:
+    """Create a mock V1WorkflowRunDetails."""
+    details = MagicMock()
+    details.shape = shape
+    details.tasks = tasks
+    details.task_events = []
+    details.run = MagicMock()
+    details.run.metadata = MagicMock()
+    details.run.metadata.id = run_id
+    return details
+
+
+class TestExtractDagTasksBasic:
+    """Test basic extraction of DAG tasks from workflow run details."""
+
+    def test_empty_shape_returns_empty_list(self):
+        details = _make_details(shape=[], tasks=[])
+        result = extract_dag_tasks(details)
+        assert result == []
+
+    def test_single_task_queued(self):
+        shape = [_make_shape_item("s1", "get_recording")]
+        tasks = [_make_task_summary("s1", status="QUEUED")]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        assert len(result) == 1
+        assert result[0].name == "get_recording"
+        assert result[0].status == DagTaskStatus.QUEUED
+        assert result[0].parents == []
+        assert result[0].started_at is None
+        assert result[0].finished_at is None
+        assert result[0].duration_seconds is None
+        assert result[0].error is None
+        assert result[0].children_total is None
+        assert result[0].children_completed is None
+        assert result[0].progress_pct is None
+
+    def test_completed_task_with_duration(self):
+        now = datetime.now(timezone.utc)
+        shape = [_make_shape_item("s1", "get_recording")]
+        tasks = [
+            _make_task_summary(
+                "s1",
+                status="COMPLETED",
+                started_at=now,
+                finished_at=now,
+                duration=1500,  # milliseconds
+            )
+        ]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        assert result[0].status == DagTaskStatus.COMPLETED
+        assert result[0].duration_seconds == 1.5
+        assert result[0].started_at == now
+        assert result[0].finished_at == now
+
+    def test_failed_task_with_error(self):
+        shape = [_make_shape_item("s1", "get_recording")]
+        tasks = [
+            _make_task_summary(
+                "s1",
+                status="FAILED",
+                error_message="Traceback (most recent call last):\n  File something\nConnectionError: connection refused",
+            )
+        ]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        assert result[0].status == DagTaskStatus.FAILED
+        assert result[0].error == "ConnectionError: connection refused"
+
+    def test_running_task(self):
+        now = datetime.now(timezone.utc)
+        shape = [_make_shape_item("s1", "mixdown_tracks")]
+        tasks = [
+            _make_task_summary(
+                "s1",
+                status="RUNNING",
+                started_at=now,
+                duration=5000,
+            )
+        ]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        assert result[0].status == DagTaskStatus.RUNNING
+        assert result[0].started_at == now
+        assert result[0].duration_seconds == 5.0
+
+    def test_cancelled_task(self):
+        shape = [_make_shape_item("s1", "post_zulip")]
+        tasks = [_make_task_summary("s1", status="CANCELLED")]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        assert result[0].status == DagTaskStatus.CANCELLED
+
+
+class TestExtractDagTasksTopology:
+    """Test topological ordering and parent extraction."""
+
+    def test_linear_chain_parents(self):
+        """A -> B -> C should produce correct parents."""
+        shape = [
+            _make_shape_item("s1", "get_recording", children_step_ids=["s2"]),
+            _make_shape_item("s2", "get_participants", children_step_ids=["s3"]),
+            _make_shape_item("s3", "process_tracks"),
+        ]
+        tasks = [
+            _make_task_summary("s1", status="COMPLETED"),
+            _make_task_summary("s2", status="COMPLETED"),
+            _make_task_summary("s3", status="QUEUED"),
+        ]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        assert [t.name for t in result] == [
+            "get_recording",
+            "get_participants",
+            "process_tracks",
+        ]
+        assert result[0].parents == []
+        assert result[1].parents == ["get_recording"]
+        assert result[2].parents == ["get_participants"]
+
+    def test_diamond_dag(self):
+        """
+        A -> B, A -> C, B -> D, C -> D
+        D should have parents [B, C] (or [C, B] depending on sort).
+        """
+        shape = [
+            _make_shape_item("s1", "get_recording", children_step_ids=["s2", "s3"]),
+            _make_shape_item("s2", "mixdown_tracks", children_step_ids=["s4"]),
+            _make_shape_item("s3", "detect_topics", children_step_ids=["s4"]),
+            _make_shape_item("s4", "finalize"),
+        ]
+        tasks = [
+            _make_task_summary("s1", status="COMPLETED"),
+            _make_task_summary("s2", status="RUNNING"),
+            _make_task_summary("s3", status="RUNNING"),
+            _make_task_summary("s4", status="QUEUED"),
+        ]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        # Topological: s1 first, s2/s3 in some order, s4 last
+        assert result[0].name == "get_recording"
+        assert result[-1].name == "finalize"
+        finalize = result[-1]
+        assert set(finalize.parents) == {"mixdown_tracks", "detect_topics"}
+
+    def test_topological_order_is_stable(self):
+        """Verify deterministic ordering (sorted queue in Kahn's)."""
+        shape = [
+            _make_shape_item("s_c", "task_c"),
+            _make_shape_item("s_a", "task_a", children_step_ids=["s_c"]),
+            _make_shape_item("s_b", "task_b", children_step_ids=["s_c"]),
+        ]
+        tasks = [
+            _make_task_summary("s_c", status="QUEUED"),
+            _make_task_summary("s_a", status="COMPLETED"),
+            _make_task_summary("s_b", status="COMPLETED"),
+        ]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        # s_a and s_b both roots with in-degree 0; sorted alphabetically by step_id
+        names = [t.name for t in result]
+        assert names[-1] == "task_c"
+        # First two should be task_a, task_b (sorted by step_id: s_a < s_b)
+        assert names[0] == "task_a"
+        assert names[1] == "task_b"
+
+    def test_production_dag_shape(self):
+        """Test the real 15-task pipeline topology with mixed statuses.
+
+        Simulates a mid-pipeline state where early tasks completed,
+        middle tasks running, and later tasks still queued.
+        """
+        # Production DAG edges (parent -> children):
+        # get_recording -> get_participants
+        # get_participants -> process_tracks
+        # process_tracks -> mixdown_tracks, detect_topics, finalize
+        # mixdown_tracks -> generate_waveform
+        # detect_topics -> generate_title, extract_subjects
+        # extract_subjects -> process_subjects, identify_action_items
+        # process_subjects -> generate_recap
+        # generate_title -> finalize
+        # generate_recap -> finalize
+        # identify_action_items -> finalize
+        # finalize -> cleanup_consent
+        # cleanup_consent -> post_zulip, send_webhook
+        shape = [
+            _make_shape_item(
+                "s_get_recording", TaskName.GET_RECORDING, ["s_get_participants"]
+            ),
+            _make_shape_item(
+                "s_get_participants", TaskName.GET_PARTICIPANTS, ["s_process_tracks"]
+            ),
+            _make_shape_item(
+                "s_process_tracks",
+                TaskName.PROCESS_TRACKS,
+                ["s_mixdown_tracks", "s_detect_topics", "s_finalize"],
+            ),
+            _make_shape_item(
+                "s_mixdown_tracks", TaskName.MIXDOWN_TRACKS, ["s_generate_waveform"]
+            ),
+            _make_shape_item("s_generate_waveform", TaskName.GENERATE_WAVEFORM),
+            _make_shape_item(
+                "s_detect_topics",
+                TaskName.DETECT_TOPICS,
+                ["s_generate_title", "s_extract_subjects"],
+            ),
+            _make_shape_item(
+                "s_generate_title", TaskName.GENERATE_TITLE, ["s_finalize"]
+            ),
+            _make_shape_item(
+                "s_extract_subjects",
+                TaskName.EXTRACT_SUBJECTS,
+                ["s_process_subjects", "s_identify_action_items"],
+            ),
+            _make_shape_item(
+                "s_process_subjects", TaskName.PROCESS_SUBJECTS, ["s_generate_recap"]
+            ),
+            _make_shape_item(
+                "s_generate_recap", TaskName.GENERATE_RECAP, ["s_finalize"]
+            ),
+            _make_shape_item(
+                "s_identify_action_items",
+                TaskName.IDENTIFY_ACTION_ITEMS,
+                ["s_finalize"],
+            ),
+            _make_shape_item("s_finalize", TaskName.FINALIZE, ["s_cleanup_consent"]),
+            _make_shape_item(
+                "s_cleanup_consent",
+                TaskName.CLEANUP_CONSENT,
+                ["s_post_zulip", "s_send_webhook"],
+            ),
+            _make_shape_item("s_post_zulip", TaskName.POST_ZULIP),
+            _make_shape_item("s_send_webhook", TaskName.SEND_WEBHOOK),
+        ]
+
+        # Mid-pipeline: early tasks done, middle running, later queued
+        tasks = [
+            _make_task_summary("s_get_recording", status="COMPLETED"),
+            _make_task_summary("s_get_participants", status="COMPLETED"),
+            _make_task_summary("s_process_tracks", status="COMPLETED"),
+            _make_task_summary("s_mixdown_tracks", status="RUNNING"),
+            _make_task_summary("s_generate_waveform", status="QUEUED"),
+            _make_task_summary("s_detect_topics", status="RUNNING"),
+            _make_task_summary("s_generate_title", status="QUEUED"),
+            _make_task_summary("s_extract_subjects", status="QUEUED"),
+            _make_task_summary("s_process_subjects", status="QUEUED"),
+            _make_task_summary("s_generate_recap", status="QUEUED"),
+            _make_task_summary("s_identify_action_items", status="QUEUED"),
+            _make_task_summary("s_finalize", status="QUEUED"),
+            _make_task_summary("s_cleanup_consent", status="QUEUED"),
+            _make_task_summary("s_post_zulip", status="QUEUED"),
+            _make_task_summary("s_send_webhook", status="QUEUED"),
+        ]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        # All 15 tasks present
+        assert len(result) == 15
+        result_names = [t.name for t in result]
+        assert set(result_names) == {
+            TaskName.GET_RECORDING,
+            TaskName.GET_PARTICIPANTS,
+            TaskName.PROCESS_TRACKS,
+            TaskName.MIXDOWN_TRACKS,
+            TaskName.GENERATE_WAVEFORM,
+            TaskName.DETECT_TOPICS,
+            TaskName.GENERATE_TITLE,
+            TaskName.EXTRACT_SUBJECTS,
+            TaskName.PROCESS_SUBJECTS,
+            TaskName.GENERATE_RECAP,
+            TaskName.IDENTIFY_ACTION_ITEMS,
+            TaskName.FINALIZE,
+            TaskName.CLEANUP_CONSENT,
+            TaskName.POST_ZULIP,
+            TaskName.SEND_WEBHOOK,
+        }
+
+        # Topological order invariant: no task appears before its parents
+        name_to_index = {t.name: i for i, t in enumerate(result)}
+        for task in result:
+            for parent_name in task.parents:
+                assert name_to_index[parent_name] < name_to_index[task.name], (
+                    f"Parent {parent_name} (idx {name_to_index[parent_name]}) "
+                    f"must appear before {task.name} (idx {name_to_index[task.name]})"
+                )
+
+        # finalize has exactly 4 parents
+        finalize = next(t for t in result if t.name == TaskName.FINALIZE)
+        assert set(finalize.parents) == {
+            TaskName.PROCESS_TRACKS,
+            TaskName.GENERATE_TITLE,
+            TaskName.GENERATE_RECAP,
+            TaskName.IDENTIFY_ACTION_ITEMS,
+        }
+
+        # cleanup_consent has 1 parent (finalize)
+        cleanup = next(t for t in result if t.name == TaskName.CLEANUP_CONSENT)
+        assert cleanup.parents == [TaskName.FINALIZE]
+
+        # post_zulip and send_webhook both have cleanup_consent as parent
+        post_zulip = next(t for t in result if t.name == TaskName.POST_ZULIP)
+        send_webhook = next(t for t in result if t.name == TaskName.SEND_WEBHOOK)
+        assert post_zulip.parents == [TaskName.CLEANUP_CONSENT]
+        assert send_webhook.parents == [TaskName.CLEANUP_CONSENT]
+
+        # Verify statuses propagated correctly
+        assert (
+            next(t for t in result if t.name == TaskName.GET_RECORDING).status
+            == DagTaskStatus.COMPLETED
+        )
+        assert (
+            next(t for t in result if t.name == TaskName.MIXDOWN_TRACKS).status
+            == DagTaskStatus.RUNNING
+        )
+        assert (
+            next(t for t in result if t.name == TaskName.FINALIZE).status
+            == DagTaskStatus.QUEUED
+        )
+
+    def test_topological_sort_invariant_complex_dag(self):
+        """For a complex DAG, every task's parents appear earlier in the list.
+
+        Uses a wider branching/merging DAG than diamond to stress the invariant.
+        """
+        # DAG: A -> B, A -> C, A -> D, B -> E, C -> E, C -> F, D -> F, E -> G, F -> G
+        shape = [
+            _make_shape_item("s_a", "task_a", ["s_b", "s_c", "s_d"]),
+            _make_shape_item("s_b", "task_b", ["s_e"]),
+            _make_shape_item("s_c", "task_c", ["s_e", "s_f"]),
+            _make_shape_item("s_d", "task_d", ["s_f"]),
+            _make_shape_item("s_e", "task_e", ["s_g"]),
+            _make_shape_item("s_f", "task_f", ["s_g"]),
+            _make_shape_item("s_g", "task_g"),
+        ]
+        tasks = [
+            _make_task_summary("s_a", status="COMPLETED"),
+            _make_task_summary("s_b", status="COMPLETED"),
+            _make_task_summary("s_c", status="RUNNING"),
+            _make_task_summary("s_d", status="COMPLETED"),
+            _make_task_summary("s_e", status="QUEUED"),
+            _make_task_summary("s_f", status="QUEUED"),
+            _make_task_summary("s_g", status="QUEUED"),
+        ]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        assert len(result) == 7
+        name_to_index = {t.name: i for i, t in enumerate(result)}
+
+        # Verify invariant: every parent appears before its child
+        for task in result:
+            for parent_name in task.parents:
+                assert name_to_index[parent_name] < name_to_index[task.name], (
+                    f"Parent {parent_name} (idx {name_to_index[parent_name]}) "
+                    f"must appear before {task.name} (idx {name_to_index[task.name]})"
+                )
+
+        # task_g has 2 parents
+        task_g = next(t for t in result if t.name == "task_g")
+        assert set(task_g.parents) == {"task_e", "task_f"}
+
+        # task_e has 2 parents
+        task_e = next(t for t in result if t.name == "task_e")
+        assert set(task_e.parents) == {"task_b", "task_c"}
+
+        # task_a is root (first in topological order)
+        assert result[0].name == "task_a"
+        assert result[0].parents == []
+
+
+class TestExtractDagTasksFanOut:
+    """Test fan-out tasks with spawned children."""
+
+    def test_fan_out_children_counts(self):
+        from hatchet_sdk.clients.rest.models import V1TaskStatus
+
+        child_mocks = []
+        for status in ["COMPLETED", "COMPLETED", "RUNNING", "QUEUED"]:
+            child = MagicMock()
+            child.status = V1TaskStatus(status)
+            child_mocks.append(child)
+
+        shape = [_make_shape_item("s1", "process_tracks")]
+        tasks = [
+            _make_task_summary(
+                "s1",
+                status="RUNNING",
+                num_spawned_children=4,
+                children=child_mocks,
+            )
+        ]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        assert result[0].children_total == 4
+        assert result[0].children_completed == 2
+
+    def test_no_children_when_no_spawn(self):
+        shape = [_make_shape_item("s1", "get_recording")]
+        tasks = [
+            _make_task_summary("s1", status="COMPLETED", num_spawned_children=None)
+        ]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        assert result[0].children_total is None
+        assert result[0].children_completed is None
+
+    def test_zero_spawned_children(self):
+        shape = [_make_shape_item("s1", "process_tracks")]
+        tasks = [_make_task_summary("s1", status="COMPLETED", num_spawned_children=0)]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        assert result[0].children_total is None
+        assert result[0].children_completed is None
+
+
+class TestExtractDagTasksErrorExtraction:
+    """Test error message extraction logic."""
+
+    def test_simple_error(self):
+        shape = [_make_shape_item("s1", "mixdown_tracks")]
+        tasks = [
+            _make_task_summary(
+                "s1", status="FAILED", error_message="ValueError: no tracks"
+            )
+        ]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+        assert result[0].error == "ValueError: no tracks"
+
+    def test_traceback_extracts_meaningful_line(self):
+        error = (
+            "Traceback (most recent call last):\n"
+            '  File "/app/something.py", line 42\n'
+            "RuntimeError: out of memory"
+        )
+        shape = [_make_shape_item("s1", "mixdown_tracks")]
+        tasks = [_make_task_summary("s1", status="FAILED", error_message=error)]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+        assert result[0].error == "RuntimeError: out of memory"
+
+    def test_no_error_when_none(self):
+        shape = [_make_shape_item("s1", "get_recording")]
+        tasks = [_make_task_summary("s1", status="COMPLETED", error_message=None)]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+        assert result[0].error is None
+
+    def test_empty_error_message(self):
+        shape = [_make_shape_item("s1", "get_recording")]
+        tasks = [_make_task_summary("s1", status="FAILED", error_message="")]
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+        assert result[0].error is None
+
+
+class TestExtractDagTasksMissingData:
+    """Test edge cases with missing task data."""
+
+    def test_shape_without_matching_task(self):
+        """Shape has a step but tasks list doesn't contain it."""
+        shape = [_make_shape_item("s1", "get_recording")]
+        tasks = []  # No matching task
+        details = _make_details(shape, tasks)
+
+        result = extract_dag_tasks(details)
+
+        assert len(result) == 1
+        assert result[0].name == "get_recording"
+        assert result[0].status == DagTaskStatus.QUEUED  # default when no task data
+        assert result[0].started_at is None
+
+    def test_none_shape_returns_empty(self):
+        details = _make_details(shape=[], tasks=[])
+        details.shape = None
+
+        result = extract_dag_tasks(details)
+        assert result == []
+
+
+class TestDagStatusData:
+    """Test DagStatusData model serialization."""
+
+    def test_serialization(self):
+        task = DagTask(
+            name="get_recording",
+            status=DagTaskStatus.COMPLETED,
+            started_at=datetime(2025, 1, 1, tzinfo=timezone.utc),
+            finished_at=datetime(2025, 1, 1, 0, 0, 1, tzinfo=timezone.utc),
+            duration_seconds=1.0,
+            parents=[],
+            error=None,
+            children_total=None,
+            children_completed=None,
+            progress_pct=None,
+        )
+        data = DagStatusData(workflow_run_id="test-123", tasks=[task])
+        dumped = data.model_dump(mode="json")
+
+        assert dumped["workflow_run_id"] == "test-123"
+        assert len(dumped["tasks"]) == 1
+        assert dumped["tasks"][0]["name"] == "get_recording"
+        assert dumped["tasks"][0]["status"] == "completed"
+        assert dumped["tasks"][0]["duration_seconds"] == 1.0
+
+
+class AsyncContextManager:
+    """No-op async context manager for mocking fresh_db_connection."""
+
+    async def __aenter__(self):
+        return None
+
+    async def __aexit__(self, *args):
+        return None
+
+
+class TestBroadcastDagStatus:
+    """Test broadcast_dag_status function.
+
+    broadcast_dag_status uses deferred imports inside its function body.
+    We mock the source modules/objects before calling the function.
+    Importing daily_multitrack_pipeline triggers a cascade
+    (subject_processing -> HatchetClientManager.get_client at module level),
+    so we set _instance before the import to prevent real SDK init.
+    """
+
+    @pytest.fixture(autouse=True)
+    def _setup_hatchet_mock(self):
+        """Set HatchetClientManager._instance to a mock to prevent real SDK init.
+
+        Module-level code in workflow files calls get_client() during import.
+        Setting _instance before import avoids ClientConfig validation.
+        """
+        from reflector.hatchet.client import HatchetClientManager
+
+        original = HatchetClientManager._instance
+        HatchetClientManager._instance = MagicMock()
+        yield
+        HatchetClientManager._instance = original
+
+    @pytest.mark.asyncio
+    async def test_broadcasts_dag_status(self):
+        """broadcast_dag_status fetches run, transforms, and broadcasts."""
+        mock_transcript = MagicMock()
+        mock_transcript.id = "t-123"
+
+        mock_details = _make_details(
+            shape=[_make_shape_item("s1", "get_recording")],
+            tasks=[_make_task_summary("s1", status="COMPLETED")],
+            run_id="wf-abc",
+        )
+
+        mock_client = MagicMock()
+        mock_client.runs.aio_get = AsyncMock(return_value=mock_details)
+
+        with (
+            patch(
+                "reflector.hatchet.client.HatchetClientManager.get_client",
+                return_value=mock_client,
+            ),
+            patch(
+                "reflector.hatchet.broadcast.append_event_and_broadcast",
+                new_callable=AsyncMock,
+            ) as mock_broadcast,
+            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",
+                return_value=AsyncContextManager(),
+            ),
+        ):
+            from reflector.hatchet.dag_progress import broadcast_dag_status
+
+            await broadcast_dag_status("t-123", "wf-abc")
+
+            mock_client.runs.aio_get.assert_called_once_with("wf-abc")
+            mock_broadcast.assert_called_once()
+            call_args = mock_broadcast.call_args
+            assert call_args[0][0] == "t-123"  # transcript_id
+            assert call_args[0][1] is mock_transcript  # transcript
+            assert call_args[0][2] == "DAG_STATUS"  # event_name
+            data = call_args[0][3]
+            assert isinstance(data, DagStatusData)
+            assert data.workflow_run_id == "wf-abc"
+            assert len(data.tasks) == 1
+
+    @pytest.mark.asyncio
+    async def test_swallows_exceptions(self):
+        """broadcast_dag_status never raises even when internals fail."""
+        from reflector.hatchet.dag_progress import broadcast_dag_status
+
+        with patch(
+            "reflector.hatchet.workflows.daily_multitrack_pipeline.fresh_db_connection",
+            side_effect=RuntimeError("db exploded"),
+        ):
+            # Should not raise
+            await broadcast_dag_status("t-123", "wf-abc")
+
+    @pytest.mark.asyncio
+    async def test_no_broadcast_when_transcript_not_found(self):
+        """broadcast_dag_status does not broadcast if transcript is None."""
+        mock_details = _make_details(
+            shape=[_make_shape_item("s1", "get_recording")],
+            tasks=[_make_task_summary("s1", status="COMPLETED")],
+        )
+
+        mock_client = MagicMock()
+        mock_client.runs.aio_get = AsyncMock(return_value=mock_details)
+
+        with (
+            patch(
+                "reflector.hatchet.client.HatchetClientManager.get_client",
+                return_value=mock_client,
+            ),
+            patch(
+                "reflector.hatchet.workflows.daily_multitrack_pipeline.fresh_db_connection",
+                return_value=AsyncContextManager(),
+            ),
+            patch(
+                "reflector.db.transcripts.transcripts_controller.get_by_id",
+                new_callable=AsyncMock,
+                return_value=None,
+            ),
+            patch(
+                "reflector.hatchet.broadcast.append_event_and_broadcast",
+                new_callable=AsyncMock,
+            ) as mock_broadcast,
+        ):
+            from reflector.hatchet.dag_progress import broadcast_dag_status
+
+            await broadcast_dag_status("t-123", "wf-abc")
+
+            mock_broadcast.assert_not_called()
+
+
+class TestMakeAudioProgressLoggerWithBroadcast:
+    """Test make_audio_progress_logger with transcript_id for transient broadcasts."""
+
+    @pytest.fixture(autouse=True)
+    def _setup_hatchet_mock(self):
+        """Set HatchetClientManager._instance to prevent real SDK init on import."""
+        from reflector.hatchet.client import HatchetClientManager
+
+        original = HatchetClientManager._instance
+        if original is None:
+            HatchetClientManager._instance = MagicMock()
+        yield
+        HatchetClientManager._instance = original
+
+    def test_broadcasts_transient_progress_event(self):
+        """When transcript_id provided and progress_pct not None, broadcasts event."""
+        import asyncio
+
+        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
+            make_audio_progress_logger,
+        )
+
+        ctx = MagicMock()
+        ctx.log = MagicMock()
+
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        mock_broadcast = AsyncMock()
+        tasks_created = []
+
+        original_create_task = loop.create_task
+
+        def capture_create_task(coro):
+            task = original_create_task(coro)
+            tasks_created.append(task)
+            return task
+
+        try:
+            with (
+                patch(
+                    "reflector.hatchet.broadcast.broadcast_event",
+                    mock_broadcast,
+                ),
+                patch.object(loop, "create_task", side_effect=capture_create_task),
+            ):
+                callback = make_audio_progress_logger(
+                    ctx, TaskName.MIXDOWN_TRACKS, interval=0.0, transcript_id="t-123"
+                )
+                callback(50.0, 100.0)
+
+                # Run pending tasks
+                if tasks_created:
+                    loop.run_until_complete(asyncio.gather(*tasks_created))
+
+                mock_broadcast.assert_called_once()
+                event_arg = mock_broadcast.call_args[0][1]
+                assert event_arg.event == "DAG_TASK_PROGRESS"
+                assert event_arg.data["task_name"] == TaskName.MIXDOWN_TRACKS
+                assert event_arg.data["progress_pct"] == 50.0
+        finally:
+            loop.close()
+
+    def test_no_broadcast_without_transcript_id(self):
+        """When transcript_id is None, no broadcast happens."""
+        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
+            make_audio_progress_logger,
+        )
+
+        ctx = MagicMock()
+
+        with patch(
+            "reflector.hatchet.broadcast.broadcast_event",
+            new_callable=AsyncMock,
+        ) as mock_broadcast:
+            callback = make_audio_progress_logger(
+                ctx, TaskName.MIXDOWN_TRACKS, interval=0.0, transcript_id=None
+            )
+            callback(50.0, 100.0)
+            mock_broadcast.assert_not_called()
+
+    def test_no_broadcast_when_progress_pct_is_none(self):
+        """When progress_pct is None, no broadcast happens even with transcript_id."""
+        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
+            make_audio_progress_logger,
+        )
+
+        ctx = MagicMock()
+
+        with patch(
+            "reflector.hatchet.broadcast.broadcast_event",
+            new_callable=AsyncMock,
+        ) as mock_broadcast:
+            callback = make_audio_progress_logger(
+                ctx, TaskName.MIXDOWN_TRACKS, interval=0.0, transcript_id="t-123"
+            )
+            callback(None, 100.0)
+            mock_broadcast.assert_not_called()
+
+    def test_logging_throttled_by_interval(self):
+        """With interval=5.0, rapid calls only log once until interval elapses.
+
+        The throttle applies to ctx.log() calls. Broadcasts (fire-and-forget)
+        are not throttled — they occur every call when transcript_id + progress_pct set.
+        """
+        import asyncio
+        import time as time_mod
+
+        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
+            make_audio_progress_logger,
+        )
+
+        ctx = MagicMock()
+        ctx.log = MagicMock()
+
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        mock_broadcast = AsyncMock()
+        tasks_created = []
+        original_create_task = loop.create_task
+
+        def capture_create_task(coro):
+            task = original_create_task(coro)
+            tasks_created.append(task)
+            return task
+
+        # Controlled monotonic values for the 4 calls from make_audio_progress_logger:
+        # init (start_time, last_log_time), call1 (now), call2 (now), call3 (now)
+        # After those, fall back to real time.monotonic() for asyncio internals.
+        controlled_values = [100.0, 100.0, 101.0, 106.0]
+        call_index = [0]
+        real_monotonic = time_mod.monotonic
+
+        def mock_monotonic():
+            if call_index[0] < len(controlled_values):
+                val = controlled_values[call_index[0]]
+                call_index[0] += 1
+                return val
+            return real_monotonic()
+
+        try:
+            with (
+                patch(
+                    "reflector.hatchet.workflows.daily_multitrack_pipeline.time.monotonic",
+                    side_effect=mock_monotonic,
+                ),
+                patch(
+                    "reflector.hatchet.broadcast.broadcast_event",
+                    mock_broadcast,
+                ),
+                patch.object(loop, "create_task", side_effect=capture_create_task),
+            ):
+                callback = make_audio_progress_logger(
+                    ctx, TaskName.MIXDOWN_TRACKS, interval=5.0, transcript_id="t-123"
+                )
+
+                # Call 1 at t=100.0: 100.0 - 100.0 = 0.0 < 5.0 => no log
+                callback(25.0, 50.0)
+                assert ctx.log.call_count == 0
+
+                # Call 2 at t=101.0: 101.0 - 100.0 = 1.0 < 5.0 => no log
+                callback(50.0, 100.0)
+                assert ctx.log.call_count == 0
+
+                # Call 3 at t=106.0: 106.0 - 100.0 = 6.0 >= 5.0 => logs
+                callback(75.0, 150.0)
+                assert ctx.log.call_count == 1
+
+                # Run pending broadcast tasks
+                if tasks_created:
+                    loop.run_until_complete(asyncio.gather(*tasks_created))
+
+                # Broadcasts happen on every call (not throttled) — 3 calls total
+                assert mock_broadcast.call_count == 3
+        finally:
+            loop.close()
+
+    def test_uses_broadcast_event_not_append_event_and_broadcast(self):
+        """Progress events use broadcast_event (transient), not append_event_and_broadcast (persisted)."""
+        import asyncio
+
+        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
+            make_audio_progress_logger,
+        )
+
+        ctx = MagicMock()
+        ctx.log = MagicMock()
+
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        mock_broadcast_event = AsyncMock()
+        mock_append = AsyncMock()
+        tasks_created = []
+        original_create_task = loop.create_task
+
+        def capture_create_task(coro):
+            task = original_create_task(coro)
+            tasks_created.append(task)
+            return task
+
+        try:
+            with (
+                patch(
+                    "reflector.hatchet.broadcast.broadcast_event",
+                    mock_broadcast_event,
+                ),
+                patch(
+                    "reflector.hatchet.broadcast.append_event_and_broadcast",
+                    mock_append,
+                ),
+                patch.object(loop, "create_task", side_effect=capture_create_task),
+            ):
+                callback = make_audio_progress_logger(
+                    ctx, TaskName.MIXDOWN_TRACKS, interval=0.0, transcript_id="t-123"
+                )
+                callback(50.0, 100.0)
+
+                if tasks_created:
+                    loop.run_until_complete(asyncio.gather(*tasks_created))
+
+                # broadcast_event (transient) IS called
+                mock_broadcast_event.assert_called_once()
+                # append_event_and_broadcast (persisted) is NOT called
+                mock_append.assert_not_called()
+        finally:
+            loop.close()

server/tests/test_dag_progress_decorator.py

@@ -0,0 +1,181 @@
+"""Tests for with_error_handling decorator integration with broadcast_dag_status.
+
+The decorator wraps each pipeline task and calls broadcast_dag_status on both
+success and failure paths. These tests verify that integration rather than
+testing broadcast_dag_status in isolation (which test_dag_progress.py covers).
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from reflector.hatchet.constants import TaskName
+
+
+class TestWithErrorHandlingBroadcast:
+    """Test with_error_handling decorator's integration with broadcast_dag_status."""
+
+    @pytest.fixture(autouse=True)
+    def _setup_hatchet_mock(self):
+        """Set HatchetClientManager._instance to a mock to prevent real SDK init.
+
+        Module-level code in workflow files calls get_client() during import.
+        Setting _instance before import avoids ClientConfig validation.
+        """
+        from reflector.hatchet.client import HatchetClientManager
+
+        original = HatchetClientManager._instance
+        HatchetClientManager._instance = MagicMock()
+        yield
+        HatchetClientManager._instance = original
+
+    def _make_input(self, transcript_id: str = "t-123") -> MagicMock:
+        """Create a mock PipelineInput with transcript_id."""
+        inp = MagicMock()
+        inp.transcript_id = transcript_id
+        return inp
+
+    def _make_ctx(self, workflow_run_id: str = "wf-abc") -> MagicMock:
+        """Create a mock Context with workflow_run_id."""
+        ctx = MagicMock()
+        ctx.workflow_run_id = workflow_run_id
+        return ctx
+
+    @pytest.mark.asyncio
+    async def test_calls_broadcast_on_success(self):
+        """Decorator calls broadcast_dag_status once when task succeeds."""
+        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
+            with_error_handling,
+        )
+
+        inner = AsyncMock(return_value="ok")
+        wrapped = with_error_handling(TaskName.GET_RECORDING)(inner)
+
+        with patch(
+            "reflector.hatchet.dag_progress.broadcast_dag_status",
+            new_callable=AsyncMock,
+        ) as mock_broadcast:
+            result = await wrapped(self._make_input(), self._make_ctx())
+
+        assert result == "ok"
+        mock_broadcast.assert_called_once_with("t-123", "wf-abc")
+
+    @pytest.mark.asyncio
+    async def test_calls_broadcast_on_failure(self):
+        """Decorator calls broadcast_dag_status once when task raises."""
+        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
+            with_error_handling,
+        )
+
+        inner = AsyncMock(side_effect=RuntimeError("boom"))
+        wrapped = with_error_handling(TaskName.GET_RECORDING)(inner)
+
+        with (
+            patch(
+                "reflector.hatchet.dag_progress.broadcast_dag_status",
+                new_callable=AsyncMock,
+            ) as mock_broadcast,
+            patch(
+                "reflector.hatchet.workflows.daily_multitrack_pipeline.set_workflow_error_status",
+                new_callable=AsyncMock,
+            ),
+        ):
+            with pytest.raises(RuntimeError, match="boom"):
+                await wrapped(self._make_input(), self._make_ctx())
+
+        mock_broadcast.assert_called_once_with("t-123", "wf-abc")
+
+    @pytest.mark.asyncio
+    async def test_swallows_broadcast_exception_on_success(self):
+        """Broadcast failure does not crash the task on the success path."""
+        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
+            with_error_handling,
+        )
+
+        inner = AsyncMock(return_value="ok")
+        wrapped = with_error_handling(TaskName.GET_RECORDING)(inner)
+
+        with patch(
+            "reflector.hatchet.dag_progress.broadcast_dag_status",
+            new_callable=AsyncMock,
+            side_effect=RuntimeError("broadcast exploded"),
+        ):
+            result = await wrapped(self._make_input(), self._make_ctx())
+
+        assert result == "ok"
+
+    @pytest.mark.asyncio
+    async def test_swallows_broadcast_exception_on_failure(self):
+        """Original task exception propagates even when broadcast also fails."""
+        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
+            with_error_handling,
+        )
+
+        inner = AsyncMock(side_effect=ValueError("original error"))
+        wrapped = with_error_handling(TaskName.GET_RECORDING)(inner)
+
+        with (
+            patch(
+                "reflector.hatchet.dag_progress.broadcast_dag_status",
+                new_callable=AsyncMock,
+                side_effect=RuntimeError("broadcast exploded"),
+            ),
+            patch(
+                "reflector.hatchet.workflows.daily_multitrack_pipeline.set_workflow_error_status",
+                new_callable=AsyncMock,
+            ),
+        ):
+            with pytest.raises(ValueError, match="original error"):
+                await wrapped(self._make_input(), self._make_ctx())
+
+    @pytest.mark.asyncio
+    async def test_calls_set_workflow_error_status_on_failure(self):
+        """On task failure with set_error_status=True (default), calls set_workflow_error_status."""
+        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
+            with_error_handling,
+        )
+
+        inner = AsyncMock(side_effect=RuntimeError("boom"))
+        wrapped = with_error_handling(TaskName.GET_RECORDING)(inner)
+
+        with (
+            patch(
+                "reflector.hatchet.dag_progress.broadcast_dag_status",
+                new_callable=AsyncMock,
+            ),
+            patch(
+                "reflector.hatchet.workflows.daily_multitrack_pipeline.set_workflow_error_status",
+                new_callable=AsyncMock,
+            ) as mock_set_error,
+        ):
+            with pytest.raises(RuntimeError, match="boom"):
+                await wrapped(self._make_input(), self._make_ctx())
+
+        mock_set_error.assert_called_once_with("t-123")
+
+    @pytest.mark.asyncio
+    async def test_no_set_workflow_error_status_when_disabled(self):
+        """With set_error_status=False, set_workflow_error_status is NOT called on failure."""
+        from reflector.hatchet.workflows.daily_multitrack_pipeline import (
+            with_error_handling,
+        )
+
+        inner = AsyncMock(side_effect=RuntimeError("boom"))
+        wrapped = with_error_handling(TaskName.GET_RECORDING, set_error_status=False)(
+            inner
+        )
+
+        with (
+            patch(
+                "reflector.hatchet.dag_progress.broadcast_dag_status",
+                new_callable=AsyncMock,
+            ),
+            patch(
+                "reflector.hatchet.workflows.daily_multitrack_pipeline.set_workflow_error_status",
+                new_callable=AsyncMock,
+            ) as mock_set_error,
+        ):
+            with pytest.raises(RuntimeError, match="boom"):
+                await wrapped(self._make_input(), self._make_ctx())
+
+        mock_set_error.assert_not_called()

server/tests/test_dag_progress_rest.py

@@ -0,0 +1,421 @@
+"""Tests for DAG status REST enrichment on search and transcript GET endpoints."""
+
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+import reflector.db.search as search_module
+from reflector.db.search import SearchResult, _fetch_dag_statuses
+from reflector.db.transcripts import TranscriptEvent
+
+
+class TestFetchDagStatuses:
+    """Test the _fetch_dag_statuses helper."""
+
+    @pytest.mark.asyncio
+    async def test_returns_empty_for_empty_ids(self):
+        result = await _fetch_dag_statuses([])
+        assert result == {}
+
+    @pytest.mark.asyncio
+    async def test_extracts_last_dag_status(self):
+        events = [
+            {"event": "STATUS", "data": {"value": "processing"}},
+            {
+                "event": "DAG_STATUS",
+                "data": {
+                    "workflow_run_id": "r1",
+                    "tasks": [{"name": "get_recording", "status": "completed"}],
+                },
+            },
+            {
+                "event": "DAG_STATUS",
+                "data": {
+                    "workflow_run_id": "r1",
+                    "tasks": [
+                        {"name": "get_recording", "status": "completed"},
+                        {"name": "process_tracks", "status": "running"},
+                    ],
+                },
+            },
+        ]
+        mock_row = {"id": "t1", "events": events}
+
+        with patch("reflector.db.search.get_database") as mock_db:
+            mock_db.return_value.fetch_all = AsyncMock(return_value=[mock_row])
+            result = await _fetch_dag_statuses(["t1"])
+
+        assert "t1" in result
+        assert len(result["t1"]) == 2  # Last DAG_STATUS had 2 tasks
+
+    @pytest.mark.asyncio
+    async def test_skips_transcripts_without_events(self):
+        mock_row = {"id": "t1", "events": None}
+
+        with patch("reflector.db.search.get_database") as mock_db:
+            mock_db.return_value.fetch_all = AsyncMock(return_value=[mock_row])
+            result = await _fetch_dag_statuses(["t1"])
+
+        assert result == {}
+
+    @pytest.mark.asyncio
+    async def test_skips_transcripts_without_dag_status(self):
+        events = [
+            {"event": "STATUS", "data": {"value": "processing"}},
+            {"event": "DURATION", "data": {"duration": 1000}},
+        ]
+        mock_row = {"id": "t1", "events": events}
+
+        with patch("reflector.db.search.get_database") as mock_db:
+            mock_db.return_value.fetch_all = AsyncMock(return_value=[mock_row])
+            result = await _fetch_dag_statuses(["t1"])
+
+        assert result == {}
+
+    @pytest.mark.asyncio
+    async def test_handles_json_string_events(self):
+        """Events stored as JSON string rather than already-parsed list."""
+        import json
+
+        events = [
+            {
+                "event": "DAG_STATUS",
+                "data": {
+                    "workflow_run_id": "r1",
+                    "tasks": [{"name": "transcribe", "status": "running"}],
+                },
+            },
+        ]
+        mock_row = {"id": "t1", "events": json.dumps(events)}
+
+        with patch("reflector.db.search.get_database") as mock_db:
+            mock_db.return_value.fetch_all = AsyncMock(return_value=[mock_row])
+            result = await _fetch_dag_statuses(["t1"])
+
+        assert "t1" in result
+        assert len(result["t1"]) == 1
+        assert result["t1"][0]["name"] == "transcribe"
+
+    @pytest.mark.asyncio
+    async def test_multiple_transcripts(self):
+        """Handles multiple transcripts in one call."""
+        events_t1 = [
+            {
+                "event": "DAG_STATUS",
+                "data": {
+                    "workflow_run_id": "r1",
+                    "tasks": [{"name": "a", "status": "completed"}],
+                },
+            },
+        ]
+        events_t2 = [
+            {
+                "event": "DAG_STATUS",
+                "data": {
+                    "workflow_run_id": "r2",
+                    "tasks": [{"name": "b", "status": "running"}],
+                },
+            },
+        ]
+        mock_rows = [
+            {"id": "t1", "events": events_t1},
+            {"id": "t2", "events": events_t2},
+        ]
+
+        with patch("reflector.db.search.get_database") as mock_db:
+            mock_db.return_value.fetch_all = AsyncMock(return_value=mock_rows)
+            result = await _fetch_dag_statuses(["t1", "t2"])
+
+        assert "t1" in result
+        assert "t2" in result
+        assert result["t1"][0]["name"] == "a"
+        assert result["t2"][0]["name"] == "b"
+
+    @pytest.mark.asyncio
+    async def test_dag_status_without_tasks_key_skipped(self):
+        """DAG_STATUS event with no tasks key in data should be skipped."""
+        events = [
+            {"event": "DAG_STATUS", "data": {"workflow_run_id": "r1"}},
+        ]
+        mock_row = {"id": "t1", "events": events}
+
+        with patch("reflector.db.search.get_database") as mock_db:
+            mock_db.return_value.fetch_all = AsyncMock(return_value=[mock_row])
+            result = await _fetch_dag_statuses(["t1"])
+
+        assert result == {}
+
+
+def _extract_dag_status_from_transcript(transcript):
+    """Replicate the dag_status extraction logic from transcript_get view.
+
+    This mirrors the code in reflector/views/transcripts.py lines 495-500:
+        dag_status = None
+        if transcript.status == "processing" and transcript.events:
+            for ev in reversed(transcript.events):
+                if ev.event == "DAG_STATUS":
+                    dag_status = ev.data.get("tasks") if isinstance(ev.data, dict) else None
+                    break
+    """
+    dag_status = None
+    if transcript.status == "processing" and transcript.events:
+        for ev in reversed(transcript.events):
+            if ev.event == "DAG_STATUS":
+                dag_status = ev.data.get("tasks") if isinstance(ev.data, dict) else None
+                break
+    return dag_status
+
+
+class TestTranscriptGetDagStatusExtraction:
+    """Test dag_status extraction logic from transcript_get endpoint.
+
+    The actual endpoint is complex to set up, so we test the extraction
+    logic directly using the same code pattern from the view.
+    """
+
+    def test_processing_transcript_with_dag_status_events(self):
+        """Processing transcript with DAG_STATUS events returns tasks from last event."""
+        transcript = SimpleNamespace(
+            status="processing",
+            events=[
+                TranscriptEvent(event="STATUS", data={"value": "processing"}),
+                TranscriptEvent(
+                    event="DAG_STATUS",
+                    data={
+                        "workflow_run_id": "r1",
+                        "tasks": [{"name": "get_recording", "status": "completed"}],
+                    },
+                ),
+                TranscriptEvent(
+                    event="DAG_STATUS",
+                    data={
+                        "workflow_run_id": "r1",
+                        "tasks": [
+                            {"name": "get_recording", "status": "completed"},
+                            {"name": "transcribe", "status": "running"},
+                        ],
+                    },
+                ),
+            ],
+        )
+
+        result = _extract_dag_status_from_transcript(transcript)
+
+        assert result is not None
+        assert len(result) == 2
+        assert result[0]["name"] == "get_recording"
+        assert result[1]["name"] == "transcribe"
+        assert result[1]["status"] == "running"
+
+    def test_processing_transcript_without_dag_status_events(self):
+        """Processing transcript with only non-DAG_STATUS events returns None."""
+        transcript = SimpleNamespace(
+            status="processing",
+            events=[
+                TranscriptEvent(event="STATUS", data={"value": "processing"}),
+                TranscriptEvent(event="DURATION", data={"duration": 1000}),
+            ],
+        )
+
+        result = _extract_dag_status_from_transcript(transcript)
+        assert result is None
+
+    def test_ended_transcript_with_dag_status_events(self):
+        """Ended transcript with DAG_STATUS events returns None (status check)."""
+        transcript = SimpleNamespace(
+            status="ended",
+            events=[
+                TranscriptEvent(
+                    event="DAG_STATUS",
+                    data={
+                        "workflow_run_id": "r1",
+                        "tasks": [{"name": "transcribe", "status": "completed"}],
+                    },
+                ),
+            ],
+        )
+
+        result = _extract_dag_status_from_transcript(transcript)
+        assert result is None
+
+    def test_processing_transcript_with_empty_events(self):
+        """Processing transcript with empty events list returns None."""
+        transcript = SimpleNamespace(
+            status="processing",
+            events=[],
+        )
+
+        result = _extract_dag_status_from_transcript(transcript)
+        assert result is None
+
+    def test_processing_transcript_with_none_events(self):
+        """Processing transcript with None events returns None."""
+        transcript = SimpleNamespace(
+            status="processing",
+            events=None,
+        )
+
+        result = _extract_dag_status_from_transcript(transcript)
+        assert result is None
+
+    def test_extracts_last_dag_status_not_first(self):
+        """Should pick the last DAG_STATUS event (most recent), not the first."""
+        transcript = SimpleNamespace(
+            status="processing",
+            events=[
+                TranscriptEvent(
+                    event="DAG_STATUS",
+                    data={
+                        "workflow_run_id": "r1",
+                        "tasks": [{"name": "a", "status": "running"}],
+                    },
+                ),
+                TranscriptEvent(event="STATUS", data={"value": "processing"}),
+                TranscriptEvent(
+                    event="DAG_STATUS",
+                    data={
+                        "workflow_run_id": "r1",
+                        "tasks": [
+                            {"name": "a", "status": "completed"},
+                            {"name": "b", "status": "running"},
+                        ],
+                    },
+                ),
+            ],
+        )
+
+        result = _extract_dag_status_from_transcript(transcript)
+        assert len(result) == 2
+        assert result[0]["status"] == "completed"
+        assert result[1]["name"] == "b"
+
+
+class TestSearchEnrichmentIntegration:
+    """Test DAG status enrichment in search results.
+
+    The search function enriches processing transcripts with dag_status
+    by calling _fetch_dag_statuses for processing IDs and assigning results.
+    We test this enrichment logic by mocking _fetch_dag_statuses.
+    """
+
+    def _make_search_result(self, id: str, status: str) -> SearchResult:
+        """Create a minimal SearchResult for testing."""
+        return SearchResult(
+            id=id,
+            title=f"Transcript {id}",
+            user_id="u1",
+            room_id=None,
+            room_name=None,
+            source_kind="live",
+            created_at=datetime(2024, 1, 1, tzinfo=timezone.utc),
+            status=status,
+            rank=1.0,
+            duration=60.0,
+            search_snippets=[],
+            total_match_count=0,
+            dag_status=None,
+        )
+
+    @pytest.mark.asyncio
+    async def test_processing_result_gets_dag_status(self):
+        """SearchResult with status='processing' and matching DAG_STATUS events
+        gets dag_status populated."""
+        results = [self._make_search_result("t1", "processing")]
+        dag_tasks = [
+            {"name": "get_recording", "status": "completed"},
+            {"name": "transcribe", "status": "running"},
+        ]
+
+        with patch.object(
+            search_module,
+            "_fetch_dag_statuses",
+            new_callable=AsyncMock,
+            return_value={"t1": dag_tasks},
+        ) as mock_fetch:
+            # Replicate the enrichment logic from SearchController.search_transcripts
+            processing_ids = [r.id for r in results if r.status == "processing"]
+            if processing_ids:
+                dag_statuses = await search_module._fetch_dag_statuses(processing_ids)
+                for r in results:
+                    if r.id in dag_statuses:
+                        r.dag_status = dag_statuses[r.id]
+
+            mock_fetch.assert_called_once_with(["t1"])
+
+        assert results[0].dag_status == dag_tasks
+
+    @pytest.mark.asyncio
+    async def test_ended_result_does_not_trigger_fetch(self):
+        """SearchResult with status='ended' does NOT trigger _fetch_dag_statuses."""
+        results = [self._make_search_result("t1", "ended")]
+
+        with patch.object(
+            search_module,
+            "_fetch_dag_statuses",
+            new_callable=AsyncMock,
+            return_value={},
+        ) as mock_fetch:
+            processing_ids = [r.id for r in results if r.status == "processing"]
+            if processing_ids:
+                dag_statuses = await search_module._fetch_dag_statuses(processing_ids)
+                for r in results:
+                    if r.id in dag_statuses:
+                        r.dag_status = dag_statuses[r.id]
+
+            mock_fetch.assert_not_called()
+
+        assert results[0].dag_status is None
+
+    @pytest.mark.asyncio
+    async def test_mixed_processing_and_ended_results(self):
+        """Only processing results get enriched; ended results stay None."""
+        results = [
+            self._make_search_result("t1", "processing"),
+            self._make_search_result("t2", "ended"),
+            self._make_search_result("t3", "processing"),
+        ]
+        dag_tasks_t1 = [{"name": "transcribe", "status": "running"}]
+        dag_tasks_t3 = [{"name": "diarize", "status": "completed"}]
+
+        with patch.object(
+            search_module,
+            "_fetch_dag_statuses",
+            new_callable=AsyncMock,
+            return_value={"t1": dag_tasks_t1, "t3": dag_tasks_t3},
+        ) as mock_fetch:
+            processing_ids = [r.id for r in results if r.status == "processing"]
+            if processing_ids:
+                dag_statuses = await search_module._fetch_dag_statuses(processing_ids)
+                for r in results:
+                    if r.id in dag_statuses:
+                        r.dag_status = dag_statuses[r.id]
+
+            mock_fetch.assert_called_once_with(["t1", "t3"])
+
+        assert results[0].dag_status == dag_tasks_t1
+        assert results[1].dag_status is None
+        assert results[2].dag_status == dag_tasks_t3
+
+    @pytest.mark.asyncio
+    async def test_processing_result_without_dag_events_stays_none(self):
+        """Processing result with no DAG_STATUS events in DB stays dag_status=None."""
+        results = [self._make_search_result("t1", "processing")]
+
+        with patch.object(
+            search_module,
+            "_fetch_dag_statuses",
+            new_callable=AsyncMock,
+            return_value={},
+        ) as mock_fetch:
+            processing_ids = [r.id for r in results if r.status == "processing"]
+            if processing_ids:
+                dag_statuses = await search_module._fetch_dag_statuses(processing_ids)
+                for r in results:
+                    if r.id in dag_statuses:
+                        r.dag_status = dag_statuses[r.id]
+
+            mock_fetch.assert_called_once_with(["t1"])
+
+        assert results[0].dag_status is None

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_ws_dag_broadcast.py

@@ -0,0 +1,331 @@
+"""WebSocket broadcast delivery tests for STATUS and DAG_STATUS events.
+
+Tests the full chain identified in DEBUG.md:
+  broadcast_event() → ws_manager.send_json() → Redis/in-memory pub/sub
+    → _pubsub_data_reader() → socket.send_json() → WebSocket client
+
+Covers:
+1. STATUS event delivery to transcript room WS
+2. DAG_STATUS event delivery to transcript room WS
+3. Full broadcast_event() chain (requires broadcast.py patching)
+4. _pubsub_data_reader resilience when a client disconnects
+"""
+
+import asyncio
+import threading
+import time
+
+import pytest
+from httpx import AsyncClient
+from httpx_ws import aconnect_ws
+from uvicorn import Config, Server
+
+
+@pytest.fixture
+def appserver_ws_broadcast(setup_database, monkeypatch):
+    """Start real uvicorn server for WebSocket broadcast tests.
+
+    Also patches broadcast.py's get_ws_manager (missing from conftest autouse fixture).
+    """
+    # Patch broadcast.py's get_ws_manager — conftest.py misses this module.
+    # Without this, broadcast_event() creates a real Redis ws_manager.
+    import reflector.ws_manager as ws_mod
+    from reflector.app import app
+    from reflector.db import get_database
+
+    monkeypatch.setattr(
+        "reflector.hatchet.broadcast.get_ws_manager", ws_mod.get_ws_manager
+    )
+
+    host = "127.0.0.1"
+    port = 1259
+    server_started = threading.Event()
+    server_exception = None
+    server_instance = None
+
+    def run_server():
+        nonlocal server_exception, server_instance
+        try:
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            config = Config(app=app, host=host, port=port, loop=loop)
+            server_instance = Server(config)
+
+            async def start_server():
+                database = get_database()
+                await database.connect()
+                try:
+                    await server_instance.serve()
+                finally:
+                    await database.disconnect()
+
+            server_started.set()
+            loop.run_until_complete(start_server())
+        except Exception as e:
+            server_exception = e
+            server_started.set()
+        finally:
+            loop.close()
+
+    server_thread = threading.Thread(target=run_server, daemon=True)
+    server_thread.start()
+
+    server_started.wait(timeout=30)
+    if server_exception:
+        raise server_exception
+
+    time.sleep(0.5)
+
+    yield host, port
+
+    if server_instance:
+        server_instance.should_exit = True
+        server_thread.join(timeout=2.0)
+
+    from reflector.ws_manager import reset_ws_manager
+
+    reset_ws_manager()
+
+
+async def _create_transcript(host: str, port: int, name: str) -> str:
+    """Create a transcript via ASGI transport and return its ID."""
+    from reflector.app import app
+
+    async with AsyncClient(app=app, base_url=f"http://{host}:{port}/v1") as ac:
+        resp = await ac.post("/transcripts", json={"name": name})
+        assert resp.status_code == 200, f"Failed to create transcript: {resp.text}"
+        return resp.json()["id"]
+
+
+async def _drain_historical_events(ws, timeout: float = 0.5) -> list[dict]:
+    """Read all historical events sent on WS connect (non-blocking drain)."""
+    events = []
+    deadline = asyncio.get_event_loop().time() + timeout
+    while asyncio.get_event_loop().time() < deadline:
+        try:
+            msg = await asyncio.wait_for(ws.receive_json(), timeout=0.1)
+            events.append(msg)
+        except (asyncio.TimeoutError, Exception):
+            break
+    return events
+
+
+# ---------------------------------------------------------------------------
+# Test 1: STATUS event delivery via ws_manager.send_json
+# ---------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_transcript_ws_receives_status_via_send_json(appserver_ws_broadcast):
+    """STATUS event published via ws_manager.send_json() arrives at transcript room WS."""
+    host, port = appserver_ws_broadcast
+    transcript_id = await _create_transcript(host, port, "Status send_json test")
+
+    ws_url = f"http://{host}:{port}/v1/transcripts/{transcript_id}/events"
+    async with aconnect_ws(ws_url) as ws:
+        await _drain_historical_events(ws)
+
+        import reflector.ws_manager as ws_mod
+
+        ws_manager = ws_mod.get_ws_manager()
+        await ws_manager.send_json(
+            room_id=f"ts:{transcript_id}",
+            message={"event": "STATUS", "data": {"value": "processing"}},
+        )
+
+        msg = await asyncio.wait_for(ws.receive_json(), timeout=5.0)
+        assert msg["event"] == "STATUS"
+        assert msg["data"]["value"] == "processing"
+
+
+# ---------------------------------------------------------------------------
+# Test 2: DAG_STATUS event delivery via ws_manager.send_json
+# ---------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_transcript_ws_receives_dag_status_via_send_json(appserver_ws_broadcast):
+    """DAG_STATUS event published via ws_manager.send_json() arrives at transcript room WS."""
+    host, port = appserver_ws_broadcast
+    transcript_id = await _create_transcript(host, port, "DAG_STATUS send_json test")
+
+    dag_payload = {
+        "event": "DAG_STATUS",
+        "data": {
+            "workflow_run_id": "test-run-123",
+            "tasks": [
+                {
+                    "name": "get_recording",
+                    "status": "completed",
+                    "started_at": "2025-01-01T00:00:00Z",
+                    "finished_at": "2025-01-01T00:00:05Z",
+                    "duration_seconds": 5.0,
+                    "parents": [],
+                    "error": None,
+                    "children_total": None,
+                    "children_completed": None,
+                    "progress_pct": None,
+                },
+                {
+                    "name": "process_tracks",
+                    "status": "running",
+                    "started_at": "2025-01-01T00:00:05Z",
+                    "finished_at": None,
+                    "duration_seconds": None,
+                    "parents": ["get_recording"],
+                    "error": None,
+                    "children_total": 3,
+                    "children_completed": 1,
+                    "progress_pct": 33.3,
+                },
+            ],
+        },
+    }
+
+    ws_url = f"http://{host}:{port}/v1/transcripts/{transcript_id}/events"
+    async with aconnect_ws(ws_url) as ws:
+        await _drain_historical_events(ws)
+
+        import reflector.ws_manager as ws_mod
+
+        ws_manager = ws_mod.get_ws_manager()
+        await ws_manager.send_json(
+            room_id=f"ts:{transcript_id}",
+            message=dag_payload,
+        )
+
+        msg = await asyncio.wait_for(ws.receive_json(), timeout=5.0)
+        assert msg["event"] == "DAG_STATUS"
+        assert msg["data"]["workflow_run_id"] == "test-run-123"
+        assert len(msg["data"]["tasks"]) == 2
+        assert msg["data"]["tasks"][0]["name"] == "get_recording"
+        assert msg["data"]["tasks"][0]["status"] == "completed"
+        assert msg["data"]["tasks"][1]["name"] == "process_tracks"
+        assert msg["data"]["tasks"][1]["children_completed"] == 1
+
+
+# ---------------------------------------------------------------------------
+# Test 3: Full broadcast_event() chain for STATUS
+# ---------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_broadcast_event_delivers_status_to_transcript_ws(appserver_ws_broadcast):
+    """broadcast_event() end-to-end: STATUS event reaches transcript room WS."""
+    host, port = appserver_ws_broadcast
+    transcript_id = await _create_transcript(host, port, "broadcast_event STATUS test")
+
+    ws_url = f"http://{host}:{port}/v1/transcripts/{transcript_id}/events"
+    async with aconnect_ws(ws_url) as ws:
+        await _drain_historical_events(ws)
+
+        from reflector.db.transcripts import TranscriptEvent
+        from reflector.hatchet.broadcast import broadcast_event
+        from reflector.logger import logger
+
+        log = logger.bind(transcript_id=transcript_id)
+        event = TranscriptEvent(event="STATUS", data={"value": "processing"})
+        await broadcast_event(transcript_id, event, logger=log)
+
+        msg = await asyncio.wait_for(ws.receive_json(), timeout=5.0)
+        assert msg["event"] == "STATUS"
+        assert msg["data"]["value"] == "processing"
+
+
+# ---------------------------------------------------------------------------
+# Test 4: Full broadcast_event() chain for DAG_STATUS
+# ---------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_broadcast_event_delivers_dag_status_to_transcript_ws(
+    appserver_ws_broadcast,
+):
+    """broadcast_event() end-to-end: DAG_STATUS event reaches transcript room WS."""
+    host, port = appserver_ws_broadcast
+    transcript_id = await _create_transcript(host, port, "broadcast_event DAG test")
+
+    ws_url = f"http://{host}:{port}/v1/transcripts/{transcript_id}/events"
+    async with aconnect_ws(ws_url) as ws:
+        await _drain_historical_events(ws)
+
+        from reflector.db.transcripts import TranscriptEvent
+        from reflector.hatchet.broadcast import broadcast_event
+        from reflector.logger import logger
+
+        log = logger.bind(transcript_id=transcript_id)
+        event = TranscriptEvent(
+            event="DAG_STATUS",
+            data={
+                "workflow_run_id": "test-run-456",
+                "tasks": [
+                    {
+                        "name": "get_recording",
+                        "status": "running",
+                        "started_at": None,
+                        "finished_at": None,
+                        "duration_seconds": None,
+                        "parents": [],
+                        "error": None,
+                        "children_total": None,
+                        "children_completed": None,
+                        "progress_pct": None,
+                    }
+                ],
+            },
+        )
+        await broadcast_event(transcript_id, event, logger=log)
+
+        msg = await asyncio.wait_for(ws.receive_json(), timeout=5.0)
+        assert msg["event"] == "DAG_STATUS"
+        assert msg["data"]["tasks"][0]["name"] == "get_recording"
+
+
+# ---------------------------------------------------------------------------
+# Test 5: Multiple rapid events arrive in order
+# ---------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_multiple_events_arrive_in_order(appserver_ws_broadcast):
+    """Multiple STATUS then DAG_STATUS events arrive in correct order."""
+    host, port = appserver_ws_broadcast
+    transcript_id = await _create_transcript(host, port, "ordering test")
+
+    ws_url = f"http://{host}:{port}/v1/transcripts/{transcript_id}/events"
+    async with aconnect_ws(ws_url) as ws:
+        await _drain_historical_events(ws)
+
+        import reflector.ws_manager as ws_mod
+
+        ws_manager = ws_mod.get_ws_manager()
+
+        await ws_manager.send_json(
+            room_id=f"ts:{transcript_id}",
+            message={"event": "STATUS", "data": {"value": "processing"}},
+        )
+        await ws_manager.send_json(
+            room_id=f"ts:{transcript_id}",
+            message={
+                "event": "DAG_STATUS",
+                "data": {"workflow_run_id": "r1", "tasks": []},
+            },
+        )
+        await ws_manager.send_json(
+            room_id=f"ts:{transcript_id}",
+            message={
+                "event": "DAG_STATUS",
+                "data": {
+                    "workflow_run_id": "r1",
+                    "tasks": [{"name": "a", "status": "running"}],
+                },
+            },
+        )
+        await ws_manager.send_json(
+            room_id=f"ts:{transcript_id}",
+            message={"event": "STATUS", "data": {"value": "ended"}},
+        )
+
+        msgs = []
+        for _ in range(4):
+            msg = await asyncio.wait_for(ws.receive_json(), timeout=5.0)
+            msgs.append(msg)
+
+        assert msgs[0]["event"] == "STATUS"
+        assert msgs[0]["data"]["value"] == "processing"
+        assert msgs[1]["event"] == "DAG_STATUS"
+        assert msgs[1]["data"]["tasks"] == []
+        assert msgs[2]["event"] == "DAG_STATUS"
+        assert len(msgs[2]["data"]["tasks"]) == 1
+        assert msgs[3]["event"] == "STATUS"
+        assert msgs[3]["data"]["value"] == "ended"

www/app/(app)/browse/_components/DagProgressDots.tsx

@@ -0,0 +1,61 @@
+import React from "react";
+import { Box, Flex } from "@chakra-ui/react";
+import type { DagTask } from "../../../lib/UserEventsProvider";
+
+const pulseKeyframes = `
+  @keyframes dagDotPulse {
+    0%, 100% { opacity: 1; }
+    50% { opacity: 0.3; }
+  }
+`;
+
+function humanizeTaskName(name: string): string {
+  return name
+    .split("_")
+    .map((word) => word.charAt(0).toUpperCase() + word.slice(1))
+    .join(" ");
+}
+
+function dotProps(status: DagTask["status"]): Record<string, unknown> {
+  switch (status) {
+    case "completed":
+      return { bg: "green.500" };
+    case "running":
+      return {
+        bg: "blue.500",
+        style: { animation: "dagDotPulse 1.5s ease-in-out infinite" },
+      };
+    case "failed":
+      return { bg: "red.500" };
+    case "cancelled":
+      return { bg: "gray.400" };
+    case "queued":
+    default:
+      return {
+        bg: "transparent",
+        border: "1px solid",
+        borderColor: "gray.400",
+      };
+  }
+}
+
+export default function DagProgressDots({ tasks }: { tasks: DagTask[] }) {
+  return (
+    <>
+      <style>{pulseKeyframes}</style>
+      <Flex gap="2px" alignItems="center" flexWrap="wrap">
+        {tasks.map((task) => (
+          <Box
+            key={task.name}
+            w="4px"
+            h="4px"
+            borderRadius="full"
+            flexShrink={0}
+            title={humanizeTaskName(task.name)}
+            {...dotProps(task.status)}
+          />
+        ))}
+      </Flex>
+    </>
+  );
+}

www/app/(app)/browse/_components/TranscriptCards.tsx

@@ -19,6 +19,7 @@ import {
   generateTextFragment,
 } from "../../../lib/textHighlight";
 import type { components } from "../../../reflector-api";
+import type { DagTask } from "../../../lib/UserEventsProvider";
 
 type SearchResult = components["schemas"]["SearchResult"];
 type SourceKind = components["schemas"]["SourceKind"];
@@ -29,6 +30,7 @@ interface TranscriptCardsProps {
   isLoading?: boolean;
   onDelete: (transcriptId: string) => void;
   onReprocess: (transcriptId: string) => void;
+  dagStatusMap?: Map<string, DagTask[]>;
 }
 
 function highlightText(text: string, query: string): React.ReactNode {
@@ -102,11 +104,13 @@ function TranscriptCard({
   query,
   onDelete,
   onReprocess,
+  dagStatusMap,
 }: {
   result: SearchResult;
   query: string;
   onDelete: (transcriptId: string) => void;
   onReprocess: (transcriptId: string) => void;
+  dagStatusMap?: Map<string, DagTask[]>;
 }) {
   const [isExpanded, setIsExpanded] = useState(false);
 
@@ -137,7 +141,16 @@ function TranscriptCard({
     <Box borderWidth={1} p={4} borderRadius="md" fontSize="sm">
       <Flex justify="space-between" alignItems="flex-start" gap="2">
         <Box>
-          <TranscriptStatusIcon status={result.status} />
+          <TranscriptStatusIcon
+            status={result.status}
+            dagStatus={
+              dagStatusMap?.get(result.id) ??
+              ((result as Record<string, unknown>).dag_status as
+                | DagTask[]
+                | null) ??
+              null
+            }
+          />
         </Box>
         <Box flex="1">
           {/* Title with highlighting and text fragment for deep linking */}
@@ -284,6 +297,7 @@ export default function TranscriptCards({
   isLoading,
   onDelete,
   onReprocess,
+  dagStatusMap,
 }: TranscriptCardsProps) {
   return (
     <Box position="relative">
@@ -315,6 +329,7 @@ export default function TranscriptCards({
               query={query}
               onDelete={onDelete}
               onReprocess={onReprocess}
+              dagStatusMap={dagStatusMap}
             />
           ))}
         </Stack>

www/app/(app)/browse/_components/TranscriptStatusIcon.tsx

@@ -8,13 +8,17 @@ import {
   FaGear,
 } from "react-icons/fa6";
 import { TranscriptStatus } from "../../../lib/transcript";
+import type { DagTask } from "../../../lib/UserEventsProvider";
+import DagProgressDots from "./DagProgressDots";
 
 interface TranscriptStatusIconProps {
   status: TranscriptStatus;
+  dagStatus?: DagTask[] | null;
 }
 
 export default function TranscriptStatusIcon({
   status,
+  dagStatus,
 }: TranscriptStatusIconProps) {
   switch (status) {
     case "ended":
@@ -36,6 +40,9 @@ export default function TranscriptStatusIcon({
         </Box>
       );
     case "processing":
+      if (dagStatus && dagStatus.length > 0) {
+        return <DagProgressDots tasks={dagStatus} />;
+      }
       return (
         <Box as="span" title="Processing in progress">
           <Icon color="gray.500" as={FaGear} />

www/app/(app)/browse/page.tsx

@@ -43,6 +43,7 @@ import DeleteTranscriptDialog from "./_components/DeleteTranscriptDialog";
 import { formatLocalDate } from "../../lib/time";
 import { RECORD_A_MEETING_URL } from "../../api/urls";
 import { useUserName } from "../../lib/useUserName";
+import { useDagStatusMap } from "../../lib/UserEventsProvider";
 
 const SEARCH_FORM_QUERY_INPUT_NAME = "query" as const;
 
@@ -273,6 +274,7 @@ export default function TranscriptBrowser() {
   }, [JSON.stringify(searchFilters)]);
 
   const userName = useUserName();
+  const dagStatusMap = useDagStatusMap();
   const [deletionLoading, setDeletionLoading] = useState(false);
   const cancelRef = React.useRef(null);
   const [transcriptToDeleteId, setTranscriptToDeleteId] =
@@ -408,6 +410,7 @@ export default function TranscriptBrowser() {
             isLoading={searchLoading}
             onDelete={setTranscriptToDeleteId}
             onReprocess={handleProcessTranscript}
+            dagStatusMap={dagStatusMap}
           />
 
           {!searchLoading && results.length === 0 && (

www/app/(app)/transcripts/[transcriptId]/processing/DagProgressTable.tsx

@@ -0,0 +1,190 @@
+"use client";
+
+import { useEffect, useState } from "react";
+import { Table, Box, Icon, Spinner, Text, Badge } from "@chakra-ui/react";
+import { FaCheck, FaXmark, FaClock, FaMinus } from "react-icons/fa6";
+import type { DagTask, DagTaskStatus } from "../../useWebSockets";
+
+function humanizeTaskName(name: string): string {
+  return name
+    .split("_")
+    .map((word) => word.charAt(0).toUpperCase() + word.slice(1))
+    .join(" ");
+}
+
+function formatDuration(seconds: number): string {
+  if (seconds < 60) {
+    return `${Math.round(seconds)}s`;
+  }
+  const minutes = Math.floor(seconds / 60);
+  const remainingSeconds = Math.round(seconds % 60);
+  return `${minutes}m ${remainingSeconds}s`;
+}
+
+function StatusIcon({ status }: { status: DagTaskStatus }) {
+  switch (status) {
+    case "completed":
+      return (
+        <Box as="span" title="Completed">
+          <Icon color="green.500" as={FaCheck} />
+        </Box>
+      );
+    case "running":
+      return <Spinner size="sm" color="blue.500" />;
+    case "failed":
+      return (
+        <Box as="span" title="Failed">
+          <Icon color="red.500" as={FaXmark} />
+        </Box>
+      );
+    case "queued":
+      return (
+        <Box as="span" title="Queued">
+          <Icon color="gray.400" as={FaClock} />
+        </Box>
+      );
+    case "cancelled":
+      return (
+        <Box as="span" title="Cancelled">
+          <Icon color="gray.400" as={FaMinus} />
+        </Box>
+      );
+    default:
+      return null;
+  }
+}
+
+function ElapsedTimer({ startedAt }: { startedAt: string }) {
+  const [elapsed, setElapsed] = useState<number>(() => {
+    return (Date.now() - new Date(startedAt).getTime()) / 1000;
+  });
+
+  useEffect(() => {
+    const interval = setInterval(() => {
+      setElapsed((Date.now() - new Date(startedAt).getTime()) / 1000);
+    }, 1000);
+    return () => clearInterval(interval);
+  }, [startedAt]);
+
+  return <Text fontSize="sm">{formatDuration(elapsed)}</Text>;
+}
+
+function DurationCell({ task }: { task: DagTask }) {
+  if (task.status === "completed" && task.duration_seconds !== null) {
+    return <Text fontSize="sm">{formatDuration(task.duration_seconds)}</Text>;
+  }
+  if (task.status === "running" && task.started_at) {
+    return <ElapsedTimer startedAt={task.started_at} />;
+  }
+  return (
+    <Text fontSize="sm" color="gray.400">
+      --
+    </Text>
+  );
+}
+
+function ProgressCell({ task }: { task: DagTask }) {
+  if (task.progress_pct === null && task.children_total === null) {
+    return null;
+  }
+
+  return (
+    <Box>
+      {task.progress_pct !== null && (
+        <Box
+          w="100%"
+          h="6px"
+          bg="gray.200"
+          borderRadius="full"
+          overflow="hidden"
+        >
+          <Box
+            h="100%"
+            w={`${Math.min(100, Math.max(0, task.progress_pct))}%`}
+            bg={task.status === "failed" ? "red.400" : "blue.400"}
+            borderRadius="full"
+            transition="width 0.3s ease"
+          />
+        </Box>
+      )}
+      {task.children_total !== null && (
+        <Badge
+          size="sm"
+          colorPalette="gray"
+          mt={task.progress_pct !== null ? 1 : 0}
+        >
+          {task.children_completed ?? 0}/{task.children_total}
+        </Badge>
+      )}
+    </Box>
+  );
+}
+
+function TaskRow({ task }: { task: DagTask }) {
+  const [expanded, setExpanded] = useState(false);
+  const hasFailed = task.status === "failed" && task.error;
+
+  return (
+    <>
+      <Table.Row
+        cursor={hasFailed ? "pointer" : "default"}
+        onClick={hasFailed ? () => setExpanded((prev) => !prev) : undefined}
+        _hover={hasFailed ? { bg: "gray.50" } : undefined}
+      >
+        <Table.Cell>
+          <Text fontSize="sm" fontWeight="medium">
+            {humanizeTaskName(task.name)}
+          </Text>
+        </Table.Cell>
+        <Table.Cell>
+          <StatusIcon status={task.status} />
+        </Table.Cell>
+        <Table.Cell>
+          <DurationCell task={task} />
+        </Table.Cell>
+        <Table.Cell>
+          <ProgressCell task={task} />
+        </Table.Cell>
+      </Table.Row>
+      {hasFailed && expanded && (
+        <Table.Row>
+          <Table.Cell colSpan={4}>
+            <Box bg="red.50" p={3} borderRadius="md">
+              <Text fontSize="xs" color="red.700" whiteSpace="pre-wrap">
+                {task.error}
+              </Text>
+            </Box>
+          </Table.Cell>
+        </Table.Row>
+      )}
+    </>
+  );
+}
+
+export default function DagProgressTable({ tasks }: { tasks: DagTask[] }) {
+  return (
+    <Box w="100%" overflowX="auto">
+      <Table.Root size="sm">
+        <Table.Header>
+          <Table.Row>
+            <Table.ColumnHeader fontWeight="600">Task</Table.ColumnHeader>
+            <Table.ColumnHeader fontWeight="600" width="80px">
+              Status
+            </Table.ColumnHeader>
+            <Table.ColumnHeader fontWeight="600" width="100px">
+              Duration
+            </Table.ColumnHeader>
+            <Table.ColumnHeader fontWeight="600" width="140px">
+              Progress
+            </Table.ColumnHeader>
+          </Table.Row>
+        </Table.Header>
+        <Table.Body>
+          {tasks.map((task) => (
+            <TaskRow key={task.name} task={task} />
+          ))}
+        </Table.Body>
+      </Table.Root>
+    </Box>
+  );
+}

www/app/(app)/transcripts/[transcriptId]/processing/page.tsx

@@ -11,6 +11,10 @@ import {
 import { useRouter } from "next/navigation";
 import { useTranscriptGet } from "../../../../lib/apiHooks";
 import { parseNonEmptyString } from "../../../../lib/utils";
+import { useWebSockets } from "../../useWebSockets";
+import type { DagTask } from "../../useWebSockets";
+import { useDagStatusMap } from "../../../../lib/UserEventsProvider";
+import DagProgressTable from "./DagProgressTable";
 
 type TranscriptProcessing = {
   params: Promise<{
@@ -24,9 +28,21 @@ export default function TranscriptProcessing(details: TranscriptProcessing) {
   const router = useRouter();
 
   const transcript = useTranscriptGet(transcriptId);
+  const { status: wsStatus, dagStatus: wsDagStatus } =
+    useWebSockets(transcriptId);
+  const userDagStatusMap = useDagStatusMap();
+  const userDagStatus = userDagStatusMap.get(transcriptId) ?? null;
+
+  const restDagStatus: DagTask[] | null =
+    ((transcript.data as Record<string, unknown>)?.dag_status as
+      | DagTask[]
+      | null) ?? null;
+
+  // Prefer transcript room WS (most granular), then user room WS, then REST
+  const dagStatus = wsDagStatus ?? userDagStatus ?? restDagStatus;
 
   useEffect(() => {
-    const status = transcript.data?.status;
+    const status = wsStatus?.value ?? transcript.data?.status;
     if (!status) return;
 
     if (status === "ended" || status === "error") {
@@ -41,6 +57,7 @@ export default function TranscriptProcessing(details: TranscriptProcessing) {
       router.replace(dest);
     }
   }, [
+    wsStatus?.value,
     transcript.data?.status,
     transcript.data?.source_kind,
     router,
@@ -74,11 +91,29 @@ export default function TranscriptProcessing(details: TranscriptProcessing) {
         w={{ base: "full", md: "container.xl" }}
       >
         <Center h={"full"} w="full">
-          <VStack gap={10} bg="gray.100" p={10} borderRadius="md" maxW="500px">
-            <Spinner size="xl" color="blue.500" />
-            <Heading size={"md"} textAlign="center">
-              Processing recording
-            </Heading>
+          <VStack
+            gap={10}
+            bg="gray.100"
+            p={10}
+            borderRadius="md"
+            maxW="600px"
+            w="full"
+          >
+            {dagStatus ? (
+              <>
+                <Heading size={"md"} textAlign="center">
+                  Processing recording
+                </Heading>
+                <DagProgressTable tasks={dagStatus} />
+              </>
+            ) : (
+              <>
+                <Spinner size="xl" color="blue.500" />
+                <Heading size={"md"} textAlign="center">
+                  Processing recording
+                </Heading>
+              </>
+            )}
             <Text color="gray.600" textAlign="center">
               You can safely return to the library while your recording is being
               processed.

www/app/(app)/transcripts/useWebSockets.ts

@@ -14,6 +14,9 @@ import {
 } from "../../lib/apiHooks";
 import { NonEmptyString } from "../../lib/utils";
 
+import type { DagTask } from "../../lib/dagTypes";
+export type { DagTask, DagTaskStatus } from "../../lib/dagTypes";
+
 export type UseWebSockets = {
   transcriptTextLive: string;
   translateText: string;
@@ -24,6 +27,7 @@ export type UseWebSockets = {
   status: Status | null;
   waveform: AudioWaveform | null;
   duration: number | null;
+  dagStatus: DagTask[] | null;
 };
 
 export const useWebSockets = (transcriptId: string | null): UseWebSockets => {
@@ -40,6 +44,7 @@ export const useWebSockets = (transcriptId: string | null): UseWebSockets => {
     summary: "",
   });
   const [status, setStatus] = useState<Status | null>(null);
+  const [dagStatus, setDagStatus] = useState<DagTask[] | null>(null);
   const { setError } = useError();
 
   const queryClient = useQueryClient();
@@ -431,11 +436,31 @@ export const useWebSockets = (transcriptId: string | null): UseWebSockets => {
               );
             }
             setStatus(message.data);
+            invalidateTranscript(queryClient, transcriptId as NonEmptyString);
             if (message.data.value === "ended") {
               ws.close();
             }
             break;
 
+          case "DAG_STATUS":
+            if (message.data?.tasks) {
+              setDagStatus(message.data.tasks);
+            }
+            break;
+
+          case "DAG_TASK_PROGRESS":
+            if (message.data) {
+              setDagStatus(
+                (prev) =>
+                  prev?.map((t) =>
+                    t.name === message.data.task_name
+                      ? { ...t, progress_pct: message.data.progress_pct }
+                      : t,
+                  ) ?? null,
+              );
+            }
+            break;
+
           default:
             setError(
               new Error(`Received unknown WebSocket event: ${message.event}`),
@@ -493,5 +518,6 @@ export const useWebSockets = (transcriptId: string | null): UseWebSockets => {
     status,
     waveform,
     duration,
+    dagStatus,
   };
 };

www/app/lib/UserEventsProvider.tsx

@@ -1,11 +1,25 @@
 "use client";
 
-import React, { useEffect, useRef } from "react";
+import React, { useEffect, useRef, useState } from "react";
 import { useQueryClient } from "@tanstack/react-query";
 import { WEBSOCKET_URL } from "./apiClient";
 import { useAuth } from "./AuthProvider";
 import { z } from "zod";
-import { invalidateTranscriptLists, TRANSCRIPT_SEARCH_URL } from "./apiHooks";
+import {
+  invalidateTranscript,
+  invalidateTranscriptLists,
+  TRANSCRIPT_SEARCH_URL,
+} from "./apiHooks";
+import type { NonEmptyString } from "./utils";
+
+import type { DagTask } from "./dagTypes";
+export type { DagTask, DagTaskStatus } from "./dagTypes";
+
+const DagStatusContext = React.createContext<Map<string, DagTask[]>>(new Map());
+
+export function useDagStatusMap() {
+  return React.useContext(DagStatusContext);
+}
 
 const UserEvent = z.object({
   event: z.string(),
@@ -95,6 +109,9 @@ export function UserEventsProvider({
   const queryClient = useQueryClient();
   const tokenRef = useRef<string | null>(null);
   const detachRef = useRef<(() => void) | null>(null);
+  const [dagStatusMap, setDagStatusMap] = useState<Map<string, DagTask[]>>(
+    new Map(),
+  );
 
   useEffect(() => {
     // Only tear down when the user is truly unauthenticated
@@ -133,20 +150,52 @@ export function UserEventsProvider({
     if (!detachRef.current) {
       const onMessage = (event: MessageEvent) => {
         try {
-          const msg = UserEvent.parse(JSON.parse(event.data));
+          const fullMsg = JSON.parse(event.data);
+          const msg = UserEvent.parse(fullMsg);
           const eventName = msg.event;
-
           const invalidateList = () => invalidateTranscriptLists(queryClient);
 
           switch (eventName) {
             case "TRANSCRIPT_CREATED":
             case "TRANSCRIPT_DELETED":
-            case "TRANSCRIPT_STATUS":
             case "TRANSCRIPT_FINAL_TITLE":
             case "TRANSCRIPT_DURATION":
               invalidateList().then(() => {});
               break;
 
+            case "TRANSCRIPT_STATUS": {
+              invalidateList().then(() => {});
+              const transcriptId = fullMsg.data?.id as string | undefined;
+              if (transcriptId) {
+                invalidateTranscript(
+                  queryClient,
+                  transcriptId as NonEmptyString,
+                ).then(() => {});
+              }
+              const status = fullMsg.data?.value as string | undefined;
+              if (transcriptId && status && status !== "processing") {
+                setDagStatusMap((prev) => {
+                  const next = new Map(prev);
+                  next.delete(transcriptId);
+                  return next;
+                });
+              }
+              break;
+            }
+
+            case "TRANSCRIPT_DAG_STATUS": {
+              const transcriptId = fullMsg.data?.id as string | undefined;
+              const tasks = fullMsg.data?.tasks as DagTask[] | undefined;
+              if (transcriptId && tasks) {
+                setDagStatusMap((prev) => {
+                  const next = new Map(prev);
+                  next.set(transcriptId, tasks);
+                  return next;
+                });
+              }
+              break;
+            }
+
             default:
               // Ignore other content events for list updates
               break;
@@ -176,5 +225,9 @@ export function UserEventsProvider({
     };
   }, []);
 
-  return <>{children}</>;
+  return (
+    <DagStatusContext.Provider value={dagStatusMap}>
+      {children}
+    </DagStatusContext.Provider>
+  );
 }

www/app/lib/dagTypes.ts

@@ -0,0 +1,19 @@
+export type DagTaskStatus =
+  | "queued"
+  | "running"
+  | "completed"
+  | "failed"
+  | "cancelled";
+
+export type DagTask = {
+  name: string;
+  status: DagTaskStatus;
+  started_at: string | null;
+  finished_at: string | null;
+  duration_seconds: number | null;
+  parents: string[];
+  error: string | null;
+  children_total: number | null;
+  children_completed: number | null;
+  progress_pct: number | null;
+};