diff --git a/server/reflector/hatchet/dag_zulip.py b/server/reflector/hatchet/dag_zulip.py new file mode 100644 index 00000000..aa0c0bd7 --- /dev/null +++ b/server/reflector/hatchet/dag_zulip.py @@ -0,0 +1,144 @@ +""" +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, + ) diff --git a/server/reflector/hatchet/workflows/daily_multitrack_pipeline.py b/server/reflector/hatchet/workflows/daily_multitrack_pipeline.py index 188133c7..868edaf7 100644 --- a/server/reflector/hatchet/workflows/daily_multitrack_pipeline.py +++ b/server/reflector/hatchet/workflows/daily_multitrack_pipeline.py @@ -45,6 +45,7 @@ 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, @@ -238,7 +239,14 @@ def with_error_handling( @functools.wraps(func) async def wrapper(input: PipelineInput, ctx: Context) -> R: try: - return await func(input, ctx) + result = await func(input, ctx) + try: + await update_dag_zulip_message( + input.transcript_id, ctx.workflow_run_id + ) + except Exception: + pass + return result except Exception as e: logger.error( f"[Hatchet] {step_name} failed", @@ -246,6 +254,14 @@ def with_error_handling( error=str(e), exc_info=True, ) + try: + await update_dag_zulip_message( + input.transcript_id, + ctx.workflow_run_id, + error_message=f"{step_name} failed: {e}", + ) + except Exception: + pass if set_error_status: await set_workflow_error_status(input.transcript_id) raise @@ -1294,6 +1310,11 @@ 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: diff --git a/server/reflector/services/transcript_process.py b/server/reflector/services/transcript_process.py index 13847a49..24a6fbe5 100644 --- a/server/reflector/services/transcript_process.py +++ b/server/reflector/services/transcript_process.py @@ -17,6 +17,7 @@ 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 @@ -266,6 +267,16 @@ async def dispatch_transcript_processing( transcript, {"workflow_run_id": workflow_id} ) + try: + await create_dag_zulip_message(config.transcript_id, workflow_id) + except Exception: + logger.warning( + "[DAG Zulip] Failed to create DAG message at dispatch", + transcript_id=config.transcript_id, + workflow_id=workflow_id, + exc_info=True, + ) + logger.info("Hatchet workflow dispatched", workflow_id=workflow_id) return None diff --git a/server/reflector/settings.py b/server/reflector/settings.py index 4955d568..125931f2 100644 --- a/server/reflector/settings.py +++ b/server/reflector/settings.py @@ -161,6 +161,9 @@ 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 diff --git a/server/reflector/worker/process.py b/server/reflector/worker/process.py index c1d26d96..27e597e0 100644 --- a/server/reflector/worker/process.py +++ b/server/reflector/worker/process.py @@ -25,6 +25,7 @@ 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 @@ -372,6 +373,16 @@ 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 @@ -1079,6 +1090,16 @@ 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, diff --git a/server/reflector/zulip.py b/server/reflector/zulip.py index 2b83ddf0..3e33a6ed 100644 --- a/server/reflector/zulip.py +++ b/server/reflector/zulip.py @@ -12,9 +12,16 @@ 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 httpx.AsyncClient() as client: + async with _zulip_client() 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), @@ -29,7 +36,7 @@ async def get_zulip_topics(stream_id: int) -> list[dict]: async def get_zulip_streams() -> list[dict]: try: - async with httpx.AsyncClient() as client: + async with _zulip_client() as client: response = await client.get( f"https://{settings.ZULIP_REALM}/api/v1/streams", auth=(settings.ZULIP_BOT_EMAIL, settings.ZULIP_API_KEY), @@ -44,7 +51,7 @@ async def get_zulip_streams() -> list[dict]: async def send_message_to_zulip(stream: str, topic: str, content: str): try: - async with httpx.AsyncClient() as client: + async with _zulip_client() as client: response = await client.post( f"https://{settings.ZULIP_REALM}/api/v1/messages", data={ @@ -66,7 +73,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 httpx.AsyncClient() as client: + async with _zulip_client() as client: response = await client.patch( f"https://{settings.ZULIP_REALM}/api/v1/messages/{message_id}", data={ @@ -90,6 +97,27 @@ 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}" diff --git a/server/tests/test_dag_zulip.py b/server/tests/test_dag_zulip.py new file mode 100644 index 00000000..174753b7 --- /dev/null +++ b/server/tests/test_dag_zulip.py @@ -0,0 +1,536 @@ +""" +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"