feat: add redial option in campaigns

api/db/campaign_client.py

@@ -1,7 +1,7 @@
 from datetime import UTC, datetime
 from typing import Any, Dict, List, Optional
 
-from sqlalchemy import func
+from sqlalchemy import func, text
 from sqlalchemy.future import select
 
 from api.db.base_client import BaseDBClient
@@ -271,6 +271,153 @@ class CampaignClient(BaseDBClient):
             ]
             return runs, total_count
 
+    async def create_redial_campaign(
+        self,
+        parent_campaign: CampaignModel,
+        new_name: str,
+        retry_config: Optional[dict],
+        queued_runs_data: list[dict],
+    ) -> CampaignModel:
+        """Atomically create a redial child campaign, seed its queued_runs, and
+        link the parent.
+
+        - The child inherits `workflow_id`, `source_type`, `source_id`,
+          `created_by`, `organization_id`, and orchestrator settings
+          (`max_concurrency`, `schedule_config`, `circuit_breaker`) from the
+          parent. `parent_campaign_id` is stored in the child's
+          orchestrator_metadata.
+        - `queued_runs_data` should be pre-built dicts with campaign_id set to 0
+          (will be replaced once the child id is known).
+        - Parent's orchestrator_metadata gets `redialed_campaign_id` set.
+        - All inserts/updates happen in a single transaction.
+        """
+        async with self.async_session() as session:
+            parent_meta = dict(parent_campaign.orchestrator_metadata or {})
+            if parent_meta.get("redialed_campaign_id"):
+                raise ValueError(
+                    f"Campaign {parent_campaign.id} has already been redialed"
+                )
+
+            child_meta = {
+                k: v
+                for k, v in parent_meta.items()
+                if k in ("max_concurrency", "schedule_config", "circuit_breaker")
+            }
+            child_meta["parent_campaign_id"] = parent_campaign.id
+
+            child = CampaignModel(
+                name=new_name,
+                workflow_id=parent_campaign.workflow_id,
+                source_type=parent_campaign.source_type,
+                source_id=parent_campaign.source_id,
+                created_by=parent_campaign.created_by,
+                organization_id=parent_campaign.organization_id,
+                retry_config=retry_config
+                if retry_config
+                else CampaignModel.retry_config.default.arg,
+                orchestrator_metadata=child_meta,
+                rate_limit_per_second=parent_campaign.rate_limit_per_second,
+                total_rows=len(queued_runs_data),
+                source_sync_status="completed",
+            )
+            session.add(child)
+            await session.flush()  # assign child.id
+
+            for data in queued_runs_data:
+                data["campaign_id"] = child.id
+            session.add_all([QueuedRunModel(**data) for data in queued_runs_data])
+
+            parent_meta["redialed_campaign_id"] = child.id
+            parent_stmt = select(CampaignModel).where(
+                CampaignModel.id == parent_campaign.id
+            )
+            parent_result = await session.execute(parent_stmt)
+            parent_row = parent_result.scalar_one()
+            parent_row.orchestrator_metadata = parent_meta
+            parent_row.updated_at = datetime.now(UTC)
+
+            try:
+                await session.commit()
+            except Exception as e:
+                await session.rollback()
+                raise e
+            await session.refresh(child)
+            return child
+
+    async def get_redial_candidates(
+        self,
+        campaign_id: int,
+        include_voicemail: bool,
+        include_no_answer: bool,
+        include_busy: bool,
+    ) -> list[dict]:
+        """Return root context_variables for subscribers whose LATEST
+        workflow_run indicates the call should be redialed.
+
+        A subscriber (identified by `source_uuid`) is a redial candidate iff
+        the latest workflow_run (by created_at) for that source_uuid has a
+        `call_tags` entry matching any of the selected failure reasons. Uses
+        the root queued_run (retry_count=0) for the original context.
+        """
+        tag_clauses = []
+        if include_voicemail:
+            tag_clauses.append(
+                "(lr.gathered_context::jsonb -> 'call_tags') @> '[\"voicemail_detected\"]'::jsonb"
+            )
+        if include_no_answer:
+            tag_clauses.append(
+                "(lr.gathered_context::jsonb -> 'call_tags') @> '[\"telephony_no-answer\"]'::jsonb"
+            )
+        if include_busy:
+            tag_clauses.append(
+                "(lr.gathered_context::jsonb -> 'call_tags') @> '[\"telephony_busy\"]'::jsonb"
+            )
+
+        if not tag_clauses:
+            return []
+
+        tag_filter = " OR ".join(tag_clauses)
+        # Retries create new queued_runs with suffixed source_uuids linked via
+        # parent_queued_run_id, so group by the ROOT queued_run using a
+        # recursive walk and pick the latest workflow_run across the tree.
+        sql = text(
+            f"""
+            WITH RECURSIVE run_tree AS (
+                SELECT id AS root_id, id AS run_id
+                FROM queued_runs
+                WHERE campaign_id = :cid
+                  AND parent_queued_run_id IS NULL
+                UNION ALL
+                SELECT rt.root_id, q.id
+                FROM run_tree rt
+                JOIN queued_runs q ON q.parent_queued_run_id = rt.run_id
+                WHERE q.campaign_id = :cid
+            ),
+            latest_run_per_root AS (
+                SELECT DISTINCT ON (rt.root_id)
+                    rt.root_id,
+                    wr.gathered_context
+                FROM run_tree rt
+                JOIN workflow_runs wr
+                  ON wr.queued_run_id = rt.run_id
+                 AND wr.campaign_id = :cid
+                ORDER BY rt.root_id, wr.created_at DESC
+            )
+            SELECT q0.source_uuid, q0.context_variables
+            FROM queued_runs q0
+            JOIN latest_run_per_root lr ON lr.root_id = q0.id
+            WHERE q0.campaign_id = :cid
+              AND ({tag_filter})
+            """
+        )
+
+        async with self.async_session() as session:
+            result = await session.execute(sql, {"cid": campaign_id})
+            return [
+                {"source_uuid": row[0], "context_variables": row[1]}
+                for row in result.all()
+            ]
+
     async def get_campaign_by_id(self, campaign_id: int) -> Optional[CampaignModel]:
         """Get campaign by ID without organization check (for internal use)"""
         async with self.async_session() as session:
@@ -352,6 +499,35 @@ class CampaignClient(BaseDBClient):
             result = await session.execute(query)
             return result.scalar() or 0
 
+    async def get_queued_runs_stats_for_campaigns(
+        self, campaign_ids: List[int]
+    ) -> Dict[int, Dict[str, int]]:
+        """Return {campaign_id: {"total": N, "executed": M}} for given campaigns.
+
+        "executed" means queued runs in the "processed" state.
+        """
+        if not campaign_ids:
+            return {}
+        async with self.async_session() as session:
+            query = (
+                select(
+                    QueuedRunModel.campaign_id,
+                    QueuedRunModel.state,
+                    func.count(QueuedRunModel.id),
+                )
+                .where(QueuedRunModel.campaign_id.in_(campaign_ids))
+                .group_by(QueuedRunModel.campaign_id, QueuedRunModel.state)
+            )
+            result = await session.execute(query)
+            stats: Dict[int, Dict[str, int]] = {
+                cid: {"total": 0, "executed": 0} for cid in campaign_ids
+            }
+            for campaign_id, state, count in result.all():
+                stats[campaign_id]["total"] += count
+                if state == "processed":
+                    stats[campaign_id]["executed"] += count
+            return stats
+
     async def get_workflow_runs_by_campaign(
         self, campaign_id: int
     ) -> list[WorkflowRunModel]:
@@ -367,22 +543,31 @@ class CampaignClient(BaseDBClient):
 
     async def get_completed_runs_for_report(
         self,
-        campaign_id: int,
+        *,
+        campaign_id: Optional[int] = None,
+        workflow_id: Optional[int] = None,
         start_date: Optional[datetime] = None,
         end_date: Optional[datetime] = None,
     ) -> list:
-        """Get completed workflow runs for campaign report CSV.
+        """Get completed workflow runs for a run report CSV.
 
+        Scope the query by exactly one of campaign_id or workflow_id.
         Returns rows with only the columns needed for report generation.
         """
+        if (campaign_id is None) == (workflow_id is None):
+            raise ValueError("Provide exactly one of campaign_id or workflow_id")
+
         async with self.async_session() as session:
             conditions = [
-                WorkflowRunModel.campaign_id == campaign_id,
                 WorkflowRunModel.is_completed.is_(True),
                 WorkflowRunModel.cost_info["call_duration_seconds"]
                 .as_string()
                 .isnot(None),
             ]
+            if campaign_id is not None:
+                conditions.append(WorkflowRunModel.campaign_id == campaign_id)
+            if workflow_id is not None:
+                conditions.append(WorkflowRunModel.workflow_id == workflow_id)
             if start_date is not None:
                 conditions.append(WorkflowRunModel.created_at >= start_date)
             if end_date is not None:
@@ -391,11 +576,13 @@ class CampaignClient(BaseDBClient):
             query = (
                 select(
                     WorkflowRunModel.id,
+                    WorkflowRunModel.workflow_id,
+                    WorkflowRunModel.definition_id,
+                    WorkflowRunModel.campaign_id,
                     WorkflowRunModel.created_at,
                     WorkflowRunModel.initial_context,
                     WorkflowRunModel.gathered_context,
                     WorkflowRunModel.cost_info,
-                    WorkflowRunModel.logs,
                     WorkflowRunModel.public_access_token,
                 )
                 .where(*conditions)

api/routes/campaign.py

@@ -183,6 +183,10 @@ class CampaignResponse(BaseModel):
     max_concurrency: Optional[int] = None
     schedule_config: Optional[ScheduleConfigResponse] = None
     circuit_breaker: Optional[CircuitBreakerConfigResponse] = None
+    executed_count: int = 0
+    total_queued_count: int = 0
+    parent_campaign_id: Optional[int] = None
+    redialed_campaign_id: Optional[int] = None
 
 
 class CampaignsResponse(BaseModel):
@@ -223,7 +227,12 @@ class CampaignProgressResponse(BaseModel):
 # Default retry config for campaigns
 
 
-def _build_campaign_response(campaign, workflow_name: str) -> CampaignResponse:
+def _build_campaign_response(
+    campaign,
+    workflow_name: str,
+    executed_count: int = 0,
+    total_queued_count: int = 0,
+) -> CampaignResponse:
     """Build a CampaignResponse from a campaign model."""
     # Get retry_config from campaign or use defaults
     retry_config = (
@@ -236,6 +245,8 @@ def _build_campaign_response(campaign, workflow_name: str) -> CampaignResponse:
     max_concurrency = None
     schedule_config = None
     circuit_breaker_config = CircuitBreakerConfigResponse()
+    parent_campaign_id = None
+    redialed_campaign_id = None
     if campaign.orchestrator_metadata:
         max_concurrency = campaign.orchestrator_metadata.get("max_concurrency")
         sc = campaign.orchestrator_metadata.get("schedule_config")
@@ -248,6 +259,10 @@ def _build_campaign_response(campaign, workflow_name: str) -> CampaignResponse:
         cb = campaign.orchestrator_metadata.get("circuit_breaker")
         if cb:
             circuit_breaker_config = CircuitBreakerConfigResponse(**cb)
+        parent_campaign_id = campaign.orchestrator_metadata.get("parent_campaign_id")
+        redialed_campaign_id = campaign.orchestrator_metadata.get(
+            "redialed_campaign_id"
+        )
 
     return CampaignResponse(
         id=campaign.id,
@@ -267,9 +282,20 @@ def _build_campaign_response(campaign, workflow_name: str) -> CampaignResponse:
         max_concurrency=max_concurrency,
         schedule_config=schedule_config,
         circuit_breaker=circuit_breaker_config,
+        executed_count=executed_count,
+        total_queued_count=total_queued_count,
+        parent_campaign_id=parent_campaign_id,
+        redialed_campaign_id=redialed_campaign_id,
     )
 
 
+async def _get_campaign_stats(campaign_id: int) -> tuple[int, int]:
+    """Return (executed_count, total_queued_count) for a campaign."""
+    stats_map = await db_client.get_queued_runs_stats_for_campaigns([campaign_id])
+    s = stats_map.get(campaign_id, {})
+    return s.get("executed", 0), s.get("total", 0)
+
+
 @router.post("/create")
 async def create_campaign(
     request: CreateCampaignRequest,
@@ -374,8 +400,17 @@ async def get_campaigns(
     )
     workflow_map = {w.id: w.name for w in workflows}
 
+    stats_map = await db_client.get_queued_runs_stats_for_campaigns(
+        [c.id for c in campaigns]
+    )
+
     campaign_responses = [
-        _build_campaign_response(c, workflow_map.get(c.workflow_id, "Unknown"))
+        _build_campaign_response(
+            c,
+            workflow_map.get(c.workflow_id, "Unknown"),
+            executed_count=stats_map.get(c.id, {}).get("executed", 0),
+            total_queued_count=stats_map.get(c.id, {}).get("total", 0),
+        )
         for c in campaigns
     ]
 
@@ -394,7 +429,10 @@ async def get_campaign(
 
     workflow_name = await db_client.get_workflow_name(campaign.workflow_id, user.id)
 
-    return _build_campaign_response(campaign, workflow_name or "Unknown")
+    executed, total = await _get_campaign_stats(campaign.id)
+    return _build_campaign_response(
+        campaign, workflow_name or "Unknown", executed, total
+    )
 
 
 @router.post("/{campaign_id}/start")
@@ -435,7 +473,10 @@ async def start_campaign(
     campaign = await db_client.get_campaign(campaign_id, user.selected_organization_id)
     workflow_name = await db_client.get_workflow_name(campaign.workflow_id, user.id)
 
-    return _build_campaign_response(campaign, workflow_name or "Unknown")
+    executed, total = await _get_campaign_stats(campaign.id)
+    return _build_campaign_response(
+        campaign, workflow_name or "Unknown", executed, total
+    )
 
 
 @router.post("/{campaign_id}/pause")
@@ -459,7 +500,10 @@ async def pause_campaign(
     campaign = await db_client.get_campaign(campaign_id, user.selected_organization_id)
     workflow_name = await db_client.get_workflow_name(campaign.workflow_id, user.id)
 
-    return _build_campaign_response(campaign, workflow_name or "Unknown")
+    executed, total = await _get_campaign_stats(campaign.id)
+    return _build_campaign_response(
+        campaign, workflow_name or "Unknown", executed, total
+    )
 
 
 @router.patch("/{campaign_id}")
@@ -519,7 +563,10 @@ async def update_campaign(
     campaign = await db_client.get_campaign(campaign_id, user.selected_organization_id)
     workflow_name = await db_client.get_workflow_name(campaign.workflow_id, user.id)
 
-    return _build_campaign_response(campaign, workflow_name or "Unknown")
+    executed, total = await _get_campaign_stats(campaign.id)
+    return _build_campaign_response(
+        campaign, workflow_name or "Unknown", executed, total
+    )
 
 
 @router.get("/{campaign_id}/runs")
@@ -586,6 +633,101 @@ async def get_campaign_runs(
     )
 
 
+class RedialCampaignRequest(BaseModel):
+    name: Optional[str] = Field(
+        None, min_length=1, max_length=255, description="Name for the redial campaign"
+    )
+    retry_on_voicemail: bool = True
+    retry_on_no_answer: bool = True
+    retry_on_busy: bool = True
+    retry_config: Optional[RetryConfigRequest] = None
+
+    @model_validator(mode="after")
+    def validate_at_least_one_reason(self):
+        if not (
+            self.retry_on_voicemail or self.retry_on_no_answer or self.retry_on_busy
+        ):
+            raise ValueError(
+                "At least one of retry_on_voicemail, retry_on_no_answer, "
+                "retry_on_busy must be true"
+            )
+        return self
+
+
+@router.post("/{campaign_id}/redial")
+async def redial_campaign(
+    campaign_id: int,
+    request: RedialCampaignRequest,
+    user: UserModel = Depends(get_user),
+) -> CampaignResponse:
+    """Create a new campaign that re-dials unique subscribers from a completed
+    campaign whose latest call resulted in voicemail, no-answer, or busy.
+
+    The new campaign is created in 'created' state with queued_runs pre-seeded
+    from the parent's original initial contexts. A campaign can be redialed at
+    most once.
+    """
+    parent = await db_client.get_campaign(campaign_id, user.selected_organization_id)
+    if not parent:
+        raise HTTPException(status_code=404, detail="Campaign not found")
+
+    if parent.state != "completed":
+        raise HTTPException(
+            status_code=400,
+            detail=f"Only completed campaigns can be redialed (current state: {parent.state})",
+        )
+
+    parent_meta = parent.orchestrator_metadata or {}
+    if parent_meta.get("redialed_campaign_id"):
+        raise HTTPException(
+            status_code=400,
+            detail="This campaign has already been redialed",
+        )
+
+    candidates = await db_client.get_redial_candidates(
+        campaign_id=parent.id,
+        include_voicemail=request.retry_on_voicemail,
+        include_no_answer=request.retry_on_no_answer,
+        include_busy=request.retry_on_busy,
+    )
+    if not candidates:
+        raise HTTPException(
+            status_code=400,
+            detail="No subscribers match the selected redial criteria",
+        )
+
+    queued_runs_data = [
+        {
+            "campaign_id": 0,  # replaced inside create_redial_campaign
+            "source_uuid": c["source_uuid"],
+            "context_variables": c["context_variables"],
+            "state": "queued",
+        }
+        for c in candidates
+    ]
+
+    retry_config = (
+        request.retry_config.model_dump()
+        if request.retry_config
+        else parent.retry_config
+    )
+    new_name = request.name or f"{parent.name} (Redial)"
+
+    try:
+        child = await db_client.create_redial_campaign(
+            parent_campaign=parent,
+            new_name=new_name,
+            retry_config=retry_config,
+            queued_runs_data=queued_runs_data,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+    workflow_name = await db_client.get_workflow_name(child.workflow_id, user.id)
+    executed, total = await _get_campaign_stats(child.id)
+    return _build_campaign_response(child, workflow_name or "Unknown", executed, total)
+
+
 @router.post("/{campaign_id}/resume")
 async def resume_campaign(
     campaign_id: int,
@@ -624,7 +766,10 @@ async def resume_campaign(
     campaign = await db_client.get_campaign(campaign_id, user.selected_organization_id)
     workflow_name = await db_client.get_workflow_name(campaign.workflow_id, user.id)
 
-    return _build_campaign_response(campaign, workflow_name or "Unknown")
+    executed, total = await _get_campaign_stats(campaign.id)
+    return _build_campaign_response(
+        campaign, workflow_name or "Unknown", executed, total
+    )
 
 
 @router.get("/{campaign_id}/progress")

api/routes/workflow.py

@@ -5,6 +5,7 @@ from datetime import datetime
 from typing import List, Literal, Optional
 
 from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi.responses import StreamingResponse
 from httpx import HTTPStatusError
 from loguru import logger
 from pydantic import BaseModel, Field, ValidationError
@@ -16,6 +17,7 @@ from api.db.workflow_template_client import WorkflowTemplateClient
 from api.enums import CallType, PostHogEvent, StorageBackend
 from api.schemas.workflow import WorkflowRunResponseSchema
 from api.services.auth.depends import get_user
+from api.services.campaign.report import generate_workflow_report_csv
 from api.services.configuration.check_validity import UserConfigurationValidator
 from api.services.configuration.masking import (
     mask_workflow_definition,
@@ -1002,6 +1004,37 @@ async def get_workflow_runs(
     )
 
 
+@router.get("/{workflow_id}/report")
+async def download_workflow_report(
+    workflow_id: int,
+    user: UserModel = Depends(get_user),
+    start_date: Optional[datetime] = Query(
+        None, description="Filter runs created on or after this datetime (ISO 8601)"
+    ),
+    end_date: Optional[datetime] = Query(
+        None, description="Filter runs created on or before this datetime (ISO 8601)"
+    ),
+) -> StreamingResponse:
+    """Download a CSV report of completed runs for a workflow."""
+    workflow = await db_client.get_workflow(
+        workflow_id, organization_id=user.selected_organization_id
+    )
+    if workflow is None:
+        raise HTTPException(
+            status_code=404, detail=f"Workflow with id {workflow_id} not found"
+        )
+
+    output, filename = await generate_workflow_report_csv(
+        workflow_id, start_date=start_date, end_date=end_date
+    )
+
+    return StreamingResponse(
+        output,
+        media_type="text/csv",
+        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+    )
+
+
 @router.get("/templates")
 async def get_workflow_templates() -> List[WorkflowTemplateResponse]:
     """

api/services/campaign/report.py

@@ -5,15 +5,12 @@ from typing import Any, List, Optional
 
 from api.constants import BACKEND_API_ENDPOINT
 from api.db import db_client
-from api.utils.transcript import generate_transcript_text
 
 
-def _transcript_from_logs(logs: dict | None) -> str:
-    """Extract transcript text from workflow run logs JSON."""
-    if not logs:
+def _artifact_url(token: str | None, artifact: str) -> str:
+    if not token:
         return ""
-    events = logs.get("realtime_feedback_events", [])
-    return generate_transcript_text(events).strip()
+    return f"{BACKEND_API_ENDPOINT}/api/v1/public/download/workflow/{token}/{artifact}"
 
 
 def _collect_extracted_variable_keys(runs: List[Any]) -> list[str]:
@@ -28,20 +25,11 @@ def _collect_extracted_variable_keys(runs: List[Any]) -> list[str]:
     return list(keys)
 
 
-async def generate_campaign_report_csv(
-    campaign_id: int,
-    start_date: Optional[datetime] = None,
-    end_date: Optional[datetime] = None,
-) -> tuple[io.StringIO, str]:
-    """Generate a CSV report for a campaign.
+def _build_run_report_csv(runs: List[Any]) -> io.StringIO:
+    """Build a CSV from completed workflow runs.
 
-    Returns a tuple of (csv_output, filename).
+    Shared between campaign-scoped and workflow-scoped reports.
     """
-    runs = await db_client.get_completed_runs_for_report(
-        campaign_id, start_date=start_date, end_date=end_date
-    )
-
-    # Collect dynamic extracted variable columns
     extracted_var_keys = _collect_extracted_variable_keys(runs)
 
     output = io.StringIO()
@@ -49,6 +37,9 @@ async def generate_campaign_report_csv(
 
     pre_headers = [
         "Run ID",
+        "Campaign ID",
+        "Agent ID",
+        "Agent Definition ID",
         "Created At",
         "Phone Number",
         "Call Disposition",
@@ -56,7 +47,7 @@ async def generate_campaign_report_csv(
     ]
     post_headers = [
         "Call Tags",
-        "Transcript",
+        "Transcript URL",
         "Recording URL",
     ]
     writer.writerow(pre_headers + extracted_var_keys + post_headers)
@@ -66,19 +57,15 @@ async def generate_campaign_report_csv(
         gathered = run.gathered_context or {}
         cost = run.cost_info or {}
 
-        recording_url = ""
-        if run.public_access_token:
-            recording_url = (
-                f"{BACKEND_API_ENDPOINT}/api/v1/public/download/workflow"
-                f"/{run.public_access_token}/recording"
-            )
-
         call_tags = gathered.get("call_tags", [])
         if isinstance(call_tags, list):
             call_tags = ", ".join(str(t) for t in call_tags)
 
         pre_values = [
             run.id,
+            run.campaign_id if run.campaign_id is not None else "",
+            run.workflow_id,
+            run.definition_id if run.definition_id is not None else "",
             run.created_at.isoformat() if run.created_at else "",
             initial.get("phone_number", ""),
             gathered.get("mapped_call_disposition", ""),
@@ -92,12 +79,35 @@ async def generate_campaign_report_csv(
 
         post_values = [
             call_tags,
-            _transcript_from_logs(run.logs),
-            recording_url,
+            _artifact_url(run.public_access_token, "transcript"),
+            _artifact_url(run.public_access_token, "recording"),
         ]
 
         writer.writerow(pre_values + extracted_values + post_values)
 
     output.seek(0)
-    filename = f"campaign_{campaign_id}_report.csv"
-    return output, filename
+    return output
+
+
+async def generate_campaign_report_csv(
+    campaign_id: int,
+    start_date: Optional[datetime] = None,
+    end_date: Optional[datetime] = None,
+) -> tuple[io.StringIO, str]:
+    """Generate a CSV report for a campaign."""
+    runs = await db_client.get_completed_runs_for_report(
+        campaign_id=campaign_id, start_date=start_date, end_date=end_date
+    )
+    return _build_run_report_csv(runs), f"campaign_{campaign_id}_report.csv"
+
+
+async def generate_workflow_report_csv(
+    workflow_id: int,
+    start_date: Optional[datetime] = None,
+    end_date: Optional[datetime] = None,
+) -> tuple[io.StringIO, str]:
+    """Generate a CSV report for all completed runs of a workflow."""
+    runs = await db_client.get_completed_runs_for_report(
+        workflow_id=workflow_id, start_date=start_date, end_date=end_date
+    )
+    return _build_run_report_csv(runs), f"workflow_{workflow_id}_report.csv"

api/services/campaign/runner.py

@@ -4,6 +4,9 @@ from typing import Any, Dict
 from loguru import logger
 
 from api.db import db_client
+from api.services.campaign.campaign_event_publisher import (
+    get_campaign_event_publisher,
+)
 from api.services.campaign.circuit_breaker import circuit_breaker
 from api.tasks.arq import enqueue_job
 from api.tasks.function_names import FunctionNames
@@ -24,6 +27,29 @@ class CampaignRunnerService:
                 f"Campaign must be in 'created' state to start, current state: {campaign.state}"
             )
 
+        # Redial campaigns have queued_runs pre-seeded from the parent campaign,
+        # so skip source sync and transition straight to 'running'.
+        is_redial = bool(
+            (campaign.orchestrator_metadata or {}).get("parent_campaign_id")
+        )
+        if is_redial:
+            now = datetime.now(UTC)
+            await db_client.update_campaign(
+                campaign_id=campaign_id,
+                state="running",
+                started_at=now,
+                source_last_synced_at=now,
+            )
+            publisher = await get_campaign_event_publisher()
+            await publisher.publish_sync_completed(
+                campaign_id=campaign_id,
+                total_rows=campaign.total_rows or 0,
+                source_type=campaign.source_type,
+                source_id=campaign.source_id,
+            )
+            logger.info(f"Redial campaign {campaign_id} started, source sync skipped")
+            return
+
         # Update campaign state to syncing
         await db_client.update_campaign(
             campaign_id=campaign_id,

api/services/filesystem/s3.py

@@ -64,14 +64,29 @@ class S3FileSystem(BaseFileSystem):
             ) as s3_client:
                 params = {"Bucket": self.bucket_name, "Key": file_path}
 
-                # Make transcripts viewable inline in the browser when requested
-                if force_inline and file_path.endswith(".txt"):
-                    params.update(
-                        {
-                            "ResponseContentType": "text/plain",
-                            "ResponseContentDisposition": "inline",
-                        }
-                    )
+                # Make artifacts viewable inline in the browser when requested
+                if force_inline:
+                    if file_path.endswith(".txt"):
+                        params.update(
+                            {
+                                "ResponseContentType": "text/plain",
+                                "ResponseContentDisposition": "inline",
+                            }
+                        )
+                    elif file_path.endswith(".wav"):
+                        params.update(
+                            {
+                                "ResponseContentType": "audio/wav",
+                                "ResponseContentDisposition": "inline",
+                            }
+                        )
+                    elif file_path.endswith(".mp3"):
+                        params.update(
+                            {
+                                "ResponseContentType": "audio/mpeg",
+                                "ResponseContentDisposition": "inline",
+                            }
+                        )
 
                 url = await s3_client.generate_presigned_url(
                     "get_object",

api/services/workflow/pipecat_engine.py

@@ -672,6 +672,13 @@ class PipecatEngine:
             self._gathered_context["call_disposition"] = reason
             self._gathered_context["mapped_call_disposition"] = mapped_disposition
 
+        effective_disposition = self._gathered_context.get("call_disposition", "")
+        if effective_disposition:
+            call_tags = self._gathered_context.get("call_tags", [])
+            if effective_disposition not in call_tags:
+                call_tags.append(effective_disposition)
+            self._gathered_context["call_tags"] = call_tags
+
         logger.debug(
             f"Finishing run with reason: {reason}, disposition: {mapped_disposition} queueing frame {frame_to_push}"
         )

api/services/workflow/pipecat_engine_custom_tools.py

@@ -370,8 +370,8 @@ class CustomToolManager:
                     logger.info(f"End call reason: {reason}")
                     self._engine._gathered_context["call_disposition"] = reason
                     call_tags = self._engine._gathered_context.get("call_tags", [])
-                    if reason not in call_tags:
-                        call_tags.extend([reason, "end_call_tool"])
+                    if "end_call_tool" not in call_tags:
+                        call_tags.append("end_call_tool")
                     self._engine._gathered_context["call_tags"] = call_tags
 
                 # Send result callback first