fix: attribute a turn's rows by job id as well as sequence

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

chat-sdk/src/chat.ts

@@ -46,6 +46,8 @@ interface Turn {
   /** Id of the turn's user message; the answer is whatever follows it. */
   userMessageId: string
   jobId?: string
+  /** The flow job and its step jobs; a persisted answer carries one of them as `job_id`. */
+  jobIds?: Set<string>
   /** Id of the streaming assistant message; cleared when a tool call ends the round. */
   assistantId?: string
   streamedText: boolean
@@ -572,6 +574,8 @@ class ChatImpl implements Chat {
   async #finishTurn(turn: Turn, result: unknown, isNew: boolean): Promise<void> {
     if (!this.#turnActive(turn)) return
     if (this.#state.history === 'server') {
+      turn.jobIds = await this.#turnJobIds(turn)
+      if (!this.#turnActive(turn)) return
       const reconciled = await this.#reconcileTurn(turn)
       if (!this.#turnActive(turn)) return
       if (reconciled) {
@@ -649,24 +653,52 @@ class ChatImpl implements Chat {
    * a badly delayed one can invert that order at the cost of the reconcile
    * retries). The content is not compared with the flow result: an image answer, a
    * structured one and a forwarded agent result are all persisted in a shape the
-   * result does not reproduce. Every row created after the user message is the
-   * turn's: the server refuses a turn while the conversation's previous run is still
-   * queued, so no other run of this conversation writes meanwhile. Until the user
+   * result does not reproduce. A row belongs to the turn when it was created after
+   * the user message and, when it carries a job id, the job is one of the turn's. The
+   * server refuses a turn while the previous run is queued, but an agent writes its
+   * answer row from a task the run does not wait for, so that row can still land after
+   * the next user message; its job says whose it is. A tool row without a job (an MCP
+   * call runs inside the agent step) belongs to the turn under way. Until the user
    * message's own row has been read, its position in the list stands in for its seq.
    */
   #answered(turn: Turn): boolean {
     const messages = this.#state.messages
     const from = messages.findIndex((m) => m.id === turn.userMessageId)
     const userSeq = messages[from]?.seq
+    const ownJob = (m: ChatMessage) =>
+      turn.jobIds === undefined || (m.jobId === undefined ? m.role === 'tool' : turn.jobIds.has(m.jobId))
     let latest: ChatMessage | undefined
     messages.forEach((m, i) => {
-      if (m.seq === undefined || m.role === 'user') return
+      if (m.seq === undefined || m.role === 'user' || !ownJob(m)) return
       if (userSeq !== undefined ? m.seq <= userSeq : i <= from) return
       if (latest === undefined || m.seq > latest.seq!) latest = m
     })
     return latest?.role === 'assistant'
   }
 
+  /**
+   * 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.
+   */
+  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)
+      }
+      return ids
+    } catch (e) {
+      if (isAbortError(e)) throw e
+      return undefined
+    }
+  }
+
   #failTurn(turn: Turn, e: unknown): void {
     if (!this.#turnActive(turn)) return
     const error = toError(e)

chat-sdk/test/chat.test.ts

@@ -745,7 +745,7 @@ describe('createChat with server history', () => {
   })
 
   test('resuming a turn whose message is off the first page replays it without duplicating rows', async () => {
-    const { fetch, calls } = fetchMock(
+    const { fetch } = fetchMock(
       (c) =>
         c.url.pathname === streamPath && !c.url.searchParams.has('stream_offset')
           ? sse([
@@ -780,7 +780,37 @@ describe('createChat with server history', () => {
       ['row-52', 'assistant', 'Done', false]
     ])
     expect(chat.getState().status).toBe('idle')
-    expect(calls.some((c) => c.url.pathname.includes('/jobs_u/get/'))).toBe(false)
+  })
+
+  test("the previous run's answer landing after the next message is not that turn's answer", async () => {
+    let reads = 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),
+      (c) => {
+        if (!c.url.pathname.endsWith('/messages')) return undefined
+        if (!c.url.searchParams.has('after_seq')) return json([messageRow(71, 'user', 'first')])
+        // The earlier agent wrote its answer from a task its run did not wait for.
+        return json(
+          ++reads === 1
+            ? [messageRow(72, 'user', 'second'), messageRow(73, 'assistant', 'first answer, late', { job_id: 'step-1' })]
+            : [messageRow(74, 'assistant', 'second answer', { job_id: 'step-2' })]
+        )
+      }
+    )
+    const chat = createChat(options({}, fetch))
+    await chat.selectConversation('conv')
+    await chat.sendMessage('second')
+    expect(chat.getState().messages.map((m) => [m.role, m.content, m.serverId])).toEqual([
+      ['user', 'first', 'row-71'],
+      ['user', 'second', 'row-72'],
+      ['assistant', 'first answer, late', 'row-73'],
+      ['assistant', 'second answer', 'row-74']
+    ])
   })
 
   test('an answer cut by a lost stream gives way to the polled result', async () => {