fix: count only real stream updates and retry the job-id read

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

chat-sdk/src/chat.ts

@@ -682,22 +682,29 @@ class ChatImpl implements Chat {
    * The flow job plus every step job it ran, the failure and preprocessor steps
    * included (a failure handler's answer is persisted under its own job), and the
    * jobs an agent step's tool calls ran as (a tool row is persisted under its own
-   * job too). Unknown when the read fails.
+   * job too). A failed read is retried like the rows are; unknown when it keeps
+   * failing or the credential may not read jobs. Unknown accepts every row after the
+   * question: refusing them would leave a token without job access with no turn ever
+   * answered, each one finished a second time from its result.
    */
   async #turnJobIds(turn: Turn): Promise<Set<string> | undefined> {
-    try {
-      const job = await this.#api.getFlowJob(turn.jobId!, turn.controller.signal)
-      const ids = new Set([turn.jobId!])
-      const status = job.flow_status
-      for (const m of [...(status?.modules ?? []), status?.failure_module, status?.preprocessor_module]) {
-        if (m?.job) ids.add(m.job)
-        for (const j of m?.flow_jobs ?? []) ids.add(j)
-        for (const a of m?.agent_actions ?? []) if (a.job_id) ids.add(a.job_id)
+    for (let attempt = 1; ; attempt++) {
+      try {
+        const job = await this.#api.getFlowJob(turn.jobId!, turn.controller.signal)
+        const ids = new Set([turn.jobId!])
+        const status = job.flow_status
+        for (const m of [...(status?.modules ?? []), status?.failure_module, status?.preprocessor_module]) {
+          if (m?.job) ids.add(m.job)
+          for (const j of m?.flow_jobs ?? []) ids.add(j)
+          for (const a of m?.agent_actions ?? []) if (a.job_id) ids.add(a.job_id)
+        }
+        return ids
+      } catch (e) {
+        if (isAbortError(e)) throw e
+        const refused = e instanceof WindmillApiError && e.status >= 400 && e.status < 500
+        if (refused || attempt === RECONCILE_ATTEMPTS) return undefined
+        await sleep(RECONCILE_DELAY_MS, turn.controller.signal)
       }
-      return ids
-    } catch (e) {
-      if (isAbortError(e)) throw e
-      return undefined
     }
   }
 

chat-sdk/src/follow.ts

@@ -41,8 +41,10 @@ export async function* followJob(
     let reopen = false
     try {
       for await (const update of api.streamJob(jobId, { streamOffset: offset, signal: options.signal })) {
-        failures = 0
+        // A ping proves the connection opened, not that it carries the job: only an update
+        // clears the count, or a connection that pings and drops would never reach polling.
         if (update.type === 'ping') continue
+        failures = 0
         if (update.type === 'timeout') {
           reopen = true
           break

chat-sdk/test/chat.test.ts

@@ -1005,13 +1005,20 @@ describe('createChat with server history', () => {
 
   test("the previous run's answer landing after the next message is not that turn's answer", async () => {
     let reads = 0
+    let jobReads = 0
     const { fetch } = fetchMock(
       run,
       (c) =>
         c.url.pathname === streamPath
           ? sse([{ type: 'update', completed: true, only_result: { windmill_chat_answer: 'second answer' } }])
           : undefined,
-      (c) => (c.url.pathname.endsWith('/jobs_u/get/job-1') ? json({ flow_status: { modules: [{ job: 'step-2' }] } }) : undefined),
+      // A transient failure of the job read is retried, not taken as "any row counts".
+      (c) =>
+        c.url.pathname.endsWith('/jobs_u/get/job-1')
+          ? ++jobReads === 1
+            ? text('bad gateway', 502)
+            : json({ flow_status: { modules: [{ job: 'step-2' }] } })
+          : undefined,
       (c) => {
         if (!c.url.pathname.endsWith('/messages')) return undefined
         if (!c.url.searchParams.has('after_seq')) return json([messageRow(71, 'user', 'first')])
@@ -1058,9 +1065,20 @@ describe('createChat with server history', () => {
   }, 15000)
 
   test('a stream that keeps failing hands the turn to polling the job', async () => {
+    // Each connection opens, sends the server's ping, then drops.
+    const pingThenDrop = () =>
+      new Response(
+        new ReadableStream<Uint8Array>({
+          start(controller) {
+            controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify({ type: 'ping' })}\n\n`))
+            setTimeout(() => controller.error(new TypeError('network connection was lost')), 20)
+          }
+        }),
+        { status: 200, headers: { 'content-type': 'text/event-stream' } }
+      )
     const { fetch } = fetchMock(
       run,
-      (c) => (c.url.pathname === streamPath ? text('bad gateway', 502) : undefined),
+      (c) => (c.url.pathname === streamPath ? pingThenDrop() : undefined),
       (c) =>
         c.url.pathname.endsWith('/get_result_maybe/job-1')
           ? json({ completed: true, success: true, result: { windmill_chat_answer: 'polled' } })