diff --git a/products/tasks/backend/facade/api.py b/products/tasks/backend/facade/api.py index ab3a86bd9776..4d3bc1393cd0 100644 --- a/products/tasks/backend/facade/api.py +++ b/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( diff --git a/products/tasks/backend/facade/contracts.py b/products/tasks/backend/facade/contracts.py index e7875b6dfc77..65350387ae16 100644 --- a/products/tasks/backend/facade/contracts.py +++ b/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 diff --git a/products/tasks/backend/presentation/serializers.py b/products/tasks/backend/presentation/serializers.py index 16e73a92884a..9d754357e270 100644 --- a/products/tasks/backend/presentation/serializers.py +++ b/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", diff --git a/products/tasks/backend/tests/test_api.py b/products/tasks/backend/tests/test_api.py index 11da5408f6a1..608d33a82ba3 100644 --- a/products/tasks/backend/tests/test_api.py +++ b/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") diff --git a/products/tasks/frontend/generated/api.schemas.ts b/products/tasks/frontend/generated/api.schemas.ts index 576814191853..4e278410979f 100644 --- a/products/tasks/frontend/generated/api.schemas.ts +++ b/products/tasks/frontend/generated/api.schemas.ts @@ -1450,6 +1450,8 @@ export interface TaskRunDetailDTOApi { * @nullable */ log_url?: string | null + /** 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. */ + log_urls?: string[] /** @nullable */ error_message: string | null /** @nullable */ diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts index 1a31db481ed2..193d8f1314c9 100644 --- a/services/mcp/src/api/generated.ts +++ b/services/mcp/src/api/generated.ts @@ -39900,6 +39900,8 @@ export namespace Schemas { * @nullable */ log_url?: string | null; + /** 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. */ + log_urls?: string[]; /** @nullable */ error_message: string | null; /** @nullable */