diff --git a/apps/client/src/features/ai-chat/state/run-fsm.spec.md b/apps/client/src/features/ai-chat/state/run-fsm.spec.md new file mode 100644 index 00000000..d7c7557c --- /dev/null +++ b/apps/client/src/features/ai-chat/state/run-fsm.spec.md @@ -0,0 +1,161 @@ +# AI-chat run-lifecycle FSM — design spec (#488) + +This is the written design that `run-fsm.ts` implements. It ships in the PR (issue +#488 commit 1: "the spec is written FIRST and enters the PR"). It has four parts: +(1) the event × state transition table, (2) the map of every `chat-thread.tsx` ref +to {FSM state | FSM context | stays data}, (3) the run-fact protocol, (4) the +invariants. + +The reducer is a **pure function** `reduce(machine, event) → machine`. The returned +machine carries the **command effects** for that transition; a thin runtime in +`chat-thread.tsx` dispatches events and executes effects. Because it is pure, the +whole machine is enumerable and unit-tested directly (event × state → next state is +the observable property) — see `run-fsm.test.ts`. + +--- + +## 1. Event × state transition table + +Phases: `idle | sending | streaming | attaching | reconnecting(attempt,failed) | +polling(reason) | stalled | stopping | superseding | error(kind)`. +Context (orthogonal): `epoch`, `ownership: local|observer`, `runFact: {runId}|null`. + +Legend: **†** = command-transition (bumps `epoch`, I1). Effects in `[…]`. + +| Event (source) | From phase(s) | → To phase | Effects / ctx | +|---|---|---|---| +| `SEND_LOCAL` (user send) | idle, error, polling, stalled, reconnecting | sending **†** | `[cancelReconnect, disarmPoll]`, ownership=local | +| `STREAM_START{runId}` (SDK `start` metadata) | sending, attaching, reconnecting, superseding | streaming | `[cancelReconnect, disarmPoll]`, runFact←runId | +| `FINISH_CLEAN` (onFinish clean) | streaming, … | idle | `[disarmPoll, cancelReconnect]`, runFact←null | +| `FINISH_ABORT` (onFinish isAbort) | streaming, stopping | idle | `[disarmPoll, cancelReconnect]`, runFact←null (I4 exits stopping by this DATA) | +| `FINISH_DISCONNECT{hasVisibleContent}` (onFinish isDisconnect) | streaming | reconnecting(1) **†** *iff runFact* | `[scheduleReconnect(1)]` (+`armPoll(disconnect-visible)` if visible), ownership=observer | +| `FINISH_DISCONNECT` (no runFact) | streaming | idle | runFact←null (plain terminal "connection lost") | +| `FINISH_ERROR{kind}` (onFinish isError) | any | error(kind) | `[disarmPoll, cancelReconnect]`, runFact←null | +| `ATTACH_START{runId}` (mount resume) | idle | attaching **†** | `[resumeStream]`, ownership=observer, runFact←runId | +| `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 e.type); +} +function hasEffect(m: Machine, type: Effect["type"]): boolean { + return m.effects.some((e) => e.type === type); +} + +describe("run-fsm — epoch invariant (I1)", () => { + it("drops an outcome carrying a stale epoch", () => { + // A command bumps the epoch; an outcome stamped with the OLD epoch is dropped. + const m0 = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" }); // epoch 0->1, attaching + expect(m0.ctx.epoch).toBe(1); + expect(m0.phase.name).toBe("attaching"); + // A late ATTACH_LIVE from a SUPERSEDED attempt (epoch 0) must NOT drive us. + const stale = reduce(m0, { type: "ATTACH_LIVE", epoch: 0 }); + expect(stale.phase.name).toBe("attaching"); + expect(stale.effects).toEqual([]); + }); + + it("applies an outcome carrying the current epoch", () => { + const m0 = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" }); + const live = reduce(m0, { type: "ATTACH_LIVE", epoch: m0.ctx.epoch }); + expect(live.phase.name).toBe("streaming"); + }); + + it("an outcome with no epoch is never dropped (trigger events)", () => { + const m0 = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" }); + const disposed = reduce(m0, { type: "DISPOSE" }); + expect(disposed.phase.name).toBe("idle"); + expect(hasEffect(disposed, "abortAttach")).toBe(true); + }); + + it("every command-transition increments the epoch exactly once", () => { + let m = initialMachine(); + const before = m.ctx.epoch; + m = reduce(m, { type: "SEND_LOCAL" }); + expect(m.ctx.epoch).toBe(before + 1); + m = reduce(m, { type: "STOP_REQUESTED" }); + expect(m.ctx.epoch).toBe(before + 2); + }); +}); + +describe("run-fsm — local turn", () => { + it("SEND_LOCAL → sending, local ownership, cancels recovery", () => { + const m = reduce(withRunFact(), { type: "SEND_LOCAL" }); + expect(m.phase.name).toBe("sending"); + expect(m.ctx.ownership).toBe("local"); + expect(effectTypes(m)).toEqual( + expect.arrayContaining(["cancelReconnect", "disarmPoll"]), + ); + }); + + it("STREAM_START adopts the runId into the run-fact and goes streaming", () => { + const m = run(initialMachine(), { type: "SEND_LOCAL" }); + const s = reduce(m, { type: "STREAM_START", runId: "run-9", epoch: m.ctx.epoch }); + expect(s.phase.name).toBe("streaming"); + expect(s.ctx.runFact).toEqual({ runId: "run-9" }); + }); + + it("FINISH_CLEAN → idle, run-fact cleared, poll/reconnect disarmed", () => { + const streaming = run(initialMachine(), { type: "SEND_LOCAL" }, { type: "STREAM_START", runId: "r" }); + const done = reduce(streaming, { type: "FINISH_CLEAN" }); + expect(done.phase.name).toBe("idle"); + expect(done.ctx.runFact).toBeNull(); + }); +}); + +// #488 commit 2 — SSE break BEFORE the first assistant frame must still recover. +describe("run-fsm — commit 2: reconnect by run-fact, not by assistant message", () => { + it("FINISH_DISCONNECT with an active run-fact → reconnecting (even with no visible content)", () => { + // Setup-phase break: no assistant frame yet, but a run-fact exists. + const streaming = withRunFact("run-2"); + const m = reduce(streaming, { + type: "FINISH_DISCONNECT", + hasVisibleContent: false, + epoch: streaming.ctx.epoch, + }); + expect(m.phase.name).toBe("reconnecting"); + if (m.phase.name === "reconnecting") expect(m.phase.attempt).toBe(1); + expect(m.ctx.ownership).toBe("observer"); + expect(hasEffect(m, "scheduleReconnect")).toBe(true); + // No visible content -> no poll arm yet (the reconnect ladder rebuilds it). + expect(hasEffect(m, "armPoll")).toBe(false); + }); + + it("FINISH_DISCONNECT WITH visible content also arms the poll", () => { + const m = reduce(withRunFact("run-2"), { + type: "FINISH_DISCONNECT", + hasVisibleContent: true, + epoch: 0, + }); + expect(m.phase.name).toBe("reconnecting"); + expect(hasEffect(m, "armPoll")).toBe(true); + }); + + it("FINISH_DISCONNECT with NO run-fact → idle (plain connection-lost)", () => { + const m = reduce(initialMachine(), { + type: "FINISH_DISCONNECT", + hasVisibleContent: true, + epoch: 0, + }); + expect(m.phase.name).toBe("idle"); + }); +}); + +// #488 commit 3 — a SECOND break after a successful re-attach starts a NEW ladder. +describe("run-fsm — commit 3: repeated reconnect cycles", () => { + it("two breaks in a row produce two reconnect cycles (counter resets on attach)", () => { + let m = withRunFact("run-3"); + // First break -> reconnecting(1). + m = reduce(m, { type: "FINISH_DISCONNECT", hasVisibleContent: false, epoch: m.ctx.epoch }); + expect(m.phase.name).toBe("reconnecting"); + // Attempt fires, re-attaches live. + m = reduce(m, { type: "RECONNECT_ATTEMPT", attempt: 1, epoch: m.ctx.epoch }); + m = reduce(m, { type: "RECONNECT_ATTACHED", epoch: m.ctx.epoch }); + expect(m.phase.name).toBe("streaming"); + // SECOND break: the counter was reset, so a fresh ladder starts at attempt 1 + // (the old one-shot !wasResumed gate would have sent this to silent poll). + m = reduce(m, { type: "FINISH_DISCONNECT", hasVisibleContent: false, epoch: m.ctx.epoch }); + expect(m.phase.name).toBe("reconnecting"); + if (m.phase.name === "reconnecting") expect(m.phase.attempt).toBe(1); + expect(hasEffect(m, "scheduleReconnect")).toBe(true); + }); + + it("RECONNECT_NONE backs off through the ladder, then fails at the cap", () => { + let m = withRunFact("run-3"); + m = reduce(m, { type: "FINISH_DISCONNECT", hasVisibleContent: false, epoch: m.ctx.epoch }); + for (let n = 1; n < RECONNECT_MAX_ATTEMPTS; n++) { + m = reduce(m, { type: "RECONNECT_ATTEMPT", attempt: n, epoch: m.ctx.epoch }); + m = reduce(m, { type: "RECONNECT_NONE", epoch: m.ctx.epoch }); + expect(m.phase.name).toBe("reconnecting"); + if (m.phase.name === "reconnecting") { + expect(m.phase.attempt).toBe(n + 1); + expect(m.phase.failed).toBe(false); + } + // The belt-and-suspenders poll is armed each failed attempt. + expect(hasEffect(m, "armPoll")).toBe(true); + } + // Final attempt fails -> failed banner (Retry), poll armed. + m = reduce(m, { type: "RECONNECT_ATTEMPT", attempt: RECONNECT_MAX_ATTEMPTS, epoch: m.ctx.epoch }); + m = reduce(m, { type: "RECONNECT_NONE", epoch: m.ctx.epoch }); + expect(m.phase.name).toBe("reconnecting"); + if (m.phase.name === "reconnecting") expect(m.phase.failed).toBe(true); + // RETRY restarts at attempt 1. + m = reduce(m, { type: "RETRY" }); + expect(m.phase.name).toBe("reconnecting"); + if (m.phase.name === "reconnecting") { + expect(m.phase.attempt).toBe(1); + expect(m.phase.failed).toBe(false); + } + expect(hasEffect(m, "resumeStream")).toBe(true); + }); + + it("reconnectDelayMs is the exponential backoff 1s,2s,4s,8s,16s", () => { + expect([1, 2, 3, 4, 5].map(reconnectDelayMs)).toEqual([1000, 2000, 4000, 8000, 16000]); + }); +}); + +// #488 commit 4 — polling stalled-state + user-tail gating. +describe("run-fsm — commit 4: stalled + run-fact gating", () => { + it("POLL_IDLE_CAP: polling → stalled with a banner (poll disarmed), not silent", () => { + let m = reduce(withRunFact(), { type: "ATTACH_START", runId: "r" }); + m = reduce(m, { type: "ATTACH_NONE", epoch: m.ctx.epoch }); + expect(m.phase.name).toBe("polling"); + m = reduce(m, { type: "POLL_IDLE_CAP" }); + expect(m.phase.name).toBe("stalled"); + expect(hasEffect(m, "disarmPoll")).toBe(true); + }); + + it("RETRY from stalled re-arms the poll", () => { + let m = reduce(withRunFact(), { type: "ATTACH_START", runId: "r" }); + m = reduce(m, { type: "ATTACH_NONE", epoch: m.ctx.epoch }); + m = reduce(m, { type: "POLL_IDLE_CAP" }); + m = reduce(m, { type: "RETRY" }); + expect(m.phase.name).toBe("polling"); + expect(hasEffect(m, "armPoll")).toBe(true); + }); + + it("a fresh NEGATIVE run-fact while attaching cancels recovery (user-tail, no active run)", () => { + // The mount POST /run returns no active run: attaching → idle, no poll armed. + let m = reduce(withRunFact(), { type: "ATTACH_START", runId: "r" }); + m = reduce(m, { type: "RUN_FACT", runFact: null, epoch: m.ctx.epoch }); + expect(m.phase.name).toBe("idle"); + expect(m.ctx.runFact).toBeNull(); + expect(hasEffect(m, "disarmPoll")).toBe(true); + }); + + it("a negative run-fact while polling stops the poll", () => { + let m = reduce(withRunFact(), { type: "ATTACH_START", runId: "r" }); + m = reduce(m, { type: "ATTACH_NONE", epoch: m.ctx.epoch }); + m = reduce(m, { type: "RUN_FACT", runFact: null, epoch: m.ctx.epoch }); + expect(m.phase.name).toBe("idle"); + }); + + it("POLL_TERMINAL settles polling → idle (I4 data-driven exit)", () => { + let m = reduce(withRunFact(), { type: "ATTACH_START", runId: "r" }); + m = reduce(m, { type: "ATTACH_NONE", epoch: m.ctx.epoch }); + m = reduce(m, { type: "POLL_TERMINAL" }); + expect(m.phase.name).toBe("idle"); + expect(m.ctx.runFact).toBeNull(); + }); +}); + +// #488 commit 5 — error classification + supersede CAS transitions. +describe("run-fsm — commit 5: supersede CAS + error classification", () => { + it("SUPERSEDE_REQUESTED → superseding, fires the CAS effect, bumps epoch", () => { + const streaming = withRunFact("run-old"); + const m = reduce(streaming, { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" }); + expect(m.phase.name).toBe("superseding"); + expect(m.ctx.epoch).toBe(streaming.ctx.epoch + 1); + const sup = m.effects.find((e) => e.type === "supersede"); + expect(sup).toEqual({ type: "supersede", targetRunId: "run-old" }); + }); + + it("SUPERSEDE_READY → streaming as the new local owner", () => { + let m = reduce(withRunFact("run-old"), { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" }); + m = reduce(m, { type: "SUPERSEDE_READY", runId: "run-new", epoch: m.ctx.epoch }); + expect(m.phase.name).toBe("streaming"); + expect(m.ctx.ownership).toBe("local"); + expect(m.ctx.runFact).toEqual({ runId: "run-new" }); + }); + + it("SUPERSEDE_MISMATCH → error(supersede-mismatch) + verify via /run (no blind banner)", () => { + let m = reduce(withRunFact("run-old"), { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" }); + m = reduce(m, { type: "SUPERSEDE_MISMATCH", currentRunId: "run-x", epoch: m.ctx.epoch }); + expect(m.phase).toEqual({ name: "error", kind: "supersede-mismatch" }); + expect(hasEffect(m, "postRun")).toBe(true); + expect(m.ctx.runFact).toEqual({ runId: "run-x" }); + }); + + it("SUPERSEDE_TIMEOUT → error(supersede-timeout), no auto-retry effect", () => { + let m = reduce(withRunFact("run-old"), { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" }); + m = reduce(m, { type: "SUPERSEDE_TIMEOUT", epoch: m.ctx.epoch }); + expect(m.phase).toEqual({ name: "error", kind: "supersede-timeout" }); + expect(m.effects).toEqual([]); + }); + + it("SUPERSEDE_INVALID → error(supersede-invalid)", () => { + let m = reduce(withRunFact("run-old"), { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" }); + m = reduce(m, { type: "SUPERSEDE_INVALID", epoch: m.ctx.epoch }); + expect(m.phase).toEqual({ name: "error", kind: "supersede-invalid" }); + }); + + it("a stale SUPERSEDE outcome from a superseded epoch is dropped", () => { + let m = reduce(withRunFact("run-old"), { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" }); + const supersedingEpoch = m.ctx.epoch; + // The user retriggers, bumping the epoch again. + m = reduce(m, { type: "SUPERSEDE_REQUESTED", targetRunId: "run-old" }); + // The first CAS's late TIMEOUT (old epoch) must NOT knock us out of superseding. + const late = reduce(m, { type: "SUPERSEDE_TIMEOUT", epoch: supersedingEpoch }); + expect(late.phase.name).toBe("superseding"); + }); + + it("RUN_ALREADY_ACTIVE (plain POST gate) → error(run-already-active), no retry effect", () => { + const m = reduce(run(initialMachine(), { type: "SEND_LOCAL" }), { type: "RUN_ALREADY_ACTIVE" }); + 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 observer = { ...withRunFact("run-dead"), ctx: { epoch: 0, ownership: "observer" as const, runFact: { runId: "run-dead" } } }; + const streaming: Machine = { phase: { name: "streaming" }, ctx: observer.ctx, effects: [] }; + const m = reduce(streaming, { type: "RUN_SUPERSEDED" }); + expect(m.phase.name).toBe("attaching"); + expect(m.effects.find((e) => e.type === "postRun")).toEqual({ + type: "postRun", + reason: "observer-follow", + }); + }); +}); + +describe("run-fsm — stop (I4: exit by data)", () => { + it("STOP_REQUESTED → stopping, fires stopRun + abortAttach, no data-independent exit", () => { + const m = reduce(withRunFact(), { type: "STOP_REQUESTED" }); + expect(m.phase.name).toBe("stopping"); + expect(effectTypes(m)).toEqual(expect.arrayContaining(["stopRun", "abortAttach"])); + }); + + it("stopping exits on a terminal FINISH_ABORT (data), not on the stopRun response", () => { + let m = reduce(withRunFact(), { type: "STOP_REQUESTED" }); + // No transition keys off the HTTP result — only the terminal finish moves us. + m = reduce(m, { type: "FINISH_ABORT", epoch: m.ctx.epoch }); + expect(m.phase.name).toBe("idle"); + expect(m.ctx.runFact).toBeNull(); + }); + + it("stopping exits on a negative run-fact (data)", () => { + let m = reduce(withRunFact(), { type: "STOP_REQUESTED" }); + m = reduce(m, { type: "RUN_FACT", runFact: null, epoch: m.ctx.epoch }); + expect(m.phase.name).toBe("idle"); + }); +}); + +describe("run-fsm — ownership (I2) is context, orthogonal to phase", () => { + it("attach/reconnect set observer; send/supersede-ready set local", () => { + let m = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" }); + expect(m.ctx.ownership).toBe("observer"); + m = reduce(m, { type: "ATTACH_LIVE", epoch: m.ctx.epoch }); + expect(m.phase.name).toBe("streaming"); + expect(m.ctx.ownership).toBe("observer"); // still observing a detached run + // A local send flips ownership back to local. + m = reduce(m, { type: "SEND_LOCAL" }); + expect(m.ctx.ownership).toBe("local"); + }); +}); + +describe("run-fsm — dispose (I5)", () => { + it("DISPOSE from any phase aborts controllers and bumps epoch", () => { + let m = reduce(withRunFact(), { type: "ATTACH_START", runId: "r" }); + const before = m.ctx.epoch; + m = reduce(m, { type: "DISPOSE" }); + expect(m.phase.name).toBe("idle"); + expect(m.ctx.epoch).toBe(before + 1); + expect(effectTypes(m)).toEqual( + expect.arrayContaining(["abortAttach", "cancelReconnect", "disarmPoll"]), + ); + }); +}); diff --git a/apps/client/src/features/ai-chat/state/run-fsm.ts b/apps/client/src/features/ai-chat/state/run-fsm.ts new file mode 100644 index 00000000..89348384 --- /dev/null +++ b/apps/client/src/features/ai-chat/state/run-fsm.ts @@ -0,0 +1,513 @@ +/** + * Run-lifecycle finite state machine for a single AI-chat thread (#488). + * + * ============================================================================ + * WHY THIS EXISTS + * ---------------------------------------------------------------------------- + * The resume/reconnect/poll/stop/supersede lifecycle used to be spread across + * ~26 `useRef` one-shot flags in `chat-thread.tsx`, each disarmed "on every + * path". Ownerless flag combinations produced silent UI freezes, and every fix + * added another ref (the #381 -> #432 -> #456 spiral). This module replaces that + * ref-zoo with ONE pure reducer whose transitions are enumerable and unit- + * testable in isolation (event x state -> next state is the observable property). + * + * The reducer is PURE: it owns no timers, no fetches, no React state. It maps + * `(machine, event) -> machine`, where the returned machine carries the list of + * COMMAND EFFECTS to run for that transition. A thin runtime in `chat-thread.tsx` + * dispatches events (from SDK callbacks / HTTP outcomes) and executes the + * effects (attach GET, POST /stream, POST /run, POST /stop, backoff timers, + * poll arm/disarm). The runtime lives in a THREAD, not the window, so a late SDK + * callback dies with the owner (kills the "event from a dead view" class, #161). + * + * ============================================================================ + * INVARIANTS (see run-fsm.spec.md for the full spec + tables) + * ---------------------------------------------------------------------------- + * I1 EPOCH (generation counter). Commands (`resumeStream`, `postRun`, `stop`, + * `supersede`, `scheduleReconnect`) are async; their outcomes arrive on the + * SAME SDK/HTTP callbacks. Every command-emitting transition increments + * `ctx.epoch`; every OUTCOME event carries the epoch it was issued under; + * the reducer DROPS an outcome whose epoch != the current epoch. This is + * what the one-shot-ref zoo used to approximate by hand. + * I2 OWNERSHIP is a CONTEXT FIELD (`'local' | 'observer'`), not a state — + * orthogonal to the transport phase. The queue is flushed ONLY by a local + * owner (an observer following a detached run never flushes). + * I3 RUN-FACT ("a run is active") is first-class from the server: `runFact` + * holds the server-confirmed active run id (POST /run on mount, the `start` + * metadata runId, attach outcomes). Reconnect is entered by the RUN-FACT, + * not by the presence of an assistant message (#488 commit 2). A fresh + * negative fact (null) cancels reconnect immediately. + * I4 Exit `stopping` by DATA (a terminal row / negative run-fact), NEVER by the + * stopRun HTTP response (which returns after abort, before finalization). + * I5 Command controllers are effect-owned (abort in cleanup), NOT render-phase + * refs — expressed here as the `abortAttach` effect on disposing transitions. + * ============================================================================ + */ + +// --------------------------------------------------------------------------- +// Phases (the transport lifecycle). Ownership / runFact are CONTEXT, not here. +// --------------------------------------------------------------------------- + +/** Why the degraded poll is the active recovery. */ +export type PollReason = + | "attach-none" // mount attach returned 204 / error — nothing live to attach + | "starved" // a resumed finish carried no visible content + | "disconnect-visible" // a live disconnect WITH on-screen content — poll to terminal + | "reconnect-exhausted"; // the live re-attach ladder gave up + +/** The classified error kind (drives the banner text + composer behavior). */ +export type ErrorKind = + | "stream" // a generic provider/network stream error (useChat error) + | "run-already-active" // 409 A_RUN_ALREADY_ACTIVE (a plain POST hit the gate) + | "supersede-mismatch" // 409 SUPERSEDE_TARGET_MISMATCH (CAS target moved) + | "supersede-timeout" // 409 SUPERSEDE_TIMEOUT (old run did not settle in W) + | "supersede-invalid" // 409 SUPERSEDE_INVALID (bad supersede target) + | "begin-failed"; // 503 A_RUN_BEGIN_FAILED (could not start the run) + +export type Phase = + | { name: "idle" } + | { name: "sending" } // local POST in flight, before the first frame + | { name: "streaming" } // receiving frames + | { name: "attaching" } // mount-time attach GET in flight + | { name: "reconnecting"; attempt: number; failed: boolean } + | { name: "polling"; reason: PollReason } + | { name: "stalled" } // poll hit the inactivity cap — banner + Retry + | { name: "stopping" } + | { name: "superseding" } + | { name: "error"; kind: ErrorKind }; + +export type Ownership = "local" | "observer"; + +/** The server-confirmed active run, or null when no run is active. */ +export type RunFact = { runId: string } | null; + +export interface Ctx { + /** I1: generation counter — every command-transition increments it. */ + epoch: number; + /** I2: does THIS client own the turn's writes (local streamer) or observe? */ + ownership: Ownership; + /** I3: the server-confirmed active run. */ + runFact: RunFact; +} + +export interface Machine { + phase: Phase; + ctx: Ctx; + /** Command effects to run for the transition that produced THIS machine. + * The runtime executes them and does not read them again. */ + effects: Effect[]; +} + +// --------------------------------------------------------------------------- +// Command effects (the reducer's only side-channel — executed by the runtime). +// --------------------------------------------------------------------------- + +export type Effect = + /** POST /run to (re)establish or verify the run-fact. `reason` is diagnostic. */ + | { type: "postRun"; reason: "mount" | "verify" | "observer-follow" } + /** Trigger the SDK `resumeStream()` (attach GET via prepareReconnectToStream). */ + | { type: "resumeStream" } + /** Schedule a reconnect attempt after a backoff, then dispatch RECONNECT_ATTEMPT. */ + | { type: "scheduleReconnect"; attempt: number; delayMs: number } + /** Cancel any pending reconnect backoff timer. */ + | { type: "cancelReconnect" } + /** Arm the degraded poll (the window's dumb timer follows the run in the DB). */ + | { type: "armPoll"; reason: PollReason } + /** Disarm the degraded poll. */ + | { type: "disarmPoll" } + /** POST /stop the chat's active run (authoritative detached-run stop). */ + | { type: "stopRun" } + /** POST /stream { supersede: { runId } } — the CAS "interrupt and send now". */ + | { type: "supersede"; targetRunId: string } + /** Abort the in-flight attach/reconnect GET controller (dispose / observer stop). */ + | { type: "abortAttach" }; + +// --------------------------------------------------------------------------- +// Events. An OUTCOME event MAY carry `epoch`; if it does and it does not equal +// the current epoch, the reducer drops it (I1). Trigger events (user actions, +// fresh disconnects) carry no epoch and are never dropped. +// --------------------------------------------------------------------------- + +export type Event = + // -- local turn -- + | { type: "SEND_LOCAL" } + | { type: "STREAM_START"; runId?: string; epoch?: number } + | { type: "FINISH_CLEAN"; epoch?: number } + | { type: "FINISH_ABORT"; epoch?: number } + | { type: "FINISH_DISCONNECT"; hasVisibleContent: boolean; epoch?: number } + | { type: "FINISH_ERROR"; kind: ErrorKind; epoch?: number } + // -- mount attach (resume) -- + | { 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" } + | { 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) -- + | { type: "RUN_FACT"; runFact: RunFact; epoch?: number } + // -- stop -- + | { type: "STOP_REQUESTED" } + // -- supersede (CAS) -- + | { type: "SUPERSEDE_REQUESTED"; targetRunId: string } + | { type: "SUPERSEDE_READY"; runId?: string; epoch?: number } + | { type: "SUPERSEDE_MISMATCH"; currentRunId?: string; epoch?: number } + | { 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" }; + +export const RECONNECT_MAX_ATTEMPTS = 5; +export const RECONNECT_BASE_DELAY_MS = 1000; +/** Backoff before attempt N (1-based): 1s, 2s, 4s, 8s, 16s. */ +export function reconnectDelayMs(attempt: number): number { + return RECONNECT_BASE_DELAY_MS * 2 ** (attempt - 1); +} + +// --------------------------------------------------------------------------- +// Constructors / helpers. +// --------------------------------------------------------------------------- + +export function initialMachine(overrides?: Partial): Machine { + return { + phase: { name: "idle" }, + ctx: { epoch: 0, ownership: "local", runFact: null, ...overrides }, + effects: [], + }; +} + +/** Build a machine result: a phase, optional ctx patch, and effects. Empty + * effects by default. Never mutates the input. */ +function to( + m: Machine, + phase: Phase, + opts?: { ctx?: Partial; effects?: Effect[] }, +): Machine { + return { + phase, + ctx: { ...m.ctx, ...(opts?.ctx ?? {}) }, + effects: opts?.effects ?? [], + }; +} + +/** No transition: keep the phase, clear effects (so a re-run does not re-fire). */ +function stay(m: Machine): Machine { + return { phase: m.phase, ctx: m.ctx, effects: [] }; +} + +/** A command-transition: same as `to` but bumps the epoch (I1). Any outcome + * event issued under the old epoch is dropped once this lands. */ +function command( + m: Machine, + phase: Phase, + effects: Effect[], + ctx?: Partial, +): Machine { + return { + phase, + ctx: { ...m.ctx, ...(ctx ?? {}), epoch: m.ctx.epoch + 1 }, + effects, + }; +} + +// --------------------------------------------------------------------------- +// The pure reducer. +// --------------------------------------------------------------------------- + +export function reduce(m: Machine, event: Event): Machine { + // I1: drop a stale outcome (an event issued under a superseded epoch). + if ("epoch" in event && event.epoch !== undefined && event.epoch !== m.ctx.epoch) { + return stay(m); + } + + switch (event.type) { + // ---- local turn ---------------------------------------------------- + case "SEND_LOCAL": + // A local send owns the view: leave any recovery, become the local + // streamer, disarm poll/reconnect. epoch++ so a late recovery outcome + // from the previous phase is dropped. + return command( + m, + { name: "sending" }, + [{ type: "cancelReconnect" }, { type: "disarmPoll" }], + { ownership: "local" }, + ); + + case "STREAM_START": { + // First frame arrived. Adopt the run-fact runId if present. sending -> + // streaming; a reconnect/attach that just went live also lands here. + const runFact = event.runId ? { runId: event.runId } : m.ctx.runFact; + return to(m, { name: "streaming" }, { + ctx: { runFact }, + effects: [{ type: "cancelReconnect" }, { type: "disarmPoll" }], + }); + } + + case "FINISH_CLEAN": + // A clean terminal outcome. The run is done — clear the run-fact and go + // idle. (The queue flush is a component concern gated by ownership; the + // FSM only models the phase.) + return to(m, { name: "idle" }, { + ctx: { runFact: null }, + effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }], + }); + + case "FINISH_ABORT": + // A user Stop / intentional abort finished. If we were stopping, the + // terminal data has now arrived (I4) — go idle. The run-fact is cleared. + return to(m, { name: "idle" }, { + ctx: { runFact: null }, + effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }], + }); + + case "FINISH_DISCONNECT": + // A LIVE SSE drop. If a run is (or may be) active, recover: + // - visible content already on screen -> keep it, poll to terminal + // (a full replay could clobber the fuller live tail), AND begin the + // live re-attach ladder; + // - no visible content -> the reconnect ladder rebuilds it. + // #488 commit 2: recovery is gated on the RUN-FACT, not on the presence + // of an assistant message — a break during the setup phase (before the + // first assistant frame) still has an active detached run to pick up. + if (m.ctx.runFact) { + const effects: Effect[] = [ + { type: "scheduleReconnect", attempt: 1, delayMs: reconnectDelayMs(1) }, + ]; + if (event.hasVisibleContent) effects.push({ type: "armPoll", reason: "disconnect-visible" }); + return command(m, { name: "reconnecting", attempt: 1, failed: false }, effects, { + ownership: "observer", + }); + } + // No run to recover: a plain disconnect. Surface the terminal notice. + return to(m, { name: "idle" }, { ctx: { runFact: null } }); + + case "FINISH_ERROR": + return to(m, { name: "error", kind: event.kind }, { + ctx: { runFact: null }, + effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }], + }); + + // ---- mount attach (resume) ---------------------------------------- + case "ATTACH_START": + // A reopened tab attaches to a still-running run: observer ownership. + return command(m, { name: "attaching" }, [{ type: "resumeStream" }], { + ownership: "observer", + runFact: event.runId ? { runId: event.runId } : m.ctx.runFact, + }); + + case "ATTACH_LIVE": + // The attach GET returned a live 2xx stream — follow it as an observer. + return to(m, { name: "streaming" }); + + case "ATTACH_NONE": + // 204 / non-2xx / throw: nothing live to attach. Arm the degraded poll to + // follow the run to terminal from the DB. This is a soft-negative run-fact + // (204 on a non-stripped path is authoritative-negative; the runtime may + // pass a RUN_FACT null separately). Keep the run-fact as-is here. + return to(m, { name: "polling", reason: "attach-none" }, { + effects: [{ type: "armPoll", reason: "attach-none" }], + }); + + // ---- 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. + if (m.phase.name !== "reconnecting") return stay(m); + return command( + m, + { name: "reconnecting", attempt: event.attempt, failed: false }, + [{ type: "resumeStream" }], + ); + + case "RECONNECT_ATTACHED": + // #488 commit 3: a live re-attach succeeded. Reset to streaming — the + // attempt counter is dropped, so a LATER disconnect can start a fresh + // ladder from attempt 1 (the old one-shot `!wasResumed` gate forbade a + // second cycle, sending the second break to silent poll). + return to(m, { name: "streaming" }, { + effects: [{ type: "cancelReconnect" }, { type: "disarmPoll" }], + }); + + case "RECONNECT_NONE": { + // 204 / error during a reconnect attempt. Arm the degraded poll as the + // belt-and-suspenders fallback, then either back off to the next attempt + // or, at the cap, surface the manual Retry ("failed"). + if (m.phase.name !== "reconnecting") return stay(m); + const attempt = m.phase.attempt; + if (attempt < RECONNECT_MAX_ATTEMPTS) { + return command( + m, + { name: "reconnecting", attempt: attempt + 1, failed: false }, + [ + { type: "armPoll", reason: "attach-none" }, + { type: "scheduleReconnect", attempt: attempt + 1, delayMs: reconnectDelayMs(attempt + 1) }, + ], + ); + } + return to(m, { name: "reconnecting", attempt, failed: true }, { + effects: [{ type: "armPoll", reason: "reconnect-exhausted" }], + }); + } + + case "RETRY": + // Manual Retry from the "failed" reconnect banner OR the stalled banner. + if (m.phase.name === "reconnecting" && m.phase.failed) { + return command( + m, + { name: "reconnecting", attempt: 1, failed: false }, + [{ type: "resumeStream" }], + ); + } + if (m.phase.name === "stalled") { + // Re-arm the poll to try to catch the run up again. + return command(m, { name: "polling", reason: "attach-none" }, [ + { type: "armPoll", reason: "attach-none" }, + ]); + } + 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 + // from `stopping`). + return to(m, { name: "idle" }, { + ctx: { runFact: null }, + effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }], + }); + + case "POLL_IDLE_CAP": + // #488 commit 4a: the poll hit the inactivity cap. Instead of going SILENT + // (the old "forever half-done answer"), surface a stalled banner + Retry. + if (m.phase.name !== "polling" && m.phase.name !== "reconnecting") return stay(m); + return to(m, { name: "stalled" }, { + effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }], + }); + + // ---- run-fact ------------------------------------------------------ + case "RUN_FACT": { + const runFact = event.runFact; + // A fresh NEGATIVE fact (no active run) cancels recovery immediately (I3): + // there is nothing to reconnect to / poll for. + if (!runFact) { + if ( + m.phase.name === "reconnecting" || + m.phase.name === "attaching" || + m.phase.name === "polling" || + m.phase.name === "stopping" + ) { + return to(m, { name: "idle" }, { + ctx: { runFact: null }, + effects: [{ type: "cancelReconnect" }, { type: "disarmPoll" }], + }); + } + return to(m, m.phase, { ctx: { runFact: null } }); + } + // A positive fact just updates the context (pessimism toward an attempt: a + // stale-but-positive fact permits entering recovery; a 204 will cut it). + return to(m, m.phase, { ctx: { runFact } }); + } + + // ---- stop ---------------------------------------------------------- + case "STOP_REQUESTED": + // Authoritative stop of a detached run. Enter `stopping` and fire stopRun + + // abort the local/attach reader. Exit is by DATA (I4), never by the HTTP + // response — so no transition keys off stopRun's return. + return command( + m, + { name: "stopping" }, + [{ type: "stopRun" }, { type: "abortAttach" }, { type: "cancelReconnect" }], + ); + + // ---- supersede (CAS) ---------------------------------------------- + case "SUPERSEDE_REQUESTED": + // "Interrupt and send now": CAS POST /stream { supersede }. epoch++ so a + // late outcome of the interrupted run is dropped. + return command( + m, + { name: "superseding" }, + [{ type: "supersede", targetRunId: event.targetRunId }, { type: "cancelReconnect" }, { type: "disarmPoll" }], + ); + + case "SUPERSEDE_READY": { + // CAS succeeded (old run stopped/settled, slot taken, new run begun). We + // are now the local streamer of the NEW run. Adopt its runId if provided. + const runFact = event.runId ? { runId: event.runId } : m.ctx.runFact; + return to(m, { name: "streaming" }, { ctx: { ownership: "local", runFact } }); + } + + case "SUPERSEDE_MISMATCH": + // The active run moved between the click and the CAS. Per the spec: verify + // via /run rather than blindly banner — the mismatch may be our own already- + // superseded run. Surface a classified error AND fire a run-fact verify. + return to(m, { name: "error", kind: "supersede-mismatch" }, { + ctx: { runFact: event.currentRunId ? { runId: event.currentRunId } : m.ctx.runFact }, + effects: [{ type: "postRun", reason: "verify" }], + }); + + case "SUPERSEDE_TIMEOUT": + // The old run did not settle within W. Nothing persisted; the composer keeps + // its text. Classified error, NO auto-retry (the old client retry ladder is + // removed in #488 commit 5). + return to(m, { name: "error", kind: "supersede-timeout" }); + + case "SUPERSEDE_INVALID": + return to(m, { name: "error", kind: "supersede-invalid" }); + + case "RUN_ALREADY_ACTIVE": + // A plain POST hit the one-active-run gate. NO auto-retry — the composer + // 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 + // NO late callback can drive this (now dead) machine (I5). + return command(m, { name: "idle" }, [ + { type: "abortAttach" }, + { type: "cancelReconnect" }, + { type: "disarmPoll" }, + ]); + + default: { + // Exhaustiveness guard. + const _never: never = event; + void _never; + return stay(m); + } + } +}