fix(core): harden archived task hydration

Generated-By: PostHog Code
Task-Id: c1bbe3cf-742b-4b24-bf96-d11a18b4cf22

Author: richardsolomou

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

@@ -34,6 +34,7 @@ const SSE_HEALTHY_CONNECTION_MS = 60_000;
 const EVENT_BATCH_FLUSH_MS = 16;
 const EVENT_BATCH_MAX_SIZE = 50;
 const SESSION_LOG_PAGE_LIMIT = 5_000;
+const ARCHIVED_LOG_FETCH_TIMEOUT_MS = 15_000;
 const MAX_HANDLED_RELAY_REQUEST_IDS = 1_000;
 const MCP_RELAY_METHODS_WITHOUT_APPROVAL = new Set([
   "initialize",
@@ -2162,16 +2163,24 @@ export class CloudTaskEngine extends TypedEventEmitter<CloudTaskEvents> {
     logUrl: string,
   ): Promise<StoredLogEntry[] | null> {
     try {
-      const response = await this.streamFetch(logUrl);
+      const response = await this.streamFetch(logUrl, {
+        signal: AbortSignal.timeout(ARCHIVED_LOG_FETCH_TIMEOUT_MS),
+      });
       if (!response.ok) return null;
       const content = await response.text();
       if (!content.trim()) return [];
       return content
         .trim()
         .split("\n")
-        .map((line) => JSON.parse(line) as StoredLogEntry);
+        .flatMap((line) => {
+          try {
+            return [JSON.parse(line) as StoredLogEntry];
+          } catch {
+            return [];
+          }
+        });
     } catch (error) {
-      this.log.warn("Cloud task archived logs fetch error", { logUrl, error });
+      this.log.warn("Cloud task archived logs fetch error", { error });
       return null;
     }
   }

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

@@ -2578,6 +2578,56 @@ describe("CloudTaskEngine", () => {
         status: "completed",
       }),
     );
+    expect(
+      mockNetFetch.mock.calls.find(
+        ([input]) => input === "https://logs.example.com/run-1.jsonl",
+      )?.[1]?.signal,
+    ).toBeInstanceOf(AbortSignal);
+  });
+
+  it("keeps valid archived entries around malformed lines", async () => {
+    const updates: unknown[] = [];
+    service.on(CloudTaskEvent.Update, (payload) => updates.push(payload));
+    const archivedEntry = {
+      type: "notification",
+      timestamp: "2026-01-01T00:00:00Z",
+    };
+
+    mockNetFetch.mockImplementation((input: string | Request) => {
+      const url = typeof input === "string" ? input : input.url;
+      if (url.includes("/session_logs/")) {
+        return Promise.resolve(
+          createJsonResponse([], 200, { "X-Has-More": "false" }),
+        );
+      }
+      if (url === "https://logs.example.com/run-1.jsonl") {
+        return Promise.resolve(
+          new Response(`invalid\n${JSON.stringify(archivedEntry)}\n`, {
+            status: 200,
+          }),
+        );
+      }
+      return Promise.resolve(
+        createJsonResponse({
+          id: "run-1",
+          status: "completed",
+          log_url: "https://logs.example.com/run-1.jsonl",
+          updated_at: "2026-01-01T00:00:00Z",
+        }),
+      );
+    });
+
+    service.watch({
+      taskId: "task-1",
+      runId: "run-1",
+      apiHost: "https://app.example.com",
+      teamId: 2,
+    });
+
+    await waitFor(() => updates.length === 1);
+    expect(updates[0]).toEqual(
+      expect.objectContaining({ newEntries: [archivedEntry] }),
+    );
   });
 
   const guardedFetchStatusExpectations = [