diff --git a/crates/giskard-proto/src/lib.rs b/crates/giskard-proto/src/lib.rs index cf8b664..05c9601 100644 --- a/crates/giskard-proto/src/lib.rs +++ b/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 diff --git a/crates/giskard-server/src/routes.rs b/crates/giskard-server/src/routes.rs index 8332b33..3187da4 100644 --- a/crates/giskard-server/src/routes.rs +++ b/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,6 +4633,7 @@ async fn broadcast_thread_state( return; } }; + let active_turn = state.registry.thread_has_active_turn(thread_id).await; state .hub .broadcast( @@ -4634,6 +4641,7 @@ async fn broadcast_thread_state( ServerMessage::ThreadState(giskard_proto::ThreadState { thread_id, state: value, + active_turn, }), ) .await; diff --git a/crates/giskard-server/static/app.js b/crates/giskard-server/static/app.js index a21b7ac..c91a889 100644 --- a/crates/giskard-server/static/app.js +++ b/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 = diff --git a/crates/giskard-server/tests/e2e_smoke.rs b/crates/giskard-server/tests/e2e_smoke.rs index 8225fec..c86298b 100644 --- a/crates/giskard-server/tests/e2e_smoke.rs +++ b/crates/giskard-server/tests/e2e_smoke.rs @@ -2313,7 +2313,7 @@ async fn wait_for_thread_state( tokio_tungstenite::MaybeTlsStream, >, 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 { @@ -2321,7 +2321,7 @@ async fn wait_for_thread_state( 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; diff --git a/crates/giskard-server/tests/ui.rs b/crates/giskard-server/tests/ui.rs index e728853..a66a9da 100644 --- a/crates/giskard-server/tests/ui.rs +++ b/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( diff --git a/specs/giskard-specification.md b/specs/giskard-specification.md index 1a4a63f..d64b718 100644 --- a/specs/giskard-specification.md +++ b/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 `.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 diff --git a/tests/e2e/tests/draft-composer.spec.ts b/tests/e2e/tests/draft-composer.spec.ts index 27c0be2..6716485 100644 --- a/tests/e2e/tests/draft-composer.spec.ts +++ b/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