feat(tasks): summarize endpoint

posthog/api/mixins.py

@@ -92,7 +92,7 @@ def my_action(self, request: ValidatedRequest, **kwargs):
     """
 
     def decorator(view_func: Callable) -> Callable:
-        parameters = []
+        parameters: list = list(extend_schema_kwargs.pop("parameters", None) or [])
         if query_serializer is not None:
             parameters.append(query_serializer)
 

posthog/settings/web.py

@@ -417,6 +417,8 @@ def static_varies_origin(headers, path, url):
         "LLMProviderEnum": "products.llm_analytics.backend.models.provider_keys.LLMProvider",
         "HogFlowStatusEnum": "posthog.models.hog_flow.hog_flow.HogFlow.State",
         "MCPAuthTypeEnum": "products.mcp_store.backend.models.AUTH_TYPE_CHOICES",
+        "TaskRunStatusEnum": "products.tasks.backend.models.TaskRun.Status",
+        "TaskRunEnvironmentEnum": "products.tasks.backend.models.TaskRun.Environment",
         # --- Inline value lists (type-hint enums, no x-spec-enum-id) ---
         "PropertyGroupOperator": ["AND", "OR"],
         "PropertyFilterTypeEnum": [

products/tasks/backend/api.py

@@ -12,13 +12,15 @@
 from django.core.cache import cache
 from django.db import models, transaction
 from django.db.models import F, OuterRef, Q, Subquery
+from django.db.models.functions import JSONObject
 from django.http import HttpResponse, JsonResponse, StreamingHttpResponse
 from django.utils import timezone
 
 import requests as http_requests
 import jsonschema
 import posthoganalytics
-from drf_spectacular.utils import OpenApiResponse, extend_schema
+from drf_spectacular.types import OpenApiTypes
+from drf_spectacular.utils import OpenApiParameter, OpenApiResponse, extend_schema
 from rest_framework import status, viewsets
 from rest_framework.authentication import SessionAuthentication
 from rest_framework.decorators import action
@@ -85,6 +87,8 @@
     TaskStagedArtifactsFinalizeUploadResponseSerializer,
     TaskStagedArtifactsPrepareUploadRequestSerializer,
     TaskStagedArtifactsPrepareUploadResponseSerializer,
+    TaskSummariesRequestSerializer,
+    TaskSummarySerializer,
     build_task_run_artifact_size_error,
     get_task_run_artifact_max_size_bytes,
 )
@@ -274,6 +278,62 @@ def repositories(self, request, **kwargs):
         serializer = TaskRepositoriesResponseSerializer({"repositories": list(repositories)})
         return Response(serializer.data)
 
+    @validated_request(
+        request_serializer=TaskSummariesRequestSerializer,
+        responses={
+            200: OpenApiResponse(
+                response=TaskSummarySerializer(many=True),
+                description="Summary fields for the requested tasks",
+            ),
+        },
+        summary="Fetch task summaries by ID",
+        description=(
+            "Returns summary for the requested tasks: `id`, `title`, `repository`, `created_at`, "
+            "`updated_at`, and the latest run's `status` and `environment`."
+        ),
+        parameters=[
+            OpenApiParameter(
+                name="limit",
+                type=OpenApiTypes.INT,
+                location=OpenApiParameter.QUERY,
+                required=False,
+                description="Page size for the paginated response.",
+            ),
+            OpenApiParameter(
+                name="offset",
+                type=OpenApiTypes.INT,
+                location=OpenApiParameter.QUERY,
+                required=False,
+                description="Offset into the result set for pagination.",
+            ),
+        ],
+    )
+    @action(
+        detail=False,
+        methods=["post"],
+        url_path="summaries",
+        required_scopes=["task:read"],
+        filter_backends=[],
+    )
+    def summaries(self, request, **kwargs):
+        ids = request.validated_data["ids"]
+        latest_run = (
+            TaskRun.objects.filter(task=OuterRef("pk"))
+            .order_by("-created_at", "-id")
+            .annotate(_data=JSONObject(status="status", environment="environment"))
+        )
+        tasks = (
+            Task.objects.filter(team=self.team, deleted=False, id__in=ids)
+            .annotate(_latest_run=Subquery(latest_run.values("_data")[:1]))
+            .order_by("-created_at", "id")
+        )
+        page = self.paginate_queryset(tasks)
+        if page is not None:
+            serializer = TaskSummarySerializer(page, many=True)
+            return self.get_paginated_response(serializer.data)
+        serializer = TaskSummarySerializer(tasks, many=True)
+        return Response(serializer.data)
+
     @validated_request(
         query_serializer=RepositoryReadinessQuerySerializer,
         responses={

products/tasks/backend/serializers.py

@@ -790,6 +790,39 @@ class TaskRunArtifactPresignResponseSerializer(serializers.Serializer):
     expires_in = serializers.IntegerField(help_text="URL expiry in seconds")
 
 
+TASK_SUMMARIES_MAX_IDS = 5000
+
+
+class TaskSummariesRequestSerializer(serializers.Serializer):
+    ids = serializers.ListField(
+        child=serializers.UUIDField(),
+        allow_empty=False,
+        max_length=TASK_SUMMARIES_MAX_IDS,
+        help_text=(
+            f"Task IDs to fetch summaries for (max {TASK_SUMMARIES_MAX_IDS}). Response is paginated; "
+            f"follow the `next` cursor to retrieve all results."
+        ),
+    )
+
+
+class TaskRunSummarySerializer(serializers.Serializer):
+    status = serializers.ChoiceField(choices=TaskRun.Status.choices, allow_null=True)
+    environment = serializers.ChoiceField(choices=TaskRun.Environment.choices, allow_null=True)
+
+
+class TaskSummarySerializer(serializers.ModelSerializer):
+    latest_run = serializers.SerializerMethodField()
+
+    class Meta:
+        model = Task
+        fields = ["id", "title", "repository", "created_at", "updated_at", "latest_run"]
+        read_only_fields = fields
+
+    @extend_schema_field(TaskRunSummarySerializer(allow_null=True))
+    def get_latest_run(self, obj):
+        return getattr(obj, "_latest_run", None)
+
+
 class TaskListQuerySerializer(serializers.Serializer):
     """Query parameters for listing tasks"""
 

products/tasks/backend/tests/test_api.py

@@ -2048,6 +2048,108 @@ def test_internal_field_defaults_to_false_on_create(self):
         self.assertFalse(response.json()["internal"])
 
 
+class TestTaskSummariesAPI(BaseTaskAPITest):
+    SUMMARIES_URL = "/api/projects/@current/tasks/summaries/"
+    SUMMARY_FIELDS = {"id", "title", "repository", "created_at", "updated_at", "latest_run"}
+
+    def post_summaries(self, ids):
+        return self.client.post(self.SUMMARIES_URL, {"ids": ids}, format="json")
+
+    def assertReturnsTaskIds(self, response, expected_ids):
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(
+            sorted(t["id"] for t in response.json()["results"]),
+            sorted(str(i) for i in expected_ids),
+        )
+
+    def test_summaries_returns_only_requested_ids_scoped_to_team(self):
+        task_a = self.create_task("Task A")
+        task_b = self.create_task("Task B")
+        self.create_task("Task C")
+        other_team = Team.objects.create(organization=self.organization, name="Other Team")
+        other_task = Task.objects.create(
+            team=other_team, title="Other", description="", origin_product=Task.OriginProduct.USER_CREATED
+        )
+
+        response = self.post_summaries([str(task_a.id), str(task_b.id), str(other_task.id), str(uuid.uuid4())])
+
+        self.assertReturnsTaskIds(response, [task_a.id, task_b.id])
+
+    def test_summaries_returns_internal_tasks_when_requested_by_id(self):
+        internal_task = Task.objects.create(
+            team=self.team,
+            title="Internal",
+            description="",
+            origin_product=Task.OriginProduct.USER_CREATED,
+            internal=True,
+        )
+
+        response = self.post_summaries([str(internal_task.id)])
+
+        self.assertReturnsTaskIds(response, [internal_task.id])
+
+    @parameterized.expand(
+        [
+            ("with_run", True),
+            ("no_runs", False),
+        ]
+    )
+    def test_summaries_response_shape(self, _name, with_run):
+        task = self.create_task("Task")
+        run = (
+            TaskRun.objects.create(
+                team=self.team,
+                task=task,
+                status=TaskRun.Status.IN_PROGRESS,
+                environment=TaskRun.Environment.LOCAL,
+            )
+            if with_run
+            else None
+        )
+
+        response = self.post_summaries([str(task.id)])
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        [payload] = response.json()["results"]
+        self.assertEqual(set(payload.keys()), self.SUMMARY_FIELDS)
+        expected_run = {"status": run.status, "environment": run.environment} if run else None
+        self.assertEqual(payload["latest_run"], expected_run)
+
+    def test_summaries_paginates_large_id_sets(self):
+        tasks = [self.create_task(f"Task {i}") for i in range(3)]
+        ids = [str(t.id) for t in tasks]
+
+        response = self.client.post(f"{self.SUMMARIES_URL}?limit=2", {"ids": ids}, format="json")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        payload = response.json()
+        self.assertEqual(payload["count"], 3)
+        self.assertEqual(len(payload["results"]), 2)
+        self.assertIsNotNone(payload["next"])
+
+        # Follow `next` cursor; combined results should equal the full set without duplicates.
+        next_url = payload["next"]
+        next_path = next_url[next_url.index("/api/") :]
+        response2 = self.client.post(next_path, {"ids": ids}, format="json")
+        self.assertEqual(response2.status_code, status.HTTP_200_OK)
+        payload2 = response2.json()
+        self.assertEqual(len(payload2["results"]), 1)
+        self.assertIsNone(payload2["next"])
+
+        seen_ids = [r["id"] for r in payload["results"]] + [r["id"] for r in payload2["results"]]
+        self.assertEqual(sorted(seen_ids), sorted(ids))
+        self.assertEqual(len(set(seen_ids)), 3)
+
+    @parameterized.expand(
+        [
+            ("empty", lambda: []),
+            ("invalid_uuid", lambda: ["not-a-uuid"]),
+        ]
+    )
+    def test_summaries_rejects_invalid_payload(self, _name, ids_factory):
+        response = self.post_summaries(ids_factory())
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+
 class TestTaskAutomationAPI(BaseTaskAPITest):
     @patch("products.tasks.backend.api.sync_automation_schedule")
     def test_create_automation(self, mock_sync_schedule):