diff --git a/web/src/structured.test.ts b/web/src/structured.test.ts index e5faf7d..7873a0b 100644 --- a/web/src/structured.test.ts +++ b/web/src/structured.test.ts @@ -147,6 +147,8 @@ describe("structured connection: turn/composer gating", () => { input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true })); expect(sendStructuredPrompt).toHaveBeenCalledWith("s1", "do the thing"); expect(input.value).toBe(""); + ws.emit({ type: "reduced_state", state: { turn: { responding: true, thinking: false, active_tool_calls: [] } } }); + ws.emit({ type: "reduced_state", state: idleState }); vi.mocked(sendStructuredPrompt).mockRejectedValueOnce(new Error("boom")); input.value = "retry me"; @@ -158,6 +160,103 @@ describe("structured connection: turn/composer gating", () => { conn.dispose(); }); + it("disables immediately and dispatches only once before begin-turn is projected", () => { + vi.mocked(sendStructuredPrompt).mockImplementationOnce(() => new Promise(() => {})); + const ws = new FakeSocket(); + const conn = mount({ sockets: [ws] }); + ws.open(); + ws.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } }); + ws.emit({ type: "reduced_state", state: idleState }); + const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement; + const send = conn.el.querySelector(".structured-send") as HTMLButtonElement; + const form = conn.el.querySelector(".structured-composer") as HTMLFormElement; + input.value = "only once"; + + form.dispatchEvent(new SubmitEvent("submit", { bubbles: true, cancelable: true })); + form.dispatchEvent(new SubmitEvent("submit", { bubbles: true, cancelable: true })); + + expect(input.disabled).toBe(true); + expect(send.disabled).toBe(true); + expect(sendStructuredPrompt).toHaveBeenCalledTimes(1); + expect(sendStructuredPrompt).toHaveBeenCalledWith("s1", "only once"); + conn.dispose(); + }); + + it("does not overwrite a newer operator draft when the matching request fails", async () => { + let reject!: (error: unknown) => void; + vi.mocked(sendStructuredPrompt).mockImplementationOnce( + () => new Promise((_, rejectPromise) => { reject = rejectPromise; }), + ); + const ws = new FakeSocket(); + const conn = mount({ sockets: [ws] }); + ws.open(); + ws.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } }); + ws.emit({ type: "reduced_state", state: idleState }); + const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement; + input.value = "first prompt"; + input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true })); + ws.emit({ type: "reduced_state", state: { turn: { responding: true, thinking: false, active_tool_calls: [] } } }); + ws.emit({ type: "reduced_state", state: idleState }); + input.value = "newer draft"; + + reject(new Error("late failure")); + await Promise.resolve().then(() => Promise.resolve()); + + expect(input.value).toBe("newer draft"); + expect(conn.el.textContent).toContain("late failure"); + conn.dispose(); + }); + + it("ignores an older failure after a newer submission generation starts", async () => { + let rejectFirst!: (error: unknown) => void; + vi.mocked(sendStructuredPrompt).mockImplementationOnce( + () => new Promise((_, rejectPromise) => { rejectFirst = rejectPromise; }), + ); + const ws = new FakeSocket(); + const conn = mount({ sockets: [ws] }); + ws.open(); + ws.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } }); + ws.emit({ type: "reduced_state", state: idleState }); + const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement; + input.value = "first prompt"; + input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true })); + ws.emit({ type: "reduced_state", state: { turn: { responding: true, thinking: false, active_tool_calls: [] } } }); + ws.emit({ type: "reduced_state", state: idleState }); + input.value = "second prompt"; + input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true })); + + rejectFirst(new Error("stale failure")); + await Promise.resolve().then(() => Promise.resolve()); + + expect(sendStructuredPrompt).toHaveBeenCalledTimes(2); + expect(input.value).toBe(""); + expect(input.disabled).toBe(true); + expect(conn.el.textContent).not.toContain("stale failure"); + conn.dispose(); + }); + + it("ignores a submission failure after disposal", async () => { + let reject!: (error: unknown) => void; + vi.mocked(sendStructuredPrompt).mockImplementationOnce( + () => new Promise((_, rejectPromise) => { reject = rejectPromise; }), + ); + const ws = new FakeSocket(); + const conn = mount({ sockets: [ws] }); + ws.open(); + ws.emit({ type: "reduced_state", state: idleState }); + const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement; + input.value = "do not restore"; + input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true })); + + conn.dispose(); + input.value = "detached draft"; + reject(new Error("late failure")); + await Promise.resolve().then(() => Promise.resolve()); + + expect(input.value).toBe("detached draft"); + expect(conn.el.querySelector(".structured-error")).toBeNull(); + }); + it("cancels the active turn via the cancel button", () => { const ws = new FakeSocket(); const conn = mount({ sockets: [ws] }); @@ -301,6 +400,8 @@ describe("structured connection: reconnection/error/loading", () => { vi.useFakeTimers(); }); + afterEach(() => vi.useRealTimers()); + it("shows connecting, then connected, then reconnecting with backoff, and rebuilds from the fresh snapshot", () => { const first = new FakeSocket(); const second = new FakeSocket(); @@ -327,6 +428,154 @@ describe("structured connection: reconnection/error/loading", () => { conn.dispose(); }); + it("keeps a local submission gated across a reconnect until begin-turn arrives", () => { + vi.mocked(sendStructuredPrompt).mockImplementationOnce(() => new Promise(() => {})); + const first = new FakeSocket(); + const second = new FakeSocket(); + const conn = mount({ sockets: [first, second], shouldReconnect: () => true }); + first.open(); + first.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } }); + first.emit({ type: "reduced_state", state: idleState }); + const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement; + input.value = "survive reconnect"; + input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true })); + + first.drop(); + vi.advanceTimersByTime(1000); + second.open(); + second.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } }); + second.emit({ type: "reduced_state", state: idleState }); + expect(input.disabled).toBe(true); + + second.emit({ type: "reduced_state", state: { turn: { responding: true, thinking: false, active_tool_calls: [] } } }); + second.emit({ type: "reduced_state", state: idleState }); + expect(input.disabled).toBe(false); + conn.dispose(); + }); + + it("releases an accepted submission from a post-acceptance idle snapshot", async () => { + vi.mocked(sendStructuredPrompt).mockResolvedValueOnce(undefined); + const first = new FakeSocket(); + const second = new FakeSocket(); + const conn = mount({ sockets: [first, second], shouldReconnect: () => true }); + first.open(); + first.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } }); + first.emit({ type: "reduced_state", state: idleState }); + const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement; + input.value = "finishes while disconnected"; + input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true })); + await Promise.resolve().then(() => Promise.resolve()); + + first.drop(); + vi.advanceTimersByTime(1000); + second.open(); + second.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } }); + second.emit({ type: "reduced_state", state: idleState }); + + expect(input.disabled).toBe(false); + conn.dispose(); + }); + + it("does not release from reconnect idle cached before submission acceptance", async () => { + let accept!: () => void; + vi.mocked(sendStructuredPrompt).mockImplementationOnce( + () => new Promise((resolve) => { accept = resolve; }), + ); + const first = new FakeSocket(); + const second = new FakeSocket(); + const third = new FakeSocket(); + const conn = mount({ sockets: [first, second, third], shouldReconnect: () => true }); + first.open(); + first.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } }); + first.emit({ type: "reduced_state", state: idleState }); + const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement; + input.value = "accepted after reconnect"; + input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true })); + + first.drop(); + vi.advanceTimersByTime(1000); + second.open(); + second.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } }); + second.emit({ type: "reduced_state", state: idleState }); + expect(input.disabled).toBe(true); + + accept(); + await Promise.resolve().then(() => Promise.resolve()); + + expect(input.disabled).toBe(true); + expect(second.closed).toBe(true); + + third.open(); + third.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } }); + third.emit({ type: "reduced_state", state: idleState }); + expect(input.disabled).toBe(false); + conn.dispose(); + }); + + it("cancels pending backoff when acceptance forces reconciliation", async () => { + let accept!: () => void; + vi.mocked(sendStructuredPrompt).mockImplementationOnce( + () => new Promise((resolve) => { accept = resolve; }), + ); + const first = new FakeSocket(); + const second = new FakeSocket(); + const reconciliation = new FakeSocket(); + const duplicate = new FakeSocket(); + const conn = mount({ + sockets: [first, second, reconciliation, duplicate], + shouldReconnect: () => true, + }); + first.open(); + first.emit({ type: "reduced_state", state: idleState }); + const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement; + input.value = "accepted during backoff"; + input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true })); + + first.drop(); + vi.advanceTimersByTime(1000); + second.drop(); + accept(); + await Promise.resolve().then(() => Promise.resolve()); + + expect(reconciliation.onopen).not.toBeNull(); + vi.advanceTimersByTime(60_000); + expect(duplicate.onopen).toBeNull(); + + reconciliation.open(); + reconciliation.emit({ type: "reduced_state", state: idleState }); + expect(input.disabled).toBe(false); + conn.dispose(); + expect(reconciliation.closed).toBe(true); + }); + + it("does not reconcile when submission acceptance arrives after disposal", async () => { + let accept!: () => void; + vi.mocked(sendStructuredPrompt).mockImplementationOnce( + () => new Promise((resolve) => { accept = resolve; }), + ); + const first = new FakeSocket(); + const second = new FakeSocket(); + const late = new FakeSocket(); + const conn = mount({ sockets: [first, second, late], shouldReconnect: () => true }); + first.open(); + first.emit({ type: "reduced_state", state: idleState }); + const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement; + input.value = "accepted after disposal"; + input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true })); + + first.drop(); + vi.advanceTimersByTime(1000); + second.open(); + second.emit({ type: "reduced_state", state: idleState }); + conn.dispose(); + accept(); + await Promise.resolve().then(() => Promise.resolve()); + vi.advanceTimersByTime(60_000); + + expect(second.closed).toBe(true); + expect(late.onopen).toBeNull(); + }); + it("reports a daemon rejection reason and does not schedule a retry", () => { const only = new FakeSocket(); const conn = mount({ sockets: [only], shouldReconnect: () => true }); diff --git a/web/src/structured.ts b/web/src/structured.ts index 994c176..101af49 100644 --- a/web/src/structured.ts +++ b/web/src/structured.ts @@ -109,6 +109,12 @@ export function createStructuredConnection( let state: ReducedState = { turn: idleTurn() }; let pendingApprovalRowId: string | null = null; let latestSession: Session | undefined = store?.snapshot().sessions.get(sessionId); + let submissionGeneration = 0; + let awaitingBeginGeneration: number | null = null; + let acceptedSubmission: { generation: number; socketGeneration: number } | null = null; + let socketGeneration = 0; + let stateSocketGeneration = 0; + let disposed = false; function lastRowIdOfKind(kind: TranscriptRowKind): string | null { for (let i = rows.length - 1; i >= 0; i--) if (rows[i].kind === kind) return rows[i].id; @@ -138,7 +144,7 @@ export function createStructuredConnection( // request was in flight (superseded, cancelled, answered elsewhere), // the state-driven re-render already replaced the row; reviving // buttons on it would be reviving a decision that no longer applies. - if (rowId === pendingApprovalRowId) { + if (!disposed && rowId === pendingApprovalRowId) { showError(String(e)); throw e; } @@ -217,14 +223,28 @@ export function createStructuredConnection( } function renderComposerState(): void { - const active = state.turn.responding || state.turn.thinking; + const active = state.turn.responding || state.turn.thinking || awaitingBeginGeneration !== null; input.disabled = active; sendBtn.disabled = active; - cancelBtn.hidden = !active; + cancelBtn.hidden = !state.turn.responding && !state.turn.thinking; } - function applyState(next: ReducedState): void { + function reconcileAwaitingBegin(): void { + if (state.turn.responding || state.turn.thinking) { + awaitingBeginGeneration = null; + } else if ( + awaitingBeginGeneration !== null + && acceptedSubmission?.generation === awaitingBeginGeneration + && stateSocketGeneration > acceptedSubmission.socketGeneration + ) { + awaitingBeginGeneration = null; + } + } + + function applyState(next: ReducedState, connectionGeneration: number): void { state = next; + stateSocketGeneration = connectionGeneration; + reconcileAwaitingBegin(); const newPendingId = state.pending_approval ? lastRowIdOfKind("approval") : null; if (newPendingId !== pendingApprovalRowId) { const prev = pendingApprovalRowId; @@ -268,18 +288,44 @@ export function createStructuredConnection( e.preventDefault(); const text = input.value; if (!text.trim() || input.disabled) return; + const generation = ++submissionGeneration; + const submissionSocketGeneration = socketGeneration; + awaitingBeginGeneration = generation; + acceptedSubmission = null; input.value = ""; autoResize(); clearError(); - sendStructuredPrompt(sessionId, text).catch((e: unknown) => { - input.value = text; - autoResize(); - showError(String(e)); - }); + renderComposerState(); + sendStructuredPrompt(sessionId, text).then( + () => { + if ( + disposed + || generation !== submissionGeneration + || awaitingBeginGeneration !== generation + ) return; + const acceptanceSocketGeneration = socketGeneration; + acceptedSubmission = { generation, socketGeneration: acceptanceSocketGeneration }; + // A replacement socket may have captured idle before the POST claimed + // the prompt. Reconnect again so its snapshot cannot predate the 204. + if (acceptanceSocketGeneration > submissionSocketGeneration) reconnectForReconciliation(); + }, + (e: unknown) => { + if (disposed || generation !== submissionGeneration) return; + if (awaitingBeginGeneration === generation) awaitingBeginGeneration = null; + if (input.value === "") { + input.value = text; + autoResize(); + } + showError(String(e)); + renderComposerState(); + }, + ); }); cancelBtn.addEventListener("click", () => { clearError(); - cancelStructuredTurn(sessionId).catch((e: unknown) => showError(String(e))); + cancelStructuredTurn(sessionId).catch((e: unknown) => { + if (!disposed) showError(String(e)); + }); }); // ---- narrow-pane rail collapse (a split pane, not just the viewport) ---- @@ -292,23 +338,59 @@ export function createStructuredConnection( // ---- live worktree/git-status rail data ---- const unsubscribeStore = store?.subscribe((catalog: CatalogState) => { + if (disposed) return; latestSession = catalog.sessions.get(sessionId); renderRailNow(); }); // ---- socket lifecycle ---- - let ws: WebSocket; + let ws: WebSocket | null = null; let reconnectMs = INITIAL_RECONNECT_MS; let reconnectTimer: number | null = null; - let disposed = false; + + function cancelReconnect(): void { + if (reconnectTimer === null) return; + window.clearTimeout(reconnectTimer); + reconnectTimer = null; + } + + function closeCurrentSocket(): void { + const socket = ws; + ws = null; + if (!socket) return; + socket.onopen = null; + socket.onmessage = null; + socket.onerror = null; + socket.onclose = null; + try { + socket.close(); + } catch { + // ignore + } + } + + function reconnectForReconciliation(): void { + if (disposed) return; + cancelReconnect(); + closeCurrentSocket(); + setConnState("reconnecting…", "connecting"); + connect(); + } function connect(): void { - ws = socketFactory(sessionId); - ws.onopen = () => { + if (disposed) return; + cancelReconnect(); + closeCurrentSocket(); + const connectionGeneration = ++socketGeneration; + const socket = socketFactory(sessionId); + ws = socket; + socket.onopen = () => { + if (disposed || ws !== socket) return; reconnectMs = INITIAL_RECONNECT_MS; setConnState("connected", "open"); }; - ws.onmessage = (ev) => { + socket.onmessage = (ev) => { + if (disposed || ws !== socket) return; if (typeof ev.data !== "string") return; let msg: StructuredEvent; try { @@ -321,28 +403,33 @@ export function createStructuredConnection( applySnapshot(msg.snapshot); break; case "reduced_state": - applyState(msg.state); + applyState(msg.state, connectionGeneration); break; case "projection_update": applyDelta(msg.transcript); - applyState(msg.state); + applyState(msg.state, connectionGeneration); break; } }; - ws.onerror = () => setConnState("error", "error"); - ws.onclose = (ev) => { + socket.onerror = () => { + if (disposed || ws !== socket) return; + setConnState("error", "error"); + }; + socket.onclose = (ev) => { + if (disposed || ws !== socket) return; + ws = null; const reason = ev.reason?.trim(); - if (ev.code >= 4000 && reason && !disposed) { + if (ev.code >= 4000 && reason) { setConnState(reason, "error"); showError(`cannot attach: ${reason}`); return; } - if (disposed) return; if (opts.shouldReconnect?.(sessionId, ev.code)) { const delay = Math.min(reconnectMs, MAX_RECONNECT_MS); setConnState("reconnecting…", "connecting"); reconnectTimer = window.setTimeout(() => { reconnectTimer = null; + if (disposed) return; reconnectMs = Math.min(reconnectMs * 2, MAX_RECONNECT_MS); connect(); }, delay); @@ -368,7 +455,7 @@ export function createStructuredConnection( const focusOnMount = mountOpts?.focus !== false; const token = ++mountToken; requestAnimationFrame(() => { - if (token !== mountToken) return; // parked/remounted since this was scheduled + if (disposed || token !== mountToken) return; // parked/remounted since this was scheduled if (focusOnMount && !input.disabled) input.focus(); }); }, @@ -385,14 +472,11 @@ export function createStructuredConnection( dispose() { if (disposed) return; disposed = true; - if (reconnectTimer !== null) window.clearTimeout(reconnectTimer); + mountToken++; + cancelReconnect(); resizeObserver.disconnect(); unsubscribeStore?.(); - try { - ws.close(); - } catch { - // ignore - } + closeCurrentSocket(); el.remove(); }, };