fix(app): drive request slots and draft image retention through state machines (#61970)

packages/app/src/runtime/persistence/drafts-machine.test.ts

@@ -0,0 +1,214 @@
+import { describe, expect, test } from "bun:test"
+import {
+  createDraftRetentionState,
+  isCurrentSave,
+  selectBlobStoreId,
+  selectDocumentRevision,
+  selectLatestMutation,
+  selectRetainedBlob,
+  transitionDraftRetention,
+  type DraftRetentionEffect,
+  type DraftRetentionEvent,
+  type DraftRetentionState,
+} from "./drafts-machine"
+
+const graceMs = 30
+
+const blob = new Blob(["image"])
+
+const doc = { storeId: 1, key: "draft" }
+
+function run(events: readonly DraftRetentionEvent[], from = createDraftRetentionState()) {
+  return events.reduce<{ state: DraftRetentionState; effects: DraftRetentionEffect[] }>(
+    (acc, event) => {
+      const next = transitionDraftRetention(acc.state, event)
+
+      return { state: next.state, effects: [...acc.effects, ...next.effects] }
+    },
+    { state: from, effects: [] },
+  )
+}
+
+const materialize = (id: string, grace: number | undefined = graceMs): DraftRetentionEvent => ({
+  type: "blob.materialized",
+  id,
+  blob,
+  url: `blob:${id}`,
+  graceMs: grace,
+})
+
+const status = (state: DraftRetentionState, id: string) => state.blobs.get(id)?.status
+
+describe("transitionDraftRetention", () => {
+  test("a save that outlasts the grace keeps the pasted image it is writing", () => {
+    const pasted = run([{ type: "store.registered" }, materialize("img")])
+    expect(pasted.effects).toEqual([{ type: "timer.arm", id: "img", attempt: 1, delayMs: graceMs }])
+
+    const saving = run([{ type: "save.started", ...doc, ids: new Set(["img"]), graceMs }], pasted.state)
+    expect(saving.effects).toEqual([{ type: "timer.cancel", id: "img" }])
+    expect(status(saving.state, "img")).toBe("pinned")
+
+    // The timer callback already queued before the cancel lands late and must not revoke.
+    const late = transitionDraftRetention(saving.state, { type: "grace.expired", id: "img", attempt: 1 })
+    expect(late.state).toBe(saving.state)
+    expect(late.effects).toEqual([])
+
+    const attempt = selectLatestMutation(saving.state, 1, "draft")
+    const saved = run([{ type: "save.committed", ...doc, attempt, ids: new Set(["img"]), graceMs }], saving.state)
+    expect(saved.effects).toEqual([])
+    expect(status(saved.state, "img")).toBe("pinned")
+  })
+
+  test("a load that read the document before a save replaced it does not unpin the new image", () => {
+    const start = run([{ type: "store.registered" }, materialize("new")]).state
+    const revision = selectDocumentRevision(start, 1, "draft")
+
+    const saved = run(
+      [
+        { type: "save.started", ...doc, ids: new Set(["new"]), graceMs },
+        { type: "save.committed", ...doc, attempt: 2, ids: new Set(["new"]), graceMs },
+      ],
+      start,
+    )
+
+    const late = transitionDraftRetention(saved.state, {
+      type: "load.succeeded",
+      ...doc,
+      revision,
+      ids: new Set(["old"]),
+      graceMs,
+    })
+
+    expect(late.state).toBe(saved.state)
+    expect(status(late.state, "new")).toBe("pinned")
+  })
+
+  test("a load that starts while a save is writing is fenced by the save's commit", () => {
+    const saving = run([
+      { type: "store.registered" },
+      materialize("new"),
+      { type: "save.started", ...doc, ids: new Set(["new"]), graceMs },
+    ]).state
+
+    const revision = selectDocumentRevision(saving, 1, "draft")
+    const attempt = selectLatestMutation(saving, 1, "draft")
+    const saved = run([{ type: "save.committed", ...doc, attempt, ids: new Set(["new"]), graceMs }], saving).state
+
+    expect(
+      transitionDraftRetention(saved, { type: "load.succeeded", ...doc, revision, ids: new Set(), graceMs }).state,
+    ).toBe(saved)
+  })
+
+  test("a remove that a later save overtook does not unpin the save's image", () => {
+    const removing = run([{ type: "store.registered" }, materialize("img"), { type: "remove.started", ...doc }]).state
+    const removal = selectLatestMutation(removing, 1, "draft")
+    const saved = run([{ type: "save.started", ...doc, ids: new Set(["img"]), graceMs }], removing)
+    const attempt = selectLatestMutation(saved.state, 1, "draft")
+    const committed = run([{ type: "save.committed", ...doc, attempt, ids: new Set(["img"]), graceMs }], saved.state)
+
+    const late = transitionDraftRetention(committed.state, {
+      type: "remove.committed",
+      ...doc,
+      attempt: removal,
+      graceMs,
+    })
+
+    expect(late.state).toBe(committed.state)
+    expect(status(late.state, "img")).toBe("pinned")
+  })
+
+  test("a superseded save neither writes nor replaces the committed document", () => {
+    const first = run([
+      { type: "store.registered" },
+      materialize("a"),
+      materialize("b"),
+      { type: "save.started", ...doc, ids: new Set(["a"]), graceMs },
+    ]).state
+
+    const stale = selectLatestMutation(first, 1, "draft")
+    const second = run([{ type: "save.started", ...doc, ids: new Set(["b"]), graceMs }], first).state
+    expect(isCurrentSave(second, 1, "draft", stale)).toBe(false)
+
+    const current = selectLatestMutation(second, 1, "draft")
+
+    const done = run(
+      [
+        { type: "save.committed", ...doc, attempt: current, ids: new Set(["b"]), graceMs },
+        { type: "save.committed", ...doc, attempt: stale, ids: new Set(["a"]), graceMs },
+      ],
+      second,
+    )
+
+    expect(status(done.state, "a")).toBe("grace")
+    expect(status(done.state, "b")).toBe("pinned")
+  })
+
+  test("a failed save releases images only the save held", () => {
+    const first = run([
+      { type: "store.registered" },
+      materialize("kept"),
+      materialize("new"),
+      { type: "save.started", ...doc, ids: new Set(["kept"]), graceMs },
+    ]).state
+
+    const committed = run(
+      [
+        {
+          type: "save.committed",
+          ...doc,
+          attempt: selectLatestMutation(first, 1, "draft"),
+          ids: new Set(["kept"]),
+          graceMs,
+        },
+        { type: "save.started", ...doc, ids: new Set(["kept", "new"]), graceMs },
+      ],
+      first,
+    ).state
+
+    const attempt = selectLatestMutation(committed, 1, "draft")
+    const failed = transitionDraftRetention(committed, { type: "save.aborted", ...doc, attempt, graceMs })
+    expect(failed.effects).toEqual([{ type: "timer.arm", id: "new", attempt: expect.any(Number), delayMs: graceMs }])
+    expect(status(failed.state, "kept")).toBe("pinned")
+    expect(status(failed.state, "new")).toBe("grace")
+  })
+
+  test("pasting the same image again during its grace restarts the release, and a page reference keeps it", () => {
+    const pasted = run([materialize("img")]).state
+    const again = transitionDraftRetention(pasted, { type: "blob.touched", id: "img", graceMs })
+    expect(again.effects).toEqual([{ type: "timer.arm", id: "img", attempt: 2, delayMs: graceMs }])
+    expect(transitionDraftRetention(again.state, { type: "grace.expired", id: "img", attempt: 1 }).state).toBe(
+      again.state,
+    )
+
+    const permanent = transitionDraftRetention(again.state, { type: "blob.touched", id: "img", graceMs: undefined })
+    expect(permanent.effects).toEqual([{ type: "timer.cancel", id: "img" }])
+    expect(status(permanent.state, "img")).toBe("permanent")
+    expect(transitionDraftRetention(permanent.state, { type: "grace.expired", id: "img", attempt: 2 }).state).toBe(
+      permanent.state,
+    )
+  })
+
+  test("a released image revokes its URL and drops the aliases that pointed at it", () => {
+    const aliased = run([materialize("stored"), { type: "alias.recorded", from: "live", to: "stored" }]).state
+    expect(selectRetainedBlob(aliased, "live")?.url).toBe("blob:stored")
+
+    const expired = transitionDraftRetention(aliased, { type: "grace.expired", id: "stored", attempt: 1 })
+    expect(expired.effects).toEqual([{ type: "url.revoke", url: "blob:stored" }])
+    expect(expired.state.aliases.size).toBe(0)
+    expect(selectRetainedBlob(expired.state, "live")).toBeUndefined()
+  })
+
+  test("an image resolves through the store whose document references it", () => {
+    const state = run([
+      { type: "store.registered" },
+      { type: "load.succeeded", ...doc, revision: 0, ids: new Set(["first"]), graceMs },
+      { type: "store.registered" },
+      { type: "load.succeeded", storeId: 2, key: "draft", revision: 0, ids: new Set(), graceMs },
+    ]).state
+
+    expect(selectBlobStoreId(state, "first")).toBe(1)
+    expect(selectBlobStoreId(state, "unknown")).toBe(2)
+    // Each store keeps its own copy of a key.
+    expect(state.documents.size).toBe(2)
+  })
+})

packages/app/src/runtime/persistence/drafts-machine.ts

@@ -0,0 +1,385 @@
+// Decoded image bytes the renderer pins through object URLs. Every consumer of a `blob.url` is a
+// persisted draft document (composer prompt, prompt history), so an image is pinned exactly while a
+// stored document, or a save writing one, references it. Once the last reference disappears (removed
+// from a draft, sent, or a discarded duplicate paste) the URL is revoked after this grace, which
+// covers the persist delay between a paste and the save that references it, and the submit → history
+// handoff.
+export const RETAINED_BLOB_GRACE_MS = 30_000
+
+/**
+ * One image's `Blob` and object URL. Absent from `blobs` means not loaded yet, or revoked.
+ * - `permanent`: a page-lifetime reference (`createBlobReference`, `pinMediaBlob`); never revoked.
+ * - `pinned`: a committed document or a running save references it.
+ * - `grace`: nothing references it; revoked when the timer for `attempt` fires.
+ */
+export type DraftBlobSlot =
+  | { readonly status: "permanent"; readonly blob: Blob; readonly url: string }
+  | { readonly status: "pinned"; readonly blob: Blob; readonly url: string }
+  | { readonly status: "grace"; readonly blob: Blob; readonly url: string; readonly attempt: number }
+
+/** One stored document of one draft store. Never pruned: `revision` must outlive the document to fence late results. */
+export type DraftDocumentSlot = {
+  readonly storeId: number
+  /** The newest save or remove attempt. A save or remove that started before it is stale. */
+  readonly latestMutation: number
+  /**
+   * Moves when a save or remove starts or commits. A load that saw another revision may have read a document that is
+   * no longer stored.
+   */
+  readonly revision: number
+  /** Image ids of the document as last loaded or saved. */
+  readonly committed: ReadonlySet<string>
+  /** Image ids of each running save, keyed by its attempt, pinned until it commits or fails. */
+  readonly saves: ReadonlyMap<number, ReadonlySet<string>>
+}
+
+export type DraftRetentionState = {
+  readonly nextAttempt: number
+  readonly nextStoreId: number
+  /** Resolves image ids no document references yet. */
+  readonly latestStoreId: number | undefined
+  readonly blobs: ReadonlyMap<string, DraftBlobSlot>
+  readonly documents: ReadonlyMap<string, DraftDocumentSlot>
+  /**
+   * Image ids that were stored under a different id (a store without WebCrypto assigns fresh ones, and a `media:` ref
+   * is copied into the store); live references still carry the original.
+   */
+  readonly aliases: ReadonlyMap<string, string>
+}
+
+type DocumentRef = { readonly storeId: number; readonly key: string }
+
+export type DraftRetentionEvent =
+  | { readonly type: "store.registered" }
+  | {
+      readonly type: "blob.materialized"
+      readonly id: string
+      readonly blob: Blob
+      readonly url: string
+      /** Undefined for a page-lifetime reference. */
+      readonly graceMs: number | undefined
+    }
+  | { readonly type: "blob.touched"; readonly id: string; readonly graceMs: number | undefined }
+  | { readonly type: "alias.recorded"; readonly from: string; readonly to: string }
+  | (DocumentRef & {
+      readonly type: "load.succeeded"
+      /** The document's `revision` when the read started. */
+      readonly revision: number
+      readonly ids: ReadonlySet<string>
+      readonly graceMs: number
+    })
+  | (DocumentRef & { readonly type: "save.started"; readonly ids: ReadonlySet<string>; readonly graceMs: number })
+  | (DocumentRef & {
+      readonly type: "save.referenced"
+      readonly attempt: number
+      readonly ids: ReadonlySet<string>
+      readonly graceMs: number
+    })
+  | (DocumentRef & {
+      readonly type: "save.committed"
+      readonly attempt: number
+      readonly ids: ReadonlySet<string>
+      readonly graceMs: number
+    })
+  | (DocumentRef & { readonly type: "save.aborted"; readonly attempt: number; readonly graceMs: number })
+  | (DocumentRef & { readonly type: "remove.started" })
+  | (DocumentRef & { readonly type: "remove.committed"; readonly attempt: number; readonly graceMs: number })
+  | { readonly type: "grace.expired"; readonly id: string; readonly attempt: number }
+
+export type DraftRetentionEffect =
+  | { readonly type: "timer.arm"; readonly id: string; readonly attempt: number; readonly delayMs: number }
+  | { readonly type: "timer.cancel"; readonly id: string }
+  | { readonly type: "url.revoke"; readonly url: string }
+
+export type DraftRetentionTransition = {
+  readonly state: DraftRetentionState
+  readonly effects: readonly DraftRetentionEffect[]
+}
+
+const none: readonly DraftRetentionEffect[] = []
+
+const empty: ReadonlySet<string> = new Set()
+
+const noSaves: ReadonlyMap<number, ReadonlySet<string>> = new Map()
+
+export function createDraftRetentionState(): DraftRetentionState {
+  return {
+    nextAttempt: 1,
+    nextStoreId: 1,
+    latestStoreId: undefined,
+    blobs: new Map(),
+    documents: new Map(),
+    aliases: new Map(),
+  }
+}
+
+export function transitionDraftRetention(
+  state: DraftRetentionState,
+  event: DraftRetentionEvent,
+): DraftRetentionTransition {
+  if (event.type === "store.registered")
+    return { state: { ...state, nextStoreId: state.nextStoreId + 1, latestStoreId: state.nextStoreId }, effects: none }
+
+  if (event.type === "blob.materialized") return materialized(state, event)
+
+  if (event.type === "blob.touched") return touched(state, event.id, event.graceMs)
+
+  if (event.type === "alias.recorded") return aliased(state, event.from, event.to)
+
+  if (event.type === "grace.expired") return expired(state, event.id, event.attempt)
+
+  const name = documentName(event)
+
+  const doc = state.documents.get(name) ?? {
+    storeId: event.storeId,
+    latestMutation: 0,
+    revision: 0,
+    committed: empty,
+    saves: noSaves,
+  }
+
+  // Claims the images synchronously, before the save's first await, so a slow encode or write cannot outlast the grace.
+  if (event.type === "save.started") {
+    const attempt = state.nextAttempt
+    const saves = new Map(doc.saves).set(attempt, resolveIds(state, event.ids))
+
+    return reconcile(
+      { ...state, nextAttempt: attempt + 1 },
+      name,
+      { ...doc, latestMutation: attempt, revision: attempt, saves },
+      event.graceMs,
+    )
+  }
+
+  if (event.type === "remove.started") {
+    const attempt = state.nextAttempt
+    const documents = new Map(state.documents).set(name, { ...doc, latestMutation: attempt, revision: attempt })
+
+    return { state: { ...state, nextAttempt: attempt + 1, documents }, effects: none }
+  }
+
+  if (event.type === "load.succeeded") {
+    if (doc.revision !== event.revision) return { state, effects: none }
+
+    return reconcile(state, name, { ...doc, committed: resolveIds(state, event.ids) }, event.graceMs)
+  }
+
+  if (event.type === "remove.committed") {
+    if (doc.latestMutation !== event.attempt) return { state, effects: none }
+
+    return reconcile(
+      { ...state, nextAttempt: state.nextAttempt + 1 },
+      name,
+      { ...doc, revision: state.nextAttempt, committed: empty },
+      event.graceMs,
+    )
+  }
+
+  const current = doc.saves.get(event.attempt)
+
+  if (!current) return { state, effects: none }
+
+  if (event.type === "save.referenced") {
+    if (doc.latestMutation !== event.attempt) return { state, effects: none }
+
+    const ids = new Set([...current, ...resolveIds(state, event.ids)])
+
+    return reconcile(state, name, { ...doc, saves: new Map(doc.saves).set(event.attempt, ids) }, event.graceMs)
+  }
+
+  const saves = new Map(doc.saves)
+  saves.delete(event.attempt)
+
+  // A save that failed, or that a later save or remove superseded, leaves the committed document as it was.
+  if (event.type === "save.aborted" || doc.latestMutation !== event.attempt)
+    return reconcile(state, name, { ...doc, saves }, event.graceMs)
+
+  return reconcile(
+    { ...state, nextAttempt: state.nextAttempt + 1 },
+    name,
+    { ...doc, revision: state.nextAttempt, saves, committed: resolveIds(state, event.ids) },
+    event.graceMs,
+  )
+}
+
+/** The live bytes and URL behind an image id, following its alias. */
+export function selectRetainedBlob(state: DraftRetentionState, id: string) {
+  const slot = state.blobs.get(resolveBlobAlias(state, id))
+
+  return slot && { blob: slot.blob, url: slot.url }
+}
+
+export function resolveBlobAlias(state: DraftRetentionState, id: string) {
+  return state.aliases.get(id) ?? id
+}
+
+/** Fences a load: read it when the load starts and pass it back with `load.succeeded`. */
+export function selectDocumentRevision(state: DraftRetentionState, storeId: number, key: string) {
+  return state.documents.get(documentName({ storeId, key }))?.revision ?? 0
+}
+
+/** The attempt of the newest save or remove, read right after `save.started` or `remove.started`. */
+export function selectLatestMutation(state: DraftRetentionState, storeId: number, key: string) {
+  return state.documents.get(documentName({ storeId, key }))?.latestMutation ?? 0
+}
+
+/** False once a later save or remove superseded the save, so it must not write. */
+export function isCurrentSave(state: DraftRetentionState, storeId: number, key: string, attempt: number) {
+  const doc = state.documents.get(documentName({ storeId, key }))
+
+  return doc?.latestMutation === attempt && doc.saves.has(attempt)
+}
+
+/** The store whose documents reference the image, else the newest store. */
+export function selectBlobStoreId(state: DraftRetentionState, id: string) {
+  const target = resolveBlobAlias(state, id)
+  const owner = [...state.documents.values()].findLast((doc) => references(doc, id) || references(doc, target))
+
+  return owner?.storeId ?? state.latestStoreId
+}
+
+function materialized(
+  state: DraftRetentionState,
+  event: Extract<DraftRetentionEvent, { type: "blob.materialized" }>,
+): DraftRetentionTransition {
+  const existing = state.blobs.get(event.id)
+
+  // Another URL already serves these bytes; keep it and drop the duplicate.
+  if (existing) {
+    const touch = touched(state, event.id, event.graceMs)
+
+    return { state: touch.state, effects: [{ type: "url.revoke", url: event.url }, ...touch.effects] }
+  }
+
+  const blobs = new Map(state.blobs)
+
+  if (event.graceMs === undefined) {
+    blobs.set(event.id, { status: "permanent", blob: event.blob, url: event.url })
+
+    return { state: { ...state, blobs }, effects: none }
+  }
+
+  if (referenced(state.documents, event.id)) {
+    blobs.set(event.id, { status: "pinned", blob: event.blob, url: event.url })
+
+    return { state: { ...state, blobs }, effects: none }
+  }
+
+  const attempt = state.nextAttempt
+  blobs.set(event.id, { status: "grace", blob: event.blob, url: event.url, attempt })
+
+  return {
+    state: { ...state, nextAttempt: attempt + 1, blobs },
+    effects: [{ type: "timer.arm", id: event.id, attempt, delayMs: event.graceMs }],
+  }
+}
+
+// The same bytes were uploaded or referenced again: a pending release restarts, and a page-lifetime reference keeps them.
+function touched(state: DraftRetentionState, id: string, graceMs: number | undefined): DraftRetentionTransition {
+  const slot = state.blobs.get(id)
+
+  if (!slot || slot.status === "permanent") return { state, effects: none }
+
+  if (graceMs === undefined)
+    return {
+      state: { ...state, blobs: new Map(state.blobs).set(id, { status: "permanent", blob: slot.blob, url: slot.url }) },
+      effects: slot.status === "grace" ? [{ type: "timer.cancel", id }] : none,
+    }
+
+  if (slot.status === "pinned") return { state, effects: none }
+
+  const attempt = state.nextAttempt
+
+  return {
+    state: { ...state, nextAttempt: attempt + 1, blobs: new Map(state.blobs).set(id, { ...slot, attempt }) },
+    effects: [{ type: "timer.arm", id, attempt, delayMs: graceMs }],
+  }
+}
+
+// Later encodes of the still-live reference resolve straight to the new id. Re-point any earlier
+// alias chain so lookups stay one step.
+function aliased(state: DraftRetentionState, from: string, to: string): DraftRetentionTransition {
+  if (from === to || state.aliases.get(from) === to) return { state, effects: none }
+
+  const aliases = new Map([...state.aliases].map(([key, value]) => [key, value === from ? to : value]))
+  aliases.set(from, to)
+
+  return { state: { ...state, aliases }, effects: none }
+}
+
+function expired(state: DraftRetentionState, id: string, attempt: number): DraftRetentionTransition {
+  const slot = state.blobs.get(id)
+
+  if (slot?.status !== "grace" || slot.attempt !== attempt) return { state, effects: none }
+
+  const blobs = new Map(state.blobs)
+  blobs.delete(id)
+
+  return {
+    state: { ...state, blobs, aliases: new Map([...state.aliases].filter(([, value]) => value !== id)) },
+    effects: [{ type: "url.revoke", url: slot.url }],
+  }
+}
+
+/** Writes the document, then pins the images it gained and starts the release of those nothing references anymore. */
+function reconcile(
+  state: DraftRetentionState,
+  name: string,
+  doc: DraftDocumentSlot,
+  graceMs: number,
+): DraftRetentionTransition {
+  const previous = state.documents.get(name)
+  const documents = new Map(state.documents).set(name, doc)
+  const touchedIds = new Set([...ids(previous), ...ids(doc)])
+  const blobs = new Map(state.blobs)
+  const effects: DraftRetentionEffect[] = []
+  let attempt = state.nextAttempt
+
+  touchedIds.forEach((id) => {
+    const slot = blobs.get(id)
+
+    if (!slot || slot.status === "permanent") return
+
+    const live = referenced(documents, id)
+
+    if (live && slot.status === "grace") {
+      blobs.set(id, { status: "pinned", blob: slot.blob, url: slot.url })
+      effects.push({ type: "timer.cancel", id })
+    }
+
+    if (!live && slot.status === "pinned") {
+      blobs.set(id, { status: "grace", blob: slot.blob, url: slot.url, attempt })
+      effects.push({ type: "timer.arm", id, attempt, delayMs: graceMs })
+      attempt++
+    }
+  })
+
+  return {
+    state: { ...state, nextAttempt: attempt, documents, blobs: effects.length ? blobs : state.blobs },
+    effects,
+  }
+}
+
+function resolveIds(state: DraftRetentionState, ids: ReadonlySet<string>): ReadonlySet<string> {
+  if (!state.aliases.size) return ids
+
+  return new Set([...ids].map((id) => resolveBlobAlias(state, id)))
+}
+
+function ids(doc: DraftDocumentSlot | undefined) {
+  if (!doc) return []
+
+  return [...doc.committed, ...[...doc.saves.values()].flatMap((save) => [...save])]
+}
+
+function references(doc: DraftDocumentSlot, id: string) {
+  return doc.committed.has(id) || [...doc.saves.values()].some((save) => save.has(id))
+}
+
+function referenced(documents: ReadonlyMap<string, DraftDocumentSlot>, id: string) {
+  return [...documents.values()].some((doc) => references(doc, id))
+}
+
+function documentName(ref: DocumentRef) {
+  return `${ref.storeId}\0${ref.key}`
+}

packages/app/src/runtime/persistence/drafts.test.ts

@@ -345,6 +345,57 @@ describe("draft store image retention", () => {
     fresh()
     expect(await resolveBlobUrl({ id: "gone" })).toBeUndefined()
   })
+
+  test("a save that takes longer than the grace keeps the image it writes", async () => {
+    const { memory, store } = fresh(10)
+    const pasted = await store.putBlob(image(7))
+    const gate = Promise.withResolvers<void>()
+    const set = memory.driver.set
+    memory.driver.set = async (key, value, strict) => {
+      await gate.promise
+
+      return set(key, value, strict)
+    }
+
+    const saving = store.setDocument("slow", { prompt: [{ type: "image", blob: pasted }] })
+    await new Promise((resolve) => setTimeout(resolve, 40))
+    gate.resolve()
+    await saving
+    await tick()
+    expect(await released(store, 7, pasted.url)).toBe(false)
+  })
+
+  test("a load that resolves after a save replaced the image does not release the new one", async () => {
+    const { memory, store } = fresh()
+    const old = await memory.driver.putBlob(image(8))
+    memory.documents.set("raced", JSON.stringify({ prompt: [{ type: "image", blob: { id: old } }] }))
+    const gate = Promise.withResolvers<void>()
+    const get = memory.driver.get
+    memory.driver.get = async (key) => {
+      const value = await get(key)
+      await gate.promise
+
+      return value
+    }
+
+    const loading = store.getItem("raced")
+    const pasted = await store.putBlob(image(9))
+    await store.setDocument("raced", { prompt: [{ type: "image", blob: pasted }] })
+    gate.resolve()
+    await loading
+    await tick()
+    expect(await released(store, 9, pasted.url)).toBe(false)
+  })
+
+  test("an image a store's document references resolves through that store after another store is created", async () => {
+    const first = fresh()
+    const id = await first.memory.driver.putBlob(image(10))
+    first.memory.documents.set("loaded", JSON.stringify({ prompt: [{ type: "image", blob: { id } }] }))
+    await first.store.getItem("loaded")
+    fresh()
+    expect((await resolveBlobUrl({ id }))?.startsWith("blob:")).toBe(true)
+    expect(await resolveBlobUrl({ id: "gone" })).toBeUndefined()
+  })
 })
 
 describe("draft store media refs", () => {

packages/app/src/runtime/persistence/drafts.ts

@@ -1,6 +1,18 @@
 import type { AsyncStorage } from "@solid-primitives/storage"
 import { Option, Schema } from "effect"
 import { mediaHashFromUri } from "@/runtime/server/media"
+import {
+  createDraftRetentionState,
+  isCurrentSave,
+  RETAINED_BLOB_GRACE_MS,
+  resolveBlobAlias,
+  selectBlobStoreId,
+  selectDocumentRevision,
+  selectLatestMutation,
+  selectRetainedBlob,
+  transitionDraftRetention,
+  type DraftRetentionEvent,
+} from "./drafts-machine"
 
 /* oxlint-disable anti-slop/no-unknown-parameters, anti-slop/no-unknown-returns, anti-slop/no-runtime-typeof, anti-slop/no-unsafe-dictionary-type, anti-slop/require-safety-comment-for-type-assertion, anti-slop/no-known-value-widening -- the store walks opaque persisted JSON in every shape earlier releases wrote, and IndexedDB hands back untyped values */
 
@@ -33,33 +45,42 @@ export const draftTextChunk = 64 * 1024
 
 const textCacheLimit = 64
 
-// Decoded image bytes the renderer pins through object URLs. Every consumer of a `blob.url` is a
-// persisted draft document (composer prompt, prompt history), so an image is pinned exactly while a
-// stored document references it. Once the last reference disappears (removed from a draft, sent, or
-// a discarded duplicate paste) the URL is revoked after this grace, which covers the persist delay
-// between a paste and the save that references it, and the submit → history handoff.
-export const retainedBlobGrace = 30_000
+export const retainedBlobGrace = RETAINED_BLOB_GRACE_MS
 
-type Retained = { blob: Blob; url: string; release: ReturnType<typeof setTimeout> | undefined }
+let retention = createDraftRetentionState()
 
-const retained = new Map<string, Retained>()
+const timers = new Map<string, ReturnType<typeof setTimeout>>()
 
-// Document keys that reference each image id; an id with no keys is released after the grace.
-const refs = new Map<string, Set<string>>()
+// Fetches image bytes from the store whose documents reference them. Documents load without their
+// bytes; a consumer that renders or sends an image resolves its URL through here, so a history full
+// of large attachments costs nothing at startup.
+const storeLoaders = new Map<number, (id: string) => Promise<string | undefined>>()
 
-// Image ids that were restored under a different id (a store without WebCrypto assigns fresh
-// ones); live references still carry the original.
-const aliases = new Map<string, string>()
-
-// Fetches image bytes from the store created last. Documents load without their bytes; a consumer
-// that renders or sends an image resolves its URL through here, so a history full of large
-// attachments costs nothing at startup.
 type MediaBlobLoader = (uri: string) => Promise<string | undefined>
 
-let loader: ((id: string) => Promise<string | undefined>) | undefined
-
 const mediaLoaders = new Map<string, MediaBlobLoader>()
 
+function dispatch(event: DraftRetentionEvent) {
+  const result = transitionDraftRetention(retention, event)
+  retention = result.state
+  result.effects.forEach((effect) => {
+    if (effect.type === "url.revoke") return URL.revokeObjectURL(effect.url)
+
+    clearTimeout(timers.get(effect.id))
+    timers.delete(effect.id)
+
+    if (effect.type === "timer.cancel") return
+
+    timers.set(
+      effect.id,
+      setTimeout(() => {
+        timers.delete(effect.id)
+        dispatch({ type: "grace.expired", id: effect.id, attempt: effect.attempt })
+      }, effect.delayMs),
+    )
+  })
+}
+
 export function setMediaBlobLoader(key: string, next: MediaBlobLoader | undefined) {
   mediaLoaders.delete(key)
 
@@ -103,7 +124,7 @@ async function loadMediaBlob(uri: string, serverKey?: string) {
 /** The object URL for an image reference, loading its bytes from the draft store on first use. */
 export function resolveBlobUrl(blob: { id: string; url?: string; server?: string }) {
   if (blob.url && !mediaHashFromUri(blob.url)) return Promise.resolve(blob.url)
-  const existing = retained.get(aliases.get(blob.id) ?? blob.id)
+  const existing = selectRetainedBlob(retention, blob.id)
 
   if (existing) return Promise.resolve(existing.url)
 
@@ -111,54 +132,25 @@ export function resolveBlobUrl(blob: { id: string; url?: string; server?: string
 
   if (blob.url) return Promise.resolve(blob.url)
 
-  return loader?.(blob.id) ?? Promise.resolve(undefined)
+  return storeLoaders.get(selectBlobStoreId(retention, blob.id) ?? 0)?.(blob.id) ?? Promise.resolve(undefined)
 }
 
-function blobUrl(id: string, blob: Blob, grace?: number) {
-  const existing = retained.get(id)
+// Without a grace the image has no store to reference it from and stays for the page's lifetime.
+function blobUrl(id: string, blob: Blob, graceMs?: number) {
+  const existing = retention.blobs.get(id)
+
+  if (existing) {
+    dispatch({ type: "blob.touched", id, graceMs })
+
+    return existing.url
+  }
 
-  if (existing) return existing.url
   const url = URL.createObjectURL(blob)
-  // Without a grace the image has no store to reference it from and stays for the page's lifetime.
-  const release = grace === undefined || refs.get(id)?.size ? undefined : setTimeout(() => revoke(id), grace)
-  retained.set(id, { blob, url, release })
+  dispatch({ type: "blob.materialized", id, blob, url, graceMs })
 
   return url
 }
 
-// Record which image ids `key` now references; ids it dropped are released once no other document
-// references them, ids it gained stay pinned.
-function retain(key: string, ids: ReadonlySet<string>, grace: number) {
-  for (const [id, keys] of refs) {
-    if (ids.has(id) || !keys.delete(key) || keys.size) continue
-    refs.delete(id)
-    const entry = retained.get(id)
-
-    if (entry) entry.release = setTimeout(() => revoke(id), grace)
-  }
-
-  for (const id of ids) {
-    const keys = refs.get(id) ?? new Set<string>()
-    keys.add(key)
-    refs.set(id, keys)
-    const entry = retained.get(id)
-
-    if (!entry) continue
-    clearTimeout(entry.release)
-    entry.release = undefined
-  }
-}
-
-function revoke(id: string) {
-  const entry = retained.get(id)
-
-  if (!entry) return
-  URL.revokeObjectURL(entry.url)
-  retained.delete(id)
-
-  for (const [from, to] of aliases) if (to === id) aliases.delete(from)
-}
-
 // Image ids a document references: `{ blob: { id } }` parts, not text chunk lists.
 function imageIDs(value: unknown, into = new Set<string>()): Set<string> {
   if (Array.isArray(value)) {
@@ -204,11 +196,12 @@ export async function createBlobReference(blob: Blob): Promise<BlobReference> {
 
 export function createDraftStore(driver: Driver, options: { grace?: number } = {}): DraftStore {
   const grace = options.grace ?? retainedBlobGrace
-  const versions = new Map<string, number>()
+  const storeId = retention.nextStoreId
+  dispatch({ type: "store.registered" })
   const loading = new Map<string, Promise<string | undefined>>()
 
   const loadBlobUrl = (id: string) => {
-    const existing = retained.get(id)
+    const existing = selectRetainedBlob(retention, id)
 
     if (existing) return Promise.resolve(existing.url)
     const pending = loading.get(id)
@@ -225,7 +218,7 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = {
     return next
   }
 
-  loader = loadBlobUrl
+  storeLoaders.set(storeId, loadBlobUrl)
 
   const putBlob = async (blob: Blob) => {
     const id = await driver.putBlob(blob)
@@ -280,7 +273,10 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = {
   // bytes again: the chunk text itself, or the image Blob (or object URL) the reference carries.
   type Sources = Map<string, { blob: () => Promise<Blob>; chunk?: string }>
 
-  const encode = async (value: unknown, sources: Sources): Promise<unknown> => {
+  // `pin` claims an image id for the running save before its bytes get an object URL.
+  type Pin = (id: string) => void
+
+  const encode = async (value: unknown, sources: Sources, pin: Pin): Promise<unknown> => {
     if (typeof value === "string" && value.length >= draftTextThreshold) {
       const pieces = split(value)
       const ids = await externalize(value)
@@ -291,7 +287,7 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = {
       return { blob: { kind: "text", ids } }
     }
 
-    if (Array.isArray(value)) return Promise.all(value.map((entry) => encode(entry, sources)))
+    if (Array.isArray(value)) return Promise.all(value.map((entry) => encode(entry, sources, pin)))
 
     if (!value || typeof value !== "object") return value
     const item = value as Record<string, unknown>
@@ -320,14 +316,19 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = {
 
       // Drafts own their bytes: the server sweeps a media blob once no message or inbox item references
       // it, so the first save copies the bytes into the draft store and later saves reuse that copy.
-      if (typeof blob.id === "string" && mediaHashFromUri(blob.id) && !retained.has(aliases.get(blob.id) ?? "")) {
+      if (
+        typeof blob.id === "string" &&
+        mediaHashFromUri(blob.id) &&
+        !(retention.aliases.has(blob.id) && selectRetainedBlob(retention, blob.id))
+      ) {
         const data = await loadMediaBlob(blob.id, typeof blob.server === "string" ? blob.server : undefined)
 
         // Nothing serves the bytes anymore; the reference is dropped on read like any blob with no bytes.
         if (!data) return { ...item, blob: { id: blob.id } }
         const id = await driver.putBlob(data)
+        pin(id)
         blobUrl(id, data, grace)
-        aliases.set(blob.id, id)
+        dispatch({ type: "alias.recorded", from: blob.id, to: id })
         sources.set(id, { blob: async () => data })
 
         return { ...item, blob: { id } }
@@ -335,9 +336,10 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = {
 
       if (typeof blob.id === "string") {
         // A live reference keeps the id it was created with; publish the id its bytes now live under.
-        const id = aliases.get(blob.id) ?? blob.id
-        const kept = retained.get(id)?.blob
-        const url = typeof blob.url === "string" && !mediaHashFromUri(blob.url) ? blob.url : retained.get(id)?.url
+        const id = resolveBlobAlias(retention, blob.id)
+        const retained = selectRetainedBlob(retention, id)
+        const kept = retained?.blob
+        const url = typeof blob.url === "string" && !mediaHashFromUri(blob.url) ? blob.url : retained?.url
 
         if (kept) sources.set(id, { blob: async () => kept })
         else if (url) sources.set(id, { blob: () => fetch(url).then((response) => response.blob()) })
@@ -349,7 +351,7 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = {
     }
 
     return Object.fromEntries(
-      await Promise.all(Object.entries(item).map(async ([key, entry]) => [key, await encode(entry, sources)])),
+      await Promise.all(Object.entries(item).map(async ([key, entry]) => [key, await encode(entry, sources, pin)])),
     )
   }
 
@@ -369,7 +371,7 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = {
       // Bytes stay in the store until something renders or sends the image (see resolveBlobUrl);
       // only an image already pinned in this page gets its URL back immediately.
       if (typeof ref.id === "string") {
-        const url = retained.get(aliases.get(ref.id) ?? ref.id)?.url
+        const url = selectRetainedBlob(retention, ref.id)?.url
 
         return { ...item, blob: url ? { id: ref.id, url } : { id: ref.id } }
       }
@@ -383,7 +385,7 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = {
   // Upload the bytes behind `ids` again and return the ids they were stored under. Ids are
   // usually content hashes and come back unchanged, but a store without WebCrypto assigns fresh
   // ones, so callers must rename references rather than assume.
-  const restore = async (ids: readonly string[], sources: Sources) => {
+  const restore = async (ids: readonly string[], sources: Sources, pin: Pin) => {
     const renamed = new Map<string, string>()
     await Promise.all(
       ids.map(async (id) => {
@@ -401,14 +403,9 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = {
           return
         }
 
+        pin(next)
         blobUrl(next, blob, grace)
-
-        if (next === id) return
-
-        // Later encodes of the still-live reference resolve straight to the new id. Re-point any
-        // earlier alias chain so lookups stay one step.
-        for (const [from, to] of aliases) if (to === id) aliases.set(from, next)
-        aliases.set(id, next)
+        dispatch({ type: "alias.recorded", from: id, to: next })
       }),
     )
 
@@ -433,27 +430,26 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = {
     return Object.fromEntries(Object.entries(item).map(([key, entry]) => [key, rename(entry, renamed)]))
   }
 
-  const setDocument = async (key: string, document: unknown) => {
-    const version = (versions.get(key) ?? 0) + 1
-    versions.set(key, version)
-    const sources: Sources = new Map()
-    const encoded = await encode(document, sources)
+  // The image ids the stored document now references, or undefined when a later save or remove
+  // superseded this one before it wrote.
+  const write = async (key: string, document: unknown, attempt: number) => {
+    const pin = (id: string) =>
+      dispatch({ type: "save.referenced", storeId, key, attempt, ids: new Set([id]), graceMs: grace })
 
-    if (versions.get(key) !== version) return
+    const sources: Sources = new Map()
+    const encoded = await encode(document, sources, pin)
+
+    if (!isCurrentSave(retention, storeId, key, attempt)) return undefined
     // The store refuses the write while any referenced blob is missing, so the previous document
     // stays visible until the bytes are back. Covers a blob collected while a cache, another tab,
     // or the composer's history still held its id.
     const missing = await driver.set(key, JSON.stringify(encoded), true)
 
-    if (missing.length === 0) {
-      retain(key, imageIDs(encoded), grace)
+    if (missing.length === 0) return imageIDs(encoded)
 
-      return
-    }
+    const renamed = await restore(missing, sources, pin)
 
-    const renamed = await restore(missing, sources)
-
-    if (versions.get(key) !== version) return
+    if (!isCurrentSave(retention, storeId, key, attempt)) return undefined
     const unrestored = missing.filter((id) => !renamed.has(id))
 
     if (unrestored.length)
@@ -461,11 +457,31 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = {
     // Anything still missing has no bytes anywhere; the owning codec drops such references on read.
     const final = rename(encoded, renamed)
     await driver.set(key, JSON.stringify(final), false)
-    retain(key, imageIDs(final), grace)
+
+    return imageIDs(final)
+  }
+
+  const setDocument = async (key: string, document: unknown) => {
+    dispatch({ type: "save.started", storeId, key, ids: imageIDs(document), graceMs: grace })
+    const attempt = selectLatestMutation(retention, storeId, key)
+
+    await write(key, document, attempt).then(
+      (ids) =>
+        dispatch(
+          ids
+            ? { type: "save.committed", storeId, key, attempt, ids, graceMs: grace }
+            : { type: "save.aborted", storeId, key, attempt, graceMs: grace },
+        ),
+      (error: unknown) => {
+        dispatch({ type: "save.aborted", storeId, key, attempt, graceMs: grace })
+        throw error
+      },
+    )
   }
 
   return {
     getItem: async (key) => {
+      const revision = selectDocumentRevision(retention, storeId, key)
       const value = await driver.get(key)
 
       if (value === null) return null
@@ -474,16 +490,17 @@ export function createDraftStore(driver: Driver, options: { grace?: number } = {
       // Let the owning persistence codec apply its invalid-document policy.
       if (Option.isNone(parsed)) return value
       // A loaded document is live in the composer: pin its images before decode mints their URLs.
-      retain(key, imageIDs(parsed.value), grace)
+      dispatch({ type: "load.succeeded", storeId, key, revision, ids: imageIDs(parsed.value), graceMs: grace })
 
       return JSON.stringify(await decode(parsed.value))
     },
     setItem: (key, value) => setDocument(key, JSON.parse(value)),
     setDocument,
     removeItem: async (key) => {
-      versions.set(key, (versions.get(key) ?? 0) + 1)
+      dispatch({ type: "remove.started", storeId, key })
+      const attempt = selectLatestMutation(retention, storeId, key)
       await driver.remove(key)
-      retain(key, new Set(), grace)
+      dispatch({ type: "remove.committed", storeId, key, attempt, graceMs: grace })
     },
     putBlob,
   }
@@ -619,7 +636,7 @@ function referenced(json: string) {
 // Submits always carry the bytes, so a send never depends on a server blob that may have been swept.
 // A `media:` ref goes out only when no bytes are reachable here; the server still resolves it if it can.
 export async function blobDataUrl(blob: BlobReference, mime: string) {
-  const kept = retained.get(aliases.get(blob.id) ?? blob.id)
+  const kept = selectRetainedBlob(retention, blob.id)
   const url = kept ? undefined : await resolveBlobUrl(blob)
 
   if (!kept && !url && mediaHashFromUri(blob.id)) return blob.id

packages/app/src/runtime/server/client.tsx

@@ -161,7 +161,10 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
     )
   }
 
-  onCleanup(() => dispatchEndpoint({ type: "disposed" }))
+  onCleanup(() => {
+    dispatchEndpoint({ type: "disposed" })
+    transport.dispose()
+  })
 
   return {
     server,
@@ -218,6 +221,8 @@ export interface ServerTransport {
   readonly http: ServerConnection.HttpBase
   /** The current API client. */
   readonly api: ServerApi
+  /** Cancels every queued and in-flight request; later requests reject at once. */
+  dispose(): void
 }
 
 export function createServerTransport(input: {
@@ -252,6 +257,7 @@ export function createServerTransport(input: {
     get api() {
       return state.current.api
     },
+    dispose: queue.dispose,
   }
 }
 

packages/app/src/runtime/server/request-queue-machine.test.ts

@@ -0,0 +1,127 @@
+import { describe, expect, test } from "bun:test"
+import {
+  createRequestQueueState,
+  selectInflightCount,
+  selectQueuedCount,
+  transitionRequestQueue,
+  type RequestPriority,
+  type RequestQueueEffect,
+  type RequestQueueEvent,
+  type RequestQueueState,
+} from "./request-queue-machine"
+
+const config = { limit: 1, slowLimit: 1, stallMs: 2_000 }
+
+function run(events: readonly RequestQueueEvent[], options = config, from = createRequestQueueState()) {
+  return events.reduce<{ state: RequestQueueState; effects: RequestQueueEffect[] }>(
+    (acc, event) => {
+      const next = transitionRequestQueue(acc.state, event, options)
+
+      return { state: next.state, effects: [...acc.effects, ...next.effects] }
+    },
+    { state: from, effects: [] },
+  )
+}
+
+const arrive = (url: string, now: number, priority: RequestPriority = 0, slow = false): RequestQueueEvent => ({
+  type: "request.arrived",
+  method: priority === 2 ? "POST" : "GET",
+  url,
+  slow,
+  priority,
+  now,
+})
+
+describe("transitionRequestQueue", () => {
+  test("an aborted waiter leaves the queue at once and is never started", () => {
+    const queued = run([arrive("/stuck", 0), arrive("/second", 0)])
+    expect(selectQueuedCount(queued.state)).toBe(1)
+
+    const aborted = transitionRequestQueue(queued.state, { type: "request.aborted", id: 2, now: 10 }, config)
+    expect(selectQueuedCount(aborted.state)).toBe(0)
+    expect(selectInflightCount(aborted.state)).toBe(1)
+    expect(aborted.effects).toEqual([{ type: "watcher.stop" }])
+
+    const settled = transitionRequestQueue(aborted.state, { type: "request.settled", id: 1, now: 20 }, config)
+    expect(settled.effects).toEqual([])
+    expect(selectInflightCount(settled.state)).toBe(0)
+  })
+
+  test("a starving background request is reported even when higher priorities queue ahead of it", () => {
+    const queued = run([arrive("/busy", 0), arrive("/background", 0), arrive("/write", 1_900, 2)])
+    expect(queued.state.status === "queued" && queued.state.waiting.map((entry) => entry.url)).toEqual([
+      "/write",
+      "/background",
+    ])
+
+    const due = transitionRequestQueue(queued.state, { type: "watch.due", now: 2_000 }, config)
+    expect(due.effects.map((effect) => effect.type)).toEqual(["watcher.start", "log.stall"])
+  })
+
+  test("after the oldest waiter leaves, the watcher waits out the next one's remaining budget", () => {
+    const queued = run([
+      arrive("/busy", 0),
+      arrive("/a", 0),
+      arrive("/b", 1_200),
+      { type: "request.aborted", id: 2, now: 1_500 },
+    ])
+
+    const early = transitionRequestQueue(queued.state, { type: "watch.due", now: 2_000 }, config)
+    expect(early.effects).toEqual([{ type: "watcher.start", delayMs: 1_200 }])
+    expect(early.state).toBe(queued.state)
+
+    const late = transitionRequestQueue(early.state, { type: "watch.due", now: 3_200 }, config)
+    expect(late.effects.map((effect) => effect.type)).toEqual(["watcher.start", "log.stall"])
+  })
+
+  test("fast reads take free slots while a slow request waits for the slow share", () => {
+    const options = { limit: 3, slowLimit: 1, stallMs: 2_000 }
+
+    const result = run(
+      [arrive("/api/vcs", 0, 0, true), arrive("/api/worktree", 0, 0, true), arrive("/fast", 0)],
+      options,
+    )
+
+    expect(result.effects).toEqual([
+      { type: "request.start", id: 1 },
+      { type: "watcher.start", delayMs: 2_000 },
+      { type: "request.start", id: 3 },
+    ])
+
+    // A fast request finishing frees no slow slot; the slow one finishing does.
+    const fast = transitionRequestQueue(result.state, { type: "request.settled", id: 3, now: 5 }, options)
+    expect(fast.effects).toEqual([])
+    const slow = transitionRequestQueue(fast.state, { type: "request.settled", id: 1, now: 6 }, options)
+    expect(slow.effects).toEqual([{ type: "watcher.stop" }, { type: "request.start", id: 2 }])
+    expect(slow.state.status).toBe("running")
+  })
+
+  test("events that do not apply return the same state", () => {
+    const idle = createRequestQueueState()
+    expect(transitionRequestQueue(idle, { type: "watch.due", now: 1 }, config).state).toBe(idle)
+    expect(transitionRequestQueue(idle, { type: "request.settled", id: 9, now: 1 }, config).state).toBe(idle)
+
+    const queued = run([arrive("/busy", 0), arrive("/a", 0)]).state
+    expect(transitionRequestQueue(queued, { type: "request.settled", id: 2, now: 1 }, config).state).toBe(queued)
+    expect(transitionRequestQueue(queued, { type: "request.aborted", id: 9, now: 1 }, config).state).toBe(queued)
+  })
+
+  test("disposal cancels queued and in-flight requests, stops the watcher and ignores later events", () => {
+    const queued = run([arrive("/busy", 0), arrive("/a", 0), arrive("/b", 0)]).state
+    const disposed = transitionRequestQueue(queued, { type: "disposed" }, config)
+    expect(disposed.effects).toEqual([{ type: "watcher.stop" }, { type: "request.cancel", ids: [2, 3, 1] }])
+    expect(selectInflightCount(disposed.state)).toBe(0)
+
+    for (const event of [
+      arrive("/late", 1),
+      { type: "request.settled", id: 1, now: 1 } as const,
+      { type: "disposed" } as const,
+    ])
+      expect(transitionRequestQueue(disposed.state, event, config)).toEqual({ state: disposed.state, effects: [] })
+
+    const running = run([arrive("/busy", 0)]).state
+    expect(transitionRequestQueue(running, { type: "disposed" }, config).effects).toEqual([
+      { type: "request.cancel", ids: [1] },
+    ])
+  })
+})

packages/app/src/runtime/server/request-queue-machine.ts

@@ -0,0 +1,305 @@
+// Chromium allows six connections per origin. The event stream holds one for the life of the
+// connection and health probes use their own fetch, so the app's API calls stay below that or
+// a burst stalls probes and user actions inside the browser where nothing can observe it.
+export const REQUEST_QUEUE_LIMIT = 4
+
+// Endpoints that shell out to git or walk the filesystem take seconds on a large repository. They
+// may hold at most this many slots, so a session mount's small reads never queue behind them.
+export const REQUEST_QUEUE_SLOW_LIMIT = 2
+
+// A mount legitimately fires a dozen requests at once; only a request that has waited this long
+// for a slot indicates the server is not keeping up.
+export const REQUEST_STALL_MS = 2_000
+
+export const REQUEST_STALL_COOLDOWN_MS = 10_000
+
+const slowRequestPaths = ["/api/vcs", "/api/worktree"]
+
+export type RequestQueueConfig = {
+  readonly limit: number
+  readonly slowLimit: number
+  readonly stallMs: number
+}
+
+export type RequestPriority = 0 | 1 | 2
+
+export type RequestEntry = {
+  readonly id: number
+  readonly method: string
+  readonly url: string
+  /** When the request arrived while it waits, and when it started once in flight. */
+  readonly at: number
+  readonly slow: boolean
+  readonly priority: RequestPriority
+}
+
+export type RequestStallItem = { readonly method: string; readonly url: string; readonly ms: number }
+
+export type RequestStallSnapshot = {
+  readonly limit: number
+  readonly inflight: readonly RequestStallItem[]
+  readonly queued: readonly RequestStallItem[]
+}
+
+/**
+ * One transport's HTTP slots.
+ * - `running`: every request has a slot (possibly none in flight); the stall watcher is off.
+ * - `queued`: at least one request waits for a slot; the stall watcher is armed.
+ * - `disposed`: the transport is gone; every request it held was cancelled.
+ */
+export type RequestQueueState =
+  | {
+      readonly status: "running"
+      readonly nextId: number
+      readonly warnedAt: number
+      readonly inflight: readonly RequestEntry[]
+    }
+  | {
+      readonly status: "queued"
+      readonly nextId: number
+      readonly warnedAt: number
+      readonly inflight: readonly RequestEntry[]
+      readonly waiting: readonly RequestEntry[]
+    }
+  | { readonly status: "disposed" }
+
+export type RequestQueueEvent =
+  | {
+      readonly type: "request.arrived"
+      readonly method: string
+      readonly url: string
+      readonly slow: boolean
+      readonly priority: RequestPriority
+      readonly now: number
+    }
+  | { readonly type: "request.settled"; readonly id: number; readonly now: number }
+  | { readonly type: "request.aborted"; readonly id: number; readonly now: number }
+  | { readonly type: "watch.due"; readonly now: number }
+  | { readonly type: "disposed" }
+
+export type RequestQueueEffect =
+  | { readonly type: "request.start"; readonly id: number }
+  | { readonly type: "request.cancel"; readonly ids: readonly number[] }
+  | { readonly type: "watcher.start"; readonly delayMs: number }
+  | { readonly type: "watcher.stop" }
+  | { readonly type: "log.stall"; readonly data: RequestStallSnapshot }
+
+export type RequestQueueTransition = {
+  readonly state: RequestQueueState
+  readonly effects: readonly RequestQueueEffect[]
+}
+
+export type RequestClassification =
+  | { readonly bypass: true }
+  | { readonly bypass: false; readonly slow: boolean; readonly setup: boolean; readonly priority: RequestPriority }
+
+const none: readonly RequestQueueEffect[] = []
+
+const stop: RequestQueueEffect = { type: "watcher.stop" }
+
+const bypass: RequestClassification = { bypass: true }
+
+export function createRequestQueueState(): RequestQueueState {
+  return { status: "running", nextId: 1, warnedAt: -Infinity, inflight: [] }
+}
+
+export function transitionRequestQueue(
+  state: RequestQueueState,
+  event: RequestQueueEvent,
+  config: RequestQueueConfig,
+): RequestQueueTransition {
+  if (state.status === "disposed") return { state, effects: none }
+
+  if (event.type === "request.arrived") return arrived(state, event, config)
+
+  if (event.type === "request.settled" || event.type === "request.aborted") return left(state, event, config)
+
+  if (event.type === "watch.due") return watched(state, event.now, config)
+
+  if (state.status === "running")
+    return {
+      state: { status: "disposed" },
+      effects: state.inflight.length
+        ? [{ type: "request.cancel", ids: state.inflight.map((entry) => entry.id) }]
+        : none,
+    }
+
+  return {
+    state: { status: "disposed" },
+    effects: [stop, { type: "request.cancel", ids: [...state.waiting, ...state.inflight].map((entry) => entry.id) }],
+  }
+}
+
+export function selectInflightCount(state: RequestQueueState) {
+  return state.status === "disposed" ? 0 : state.inflight.length
+}
+
+export function selectQueuedCount(state: RequestQueueState) {
+  return state.status === "queued" ? state.waiting.length : 0
+}
+
+export function selectQueueDisposed(state: RequestQueueState) {
+  return state.status === "disposed"
+}
+
+export function isRequestInflight(state: RequestQueueState, id: number) {
+  return state.status !== "disposed" && state.inflight.some((entry) => entry.id === id)
+}
+
+/**
+ * The event stream bypasses the queue. Writes go first. Session lists and the first transcript page unblock
+ * navigation, so they go ahead of location reads. Older pages and filtered indexes keep background priority.
+ */
+export function classifyRequest(method: string, url: URL): RequestClassification {
+  const pathname = url.pathname
+
+  // The event stream is long-lived; never count it against the request budget.
+  if (pathname === "/api/event") return bypass
+
+  const initialMessages =
+    /^\/api\/session\/[^/]+\/message$/.test(pathname) &&
+    !url.searchParams.has("cursor") &&
+    !url.searchParams.has("before") &&
+    !url.searchParams.has("type") &&
+    url.searchParams.get("order") !== "asc"
+
+  return {
+    bypass: false,
+    slow: slowRequestPaths.some((path) => pathname === path || pathname.startsWith(`${path}/`)),
+    setup: method === "POST" && pathname === "/api/worktree",
+    priority: method !== "GET" ? 2 : pathname === "/api/session" || initialMessages ? 1 : 0,
+  }
+}
+
+type OpenState = Exclude<RequestQueueState, { status: "disposed" }>
+
+function arrived(
+  state: OpenState,
+  event: Extract<RequestQueueEvent, { type: "request.arrived" }>,
+  config: RequestQueueConfig,
+): RequestQueueTransition {
+  const entry: RequestEntry = {
+    id: state.nextId,
+    method: event.method,
+    url: event.url,
+    at: event.now,
+    slow: event.slow,
+    priority: event.priority,
+  }
+
+  const nextId = entry.id + 1
+
+  if (canStart(state.inflight, entry.slow, config))
+    return {
+      state: { ...state, nextId, inflight: [...state.inflight, entry] },
+      effects: [{ type: "request.start", id: entry.id }],
+    }
+
+  if (state.status === "running")
+    return {
+      state: { status: "queued", nextId, warnedAt: state.warnedAt, inflight: state.inflight, waiting: [entry] },
+      effects: [{ type: "watcher.start", delayMs: config.stallMs }],
+    }
+
+  // FIFO within each priority tier.
+  const position = state.waiting.findIndex((item) => item.priority < entry.priority)
+
+  return {
+    state: {
+      ...state,
+      nextId,
+      waiting: position === -1 ? [...state.waiting, entry] : state.waiting.toSpliced(position, 0, entry),
+    },
+    effects: none,
+  }
+}
+
+function left(
+  state: OpenState,
+  event: Extract<RequestQueueEvent, { type: "request.settled" | "request.aborted" }>,
+  config: RequestQueueConfig,
+): RequestQueueTransition {
+  const index = state.inflight.findIndex((entry) => entry.id === event.id)
+
+  if (index === -1) {
+    // Only an abort can take a request out of the queue before it starts.
+    if (state.status === "running" || event.type === "request.settled") return { state, effects: none }
+
+    const position = state.waiting.findIndex((entry) => entry.id === event.id)
+
+    if (position === -1) return { state, effects: none }
+
+    const waiting = state.waiting.toSpliced(position, 1)
+
+    if (waiting.length) return { state: { ...state, waiting }, effects: none }
+
+    return { state: running(state, state.inflight), effects: [stop] }
+  }
+
+  const inflight = state.inflight.toSpliced(index, 1)
+
+  if (state.status === "running") return { state: { ...state, inflight }, effects: none }
+
+  // A slow request waits its turn behind faster ones while the slow slots are full.
+  const next = state.waiting.findIndex((entry) => canStart(inflight, entry.slow, config))
+
+  if (next === -1) return { state: { ...state, inflight }, effects: none }
+
+  const promoted = { ...state.waiting[next]!, at: event.now }
+  const waiting = state.waiting.toSpliced(next, 1)
+  const start: RequestQueueEffect = { type: "request.start", id: promoted.id }
+
+  if (waiting.length) return { state: { ...state, inflight: [...inflight, promoted], waiting }, effects: [start] }
+
+  return { state: running(state, [...inflight, promoted]), effects: [stop, start] }
+}
+
+// Debug exports include the console, so list what the server is busy with while requests wait. Priority
+// insertion puts newer requests ahead of older ones, so the longest wait is anywhere in the queue.
+function watched(state: OpenState, now: number, config: RequestQueueConfig): RequestQueueTransition {
+  if (state.status !== "queued") return { state, effects: none }
+
+  const waited = now - state.waiting.reduce((oldest, entry) => Math.min(oldest, entry.at), now)
+
+  if (waited >= config.stallMs && now - state.warnedAt >= REQUEST_STALL_COOLDOWN_MS)
+    return {
+      state: { ...state, warnedAt: now },
+      effects: [
+        { type: "watcher.start", delayMs: config.stallMs },
+        {
+          type: "log.stall",
+          data: {
+            limit: config.limit,
+            inflight: state.inflight.map((entry) => describe(entry, now)),
+            queued: state.waiting.map((entry) => describe(entry, now)),
+          },
+        },
+      ],
+    }
+
+  return {
+    state,
+    effects: [
+      {
+        type: "watcher.start",
+        delayMs: waited >= config.stallMs ? config.stallMs : Math.max(1, config.stallMs - waited),
+      },
+    ],
+  }
+}
+
+function running(state: OpenState, inflight: readonly RequestEntry[]): RequestQueueState {
+  return { status: "running", nextId: state.nextId, warnedAt: state.warnedAt, inflight }
+}
+
+function canStart(inflight: readonly RequestEntry[], slow: boolean, config: RequestQueueConfig) {
+  if (inflight.length >= config.limit) return false
+
+  if (!slow) return true
+
+  return inflight.reduce((count, entry) => (entry.slow ? count + 1 : count), 0) < config.slowLimit
+}
+
+function describe(entry: RequestEntry, now: number): RequestStallItem {
+  return { method: entry.method, url: entry.url, ms: now - entry.at }
+}

packages/app/src/runtime/server/request-queue.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, test } from "bun:test"
 import { createRequestQueue } from "./request-queue"
+import type { RequestStallSnapshot } from "./request-queue-machine"
 
 function setup(input?: {
   limit?: number
@@ -9,7 +10,7 @@ function setup(input?: {
   setupHeadersTimeoutMs?: number
 }) {
   const pending: Array<{ url: string; signal: AbortSignal; resolve: () => void }> = []
-  const logs: Array<{ message: string; data: Record<string, unknown> }> = []
+  const logs: Array<{ message: string; data: RequestStallSnapshot }> = []
   let clock = 0
 
   const queue = createRequestQueue({
@@ -158,17 +159,41 @@ describe("createRequestQueue", () => {
     expect(input.queue.inflight()).toBe(1)
   })
 
-  test("aborted requests leave the queue without being sent", async () => {
-    const input = setup({ limit: 1 })
+  test("a queued request leaves without being sent as soon as its caller aborts, while every slot stays stuck", async () => {
+    const input = setup({ limit: 1, stallMs: 5 })
     const controller = new AbortController()
-    void input.queue.fetch("http://server/api/first")
+    void input.queue.fetch("http://server/api/stuck").catch(() => undefined)
     const aborted = input.queue.fetch("http://server/api/second", { signal: controller.signal })
+    expect(input.queue.queued()).toBe(1)
     controller.abort()
-    await input.settle()
-    input.pending[0]!.resolve()
     await expect(aborted).rejects.toBeInstanceOf(DOMException)
-    expect(input.pending.map((item) => new URL(item.url).pathname)).toEqual(["/api/first"])
+    expect(input.queue.queued()).toBe(0)
+    input.tick(1_000)
+    await new Promise((resolve) => setTimeout(resolve, 20))
+    expect(input.logs).toEqual([])
+    expect(input.pending.map((item) => new URL(item.url).pathname)).toEqual(["/api/stuck"])
+  })
+
+  test("an already-aborted request never joins a full queue", async () => {
+    const input = setup({ limit: 1 })
+    void input.queue.fetch("http://server/api/stuck").catch(() => undefined)
+    await expect(input.queue.fetch("http://server/api/next", { signal: AbortSignal.abort() })).rejects.toBeInstanceOf(
+      DOMException,
+    )
+    expect(input.queue.queued()).toBe(0)
+  })
+
+  test("disposing cancels queued and in-flight requests and rejects later ones", async () => {
+    const input = setup({ limit: 1 })
+    const inflight = input.queue.fetch("http://server/api/a")
+    const queued = input.queue.fetch("http://server/api/b")
+    input.queue.dispose()
+    const errors = await Promise.all([inflight, queued].map((request) => request.catch((cause: unknown) => cause)))
+    expect(errors.map((error) => error instanceof DOMException && error.name)).toEqual(["AbortError", "AbortError"])
+    expect(input.pending.map((item) => new URL(item.url).pathname)).toEqual(["/api/a"])
     expect(input.queue.inflight()).toBe(0)
+    await expect(input.queue.fetch("http://server/api/c")).rejects.toBeInstanceOf(DOMException)
+    expect(input.pending).toHaveLength(1)
   })
 
   test("a request the server never answers times out and frees its slot", async () => {
@@ -178,7 +203,7 @@ describe("createRequestQueue", () => {
     expect(input.queue.queued()).toBe(1)
     const error = await dead.catch((cause: unknown) => cause)
     expect(error).toBeInstanceOf(DOMException)
-    expect((error as DOMException).name).toBe("TimeoutError")
+    expect(error instanceof DOMException && error.name).toBe("TimeoutError")
     expect(input.pending.map((item) => new URL(item.url).pathname)).toEqual(["/api/dead", "/api/next"])
     input.pending[1]!.resolve()
     await expect(next).resolves.toBeInstanceOf(Response)
@@ -196,7 +221,7 @@ describe("createRequestQueue", () => {
     ]
 
     const errors = await Promise.all(others.map((request) => request.catch((cause: unknown) => cause)))
-    expect(errors.map((error) => (error as DOMException).name)).toEqual([
+    expect(errors.map((error) => error instanceof DOMException && error.name)).toEqual([
       "TimeoutError",
       "TimeoutError",
       "TimeoutError",

packages/app/src/runtime/server/request-queue.ts

@@ -1,19 +1,18 @@
-type Entry = { method: string; url: string; at: number; slow: boolean; priority: number }
-
-// Chromium allows six connections per origin. The event stream holds one for the life of the
-// connection and health probes use their own fetch, so the app's API calls stay below that or
-// a burst stalls probes and user actions inside the browser where nothing can observe it.
-const requestQueueLimit = 4
-
-// Endpoints that shell out to git or walk the filesystem take seconds on a large repository. They
-// may hold at most this many slots, so a session mount's small reads never queue behind them.
-const requestQueueSlowLimit = 2
-
-const slowRequestPaths = ["/api/vcs", "/api/worktree"]
-
-// A mount legitimately fires a dozen requests at once; only a request that has waited this long
-// for a slot indicates the server is not keeping up.
-const requestStallMs = 2_000
+import {
+  classifyRequest,
+  createRequestQueueState,
+  isRequestInflight,
+  REQUEST_QUEUE_LIMIT,
+  REQUEST_QUEUE_SLOW_LIMIT,
+  REQUEST_STALL_MS,
+  selectInflightCount,
+  selectQueueDisposed,
+  selectQueuedCount,
+  transitionRequestQueue,
+  type RequestQueueEffect,
+  type RequestQueueEvent,
+  type RequestStallSnapshot,
+} from "./request-queue-machine"
 
 // A socket that dies while the device sleeps can leave fetch waiting for response headers until the
 // OS gives up on TCP retransmits, which takes minutes. Bound that so a dead request frees its slot
@@ -25,14 +24,6 @@ const requestHeadersTimeoutMs = 60_000
 // kills the script midway and leaves a registered but half-initialised worktree behind.
 const setupRequestHeadersTimeoutMs = 10 * 60_000
 
-function isSlowRequest(pathname: string) {
-  return slowRequestPaths.some((path) => pathname === path || pathname.startsWith(`${path}/`))
-}
-
-function isSetupRequest(method: string, pathname: string) {
-  return method === "POST" && pathname === "/api/worktree"
-}
-
 export function createRequestQueue(input: {
   fetch: typeof globalThis.fetch
   limit?: number
@@ -40,121 +31,122 @@ export function createRequestQueue(input: {
   stallMs?: number
   headersTimeoutMs?: number
   setupHeadersTimeoutMs?: number
-  log?: (message: string, data: Record<string, unknown>) => void
+  log?: (message: string, data: RequestStallSnapshot) => void
   now?: () => number
 }) {
-  const limit = input.limit ?? requestQueueLimit
-  const slowLimit = input.slowLimit ?? requestQueueSlowLimit
-  const stallMs = input.stallMs ?? requestStallMs
+  const config = {
+    limit: input.limit ?? REQUEST_QUEUE_LIMIT,
+    slowLimit: input.slowLimit ?? REQUEST_QUEUE_SLOW_LIMIT,
+    stallMs: input.stallMs ?? REQUEST_STALL_MS,
+  }
+
   const headersTimeoutMs = input.headersTimeoutMs ?? requestHeadersTimeoutMs
   const setupHeadersTimeoutMs = input.setupHeadersTimeoutMs ?? setupRequestHeadersTimeoutMs
   // Call the browser fetch unbound; `input.fetch(...)` would make `this` the options object.
   const base = input.fetch
   const now = input.now ?? Date.now
   const log = input.log ?? ((message, data) => console.warn(`[server-request-queue] ${message}`, data))
-  const inflight = new Set<Entry>()
-  const waiting: Array<{ entry: Entry; start: () => void }> = []
-  let warned = -Infinity
+  let queue = createRequestQueueState()
   let watcher: ReturnType<typeof setTimeout> | undefined
+  // Callers suspended until their queued request gets a slot, and the controllers of requests in flight.
+  const waiters = new Map<number, { grant: () => void; cancel: () => void }>()
+  const controllers = new Map<number, AbortController>()
 
-  const describe = (entry: Entry) => ({ method: entry.method, url: entry.url, ms: now() - entry.at })
+  const run = (effect: RequestQueueEffect) => {
+    if (effect.type === "request.start") return waiters.get(effect.id)?.grant()
 
-  // Debug exports include the console, so list what the server is busy with while requests wait.
-  const watch = () => {
-    watcher = undefined
-    const oldest = waiting[0]?.entry
-
-    if (!oldest) return
-
-    if (now() - oldest.at >= stallMs && now() - warned >= 10_000) {
-      warned = now()
-      log("server thrashing detected", {
-        limit,
-        inflight: [...inflight].map(describe),
-        queued: waiting.map((item) => describe(item.entry)),
+    if (effect.type === "request.cancel")
+      return effect.ids.forEach((id) => {
+        waiters.get(id)?.cancel()
+        controllers.get(id)?.abort(aborted())
       })
-    }
 
-    watcher = setTimeout(watch, stallMs)
+    if (effect.type === "log.stall") return log("server thrashing detected", effect.data)
+
+    clearTimeout(watcher)
+    watcher =
+      effect.type === "watcher.start"
+        ? setTimeout(() => dispatch({ type: "watch.due", now: now() }), effect.delayMs)
+        : undefined
   }
 
-  const canStart = (entry: Entry) => {
-    if (inflight.size >= limit) return false
-
-    if (!entry.slow) return true
-
-    return [...inflight].filter((item) => item.slow).length < slowLimit
+  const dispatch = (event: RequestQueueEvent) => {
+    const result = transitionRequestQueue(queue, event, config)
+    queue = result.state
+    result.effects.forEach(run)
   }
 
-  // FIFO within each priority tier, except a slow request waits its turn behind faster ones while the slow slots are full.
-  const release = (entry: Entry) => {
-    inflight.delete(entry)
-    const index = waiting.findIndex((item) => canStart(item.entry))
-
-    if (index === -1) return
-    waiting.splice(index, 1)[0]?.start()
-  }
-
-  const acquire = (entry: Entry) => {
-    // A free slot must start fetch before the caller's synchronous UI work.
-    // Awaiting an already-resolved promise postpones that dispatch until after it.
-    if (canStart(entry)) {
-      inflight.add(entry)
-
-      return
-    }
-
-    return new Promise<void>((resolve) => {
-      const start = () => {
-        entry.at = now()
-        inflight.add(entry)
-        resolve()
+  const waitForSlot = (id: number, signal: AbortSignal) =>
+    new Promise<void>((resolve, reject) => {
+      const leave = () => {
+        signal.removeEventListener("abort", abort)
+        waiters.delete(id)
       }
 
-      const position = waiting.findIndex((item) => item.entry.priority < entry.priority)
-      waiting.splice(position === -1 ? waiting.length : position, 0, { entry, start })
-      watcher ??= setTimeout(watch, stallMs)
+      // A caller that gives up while queued leaves at once instead of waiting for a slot it no longer needs.
+      const abort = () => {
+        leave()
+        dispatch({ type: "request.aborted", id, now: now() })
+        reject(reason(signal))
+      }
+
+      signal.addEventListener("abort", abort, { once: true })
+      waiters.set(id, {
+        grant: () => {
+          leave()
+          resolve()
+        },
+        cancel: () => {
+          leave()
+          reject(aborted())
+        },
+      })
     })
-  }
 
   const fetch: typeof globalThis.fetch = Object.assign(
     async (resource: RequestInfo | URL, init?: RequestInit) => {
       const request = new Request(resource, init)
-      const url = new URL(request.url)
-      const pathname = url.pathname
+      const kind = classifyRequest(request.method, new URL(request.url))
 
-      // The event stream is long-lived; never count it against the request budget.
-      if (pathname === "/api/event") return base(request)
-      // Writes go first. Session lists and the first transcript page unblock navigation, so they
-      // go ahead of location reads. Older pages and filtered indexes keep background priority.
-      const initialMessages =
-        /^\/api\/session\/[^/]+\/message$/.test(pathname) &&
-        !url.searchParams.has("cursor") &&
-        !url.searchParams.has("before") &&
-        !url.searchParams.has("type") &&
-        url.searchParams.get("order") !== "asc"
-      const priority = request.method !== "GET" ? 2 : pathname === "/api/session" || initialMessages ? 1 : 0
-      const entry = { method: request.method, url: request.url, at: now(), slow: isSlowRequest(pathname), priority }
-      const queued = acquire(entry)
+      if (kind.bypass) return base(request)
 
-      if (queued) await queued
+      if (request.signal.aborted || queue.status === "disposed") throw reason(request.signal)
 
-      if (request.signal.aborted) {
-        release(entry)
-        throw request.signal.reason ?? new DOMException("The operation was aborted.", "AbortError")
+      const id = queue.nextId
+      dispatch({
+        type: "request.arrived",
+        method: request.method,
+        url: request.url,
+        slow: kind.slow,
+        priority: kind.priority,
+        now: now(),
+      })
+
+      // A free slot must start fetch before the caller's synchronous UI work.
+      // Awaiting an already-resolved promise postpones that dispatch until after it.
+      if (!isRequestInflight(queue, id)) await waitForSlot(id, request.signal)
+
+      // Aborted or disposed between the grant and this continuation.
+      if (request.signal.aborted || selectQueueDisposed(queue)) {
+        dispatch({ type: "request.aborted", id, now: now() })
+        throw reason(request.signal)
       }
 
       const controller = new AbortController()
-      request.signal.addEventListener("abort", () => controller.abort(request.signal.reason), { once: true })
+      controllers.set(id, controller)
+      const abort = () => controller.abort(request.signal.reason)
+      request.signal.addEventListener("abort", abort, { once: true })
 
       const timer = setTimeout(
         () => controller.abort(new DOMException("Timed out waiting for the server to respond", "TimeoutError")),
-        isSetupRequest(request.method, pathname) ? setupHeadersTimeoutMs : headersTimeoutMs,
+        kind.setup ? setupHeadersTimeoutMs : headersTimeoutMs,
       )
 
       return base(new Request(request, { signal: controller.signal })).finally(() => {
         clearTimeout(timer)
-        release(entry)
+        request.signal.removeEventListener("abort", abort)
+        controllers.delete(id)
+        dispatch({ type: "request.settled", id, now: now() })
       })
     },
     // Bun's fetch type carries preconnect; the browser never calls it.
@@ -163,7 +155,17 @@ export function createRequestQueue(input: {
 
   return {
     fetch,
-    inflight: () => inflight.size,
-    queued: () => waiting.length,
+    inflight: () => selectInflightCount(queue),
+    queued: () => selectQueuedCount(queue),
+    /** Cancels every queued and in-flight request; later requests reject at once. */
+    dispose: () => dispatch({ type: "disposed" }),
   }
 }
+
+function aborted() {
+  return new DOMException("The operation was aborted.", "AbortError")
+}
+
+function reason(signal: AbortSignal) {
+  return signal.aborted ? (signal.reason ?? aborted()) : aborted()
+}