retry stream target resolution on transient errors

Author: charlesvien

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

@@ -1017,6 +1017,78 @@ describe("CloudTaskService", () => {
     expect(mockStreamTokenFetch.mock.calls.length).toBe(1);
   });
 
+  it("a transient stream_token failure retries resolution instead of pinning to Django", async () => {
+    vi.useFakeTimers();
+
+    // The endpoint is momentarily down (503): unlike a 404, this must not cache a Django fallback.
+    // The next reconnect re-resolves and the watch upgrades to the durable proxy leg.
+    mockStreamTokenFetch
+      .mockImplementationOnce(() =>
+        Promise.resolve(createJsonResponse({ detail: "unavailable" }, 503)),
+      )
+      .mockImplementation(() =>
+        Promise.resolve(
+          createJsonResponse({
+            token: "fresh-token",
+            stream_base_url: "https://proxy.example",
+          }),
+        ),
+      );
+
+    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" }),
+        );
+      }
+      return Promise.resolve(
+        createJsonResponse({
+          id: "run-1",
+          status: "in_progress",
+          stage: null,
+          output: null,
+          error_message: null,
+          branch: "main",
+          updated_at: "2026-01-01T00:00:00Z",
+        }),
+      );
+    });
+
+    // Django leg (transient round) just EOFs to force a reconnect; once the proxy resolves it
+    // emits stream-end so the watch completes, proving durable streaming engaged after the retry.
+    const usedProxyLeg = (input: string | Request): boolean => {
+      const url = typeof input === "string" ? input : input.url;
+      return url.includes("proxy.example");
+    };
+    mockStreamFetch.mockImplementation((input: string | Request) =>
+      Promise.resolve(
+        usedProxyLeg(input)
+          ? createSseResponse("event: stream-end\ndata: {}\n\n")
+          : createSseResponse(""),
+      ),
+    );
+
+    service.watch({
+      taskId: "task-1",
+      runId: "run-1",
+      apiHost: "https://app.example.com",
+      teamId: 2,
+    });
+
+    const hasWatcher = (): boolean =>
+      (service as unknown as { watchers: Map<string, unknown> }).watchers.has(
+        "task-1:run-1",
+      );
+    await waitFor(() => !hasWatcher(), 10_000);
+
+    // The 503 was not cached: resolution retried and the stream switched to the durable proxy leg.
+    expect(mockStreamTokenFetch.mock.calls.length).toBeGreaterThanOrEqual(2);
+    expect(
+      mockStreamFetch.mock.calls.some(([input]) => usedProxyLeg(input)),
+    ).toBe(true);
+  });
+
   it("proxy 401 re-resolves the read target and resumes with a fresh token", async () => {
     vi.useFakeTimers();
 

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

@@ -263,6 +263,12 @@ function shouldFailWatcherForFetchStatus(status: number): boolean {
   return status === 401 || status === 403 || status === 404;
 }
 
+// 5xx and 429 are momentary: the stream-token endpoint exists but is briefly unavailable, so the
+// target stays unresolved and the next reconnect retries instead of caching a Django fallback.
+function isTransientStreamTargetStatus(status: number): boolean {
+  return status >= 500 || status === 429;
+}
+
 @injectable()
 export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
   private watchers = new Map<string, WatcherState>();
@@ -1573,9 +1579,19 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
         method: "GET",
       });
       if (!response.ok) {
-        // Refused, or an old server without the endpoint: read from Django with status polling.
         watcher.streamBaseUrl = null;
         watcher.streamReadToken = null;
+        if (isTransientStreamTargetStatus(response.status)) {
+          // Transient: read from Django this round but leave the target unresolved so the next
+          // reconnect retries durable resolution instead of pinning the run to status polling.
+          this.log.warn("Cloud task stream target temporarily unavailable", {
+            taskId: watcher.taskId,
+            runId: watcher.runId,
+            status: response.status,
+          });
+          return;
+        }
+        // Refused, or an old server without the endpoint: read from Django with status polling.
         watcher.durableStreamEnabled = false;
         watcher.streamTargetResolved = true;
         this.log.info("Cloud task stream reading from API host", {