feat(tasks): expose resume-chain presigned log urls on run detail

products/tasks/backend/facade/api.py

@@ -338,12 +338,42 @@ def _task_run_log_url(run: TaskRun) -> str | None:
     return presigned_url
 
 
-def _task_run_detail_to_dto(run: TaskRun) -> contracts.TaskRunDetailDTO:
+def _task_run_chain_log_urls(run: TaskRun) -> list[str]:
+    """Presigned S3 URLs for every log in the run's resume chain, oldest first, cached.
+
+    Lets clients bootstrap a run's full session history straight from object storage
+    instead of paging it through ``session_logs`` (which re-reads and re-parses the whole
+    chain per page). A run's ancestor chain is fixed at creation, so the list is safe to
+    cache; empty when presigning is unavailable so callers can fall back to the API.
+    """
+    from posthog.storage import object_storage  # noqa: PLC0415 — keep storage deps off the api import path
+
+    from products.tasks.backend.redis import get_tasks_cache  # noqa: PLC0415 — keep redis off the api import path
+
+    cache_key = f"task_run_chain_log_urls:{run.id}"
+    cached_urls = get_tasks_cache().get(cache_key)
+    if cached_urls:
+        return cached_urls
+
+    urls: list[str] = []
+    for ancestor in run.get_resume_chain():
+        presigned_url = object_storage.get_presigned_url(ancestor.log_url, expiration=3600)
+        if presigned_url is None:
+            return []
+        urls.append(presigned_url)
+    if urls:
+        get_tasks_cache().set(cache_key, urls, timeout=_TASK_RUN_LOG_URL_CACHE_TTL)
+    return urls
+
+
+def _task_run_detail_to_dto(run: TaskRun, *, include_log_urls: bool = False) -> contracts.TaskRunDetailDTO:
     """Map a ``TaskRun`` to its HTTP detail DTO.
 
     Reproduces the SMF-derived fields ``TaskRunDetailSerializer`` computed: ``log_url`` does
     presigned-URL I/O (with caching), and ``runtime_adapter`` / ``provider`` / ``model`` /
-    ``reasoning_effort`` are parsed off the run ``state``.
+    ``reasoning_effort`` are parsed off the run ``state``. ``log_urls`` (the resume-chain
+    presigns) needs a chain walk per run, so only the run-detail retrieve path — the one
+    clients bootstrap history from — opts in; every other caller gets an empty list.
     """
     from products.tasks.backend.temporal.process_task.utils import (  # noqa: PLC0415 — keep temporalio off the api import path
         parse_run_state,
@@ -362,6 +392,7 @@ def _task_run_detail_to_dto(run: TaskRun) -> contracts.TaskRunDetailDTO:
         model=state.model,
         reasoning_effort=state.reasoning_effort.value if state.reasoning_effort is not None else None,
         log_url=_task_run_log_url(run),
+        log_urls=_task_run_chain_log_urls(run) if include_log_urls else [],
         error_message=run.error_message,
         output=run.output,
         state=run.state or {},
@@ -1669,7 +1700,7 @@ def list_task_runs(task_id: str | UUID, team_id: int) -> list[contracts.TaskRunD
 def get_task_run_detail(run_id: str | UUID, task_id: str | UUID, team_id: int) -> contracts.TaskRunDetailDTO | None:
     """A single run as a detail DTO, scoped to its task + team."""
     run = _get_visible_run(run_id, task_id, team_id)
-    return _task_run_detail_to_dto(run) if run is not None else None
+    return _task_run_detail_to_dto(run, include_log_urls=True) if run is not None else None
 
 
 def get_task_run_stream_info(

products/tasks/backend/facade/contracts.py

@@ -383,10 +383,12 @@ class TaskRunDetailDTO:
     Mirrors exactly the fields ``TaskRunDetailSerializer`` emits, in order. ``task`` is the
     parent task id (rendered as a string, matching the original ``PrimaryKeyRelatedField``).
     The SMF-derived fields are computed in the facade mapper ``_task_run_detail_to_dto``:
-    ``log_url`` is a presigned S3 URL (cached); ``runtime_adapter`` / ``provider`` / ``model`` /
-    ``reasoning_effort`` are parsed off the run ``state``. ``artifacts`` carries the run's
-    artifact manifest entries verbatim. Reused by the run-detail responses and nested as
-    ``latest_run`` by the task detail response.
+    ``log_url`` is a presigned S3 URL (cached); ``log_urls`` carries presigned URLs for every
+    log in the run's resume chain, oldest first — populated only by the run-detail retrieve
+    path (empty for other consumers and when presigning is unavailable); ``runtime_adapter`` /
+    ``provider`` / ``model`` / ``reasoning_effort`` are parsed off the run ``state``.
+    ``artifacts`` carries the run's artifact manifest entries verbatim. Reused by the
+    run-detail responses and nested as ``latest_run`` by the task detail response.
     """
 
     id: UUID
@@ -404,6 +406,7 @@ class TaskRunDetailDTO:
     output: dict | None
     state: dict
     artifacts: list = Field(default_factory=list)
+    log_urls: list[str] = Field(default_factory=list)
     created_at: datetime | None = None
     updated_at: datetime | None = None
     completed_at: datetime | None = None

products/tasks/backend/presentation/serializers.py

@@ -262,6 +262,16 @@ class TaskRunDetailSerializer(DataclassSerializer):
     log_url = serializers.URLField(
         allow_null=True, required=False, help_text="Presigned S3 URL for log access (valid for 1 hour)."
     )
+    log_urls = serializers.ListField(
+        child=serializers.URLField(help_text="Presigned S3 URL for one log in the resume chain."),
+        required=False,
+        help_text=(
+            "Presigned S3 URLs for every JSONL log in the run's resume chain, oldest first "
+            "(each valid for 1 hour). Concatenated they form the run's full session history. "
+            "Only populated on the run detail endpoint; empty elsewhere or when presigning "
+            "is unavailable — fall back to the session_logs endpoint."
+        ),
+    )
     artifacts = TaskRunArtifactResponseSerializer(many=True, read_only=True)
     runtime_adapter = serializers.ChoiceField(
         choices=[adapter.value for adapter in RuntimeAdapter],
@@ -299,6 +309,7 @@ class Meta:
             "model",
             "reasoning_effort",
             "log_url",
+            "log_urls",
             "error_message",
             "output",
             "state",

products/tasks/backend/tests/test_api.py

@@ -4243,6 +4243,47 @@ def test_retrieve_specific_run(self):
         self.assertIn("log_url", data)
         self.assertIsNotNone(data["log_url"])
         self.assertTrue(data["log_url"].startswith("http"))
+        self.assertEqual(len(data["log_urls"]), 1)
+        self.assertIn(f"run_{run.id}.jsonl", data["log_urls"][0])
+
+    def test_retrieve_run_log_urls_walk_resume_chain_oldest_first(self):
+        task = self.create_task()
+        ancestor = TaskRun.objects.create(task=task, team=self.team, status=TaskRun.Status.COMPLETED)
+        resumed = TaskRun.objects.create(
+            task=task,
+            team=self.team,
+            status=TaskRun.Status.IN_PROGRESS,
+            state={"resume_from_run_id": str(ancestor.id)},
+        )
+
+        response = self.client.get(f"/api/projects/@current/tasks/{task.id}/runs/{resumed.id}/")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        log_urls = response.json()["log_urls"]
+        self.assertEqual(len(log_urls), 2)
+        self.assertIn(f"run_{ancestor.id}.jsonl", log_urls[0])
+        self.assertIn(f"run_{resumed.id}.jsonl", log_urls[1])
+
+    @patch.object(object_storage, "get_presigned_url", return_value=None)
+    def test_retrieve_run_log_urls_empty_when_presigning_unavailable(self, _mock_presign):
+        task = self.create_task()
+        run = TaskRun.objects.create(task=task, team=self.team, status=TaskRun.Status.IN_PROGRESS)
+
+        response = self.client.get(f"/api/projects/@current/tasks/{task.id}/runs/{run.id}/")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        data = response.json()
+        self.assertIsNone(data["log_url"])
+        self.assertEqual(data["log_urls"], [])
+
+    def test_list_runs_omits_chain_log_urls(self):
+        task = self.create_task()
+        TaskRun.objects.create(task=task, team=self.team, status=TaskRun.Status.IN_PROGRESS)
+
+        response = self.client.get(f"/api/projects/@current/tasks/{task.id}/runs/")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        self.assertEqual([run["log_urls"] for run in response.json()["results"]], [[]])
 
     def test_list_runs_only_returns_task_runs(self):
         task1 = self.create_task("Task 1")