Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion apps/mobile/src/features/tasks/api.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
36 changes: 31 additions & 5 deletions apps/mobile/src/features/tasks/lib/cloudTaskStream.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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"`.
Expand All @@ -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);
Comment thread
charlesvien marked this conversation as resolved.
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;
Expand All @@ -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.
Expand Down
3 changes: 3 additions & 0 deletions apps/mobile/src/features/tasks/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
197 changes: 197 additions & 0 deletions packages/core/src/cloud-task/cloud-task.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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();

Expand Down
Loading
Loading