From d14a78ceb8bc19ef8f68b024d1c2af86ddf28f96 Mon Sep 17 00:00:00 2001 From: Guilhem Lemouel Date: Thu, 17 Sep 2026 14:56:01 +0200 Subject: [PATCH] feat: run turns in several flow chat conversations at once Co-Authored-By: Claude Opus 5 (1M context) --- CONTEXT.md | 17 + .../tests/flow_conversation_running_turn.rs | 87 ++++++ .../src/lib.rs | 24 +- backend/windmill-api/openapi.yaml | 27 ++ .../windmill-common/src/flow_conversations.rs | 69 ++++- chat-sdk/README.md | 17 + chat-sdk/src/api.ts | 70 ++++- chat-sdk/src/chat.ts | 184 +++++++---- chat-sdk/src/follow.ts | 103 +++++-- chat-sdk/src/index.ts | 3 + chat-sdk/src/react.ts | 2 + chat-sdk/src/types.ts | 21 +- chat-sdk/test/chat.test.ts | 101 ++++-- .../lib/components/FlowPreviewContent.svelte | 6 +- .../common/badge/UnreadCountBadge.svelte | 53 ++++ .../flows/conversations/FlowChat.svelte | 88 ++++-- .../conversations/FlowChatInterface.svelte | 35 +-- .../FlowConversationsSidebar.svelte | 133 +++++--- .../flows/conversations/flowChatPool.test.ts | 126 ++++++++ .../flows/conversations/flowChatPool.ts | 290 ++++++++++++++++++ .../conversations/flowChatViewHost.svelte.ts | 77 +++-- .../conversations/flowChatViewHost.test.ts | 35 ++- .../components/sessions/SessionPicker.svelte | 19 +- .../sessions/SessionStatusDot.svelte | 20 +- 24 files changed, 1345 insertions(+), 262 deletions(-) create mode 100644 backend/tests/flow_conversation_running_turn.rs create mode 100644 frontend/src/lib/components/common/badge/UnreadCountBadge.svelte create mode 100644 frontend/src/lib/components/flows/conversations/flowChatPool.test.ts create mode 100644 frontend/src/lib/components/flows/conversations/flowChatPool.ts diff --git a/CONTEXT.md b/CONTEXT.md index fbc2592cc9..7824effd6f 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -48,8 +48,25 @@ _Avoid_: thread, session (that names an AI session, a different thing), chat (th One question and the answer to it: the run the question started, the handle that stops it, and the rows it is writing. At most one per conversation, and the chat is held for its whole length — from the moment the question takes the chat, before it has a job, until it is ended. +The server holds the same rule: a question sent to a conversation whose turn is still running +is refused, whoever sends it. Several conversations of one flow can each have a turn running. _Avoid_: request, exchange, message round +**Running turn**: +A turn whose run has not finished. Which conversations have one is the server's to say, so a +chat opened after the turn started — a reload, another tab — still sees it and follows it. +_Avoid_: busy, active, in flight + +**Queued message**: +A question typed into a conversation while its turn runs, sent when that turn ends answered. +At most one per conversation; typing another adds to it. +_Avoid_: pending message (a pending message is one already sent and not yet confirmed) + +**Unread**: +The answers that arrived in a conversation while it was not the one shown. Counted per open +chat and forgotten on reload. +_Avoid_: new messages, notifications + **Transcript**: The rows a conversation's chat holds. Not the conversation: it is the newest page plus whatever older pages the reader has scrolled back through, so a question it cannot answer diff --git a/backend/tests/flow_conversation_running_turn.rs b/backend/tests/flow_conversation_running_turn.rs new file mode 100644 index 0000000000..392ff6f5cf --- /dev/null +++ b/backend/tests/flow_conversation_running_turn.rs @@ -0,0 +1,87 @@ +//! A conversation answers one message at a time: while the run its newest user message started +//! is queued or running, the list reports that turn and a run into the conversation is refused. +//! +//! Uses runtime `sqlx::query` (not the compile-time macros) so no offline query cache is +//! needed, matching v2_job_delete_orphans.rs. + +use sqlx::{Pool, Postgres}; +use uuid::Uuid; +use windmill_common::error::Error; +use windmill_common::flow_conversations::{get_or_create_conversation_with_id, running_turns}; +use windmill_test_utils::*; + +const WS: &str = "test-workspace"; +const CONV: Uuid = Uuid::from_u128(0x5eed); + +/// Takes the conversation for a turn, as a run does, without keeping what it writes. +async fn take_conversation(db: &Pool) -> windmill_common::error::Result { + let mut tx = db.begin().await?; + let result = + get_or_create_conversation_with_id(&mut tx, WS, "f/flow", "test-user", "t", CONV, false) + .await; + tx.rollback().await?; + result.map(|c| c.id) +} + +#[sqlx::test(fixtures("base"))] +async fn test_a_conversation_refuses_a_turn_while_one_runs( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + + let job_id = Uuid::new_v4(); + sqlx::query("INSERT INTO v2_job (id, workspace_id, kind) VALUES ($1, $2, 'flow')") + .bind(job_id) + .bind(WS) + .execute(&db) + .await?; + sqlx::query( + "INSERT INTO v2_job_queue (id, workspace_id, scheduled_for) VALUES ($1, $2, now())", + ) + .bind(job_id) + .bind(WS) + .execute(&db) + .await?; + sqlx::query( + "INSERT INTO flow_conversation (id, workspace_id, flow_path, created_by) + VALUES ($1, $2, 'f/flow', 'test-user')", + ) + .bind(CONV) + .bind(WS) + .execute(&db) + .await?; + let user_seq: i64 = sqlx::query_scalar( + "INSERT INTO flow_conversation_message (conversation_id, message_type, content, job_id) + VALUES ($1, 'user', 'hi', $2) RETURNING created_seq", + ) + .bind(CONV) + .bind(job_id) + .fetch_one(&db) + .await?; + + let running = running_turns(&db, &[CONV]).await?; + let turn = running + .get(&CONV) + .expect("the queued run is the conversation's running turn"); + assert_eq!((turn.job_id, turn.user_seq), (job_id, user_seq)); + + match take_conversation(&db).await { + Err(Error::Generic(status, body)) => { + assert_eq!(status.as_u16(), 409); + assert!( + body.contains(&job_id.to_string()), + "the refusal names the running job: {body}" + ); + } + other => panic!("expected a 409 while the first turn runs, got {other:?}"), + } + + // The run ends: it leaves the queue, and the conversation takes the next message. + sqlx::query("DELETE FROM v2_job_queue WHERE id = $1") + .bind(job_id) + .execute(&db) + .await?; + assert!(running_turns(&db, &[CONV]).await?.is_empty()); + assert_eq!(take_conversation(&db).await?, CONV); + Ok(()) +} diff --git a/backend/windmill-api-flow-conversations/src/lib.rs b/backend/windmill-api-flow-conversations/src/lib.rs index 80eebfeda9..55ee8593d2 100644 --- a/backend/windmill-api-flow-conversations/src/lib.rs +++ b/backend/windmill-api-flow-conversations/src/lib.rs @@ -14,7 +14,7 @@ pub use windmill_common::flow_conversations::FlowConversation; use windmill_common::{ db::{UserDB, DB}, error::{JsonResult, Result}, - flow_conversations::MessageType, + flow_conversations::{running_turns, MessageType, RunningTurn}, utils::{not_found_if_none, paginate, truncate_with_ellipsis, Pagination}, }; @@ -68,7 +68,7 @@ async fn list_conversations( Path(w_id): Path, Query(pagination): Query, Query(query): Query, -) -> JsonResult> { +) -> JsonResult> { let (per_page, offset) = paginate(pagination); let mut tx = user_db.clone().begin(&authed).await?; @@ -110,9 +110,27 @@ async fn list_conversations( let conversations = sqlx::query_as::(&sql) .fetch_all(&mut *tx) .await?; + let ids: Vec = conversations.iter().map(|c| c.id).collect(); + let mut running = running_turns(&mut *tx, &ids).await?; tx.commit().await?; - Ok(Json(conversations)) + Ok(Json( + conversations + .into_iter() + .map(|conversation| ListedConversation { + running_turn: running.remove(&conversation.id), + conversation, + }) + .collect(), + )) +} + +#[derive(Serialize)] +pub struct ListedConversation { + #[serde(flatten)] + pub conversation: FlowConversation, + /// Lets a chat that opens after a turn started follow it, and tell which chats are busy. + pub running_turn: Option, } async fn delete_conversation( diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 60b3c2ad25..84a80b828b 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -14923,6 +14923,11 @@ paths: schema: type: string format: uuid + "409": + description: >- + Chat-enabled flow only: the conversation named by `memory_id` is still answering + a message. The body is JSON: `{ "error": string, "running_turn": { "job_id", + "user_seq" } }`. /w/{workspace}/jobs/run/fv/{version}: post: @@ -15473,6 +15478,11 @@ paths: schema: type: string format: uuid + "409": + description: >- + Chat-enabled flow only: the conversation named by `memory_id` is still answering + a message. The body is JSON: `{ "error": string, "running_turn": { "job_id", + "user_seq" } }`. /w/{workspace}/jobs/run_wait_result/preview_flow: post: @@ -28363,6 +28373,23 @@ components: is_test: type: boolean description: Started from the flow editor's test panel rather than a deployed run + running_turn: + type: object + nullable: true + description: >- + The turn the conversation is still answering, set by the list endpoint: its + newest user message, while the flow run it started is queued or running. A + run into this conversation is refused with 409 until the turn ends. + required: [job_id, user_seq] + properties: + job_id: + type: string + format: uuid + description: The flow run of the turn + user_seq: + type: integer + format: int64 + description: created_seq of the user message that started the turn FlowConversationMessage: type: object diff --git a/backend/windmill-common/src/flow_conversations.rs b/backend/windmill-common/src/flow_conversations.rs index 1abe82e5a7..c767385594 100644 --- a/backend/windmill-common/src/flow_conversations.rs +++ b/backend/windmill-common/src/flow_conversations.rs @@ -65,7 +65,9 @@ pub async fn get_or_create_conversation_with_id( is_test: bool, ) -> Result { if let Some(existing) = lock_conversation(tx, w_id, conversation_id).await? { - return same_kind(existing, is_test); + let existing = same_kind(existing, is_test)?; + refuse_running_turn(tx, conversation_id).await?; + return Ok(existing); } // Truncate title to 25 characters max @@ -100,7 +102,70 @@ pub async fn get_or_create_conversation_with_id( "conversation {conversation_id} belongs to another workspace" )) })?; - same_kind(existing, is_test) + let existing = same_kind(existing, is_test)?; + refuse_running_turn(tx, conversation_id).await?; + Ok(existing) +} + +/// The turn a conversation is still answering: its newest user message, while the flow run +/// that message started is still queued or running. +#[derive(Serialize, Debug, Clone, Copy)] +pub struct RunningTurn { + pub job_id: Uuid, + /// `created_seq` of the user message that started the turn. + pub user_seq: i64, +} + +/// One running turn per conversation holds its agent memory; a second run would write the +/// same memory concurrently. Checked under the conversation's row lock, so two runs sent at +/// once cannot both pass: the second waits, then sees the first's message and queued job. +async fn refuse_running_turn( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + conversation_id: Uuid, +) -> Result<()> { + let Some(turn) = running_turns(&mut **tx, &[conversation_id]) + .await? + .remove(&conversation_id) + else { + return Ok(()); + }; + // A JSON body, so a chat client can follow the running turn instead of failing. + Err(crate::error::Error::Generic( + axum::http::StatusCode::CONFLICT, + serde_json::json!({ + "error": "this conversation is still answering a message; wait for it to finish or stop it before sending another", + "running_turn": turn, + }) + .to_string(), + )) +} + +pub async fn running_turns<'e, E: sqlx::PgExecutor<'e>>( + executor: E, + conversation_ids: &[Uuid], +) -> Result> { + let rows = sqlx::query_as::<_, (Uuid, Uuid, i64)>( + "SELECT c.id, u.job_id, u.created_seq + FROM unnest($1::uuid[]) AS c(id) + CROSS JOIN LATERAL ( + SELECT job_id, created_seq + FROM flow_conversation_message + WHERE conversation_id = c.id AND message_type = 'user' + ORDER BY created_seq DESC + LIMIT 1 + ) u + WHERE u.job_id IS NOT NULL + AND EXISTS (SELECT 1 FROM v2_job_queue q WHERE q.id = u.job_id)", + ) + .bind(conversation_ids) + .fetch_all(executor) + .await?; + Ok(rows + .into_iter() + .map(|(conversation_id, job_id, user_seq)| { + (conversation_id, RunningTurn { job_id, user_seq }) + }) + .collect()) } /// `memory_id` is the caller's to choose, so a preview run could name a deployed diff --git a/chat-sdk/README.md b/chat-sdk/README.md index ba76d1d507..8ebff41d4d 100644 --- a/chat-sdk/README.md +++ b/chat-sdk/README.md @@ -233,6 +233,23 @@ own (`'deployed'`, the server's default) or both (`'all'`); each `Conversation` stops following the current answer; the flow keeps running and, with server history, its answer is there when you come back. +A conversation answers one message at a time, and Windmill enforces it: a message sent +while its previous turn still runs (from another tab, or before a reload) is refused. +`sendMessage` then rejects with a `TurnRunningError` and shows nothing of the message. +With server history, a listed `Conversation` also carries `runningTurn` while it is +answering. Either way, `resumeTurn(turn)` follows that turn in the selected +conversation: its answer streams in from the start and the turn finishes as if it had +been sent here, after which the message can be sent again. + +```ts +await chat.selectConversation(id) +const running = chat.getState().conversations.find((c) => c.id === id)?.runningTurn +if (running) await chat.resumeTurn(running) +``` + +A dropped connection to the answer is retried; when it keeps failing, the chat stops +streaming and waits for the flow's result instead, so the turn still ends. + ## History Windmill stores every conversation of a chat-mode flow, and each Windmill user sees diff --git a/chat-sdk/src/api.ts b/chat-sdk/src/api.ts index 4bba733196..df97bdf506 100644 --- a/chat-sdk/src/api.ts +++ b/chat-sdk/src/api.ts @@ -13,13 +13,53 @@ export interface WindmillChatApiOptions { export class WindmillApiError extends Error { constructor( message: string, - readonly status: number + readonly status: number, + /** The response body, as the server sent it. */ + readonly body?: string ) { super(message) this.name = 'WindmillApiError' } } +/** The turn a conversation is still answering, as the server reports it. */ +export interface RunningTurn { + jobId: string + /** `created_seq` of the user message that started the turn. */ + userSeq: number +} + +/** + * A message was sent to a conversation whose turn is still running. The server refuses + * it (409) so two runs never write one agent memory; `turn` is the run to follow instead. + */ +export class TurnRunningError extends Error { + constructor( + message: string, + readonly turn: RunningTurn + ) { + super(message) + this.name = 'TurnRunningError' + } +} + +/** + * The running turn named by a run's 409 body, or undefined for any other body. Exported + * for a custom `run` that calls Windmill through its own client: it rethrows the refusal + * as a `TurnRunningError` so the chat can follow the running turn. + */ +export function turnRunningError(body: string): TurnRunningError | undefined { + try { + const parsed = JSON.parse(body) as { error?: unknown; running_turn?: FlowConversation['running_turn'] } + const turn = parsed?.running_turn + if (!turn || typeof turn.job_id !== 'string' || typeof turn.user_seq !== 'number') return undefined + const message = typeof parsed.error === 'string' ? parsed.error : 'this conversation is still answering a message' + return new TurnRunningError(message, { jobId: turn.job_id, userSeq: turn.user_seq }) + } catch { + return undefined + } +} + export interface FlowConversation { id: string workspace_id: string @@ -30,6 +70,8 @@ export interface FlowConversation { created_by: string /** Started from the flow editor's test panel rather than a deployed run. */ is_test: boolean + /** Set by the list: the turn this conversation is still answering. */ + running_turn?: { job_id: string; user_seq: number } | null } /** @@ -103,18 +145,27 @@ export class WindmillChatApi { this.#pollDelayMs = options.pollDelayMs } - /** Starts a turn: runs the flow with `memory_id` set to the conversation id. Returns the job id. */ + /** + * Starts a turn: runs the flow with `memory_id` set to the conversation id. Returns the + * job id. Throws `TurnRunningError` when the conversation is still answering. + */ async runFlow( flowPath: string, args: Record, options: { memoryId: string; signal?: AbortSignal } ): Promise { - const res = await this.#request(`jobs/run/f/${encodePath(flowPath)}`, { - method: 'POST', - query: { memory_id: options.memoryId, skip_preprocessor: 'true' }, - body: args, - signal: options.signal - }) + let res: Response + try { + res = await this.#request(`jobs/run/f/${encodePath(flowPath)}`, { + method: 'POST', + query: { memory_id: options.memoryId, skip_preprocessor: 'true' }, + body: args, + signal: options.signal + }) + } catch (e) { + if (e instanceof WindmillApiError && e.status === 409) throw turnRunningError(e.body ?? '') ?? e + throw e + } return (await res.text()).trim() } @@ -249,7 +300,8 @@ export class WindmillChatApi { const text = await res.text().catch(() => '') throw new WindmillApiError( `${init.method ?? 'GET'} ${path} failed (${res.status})${text ? `: ${text}` : ''}`, - res.status + res.status, + text ) } return res diff --git a/chat-sdk/src/chat.ts b/chat-sdk/src/chat.ts index 8055ea2234..32ba39697f 100644 --- a/chat-sdk/src/chat.ts +++ b/chat-sdk/src/chat.ts @@ -1,9 +1,11 @@ import { + TurnRunningError, WindmillApiError, WindmillChatApi, type ConversationKind, type FlowConversation, - type FlowConversationMessage + type FlowConversationMessage, + type RunningTurn } from './api' import { resolveConfig, type ResolvedConfig } from './config' import { followJob } from './follow' @@ -35,6 +37,8 @@ const PERSIST_DEBOUNCE_MS = 250 /** Messages persist from spawned tasks that can land just after the flow completes. */ const RECONCILE_ATTEMPTS = 3 const RECONCILE_DELAY_MS = 400 +/** Rows per read of what a turn wrote; a fuller page is read on from its last row. */ +const ROWS_PAGE = 100 interface Turn { controller: AbortController @@ -42,8 +46,6 @@ 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 /** Id of the streaming assistant message; cleared when a tool call ends the round. */ assistantId?: string streamedText: boolean @@ -64,6 +66,8 @@ class ChatImpl implements Chat { /** The kind the caller last listed, so the refresh after a new turn lists the same rows. */ #conversationKind: ConversationKind | undefined #persistTimer: ReturnType | undefined + /** Settles once the selected conversation's first page has been read. */ + #selecting: Promise = Promise.resolve() constructor(options: ChatOptions) { this.#config = resolveConfig(options) @@ -125,9 +129,10 @@ class ChatImpl implements Chat { (c) => c.id === conversationId ) ?? { id: conversationId, title: conversationTitle(content), createdAt: timestamp, updatedAt: timestamp } const touched = { ...conversation, updatedAt: timestamp } + const listed = this.#state.conversations this.#set({ conversationId, - conversations: [touched, ...this.#state.conversations.filter((c) => c.id !== conversationId)], + conversations: [touched, ...listed.filter((c) => c.id !== conversationId)], messages: [ ...this.#state.messages, { id: turn.userMessageId, role: 'user', content, success: true, createdAt: timestamp, pending: true } @@ -143,6 +148,77 @@ class ChatImpl implements Chat { turn.jobId = this.#config.run ? await this.#config.run(args, context) : await this.#api.runFlow(this.#config.flowPath, args, context) + } catch (e) { + try { + // stop() and a conversation switch abort the turn and settle the state themselves. + if (turn.controller.signal.aborted || isAbortError(e)) return + if (e instanceof TurnRunningError) { + // Nothing of this message reached the server: it goes, and the caller decides + // whether to follow the running turn and send it again. + if (this.#turnActive(turn)) { + this.#set({ + messages: this.#state.messages.filter((m) => m.id !== turn.userMessageId), + conversations: listed, + status: 'idle' + }) + } + throw e + } + this.#failTurn(turn, e) + return + } finally { + if (this.#turn === turn) this.#turn = undefined + } + } + await this.#followTurn(turn, isNew) + } + + resumeTurn = async ({ jobId, userSeq }: RunningTurn): Promise => { + const conversationId = this.#state.conversationId + if (!conversationId || this.#turn || this.#state.history !== 'server') return + const turn: Turn = { + controller: new AbortController(), + conversationId, + userMessageId: '', + jobId, + streamedText: false + } + this.#turn = turn + this.#set({ status: 'submitted', error: undefined }) + try { + // The conversation's first page may still be on its way; it would land over the turn. + await this.#selecting + if (!this.#turnActive(turn)) return + // The stream replays the turn from its start, so the rows it already wrote go and + // come back as it replays them. The message that started it stays: it is the turn's + // anchor, and a long turn can have pushed it off the page this chat opened on. + let messages = this.#state.messages.filter((m) => m.seq !== undefined && m.seq <= userSeq) + let user = messages.find((m) => m.seq === userSeq) + if (!user) { + const [row] = await this.#api.listMessages(conversationId, { + afterSeq: userSeq - 1, + perPage: 1, + signal: turn.controller.signal + }) + if (!this.#turnActive(turn)) return + if (row?.created_seq !== userSeq || row.message_type !== 'user') { + throw new Error('windmill-chat: the message that started the running turn is gone') + } + user = fromRow(row) + messages = [...messages, user] + } + turn.userMessageId = user.id + this.#set({ messages }) + } catch (e) { + if (!(turn.controller.signal.aborted || isAbortError(e))) this.#failTurn(turn, e) + if (this.#turn === turn) this.#turn = undefined + return + } + await this.#followTurn(turn, false) + } + + async #followTurn(turn: Turn, isNew: boolean): Promise { + try { const stopPolling = this.#state.history === 'server' ? this.#startPolling(turn) : () => {} let result: unknown try { @@ -193,8 +269,14 @@ class ChatImpl implements Chat { }) } - selectConversation = async (conversationId: string): Promise => { - if (conversationId === this.#state.conversationId) return + selectConversation = (conversationId: string): Promise => { + if (conversationId === this.#state.conversationId) return this.#selecting + const selecting = this.#select(conversationId) + this.#selecting = selecting.catch(() => {}) + return selecting + } + + async #select(conversationId: string): Promise { this.#leaveConversation() this.#page = 1 this.#set({ @@ -323,9 +405,19 @@ class ChatImpl implements Chat { }) if (this.#state.conversationId !== conversationId) return const known = new Set(this.#state.messages.map((m) => m.serverId ?? m.id)) + // Pages count back from the newest row, so rows written since the first page shift + // newer rows into this one; and a resumed turn drops the rows it replays. Only what + // is older than everything held belongs above it. + const oldest = this.#state.messages.reduce( + (min, m) => (m.seq !== undefined && (min === undefined || m.seq < min) ? m.seq : min), + undefined + ) + const older = rows + .map(fromRow) + .filter((m) => !known.has(m.id) && (oldest === undefined || m.seq! < oldest)) this.#page = page this.#set({ - messages: [...rows.map(fromRow).filter((m) => !known.has(m.id)), ...this.#state.messages], + messages: [...older, ...this.#state.messages], hasMoreMessages: rows.length === this.#config.pageSize }) } finally { @@ -451,8 +543,6 @@ class ChatImpl implements Chat { async #finishTurn(turn: Turn, result: unknown, isNew: boolean): Promise { 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) { @@ -503,11 +593,7 @@ class ChatImpl implements Chat { for (let attempt = 1; attempt <= RECONCILE_ATTEMPTS; attempt++) { let rows: FlowConversationMessage[] try { - rows = await this.#api.listMessages(turn.conversationId, { - afterSeq: this.#lastSeq(), - perPage: 100, - signal: turn.controller.signal - }) + rows = await this.#rowsAfterLastSeq(turn.conversationId, turn.controller.signal) } catch (e) { if (isAbortError(e)) throw e if (this.#fallBackToLocal(e)) return false @@ -534,48 +620,24 @@ 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. Rows carrying a job id belong to the turn when the - * job is one of the turn's, which leaves out an earlier turn whose job outlived - * `stop()` (a token without `jobs:write` cannot cancel it); a tool row without one - * (an MCP call runs inside the agent step) belongs to whatever turn is under way. + * 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 + * 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 ownJob = (m: ChatMessage) => - turn.jobIds === undefined || (m.jobId === undefined ? m.role === 'tool' : turn.jobIds.has(m.jobId)) + const userSeq = messages[from]?.seq let latest: ChatMessage | undefined - for (let i = from + 1; i < messages.length; i++) { - const m = messages[i] - if (m.seq === undefined || m.role === 'user' || !ownJob(m)) continue + messages.forEach((m, i) => { + if (m.seq === undefined || m.role === 'user') 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 | 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) @@ -601,11 +663,7 @@ class ChatImpl implements Chat { } if (stopped) return try { - const rows = await this.#api.listMessages(turn.conversationId, { - afterSeq: this.#lastSeq(), - perPage: 100, - signal - }) + const rows = await this.#rowsAfterLastSeq(turn.conversationId, signal) if (!stopped && this.#turnActive(turn)) this.#mergeRows(rows) } catch { // transient; the completion reconciliation catches up @@ -618,11 +676,20 @@ class ChatImpl implements Chat { } } + /** Every row created after the newest one held, however many pages that takes. */ + async #rowsAfterLastSeq(conversationId: string, signal?: AbortSignal): Promise { + const rows: FlowConversationMessage[] = [] + let afterSeq = this.#lastSeq() + while (true) { + const page = await this.#api.listMessages(conversationId, { afterSeq, perPage: ROWS_PAGE, signal }) + rows.push(...page) + if (page.length < ROWS_PAGE) return rows + afterSeq = page[page.length - 1].created_seq + } + } + async #syncFromServer(conversationId: string): Promise { - const rows = await this.#api.listMessages(conversationId, { - afterSeq: this.#lastSeq(), - perPage: 100 - }) + const rows = await this.#rowsAfterLastSeq(conversationId) if (this.#turn || this.#state.conversationId !== conversationId) return this.#mergeRows(rows) this.#set({ messages: finalized(this.#state.messages) }) @@ -766,7 +833,10 @@ function fromConversation(row: FlowConversation): Conversation { title: row.title ?? undefined, createdAt: row.created_at, updatedAt: row.updated_at, - isTest: row.is_test + isTest: row.is_test, + runningTurn: row.running_turn + ? { jobId: row.running_turn.job_id, userSeq: row.running_turn.user_seq } + : undefined } } diff --git a/chat-sdk/src/follow.ts b/chat-sdk/src/follow.ts index 50f101393e..3f3503107a 100644 --- a/chat-sdk/src/follow.ts +++ b/chat-sdk/src/follow.ts @@ -1,8 +1,12 @@ -import type { WindmillChatApi } from './api' +import { WindmillApiError, type WindmillChatApi } from './api' import { createStreamEventParser, type AgentStreamEvent } from './stream' -import { abortError, sleep } from './utils' +import { abortError, isAbortError, sleep } from './utils' const RECONNECT_DELAY_MS = 300 +const MAX_RECONNECT_DELAY_MS = 5000 +/** Consecutive failed connections before the job is polled instead. */ +const MAX_CONNECTION_FAILURES = 3 +const RESULT_POLL_MS = 2000 export type FollowEvent = /** Agent events decoded from the job's result stream; empty when a chunk ended mid-line. */ @@ -18,6 +22,10 @@ export type FollowEvent = * The offset indexes the stream of one sub-job (`flow_stream_job_id`, the flow's * streaming step). A retried step gets a new one, so when the id changes the * offset is dropped and the connection reopened from that sub-job's start. + * + * A connection that fails (a proxy restarting, the network dropping) is retried with + * backoff. The run is still going, so after a few failures in a row the job's result is + * polled instead: the rest of the answer is not streamed, but the turn still ends. */ export async function* followJob( api: WindmillChatApi, @@ -27,45 +35,76 @@ export async function* followJob( let parser = createStreamEventParser() let offset = options.streamOffset let streamJobId: string | undefined - while (true) { + let failures = 0 + while (failures < MAX_CONNECTION_FAILURES) { let reopen = false - for await (const update of api.streamJob(jobId, { streamOffset: offset, signal: options.signal })) { - if (update.type === 'ping') continue - if (update.type === 'timeout') { - reopen = true - break - } - if (update.type === 'error') throw new Error(update.error) - if (update.type === 'notfound') throw new Error(`Job ${jobId} not found`) - if (update.flow_stream_job_id && update.flow_stream_job_id !== streamJobId) { - const switched = streamJobId !== undefined && offset !== undefined - streamJobId = update.flow_stream_job_id - if (switched) { - // This connection skipped the new sub-job's first chunks: start it over. - offset = undefined - options.onOffset?.(undefined) - parser = createStreamEventParser() + try { + for await (const update of api.streamJob(jobId, { streamOffset: offset, signal: options.signal })) { + failures = 0 + if (update.type === 'ping') continue + if (update.type === 'timeout') { reopen = true break } + if (update.type === 'error') throw new Error(update.error) + if (update.type === 'notfound') throw new Error(`Job ${jobId} not found`) + if (update.flow_stream_job_id && update.flow_stream_job_id !== streamJobId) { + const switched = streamJobId !== undefined && offset !== undefined + streamJobId = update.flow_stream_job_id + if (switched) { + // This connection skipped the new sub-job's first chunks: start it over. + offset = undefined + options.onOffset?.(undefined) + parser = createStreamEventParser() + reopen = true + break + } + } + if (update.stream_offset !== undefined) { + offset = update.stream_offset + options.onOffset?.(offset) + } + if (update.new_result_stream) { + yield { type: 'stream', events: parser.push(update.new_result_stream) } + } + if (update.completed) { + const rest = parser.flush() + if (rest.length > 0) yield { type: 'stream', events: rest } + yield { type: 'completed', result: update.only_result } + return + } } - if (update.stream_offset !== undefined) { - offset = update.stream_offset - options.onOffset?.(offset) - } - if (update.new_result_stream) { - yield { type: 'stream', events: parser.push(update.new_result_stream) } - } - if (update.completed) { - const rest = parser.flush() - if (rest.length > 0) yield { type: 'stream', events: rest } - yield { type: 'completed', result: update.only_result } - return - } + } catch (e) { + if (options.signal?.aborted || isAbortError(e) || !isConnectionFailure(e)) throw e + failures++ + if (failures >= MAX_CONNECTION_FAILURES) break + await sleep(Math.min(RECONNECT_DELAY_MS * 2 ** failures, MAX_RECONNECT_DELAY_MS), options.signal) + continue } if (options.signal?.aborted) throw abortError() // The server closes the connection after its timeout; a dropped connection looks // the same minus the event. Either way the offset lets the next one resume. if (!reopen) await sleep(RECONNECT_DELAY_MS, options.signal) } + while (true) { + await sleep(RESULT_POLL_MS, options.signal) + try { + const { completed, result } = await api.getCompletedResult(jobId, options.signal) + if (completed) { + yield { type: 'completed', result } + return + } + } catch (e) { + if (options.signal?.aborted || isAbortError(e) || !isConnectionFailure(e)) throw e + } + } +} + +/** + * A failure that says nothing about the job: the request never reached Windmill, or a + * gateway in front of it answered. A 4xx from Windmill itself (not found, refused) does. + */ +function isConnectionFailure(e: unknown): boolean { + if (e instanceof WindmillApiError) return e.status >= 500 || e.status === 0 + return e instanceof TypeError } diff --git a/chat-sdk/src/index.ts b/chat-sdk/src/index.ts index 1c18a5924b..60712f9c09 100644 --- a/chat-sdk/src/index.ts +++ b/chat-sdk/src/index.ts @@ -3,7 +3,10 @@ export { detectRawApp, type RawAppContext } from './config' export { WindmillChatApi, WindmillApiError, + TurnRunningError, + turnRunningError, readServerSentEvents, + type RunningTurn, type WindmillChatApiOptions, type ConversationKind, type FlowConversation, diff --git a/chat-sdk/src/react.ts b/chat-sdk/src/react.ts index 33193842f2..5b913ebd6b 100644 --- a/chat-sdk/src/react.ts +++ b/chat-sdk/src/react.ts @@ -6,6 +6,7 @@ export type UseWindmillChat = ChatState & Pick< Chat, | 'sendMessage' + | 'resumeTurn' | 'stop' | 'newConversation' | 'selectConversation' @@ -61,6 +62,7 @@ export function useWindmillChat(options: ChatOptions): UseWindmillChat { chat, sendMessage: (text, options) => chat.sendMessage(text, { ...options, inputs: { ...latest.current.inputs, ...options?.inputs } }), + resumeTurn: chat.resumeTurn, stop: chat.stop, newConversation: chat.newConversation, selectConversation: chat.selectConversation, diff --git a/chat-sdk/src/types.ts b/chat-sdk/src/types.ts index 5e0d673bd9..88176989a1 100644 --- a/chat-sdk/src/types.ts +++ b/chat-sdk/src/types.ts @@ -1,3 +1,5 @@ +import type { RunningTurn } from './api' + export type ChatRole = 'user' | 'assistant' | 'tool' | 'system' /** @@ -57,6 +59,11 @@ export interface Conversation { * server has listed the conversation; unset for one only this client has seen. */ isTest?: boolean + /** + * The turn the conversation was still answering when the server listed it: started in + * another tab, or before this chat was created. Pass it to `resumeTurn` to follow it. + */ + runningTurn?: RunningTurn } export interface ChatState { @@ -127,8 +134,20 @@ export interface Chat { getState(): ChatState /** Calls `listener` now and on every change; returns the unsubscribe function (Svelte store contract). */ subscribe(listener: (state: ChatState) => void): () => void - /** Sends a message in the current conversation, starting one when there is none. Resolves when the answer is complete. */ + /** + * Sends a message in the current conversation, starting one when there is none. Resolves + * when the answer is complete. Rejects with `TurnRunningError`, and shows nothing of the + * message, when the conversation is still answering one sent elsewhere: follow that turn + * with `resumeTurn`, then send again. + */ sendMessage(text: string, options?: { inputs?: Record }): Promise + /** + * Follows a turn of the current conversation that this chat did not start, as named by + * `Conversation.runningTurn` or a `TurnRunningError`: its answer streams into `messages` + * from the start and the turn finishes like one sent here. Server history only; resolves + * when the answer is complete, at once when a turn is already being followed. + */ + resumeTurn(turn: RunningTurn): Promise /** Stops following the answer and asks Windmill to cancel the run. */ stop(): Promise newConversation(): void diff --git a/chat-sdk/test/chat.test.ts b/chat-sdk/test/chat.test.ts index 89c4565942..b84237e3f3 100644 --- a/chat-sdk/test/chat.test.ts +++ b/chat-sdk/test/chat.test.ts @@ -1,4 +1,5 @@ import { describe, expect, test } from 'bun:test' +import { TurnRunningError } from '../src/api' import { createChat } from '../src/chat' import type { ChatOptions } from '../src/types' import { fetchMock, json, memoryStorage, messageRow, ndjson, sse, sseTimed, text, type Route } from './support' @@ -724,49 +725,87 @@ describe('createChat with server history', () => { ]) }) - test('a late answer from a stopped job is not taken as the next turn answer', async () => { - let jobs = 0 - let reads = 0 + test('a message refused because a turn is running leaves nothing behind and names that turn', async () => { const { fetch } = fetchMock( - (c) => (c.method === 'POST' && c.url.pathname.includes('/jobs/run/f/') ? text(`job-${++jobs}`) : undefined), - // job-1 never completes: the connection just ends, so the turn keeps waiting. - (c) => (c.url.pathname.endsWith('/getupdate_sse/job-1') ? sse([{ type: 'update' }]) : undefined), (c) => - c.url.pathname.endsWith('/getupdate_sse/job-2') - ? sse([{ type: 'update', completed: true, only_result: { windmill_chat_answer: 'second answer' } }]) + c.method === 'POST' && c.url.pathname === `/api/w/ws/jobs/run/f/${FLOW}` + ? text(JSON.stringify({ error: 'still answering', running_turn: { job_id: 'job-9', user_seq: 41 } }), 409) : undefined, - // The run-only token cannot cancel: job-1 keeps running after stop(). - (c) => (c.url.pathname.includes('/queue/cancel/') ? text('forbidden', 400) : undefined), + (c) => (c.url.pathname.endsWith('/messages') ? json([messageRow(40, 'user', 'earlier')]) : undefined) + ) + const chat = createChat(options({}, fetch)) + await chat.selectConversation('conv') + const refused = await chat.sendMessage('again').catch((e) => e) + expect(refused).toBeInstanceOf(TurnRunningError) + expect(refused.turn).toEqual({ jobId: 'job-9', userSeq: 41 }) + const state = chat.getState() + expect(state.messages.map((m) => m.content)).toEqual(['earlier']) + expect(state.status).toBe('idle') + expect(state.conversations).toEqual([]) + }) + + test('resuming a turn whose message is off the first page replays it without duplicating rows', async () => { + const { fetch, calls } = fetchMock( (c) => - c.url.pathname.endsWith('/jobs_u/get/job-2') - ? json({ flow_status: { modules: [{ job: 'step-2' }] } }) + c.url.pathname === streamPath && !c.url.searchParams.has('stream_offset') + ? sse([ + { + type: 'update', + new_result_stream: ndjson( + { type: 'tool_call', call_id: 'c1', function_name: 'lookup' }, + { type: 'tool_result', call_id: 'c1', function_name: 'lookup', result: '1', success: true }, + { type: 'token_delta', content: 'Done' } + ), + stream_offset: 3, + completed: true, + only_result: { output: 'Done', messages: [] } + } + ]) + : undefined, + (c) => { + if (!c.url.pathname.endsWith('/messages')) return undefined + const after = c.url.searchParams.get('after_seq') + // The first page holds only what the running turn wrote so far. + if (after === null) return json([messageRow(51, 'tool', 'Used lookup tool')]) + if (after === '49') return json([messageRow(50, 'user', 'hi')]) + return json([messageRow(51, 'tool', 'Used lookup tool'), messageRow(52, 'assistant', 'Done')]) + } + ) + const chat = createChat(options({}, fetch)) + void chat.selectConversation('conv') + await chat.resumeTurn({ jobId: 'job-1', userSeq: 50 }) + expect(chat.getState().messages.map((m) => [m.serverId, m.role, m.content, m.pending])).toEqual([ + ['row-50', 'user', 'hi', false], + ['row-51', 'tool', 'Used lookup tool', false], + ['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('a stream that keeps failing hands the turn to polling the job', async () => { + const { fetch } = fetchMock( + run, + (c) => (c.url.pathname === streamPath ? text('bad gateway', 502) : undefined), + (c) => + c.url.pathname.endsWith('/get_result_maybe/job-1') + ? json({ completed: true, success: true, result: { windmill_chat_answer: 'polled' } }) : undefined, - // Read 1 is stop()'s sync; the stopped job's answer lands after the second user row. (c) => c.url.pathname.endsWith('/messages') - ? json( - ++reads === 1 - ? [messageRow(71, 'user', 'first')] - : reads === 2 - ? [messageRow(72, 'user', 'second'), messageRow(73, 'assistant', 'first answer, late', { job_id: 'step-1' })] - : [messageRow(74, 'assistant', 'second answer', { job_id: 'step-2' })] - ) + ? json([messageRow(61, 'user', 'hi'), messageRow(62, 'assistant', 'polled')]) : undefined, (c) => (c.url.pathname === '/api/w/ws/flow_conversations/list' ? json([]) : undefined) ) const chat = createChat(options({}, fetch)) - const first = chat.sendMessage('first') - await new Promise((r) => setTimeout(r, 50)) - await chat.stop() - await first - 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'] + await chat.sendMessage('hi') + const state = chat.getState() + expect(state.status).toBe('idle') + expect(state.messages.map((m) => [m.role, m.content, m.success])).toEqual([ + ['user', 'hi', true], + ['assistant', 'polled', true] ]) - }) + }, 15000) test('a failure handler answer is attributed to the turn', async () => { let reads = 0 diff --git a/frontend/src/lib/components/FlowPreviewContent.svelte b/frontend/src/lib/components/FlowPreviewContent.svelte index 32b46cb2be..8c66ec81b5 100644 --- a/frontend/src/lib/components/FlowPreviewContent.svelte +++ b/frontend/src/lib/components/FlowPreviewContent.svelte @@ -1,5 +1,6 @@ + +{#if count > 0} + + {count > 9 ? '9+' : count} + +{/if} + + diff --git a/frontend/src/lib/components/flows/conversations/FlowChat.svelte b/frontend/src/lib/components/flows/conversations/FlowChat.svelte index 24c41a2ca9..4ca9d7ec0f 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChat.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowChat.svelte @@ -1,12 +1,21 @@
- {#if chat && chatState} + {#if listChat && listState && pool && poolState && shown} {#if !hideSidebar} {/if} - - {#key chat} + {#key shown} /** The flow's modules, read for the provider wiring of its AI agent steps. */ @@ -38,6 +43,8 @@ let { chat, + chatHost: chatHostProp, + isTest = undefined, deploymentInProgress = false, additionalInputsSchema, flowModules, @@ -146,30 +153,22 @@ showInputsModal = true } - // The host follows the chat it was built on for the life of this component: FlowChat - // remounts the interface under `{#key chat}`, so a later value of the prop never reaches it. - const chatHost = new FlowChatViewHost( - untrack(() => chat), - { - additionalInputs: () => (additionalInputsSchema ? { ...runInputs } : undefined), - workspace: () => workspace, - sendDisabled: () => deploymentInProgress || !!modelGap || !!wrongKindReason - } - ) + const chatHost = untrack(() => chatHostProp) + chatHost.setOptions({ + additionalInputs: () => (additionalInputsSchema ? { ...runInputs } : undefined), + workspace: () => workspace, + sendDisabled: () => deploymentInProgress || !!modelGap || !!wrongKindReason + }) setChatViewHost(chatHost) // A chat of the other kind can be read from here but not added to: the server refuses a // preview run into a deployed conversation and the reverse, so the composer says why first. const wrongKindReason = $derived.by(() => { - const { conversationId, conversations } = chatHost.state - const open = conversations.find((c) => c.id === conversationId) - if (open?.isTest === undefined || open.isTest === (conversationKind === 'test')) - return undefined - return open.isTest + if (isTest === undefined || isTest === (conversationKind === 'test')) return undefined + return isTest ? 'This chat was run from the flow editor. Start a new chat to continue here.' : 'This chat belongs to the deployed flow. Start a new chat to test.' }) - onDestroy(() => chatHost.dispose()) // What the Configure-inputs modal asks for: every flow input the composer does not // edit itself. diff --git a/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte b/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte index 49efbbc1fa..0fcc776227 100644 --- a/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte @@ -5,11 +5,13 @@ Plus, Trash2, Pen, + PencilLine, Filter, PanelLeftClose, PanelLeftOpen } from 'lucide-svelte' - import CountBadge from '$lib/components/common/badge/CountBadge.svelte' + import UnreadCountBadge from '$lib/components/common/badge/UnreadCountBadge.svelte' + import SessionStatusDot from '$lib/components/sessions/SessionStatusDot.svelte' import InfiniteList from '$lib/components/InfiniteList.svelte' import DropdownV2 from '$lib/components/DropdownV2.svelte' import Popover from '$lib/components/meltComponents/Popover.svelte' @@ -21,11 +23,16 @@ import { twMerge } from 'tailwind-merge' import { fade } from 'svelte/transition' import { tick, untrack } from 'svelte' - import type { Chat, ChatState, Conversation, ConversationKind } from 'windmill-chat' + import type { Chat, Conversation, ConversationKind } from 'windmill-chat' + import type { FlowChatPool, FlowChatPoolState } from './flowChatPool' + import type { FlowChatViewHost } from './flowChatViewHost.svelte' interface Props { - chat: Chat - chatState: ChatState + /** Lists, renames and deletes the flow's conversations; runs no turn itself. */ + listChat: Chat + /** The conversations' own chats: which one is shown, and what each is doing. */ + pool: FlowChatPool + poolState: FlowChatPoolState /** * Which conversations the list holds at first. The editor shows its own test chats, * since testing is what happens there; a deployed flow shows the chats its users @@ -40,7 +47,13 @@ canFilterKind?: boolean } - let { chat, chatState, defaultKind = 'deployed', canFilterKind = false }: Props = $props() + let { + listChat, + pool, + poolState, + defaultKind = 'deployed', + canFilterKind = false + }: Props = $props() let expanded = $state(false) let list = $state(undefined) @@ -58,13 +71,13 @@ let renameDraft = $state('') let renameInput = $state(undefined) - const turnInFlight = $derived( - chatState.status === 'submitted' || chatState.status === 'streaming' + const totalUnread = $derived( + Object.values(poolState.unread).reduce((total, count) => total + count, 0) ) $effect(() => { const l = list - const c = chat + const c = listChat if (!l) return untrack(() => { // Every load goes through here, the first one and infinite scroll included. A @@ -79,6 +92,7 @@ deletingId = id try { await c.deleteConversation(id) + pool.forget(id) sendUserToast('Conversation deleted successfully') } catch (error) { console.error('Failed to delete conversation:', error) @@ -103,10 +117,10 @@ await list?.loadData('forceRefresh') } - const draftShown = $derived(draft && !items.some((c) => c.id === chatState.conversationId)) + const draftShown = $derived(draft && poolState.selectedId === undefined) function newChat() { - chat.newConversation() + pool.newChat() draft = true } @@ -118,17 +132,16 @@ /** * Narrow the list to one kind of chat and reload it. The open conversation goes with it - * when it is not of the new kind: the composer sends into whatever is selected, and a + * when it is not of the new kind: the composer sends into whatever is shown, and a * conversation keeps the kind it was created with, so a turn sent into one the list no - * longer shows would be stored where nothing here lists it. + * longer shows would be stored where nothing here lists it. A turn running in it goes on. */ async function setKind(next: ConversationKind) { - // A turn writes into the open conversation, which a kind that excludes it would close. - if (next === kind || turnInFlight) return + if (next === kind) return kind = next - const open = items.find((c) => c.id === chatState.conversationId) + const open = items.find((c) => c.id === poolState.selectedId) const stillListed = open === undefined || next === 'all' || (next === 'test') === open.isTest - if (!stillListed) chat.newConversation() + if (!stillListed) pool.newChat() await list?.loadData('forceRefresh') } @@ -149,11 +162,11 @@ const current = items.find((c) => c.id === id) if (!current || title === '' || title === current.title) return try { - await chat.renameConversation(id, title) + await listChat.renameConversation(id, title) // The list holds its own rows, loaded through the loader: patched rather than // reloaded, so the row keeps its place without a round trip. The title is read // back from the chat, which holds it as the server stored it (a long one is cut). - const stored = chat.getState().conversations.find((c) => c.id === id)?.title ?? title + const stored = listChat.getState().conversations.find((c) => c.id === id)?.title ?? title items = items.map((c) => (c.id === id ? { ...c, title: stored } : c)) } catch (error) { console.error('Failed to rename conversation:', error) @@ -177,8 +190,34 @@ function getConversationTitle(conversation: Conversation): string { return conversation.title || `Conversation ${conversation.createdAt.slice(0, 10)}` } + + /** The session sidebar's dot vocabulary, which knows no `queued`: that is its own mark. */ + function dotStatus(conversationId: string): 'streaming' | 'error' | 'idle' { + const activity = poolState.activity[conversationId] + return activity === 'running' ? 'streaming' : activity === 'error' ? 'error' : 'idle' + } +{#snippet statusDot(conversation: Conversation)} + + + {#snippet resting()} + + {/snippet} + +{/snippet} +
+ {#snippet trigger()} @@ -233,10 +267,7 @@ unifiedSize="md" variant="subtle" startIcon={{ icon: Filter }} - disabled={turnInFlight} - title={turnInFlight - ? 'Wait for the current answer to change which chats are listed' - : `Filter conversations · ${KIND_LABELS[kind]}`} + title={`Filter conversations · ${KIND_LABELS[kind]}`} iconOnly /> {/snippet} @@ -245,7 +276,6 @@ setKind(next as ConversationKind)} - disabled={turnInFlight} noWFull > {#snippet children({ item })} @@ -274,11 +304,19 @@ unifiedSize="md" startIcon={{ icon: MessageCircle }} onClick={() => (expanded = true)} - title="{items.length} conversation{items.length !== 1 ? 's' : ''}" + title="{items.length} conversation{items.length !== 1 ? 's' : ''}{totalUnread > 0 + ? `, ${totalUnread} unread` + : ''}" variant="subtle" btnClasses="w-fit px-2 relative" > - + +
{/if} @@ -299,7 +337,6 @@ onClick={(e) => { e?.stopPropagation() draft = false - chat.newConversation() }} title="Discard draft" destructive @@ -314,7 +351,7 @@
{:else} + {@const unread = poolState.unread[conversation.id] ?? 0} + {@const queued = !!pool.get(conversation.id)?.host.queuedMessage} diff --git a/frontend/src/lib/components/sessions/SessionStatusDot.svelte b/frontend/src/lib/components/sessions/SessionStatusDot.svelte index 646163bcaf..3d37d41d54 100644 --- a/frontend/src/lib/components/sessions/SessionStatusDot.svelte +++ b/frontend/src/lib/components/sessions/SessionStatusDot.svelte @@ -8,8 +8,18 @@ let { status, isFork, - forkDetached = false - }: { status: SessionChatStatus; isFork: boolean; forkDetached?: boolean } = $props() + forkDetached = false, + resting, + restingTitle + }: { + status: SessionChatStatus + isFork: boolean + forkDetached?: boolean + /** What the slot shows when there is no live signal. Sessions leave it unset and get + * the workspace/fork mark below; another list passes its own resting mark. */ + resting?: import('svelte').Snippet + restingTitle?: string + } = $props() const statusTooltip: Record = { idle: 'No chat activity', @@ -39,7 +49,9 @@ : 'Root workspace session' ) - const title = $derived(liveOverride ? statusTooltip[status] : persistentTitle) + const title = $derived( + liveOverride ? statusTooltip[status] : (restingTitle ?? persistentTitle) + ) @@ -55,6 +67,8 @@ {:else if status === 'error'} + {:else if resting} + {@render resting()} {:else if isFork} {#if forkDetached}