feat(engineering-analytics): add workflow run activity chart endpoint

products/engineering_analytics/backend/facade/api.py

@@ -34,6 +34,7 @@
     QuarantineRequestResult,
     WorkflowHealthItem,
     WorkflowJob,
+    WorkflowRunActivity,
     WorkflowRunDetail,
     WorkflowRunnerCost,
 )
@@ -138,6 +139,27 @@ def list_workflow_runs(
     )
 
 
+def get_workflow_run_activity(
+    *,
+    team: Team,
+    repo: str,
+    workflow_name: str,
+    date_from: str | None = None,
+    date_to: str | None = None,
+    branch: str | None = None,
+    source_id: str | None = None,
+    user_access_control: "UserAccessControl | None" = None,
+) -> WorkflowRunActivity:
+    return logic.build_workflow_run_activity(
+        curated=_authorized_source(team, source_id, user_access_control),
+        repo=repo,
+        workflow_name=workflow_name,
+        date_from=date_from,
+        date_to=date_to,
+        branch=branch,
+    )
+
+
 def get_workflow_runner_costs(
     *,
     team: Team,

products/engineering_analytics/backend/facade/contracts.py

@@ -216,6 +216,42 @@ class WorkflowRunDetail:
     pr_number: int
 
 
+@dataclass(frozen=True)
+class WorkflowRunActivityPoint:
+    """A single workflow run reduced to the fields the run-activity chart plots: start time, duration,
+    conclusion, branch, and attributed PR. Deliberately leaner than ``WorkflowRunDetail`` so the chart can
+    load far more runs across the full window (for the scatter, the in-flight band, and the focus-lens
+    brush) than the capped run-detail table, without the per-row wire cost of the full detail shape.
+    """
+
+    run_id: int
+    # Raw conclusion passthrough ('success' / 'failure' / 'timed_out' / ...), or None while still running.
+    conclusion: str | None
+    # Always set here (unlike the shared WorkflowRunDetail shape): the windowed query filters on
+    # run_started_at, so a run with no parseable start timestamp is excluded — it can't be placed on the
+    # chart's time axis anyway. Non-null keeps the contract honest for this chart-only endpoint.
+    run_started_at: datetime
+    # None until the run completes — duration is only computed for completed runs.
+    duration_seconds: int | None
+    head_branch: str
+    # Attributed pull request number, or 0 when unattributed.
+    pr_number: int
+
+
+@dataclass(frozen=True)
+class WorkflowRunActivity:
+    """The run-activity chart's data for one workflow over a window: compact per-run points plus an
+    explicit truncation signal. ``points`` is capped at ``limit`` (newest first); ``truncated`` is True
+    when more runs matched than the cap, so the chart can label itself as covering only the most recent
+    runs rather than the full window. Higher-capped than the run-detail table, so the chart still spans
+    multiple days on busy workflows where the smaller table cap would collapse to a sliver.
+    """
+
+    points: list[WorkflowRunActivityPoint]
+    truncated: bool
+    limit: int
+
+
 @dataclass(frozen=True)
 class WorkflowJob:
     """One job within a workflow run, for the run's expandable job breakdown. ``estimated_cost_usd``

products/engineering_analytics/backend/logic/__init__.py

@@ -22,6 +22,7 @@
     PullRequestList,
     WorkflowHealthItem,
     WorkflowJob,
+    WorkflowRunActivity,
     WorkflowRunDetail,
     WorkflowRunnerCost,
 )
@@ -39,6 +40,7 @@
 from products.engineering_analytics.backend.logic.queries.workflow_health import query_workflow_health
 from products.engineering_analytics.backend.logic.queries.workflow_jobs import query_workflow_jobs
 from products.engineering_analytics.backend.logic.queries.workflow_run import query_workflow_run
+from products.engineering_analytics.backend.logic.queries.workflow_run_activity import query_workflow_run_activity
 from products.engineering_analytics.backend.logic.queries.workflow_run_list import query_workflow_run_list
 from products.engineering_analytics.backend.logic.sources import list_github_sources
 
@@ -124,6 +126,31 @@ def build_workflow_run_list(
     )
 
 
+def build_workflow_run_activity(
+    *,
+    curated: CuratedGitHubSource,
+    repo: str | None,
+    workflow_name: str,
+    date_from: str | None = None,
+    date_to: str | None = None,
+    branch: str | None = None,
+) -> WorkflowRunActivity:
+    owner, name = _split_repo(repo)
+    if not (owner and name):
+        raise ValueError("repo must be in 'owner/name' format")
+    parsed_from = _parse_date(curated.team, date_from or _DEFAULT_WINDOW)
+    parsed_to = _parse_date(curated.team, date_to) if date_to else None
+    return query_workflow_run_activity(
+        curated=curated,
+        repo_owner=owner,
+        repo_name=name,
+        workflow_name=workflow_name,
+        date_from=parsed_from,
+        date_to=parsed_to,
+        branch=branch,
+    )
+
+
 def build_workflow_runner_costs(
     *,
     curated: CuratedGitHubSource,

products/engineering_analytics/backend/logic/queries/workflow_run_activity.py

@@ -0,0 +1,88 @@
+"""Curated query: compact per-run points for the run-activity chart.
+
+Lists one workflow's runs over a window as the minimal shape the chart plots (start / duration /
+conclusion / branch / attributed PR), newest first, at a higher cap than the run-detail table. Kept
+separate from ``workflow_run_list`` so the chart can span the full window — enough runs for the
+scatter, the in-flight band, and the focus-lens brush — while the detail table stays capped at its
+smaller list size. Fetches ``_LIMIT + 1`` rows so an overflow is reported as ``truncated`` rather than
+silently dropped.
+
+The ``run_started_at >= date_from`` filter also excludes runs whose start timestamp didn't parse
+(``NULL``), which is intended: a run with no start time can't be placed on the chart's time axis. That
+is why ``WorkflowRunActivityPoint.run_started_at`` is non-null while the shared run-detail shape is not.
+"""
+
+from datetime import datetime
+
+from posthog.hogql import ast
+
+from products.engineering_analytics.backend.facade.contracts import WorkflowRunActivity, WorkflowRunActivityPoint
+from products.engineering_analytics.backend.logic.queries._curated import CuratedGitHubSource
+
+# The chart plots a point per run and needs enough span to cover the window: an order of magnitude
+# above the run-detail table's cap (workflow_run_list._LIMIT = 200) so the scatter and focus-lens
+# brush still see multiple days on busy workflows where the smaller cap collapses to a sub-day slice.
+# Revisit this alongside that table cap if it changes materially, so the chart keeps spanning the window.
+_LIMIT = 2000
+
+_SELECT = f"""
+    SELECT
+        id, conclusion, run_started_at, duration_seconds, head_branch, pr_number
+    FROM __RUNS_SOURCE__ AS r
+    WHERE repo_owner = {{repo_owner}} AND repo_name = {{repo_name}} AND workflow_name = {{workflow_name}}
+        AND run_started_at >= {{date_from}} __DATE_TO__ __BRANCH__
+    ORDER BY run_started_at DESC, run_attempt DESC
+    LIMIT {_LIMIT + 1}
+"""
+
+
+def query_workflow_run_activity(
+    *,
+    curated: CuratedGitHubSource,
+    repo_owner: str,
+    repo_name: str,
+    workflow_name: str,
+    date_from: datetime,
+    date_to: datetime | None,
+    branch: str | None = None,
+) -> WorkflowRunActivity:
+    placeholders: dict[str, ast.Expr] = {
+        "repo_owner": ast.Constant(value=repo_owner),
+        "repo_name": ast.Constant(value=repo_name),
+        "workflow_name": ast.Constant(value=workflow_name),
+        "date_from": ast.Constant(value=date_from),
+    }
+    date_to_clause = ""
+    if date_to is not None:
+        date_to_clause = "AND run_started_at <= {date_to}"
+        placeholders["date_to"] = ast.Constant(value=date_to)
+    # An empty/whitespace branch is "no filter", not a literal match on '' — mirrors workflow_run_list.
+    branch = branch.strip() if branch else None
+    branch_clause = ""
+    if branch:
+        branch_clause = "AND head_branch = {branch}"
+        placeholders["branch"] = ast.Constant(value=branch)
+    response = curated.run(
+        _SELECT.replace("__RUNS_SOURCE__", curated.run_source())
+        .replace("__DATE_TO__", date_to_clause)
+        .replace("__BRANCH__", branch_clause),
+        query_type="engineering_analytics.workflow_run_activity",
+        placeholders=placeholders,
+    )
+    rows = response.results or []
+    truncated = len(rows) > _LIMIT
+    points = [_to_point(row) for row in rows[:_LIMIT]]
+    return WorkflowRunActivity(points=points, truncated=truncated, limit=_LIMIT)
+
+
+def _to_point(row: tuple) -> WorkflowRunActivityPoint:
+    run_id, conclusion, run_started_at, duration_seconds, head_branch, pr_number = row
+    return WorkflowRunActivityPoint(
+        run_id=int(run_id),
+        # Empty string means "no conclusion yet" (still running) — normalize to None for the contract.
+        conclusion=conclusion or None,
+        run_started_at=run_started_at,
+        duration_seconds=int(duration_seconds) if duration_seconds is not None else None,
+        head_branch=head_branch or "",
+        pr_number=int(pr_number) if pr_number is not None else 0,
+    )

products/engineering_analytics/backend/presentation/serializers.py

@@ -33,6 +33,8 @@
     WorkflowHealthBucket,
     WorkflowHealthItem,
     WorkflowJob,
+    WorkflowRunActivity,
+    WorkflowRunActivityPoint,
     WorkflowRunDetail,
     WorkflowRunnerCost,
 )
@@ -207,6 +209,45 @@ class Meta:
         }
 
 
+class WorkflowRunActivityPointSerializer(DataclassSerializer):
+    class Meta:
+        dataclass = WorkflowRunActivityPoint
+        extra_kwargs = {
+            "run_id": {"help_text": "GitHub Actions run id."},
+            "conclusion": {
+                "help_text": "Run conclusion ('success', 'failure', 'timed_out', 'cancelled', 'skipped', ...), "
+                "or null while still in progress.",
+                "allow_null": True,
+            },
+            "run_started_at": {
+                "help_text": "When the run started. Never null on this endpoint: runs without a parseable "
+                "start timestamp are excluded from the window (they can't be plotted on the chart's time axis).",
+            },
+            "duration_seconds": {
+                "help_text": "Wall-clock duration in seconds; null until the run completes.",
+                "allow_null": True,
+            },
+            "head_branch": {"help_text": "Git branch the run was triggered on, or '' when unknown."},
+            "pr_number": {"help_text": "Attributed pull request number, or 0 when unattributed."},
+        }
+
+
+class WorkflowRunActivitySerializer(DataclassSerializer):
+    points = WorkflowRunActivityPointSerializer(
+        many=True, help_text="Per-run chart points, newest first, capped at `limit`."
+    )
+
+    class Meta:
+        dataclass = WorkflowRunActivity
+        extra_kwargs = {
+            "truncated": {
+                "help_text": "True when more runs matched than the cap; `points` is the newest `limit` runs, so the "
+                "chart covers only the most recent activity, not the full window.",
+            },
+            "limit": {"help_text": "Maximum number of run points returned in `points`."},
+        }
+
+
 class WorkflowJobSerializer(DataclassSerializer):
     class Meta:
         dataclass = WorkflowJob

products/engineering_analytics/backend/presentation/views.py

@@ -39,6 +39,7 @@
     QuarantineRequestSerializer,
     WorkflowHealthItemSerializer,
     WorkflowJobSerializer,
+    WorkflowRunActivitySerializer,
     WorkflowRunDetailSerializer,
     WorkflowRunnerCostSerializer,
 )
@@ -133,6 +134,7 @@ class EngineeringAnalyticsViewSet(TeamAndOrgViewSetMixin, viewsets.GenericViewSe
         "pr_cost",
         "workflow_run",
         "workflow_runs",
+        "workflow_run_activity",
         "workflow_runner_costs",
         "workflow_jobs",
     ]
@@ -545,6 +547,61 @@ def workflow_runs(self, request: Request, **kwargs) -> Response:
             return _bad_request(exc, fallback="Invalid date, repo, or source_id")
         return Response(WorkflowRunDetailSerializer(instance=runs, many=True).data)
 
+    @extend_schema(
+        operation_id="engineering_analytics_workflow_run_activity",
+        parameters=[
+            OpenApiParameter(
+                name="workflow_name",
+                type=OpenApiTypes.STR,
+                location=OpenApiParameter.QUERY,
+                required=True,
+                description="Workflow name to load run activity for.",
+            ),
+            OpenApiParameter(
+                name="repo",
+                type=OpenApiTypes.STR,
+                location=OpenApiParameter.QUERY,
+                required=True,
+                description="'owner/name' repository the workflow belongs to.",
+            ),
+            _DATE_FROM,
+            _DATE_TO,
+            _BRANCH,
+            _SOURCE_ID,
+        ],
+        responses={
+            200: WorkflowRunActivitySerializer,
+            400: OpenApiResponse(description="Missing workflow_name/repo, or invalid date or source_id."),
+        },
+        description=(
+            "Compact per-run points for a single workflow over a window (date_from default -30d), newest first, for "
+            "the run-activity chart: each run's start time, duration, conclusion, branch, and attributed PR. "
+            "Optionally scope to a single git branch via `branch`, matching workflow_runs. Leaner and higher-capped "
+            "than workflow_runs so the chart spans the full window even on busy workflows; `truncated` is true when "
+            "the cap is hit, so the chart covers only the most recent runs."
+        ),
+    )
+    @action(detail=False, methods=["get"], pagination_class=None)
+    def workflow_run_activity(self, request: Request, **kwargs) -> Response:
+        workflow_name = request.query_params.get("workflow_name")
+        repo = request.query_params.get("repo")
+        if not workflow_name or not repo:
+            return Response({"detail": "workflow_name and repo are required"}, status=status.HTTP_400_BAD_REQUEST)
+        try:
+            result = api.get_workflow_run_activity(
+                team=self.team,
+                repo=repo,
+                workflow_name=workflow_name,
+                date_from=request.query_params.get("date_from") or None,
+                date_to=request.query_params.get("date_to") or None,
+                branch=request.query_params.get("branch") or None,
+                source_id=request.query_params.get("source_id") or None,
+                user_access_control=self.user_access_control,
+            )
+        except ValueError as exc:
+            return _bad_request(exc, fallback="Invalid date, repo, or source_id")
+        return Response(WorkflowRunActivitySerializer(instance=result).data)
+
     @extend_schema(
         operation_id="engineering_analytics_workflow_runner_costs",
         parameters=[