fix: track realtime physical work from invocation to settlement

RT-RR-01. An effect or recovery task was registered as retained only after its
public wait expired, so a close() that arrived first saw an empty set and
reported quiescence while the raw task was still running against the authority.
Tasks are now registered when they are created and removed when they settle;
DRAINING keeps its narrower meaning through a separate timed-out set.

RT-RR-02. Admission happened when an event was queued; execution is a second
decision. A queue entry admitted before the stream entered DRAINING no longer
starts running inside it. And an abandoned task may have applied part of its
effect, so the resume token it was based on is discarded and recovery is
required explicitly — the next ordinary event can no longer skip authoritative
recovery on the strength of state a timed-out effect may have invalidated.

RT-RR-03. close() cached the first timeout forever, so a writer that later
settled could never be proved quiescent and the retained registry could never be
pruned. Only an in-flight close is shared now, every writer a close fences is
retained until its tail actually settles, and the tail prunes itself. A second
close therefore converges to success once the writer finishes.

RT-RR-04. Checkpoint work joins writer tails in the physical-task registry from
invocation to settlement and is drained on the same terms.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/adapters/realtime/live-poll-handoff-coordinator.ts

@@ -173,6 +173,23 @@ export function createLivePollHandoffCoordinator<Value>(
    * running. Membership keeps `close()` honest about quiescence.
    */
   const retiredWriters = new Set<InternalWriterLease<Value>>();
+  /**
+   * RT-RR-04. Checkpoint work is an external authority call like a writer tail,
+   * so it belongs in a physical-task registry from invocation to settlement.
+   * Racing it against a timeout bounded the public wait but left `close()` free
+   * to report success while the checkpoint was still running.
+   */
+  const checkpointTasks = new Set<Promise<unknown>>();
+
+  /** RT-RR-03. Prunes a settled writer without needing another `close()`. */
+  function trackRetiredWriter(lease: InternalWriterLease<Value>): void {
+    retiredWriters.add(lease);
+    void lease.tail
+      .catch(() => undefined)
+      .finally(() => {
+        retiredWriters.delete(lease);
+      });
+  }
 
   active = createWriterLease(dependencies.initial.writer);
 
@@ -573,6 +590,12 @@ export function createLivePollHandoffCoordinator<Value>(
         (value) => ({ kind: "VALUE" as const, value }),
         () => ({ kind: "REJECTED" as const }),
       );
+    // RT-RR-04. Registered at the moment the authority is called, and removed
+    // only when it settles, so `close()` cannot report quiescence over it.
+    checkpointTasks.add(operation);
+    void operation.finally(() => {
+      checkpointTasks.delete(operation);
+    });
     const timeout = Promise.resolve()
       .then(async () => {
         await clock.sleep(
@@ -643,6 +666,33 @@ export function createLivePollHandoffCoordinator<Value>(
     return realtimeSuccess(undefined);
   }
 
+  /**
+   * RT-RR-04. The writer-tail quiescence bound applied to any retained physical
+   * task, so checkpoint work is proved settled on the same terms.
+   */
+  async function awaitTaskQuiescence(
+    task: Promise<unknown>,
+  ): Promise<QuiescenceOutcome> {
+    const timer = new AbortController();
+    const settled = task.then(
+      () => "QUIESCED" as const,
+      () => "QUIESCED" as const,
+    );
+    const timeout = Promise.resolve()
+      .then(async () => {
+        await clock.sleep(limits.quiescenceTimeoutMs, timer.signal);
+        return "TIMED_OUT" as const;
+      })
+      .catch(() =>
+        timer.signal.aborted
+          ? ("QUIESCED" as const)
+          : ("TIMER_FAILED" as const),
+      );
+    const outcome = await Promise.race([settled, timeout]);
+    timer.abort();
+    return outcome;
+  }
+
   async function awaitQuiescence(
     lease: InternalWriterLease<Value>,
   ): Promise<QuiescenceOutcome> {
@@ -670,7 +720,13 @@ export function createLivePollHandoffCoordinator<Value>(
   }
 
   function close(): Promise<RealtimeResult<void>> {
-    closePromise ??= performClose();
+    // RT-RR-03. Only a close that is still running is shared. Caching the first
+    // timeout forever meant a writer that later settled could never be proved
+    // quiescent: every subsequent close replayed the stale failure and the
+    // retained registry could never be pruned.
+    closePromise ??= performClose().finally(() => {
+      closePromise = null;
+    });
     return closePromise;
   }
 
@@ -693,18 +749,23 @@ export function createLivePollHandoffCoordinator<Value>(
     probe = null;
     selectedProbe?.buffer.splice(0);
     if (selectedProbe) selectedProbe.bufferedBytes = 0;
-    for (const writer of writers) writer.controller.abort();
-    const outcomes = await Promise.all(
-      writers.map(async (writer) => await awaitQuiescence(writer)),
-    );
+    for (const writer of writers) {
+      // RT-RR-03. Every writer this close fences is retained until its tail
+      // actually settles, so a later close still sees a writer that has not
+      // finished — and stops seeing it the moment it does.
+      trackRetiredWriter(writer);
+      writer.controller.abort();
+    }
+    const outcomes = await Promise.all([
+      ...writers.map(async (writer) => await awaitQuiescence(writer)),
+      // RT-RR-04. Checkpoint work is drained on the same terms as a writer tail.
+      ...[...checkpointTasks].map(
+        async (task) => await awaitTaskQuiescence(task),
+      ),
+    ]);
     quiescing = null;
     transitionCandidate = null;
     transitioning = false;
-    for (const [index, writer] of writers.entries()) {
-      // Only an actually settled writer leaves the retired set; the rest keep
-      // the coordinator DRAINING.
-      if (outcomes[index] === "QUIESCED") retiredWriters.delete(writer);
-    }
     if (outcomes.includes("TIMED_OUT")) {
       return handoffFailure("IDLE_TIMEOUT", "CLOSE");
     }
@@ -802,7 +863,7 @@ export function createLivePollHandoffCoordinator<Value>(
       transitionCandidate,
     ])) {
       writer.controller.abort();
-      retiredWriters.add(writer);
+      trackRetiredWriter(writer);
     }
     active = null;
     if (probe) {

src/adapters/realtime/stream-coordinator.ts

@@ -129,10 +129,23 @@ type StreamState = {
   closed: boolean;
   lifecycle: RealtimeStreamLifecycle;
   /**
-   * Tasks whose public wait already ended but whose underlying promise has not
-   * settled. They keep the stream `DRAINING` and block new admission.
+   * RT-RR-01. Every physical task this coordinator has handed to an external
+   * authority, from the moment of the call until it settles. Registering only
+   * after a timeout meant a `close()` that arrived first saw an empty set and
+   * reported quiescence while the raw task was still running.
    */
   retainedTasks: Set<Promise<unknown>>;
+  /**
+   * The subset of `retainedTasks` whose public wait already ended. These are
+   * what keep the stream `DRAINING` and block new admission.
+   */
+  timedOutTasks: Set<Promise<unknown>>;
+  /**
+   * RT-RR-02. Set when a task was abandoned at its deadline. The resume token
+   * is discarded with it, so the next admitted event cannot skip authoritative
+   * recovery on the strength of state a timed-out effect may have invalidated.
+   */
+  recoveryRequired: boolean;
 };
 
 const SNAPSHOT_CHECKPOINT_KEYS = Object.freeze([
@@ -186,6 +199,19 @@ export function createRealtimeStreamCoordinator(
     revokeAndAbort: () => void,
   ): Promise<Value | typeof TASK_TIMED_OUT> {
     task.catch(() => {});
+    // RT-RR-01. The task is a physical effect the moment it is created, so it
+    // is registered here rather than when its public wait happens to expire.
+    state.retainedTasks.add(task);
+    void task
+      .catch(() => {})
+      .finally(() => {
+        state.retainedTasks.delete(task);
+        state.timedOutTasks.delete(task);
+        if (state.timedOutTasks.size === 0 && state.lifecycle === "DRAINING") {
+          state.lifecycle = state.closed ? "CLOSED" : "OPEN";
+          if (!state.closed) state.freshness = "STALE";
+        }
+      });
     let handle: unknown;
     const timeout = new Promise<typeof TASK_TIMED_OUT>((resolve) => {
       handle = scheduleTimeout(() => resolve(TASK_TIMED_OUT), timeoutMs);
@@ -197,16 +223,12 @@ export function createRealtimeStreamCoordinator(
     // The commit capability is revoked immediately; the work itself is not.
     revokeAndAbort();
     state.lifecycle = "DRAINING";
-    state.retainedTasks.add(task);
-    void task
-      .catch(() => {})
-      .finally(() => {
-        state.retainedTasks.delete(task);
-        if (state.retainedTasks.size === 0 && state.lifecycle === "DRAINING") {
-          state.lifecycle = state.closed ? "CLOSED" : "OPEN";
-          if (!state.closed) state.freshness = "STALE";
-        }
-      });
+    state.timedOutTasks.add(task);
+    // RT-RR-02. An abandoned task may have applied part of its effect, so the
+    // resume token it was based on is no longer authoritative evidence.
+    state.recoveryRequired = true;
+    state.resumeState = null;
+    state.freshness = "UNKNOWN";
     return TASK_TIMED_OUT;
   }
 
@@ -232,6 +254,8 @@ export function createRealtimeStreamCoordinator(
       closed: false,
       lifecycle: "OPEN",
       retainedTasks: new Set(),
+      timedOutTasks: new Set(),
+      recoveryRequired: false,
     });
   }
 
@@ -352,6 +376,10 @@ export function createRealtimeStreamCoordinator(
       return realtimeFailure("ABORTED", "RECEIVE");
     }
     if (closed || state.closed) return dropped("CLOSED");
+    // RT-RR-02. Admission happened when this event was queued; execution is a
+    // second decision. A queue entry admitted before the stream entered
+    // DRAINING must not start running inside it.
+    if (state.lifecycle === "DRAINING") return dropped("CLOSED");
     if (expectedGeneration !== state.processingGeneration) {
       return dropped("SCOPE_FENCED");
     }
@@ -367,7 +395,11 @@ export function createRealtimeStreamCoordinator(
         signal,
       );
     }
-    if (state.freshness === "UNKNOWN" || !state.resumeState) {
+    if (
+      state.recoveryRequired ||
+      state.freshness === "UNKNOWN" ||
+      !state.resumeState
+    ) {
       return recoverForAccept(state, "INITIALIZE", true, signal);
     }
     if (event.envelope.streamEpoch !== state.resumeState.streamEpoch) {
@@ -782,6 +814,9 @@ export function createRealtimeStreamCoordinator(
         validatedResumeState as RealtimeRecoveryCheckpoint;
 
       state.resumeState = resumeState;
+      // RT-RR-02. Authoritative recovery is the only thing that clears the
+      // requirement a timed-out task imposed.
+      state.recoveryRequired = false;
       state.awaitingTransportBarrier =
         recoveryRequiresTransportBarrier(state.registration);
       state.barrierCheckpoint = state.awaitingTransportBarrier

tests/unit/realtime/live-poll-handoff-coordinator.test.ts

@@ -527,12 +527,87 @@ describe("live/poll authoritative writer handoff", () => {
       ok: false,
       error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" },
     });
-    await expect(harness.coordinator.close()).resolves.toMatchObject({
+    // RT-RR-03. A second close re-runs rather than replaying a cached verdict.
+    // The writer is still hung, so it still reports a timeout.
+    const second = harness.coordinator.close();
+    await flush();
+    harness.clock.advance(100);
+    await expect(second).resolves.toMatchObject({
       ok: false,
       error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" },
     });
   });
 
+  /**
+   * RT-RR-03. Caching the first timeout forever meant a writer that later
+   * settled could never be proved quiescent: every subsequent close replayed
+   * the stale failure and the retained registry could never be pruned.
+   */
+  it("converges to success once a retired writer finally settles", async () => {
+    let release: ((value: RealtimeResult<void>) => void) | undefined;
+    const harness = createHarness({
+      apply: vi.fn(
+        async () =>
+          await new Promise<RealtimeResult<void>>((resolve) => {
+            release = resolve;
+          }),
+      ),
+    });
+    const live = harness.coordinator.currentWriter()!;
+    void live.write("late-settle", 8);
+    await flush();
+
+    const first = harness.coordinator.close();
+    await flush();
+    harness.clock.advance(100);
+    await expect(first).resolves.toMatchObject({
+      ok: false,
+      error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" },
+    });
+
+    // The writer finishes after the first close gave up.
+    release?.(realtimeSuccess(undefined));
+    await flush();
+    await flush();
+
+    const second = harness.coordinator.close();
+    await flush();
+    harness.clock.advance(100);
+    await expect(second).resolves.toMatchObject({ ok: true });
+  });
+
+  /**
+   * RT-RR-04. Checkpoint work is an external authority call like a writer
+   * tail. Racing it against a timeout bounded the public wait but left it out
+   * of the retained registry, so `close()` could report quiescence while the
+   * checkpoint was still running.
+   */
+  it("does not report quiescence while a checkpoint is still running", async () => {
+    let checkpointSignal: AbortSignal | undefined;
+    const harness = createHarness({
+      recover: vi.fn(async ({ signal }) => {
+        checkpointSignal = signal;
+        return await new Promise<RealtimeResult<void>>(() => {});
+      }),
+    });
+
+    const transition = harness.coordinator.switchToPoll();
+    await flush();
+    expect(checkpointSignal).toBeDefined();
+
+    const closing = harness.coordinator.close();
+    await flush();
+    harness.clock.advance(100);
+    await expect(closing).resolves.toMatchObject({
+      ok: false,
+      error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" },
+    });
+
+    harness.clock.advance(100);
+    await flush();
+    await transition;
+  });
+
   it("rejects invalid initial authority and resource ceilings", () => {
     expect(() =>
       createLivePollHandoffCoordinator({

tests/unit/realtime/stream-coordinator.test.ts

@@ -943,6 +943,118 @@ describe("transport-independent realtime stream coordinator", () => {
   });
 });
 
+/**
+ * RT-RR-01 / RT-RR-02. A physical effect exists from the moment the coordinator
+ * calls the authority, not from the moment its public wait expires. Registering
+ * only on timeout let a `close()` that arrived first see an empty retained set
+ * and report quiescence while the raw task was still running against the
+ * authority.
+ */
+describe("realtime physical task ownership", () => {
+  it("does not report quiescence while an apply is still running", async () => {
+    const applyGate = deferred<RealtimeResult<void>>();
+    const harness = createHarness({
+      apply: async () => await applyGate.promise,
+      taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 20 },
+    });
+    await harness.initialize();
+
+    const applying = harness.coordinator.accept(harness.event());
+    await Promise.resolve();
+    await Promise.resolve();
+
+    // close() arrives long before the effect deadline.
+    const closed = await harness.coordinator.close();
+    expect(closed.ok).toBe(false);
+    expect(closed.ok ? null : closed.error.kind).toBe("IDLE_TIMEOUT");
+    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING");
+
+    applyGate.resolve(realtimeSuccess(undefined));
+    await applying;
+  });
+
+  it("reports quiescence once the raw task settles", async () => {
+    const applyGate = deferred<RealtimeResult<void>>();
+    const harness = createHarness({
+      apply: async () => await applyGate.promise,
+      taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 200 },
+    });
+    await harness.initialize();
+
+    const applying = harness.coordinator.accept(harness.event());
+    await Promise.resolve();
+    const closing = harness.coordinator.close();
+    applyGate.resolve(realtimeSuccess(undefined));
+    await applying;
+
+    expect(await closing).toMatchObject({ ok: true });
+  });
+
+  it("does not start a queued event once the stream is DRAINING", async () => {
+    const firstApply = deferred<RealtimeResult<void>>();
+    let applyCalls = 0;
+    const harness = createHarness({
+      apply: async () => {
+        applyCalls += 1;
+        if (applyCalls === 1) return await firstApply.promise;
+        return realtimeSuccess(undefined);
+      },
+      taskLimits: { effectTimeoutMs: 15, drainTimeoutMs: 50 },
+    });
+    await harness.initialize();
+
+    const first = harness.coordinator.accept(
+      harness.event({ eventId: "event-0001", sequence: "1" }),
+    );
+    // Queued behind the first while it is still inside its deadline.
+    const second = harness.coordinator.accept(
+      harness.event({ eventId: "event-0002", sequence: "2" }),
+    );
+
+    await first;
+    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING");
+
+    // The queued event is dropped at execution time, not applied and not
+    // recovered: admission happened before DRAINING, execution happens inside
+    // it, and the second decision is the one that counts.
+    const secondResult = await second;
+    expect(secondResult).toMatchObject({
+      ok: true,
+      value: { outcome: "DROPPED", reason: "CLOSED" },
+    });
+    expect(applyCalls).toBe(1);
+
+    firstApply.resolve(realtimeSuccess(undefined));
+  });
+
+  it("discards the resume token when an effect is abandoned", async () => {
+    const firstApply = deferred<RealtimeResult<void>>();
+    let applyCalls = 0;
+    const harness = createHarness({
+      apply: async () => {
+        applyCalls += 1;
+        if (applyCalls === 1) return await firstApply.promise;
+        return realtimeSuccess(undefined);
+      },
+      taskLimits: { effectTimeoutMs: 15, drainTimeoutMs: 50 },
+    });
+    await harness.initialize();
+    expect(harness.coordinator.inspect(STREAM_ID).hasResumeState).toBe(true);
+
+    await harness.coordinator.accept(
+      harness.event({ eventId: "event-0001", sequence: "1" }),
+    );
+
+    // The abandoned effect may have applied part of its change, so the token it
+    // was based on is no longer authoritative evidence.
+    const inspection = harness.coordinator.inspect(STREAM_ID);
+    expect(inspection.hasResumeState).toBe(false);
+    expect(inspection.freshness).toBe("UNKNOWN");
+
+    firstApply.resolve(realtimeSuccess(undefined));
+  });
+});
+
 type HarnessOptions = Readonly<{
   registry?: RealtimePolicyRegistry;
   mappers?: typeof TEST_MAPPERS | Readonly<Record<string, typeof TEST_MAPPER>>;