feat: run turns in several flow chat conversations at once

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

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

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<Postgres>) -> windmill_common::error::Result<Uuid> {
+    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<Postgres>,
+) -> 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(())
+}

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<String>,
     Query(pagination): Query<Pagination>,
     Query(query): Query<ListConversationsQuery>,
-) -> JsonResult<Vec<FlowConversation>> {
+) -> JsonResult<Vec<ListedConversation>> {
     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::<Postgres, FlowConversation>(&sql)
         .fetch_all(&mut *tx)
         .await?;
+    let ids: Vec<Uuid> = 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<RunningTurn>,
 }
 
 async fn delete_conversation(

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

backend/windmill-common/src/flow_conversations.rs

@@ -65,7 +65,9 @@ pub async fn get_or_create_conversation_with_id(
     is_test: bool,
 ) -> Result<FlowConversation> {
     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<std::collections::HashMap<Uuid, RunningTurn>> {
+    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

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

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<string, unknown>,
     options: { memoryId: string; signal?: AbortSignal }
   ): Promise<string> {
-    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

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<string>
   /** 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<typeof setTimeout> | undefined
+  /** Settles once the selected conversation's first page has been read. */
+  #selecting: Promise<void> = 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<void> => {
+    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<void> {
+    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<void> => {
-    if (conversationId === this.#state.conversationId) return
+  selectConversation = (conversationId: string): Promise<void> => {
+    if (conversationId === this.#state.conversationId) return this.#selecting
+    const selecting = this.#select(conversationId)
+    this.#selecting = selecting.catch(() => {})
+    return selecting
+  }
+
+  async #select(conversationId: string): Promise<void> {
     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<number | undefined>(
+        (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<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) {
@@ -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<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)
@@ -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<FlowConversationMessage[]> {
+    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<void> {
-    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
   }
 }
 

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
 }

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,

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,

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<string, unknown> }): Promise<void>
+  /**
+   * 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<void>
   /** Stops following the answer and asks Windmill to cancel the run. */
   stop(): Promise<void>
   newConversation(): void

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

frontend/src/lib/components/FlowPreviewContent.svelte

@@ -1,5 +1,6 @@
 <script lang="ts">
 	import {
+		ApiError,
 		type Job,
 		JobService,
 		type RestartedFrom,
@@ -214,9 +215,12 @@
 			}
 			onRunPreview?.(newJobId)
 		} catch (e) {
-			sendUserToast('Could not run preview', true, undefined, e.toString())
 			isRunning = false
 			jobId = undefined
+			// The chat follows the turn its conversation is still answering rather than
+			// reporting a failed run, so the refusal goes back to it.
+			if (conversationId && e instanceof ApiError && e.status === 409) throw e
+			sendUserToast('Could not run preview', true, undefined, e.toString())
 		}
 		schemaFormWithArgPicker?.refreshHistory()
 		return newJobId

frontend/src/lib/components/common/badge/UnreadCountBadge.svelte

@@ -0,0 +1,53 @@
+<script lang="ts">
+	/**
+	 * How many messages have arrived somewhere since it was last read.
+	 *
+	 * Sits in the row's own flow by default, which is where a list uses it; `class` pins it
+	 * to a corner for a caller that has one icon standing for the whole list. Nothing renders
+	 * at zero — an absent badge is what "nothing new" looks like.
+	 */
+	import { twMerge } from 'tailwind-merge'
+
+	interface Props {
+		count: number
+		/** What the count is of, for the label a screen reader reads. */
+		noun?: string
+		/** Positioning for a caller that pins it to a corner rather than letting it sit in
+		 * the row — the collapsed rail, where the count belongs to an icon button. */
+		class?: string
+		/** The 12px form, for a corner where the row-sized badge would crowd the icon. */
+		small?: boolean
+	}
+
+	let { count, noun = 'message', class: className = '', small = false }: Props = $props()
+</script>
+
+{#if count > 0}
+	<span
+		class={twMerge(
+			'unread-badge inline-flex items-center justify-center rounded-full bg-surface-accent-primary text-white font-medium',
+			small ? 'min-w-3 h-3 px-0.5 text-[8px]' : 'min-w-3.5 h-3.5 px-1 text-[9px]',
+			// After the size, not before: tailwind-merge counts a text size as resetting
+			// line-height, so a `leading-*` ahead of one is dropped from the result.
+			'leading-none',
+			className
+		)}
+		aria-label="{count} unread {noun}{count === 1 ? '' : 's'}"
+	>
+		{count > 9 ? '9+' : count}
+	</span>
+{/if}
+
+<style>
+	/*
+	 * Centring a digit by flex centres its *line box*, which is the font's em box — and a
+	 * digit's ink does not sit in the middle of that. Inter reserves descender space a
+	 * figure never uses, so the glyph lands fractionally low; at these sizes that reads as
+	 * the badge being a pixel off. Trimming the box to cap-height and baseline makes the
+	 * ink itself what gets centred. Dropped silently where it is unsupported, which leaves
+	 * the same near-miss as before rather than anything worse.
+	 */
+	.unread-badge {
+		text-box: trim-both cap alphabetic;
+	}
+</style>

frontend/src/lib/components/flows/conversations/FlowChat.svelte

@@ -1,12 +1,21 @@
 <script lang="ts">
 	import { enterpriseLicense, workspaceStore } from '$lib/stores'
 	import { sendUserToast } from '$lib/toast'
-	import { createChat, type Chat, type ChatState } from 'windmill-chat'
+	import {
+		createChat,
+		turnRunningError,
+		WindmillChatApi,
+		type Chat,
+		type ChatOptions,
+		type ChatState
+	} from 'windmill-chat'
 	import FlowConversationsSidebar from './FlowConversationsSidebar.svelte'
 	import FlowChatInterface from './FlowChatInterface.svelte'
 	import { getContext } from 'svelte'
 	import type { FlowEditorContext } from '../types'
-	import type { FlowModule } from '$lib/gen'
+	import { ApiError, type FlowModule } from '$lib/gen'
+	import { FlowChatPool, type FlowChatPoolState } from './flowChatPool'
+	import { FlowChatViewHost } from './flowChatViewHost.svelte'
 
 	interface Props {
 		/**
@@ -53,24 +62,39 @@
 	// The editor may act on a workspace other than the nav store's (AI-session live editor).
 	const workspace = $derived(flowEditorContext?.opWorkspace?.() ?? $workspaceStore)
 
-	let chat = $state<Chat | undefined>(undefined)
-	let chatState = $state<ChatState | undefined>(undefined)
+	// The sidebar lists, renames and deletes through `listChat`; each conversation runs its
+	// turns on its own chat in the pool, so several can answer at once.
+	let listChat = $state<Chat | undefined>(undefined)
+	let listState = $state<ChatState | undefined>(undefined)
+	let pool = $state<FlowChatPool<FlowChatViewHost> | undefined>(undefined)
+	let poolState = $state<FlowChatPoolState | undefined>(undefined)
 	let sidebar = $state<FlowConversationsSidebar | undefined>(undefined)
 
 	$effect(() => {
 		const ws = workspace
 		const flowPath = path
 		if (!ws || !flowPath) return
-		const created = createChat({
+		const baseUrl = window.location.origin
+		const options: ChatOptions = {
 			flowPath,
 			workspace: ws,
-			baseUrl: window.location.origin,
+			baseUrl,
 			history: 'server',
 			// Only an enterprise server honours it; elsewhere it would just log a warning per
 			// poll. The license loads asynchronously, so a cold load may create the chat twice.
 			pollDelayMs: $enterpriseLicense ? 50 : undefined,
 			run: async ({ user_message, ...inputs }, { conversationId }) => {
-				const jobId = await onRunFlow(String(user_message), conversationId, inputs)
+				let jobId: string | undefined
+				try {
+					jobId = await onRunFlow(String(user_message), conversationId, inputs)
+				} catch (e) {
+					// The conversation is still answering a message sent elsewhere; the chat
+					// follows that turn instead of failing this one.
+					if (e instanceof ApiError && e.status === 409) {
+						throw turnRunningError(String(e.body)) ?? e
+					}
+					throw e
+				}
 				if (!jobId) throw new Error('the flow did not start')
 				// The server creates the conversation with the run, so the sidebar can list
 				// it now, whatever becomes of the turn.
@@ -78,15 +102,42 @@
 				return jobId
 			},
 			onError: (error) => sendUserToast('Failed to run flow: ' + error.message, true)
+		}
+		const api = new WindmillChatApi({ baseUrl, workspace: ws })
+		const createdList = createChat(options)
+		const createdPool = new FlowChatPool<FlowChatViewHost>({
+			createChat: () => createChat(options),
+			createHost: (chat) => new FlowChatViewHost(chat),
+			disposeHost: (host) => host.dispose(),
+			hasQueued: (host) => host.queuedMessage !== '',
+			resumeTurn: (host, turn) => host.resumeTurn(turn),
+			isRunFinished: async (jobId) => (await api.getCompletedResult(jobId)).completed
 		})
-		const unsubscribe = created.subscribe((s) => (chatState = s))
-		chat = created
+		const unsubscribeList = createdList.subscribe((s) => {
+			listState = s
+			createdPool.setListed(s.conversations)
+		})
+		const unsubscribePool = createdPool.subscribe((s) => (poolState = s))
+		listChat = createdList
+		pool = createdPool
 		return () => {
-			unsubscribe()
-			created.destroy()
+			unsubscribeList()
+			unsubscribePool()
+			createdPool.destroy()
+			createdList.destroy()
 		}
 	})
 
+	// A new chat keeps its chat once its first turn names the conversation, so the panel is
+	// only remounted when the reader moves to another one.
+	const shown = $derived.by(() => {
+		void poolState?.selectedId
+		return pool?.selected
+	})
+	const shownIsTest = $derived(
+		listState?.conversations.find((c) => c.id === poolState?.selectedId)?.isTest
+	)
+
 	// Derive additional inputs schema (excluding user_message) for chat mode
 	const additionalInputsSchema = $derived.by(() => {
 		const props = inputSchema?.properties ?? {}
@@ -103,21 +154,22 @@
 </script>
 
 <div class="flex border border-gray-200 dark:border-gray-700 rounded-lg overflow-hidden flex-1">
-	{#if chat && chatState}
+	{#if listChat && listState && pool && poolState && shown}
 		{#if !hideSidebar}
 			<FlowConversationsSidebar
 				bind:this={sidebar}
-				{chat}
-				{chatState}
+				{listChat}
+				{pool}
+				{poolState}
 				defaultKind={conversationKind}
 				canFilterKind={conversationKind !== 'deployed'}
 			/>
 		{/if}
-		<!-- The interface's host subscribes to the chat it was given, so a replaced chat
-		     (another flow or workspace) mounts a fresh interface rather than a stale host. -->
-		{#key chat}
+		{#key shown}
 			<FlowChatInterface
-				{chat}
+				chat={shown.chat}
+				chatHost={shown.host}
+				isTest={shownIsTest}
 				{deploymentInProgress}
 				{additionalInputsSchema}
 				{flowModules}

frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte

@@ -3,12 +3,12 @@
 	import { Loader2, MessageCircle, Settings2 } from 'lucide-svelte'
 	import AIChatDisplay from '$lib/components/copilot/chat/AIChatDisplay.svelte'
 	import { setChatViewHost } from '$lib/components/copilot/chat/chatViewHost'
-	import { FlowChatViewHost } from './flowChatViewHost.svelte'
+	import type { FlowChatViewHost } from './flowChatViewHost.svelte'
 	import Modal from '$lib/components/common/modal/Modal.svelte'
 	import SchemaForm from '$lib/components/SchemaForm.svelte'
 	import GfmMarkdown from '$lib/components/GfmMarkdown.svelte'
 	import { emptyString, type DynamicInput } from '$lib/utils'
-	import { onDestroy, tick, untrack } from 'svelte'
+	import { tick, untrack } from 'svelte'
 	import type { Chat } from 'windmill-chat'
 	import type { FlowModule } from '$lib/gen'
 	import { deepEqual } from 'fast-equals'
@@ -23,6 +23,11 @@
 
 	interface Props {
 		chat: Chat
+		/** The conversation's host, which outlives this panel: FlowChat remounts the panel per
+		 * conversation, under `{#key}`, so a later value of either prop never reaches it. */
+		chatHost: FlowChatViewHost
+		/** Whether the shown conversation is a test chat, once the list has said. */
+		isTest?: boolean
 		deploymentInProgress?: boolean
 		additionalInputsSchema?: Record<string, any>
 		/** 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.

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<FlowChatViewHost>
+		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<InfiniteList | undefined>(undefined)
@@ -58,13 +71,13 @@
 	let renameDraft = $state('')
 	let renameInput = $state<TextInput | undefined>(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'
+	}
 </script>
 
+{#snippet statusDot(conversation: Conversation)}
+	<!-- The AI session sidebar's dot, with the resting mark this list needs: a session rests
+	     as a workspace or a fork, a conversation as a test run or one of the deployed flow's. -->
+	<SessionStatusDot
+		status={dotStatus(conversation.id)}
+		isFork={false}
+		restingTitle={conversation.isTest
+			? 'Test chat, run from the flow editor'
+			: 'Chat on the deployed flow'}
+	>
+		{#snippet resting()}
+			<span
+				class="w-[6px] h-[6px] rounded-full {conversation.isTest
+					? 'border border-gray-400 dark:border-gray-500'
+					: 'bg-gray-300 dark:bg-gray-600'}"
+			></span>
+		{/snippet}
+	</SessionStatusDot>
+{/snippet}
+
 <div
 	class="flex flex-col h-full bg-surface border-r transition-all duration-300 {expanded
 		? 'w-60'
@@ -219,12 +258,7 @@
 				{#if canFilterKind}
 					<!-- No focus trap: opening a row's menu does not close this popover, and a
 					     trapped popover pulls focus back from the rename field that menu opens. -->
-					<Popover
-						placement="bottom-start"
-						closeButton={false}
-						disableFocusTrap
-						disabled={turnInFlight}
-					>
+					<Popover placement="bottom-start" closeButton={false} disableFocusTrap>
 						{#snippet trigger()}
 							<!-- Icon-only next to the wider New chat: which kind is listed is named in
 							     the title and by the group inside. -->
@@ -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 @@
 								<ToggleButtonGroup
 									selected={kind}
 									onSelected={(next) => 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"
 			>
-				<CountBadge count={items.length} small alwaysVisible={true} class="right-[3px] top-[3px]" />
+				<!-- The same badge the rows carry, over the one icon that stands for all of them:
+				     collapsed, what is worth a number is what arrived, not how many chats exist. -->
+				<UnreadCountBadge
+					count={totalUnread}
+					small
+					class="absolute right-[3px] top-[3px] pointer-events-none"
+				/>
 			</Button>
 		</div>
 	{/if}
@@ -299,7 +337,6 @@
 						onClick={(e) => {
 							e?.stopPropagation()
 							draft = false
-							chat.newConversation()
 						}}
 						title="Discard draft"
 						destructive
@@ -314,7 +351,7 @@
 		<InfiniteList
 			bind:this={list}
 			bind:items
-			selectedItemId={chatState.conversationId}
+			selectedItemId={poolState.selectedId}
 			noBorder={true}
 			rounded={false}
 			preventXOverflow={true}
@@ -348,19 +385,41 @@
 								/>
 							</div>
 						{:else}
+							{@const unread = poolState.unread[conversation.id] ?? 0}
+							{@const queued = !!pool.get(conversation.id)?.host.queuedMessage}
 							<Button
 								unifiedSize="md"
 								variant="subtle"
 								onClick={() => {
 									draft = false
-									chat.selectConversation(conversation.id)
+									pool.select(conversation.id)
 								}}
-								selected={chatState.conversationId === conversation.id}
-								btnClasses="transition-all duration-150 group"
+								selected={poolState.selectedId === conversation.id}
+								btnClasses="transition-all duration-150 group gap-2"
 							>
-								<span class="flex-1 text-left truncate">
+								<!-- In the slot New chat's icon occupies above, so the column lines up. Says
+								     what the chat is doing where there is something to say, and which kind of
+								     chat it is otherwise. -->
+								{@render statusDot(conversation)}
+								<span
+									class={twMerge(
+										'flex-1 text-left truncate',
+										unread > 0 ? 'font-semibold text-primary' : ''
+									)}
+								>
 									{getConversationTitle(conversation)}
 								</span>
+								{#if queued || unread > 0}
+									<span class="shrink-0 inline-flex items-center gap-1">
+										{#if queued}
+											<PencilLine
+												class="w-3 h-3 text-tertiary"
+												aria-label="Message waiting to send"
+											/>
+										{/if}
+										<UnreadCountBadge count={unread} />
+									</span>
+								{/if}
 								<!-- svelte-ignore a11y_click_events_have_key_events -->
 								<!-- svelte-ignore a11y_no_static_element_interactions -->
 								<div

frontend/src/lib/components/flows/conversations/flowChatPool.test.ts

@@ -0,0 +1,126 @@
+import { describe, expect, it, vi } from 'vitest'
+import type { Chat, ChatMessage, ChatState, Conversation } from 'windmill-chat'
+import { FlowChatPool } from './flowChatPool'
+
+/** A chat whose state the test drives; `selectConversation` only sets the id. */
+function fakeChat() {
+	let state: ChatState = {
+		conversationId: undefined,
+		messages: [],
+		status: 'idle',
+		error: undefined,
+		conversations: [],
+		history: 'server',
+		loadingMessages: false,
+		hasMoreMessages: false
+	}
+	const listeners = new Set<(s: ChatState) => void>()
+	const set = (patch: Partial<ChatState>) => {
+		state = { ...state, ...patch }
+		for (const listener of listeners) listener(state)
+	}
+	const chat = {
+		getState: () => state,
+		subscribe: (listener: (s: ChatState) => void) => {
+			listeners.add(listener)
+			listener(state)
+			return () => listeners.delete(listener)
+		},
+		sendMessage: vi.fn(async () => {}),
+		resumeTurn: vi.fn(async () => {}),
+		stop: vi.fn(async () => {}),
+		newConversation: vi.fn(),
+		selectConversation: vi.fn(async (id: string) => set({ conversationId: id })),
+		loadConversations: vi.fn(async () => []),
+		deleteConversation: vi.fn(async () => {}),
+		renameConversation: vi.fn(async () => {}),
+		loadOlderMessages: vi.fn(async () => {}),
+		destroy: vi.fn()
+	} satisfies Chat
+	return { chat, set }
+}
+
+function answer(id: string, pending = false): ChatMessage {
+	return { id, role: 'assistant', content: 'done', success: true, createdAt: '', pending }
+}
+
+function conversation(id: string, extra: Partial<Conversation> = {}): Conversation {
+	return { id, title: id, createdAt: '', updatedAt: '', ...extra }
+}
+
+function pool(
+	options: { keepSettled?: number; isRunFinished?: (jobId: string) => Promise<boolean> } = {}
+) {
+	const chats: ReturnType<typeof fakeChat>[] = []
+	const hosts = { resumeTurn: vi.fn(), dispose: vi.fn() }
+	const created = new FlowChatPool<{ chat: Chat }>({
+		createChat: () => {
+			const fake = fakeChat()
+			chats.push(fake)
+			return fake.chat
+		},
+		createHost: (chat) => ({ chat }),
+		disposeHost: hosts.dispose,
+		hasQueued: () => false,
+		resumeTurn: (host, turn) => hosts.resumeTurn(host.chat, turn),
+		isRunFinished: options.isRunFinished ?? (async () => false),
+		keepSettled: options.keepSettled,
+		pollMs: 10
+	})
+	const chatOf = (id: string) => chats.find((c) => c.chat.getState().conversationId === id)!
+	return { pool: created, chatOf, hosts }
+}
+
+describe('FlowChatPool', () => {
+	it('runs a turn in each of two conversations and counts an answer that lands out of view', () => {
+		const { pool: p, chatOf } = pool()
+		p.select('a')
+		chatOf('a').set({ status: 'streaming' })
+		p.select('b')
+		chatOf('b').set({ status: 'submitted' })
+		expect(p.getState().activity).toEqual({ a: 'running', b: 'running' })
+
+		chatOf('a').set({ status: 'streaming', messages: [answer('m1', true)] })
+		chatOf('a').set({ status: 'idle', messages: [answer('m1')] })
+		expect(p.getState().activity).toEqual({ b: 'running' })
+		expect(p.getState().unread).toEqual({ a: 1 })
+
+		p.select('a')
+		expect(p.getState().unread).toEqual({})
+		expect(chatOf('a').chat.destroy).not.toHaveBeenCalled()
+		p.destroy()
+	})
+
+	it('polls a turn another page started until its run ends, and follows it when selected', async () => {
+		let finished = false
+		const { pool: p, chatOf, hosts } = pool({ isRunFinished: async () => finished })
+		const turn = { jobId: 'job-1', userSeq: 7 }
+		p.setListed([
+			conversation('a', { runningTurn: turn }),
+			conversation('b', { runningTurn: turn })
+		])
+		expect(p.getState().activity).toEqual({ a: 'running', b: 'running' })
+
+		p.select('a')
+		expect(hosts.resumeTurn).toHaveBeenCalledWith(chatOf('a').chat, turn)
+
+		finished = true
+		await vi.waitFor(() => expect(p.getState().activity).toEqual({}))
+		p.destroy()
+	})
+
+	it('releases settled chats past the budget, never one still running', () => {
+		const { pool: p, chatOf, hosts } = pool({ keepSettled: 1 })
+		p.select('busy')
+		chatOf('busy').set({ status: 'streaming' })
+		p.select('old')
+		p.select('recent')
+		p.select('shown')
+		expect(chatOf('old').chat.destroy).toHaveBeenCalled()
+		expect(p.get('old')).toBeUndefined()
+		expect(p.get('recent')).toBeDefined()
+		expect(p.get('busy')).toBeDefined()
+		expect(hosts.dispose).toHaveBeenCalledTimes(1)
+		p.destroy()
+	})
+})

frontend/src/lib/components/flows/conversations/flowChatPool.ts

@@ -0,0 +1,290 @@
+import type { Chat, ChatMessage, ChatState, Conversation, RunningTurn } from 'windmill-chat'
+
+/**
+ * Whether the turn the user message at `index` started failed: its last row before the
+ * next user message reports `success: false`. The last row, not any row: a tool call can
+ * fail and the agent still answer, and that turn completed.
+ */
+export function turnFailed(messages: readonly ChatMessage[], index: number): boolean {
+	let last: ChatMessage | undefined
+	for (let i = index + 1; i < messages.length; i++) {
+		const message = messages[i]
+		if (message.role === 'user') break
+		last = message
+	}
+	return last?.success === false
+}
+
+/** Whether the latest turn failed, per `turnFailed`. False before any turn. */
+export function lastTurnFailed(messages: readonly ChatMessage[]): boolean {
+	for (let i = messages.length - 1; i >= 0; i--) {
+		if (messages[i].role === 'user') return turnFailed(messages, i)
+	}
+	return false
+}
+
+export function isBusy(status: ChatState['status']): boolean {
+	return status === 'submitted' || status === 'streaming'
+}
+
+/** What a conversation's row says about it. */
+export type ConversationActivity = 'running' | 'error' | 'idle'
+
+export interface FlowChatPoolState {
+	/** The conversation shown. Unset for a new chat that has not run its first turn. */
+	selectedId: string | undefined
+	/** Only conversations with something to say; absent means idle. */
+	activity: Record<string, ConversationActivity>
+	/** Answers that arrived in a conversation while another one was shown. */
+	unread: Record<string, number>
+}
+
+export interface PooledChat<H> {
+	chat: Chat
+	host: H
+}
+
+export interface FlowChatPoolOptions<H> {
+	/** A chat on the flow with no conversation selected. */
+	createChat(): Chat
+	createHost(chat: Chat): H
+	disposeHost(host: H): void
+	/** Whether a message typed during the turn waits in the host to go out. */
+	hasQueued(host: H): boolean
+	/** Follows a turn another page started, through the host so what it queues waits for it. */
+	resumeTurn(host: H, turn: RunningTurn): void
+	/** Whether a run has ended, for a running conversation this page holds no chat for. */
+	isRunFinished(jobId: string): Promise<boolean>
+	/** Settled conversations kept in memory beside the shown and the busy ones. */
+	keepSettled?: number
+	pollMs?: number
+}
+
+interface Entry<H> extends PooledChat<H> {
+	unsubscribe: () => void
+	lastShownAt: number
+	/** Assistant messages already counted, so a message counts once as it settles. */
+	counted: Set<string>
+}
+
+/**
+ * The conversations of one flow chat, each on its own `Chat` so that each can run a turn
+ * while another is shown. A `Chat` follows one conversation for its whole life and is
+ * never switched, which is what lets its turn keep running in the background.
+ *
+ * Plain TypeScript on the `windmill-chat` API, with the view host left generic: nothing
+ * here is specific to Svelte or to the app, so it can move into the SDK as is.
+ */
+export class FlowChatPool<H> {
+	readonly #options: FlowChatPoolOptions<H>
+	readonly #entries = new Map<string, Entry<H>>()
+	/** The chat a new conversation starts on; it joins `#entries` once its first turn names it. */
+	#draft: Entry<H> | undefined
+	/** Turns running in conversations this pool is not following, as the list reported them. */
+	readonly #running = new Map<string, RunningTurn>()
+	readonly #unread = new Map<string, number>()
+	readonly #listeners = new Set<(state: FlowChatPoolState) => void>()
+	#selectedId: string | undefined
+	#state: FlowChatPoolState = { selectedId: undefined, activity: {}, unread: {} }
+	#poll: ReturnType<typeof setInterval> | undefined
+	#clock = 0
+	#destroyed = false
+
+	constructor(options: FlowChatPoolOptions<H>) {
+		this.#options = options
+		this.newChat()
+	}
+
+	getState = (): FlowChatPoolState => this.#state
+
+	subscribe = (listener: (state: FlowChatPoolState) => void): (() => void) => {
+		this.#listeners.add(listener)
+		listener(this.#state)
+		return () => {
+			this.#listeners.delete(listener)
+		}
+	}
+
+	/** The chat shown now. */
+	get selected(): PooledChat<H> {
+		// The shown conversation is never evicted, and forgetting it shows a new chat.
+		return (this.#selectedId === undefined ? this.#draft : this.#entries.get(this.#selectedId))!
+	}
+
+	/** The chat of a conversation, when this pool holds one. */
+	get(conversationId: string): PooledChat<H> | undefined {
+		return this.#entries.get(conversationId)
+	}
+
+	/** Shows a new chat, reusing the one already waiting for its first message. */
+	newChat = (): PooledChat<H> => {
+		if (!this.#draft) this.#draft = this.#track(undefined)
+		this.#selectedId = undefined
+		this.#publish()
+		return this.#draft
+	}
+
+	/** Shows a conversation, following its running turn when another page started it. */
+	select = (conversationId: string): void => {
+		let entry = this.#entries.get(conversationId)
+		if (!entry) {
+			entry = this.#track(conversationId)
+			this.#entries.set(conversationId, entry)
+		}
+		this.#selectedId = conversationId
+		entry.lastShownAt = ++this.#clock
+		this.#unread.delete(conversationId)
+		const turn = this.#running.get(conversationId)
+		if (turn && !isBusy(entry.chat.getState().status)) {
+			this.#running.delete(conversationId)
+			this.#options.resumeTurn(entry.host, turn)
+		}
+		this.#evict()
+		this.#publish()
+	}
+
+	/**
+	 * What the conversation list last said. A conversation it reports running that no chat
+	 * here is following gets its run polled, so its row stops saying so when the run ends.
+	 */
+	setListed = (conversations: readonly Conversation[]): void => {
+		for (const conversation of conversations) {
+			const followed = this.#entries.get(conversation.id)
+			if (conversation.runningTurn && !(followed && isBusy(followed.chat.getState().status))) {
+				this.#running.set(conversation.id, conversation.runningTurn)
+			} else {
+				this.#running.delete(conversation.id)
+			}
+		}
+		// The shown conversation follows its turn now rather than waiting for a poll to end it.
+		if (this.#selectedId !== undefined && this.#running.has(this.#selectedId)) {
+			this.select(this.#selectedId)
+		}
+		this.#schedulePoll()
+		this.#publish()
+	}
+
+	/** Drops a conversation that no longer exists; a new chat is shown in its place. */
+	forget = (conversationId: string): void => {
+		const entry = this.#entries.get(conversationId)
+		if (entry) this.#release(entry)
+		this.#entries.delete(conversationId)
+		this.#running.delete(conversationId)
+		this.#unread.delete(conversationId)
+		if (this.#selectedId === conversationId) this.newChat()
+		else this.#publish()
+	}
+
+	destroy = (): void => {
+		this.#destroyed = true
+		clearInterval(this.#poll)
+		for (const entry of this.#entries.values()) this.#release(entry)
+		if (this.#draft) this.#release(this.#draft)
+		this.#entries.clear()
+		this.#draft = undefined
+		this.#listeners.clear()
+	}
+
+	#track(conversationId: string | undefined): Entry<H> {
+		const chat = this.#options.createChat()
+		// Selected before the host exists: a host treats a change of conversation as the
+		// reader leaving one, and this chat never leaves its conversation.
+		if (conversationId !== undefined) void chat.selectConversation(conversationId)
+		const entry: Entry<H> = {
+			chat,
+			host: this.#options.createHost(chat),
+			unsubscribe: () => {},
+			lastShownAt: ++this.#clock,
+			counted: new Set()
+		}
+		entry.unsubscribe = chat.subscribe((state) => this.#onChatState(entry, state))
+		return entry
+	}
+
+	#onChatState(entry: Entry<H>, state: ChatState): void {
+		if (this.#destroyed) return
+		if (entry === this.#draft && state.conversationId !== undefined) {
+			// The new chat's first turn named its conversation.
+			this.#draft = undefined
+			this.#entries.set(state.conversationId, entry)
+			if (this.#selectedId === undefined) this.#selectedId = state.conversationId
+		}
+		const id = state.conversationId
+		if (id === undefined) return
+		if (isBusy(state.status)) this.#running.delete(id)
+		const shown = id === this.#selectedId
+		for (const message of state.messages) {
+			if (message.role !== 'assistant' || message.pending || entry.counted.has(message.id)) continue
+			entry.counted.add(message.id)
+			if (!shown && !state.loadingMessages) this.#unread.set(id, (this.#unread.get(id) ?? 0) + 1)
+		}
+		if (!isBusy(state.status)) this.#evict()
+		this.#publish()
+	}
+
+	#activity(id: string): ConversationActivity {
+		const state = this.#entries.get(id)?.chat.getState()
+		if (this.#running.has(id) || (state && isBusy(state.status))) return 'running'
+		if (state && (state.status === 'error' || lastTurnFailed(state.messages))) return 'error'
+		return 'idle'
+	}
+
+	/** Settled chats past the budget go, least recently shown first; their unread count stays. */
+	#evict(): void {
+		const settled = [...this.#entries.entries()].filter(
+			([id, entry]) =>
+				id !== this.#selectedId &&
+				!isBusy(entry.chat.getState().status) &&
+				!this.#options.hasQueued(entry.host)
+		)
+		settled.sort(([, a], [, b]) => b.lastShownAt - a.lastShownAt)
+		for (const [id, entry] of settled.slice(this.#options.keepSettled ?? 5)) {
+			this.#release(entry)
+			this.#entries.delete(id)
+		}
+	}
+
+	#release(entry: Entry<H>): void {
+		entry.unsubscribe()
+		this.#options.disposeHost(entry.host)
+		entry.chat.destroy()
+	}
+
+	#schedulePoll(): void {
+		if (this.#running.size === 0) {
+			clearInterval(this.#poll)
+			this.#poll = undefined
+			return
+		}
+		if (this.#poll) return
+		this.#poll = setInterval(() => void this.#pollRuns(), this.#options.pollMs ?? 3000)
+	}
+
+	async #pollRuns(): Promise<void> {
+		await Promise.all(
+			[...this.#running].map(async ([id, turn]) => {
+				const finished = await this.#options.isRunFinished(turn.jobId).catch(() => false)
+				if (finished && this.#running.get(id) === turn) this.#running.delete(id)
+			})
+		)
+		if (this.#destroyed) return
+		this.#schedulePoll()
+		this.#publish()
+	}
+
+	#publish(): void {
+		if (this.#destroyed) return
+		const ids = new Set([...this.#entries.keys(), ...this.#running.keys(), ...this.#unread.keys()])
+		const activity: Record<string, ConversationActivity> = {}
+		for (const id of ids) {
+			const value = this.#activity(id)
+			if (value !== 'idle') activity[id] = value
+		}
+		this.#state = {
+			selectedId: this.#selectedId,
+			activity,
+			unread: Object.fromEntries(this.#unread)
+		}
+		for (const listener of this.#listeners) listener(this.#state)
+	}
+}

frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts

@@ -1,4 +1,11 @@
-import type { Chat, ChatMessage, ChatState } from 'windmill-chat'
+import {
+	TurnRunningError,
+	type Chat,
+	type ChatMessage,
+	type ChatState,
+	type RunningTurn
+} from 'windmill-chat'
+import { isBusy, lastTurnFailed, turnFailed } from './flowChatPool'
 import type {
 	ChatSendRequestOptions,
 	ChatViewHost
@@ -22,10 +29,6 @@ export type FlowChatViewHostOptions = {
 	sendDisabled?: () => boolean
 }
 
-function isBusy(status: ChatState['status']): boolean {
-	return status === 'submitted' || status === 'streaming'
-}
-
 /** A tool's arguments or result as the card shows them: parsed where the string is JSON. */
 function parseToolPayload(raw: string | undefined): unknown {
 	if (raw === undefined || raw === '') return undefined
@@ -36,29 +39,6 @@ function parseToolPayload(raw: string | undefined): unknown {
 	}
 }
 
-/**
- * Whether the turn the user message at `index` started failed: its last row before the
- * next user message reports `success: false`. The last row, not any row: a tool call can
- * fail and the agent still answer, and that turn completed.
- */
-export function turnFailed(messages: readonly ChatMessage[], index: number): boolean {
-	let last: ChatMessage | undefined
-	for (let i = index + 1; i < messages.length; i++) {
-		const message = messages[i]
-		if (message.role === 'user') break
-		last = message
-	}
-	return last?.success === false
-}
-
-/** Whether the latest turn failed, per `turnFailed`. False before any turn. */
-function lastTurnFailed(messages: readonly ChatMessage[]): boolean {
-	for (let i = messages.length - 1; i >= 0; i--) {
-		if (messages[i].role === 'user') return turnFailed(messages, i)
-	}
-	return false
-}
-
 export function toDisplayMessages(messages: readonly ChatMessage[]): DisplayMessage[] {
 	let userIndex = 0
 	return messages.map((message, i): DisplayMessage => {
@@ -114,7 +94,8 @@ export function toDisplayMessages(messages: readonly ChatMessage[]): DisplayMess
  * makes, so what it can offer is what the SDK's `Chat` can: a message in, an answer
  * streamed back, Stop. Every copilot-only field is answered with "no" (see ChatViewHost).
  *
- * The host owns its subscription to the chat, so a panel swapping chats mounts a new one.
+ * One host per conversation, living as long as its chat: it outlasts the panel showing it,
+ * so a message queued in a conversation still goes out once the reader has moved on.
  */
 export class FlowChatViewHost implements ChatViewHost {
 	#chat: Chat
@@ -129,6 +110,17 @@ export class FlowChatViewHost implements ChatViewHost {
 		this.#unsubscribe = chat.subscribe((state) => this.#onState(state))
 	}
 
+	/** Set by the panel showing this conversation; a message queued here is sent with what
+	 * the last panel to show it read. */
+	setOptions(options: FlowChatViewHostOptions) {
+		this.#options = options
+	}
+
+	/** Follows a turn this chat did not start, so what is queued behind it waits for it. */
+	resumeTurn = (turn: RunningTurn) => {
+		this.#turnDone = this.#chat.resumeTurn(turn)
+	}
+
 	#disposed = false
 	/** Stops following the chat, and drops what was queued: a flush still waiting on the
 	 * turn's release would otherwise start a run from a panel that is gone. The chat itself
@@ -212,16 +204,25 @@ export class FlowChatViewHost implements ChatViewHost {
 		}
 		if (this.#options.sendDisabled?.()) {
 			// Refused, not dropped: the text waits in the composer for sending to reopen.
-			this.#aiChatInput?.prependText(text)
+			this.#returnText(text)
 			return false
 		}
 		this.#automaticScroll = true
 		// A run that fails is reported through the chat's `onError` and as a failed message;
 		// the promise itself only rejects when the chat refuses the turn outright, and the
-		// text is then handed back rather than dropped.
+		// text is then handed back rather than dropped. Refused because the conversation is
+		// still answering a message sent elsewhere: that turn is followed here, and this one
+		// waits behind it as if it had been typed during it.
 		const turn = this.#chat
 			.sendMessage(text, { inputs: this.#options.additionalInputs?.() })
-			.catch(() => this.#aiChatInput?.prependText(text))
+			.catch((e) => {
+				if (e instanceof TurnRunningError && !this.#disposed) {
+					this.queueMessage(text)
+					this.resumeTurn(e.turn)
+				} else {
+					this.#returnText(text)
+				}
+			})
 		this.#turnDone = turn
 		await turn
 		return true
@@ -239,6 +240,16 @@ export class FlowChatViewHost implements ChatViewHost {
 	#aiChatInput: Parameters<ChatViewHost['setAiChatInput']>[0] = null
 	setAiChatInput: ChatViewHost['setAiChatInput'] = (aiChatInput) => {
 		this.#aiChatInput = aiChatInput
+		if (aiChatInput && this.#returned) {
+			aiChatInput.prependText(this.#returned)
+			this.#returned = ''
+		}
+	}
+	/** Text handed back while no composer shows this conversation, for the next one that does. */
+	#returned = ''
+	#returnText(text: string) {
+		if (this.#aiChatInput) this.#aiChatInput.prependText(text)
+		else this.#returned = this.#returned ? `${this.#returned}\n${text}` : text
 	}
 
 	// One message typed while the turn runs, sent whole once it settles. Enter again
@@ -260,7 +271,7 @@ export class FlowChatViewHost implements ChatViewHost {
 		const text = this.#queued
 		if (!text) return
 		this.#queued = ''
-		this.#aiChatInput?.prependText(text)
+		this.#returnText(text)
 	}
 	flushQueuedMessage = () => {
 		const text = this.#queued

frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it, vi } from 'vitest'
-import type { Chat, ChatMessage, ChatState } from 'windmill-chat'
+import { TurnRunningError, type Chat, type ChatMessage, type ChatState } from 'windmill-chat'
 import { FlowChatViewHost, toDisplayMessages } from './flowChatViewHost.svelte'
 
 function message(partial: Partial<ChatMessage> & Pick<ChatMessage, 'role'>): ChatMessage {
@@ -39,11 +39,13 @@ function fakeChat(initial: ChatState = idleState()) {
 			return () => listeners.delete(listener)
 		},
 		sendMessage: vi.fn(async () => {}),
+		resumeTurn: vi.fn(async () => {}),
 		stop: vi.fn(async () => {}),
 		newConversation: vi.fn(),
 		selectConversation: vi.fn(async () => {}),
 		loadConversations: vi.fn(async () => []),
 		deleteConversation: vi.fn(async () => {}),
+		renameConversation: vi.fn(async () => {}),
 		loadOlderMessages: vi.fn(async () => {}),
 		destroy: vi.fn()
 	} satisfies Chat
@@ -239,6 +241,37 @@ describe('FlowChatViewHost', () => {
 		host.dispose()
 	})
 
+	it('follows the turn a conversation is still answering and sends the refused message after it', async () => {
+		const { chat, set } = fakeChat()
+		const turn = { jobId: 'job-9', userSeq: 41 }
+		chat.sendMessage.mockRejectedValueOnce(new TurnRunningError('still answering', turn))
+		let releaseResumed = () => {}
+		chat.resumeTurn.mockImplementationOnce(
+			() => new Promise<void>((resolve) => (releaseResumed = resolve))
+		)
+		const host = new FlowChatViewHost(chat)
+		await host.sendRequest({ instructions: 'after it' })
+		expect(chat.resumeTurn).toHaveBeenCalledWith(turn)
+		expect(host.queuedMessage).toBe('after it')
+		set({ status: 'streaming' })
+		set({ status: 'idle' })
+		releaseResumed()
+		await new Promise((resolve) => setTimeout(resolve, 0))
+		expect(chat.sendMessage).toHaveBeenLastCalledWith('after it', { inputs: undefined })
+		host.dispose()
+	})
+
+	it('keeps text handed back while no composer is mounted for the next one', () => {
+		const { chat } = fakeChat(idleState({ status: 'streaming' }))
+		const host = new FlowChatViewHost(chat)
+		host.queueMessage('typed before leaving')
+		host.cancel()
+		const prependText = vi.fn()
+		host.setAiChatInput({ prependText } as any)
+		expect(prependText).toHaveBeenCalledWith('typed before leaving')
+		host.dispose()
+	})
+
 	it('hands the queue back to the composer on Stop and on a failed turn', async () => {
 		const { chat, set } = fakeChat(idleState({ status: 'streaming' }))
 		const host = new FlowChatViewHost(chat)

frontend/src/lib/components/sessions/SessionPicker.svelte

@@ -51,6 +51,7 @@
 		resetSessionPreviewTabs
 	} from './sessionRuntime.svelte'
 	import SessionStatusDot from './SessionStatusDot.svelte'
+	import UnreadCountBadge from '$lib/components/common/badge/UnreadCountBadge.svelte'
 	import { buildWorkspaceHierarchy } from '$lib/utils/workspaceHierarchy'
 	import SessionFilterMenu from './SessionFilterMenu.svelte'
 	import { Menu, Menubar, MenuItem } from '$lib/components/meltComponents'
@@ -801,14 +802,7 @@
 													{#if draft}
 														<PencilLine class="w-3 h-3 text-tertiary" aria-label="Unsent draft" />
 													{/if}
-													{#if unread > 0}
-														<span
-															class="inline-flex items-center justify-center rounded-full bg-surface-accent-primary text-white font-medium leading-none min-w-4 h-4 px-1 text-[10px]"
-															aria-label="{unread} unread message{unread === 1 ? '' : 's'}"
-														>
-															{unread > 9 ? '9+' : unread}
-														</span>
-													{/if}
+													<UnreadCountBadge count={unread} />
 												</span>
 											{/if}
 										</MenuItem>
@@ -1075,14 +1069,7 @@
 										{#if draft}
 											<PencilLine class="w-3 h-3 text-tertiary" aria-label="Unsent draft" />
 										{/if}
-										{#if unread > 0}
-											<span
-												class="inline-flex items-center justify-center rounded-full bg-surface-accent-primary text-white font-medium leading-none min-w-4 h-4 px-1 text-[10px]"
-												aria-label="{unread} unread message{unread === 1 ? '' : 's'}"
-											>
-												{unread > 9 ? '9+' : unread}
-											</span>
-										{/if}
+										<UnreadCountBadge count={unread} />
 									</span>
 								{/if}
 							</button>

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<SessionChatStatus, string> = {
 		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)
+	)
 </script>
 
 <span class="inline-flex items-center justify-center w-4 h-3 shrink-0" {title}>
@@ -55,6 +67,8 @@
 		<CircleHelp class="w-3 h-3 text-amber-500" />
 	{:else if status === 'error'}
 		<AlertTriangle class="w-3 h-3 text-red-500" />
+	{:else if resting}
+		{@render resting()}
 	{:else if isFork}
 		{#if forkDetached}
 			<GitPullRequestClosed class="w-3 h-3 text-red-500" />