Temporal Cloud (signalroom-713.nzg5u)
    │
    ├── Schedules (cron triggers)
    │
    └── Workflows (orchestration)
            │
            └── Activities (actual work)
                    │
                    └── dlt Pipelines / Reports / Notifications

# Sync a single source
SyncSourceWorkflow(source_name, resources, notify_on_success, notify_on_failure)

# Sync multiple sources sequentially
ScheduledSyncWorkflow(sources)

# Run and send a report
RunReportWorkflow(report_name, channel, send)

# Via script (recommended)
python scripts/trigger_workflow.py everflow -w

# Programmatically
from signalroom.temporal.config import get_temporal_client

client = await get_temporal_client()
await client.start_workflow(
    SyncSourceWorkflow.run,
    args=[...],
    id="sync-everflow-manual",
    task_queue="api-tasks"
)

# Setup/update schedules
python scripts/setup_schedules.py

# Delete all schedules
python scripts/setup_schedules.py --delete

RETRY_POLICY = RetryPolicy(
    initial_interval=timedelta(seconds=1),
    maximum_interval=timedelta(minutes=5),
    backoff_coefficient=2.0,
    maximum_attempts=5,
    non_retryable_error_types=["ValueError", "KeyError"],
)

# Local
make logs-worker

# Fly.io
fly logs

from temporalio.worker import UnsandboxedWorkflowRunner

worker = Worker(
    client,
    task_queue=settings.temporal_task_queue,
    workflows=[...],
    activities=[...],
    workflow_runner=UnsandboxedWorkflowRunner(),  # Add this
)

# WRONG - nested event loop
@activity.defn
async def my_activity():
    result = some_sync_function_that_uses_asyncio_run()  # Fails!

# RIGHT - await directly
@activity.defn
async def my_activity():
    result = await some_async_function()

@activity.defn
async def long_activity():
    for batch in batches:
        activity.heartbeat()  # Prevent timeout
        await process(batch)