starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-21 00:02:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(chat): let a turn own every hold it takes on a chat

Browse Source 
Three of the last five review rounds were the same defect with a
different trigger: some path decided a chat was free, or stuck, without
the run having said so. `Turn` gave that question an owner, but the
holds taken before a turn exists were left outside it, and the two that
were are what these fix.

The read that opens a conversation now starts a turn before it awaits.
That read is how the chat finds out whether a run owns the conversation,
so it holds the chat while it is out — and Stop, the panel going away,
and a later send all have to be able to reach it. Each ends the turn,
and a read whose turn is over writes nowhere: it can no longer drop the
rows a live turn is writing, nor resume on top of the turn that replaced
it. `#resumeRunningTurn` carries that turn on rather than starting a
second, which would end the read that produced the rows it reasons over.

A read that fails leaves the chat held, since whether a run owns the
agent memory is exactly what could not be read. `#holdUnreadableChat` is
the one place that happens, and it holds only a turn that still holds
the chat — a read outruns a Stop and fails after it, and taking the hold
back then would shut a composer the reader was just handed.

The run request is not tied to the turn's handle, so the whole of it is
a window in which Stop can land, on a turn with no job to cancel yet.
The run it returns is then nobody's: it is cancelled, and the send
reports that nothing ran.

Which turn is running is asked of the server rather than read back to
through the transcript. The kind is checked again here, because the
caller takes that row's job as the turn's: every other kind names the
agent step's own job, and a turn named with that has Stop cancel the
step while the rest of the flow runs on.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Guilhem Lemouel
2026-09-16 10:08:24 +02:00
co-authored by Claude Opus 5
parent 2199050518
commit 404419588c
3 changed files with 291 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts
+111 -77
View File
@@ -68,10 +68,6 @@ const POLL_PAGE_SIZE = 50
* stops short of is left to the next tick, which resumes from the cursor it reached. */
const POLL_MAX_REQUESTS = 20
/** Pages walked back looking for the message that started the newest turn. Bounds the search
* on a turn that wrote an implausible number of rows; the conversation's own start ends it. */
const RESUME_MAX_PAGES = 20
/**
* The agent events the worker streams, in the shape the transcript applies. The SDK names
* the same six events after the wire protocol; this is the rest of the app's vocabulary.
@@ -472,12 +468,12 @@ export class FlowChatManager {
}
/** Stop following one conversation's turn and forget what it was mid-way through. */
endTurn(conversationId: string, options?: { settled?: boolean }) {
endTurn(conversationId: string, options?: { settled?: boolean; byUser?: boolean }) {
// Out of the record first: the turn is no longer this conversation's, so anything it
// started that resolves from here on fails `#isCurrent` and writes nothing.
const turn = this.#turns[conversationId]
delete this.#turns[conversationId]
turn?.end()
turn?.end({ byUser: options?.byUser })
const status = this.#liveStatus(conversationId)
status.isLoading = false
status.isWaitingForResponse = false
@@ -738,6 +734,34 @@ export class FlowChatManager {
}
}
/**
* A run that came back to a turn that is already over, which the reader sees as a send
* that did not happen.
*
* The run request is not tied to the turn's handle, so the whole of it is a window in
* which the turn can end. Stop means the reader did not want this run, so now that it
* has a name it is cancelled; every other way a turn ends — the chat closing, a flow
* change — has never cancelled a run, and a reload picks that one back up.
*/
#abandonLaunchedRun(turn: Turn, jobId: string): false {
if (turn.endedByUser) void this.#cancelOrphanedRun(jobId)
return false
}
/** Cancel a run nothing is following, reporting only to the console: the reader has
* already been told the turn stopped, and this is the part that was still in flight. */
async #cancelOrphanedRun(jobId: string) {
try {
await JobService.cancelQueuedJob({
workspace: this.#workspace()!,
id: jobId,
requestBody: {}
})
} catch (error) {
console.error('Could not cancel a run left by a turn that was stopped:', error)
}
}
/** Stop the open conversation's turn. Every other chat's keeps running. */
async cancelCurrentJob() {
const conversationId = this.selectedConversationId
@@ -761,7 +785,7 @@ export class FlowChatManager {
}
// The turn that settled during the cancel request has already gone, and the one
// that replaced it owns a run of its own.
if (!turn || this.#isCurrent(turn)) this.endTurn(conversationId)
if (!turn || this.#isCurrent(turn)) this.endTurn(conversationId, { byUser: true })
} catch (error) {
// The run may well still be going, and freeing the chat would let the next turn
// write the same agent memory. It is left to the job to say when it is over.
@@ -770,7 +794,7 @@ export class FlowChatManager {
// The turn's own follower is still on the job and is what will settle it; another
// would ask the same question twice. With no turn left there is nothing following
// it, and the chat is released instead.
if (!turn || !this.#isCurrent(turn)) this.endTurn(conversationId)
if (!turn || !this.#isCurrent(turn)) this.endTurn(conversationId, { byUser: true })
}
}
@@ -803,15 +827,26 @@ export class FlowChatManager {
let conversationIdToUse = conversationId ?? this.selectedConversationId
if (!this.#workspace() || !conversationIdToUse) return
// The turn the opening read holds the chat with, for as long as it takes to find out
// whether a run is going. Held here so every write below can ask whether it is still
// the turn this conversation is on.
let turn: Turn | undefined
if (reset) {
// Rows already held are either what a previous load fetched or what a turn is
// writing right now; either way re-fetching would drop a live turn's temp rows.
if (this.#rowsById[conversationIdToUse]) {
// writing right now; either way re-fetching would drop a live turn's temp rows. A
// turn with no rows yet is the same chat being opened twice at once, and the
// second opening has nothing to add. A read that failed leaves neither, which is
// what lets selecting the chat again retry it.
if (this.#rowsById[conversationIdToUse] || this.#turnOf(conversationIdToUse)) {
return
}
this.isLoadingMessages = true
// Whether this chat has a run in flight is unknown until its rows are here and
// the job has answered, and a message accepted meanwhile starts a second one.
// the job has answered, and a message accepted meanwhile starts a second one. The
// hold is a turn like any other because Stop, the panel going away and a later
// send all have to reach the read it is waiting on: each ends the turn, and a read
// whose turn is over writes nowhere.
turn = this.#startTurn(conversationIdToUse)
this.#liveStatus(conversationIdToUse).isDispatchingTurn = true
} else {
this.loadingMoreMessages = true
@@ -830,10 +865,14 @@ export class FlowChatManager {
})
if (reset) {
// Stopped, or overtaken by a send that started its own turn while this was in
// flight. The rows it wrote are the live ones; these are what the conversation
// looked like before it, and writing them would drop the turn's own.
if (!turn || !this.#isCurrent(turn)) return
this.#rowsById[conversationIdToUse] = response
this.#pagedTo[conversationIdToUse] = 1
this.isLoadingMessages = false
void this.#resumeRunningTurn(conversationIdToUse)
void this.#resumeRunningTurn(turn)
await new Promise((resolve) => setTimeout(resolve, 100))
this.scrollToBottom()
} else {
@@ -858,9 +897,7 @@ export class FlowChatManager {
} catch (error) {
console.error('Failed to load messages:', error)
sendUserToast('Failed to load messages: ' + error)
// The hold stays: an empty transcript is not an idle conversation, and the run
// this could not ask about owns the agent memory a second turn would write.
// Selecting the chat again retries the load; Stop is the reader's way out.
if (turn) this.#holdUnreadableChat(turn)
} finally {
this.isLoadingMessages = false
this.loadingMoreMessages = false
@@ -1141,6 +1178,7 @@ export class FlowChatManager {
// What Stop cancels: the flow job, never the streaming step's own sub-job — which
// would leave the steps after the agent running.
this.#nameTurnJob(turn, userRowId, jobId)
if (!this.#isCurrent(turn)) return this.#abandonLaunchedRun(turn, jobId)
// start polling
turn.startPolling()
@@ -1272,17 +1310,14 @@ export class FlowChatManager {
* its flow job, which is the only thing that knows whether it is over.
*
* The chat is held from the moment the question is asked, not from the answer: a send
* accepted during that round trip is the very thing this exists to prevent.
*
* Which turn is running is inferred from the rows rather than asked of the server, so a
* turn whose own rows fill the page the transcript opened on takes a walk back through
* older pages to find the message that started it (see `#newestUserRow`).
* accepted during that round trip is the very thing this exists to prevent. The turn
* doing the holding is the one the opening read started, carried on rather than
* replaced — starting a second here would end the first, and with it the read that
* produced the rows this is about to reason over.
*/
async #resumeRunningTurn(conversationId: string) {
async #resumeRunningTurn(turn: Turn) {
const conversationId = turn.conversationId
const status = this.#liveStatus(conversationId)
// The turn starts here, before anything is awaited, so a chat torn down or stopped
// while this is in flight can still stop what it would start.
const turn = this.#startTurn(conversationId)
try {
await this.#takeOverRunningTurn(turn, status)
// The hold this attempt was given back, and only it: a turn that replaced this one
@@ -1292,76 +1327,74 @@ export class FlowChatManager {
status.isDispatchingTurn = false
}
} catch (error) {
// Whether a run owns this conversation's agent memory is exactly what could not be
// read, so the chat stays held rather than taking a message that would write it a
// second time. Stop is the reader's way out.
console.error('Could not tell whether a conversation had a run in flight:', error)
if (!this.#turnOf(conversationId)) status.isDispatchingTurn = true
this.#holdUnreadableChat(turn)
}
}
/**
* Hold a chat whose state could not be read, with nothing following a run for it.
*
* Whether a run owns this conversation's agent memory is exactly what failed to be read,
* so the chat cannot take a message that would write that memory a second time. The turn
* that was asking goes with it, having answered nothing and having no run to follow, and
* Stop is the reader's way out of what is left.
*/
#holdUnreadableChat(turn: Turn) {
// Only a turn that still holds the chat may hold it again. A read is not tied to the
// turn's handle, so it runs on past a Stop and fails afterwards — and taking the hold
// back then would shut a composer the reader has just been given, over a run that was
// cancelled. The chat going away and a later turn taking it are the same story.
if (!this.#isCurrent(turn)) return
this.endTurn(turn.conversationId)
this.#liveStatus(turn.conversationId).isDispatchingTurn = true
}
/**
* The message that started the newest turn in a conversation, which is the only turn
* that can still be running.
*
* The transcript opens on the newest page, and one turn can fill it on its own: an agent
* writes a row per round and per tool call, so a turn that used a lot of tools pushes the
* message that started it further back than a page. Pages are walked back until it turns
* up, since without it a reload cannot tell a finished conversation from one still
* running, and would offer a composer that writes a second turn into the same memory.
* Asked of the server when the rows on screen do not hold it. The transcript opens on
* the newest page and one turn can fill it on its own — an agent writes a row per round
* and per tool call — so the message that started a long turn sits an unbounded number
* of pages back, and reading towards it is a search with no end the browser can name.
* Without it a reload cannot tell a finished conversation from one still running, and
* would offer a composer that writes a second turn into the same agent memory.
*/
async #newestUserRow(
conversationId: string,
signal?: AbortSignal
): Promise<ChatMessage | undefined> {
async #newestUserRow(conversationId: string): Promise<ChatMessage | undefined> {
const held = this.#rowsOf(conversationId).findLast((row) => row.message_type === 'user')
if (held) return held
const from = (this.#pagedTo[conversationId] ?? 1) + 1
for (let page = from; page < from + RESUME_MAX_PAGES; page++) {
if (signal?.aborted) return undefined
const batch = await FlowConversationsService.listConversationMessages({
workspace: this.#workspace()!,
conversationId,
page,
perPage: this.#perPage
})
const found = batch.findLast((row) => row.message_type === 'user')
if (found) return found
// The start of the conversation, which a user row always opens — so this only runs
// out of rows on one written by something other than a chat.
if (batch.length < this.#perPage) return undefined
const [newest] = await FlowConversationsService.listConversationMessages({
workspace: this.#workspace()!,
conversationId,
messageType: 'user',
page: 1,
perPage: 1
})
// No rows at all: a conversation written by something other than a chat, and there is
// no turn to resume.
if (!newest) return undefined
// The kind is checked here rather than trusted to the request, because the caller
// takes this row's job as the turn's. Every other kind names the agent step's own
// job, and a turn named with that has Stop cancel the step while the rest of the
// flow runs on — so a server that did not honour the filter has answered a different
// question, and the caller is told so rather than handed the wrong run.
if (newest.message_type !== 'user') {
throw new Error(
`Asked conversation ${conversationId} for its newest user message and got a ` +
`${newest.message_type} one`
)
}
console.warn(
`Gave up looking for the message that started the newest turn of conversation ` +
`${conversationId} after ${RESUME_MAX_PAGES} pages`
)
return undefined
return newest
}
async #takeOverRunningTurn(turn: Turn, status: TurnStatus): Promise<boolean> {
const conversationId = turn.conversationId
// A turn already in flight here owns the chat; only the load's own hold is set.
if (status.isLoading || status.isWaitingForResponse) {
this.#endTurnIfCurrent(turn)
return false
}
if (!this.#workspace()) {
this.#endTurnIfCurrent(turn)
return false
}
// The newest turn is the only one that can still be running, and the message that
// started it is named with its flow job by the run that created it.
let startedBy: ChatMessage | undefined
try {
startedBy = await this.#newestUserRow(conversationId, turn.signal)
} catch (error) {
// The caller decides what happens to the chat's hold; what this owns is the turn
// it started, which answered nothing and has no run to follow.
this.#endTurnIfCurrent(turn)
throw error
}
// Torn down or stopped while the walk was in flight. The run is left alone either
// started it is named with its flow job by the run that created it. A read that fails
// leaves the caller to decide what becomes of the chat's hold.
const startedBy = await this.#newestUserRow(conversationId)
// Torn down or stopped while that read was in flight. The run is left alone either
// way: leaving a chat has never cancelled one, and this cannot tell the two apart.
// A turn that is no longer this conversation's has already been ended by whatever
// replaced it; ending "the conversation's turn" here would end that one.
@@ -1484,6 +1517,7 @@ export class FlowChatManager {
// Store the current job ID so it can be cancelled
this.#nameTurnJob(turn, userRowId, jobId)
if (!this.#isCurrent(turn)) return this.#abandonLaunchedRun(turn, jobId)
if (turn.listPending) {
await this.refreshConversations()
frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts
+165 -14
View File
@@ -1,5 +1,5 @@
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
import { FlowConversationsService } from '$lib/gen'
import { FlowConversationsService, JobService } from '$lib/gen'
import { createFlowChatManager } from './FlowChatManager.svelte'
vi.mock('$lib/gen', () => ({
@@ -7,7 +7,7 @@ vi.mock('$lib/gen', () => ({
listConversationMessages: vi.fn(),
deleteFlowConversation: vi.fn()
},
JobService: {},
JobService: { cancelQueuedJob: vi.fn() },
FlowService: {}
}))
vi.mock('$lib/toast', () => ({ sendUserToast: vi.fn() }))
@@ -426,7 +426,12 @@ describe('a sent message names the run it started', () => {
// These turns stream too, so they draw on the shared script the block above resets.
streamCalls.length = 0
streamScript.length = 0
vi.mocked(FlowConversationsService.listConversationMessages).mockResolvedValue([] as any)
vi.mocked(FlowConversationsService.listConversationMessages)
.mockReset()
.mockResolvedValue([] as any)
vi.mocked(JobService.cancelQueuedJob)
.mockReset()
.mockResolvedValue('' as any)
;(globalThis as any).location = { origin: 'http://localhost' }
})
@@ -473,6 +478,39 @@ describe('a sent message names the run it started', () => {
const userRow = manager.messages.find((m) => m.message_type === 'user')
expect(userRow?.job_id).toBe('job-8')
})
/**
* The whole of the run request is a window in which Stop can land, and it lands on a turn
* with no job to cancel. The run the request then returns is nobody's: nothing follows
* it, and it would hold the conversation's agent memory against the next turn.
*/
it('cancels a run that arrives after Stop ended the turn', async () => {
let launch: ((jobId: string) => void) | undefined
const manager = (live = managerWithRows())
;(manager as any).initialize(
vi.fn(() => new Promise<string>((resolve) => (launch = resolve))),
'u/admin/flow',
true
)
manager.operatingWorkspace = () => 'ws'
manager.selectedConversationId = 'a'
manager.inputMessage = 'hello'
const sent = manager.sendMessage(undefined, undefined, 'a')
await vi.waitFor(() => expect(launch).toBeTruthy())
await manager.cancelCurrentJob()
launch!('job-late')
expect(await sent).toBe(false)
await vi.waitFor(() =>
expect(vi.mocked(JobService.cancelQueuedJob)).toHaveBeenCalledWith(
expect.objectContaining({ id: 'job-late' })
)
)
// Nothing followed it either: a stream opened here would write into whatever turn
// the chat is on by the time it answers.
expect(streamCalls).toEqual([])
})
})
/**
@@ -532,9 +570,10 @@ describe('a conversation opened while its run is still going', () => {
* A turn that called a lot of tools writes more rows than the page the transcript opens
* on, pushing the message that started it out of sight. Read from that page alone the
* conversation looks finished, and the composer is handed back while the run still owns
* the agent's memory.
* the agent's memory — so the message is asked of the server by kind, which is what makes
* the answer independent of how many rows the turn wrote.
*/
it('walks back past a turn that filled the page to find the message that started it', async () => {
it('asks the server for the message that started a turn that filled the page', async () => {
const answerRows = Array.from({ length: 50 }, (_, i) => ({
id: `answer-${i}`,
conversation_id: 'a',
@@ -547,7 +586,7 @@ describe('a conversation opened while its run is still going', () => {
.mockReset()
// The newest page is this turn's answer, all of it.
.mockResolvedValueOnce(answerRows as any)
// The page before it ends with the message that started the turn.
// The newest `user` row, which is the message that started the turn.
.mockResolvedValueOnce([
{
id: 'the-question',
@@ -567,6 +606,11 @@ describe('a conversation opened while its run is still going', () => {
await vi.waitFor(() => expect(manager.isConversationBusy('a')).toBe(true))
await vi.waitFor(() => expect(streamCalls.some((call) => call.jobId === 'job-live')).toBe(true))
// Asked for by kind. Without that the newest row of any kind comes back, which for a
// turn in flight is one the agent wrote — and those name the agent step's own job.
expect(vi.mocked(FlowConversationsService.listConversationMessages)).toHaveBeenLastCalledWith(
expect.objectContaining({ messageType: 'user', page: 1, perPage: 1 })
)
// The rows the turn already wrote are dropped, since the stream replays them. The
// message that started it is not: polls only ever add what a turn wrote, so a
// transcript emptied here would come back as answers with nothing asking them.
@@ -596,15 +640,13 @@ describe('a conversation opened while its run is still going', () => {
})
/**
* The hold a failed load leaves behind gates the whole surface — New chat and every
* other conversation with it — so Stop has to be able to release it.
* Asking which turn is running is a request of its own, and it can fail on its own.
* Whether a run owns this conversation's memory is then precisely what is unknown, so the
* chat stays held rather than taking a message that would write that memory twice — and
* the hold gates the whole surface, New chat and every other conversation with it, so
* Stop has to be able to release it.
*/
/**
* The walk back for the message that started the turn is a request of its own, and it can
* fail on its own. Whether a run owns this conversation's memory is then precisely what is
* unknown, so the chat stays held rather than taking a message that would write it twice.
*/
it('keeps holding the chat when it could not read far enough to ask', async () => {
it('keeps holding the chat when it could not ask which turn is running', async () => {
const answerRows = Array.from({ length: 50 }, (_, i) => ({
id: `answer-${i}`,
conversation_id: 'a',
@@ -645,6 +687,115 @@ describe('a conversation opened while its run is still going', () => {
expect(manager.isConversationBusy('a')).toBe(false)
})
/**
* The row the resume names the turn with becomes the run Stop cancels, so it has to be
* the message that started the turn. Every other kind of row names the agent step's own
* job, and cancelling that leaves the steps after the agent running.
*/
it('holds the chat rather than resuming on a row it did not ask for', async () => {
const answerRows = Array.from({ length: 50 }, (_, i) => ({
id: `answer-${i}`,
conversation_id: 'a',
message_type: 'assistant',
content: `round ${i}`,
created_at: new Date().toISOString(),
created_seq: 100 + i
}))
vi.mocked(FlowConversationsService.listConversationMessages)
.mockReset()
.mockResolvedValueOnce(answerRows as any)
// A server that ignored the filter answers with the newest row of any kind, and
// that row carries the agent step's job.
.mockResolvedValueOnce([
{
id: 'an-answer',
conversation_id: 'a',
message_type: 'assistant',
content: 'the agent talking',
created_at: new Date().toISOString(),
created_seq: 150,
job_id: 'job-of-the-agent-step'
}
] as any)
const manager = (live = managerWithRows())
;(manager as any).initialize(vi.fn(), 'u/admin/flow', true)
manager.operatingWorkspace = () => 'ws'
await manager.selectConversation('a')
await vi.waitFor(() => expect(manager.isConversationBusy('a')).toBe(true))
expect(manager.currentJobId).toBeUndefined()
expect(streamCalls).toEqual([])
})
/**
* Stop does not reach the read that is already out, so the read fails after the composer
* has been handed back. Held against the chat then, it would shut a composer the reader
* was just given, over a run that Stop had already dealt with.
*/
it('stays released when the read fails after Stop', async () => {
let fail: ((error: Error) => void) | undefined
vi.mocked(FlowConversationsService.listConversationMessages)
.mockReset()
.mockImplementationOnce(() => new Promise((_, reject) => (fail = reject)) as any)
const manager = (live = managerWithRows())
;(manager as any).initialize(vi.fn(), 'u/admin/flow', true)
manager.operatingWorkspace = () => 'ws'
void manager.selectConversation('a')
await vi.waitFor(() => expect(fail).toBeTruthy())
await manager.cancelCurrentJob()
expect(manager.isConversationBusy('a')).toBe(false)
fail!(new Error('transcript unavailable'))
await vi.waitFor(() => expect(manager.isLoadingMessages).toBe(false))
expect(manager.isConversationBusy('a')).toBe(false)
})
/**
* The read that opens a conversation is how the chat finds out whether a run owns it, so
* it holds the chat while it is out — and Stop can release that hold and let the reader
* send before the rows arrive. What comes back is then the conversation as it was before
* that turn: written over the live one it would drop the message just sent, and resuming
* from it would end the turn now streaming.
*/
it('leaves a live turn alone when the read it replaced answers late', async () => {
let deliver: ((rows: unknown[]) => void) | undefined
vi.mocked(FlowConversationsService.listConversationMessages)
.mockReset()
.mockImplementationOnce(() => new Promise((resolve) => (deliver = resolve)) as any)
.mockResolvedValue([] as any)
const manager = (live = managerWithRows())
;(manager as any).initialize(
vi.fn(async () => 'job-new'),
'u/admin/flow',
true
)
manager.operatingWorkspace = () => 'ws'
void manager.selectConversation('a')
await vi.waitFor(() => expect(deliver).toBeTruthy())
await manager.cancelCurrentJob()
manager.inputMessage = 'ask again'
await manager.sendMessage(undefined, undefined, 'a')
deliver!([
{
id: 'before-the-send',
conversation_id: 'a',
message_type: 'user',
content: 'an older question',
created_at: new Date().toISOString(),
created_seq: 1
}
])
await vi.waitFor(() => expect(streamCalls.some((call) => call.jobId === 'job-new')).toBe(true))
expect(manager.messages.map((m) => m.content)).toEqual(['ask again'])
expect(manager.isConversationBusy('a')).toBe(true)
})
it('leaves a conversation whose run is over alone', async () => {
jobCompleted.value = true
const manager = opened('job-done')
frontend/src/lib/components/flows/conversations/turn.svelte.ts
+15 -3
View File
@@ -76,6 +76,7 @@ export class Turn {
#pollDeadline: ReturnType<typeof setTimeout> | undefined
#onPoll: () => void
#ended = false
#endedByUser = false
constructor(options: TurnOptions) {
this.conversationId = options.conversationId
@@ -93,6 +94,15 @@ export class Turn {
})
}
/**
* The reader pressed Stop on this turn, rather than it being replaced or the chat going
* away. What separates "this run was not wanted" from "nobody is watching any more":
* leaving a chat has never cancelled a run, and Stop has always meant to.
*/
get endedByUser(): boolean {
return this.#endedByUser
}
/** Stopped, or replaced by a later turn that ended this one. */
get ended(): boolean {
return this.#ended || this.#controller.signal.aborted
@@ -137,11 +147,13 @@ export class Turn {
}
/**
* Stop everything this turn started. The run itself is not cancelled — leaving a chat
* has never stopped a flow, and a turn cannot tell being stopped from being left.
* Stop everything this turn started. The run itself is not cancelled here, whoever ends
* the turn: leaving a chat has never stopped a flow, and the one caller that does mean to
* cancel owns the request itself — this only records that it was them, in `endedByUser`.
*/
end() {
end(options?: { byUser?: boolean }) {
this.#ended = true
if (options?.byUser) this.#endedByUser = true
this.stopPolling()
this.#replyReveal.reset()
this.#reasoningReveal.reset()