Files
windmill/chat-sdk/test/follow.test.ts
GuilhemandClaude Opus 5 f367eaf6d0 feat: run turns in several flow chat conversations at once (#11202)
* feat: run turns in several flow chat conversations at once

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

* fix: keep finished turns finished and cached chats current in the flow chat pool

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

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

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

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

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

* fix: keep a chat that holds an unsent draft, and take one back when its first turn is withdrawn

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

* fix: ignore a stale running-turn snapshot, keep a withdrawn chat's draft, poll after clean stream ends

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

* fix: follow the turn running now when the listing named one already over

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

* fix: keep replacement turns and SSE fallback moving

* fix: keep replacement turn handoffs active

* fix: preserve unread badge line height

* fix: settle local fallback handoffs

* fix: settle refused turn handoffs

* fix: scope turn handoffs to conversation

* fix: drop stale turn handoffs

* refactor: move the queued message and 409 handling into per-conversation turns

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

* fix: address cubic's review of the parallel flow chat turns

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

* fix: clear a stale failure on refresh, and tighten the docs and test waits

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

* fix: recover running rows past the first page, and drop the failure a re-read disproves

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

* refactor: check the running-turn query at compile time, and narrow what a refresh clears

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

* fix: settle a failed turn only from an answer that turn wrote

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

* fix: settle a failed turn from its own answer, and only while it is still the failure shown

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

* fix: drop a failure whose answer arrived even when a newer turn owns the error

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

* fix: free an answered failure whatever the turn that started meanwhile is doing

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

* fix: drop a rows read that a turn outran, rather than merging it under newer messages

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

* fix: drop a rows read whose conversation was left and opened again

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

* fix: hand over a file still being read when its composer goes

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

* fix: count a drop's routing as work in flight, so its file is handed over too

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

* fix: hold the send until every file a conversation is owed has landed

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

* refactor: keep a panel mounted per conversation instead of handing its draft over

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

* fix: keep the withdrawn chat whose composer was written in, not the empty one

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

* fix: keep the chat in front of the reader when both withdrawn composers were written in

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

* fix: keep a retry's own run arguments when a turn elsewhere refuses it

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

* fix: name panels apart across pools, and read a flow's inputs when its chat is built

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

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-29 11:36:30 +02:00

43 lines
2.0 KiB
TypeScript

import { expect, test } from 'bun:test'
import { WindmillChatApi } from '../src/api'
import { followJob } from '../src/follow'
import { fetchMock, ndjson, sse } from './support'
test('a retried streaming step reports its offset as lost before the new sub-job is followed', async () => {
let streams = 0
const { fetch } = fetchMock((c) => {
if (!c.url.pathname.endsWith('/jobs_u/getupdate_sse/job-1')) return undefined
streams++
if (streams === 1) {
return sse([
{ type: 'update', new_result_stream: ndjson({ type: 'token_delta', content: 'a' }), stream_offset: 3, flow_stream_job_id: 'agent-1' },
{ type: 'update', new_result_stream: ndjson({ type: 'token_delta', content: 'b' }), stream_offset: 4, flow_stream_job_id: 'agent-2' }
])
}
return sse([{ type: 'update', stream_offset: 1, flow_stream_job_id: 'agent-2', completed: true, only_result: 'ok' }])
})
const api = new WindmillChatApi({ baseUrl: 'http://wm.test', workspace: 'ws', token: 'tok', fetch })
const offsets: (number | undefined)[] = []
for await (const _ of followJob(api, 'job-1', { onOffset: (o) => offsets.push(o) })) {
// A resumer that stored offset 3 must not reuse it against agent-2.
}
expect(offsets).toEqual([3, undefined, 1])
})
test('a 200 carrying no stream is reconnected like any other failed connection', async () => {
let streams = 0
const { fetch } = fetchMock((c) => {
if (!c.url.pathname.endsWith('/jobs_u/getupdate_sse/job-1')) return undefined
// Something in front of Windmill answers 200 with nothing in it.
if (++streams === 1) return new Response(null, { status: 200 })
return sse([{ type: 'update', completed: true, only_result: 'ok' }])
})
const api = new WindmillChatApi({ baseUrl: 'http://wm.test', workspace: 'ws', token: 'tok', fetch })
let result: unknown
for await (const event of followJob(api, 'job-1')) {
if (event.type === 'completed') result = event.result
}
expect(streams).toBe(2)
expect(result).toBe('ok')
})