fix(client): внутреннее ревью миграции FSM — F1..F4 (#488)

F1 [correctness] — supersede: перекрытие стримов рушило свежий run. ai@6 AbstractChat.makeRequest в finally читает и обнуляет общий activeResponse, поэтому параллельные стрим-A и стрим-B корёжат друг друга, а незастемпленный onFinish мёртвого A уводил ЖИВОЙ новый run в ложный reconnect / сбрасывал runFact. Фикс: (1) FINISH_*/STREAM_INCOMPLETE штампуются per-stream generation (turnEpochRef, взводится в момент старта стрима) → фильтр I1 отбрасывает финиш чужого поколения; (2) sendNow-supersede АБОРТИТ A и стартует B только из onFinish A (микротаск, после того как finally A обнулил activeResponse) — гарантия отсутствия перекрытия. Тест на поздний isDisconnect A после SUPERSEDE_REQUESTED: машина НЕ уходит в reconnect, B отправлен. MUTATION-VERIFY: снятие epoch-штампа у FINISH_DISCONNECT → тест краснеет («Connection lost — reconnecting»). F2 [correctness] — гонка mount getRun→ATTACH_START с локальным send. Редьюсер ATTACH_START теперь игнорирует любую не-idle фазу, поэтому поздний резолв getRun не перехватывает начавшийся локальный турн в observer-attach. Тест на гонку. F3 [ghost feature] — RUN_SUPERSEDED объявлен+покрыт тестом, но НИКОГДА не диспатчился. Удалён (событие+обработчик+тест+postRun-reason observer-follow) как сознательный scope-cut: наблюдатель убитого supersede-рана и так следует за новым через деградированный поллинг (свежие строки истории, независимо от runId). F4 [hygiene] — мёртвые события. STREAM_START подключён (первый ассистент-фрейм локального турна: sending→streaming, спека↔код совпали). RECONNECT_BEGIN и POLL_ACTIVITY удалены (не диспатчились). Множество событий редьюсера = множеству диспатчащихся; редьюсер тотален. Тесты: FSM 35 + chat-thread 37 (+error 26 +adopt 16) = 114 зелёных; tsc ai-chat 0. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-11 07:23:16 +03:00
parent 77e20ecb41
commit 4b78e336a2
5 changed files with 195 additions and 68 deletions
@@ -193,9 +193,9 @@ describe("ChatThread — send now", () => {
    expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" });
  });

-  it("#488 commit 5: autonomous live sendNow with a known runId POSTs a supersede body", () => {
-    // A streaming assistant message carrying the start-metadata runId -> the FSM
-    // adopts the run-fact; sendNow then interrupts via the server CAS.
+  // A streaming assistant message carrying the start-metadata runId -> the FSM
+  // adopts the run-fact so sendNow can interrupt via the server CAS.
+  const withRunId = () => {
    h.state.status = "streaming";
    h.state.messages = [
      {
@@ -205,19 +205,58 @@ describe("ChatThread — send now", () => {
        metadata: { runId: "run-1" },
      },
    ];
+  };
+
+  it("#488 commit 5 / F1: sendNow ABORTS stream A first, then sends B (CAS) only after A finalizes", async () => {
+    withRunId();
    renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() });
    fireEvent.click(screen.getByTestId("queue-btn"));
    fireEvent.click(screen.getByLabelText("Send now"));
-    // The message is sent...
+    // F1: A is ABORTED, and B is NOT sent yet (no overlap).
+    expect(h.state.stop).toHaveBeenCalledTimes(1);
+    expect(h.state.sendMessage).not.toHaveBeenCalled();
+    // A's onFinish fires -> B is scheduled on a microtask (after A finalizes).
+    await act(async () => {
+      h.state.onFinish?.({
+        message: { id: "a1", role: "assistant", parts: [] },
+        isAbort: true,
+        isDisconnect: false,
+        isError: false,
+      });
+      await Promise.resolve();
+    });
    expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" });
-    // ...and the NEXT POST carries supersede:{runId} (read-and-cleared once).
+    // B's POST carries supersede:{runId} (read-and-cleared once).
    const prep = h.state.transport!.prepareSendMessagesRequest!;
-    const body = prep({ messages: [], body: {} }).body as Record<string, unknown>;
-    expect(body.supersede).toEqual({ runId: "run-1" });
-    // One-shot: a subsequent request has no supersede.
+    expect(prep({ messages: [], body: {} }).body.supersede).toEqual({
+      runId: "run-1",
+    });
    expect(prep({ messages: [], body: {} }).body.supersede).toBeUndefined();
  });

+  it("#488 F1: a superseded stream's LATE disconnect does NOT falsely reconnect (epoch stamp drops it)", async () => {
+    // MUTATION-VERIFY: drop `epoch: stampEpoch` from the supersede-branch
+    // FINISH_DISCONNECT dispatch and this goes red (A's disconnect -> reconnecting).
+    withRunId();
+    renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() });
+    fireEvent.click(screen.getByTestId("queue-btn"));
+    fireEvent.click(screen.getByLabelText("Send now")); // -> superseding, aborts A
+    // A ends via isDisconnect (the server CAS closed it). The OLD overlap bug would
+    // route the LIVE new run into a false reconnect banner.
+    await act(async () => {
+      h.state.onFinish?.({
+        message: { id: "a1", role: "assistant", parts: [{ type: "text", text: "x" }] },
+        isAbort: false,
+        isDisconnect: true,
+        isError: false,
+      });
+      await Promise.resolve();
+    });
+    expect(screen.queryByText(/reconnecting/i)).toBeNull();
+    // ...and B was still sent.
+    expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" });
+  });
+
  it("#488 commit 5: a supersede POST that 409s SUPERSEDE_TIMEOUT is returned as-is (NO retry ladder)", async () => {
    h.state.status = "streaming";
    h.state.messages = [
@@ -494,6 +533,31 @@ describe("ChatThread — resume (attach) machinery", () => {
    expect(onResumeFallback).not.toHaveBeenCalledWith(true);
  });

+  it("#488 F2: a local send DURING the mount getRun round-trip is NOT hijacked by the late attach", async () => {
+    // getRun stays pending while the user sends locally; its late resolve would
+    // otherwise ATTACH_START and flip the local turn into an observer-attach.
+    let resolveGetRun!: (v: {
+      run: { id: string; status: string } | null;
+      message: unknown;
+    }) => void;
+    h.state.getRun.mockReturnValue(
+      new Promise((r) => {
+        resolveGetRun = r;
+      }),
+    );
+    renderThread({ autonomousRunsEnabled: true, initialRows: userTail() });
+    // Local send in flight of getRun -> SEND_LOCAL (phase sending, ownership local).
+    fireEvent.click(screen.getByTestId("send-btn"));
+    expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "typed text" });
+    // getRun resolves with an ACTIVE run -> the F2 guard ignores ATTACH_START.
+    await act(async () => {
+      resolveGetRun({ run: { id: "run-1", status: "running" }, message: null });
+      await Promise.resolve();
+    });
+    // The local turn was NOT hijacked into a resume/attach.
+    expect(h.state.resumeStream).not.toHaveBeenCalled();
+  });
+
  it("strips the streaming tail from the seed, keeps a user tail whole", () => {
    renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail() });
    expect(h.state.seededMessages).toHaveLength(1);
@@ -313,6 +313,18 @@ export default function ChatThread({
  // (like chatIdRef/roleIdRef), the single replacement for the three REMOVED
  // one-shot flags flushOnAbortRef/interruptNextSendRef/supersedeRetryRef.
  const pendingSupersedeRef = useRef<string | null>(null);
+  // #488 F1: the interrupt-and-send text, held until the superseded stream A has
+  // FULLY finalized (its SDK `finally` cleared `activeResponse`) — only THEN is
+  // stream B started, so A cannot corrupt B (ai@6 AbstractChat.makeRequest reads
+  // `this.activeResponse.state.message` in its finally and then nulls it, so an
+  // overlapping B is clobbered). Send-plumbing DATA, not a lifecycle flag.
+  const pendingSupersedeTextRef = useRef<string | null>(null);
+  // #488 F1: the epoch under which the CURRENTLY-OWNED stream started, used to
+  // STAMP its onFinish (I1). A superseded/dead stream's late finish carries an OLD
+  // generation and is dropped by the reducer, so it cannot drive the live machine
+  // into a false reconnect or reset its run-fact. Set at each honored stream start
+  // (local send, resume/reconnect attach, the supersede B-send).
+  const turnEpochRef = useRef(0);

  // --- Effect runner: executes the reducer's command effects ---------------
  const runEffect = useCallback(
@@ -324,6 +336,9 @@ export default function ChatThread({
          // seed already stripped it), so the live replay's text-start rebuilds it
          // without duplicating parts (#430).
          pendingAttachEpochRef.current = epoch;
+          // The resumed stream's onFinish is stamped with THIS attach generation
+          // (F1), so a superseded attempt's late finish is dropped.
+          turnEpochRef.current = epoch;
          if (machineRef.current.phase.name === "reconnecting") {
            const anchor = strippedRowRef.current;
            if (anchor)
@@ -415,6 +430,8 @@ export default function ChatThread({
  // FIFO dequeue + local send of the next queued message (no-op when empty).
  const localSend = useCallback((text: string) => {
    dispatchRef.current({ type: "SEND_LOCAL" });
+    // F1: this local stream's onFinish is stamped with the just-bumped generation.
+    turnEpochRef.current = epochRef.current;
    sendMessageRef.current?.({ text });
  }, []);
  const flushNext = useCallback(() => {
@@ -581,6 +598,10 @@ export default function ChatThread({
    transport,
    experimental_throttle: STREAM_THROTTLE_MS,
    onFinish: ({ message, isAbort, isDisconnect, isError }) => {
+      // #488 F1: STAMP this finish with the generation the stream STARTED under
+      // (I1). A superseded/dead stream's late finish carries an OLD generation and
+      // is dropped by the reducer, so it cannot drive the live machine.
+      const stampEpoch = turnEpochRef.current;
      // Ownership (I2) is the FSM ctx: a resumed/attached/reconnected turn is an
      // OBSERVER; a local send is the owner. The queue flushes ONLY under local
      // ownership; an observer never flushes (invariant 7).
@@ -589,22 +610,50 @@ export default function ChatThread({
      // for #161); fires even while unmounting.
      onTurnFinished(extractServerChatId(message), threadKey);

+      // A missing message (a pre-first-frame break) has no visible content.
+      const msgHasVisible = message
+        ? assistantMessageHasVisibleContent(message)
+        : false;
+
+      // #488 F1: the SUPERSEDED stream A just finalized (we are still `superseding`
+      // and stream B has not been sent yet). Record A's terminal outcome STAMPED —
+      // I1 drops it (superseding bumped the epoch), so A cannot drive a false
+      // reconnect / reset the run-fact — then start B NOW that A is fully done. B is
+      // deferred to a microtask so A's SDK `finally` (`activeResponse = void 0`)
+      // runs BEFORE B's makeRequest sets `activeResponse`, else the dying A clobbers
+      // B (ai@6). This is the no-overlap guarantee the CAS supersede needs.
+      if (
+        machineRef.current.phase.name === "superseding" &&
+        pendingSupersedeTextRef.current !== null
+      ) {
+        if (isError) dispatch({ type: "FINISH_ERROR", kind: "stream", epoch: stampEpoch });
+        else if (isAbort) dispatch({ type: "FINISH_ABORT", epoch: stampEpoch });
+        else if (isDisconnect)
+          dispatch({ type: "FINISH_DISCONNECT", hasVisibleContent: msgHasVisible, epoch: stampEpoch });
+        else dispatch({ type: "FINISH_CLEAN", epoch: stampEpoch });
+        const text = pendingSupersedeTextRef.current;
+        pendingSupersedeTextRef.current = null;
+        setStopNotice(null);
+        queueMicrotask(() => {
+          if (!mountedRef.current) return;
+          turnEpochRef.current = epochRef.current; // B's (superseding) generation
+          sendMessageRef.current?.({ text });
+        });
+        return;
+      }
+
      if (isError) {
-        dispatch({ type: "FINISH_ERROR", kind: "stream" });
+        dispatch({ type: "FINISH_ERROR", kind: "stream", epoch: stampEpoch });
        setStopNotice(null);
        return;
      }
      if (isAbort) {
        // A user Stop / an interrupt abort finished. The FSM stopping/idle exit is
        // by DATA (this terminal outcome, I4).
-        dispatch({ type: "FINISH_ABORT" });
+        dispatch({ type: "FINISH_ABORT", epoch: stampEpoch });
        setStopNotice("manual");
        return;
      }
-      // A missing message (a pre-first-frame break) has no visible content.
-      const msgHasVisible = message
-        ? assistantMessageHasVisibleContent(message)
-        : false;
      if (isDisconnect) {
        if (wasObserver) {
          // A resumed/attached OBSERVER stream dropped. Recover via the degraded
@@ -624,6 +673,7 @@ export default function ChatThread({
            dispatch({
              type: "FINISH_DISCONNECT",
              hasVisibleContent: hasVisible,
+              epoch: stampEpoch,
            });
          }
          setStopNotice(null);
@@ -659,10 +709,15 @@ export default function ChatThread({
          dispatch({
            type: "FINISH_DISCONNECT",
            hasVisibleContent: msgHasVisible,
+            epoch: stampEpoch,
          });
          setStopNotice(null);
        } else {
-          dispatch({ type: "FINISH_DISCONNECT", hasVisibleContent: false });
+          dispatch({
+            type: "FINISH_DISCONNECT",
+            hasVisibleContent: false,
+            epoch: stampEpoch,
+          });
          setStopNotice("disconnect");
        }
        return;
@@ -680,17 +735,17 @@ export default function ChatThread({
            queryClient.invalidateQueries({
              queryKey: AI_CHAT_MESSAGES_RQ_KEY(chatIdRef.current),
            });
-            dispatch({ type: "STREAM_INCOMPLETE", reason: "starved" });
+            dispatch({ type: "STREAM_INCOMPLETE", reason: "starved", epoch: stampEpoch });
          } else {
            // Healthy resumed finish — nothing to restore/arm, just settle.
-            dispatch({ type: "FINISH_CLEAN" });
+            dispatch({ type: "FINISH_CLEAN", epoch: stampEpoch });
          }
        }
        setStopNotice(null);
        return;
      }
      // Local clean finish: settle + flush the queue (gated on liveness, #486).
-      dispatch({ type: "FINISH_CLEAN" });
+      dispatch({ type: "FINISH_CLEAN", epoch: stampEpoch });
      setStopNotice(null);
      if (mountedRef.current) flushNext();
    },
@@ -730,9 +785,16 @@ export default function ChatThread({
      if (machineRef.current.phase.name === "stopping")
        onServerStopRef.current?.(serverChatId);
    }
+    // #488 F4: the FIRST assistant frame of a LOCAL turn moves the FSM
+    // `sending -> streaming` (and adopts the runId), matching the spec's
+    // STREAM_START transition. Later frames / observer turns (already streaming)
+    // just refresh the run-fact.
    const runId = extractRunId(tail);
-    if (runId && machineRef.current.ctx.runFact?.runId !== runId)
+    if (machineRef.current.phase.name === "sending") {
+      dispatch({ type: "STREAM_START", runId, epoch: epochRef.current });
+    } else if (runId && machineRef.current.ctx.runFact?.runId !== runId) {
      dispatch({ type: "RUN_FACT", runFact: { runId } });
+    }
  }, [messages, onServerChatId, dispatch]);

  // Live "turn was interrupted" marker (a manual Stop vs a dropped connection — a
@@ -839,9 +901,16 @@ export default function ChatThread({
        runId !== "pending"
      ) {
        // CAS supersede: the server stops the old run + starts this one atomically.
+        // #488 F1: do NOT start stream B synchronously here — the local stream A is
+        // still live, and overlapping streams corrupt each other in ai@6 (A's
+        // `finally` reads/ nulls the shared `activeResponse`). Instead ABORT A now
+        // and stash the text; A's onFinish (dropped by the epoch stamp) starts B in
+        // a microtask, AFTER A is fully finalized (no overlap). The POST then
+        // carries the supersede body (pendingSupersedeRef, armed by the effect).
        setQueue(removeQueuedById(queuedRef.current, id));
+        pendingSupersedeTextRef.current = msg.text;
        dispatch({ type: "SUPERSEDE_REQUESTED", targetRunId: runId });
-        sendMessageRef.current?.({ text: msg.text }); // POST carries supersede body
+        stopFnRef.current?.(); // abort A -> its onFinish sends B
        return;
      }
      if (liveStreaming) {
@@ -36,36 +36,47 @@ Legend: **†** = command-transition (bumps `epoch`, I1). Effects in `[…]`.
 | `FINISH_DISCONNECT` (no runFact, not liveFollow) | streaming | idle | runFact←null (plain terminal "connection lost") |
 | `STREAM_INCOMPLETE{reason}` (observer starved/torn clean finish) | streaming(observer) | polling(reason) | `[armPoll(reason)]` |
 | `FINISH_ERROR{kind}` (onFinish isError) | any | error(kind) | `[disarmPoll, cancelReconnect]`, runFact←null |
-| `ATTACH_START{runId}` (mount resume) | idle | attaching **†** | `[resumeStream]`, ownership=observer, runFact←runId |
+| `STREAM_START{runId}` (first assistant frame of a local turn) | sending | streaming | runFact←runId, `[cancelReconnect, disarmPoll]` |
+| `ATTACH_START{runId}` (mount resume) | **idle only** (F2) | attaching **†** | `[resumeStream]`, ownership=observer, runFact←runId; ignored from any non-idle phase |
 | `ATTACH_LIVE` (attach GET 2xx) | attaching | streaming | — |
 | `ATTACH_NONE` (attach GET 204/err/throw) | attaching | polling(attach-none) | `[armPoll(attach-none)]` |
-| `RECONNECT_BEGIN` | streaming, idle | reconnecting(1) **†** *iff runFact* | `[scheduleReconnect(1)]`, ownership=observer |
 | `RECONNECT_ATTEMPT{n}` (backoff timer) | reconnecting | reconnecting(n) **†** | `[resumeStream]` |
 | `RECONNECT_ATTACHED` (reconnect GET 2xx) | reconnecting | streaming | `[cancelReconnect, disarmPoll]` — **counter reset** (commit 3) |
 | `RECONNECT_NONE` (reconnect GET 204/err), attempt<MAX | reconnecting | reconnecting(n+1) **†** | `[armPoll(attach-none), scheduleReconnect(n+1)]` |
 | `RECONNECT_NONE`, attempt=MAX | reconnecting | reconnecting(MAX, failed) | `[armPoll(reconnect-exhausted)]` |
 | `RETRY` (manual, failed banner) | reconnecting(failed) | reconnecting(1) **†** | `[resumeStream]` |
 | `RETRY` (manual, stalled banner) | stalled | polling(attach-none) **†** | `[armPoll]` |
-| `POLL_ACTIVITY` (rows changed) | polling, reconnecting | (same) | — (runtime resets inactivity clock) |
 | `POLL_TERMINAL` (settled tail merged) | polling, reconnecting, stopping | idle | `[disarmPoll, cancelReconnect]`, runFact←null (I4) |
 | `POLL_IDLE_CAP` (inactivity cap) | polling, reconnecting | stalled | `[disarmPoll, cancelReconnect]` (commit 4a — no more silent) |
 | `RUN_FACT{null}` (POST /run → null/terminal, 204) | reconnecting/attaching/polling/stopping | idle | `[cancelReconnect, disarmPoll]`, runFact←null (I3 fresh-negative gate) |
 | `RUN_FACT{runId}` | any | (same) | runFact←runId (pessimism toward an attempt) |
-| `STOP_REQUESTED` (user Stop) | streaming, reconnecting, polling | stopping **†** | `[stopRun, abortAttach, cancelReconnect]` |
+| `STOP_REQUESTED` (user Stop) | streaming, reconnecting, polling | stopping **†** | `[stopRun, abortAttach, cancelReconnect, armPoll]` (poll drives the terminal — I4 exit by data) |
 | `SUPERSEDE_REQUESTED{targetRunId}` (interrupt+send) | streaming, reconnecting, polling, error | superseding **†** | `[supersede(target), cancelReconnect, disarmPoll]` |
 | `SUPERSEDE_READY{runId}` (CAS ok) | superseding | streaming | ownership=local, runFact←runId |
 | `SUPERSEDE_MISMATCH{currentRunId}` (409 SUPERSEDE_TARGET_MISMATCH) | superseding | error(supersede-mismatch) | `[postRun(verify)]`, runFact←currentRunId |
 | `SUPERSEDE_TIMEOUT` (409 SUPERSEDE_TIMEOUT) | superseding | error(supersede-timeout) | — (composer keeps text; no auto-retry) |
 | `SUPERSEDE_INVALID` (409 SUPERSEDE_INVALID) | superseding | error(supersede-invalid) | — |
 | `RUN_ALREADY_ACTIVE` (409 A_RUN_ALREADY_ACTIVE, plain POST) | sending | error(run-already-active) | — (composer offers supersede; NO auto-retry) |
-| `RUN_SUPERSEDED` (observer's run aborted by supersede) | streaming(observer), polling | attaching **†** | `[postRun(observer-follow)]`, ownership=observer |
 | `DISPOSE` (unmount) | any | idle **†** | `[abortAttach, cancelReconnect, disarmPoll]` (I1/I5 — epoch++ kills late callbacks) |

 **Epoch filter (I1):** the reducer FIRST drops any event carrying an `epoch` that
 does not equal the current `ctx.epoch`. Outcome events (`STREAM_START`, `ATTACH_*`,
-`RECONNECT_*`, `SUPERSEDE_*`, `FINISH_*`, `RUN_FACT`) are stamped with the epoch of
-the command that could produce them; trigger events (user actions, fresh
-disconnects) carry no epoch.
+`RECONNECT_*`, `SUPERSEDE_*`, **`FINISH_*`/`STREAM_INCOMPLETE`**, `RUN_FACT`) are
+stamped with the generation the corresponding STREAM started under (the runtime
+holds a per-owned-stream `turnEpoch`); trigger events (user actions, fresh
+disconnects) carry no epoch. **F1:** this is what makes a SUPERSEDED stream's late
+`onFinish` (a dead stream A closing after the CAS started stream B) get dropped, so
+A cannot drive the live new run into a false reconnect or reset its run-fact. The
+supersede path additionally ABORTS A and starts B only from A's onFinish (a
+microtask), because ai@6 `AbstractChat.makeRequest` corrupts overlapping streams
+(A's `finally` reads then nulls the shared `activeResponse`).
+
+**Removed events (scope-cut, internal review):** `RUN_SUPERSEDED` (a ghost feature —
+never dispatched; the observer-superseded case is handled by the degraded poll,
+which follows the latest rows regardless of runId), `RECONNECT_BEGIN` (reconnect is
+entered by `FINISH_DISCONNECT`), and `POLL_ACTIVITY` (the window's activity clock was
+removed when the idle-cap moved into the thread). The reducer and this table now
+share exactly the dispatched event set.

 ### 409-code → event map (the real #487 contract consumed here)

@@ -309,16 +309,23 @@ describe("run-fsm — commit 5: supersede CAS + error classification", () => {
    expect(m.phase).toEqual({ name: "error", kind: "run-already-active" });
    expect(m.effects).toEqual([]);
  });
+});

-  it("RUN_SUPERSEDED (observer's run killed) → attaching + postRun(observer-follow)", () => {
-    const ctx = { epoch: 0, ownership: "observer" as const, runFact: { runId: "run-dead" }, liveFollow: false };
-    const streaming: Machine = { phase: { name: "streaming" }, ctx, effects: [] };
-    const m = reduce(streaming, { type: "RUN_SUPERSEDED" });
+// #488 F2 — a late mount `getRun → ATTACH_START` must not hijack a local turn.
+describe("run-fsm — F2: ATTACH_START only from idle", () => {
+  it("ATTACH_START from a local `sending` turn is ignored (no observer hijack)", () => {
+    const sending = reduce(initialMachine(), { type: "SEND_LOCAL" }); // idle -> sending, local
+    const m = reduce(sending, { type: "ATTACH_START", runId: "r" });
+    expect(m.phase.name).toBe("sending");
+    expect(m.ctx.ownership).toBe("local"); // NOT flipped to observer
+    expect(m.effects).toEqual([]); // no resumeStream
+  });
+
+  it("ATTACH_START from idle attaches as normal", () => {
+    const m = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" });
    expect(m.phase.name).toBe("attaching");
-    expect(m.effects.find((e) => e.type === "postRun")).toEqual({
-      type: "postRun",
-      reason: "observer-follow",
-    });
+    expect(m.ctx.ownership).toBe("observer");
+    expect(hasEffect(m, "resumeStream")).toBe(true);
  });
 });

@@ -113,7 +113,7 @@ export interface Machine {

 export type Effect =
  /** POST /run to (re)establish or verify the run-fact. `reason` is diagnostic. */
-  | { type: "postRun"; reason: "mount" | "verify" | "observer-follow" }
+  | { type: "postRun"; reason: "mount" | "verify" }
  /** Trigger the SDK `resumeStream()` (attach GET via prepareReconnectToStream). */
  | { type: "resumeStream" }
  /** Schedule a reconnect attempt after a backoff, then dispatch RECONNECT_ATTEMPT. */
@@ -153,15 +153,12 @@ export type Event =
  | { type: "ATTACH_START"; runId?: string }
  | { type: "ATTACH_LIVE"; epoch?: number }
  | { type: "ATTACH_NONE"; epoch?: number }
-  // -- reconnect after a live disconnect --
-  /** #488 commit 2: entered by the RUN-FACT, not by an assistant message. */
-  | { type: "RECONNECT_BEGIN" }
+  // -- reconnect after a live disconnect (entered by FINISH_DISCONNECT, #488 c2) --
  | { type: "RECONNECT_ATTEMPT"; attempt: number; epoch?: number }
  | { type: "RECONNECT_ATTACHED"; epoch?: number }
  | { type: "RECONNECT_NONE"; epoch?: number }
  | { type: "RETRY" }
  // -- degraded poll --
-  | { type: "POLL_ACTIVITY" }
  | { type: "POLL_TERMINAL" }
  | { type: "POLL_IDLE_CAP" }
  // -- run-fact (server-confirmed active run) --
@@ -175,10 +172,6 @@ export type Event =
  | { type: "SUPERSEDE_TIMEOUT"; epoch?: number }
  | { type: "SUPERSEDE_INVALID"; epoch?: number }
  | { type: "RUN_ALREADY_ACTIVE" }
-  /** An observer's attached run was aborted because it was superseded server-side:
-   *  ask for the latest run and follow it (else an observer tab freezes on the
-   *  killed run). */
-  | { type: "RUN_SUPERSEDED" }
  // -- lifecycle --
  | { type: "DISPOSE" };

@@ -332,6 +325,12 @@ export function reduce(m: Machine, event: Event): Machine {
    // ---- mount attach (resume) ----------------------------------------
    case "ATTACH_START":
      // A reopened tab attaches to a still-running run: observer ownership.
+      // #488 F2: ONLY from idle. The mount `getRun` round-trip resolves async, and
+      // a local send may have started meanwhile (phase `sending`, ownership local);
+      // a late ATTACH_START must NOT hijack that local turn into an observer-attach
+      // (queue would stop flushing, "Send now" would hide). Guarding in the reducer
+      // covers every dispatch source.
+      if (m.phase.name !== "idle") return stay(m);
      return command(m, { name: "attaching" }, [{ type: "resumeStream" }], {
        ownership: "observer",
        runFact: event.runId ? { runId: event.runId } : m.ctx.runFact,
@@ -351,17 +350,6 @@ export function reduce(m: Machine, event: Event): Machine {
      });

    // ---- reconnect after a live disconnect ----------------------------
-    case "RECONNECT_BEGIN":
-      // Explicit begin (used when the runtime decides to reconnect from a signal
-      // other than FINISH_DISCONNECT). Requires a run-fact (I3).
-      if (!m.ctx.runFact) return stay(m);
-      return command(
-        m,
-        { name: "reconnecting", attempt: 1, failed: false },
-        [{ type: "scheduleReconnect", attempt: 1, delayMs: reconnectDelayMs(1) }],
-        { ownership: "observer" },
-      );
-
    case "RECONNECT_ATTEMPT":
      // A scheduled backoff fired — fire the attach GET. epoch++ so the previous
      // attempt's late outcome cannot drive this one.
@@ -420,11 +408,6 @@ export function reduce(m: Machine, event: Event): Machine {
      return stay(m);

    // ---- degraded poll -------------------------------------------------
-    case "POLL_ACTIVITY":
-      // The polled rows changed (progress). No phase change — the runtime resets
-      // its inactivity clock. Only meaningful while a poll-bearing phase is active.
-      return stay(m);
-
    case "POLL_TERMINAL":
      // The run reached a terminal row via the poll (or the reconcile merge). Go
      // idle and disarm everything (I4: this is a DATA-driven exit, incl. exit
@@ -527,13 +510,6 @@ export function reduce(m: Machine, event: Event): Machine {
      // offers "interrupt and send" (supersede) instead.
      return to(m, { name: "error", kind: "run-already-active" });

-    case "RUN_SUPERSEDED":
-      // An observer's attached run was killed by a server-side supersede. Ask for
-      // the latest run and follow it, instead of freezing on the dead run.
-      return command(m, { name: "attaching" }, [{ type: "postRun", reason: "observer-follow" }], {
-        ownership: "observer",
-      });
-
    // ---- lifecycle -----------------------------------------------------
    case "DISPOSE":
      // Unmount: abort in-flight controllers, drop timers, and bump the epoch so