async def test_workflow_heartbeat_timeout_updates_sync_job_status(self, mock_dicts, mock_sync_job_id):
        """Test that heartbeat timeout (worker death) properly updates sync job status."""

        async with WorkflowEnvironment() as env:
            # Track what happened
            status_update_called = False
            final_status = None
            final_error = None

            async def mock_run_sync_activity_no_heartbeat(*args, **kwargs):
                """Simulate an activity that stops heartbeating (worker dies)."""
                # Don't send any heartbeats - simulate dead worker
                await asyncio.sleep(300)  # Sleep longer than heartbeat timeout

            async def mock_update_status_activity(*args, **kwargs):
                """Capture the status update."""
                nonlocal status_update_called, final_status, final_error

                sync_job_id, status, user_dict, error, failed_at = args

                # Verify correct sync job ID
                assert sync_job_id == mock_sync_job_id

                # Capture the final status and error
                final_status = status
                final_error = error
                status_update_called = True

                print(f"Heartbeat timeout test - Status: {status}, Error: {error}")

            # Register mocked activities
            env.set_activity_implementation(run_sync_activity, mock_run_sync_activity_no_heartbeat)
            env.set_activity_implementation(update_sync_job_status_activity, mock_update_status_activity)

            # Start the workflow with a short heartbeat timeout for testing
            # Note: In the real workflow, heartbeat_timeout is set to 2 minutes
            # For testing, we'd need to modify the workflow or wait 2 minutes
            workflow_handle = await env.client.start_workflow(
                RunSourceConnectionWorkflow.run,
                args=[
                    mock_dicts["sync_dict"],
                    mock_dicts["sync_job_dict"],
                    mock_dicts["sync_dag_dict"],
                    mock_dicts["collection_dict"],
                    mock_dicts["source_connection_dict"],
                    mock_dicts["user_dict"],
                    None,
                ],
                id=f"test-workflow-heartbeat-{mock_sync_job_id}",
                task_queue="test-queue",
            )

            # Wait for the workflow to fail due to heartbeat timeout
            with pytest.raises(WorkflowFailureError) as exc_info:
                await workflow_handle.result()

            # Verify the status update was called
            assert status_update_called, "Status update activity should have been called"
            assert final_status == "FAILED", f"Expected status to be FAILED, got {final_status}"
            # The error should mention timeout or heartbeat
            print(f"Final error message: {final_error}")
Test that heartbeat timeout (worker death) properly updates sync job status.
backend/tests/integration/temporal/test_workflow_cancellation.py