Implement a streaming state reducer

Implement reduceStream to the contract and predict the transition fixtures. Then find the defects in the fetch integration and describe a corrected flow with per-run AbortControllers. Explain how your design keeps stale-run events, duplicate deltas, and a stream that ends without a terminal event from corrupting the UI, and how the UI announces and labels each terminal state.

Each server event carries a run ID and a one-based sequence number. A delta event appends text, done marks a complete answer, and error ends the run with a safe error message. The sequence covers every persisted event for that run, so a client can tell a replay from a gap.

The client starts a run through a local action. It accepts only events that belong to the active run and requires contiguous sequence numbers. An event that was already applied is ignored, and a gap marks the response interrupted rather than silently losing text.

The duplicate rule assumes the payload for a sequence number never changes; two different payloads under one number are a protocol violation for server-side integrity checks to catch.

Local Stop means the client stopped this attempt. It does not assert that the server cancelled every operation. A separate reconciliation step can later update the persisted run record from authoritative server state.

The reducer receives the current state and one action and returns the next state. It is pure: no clock, no network, and no mutation of its inputs. It never throws: a bad sequence number becomes a failed state, not an exception. When an action changes nothing it returns the same state object, so useReducer can skip the re-render; any change returns a new object with exactly the five fields runId, status, text, lastSeq, and error.

type Status = "idle" | "streaming" | "complete" | "interrupted" | "stopped" | "failed";
type State = {
  runId: string | null;
  status: Status;
  text: string;
  lastSeq: number;
  error: string | null;
};
// seq is untrusted network data: the reducer validates it before use (rule 4).
type ServerEvent = { runId: string; seq: unknown } & (
  | { type: "delta"; text: string }
  | { type: "done" }
  | { type: "error"; message: string }
);
type Action =
  | { type: "start"; runId: string }
  | { type: "stop"; runId: string }
  | { type: "disconnect"; runId: string }
  | { type: "event"; event: ServerEvent };

const initialState: State = {
  runId: null, status: "idle", text: "", lastSeq: 0, error: null
};

declare function reduceStream(state: State, action: Action): State;

Apply these rules in order; the first one that matches decides the result.

1. start returns a fresh streaming state for its run, { runId, status: "streaming", text: "", lastSeq: 0, error: null }, whatever the current state.
2. Any other action whose run ID differs from state.runId, or any other action while status is not "streaming", returns the same state object. Complete, interrupted, stopped, and failed are therefore sticky until the next start.
3. stop sets status "stopped", and disconnect sets "interrupted". Text, lastSeq, and error are kept.
4. An event whose seq is not a safe integer of at least 1 sets status "failed" with error "Invalid event sequence".
5. An event with seq <= lastSeq was already applied: return the same state object, not an equal copy.
6. An event with seq !== lastSeq + 1 sets status "interrupted" with error "Missing event", keeping text and lastSeq.
7. Otherwise apply the event and set lastSeq to its seq: delta appends its text, done sets "complete", and error sets "failed" with the event's message as the error.

A gap ends live accumulation on purpose. Reconnecting needs a separate snapshot or resume operation (the Hard question in this topic), not more events sent into an interrupted state.

The Python version is reduce_stream(state, action) on dictionaries, with run_id and last_seq in place of runId and lastSeq. Everything else, including the status and error strings, is identical. In Python, seq must be an int that is not a bool, from 1 to 2**53 - 1.