Tell a subscribing client whether the thread's turn is still running

crates/giskard-proto/src/lib.rs

@@ -115,6 +115,17 @@ pub enum ClientMessage {
 pub struct ThreadState {
     pub thread_id: ThreadId,
     pub state: serde_json::Value,
+    /// Whether a turn is in flight for this thread *right now*, answered from the server's turn
+    /// gate rather than from anything persisted in `state`.
+    ///
+    /// A turn can be started over HTTP (`POST /threads/start`) before the browser's socket for that
+    /// thread exists, so the client cannot always learn a turn's liveness from the event stream: if
+    /// such a turn finishes before the socket attaches, its `TurnCompleted` was addressed to nobody
+    /// and no [`LiveTurnSnapshot`] follows it. This flag closes that gap. The gate is held for the
+    /// whole turn — reserved before the start request returns, released when the turn ends — so it
+    /// also covers the window before the harness emits its first event, where the live buffer is
+    /// still empty but the turn is very much running.
+    pub active_turn: bool,
 }
 
 /// Lightweight cross-thread activity update for sidebar badges and browser notifications. This is

crates/giskard-server/src/routes.rs

@@ -3388,6 +3388,11 @@ async fn handle_client_msg(
                     )
                 }
             };
+            // Registering before the snapshot below is built is what makes the snapshot's
+            // `active_turn` safe to act on: a turn that ends after this line broadcasts its
+            // `TurnCompleted` to this client, and one that ended before it is already out of the
+            // turn gate. Build the snapshot first and a turn ending in between would be reported
+            // live by a client that then never hears it finish.
             state.hub.subscribe(thread_id, client_id, tx.clone()).await;
 
             if let Some(warning) = notice {
@@ -3422,6 +3427,7 @@ async fn handle_client_msg(
                 .send(ServerMessage::ThreadState(giskard_proto::ThreadState {
                     thread_id,
                     state: thread_state,
+                    active_turn: state.registry.thread_has_active_turn(thread_id).await,
                 }))
                 .await;
 
@@ -4627,13 +4633,15 @@ async fn broadcast_thread_state(
             return;
         }
     };
+    let active_turn = state.registry.thread_has_active_turn(thread_id).await;
     state
         .hub
         .broadcast(
             thread_id,
             ServerMessage::ThreadState(giskard_proto::ThreadState {
                 thread_id,
                 state: value,
+                active_turn,
             }),
         )
         .await;

crates/giskard-server/static/app.js

@@ -2684,7 +2684,7 @@ function handleServer(msg, ws) {
   const renderStartedAtMs = browserNowMs();
   recordReconnectMessageReceived(ws, messageType);
   switch (msg.type) {
-    case "thread_state": renderThreadState(msg.state); break;
+    case "thread_state": renderThreadState(msg.state, msg.active_turn); break;
     case "history_page": renderHistoryPage(msg); break;
     case "history_delta": renderHistoryDelta(msg); break;
     case "live_turn_snapshot": renderLiveTurnSnapshot(msg); break;
@@ -3136,7 +3136,7 @@ function handleIncomingApprovalRequest(request, tid, opts) {
   renderApprovalRequest(request);
 }
 
-function renderThreadState(s) {
+function renderThreadState(s, activeTurn) {
   if (!s) return;
   const shouldResetTranscript = state.awaitingInitialThreadState || state.awaitingThreadResync;
   // An incremental resync keeps the transcript. Remember whether the viewport was pinned to the
@@ -3175,6 +3175,24 @@ function renderThreadState(s) {
   if (shouldResetTranscript) {
     resetTranscriptForAuthoritativeSnapshot();
   }
+  releaseFirstTurnLockIfIdle(activeTurn);
+}
+
+// The first turn of a draft thread is started over HTTP, before this thread's socket exists, so the
+// composer locks optimistically (`firstTurnStartingThreadId`) with nothing on the wire yet to
+// confirm the turn. Every other lock is released by something the socket delivers — `turn_completed`
+// for a turn we watched, `error` for one that failed, a resync for one we missed. This one has no
+// such release when the turn finishes *before* the socket attaches: its `turn_completed` was
+// addressed to nobody and, the turn being over, no live snapshot follows to correct us. The
+// composer would then stay locked until the thread is re-opened. The server's `active_turn` is the
+// authority here — held from before the start request returned until the turn ends — so a `false`
+// really does mean there is nothing left to wait for.
+function releaseFirstTurnLockIfIdle(activeTurn) {
+  if (activeTurn !== false) return;   // only an explicit "no turn is running" releases the lock
+  if (!state.firstTurnStartingThreadId) return;
+  if (String(state.firstTurnStartingThreadId) !== String(state.threadId)) return;
+  state.firstTurnStartingThreadId = null;
+  setTurnActive(false);
 }
 function resetTranscriptForAuthoritativeSnapshot() {
   const keepFirstTurnActive =

crates/giskard-server/tests/e2e_smoke.rs

@@ -2313,15 +2313,15 @@ async fn wait_for_thread_state(
         tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
     >,
     thread_id: ThreadId,
-) {
+) -> giskard_proto::ThreadState {
     let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(5);
     while tokio::time::Instant::now() < deadline {
         match tokio::time::timeout(tokio::time::Duration::from_secs(1), ws.next()).await {
             Ok(Some(Ok(tokio_tungstenite::tungstenite::Message::Text(text)))) => {
                 if let Ok(ServerMessage::ThreadState(state)) = serde_json::from_str(&text)
                     && state.thread_id == thread_id
                 {
-                    return;
+                    return state;
                 }
             }
             Ok(Some(Ok(_))) => {}
@@ -2611,6 +2611,137 @@ async fn send_input_rejects_second_turn_before_turn_started() {
     assert_eq!(harness.start_calls(), 2);
 }
 
+/// A thread's first turn is started over HTTP (`POST /threads/start`), before the browser has a
+/// socket for that thread. If that turn finishes before the subscribe lands, nothing else on the
+/// socket describes it: the `TurnCompleted` was broadcast to a thread nobody was subscribed to, and
+/// a turn that is over leaves no `LiveTurnSnapshot` behind. The subscribe's `ThreadState` is then
+/// the only place the browser can learn the turn is done — without it the composer stays locked on
+/// a turn that ended before it was ever watched.
+#[tokio::test]
+async fn subscribe_thread_state_reports_a_turn_that_ended_before_the_socket_attached() {
+    let (_tmp, state, port) = start_server_with_extra_config_on_available_port("").await;
+    let base = format!("http://127.0.0.1:{port}");
+    let client = reqwest::Client::new();
+    let cookie = login_cookie(&client, &base).await;
+    let (project_id, thread_id) = create_project_and_thread(&client, &base, &cookie).await;
+
+    // The whole turn happens with no socket subscribed to this thread — the window the browser
+    // races when it starts a turn over HTTP and only then opens the thread. Driven through the
+    // registry rather than `POST /threads/start` because the replay harness only streams a fixture
+    // into a thread opened with that fixture's resume key, which that endpoint does not do.
+    let thread_file = state
+        .store
+        .load_thread(project_id, thread_id)
+        .await
+        .unwrap()
+        .unwrap();
+    state
+        .registry
+        .start_turn(
+            thread_id,
+            UserInput::text("a turn nobody is subscribed to"),
+            TurnOverrides {
+                model: Some(thread_file.current_model.clone()),
+                mode: thread_file.mode,
+                permission_preset: thread_file.permission_preset,
+            },
+            thread_file.current_model,
+        )
+        .await
+        .unwrap();
+    let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(5);
+    while state.registry.thread_has_active_turn(thread_id).await {
+        if tokio::time::Instant::now() >= deadline {
+            panic!("the unsubscribed turn never finished");
+        }
+        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
+    }
+
+    let mut ws = connect_ws(port, &cookie).await;
+    ws.send(tokio_tungstenite::tungstenite::Message::Text(
+        serde_json::to_string(&ClientMessage::Subscribe {
+            thread_id,
+            since: None,
+        })
+        .unwrap()
+        .into(),
+    ))
+    .await
+    .unwrap();
+
+    let thread_state = wait_for_thread_state(&mut ws, thread_id).await;
+    assert!(
+        !thread_state.active_turn,
+        "a turn that finished before this socket subscribed must be reported as done"
+    );
+}
+
+/// The counterpart: a turn the harness has accepted but not yet said anything about. The live
+/// buffer is still empty here — there is no `TurnStarted` to put in it — so a liveness answer read
+/// from the buffer would wrongly say "idle" and unlock a composer whose turn is about to stream.
+/// The turn gate, reserved before the start request returns, is what covers this window.
+#[tokio::test]
+async fn subscribe_thread_state_reports_a_turn_the_harness_has_not_streamed_yet() {
+    let harness = Arc::new(SlowStartHarness::new());
+    let (_tmp, state, port) = start_slow_start_server_on_available_port(harness.clone()).await;
+    let base = format!("http://127.0.0.1:{port}");
+    let client = reqwest::Client::new();
+    let cookie = login_cookie(&client, &base).await;
+    let (_, thread_id) = create_project_and_thread(&client, &base, &cookie).await;
+
+    let mut sender = connect_ws(port, &cookie).await;
+    sender
+        .send(tokio_tungstenite::tungstenite::Message::Text(
+            serde_json::to_string(&ClientMessage::Subscribe {
+                thread_id,
+                since: None,
+            })
+            .unwrap()
+            .into(),
+        ))
+        .await
+        .unwrap();
+    sender
+        .send(tokio_tungstenite::tungstenite::Message::Text(
+            serde_json::to_string(&ClientMessage::SendInput {
+                thread_id,
+                text: "held inside the harness".into(),
+                attachments: Vec::new(),
+            })
+            .unwrap()
+            .into(),
+        ))
+        .await
+        .unwrap();
+    harness.wait_for_start_calls(1).await;
+    assert!(
+        !state.live_buffers.is_active(thread_id).await,
+        "the harness is still inside start_turn, so there is nothing buffered for this turn"
+    );
+
+    let mut latecomer = connect_ws(port, &cookie).await;
+    latecomer
+        .send(tokio_tungstenite::tungstenite::Message::Text(
+            serde_json::to_string(&ClientMessage::Subscribe {
+                thread_id,
+                since: None,
+            })
+            .unwrap()
+            .into(),
+        ))
+        .await
+        .unwrap();
+
+    let thread_state = wait_for_thread_state(&mut latecomer, thread_id).await;
+    assert!(
+        thread_state.active_turn,
+        "a turn the harness has accepted is running, even before it streams anything"
+    );
+
+    harness.release_first_start();
+    wait_for_turn_completed(&mut sender, thread_id).await;
+}
+
 #[tokio::test]
 async fn send_input_rejects_same_thread_during_compaction() {
     let (_tmp, state, port) = start_slow_compaction_server_on_available_port().await;

crates/giskard-server/tests/ui.rs

@@ -1698,7 +1698,7 @@ fn browser_isolates_global_and_project_model_catalogs() {
 
     let thread_state = between(
         body,
-        "function renderThreadState(s) {",
+        "function renderThreadState(s, activeTurn) {",
         "function resetTranscriptForAuthoritativeSnapshot()",
     );
     assert!(
@@ -1793,7 +1793,7 @@ fn browser_resubscribe_replaces_transient_transcript_state() {
 
     let render_thread_state = between(
         body,
-        "function renderThreadState(s) {",
+        "function renderThreadState(s, activeTurn) {",
         "function resetTranscriptForAuthoritativeSnapshot() {",
     );
     assert!(render_thread_state.contains(

specs/giskard-specification.md

@@ -2135,6 +2135,10 @@ Flow: user clicks "New project" → names it → picks a directory via the file
   returned `harness_thread_id`, writes `<thread_id>.json`, and immediately calls `start_turn`.
   If native creation fails, nothing is persisted. If persistence or synchronous `turn/start` fails
   after native creation, cleanup is best-effort and failures are logged.
+  This first turn begins before the browser subscribes to the new thread, so the composer opens
+  locked on a turn it has observed nothing of. It must not stay locked on the strength of that
+  assumption alone: the subscribe snapshot's `active_turn` (§13.6) is what settles whether the turn
+  is still running, and a turn that finished during the gap releases the composer with no re-open.
 - **Open existing:** selecting a persisted thread calls the same open endpoint with
   `thread_id = Some(existing_id)`. The server reattaches the harness using the stored native
   `harness_thread_id` but preserves the durable Giskard `ThreadId`; opening a thread is
@@ -2915,7 +2919,8 @@ interactive forwarder owns the turn gate, the passive subscriber yields before b
 above, sent once to a connecting client only and never broadcast, so a browser that missed the live
 signal still shows what is blocked; a separate message so clients can tell a replay from a live
 event and apply SB6's alert-once-per-session rule),
-`ThreadState { thread_id, state }` (persisted snapshot on subscribe/resync),
+`ThreadState { thread_id, state, active_turn }` (persisted snapshot on subscribe/resync;
+`active_turn` is live server state rather than part of the persisted snapshot — see §13.6),
 `LiveTurnSnapshot { thread_id, turn_id, user_input?, accumulated,
 answered_approvals, answered_server_requests }` (in-flight turn reconstruction on reconnect,
 carrying the turn input when the server synthesized the turn context, the `WireAgentEvent`s of the
@@ -3056,6 +3061,15 @@ events through the same event handler used for live WebSocket events.
   live snapshot. Later metadata-only `ThreadState` broadcasts are not subscribe snapshots and must
   not clear the visible transcript.
 
+  Step 2 is conditional, so its *absence* carries no information: a turn that is over and a turn
+  the harness has accepted but not yet streamed both produce no `LiveTurnSnapshot`. The
+  `ThreadState` of step 1 therefore states turn liveness outright in `active_turn`, answered from
+  the server's active-turn gate — held from before `POST /threads/start` returns until the turn
+  ends, so it covers the window where the live buffer is still empty. This is what a client needs
+  when it starts a turn over HTTP and subscribes afterwards (§7.1): if that turn finishes first,
+  its `TurnCompleted` was broadcast to a thread with no subscribers, and without `active_turn` the
+  client would keep its composer locked on a turn that is already done.
+
   This means a reconnected client sees the full in-progress turn, including still-pending
   approval and server-request prompts. The live buffer is bounded
   (coalesced/truncated for very long command output, keeping head+tail) to cap memory; the

tests/e2e/tests/draft-composer.spec.ts

@@ -266,6 +266,36 @@ test.describe("draft composer", () => {
     await expect(page.locator("#transcript")).toContainText("Start a new thread");
     await expect(page.locator(".thread.active")).toHaveCount(0);
   });
+
+  // A draft's first turn is started over HTTP, before the thread has a socket, so the composer
+  // locks with nothing on the wire yet to confirm the turn. Against a fast harness that turn is
+  // usually over before the socket attaches: its `turn_completed` was addressed to nobody, and a
+  // finished turn leaves no live snapshot to follow, so nothing ever contradicted the optimistic
+  // lock. The composer stayed disabled — Stop showing, Send gone — with the agent's reply sitting
+  // right there on screen, until the thread was re-opened from the sidebar.
+  test("unlocks the composer when the first turn ends before the socket attaches", async ({ page }) => {
+    await page.locator(".proj", { hasText: "Demo" }).locator(".project-add").click();
+    await page.locator("#input").fill("First turn of a brand new thread");
+    await expect(page.locator("#sendBtn")).toBeEnabled();
+    await page.locator("#sendBtn").click();
+
+    await expect(
+      page.locator("#transcript .msg.agent", { hasText: SCRIPTED_REPLY }),
+    ).toBeVisible();
+
+    // No re-open, no reload: this same view has to come back to life on its own.
+    await expect(page.locator("#stopBtn")).toBeHidden();
+    await expect(page.locator("#sendBtn")).toBeVisible();
+
+    // And the thread is genuinely usable, not merely repainted: a second turn sends over the
+    // socket that is now attached, and answers.
+    await page.locator("#input").fill("Second turn, same thread");
+    await expect(page.locator("#sendBtn")).toBeEnabled();
+    await page.locator("#sendBtn").click();
+    await expect(
+      page.locator("#transcript .msg.agent", { hasText: SCRIPTED_REPLY }),
+    ).toHaveCount(2);
+  });
 });
 
 // `openThread` has the same save/await/restore shape as the draft opening that lost a message, and