fix(web): close stale structured sockets

crates/maestro-tui/src/app.rs

@@ -797,10 +797,14 @@ impl App {
 
     /// Open a stream for `sid` if one isn't already running.
     fn attach(&mut self, sid: Uuid) {
+        self.attach_with_view(sid, self.session_view(sid));
+    }
+
+    fn attach_with_view(&mut self, sid: Uuid, view: SessionView) {
         if self.is_attached(sid) {
             return;
         }
-        match self.session_view(sid) {
+        match view {
             SessionView::Terminal => match self.daemon.attach_url(sid) {
                 Ok(ws_url) => {
                     let handle = SessionTask::spawn(ws_url, 24, 80);
@@ -1493,7 +1497,7 @@ impl App {
                 if session.view == SessionView::Structured {
                     self.state.mode = Mode::Normal;
                 }
-                self.attach(sid);
+                self.attach_with_view(sid, session.view);
             }
             Err(e) => {
                 // A failed POST belongs in the form it came from (#64), not a
@@ -3021,6 +3025,53 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn create_response_selects_transport_before_catalog_event() {
+        use axum::routing::post;
+        use axum::Json;
+
+        async fn spawn_session(
+            Json(req): Json<CreateSessionRequest>,
+        ) -> Json<maestro_protocol::Session> {
+            let id = match req.view {
+                SessionView::Terminal => Uuid::from_u128(187),
+                SessionView::Structured => Uuid::from_u128(188),
+            };
+            let mut session = test_session(id, req.cwd.to_str().unwrap());
+            session.view = req.view;
+            session.command = req.command;
+            Json(session)
+        }
+
+        let router =
+            axum::Router::new().route(maestro_protocol::routes::SESSIONS, post(spawn_session));
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let base = format!("http://{}", listener.local_addr().unwrap());
+        tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
+
+        let mut app = make_app_at(&base);
+        let request = |view| CreateSessionRequest {
+            view,
+            cwd: std::path::PathBuf::from("/srv/work"),
+            command: "claude".into(),
+            env: Default::default(),
+            metadata: Default::default(),
+            worktree: None,
+        };
+
+        let structured_id = Uuid::from_u128(188);
+        app.submit_spawn(request(SessionView::Structured)).await;
+        assert!(app.catalog.snapshot().sessions.is_empty());
+        assert!(app.structured.contains_key(&structured_id));
+        assert!(app.emulators.get(structured_id).is_none());
+
+        let terminal_id = Uuid::from_u128(187);
+        app.submit_spawn(request(SessionView::Terminal)).await;
+        assert!(app.catalog.snapshot().sessions.is_empty());
+        assert!(app.emulators.get(terminal_id).is_some());
+        assert!(!app.structured.contains_key(&terminal_id));
+    }
+
     /// The full data load against a live daemon stub: presets replace the
     /// built-ins, projects land sorted, and the first generated name seeds
     /// the field.

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 });

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) => {
+    renderComposerState();
-      input.value = text;
+    sendStructuredPrompt(sessionId, text).then(
-      autoResize();
+      () => {
-      showError(String(e));
+        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);
+    if (disposed) return;
-    ws.onopen = () => {
+    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");
+    socket.onerror = () => {
-    ws.onclose = (ev) => {
+      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 {
+      closeCurrentSocket();
-        ws.close();
-      } catch {
-        // ignore
-      }
       el.remove();
     },
   };