fix: webhook parity, pipeline rename, waveform constant fix (#806)

* pipeline fixes: whereby Hatchet preparation * send_webhook fixes * cleanup * self-review * comment * webhook util functions: less dependencies * remove comment --------- Co-authored-by: Igor Loskutov <igor.loskutoff@gmail.com>
2026-02-05 02:16:46 +00:00 · 2025-12-26 18:00:32 -05:00
parent 2d0df48767
commit 5f7b1ff1a6
9 changed files with 418 additions and 231 deletions
--- a/server/reflector/hatchet/workflows/init.py
+++ b/server/reflector/hatchet/workflows/init.py
@@ -1,8 +1,8 @@
 """Hatchet workflow definitions."""

-from reflector.hatchet.workflows.diarization_pipeline import (
+from reflector.hatchet.workflows.daily_multitrack_pipeline import (
    PipelineInput,
-    diarization_pipeline,
+    daily_multitrack_pipeline,
 )
 from reflector.hatchet.workflows.subject_processing import (
    SubjectInput,
@@ -15,7 +15,7 @@ from reflector.hatchet.workflows.topic_chunk_processing import (
 from reflector.hatchet.workflows.track_processing import TrackInput, track_workflow

 __all__ = [
-    "diarization_pipeline",
+    "daily_multitrack_pipeline",
    "subject_workflow",
    "topic_chunk_workflow",
    "track_workflow",
--- a/server/reflector/hatchet/workflows/daily_multitrack_pipeline.py
+++ b/server/reflector/hatchet/workflows/daily_multitrack_pipeline.py
@@ -1,9 +1,12 @@
 """
-Hatchet main workflow: DiarizationPipeline
+Hatchet main workflow: DailyMultitrackPipeline

-Multitrack diarization pipeline for Daily.co recordings.
+Multitrack processing pipeline for Daily.co recordings.
 Orchestrates the full processing flow from recording metadata to final transcript.

+Daily.co recordings don't require ML diarization - speaker identification comes from
+track index (each participant's audio is a separate track).
+
 Note: This file uses deferred imports (inside functions/tasks) intentionally.
 Hatchet workers run in forked processes; fresh imports per task ensure DB connections
 are not shared across forks, avoiding connection pooling issues.
@@ -102,7 +105,7 @@ from reflector.zulip import post_transcript_notification


 class PipelineInput(BaseModel):
-    """Input to trigger the diarization pipeline."""
+    """Input to trigger the Daily.co multitrack pipeline."""

    recording_id: NonEmptyString
    tracks: list[dict]  # List of {"s3_key": str}
@@ -113,7 +116,7 @@ class PipelineInput(BaseModel):

 hatchet = HatchetClientManager.get_client()

-diarization_pipeline = hatchet.workflow(
+daily_multitrack_pipeline = hatchet.workflow(
    name="DiarizationPipeline", input_validator=PipelineInput
 )

@@ -172,8 +175,6 @@ def _spawn_storage():


 class Loggable(Protocol):
-    """Protocol for objects with a log method."""
-
    def log(self, message: str) -> None: ...


@@ -249,7 +250,7 @@ def with_error_handling(
    return decorator


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    execution_timeout=timedelta(seconds=TIMEOUT_SHORT), retries=3
 )
@with_error_handling(TaskName.GET_RECORDING)
@@ -302,7 +303,7 @@ async def get_recording(input: PipelineInput, ctx: Context) -> RecordingResult:
    )


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    parents=[get_recording],
    execution_timeout=timedelta(seconds=TIMEOUT_SHORT),
    retries=3,
@@ -393,7 +394,7 @@ async def get_participants(input: PipelineInput, ctx: Context) -> ParticipantsRe
    )


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    parents=[get_participants],
    execution_timeout=timedelta(seconds=TIMEOUT_HEAVY),
    retries=3,
@@ -462,7 +463,7 @@ async def process_tracks(input: PipelineInput, ctx: Context) -> ProcessTracksRes
    )


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    parents=[process_tracks],
    execution_timeout=timedelta(seconds=TIMEOUT_AUDIO),
    retries=3,
@@ -559,7 +560,7 @@ async def mixdown_tracks(input: PipelineInput, ctx: Context) -> MixdownResult:
    )


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    parents=[mixdown_tracks],
    execution_timeout=timedelta(seconds=TIMEOUT_MEDIUM),
    retries=3,
@@ -627,7 +628,7 @@ async def generate_waveform(input: PipelineInput, ctx: Context) -> WaveformResul
    return WaveformResult(waveform_generated=True)


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    parents=[process_tracks],
    execution_timeout=timedelta(seconds=TIMEOUT_HEAVY),
    retries=3,
@@ -732,7 +733,7 @@ async def detect_topics(input: PipelineInput, ctx: Context) -> TopicsResult:
    return TopicsResult(topics=topics_list)


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    parents=[detect_topics],
    execution_timeout=timedelta(seconds=TIMEOUT_HEAVY),
    retries=3,
@@ -797,7 +798,7 @@ async def generate_title(input: PipelineInput, ctx: Context) -> TitleResult:
    return TitleResult(title=title_result)


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    parents=[detect_topics],
    execution_timeout=timedelta(seconds=TIMEOUT_MEDIUM),
    retries=3,
@@ -875,7 +876,7 @@ async def extract_subjects(input: PipelineInput, ctx: Context) -> SubjectsResult
    )


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    parents=[extract_subjects],
    execution_timeout=timedelta(seconds=TIMEOUT_HEAVY),
    retries=3,
@@ -917,7 +918,7 @@ async def process_subjects(input: PipelineInput, ctx: Context) -> ProcessSubject
    return ProcessSubjectsResult(subject_summaries=subject_summaries)


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    parents=[process_subjects],
    execution_timeout=timedelta(seconds=TIMEOUT_MEDIUM),
    retries=3,
@@ -1006,7 +1007,7 @@ async def generate_recap(input: PipelineInput, ctx: Context) -> RecapResult:
    return RecapResult(short_summary=short_summary, long_summary=long_summary)


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    parents=[extract_subjects],
    execution_timeout=timedelta(seconds=TIMEOUT_LONG),
    retries=3,
@@ -1074,7 +1075,7 @@ async def identify_action_items(
    return ActionItemsResult(action_items=action_items_response)


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    parents=[generate_waveform, generate_title, generate_recap, identify_action_items],
    execution_timeout=timedelta(seconds=TIMEOUT_SHORT),
    retries=3,
@@ -1159,7 +1160,7 @@ async def finalize(input: PipelineInput, ctx: Context) -> FinalizeResult:
    return FinalizeResult(status="COMPLETED")


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    parents=[finalize], execution_timeout=timedelta(seconds=TIMEOUT_SHORT), retries=3
 )
@with_error_handling(TaskName.CLEANUP_CONSENT, set_error_status=False)
@@ -1259,7 +1260,7 @@ async def cleanup_consent(input: PipelineInput, ctx: Context) -> ConsentResult:
    return ConsentResult()


-@diarization_pipeline.task(
+@daily_multitrack_pipeline.task(
    parents=[cleanup_consent],
    execution_timeout=timedelta(seconds=TIMEOUT_SHORT),
    retries=5,
@@ -1286,14 +1287,14 @@ async def post_zulip(input: PipelineInput, ctx: Context) -> ZulipResult:
    return ZulipResult(zulip_message_id=message_id)


-@diarization_pipeline.task(
-    parents=[post_zulip],
+@daily_multitrack_pipeline.task(
+    parents=[cleanup_consent],
    execution_timeout=timedelta(seconds=TIMEOUT_MEDIUM),
-    retries=30,
+    retries=5,
 )
@with_error_handling(TaskName.SEND_WEBHOOK, set_error_status=False)
 async def send_webhook(input: PipelineInput, ctx: Context) -> WebhookResult:
-    """Send completion webhook to external service."""
+    """Send completion webhook to external service with full payload and HMAC signature."""
    ctx.log(f"send_webhook: transcript_id={input.transcript_id}")

    if not input.room_id:
@@ -1302,27 +1303,39 @@ async def send_webhook(input: PipelineInput, ctx: Context) -> WebhookResult:

    async with fresh_db_connection():
        from reflector.db.rooms import rooms_controller  # noqa: PLC0415
-        from reflector.db.transcripts import transcripts_controller  # noqa: PLC0415
+        from reflector.utils.webhook import (  # noqa: PLC0415
+            fetch_transcript_webhook_payload,
+            send_webhook_request,
+        )

        room = await rooms_controller.get_by_id(input.room_id)
-        transcript = await transcripts_controller.get_by_id(input.transcript_id)
+        if not room or not room.webhook_url:
+            ctx.log("send_webhook skipped (no webhook_url configured)")
+            return WebhookResult(webhook_sent=False, skipped=True)

-        if room and room.webhook_url and transcript:
-            webhook_payload = {
-                "event": "transcript.completed",
-                "transcript_id": input.transcript_id,
-                "title": transcript.title,
-                "duration": transcript.duration,
-            }
+        payload = await fetch_transcript_webhook_payload(
+            transcript_id=input.transcript_id,
+            room_id=input.room_id,
+        )

-            async with httpx.AsyncClient() as client:
-                response = await client.post(
-                    room.webhook_url, json=webhook_payload, timeout=30
-                )
-                response.raise_for_status()
+        if isinstance(payload, str):
+            ctx.log(f"send_webhook skipped (could not build payload): {payload}")
+            return WebhookResult(webhook_sent=False, skipped=True)

-            ctx.log(f"send_webhook complete: status_code={response.status_code}")
+        ctx.log(
+            f"send_webhook: sending to {room.webhook_url} "
+            f"(topics={len(payload.transcript.topics)}, "
+            f"participants={len(payload.transcript.participants)})"
+        )

-            return WebhookResult(webhook_sent=True, response_code=response.status_code)
+        response = await send_webhook_request(
+            url=room.webhook_url,
+            payload=payload,
+            event_type="transcript.completed",
+            webhook_secret=room.webhook_secret,
+            timeout=30.0,
+        )

-    return WebhookResult(webhook_sent=False, skipped=True)
+        ctx.log(f"send_webhook complete: status_code={response.status_code}")
+
+        return WebhookResult(webhook_sent=True, response_code=response.status_code)
--- a/server/reflector/hatchet/workflows/track_processing.py
+++ b/server/reflector/hatchet/workflows/track_processing.py
@@ -4,10 +4,10 @@ Hatchet child workflow: TrackProcessing
 Handles individual audio track processing: padding and transcription.
 Spawned dynamically by the main diarization pipeline for each track.

-Architecture note: This is a separate workflow (not inline tasks in DiarizationPipeline)
+Architecture note: This is a separate workflow (not inline tasks in DailyMultitrackPipeline)
 because Hatchet workflow DAGs are defined statically, but the number of tracks varies
 at runtime. Child workflow spawning via `aio_run()` + `asyncio.gather()` is the
-standard pattern for dynamic fan-out. See `process_tracks` in diarization_pipeline.py.
+standard pattern for dynamic fan-out. See `process_tracks` in daily_multitrack_pipeline.py.

 Note: This file uses deferred imports (inside tasks) intentionally.
 Hatchet workers run in forked processes; fresh imports per task ensure