load cloud run history from presigned log urls

Author: charlesvien

apps/mobile/src/features/tasks/api.ts

@@ -719,9 +719,11 @@ export async function fetchSessionLogs(
         offset: String(options.offset ?? 0),
       });
 
+      // Big runs can take the server a long time per page (it re-reads the whole
+      // log chain each request), so this needs far more than the default budget.
       const response = await authedFetch(
         `${baseUrl}/api/projects/${projectId}/tasks/${taskId}/runs/${runId}/session_logs/?${params}`,
-        { signal: createTimeoutSignal(10_000) },
+        { signal: createTimeoutSignal(120_000) },
       );
 
       if (!response.ok) {

apps/mobile/src/features/tasks/lib/cloudTaskStream.ts

@@ -805,14 +805,18 @@ async function fetchTaskRunState(
 
 /**
  * Loads the historical log entries for the run, mirroring the desktop's
- * dual-source strategy:
- *  1. Try the paginated `session_logs/` API — the live source while a run
+ * strategy:
+ *  1. Prefer the presigned resume-chain `log_urls` (S3 NDJSON, oldest
+ *     first) when the server provides them. Downloading straight from S3
+ *     avoids the paginated API's per-page full-chain re-read, which times
+ *     out on runs with very large histories.
+ *  2. Try the paginated `session_logs/` API — the live source while a run
  *     is active. For older / archived runs this can come back empty even
  *     though the canonical log exists on S3.
- *  2. Fall back to the run's presigned `log_url` (S3 NDJSON), which is the
+ *  3. Fall back to the run's presigned `log_url` (S3 NDJSON), which is the
  *     canonical archive for completed runs.
  *
- * Returns `null` only when both sources fail outright (so the bootstrap can
+ * Returns `null` only when the sources fail outright (so the bootstrap can
  * surface a retryable error). An empty paginated result is treated as "no
  * data yet" and falls through to S3 — if S3 also has nothing we return the
  * empty array so the snapshot can still flip the session to `"connected"`.
@@ -821,6 +825,12 @@ async function fetchHistoricalEntries(
   watcher: WatcherState,
   run: TaskRun,
 ): Promise<StoredLogEntry[] | null> {
+  if (run.log_urls?.length) {
+    const chainEntries = await fetchChainLogEntries(watcher, run.log_urls);
+    if (watcher.stopped || watcher.failed) return null;
+    if (chainEntries) return chainEntries;
+  }
+
   const paginated = await fetchAllSessionLogs(watcher);
   if (watcher.stopped || watcher.failed) return null;
   if (paginated && paginated.length > 0) return paginated;
@@ -837,13 +847,29 @@ async function fetchHistoricalEntries(
   return paginated ?? null;
 }
 
+async function fetchChainLogEntries(
+  watcher: WatcherState,
+  logUrls: string[],
+): Promise<StoredLogEntry[] | null> {
+  const entries: StoredLogEntry[] = [];
+  for (const logUrl of logUrls) {
+    const chunk = await fetchS3LogEntries(watcher, logUrl);
+    if (watcher.stopped || watcher.failed) return null;
+    if (chunk === null) return null;
+    entries.push(...chunk);
+  }
+  return entries;
+}
+
 async function fetchS3LogEntries(
   watcher: WatcherState,
   logUrl: string,
 ): Promise<StoredLogEntry[] | null> {
   try {
+    // Chain logs of long-running tasks can be hundreds of MB; RN fetch buffers
+    // the whole body, so the budget covers the full download, not just TTFB.
     const response = await fetch(logUrl, {
-      signal: createTimeoutSignal(15_000),
+      signal: createTimeoutSignal(120_000),
     });
     if (response.status === 404) {
       // No archived log yet for this run — not an error, just no data.

apps/mobile/src/features/tasks/types.ts

@@ -64,6 +64,9 @@ export interface TaskRun {
   environment?: "local" | "cloud";
   status: TaskRunStatus;
   log_url: string;
+  /** Presigned S3 URLs for every log in the run's resume chain, oldest first.
+   *  Absent on old servers; empty when the server can't presign. */
+  log_urls?: string[];
   error_message: string | null;
   reasoning_effort?: string | null;
   output: Record<string, unknown> | null;

packages/core/src/cloud-task/cloud-task.test.ts

@@ -483,6 +483,203 @@ describe("CloudTaskService", () => {
     expect(messages()).toEqual(["before retry", "after retry"]);
   });
 
+  const consoleLogEntry = (message: string) => ({
+    type: "notification",
+    timestamp: "2026-01-01T00:00:01Z",
+    notification: {
+      jsonrpc: "2.0",
+      method: "_posthog/console",
+      params: { sessionId: "run-1", level: "info", message },
+    },
+  });
+
+  it("bootstraps history from presigned chain log urls without touching session_logs", async () => {
+    const updates: unknown[] = [];
+    service.on(CloudTaskEvent.Update, (payload) => updates.push(payload));
+
+    const sessionLogsCalls: string[] = [];
+    mockNetFetch.mockImplementation((input: string | Request) => {
+      const url = typeof input === "string" ? input : input.url;
+      if (url.includes("/session_logs/")) {
+        sessionLogsCalls.push(url);
+        return Promise.resolve(
+          createJsonResponse([], 200, { "X-Has-More": "false" }),
+        );
+      }
+      if (url.startsWith("https://storage.example/run-0.jsonl")) {
+        return Promise.resolve(
+          new Response(
+            `${JSON.stringify(consoleLogEntry("ancestor 1"))}\n${JSON.stringify(consoleLogEntry("ancestor 2"))}\n`,
+          ),
+        );
+      }
+      if (url.startsWith("https://storage.example/run-1.jsonl")) {
+        // No trailing newline: the final line of a log object is still a complete entry.
+        return Promise.resolve(
+          new Response(JSON.stringify(consoleLogEntry("current"))),
+        );
+      }
+      return Promise.resolve(
+        createJsonResponse({
+          id: "run-1",
+          status: "completed",
+          stage: null,
+          output: null,
+          error_message: null,
+          branch: "main",
+          updated_at: "2026-01-01T00:00:00Z",
+          log_urls: [
+            "https://storage.example/run-0.jsonl?sig=1",
+            "https://storage.example/run-1.jsonl?sig=2",
+          ],
+        }),
+      );
+    });
+
+    service.watch({
+      taskId: "task-1",
+      runId: "run-1",
+      apiHost: "https://app.example.com",
+      teamId: 2,
+    });
+
+    await waitFor(() =>
+      updates.some((u) => (u as { kind?: string }).kind === "snapshot"),
+    );
+
+    const snapshot = updates.find(
+      (u) => (u as { kind?: string }).kind === "snapshot",
+    ) as { newEntries: unknown[]; totalEntryCount: number };
+    expect(snapshot.newEntries).toEqual([
+      consoleLogEntry("ancestor 1"),
+      consoleLogEntry("ancestor 2"),
+      consoleLogEntry("current"),
+    ]);
+    expect(snapshot.totalEntryCount).toBe(3);
+    expect(sessionLogsCalls).toEqual([]);
+  });
+
+  it("falls back to the paginated API when a chain log download fails", async () => {
+    const updates: unknown[] = [];
+    service.on(CloudTaskEvent.Update, (payload) => updates.push(payload));
+
+    mockNetFetch.mockImplementation((input: string | Request) => {
+      const url = typeof input === "string" ? input : input.url;
+      if (url.includes("/session_logs/")) {
+        return Promise.resolve(
+          createJsonResponse([consoleLogEntry("from api")], 200, {
+            "X-Has-More": "false",
+          }),
+        );
+      }
+      if (url.startsWith("https://storage.example/")) {
+        return Promise.resolve(new Response("expired", { status: 403 }));
+      }
+      return Promise.resolve(
+        createJsonResponse({
+          id: "run-1",
+          status: "completed",
+          stage: null,
+          output: null,
+          error_message: null,
+          branch: "main",
+          updated_at: "2026-01-01T00:00:00Z",
+          log_urls: ["https://storage.example/run-1.jsonl?sig=1"],
+        }),
+      );
+    });
+
+    service.watch({
+      taskId: "task-1",
+      runId: "run-1",
+      apiHost: "https://app.example.com",
+      teamId: 2,
+    });
+
+    await waitFor(() =>
+      updates.some((u) => (u as { kind?: string }).kind === "snapshot"),
+    );
+
+    const snapshot = updates.find(
+      (u) => (u as { kind?: string }).kind === "snapshot",
+    ) as { newEntries: unknown[] };
+    expect(snapshot.newEntries).toEqual([consoleLogEntry("from api")]);
+  });
+
+  it("resumes paginated history from the last fetched page after a retry", async () => {
+    vi.useFakeTimers();
+    const updates: unknown[] = [];
+    service.on(CloudTaskEvent.Update, (payload) => updates.push(payload));
+
+    const sessionLogsOffsets: string[] = [];
+    let failPageFetches = true;
+    mockNetFetch.mockImplementation((input: string | Request) => {
+      const url = typeof input === "string" ? input : input.url;
+      if (url.includes("/session_logs/")) {
+        const offset = new URL(url).searchParams.get("offset") ?? "";
+        sessionLogsOffsets.push(offset);
+        if (offset === "0") {
+          return Promise.resolve(
+            createJsonResponse([consoleLogEntry("page one")], 200, {
+              "X-Has-More": "true",
+            }),
+          );
+        }
+        if (failPageFetches) {
+          return Promise.reject(new Error("socket hang up"));
+        }
+        return Promise.resolve(
+          createJsonResponse([consoleLogEntry("page two")], 200, {
+            "X-Has-More": "false",
+          }),
+        );
+      }
+      return Promise.resolve(
+        createJsonResponse({
+          id: "run-1",
+          status: "completed",
+          stage: null,
+          output: null,
+          error_message: null,
+          branch: "main",
+          updated_at: "2026-01-01T00:00:00Z",
+        }),
+      );
+    });
+
+    service.watch({
+      taskId: "task-1",
+      runId: "run-1",
+      apiHost: "https://app.example.com",
+      teamId: 2,
+    });
+
+    // Page 0 succeeds; the offset-1 page fails all retry attempts and the
+    // watcher surfaces a retryable error instead of a snapshot.
+    await waitFor(
+      () => updates.some((u) => (u as { kind?: string }).kind === "error"),
+      30_000,
+    );
+    expect(sessionLogsOffsets).toEqual(["0", "1", "1", "1"]);
+
+    failPageFetches = false;
+    await service.retry("task-1", "run-1");
+    await waitFor(
+      () => updates.some((u) => (u as { kind?: string }).kind === "snapshot"),
+      30_000,
+    );
+
+    // The retry resumed from offset 1 instead of refetching page 0.
+    expect(sessionLogsOffsets).toEqual(["0", "1", "1", "1", "1"]);
+    const snapshot = updates.find(
+      (u) => (u as { kind?: string }).kind === "snapshot",
+    ) as { newEntries: unknown[] };
+    expect(snapshot.newEntries).toEqual([
+      consoleLogEntry("page one"),
+      consoleLogEntry("page two"),
+    ]);
+  });
+
   it("reconnects with Last-Event-ID after a stream error", async () => {
     vi.useFakeTimers();