fix: retain realtime work through draining

R-02: add an OPEN/DRAINING/CLOSED lifecycle orthogonal to freshness. Effect and
recovery authorities are now awaited under a deadline: on expiry the commit
capability is revoked and the work aborted, the caller gets a bounded
non-retryable IDLE_TIMEOUT, and the underlying task is retained rather than
dropped. A draining stream refuses new events and recovery, and close() returns
a Promise that succeeds only once every retained task actually settled,
reporting IDLE_TIMEOUT otherwise.

R-03: a handoff fail-close moves active, probe, quiescing and transition leases
into a retired-writer set before clearing their references, and close() waits on
current and retired writers together, so an abandoned non-cooperative writer can
no longer make teardown report a false success.

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

docs/architecture/decisions/VD-28-realtime-events-web-push-and-bounded-polling.md

@@ -11,6 +11,26 @@
   첫 제품 stream/Web Push를 선택할 때, 또는 backend replay/hosting/provider
   protocol이 바뀔 때
 
+
+## 스트림 lifecycle은 freshness와 직교한다 (R-02, R-03)
+
+`RealtimeStreamLifecycle = OPEN | DRAINING | CLOSED`는 freshness
+(`UNKNOWN/CURRENT/STALE/RESYNCING`)와 별개다.
+
+- effect/recovery deadline에 도달하면 commit capability를 즉시 영구 무효화하고
+  abort한다. caller에는 bounded `IDLE_TIMEOUT`(non-retryable, operation
+  `APPLY`/`RECOVER`)을 반환하되 **실제 task는 버리지 않고 retain**한다.
+- retain된 task가 하나라도 있으면 stream은 `DRAINING`이고 새 event/recovery
+  admission을 거절한다. 실제 settlement가 일어나야 `STALE`로 돌아가
+  authoritative recovery를 요구하거나, close 요청이면 `CLOSED`가 된다.
+- `close()`는 `Promise<RealtimeResult<void>>`다. 모든 retain task가 실제로
+  settle해야 success이고, drain bound를 넘기면 `IDLE_TIMEOUT/CLOSE`를 반환하며
+  stream은 계속 `DRAINING`이다. teardown success가 곧 quiescence다.
+- LIVE↔POLL overflow fail-close는 active/probe/quiescing/transition lease를
+  모두 abort한 뒤 **retired writer set**으로 옮기고 나서 reference를 지운다.
+  `close()`는 current와 retired를 dedupe해 함께 기다리므로, 버려진
+  non-cooperative writer가 아직 실행 중인데 close가 성공을 보고할 수 없다.
+
 ## 배경
 
 현재 optional recipe catalog는 realtime capability에

docs/architecture/realtime-events-web-push-and-bounded-polling.md

@@ -34,6 +34,16 @@ capability가 설치됐거나 production-ready라는 뜻이 아니다.
 [Frontend ports, adapters, and boundaries](./frontend-ports-adapters-and-boundaries.md)
 를 따른다.
 
+
+## Bounded task lease와 DRAINING (R-02, R-03)
+
+non-cooperative effect/recovery authority 하나가 stream tail 전체를 영구
+wedge하지 못하도록, common coordinator는 각 task를 deadline으로 감싼다. deadline
+초과 시 commit capability는 즉시 취소되지만 task 자체는 `retainedTasks`에 남아
+stream을 `DRAINING`으로 유지한다. `close()`는 이 retain 집합이 실제로 settle해야
+성공을 반환한다. handoff coordinator도 같은 원칙으로 fail-close된 writer를
+`retiredWriters`에 보존한다.
+
 ## 0. 현재 상태와 목표 delta
 
 이 문서에서 설계 승인, reference source 존재, production 조합과 target browser의

docs/operations/adapter-remediation-ledger.md

@@ -105,8 +105,8 @@ Rollout state starts at `NOT_STARTED`; documented-unimplemented items start at
 | ID | Activation | Red test command | Fix commit/PR | Rollout state | Rollback trigger | Evidence |
 | --- | --- | --- | --- | --- | --- | --- |
 | R-01 | Browser RPC `AVAILABLE_NOT_COMPOSED` | `corepack pnpm exec vitest run tests/unit/browser-rpc/browser-rpc-runtime.test.ts` | — | `NOT_STARTED` | stream lease deadlock | — |
-| R-02 | Realtime `NOT_SELECTED` | `corepack pnpm exec vitest run tests/unit/realtime/stream-coordinator.test.ts` | — | `NOT_STARTED` | stream stuck in `DRAINING` | — |
-| R-03 | Realtime `NOT_SELECTED` | `corepack pnpm exec vitest run tests/unit/realtime/live-poll-handoff-coordinator.test.ts` | — | `NOT_STARTED` | retired-writer set growth | — |
+| R-02 | Realtime `NOT_SELECTED` | `corepack pnpm exec vitest run tests/unit/realtime/stream-coordinator.test.ts` | `fix: retain realtime work through draining` | `FIXED_NOT_RELEASED` | stream stuck in `DRAINING` | Red never-settling effect and recovery → green 27/27; `close()` returns `IDLE_TIMEOUT` while a task is retained and success only after actual settlement |
+| R-03 | Realtime `NOT_SELECTED` | `corepack pnpm exec vitest run tests/unit/realtime/live-poll-handoff-coordinator.test.ts` | `fix: retain realtime work through draining` | `FIXED_NOT_RELEASED` | retired-writer set growth | Red overflow fail-close then `close()` → green 11/11; retired writers are waited on and only removed once actually quiesced |
 | R-04 | Browser RPC `AVAILABLE_NOT_COMPOSED` | `corepack pnpm exec vitest run tests/unit/browser-rpc/browser-rpc-contract.test.ts` | — | `NOT_STARTED` | binding install rejection | — |
 | R-05 | WebSocket protocol codec | `corepack pnpm exec vitest run tests/unit/realtime/websocket-protocol.test.ts` | — | `NOT_STARTED` | frame rejection regression | — |
 | R-06 | Browser RPC `AVAILABLE_NOT_COMPOSED` | `corepack pnpm exec vitest run tests/unit/browser-rpc/browser-rpc-runtime.test.ts` | — | `NOT_STARTED` | closed-failure taxonomy drift | — |

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

@@ -168,6 +168,11 @@ export function createLivePollHandoffCoordinator<Value>(
   let quiescing: InternalWriterLease<Value> | null = null;
   let transitionCandidate: InternalWriterLease<Value> | null = null;
   let closePromise: Promise<RealtimeResult<void>> | null = null;
+  /**
+   * R-03. Writers whose lease was fail-closed but whose tail may still be
+   * running. Membership keeps `close()` honest about quiescence.
+   */
+  const retiredWriters = new Set<InternalWriterLease<Value>>();
 
   active = createWriterLease(dependencies.initial.writer);
 
@@ -673,11 +678,15 @@ export function createLivePollHandoffCoordinator<Value>(
     lifecycleGeneration += 1;
     state = "CLOSED";
     transitioning = true;
+    // R-03. Current and previously retired writers are waited on together and
+    // deduplicated, so a writer dropped by an overflow fail-close is still
+    // proved quiescent before close reports success.
     const writers = uniqueLeases([
       active,
       probe?.lease ?? null,
       quiescing,
       transitionCandidate,
+      ...retiredWriters,
     ]);
     active = null;
     const selectedProbe = probe;
@@ -691,6 +700,11 @@ export function createLivePollHandoffCoordinator<Value>(
     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");
     }
@@ -771,20 +785,33 @@ export function createLivePollHandoffCoordinator<Value>(
     );
   }
 
+  /**
+   * R-03. A fail-close aborts every lease, but the underlying writers may be
+   * non-cooperative and still running. They move into the retired set before
+   * their references are cleared, so a later `close()` cannot report success
+   * while an abandoned writer is still executing.
+   */
   function failClosed(): void {
     lifecycleGeneration += 1;
     state = "CLOSED";
     transitioning = false;
-    active?.controller.abort();
-    probe?.lease.controller.abort();
-    quiescing?.controller.abort();
-    transitionCandidate?.controller.abort();
+    for (const writer of uniqueLeases([
+      active,
+      probe?.lease ?? null,
+      quiescing,
+      transitionCandidate,
+    ])) {
+      writer.controller.abort();
+      retiredWriters.add(writer);
+    }
     active = null;
     if (probe) {
       probe.buffer.length = 0;
       probe.bufferedBytes = 0;
     }
     probe = null;
+    quiescing = null;
+    transitionCandidate = null;
   }
 
   return Object.freeze({

src/adapters/realtime/stream-coordinator.ts

@@ -41,6 +41,27 @@ import {
   type RealtimeDataSnapshot,
 } from "./result.ts";
 
+/**
+ * R-02. Lifecycle is orthogonal to freshness. `DRAINING` means the coordinator
+ * has revoked commit capability and stopped admitting work, but a
+ * non-cooperative task it started is still running and is deliberately retained
+ * until it actually settles.
+ */
+export type RealtimeStreamLifecycle = "OPEN" | "DRAINING" | "CLOSED";
+
+export type RealtimeStreamTaskLimits = Readonly<{
+  effectTimeoutMs: number;
+  recoveryTimeoutMs: number;
+  drainTimeoutMs: number;
+}>;
+
+export const DEFAULT_REALTIME_STREAM_TASK_LIMITS: RealtimeStreamTaskLimits =
+  Object.freeze({
+    effectTimeoutMs: 5_000,
+    recoveryTimeoutMs: 10_000,
+    drainTimeoutMs: 5_000,
+  });
+
 export type RealtimeStreamCoordinatorDependencies = Readonly<{
   registry: RealtimePolicyRegistry;
   mappers: Readonly<Record<string, InstalledBoundaryMapper>>;
@@ -48,6 +69,10 @@ export type RealtimeStreamCoordinatorDependencies = Readonly<{
   scope: RealtimeScopeSnapshot;
   now?: () => number;
   observe?: RealtimeEventObservationSink;
+  taskLimits?: Partial<RealtimeStreamTaskLimits>;
+  /** Test seam for the bounded task deadline. */
+  scheduleTimeout?: (callback: () => void, delayMs: number) => unknown;
+  clearScheduledTimeout?: (handle: unknown) => void;
 }>;
 
 export type RealtimeStreamCoordinator = Readonly<{
@@ -66,7 +91,13 @@ export type RealtimeStreamCoordinator = Readonly<{
   ): RealtimeResult<void>;
   getResumeState(streamId: StreamRegistrationId): RealtimeResumeState | null;
   inspect(streamId: StreamRegistrationId): RealtimeStreamInspection;
-  close(): void;
+  lifecycle(streamId: StreamRegistrationId): RealtimeStreamLifecycle;
+  /**
+   * R-02. Bounded quiescence. Success means every tracked task actually
+   * settled; `IDLE_TIMEOUT` means the coordinator is still `DRAINING` and the
+   * caller must not assume a clean teardown.
+   */
+  close(): Promise<RealtimeResult<void>>;
 }>;
 
 type DedupeEntry = Readonly<{
@@ -96,6 +127,12 @@ type StreamState = {
   awaitingTransportBarrier: boolean;
   barrierCheckpoint: RealtimeRecoveryCheckpoint | null;
   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.
+   */
+  retainedTasks: Set<Promise<unknown>>;
 };
 
 const SNAPSHOT_CHECKPOINT_KEYS = Object.freeze([
@@ -120,9 +157,59 @@ export function createRealtimeStreamCoordinator(
 ): RealtimeStreamCoordinator {
   assertDependencies(dependencies);
   const now = dependencies.now ?? Date.now;
+  const limits: RealtimeStreamTaskLimits = Object.freeze({
+    ...DEFAULT_REALTIME_STREAM_TASK_LIMITS,
+    ...dependencies.taskLimits,
+  });
+  const scheduleTimeout =
+    dependencies.scheduleTimeout ??
+    ((callback: () => void, delayMs: number) => setTimeout(callback, delayMs));
+  const clearScheduledTimeout =
+    dependencies.clearScheduledTimeout ??
+    ((handle: unknown) => {
+      clearTimeout(handle as ReturnType<typeof setTimeout>);
+    });
   const states = new Map<StreamRegistrationId, StreamState>();
   let closed = false;
 
+  const TASK_TIMED_OUT = Symbol("REALTIME_TASK_TIMED_OUT");
+
+  /**
+   * R-02. Bounds the public wait without discarding the task. A task that
+   * outlives its deadline is retained so `close()` can report honestly whether
+   * the stream is actually quiescent.
+   */
+  async function awaitTaskWithinDeadline<Value>(
+    state: StreamState,
+    task: Promise<Value>,
+    timeoutMs: number,
+    revokeAndAbort: () => void,
+  ): Promise<Value | typeof TASK_TIMED_OUT> {
+    task.catch(() => {});
+    let handle: unknown;
+    const timeout = new Promise<typeof TASK_TIMED_OUT>((resolve) => {
+      handle = scheduleTimeout(() => resolve(TASK_TIMED_OUT), timeoutMs);
+    });
+    const outcome = await Promise.race([task, timeout]);
+    clearScheduledTimeout(handle);
+    if (outcome !== TASK_TIMED_OUT) return outcome;
+
+    // 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";
+        }
+      });
+    return TASK_TIMED_OUT;
+  }
+
   for (const registration of dependencies.registry.listStreams()) {
     states.set(registration.id, {
       registration,
@@ -143,6 +230,8 @@ export function createRealtimeStreamCoordinator(
       awaitingTransportBarrier: false,
       barrierCheckpoint: null,
       closed: false,
+      lifecycle: "OPEN",
+      retainedTasks: new Set(),
     });
   }
 
@@ -158,6 +247,11 @@ export function createRealtimeStreamCoordinator(
         realtimeFailure("MALFORMED_EVENT", "RECEIVE"),
       );
     }
+    const draining = states.get(event.envelope.streamId);
+    if (draining && draining.lifecycle === "DRAINING") {
+      // R-02. New work is refused while a retained task is still running.
+      return Promise.resolve(realtimeFailure("CLOSED", "RECEIVE"));
+    }
     const state = states.get(event.envelope.streamId);
     if (!state) {
       return Promise.resolve(
@@ -384,19 +478,37 @@ export function createRealtimeStreamCoordinator(
       !effectAbort.signal.aborted &&
       scopeIsCurrent(dependencies.scope);
     let effect: unknown;
+    let effectTimedOut = false;
     try {
-      effect = await dependencies.authority.effects.apply(
-        eventType.effectProfileId,
-        mapped.value,
-        Object.freeze({
-          streamId: state.registration.id,
-          eventType: eventType.id,
-          occurredAt: event.envelope.occurredAt,
-          scopeGeneration: dependencies.scope.generation,
-          isCurrent: effectIsCurrent,
-        }),
-        effectAbort.signal,
+      const applied = await awaitTaskWithinDeadline(
+        state,
+        Promise.resolve(
+          dependencies.authority.effects.apply(
+            eventType.effectProfileId,
+            mapped.value,
+            Object.freeze({
+              streamId: state.registration.id,
+              eventType: eventType.id,
+              occurredAt: event.envelope.occurredAt,
+              scopeGeneration: dependencies.scope.generation,
+              isCurrent: effectIsCurrent,
+            }),
+            effectAbort.signal,
+          ),
+        ),
+        limits.effectTimeoutMs,
+        () => {
+          // Commit capability is revoked permanently for this attempt.
+          effectLeaseActive = false;
+          effectAbort.abort();
+        },
       );
+      if (applied === TASK_TIMED_OUT) {
+        effectTimedOut = true;
+        effect = null;
+      } else {
+        effect = applied;
+      }
     } catch {
       effect = null;
     } finally {
@@ -406,6 +518,17 @@ export function createRealtimeStreamCoordinator(
         state.activeEffectAbort = null;
       }
     }
+    if (effectTimedOut) {
+      // R-02. Bounded for the caller; the underlying task stays tracked.
+      observe({
+        operation: "APPLY",
+        outcome: "FAILED",
+        streamId: state.registration.id,
+        eventType: event.envelope.eventType,
+        reason: "IDLE_TIMEOUT",
+      });
+      return realtimeFailure("IDLE_TIMEOUT", "APPLY", false);
+    }
 
     if (closed || state.closed) {
       return dropped("CLOSED");
@@ -483,7 +606,7 @@ export function createRealtimeStreamCoordinator(
     calledFromCurrentJob: boolean,
     signal?: AbortSignal,
   ): Promise<RealtimeResult<RealtimeRecoveryCheckpoint>> {
-    if (closed || state.closed) {
+    if (closed || state.closed || state.lifecycle === "DRAINING") {
       return Promise.resolve(realtimeFailure("CLOSED", "RECOVER"));
     }
     if (!scopeIsCurrent(dependencies.scope)) {
@@ -542,22 +665,50 @@ export function createRealtimeStreamCoordinator(
 
       let recovered: unknown;
       let recoveryThrew = false;
+      let recoveryTimedOut = false;
       try {
-        recovered = await dependencies.authority.recovery.recover(
-          Object.freeze({
-            streamId: state.registration.id,
-            reason,
-            scopeGeneration: dependencies.scope.generation,
-            signal: recoveryAbort.signal,
-            isCurrent: recoveryIsCurrent,
-          }),
+        const outcome = await awaitTaskWithinDeadline(
+          state,
+          Promise.resolve(
+            dependencies.authority.recovery.recover(
+              Object.freeze({
+                streamId: state.registration.id,
+                reason,
+                scopeGeneration: dependencies.scope.generation,
+                signal: recoveryAbort.signal,
+                isCurrent: recoveryIsCurrent,
+              }),
+            ),
+          ),
+          limits.recoveryTimeoutMs,
+          () => {
+            recoveryLeaseActive = false;
+            recoveryAbort.abort();
+          },
         );
+        if (outcome === TASK_TIMED_OUT) {
+          recoveryTimedOut = true;
+          recovered = null;
+        } else {
+          recovered = outcome;
+        }
       } catch {
         recovered = null;
         recoveryThrew = true;
       } finally {
         recoveryLeaseActive = false;
       }
+      if (recoveryTimedOut) {
+        // R-02. A late checkpoint from this attempt can never commit.
+        state.freshness = "UNKNOWN";
+        observe({
+          operation: "RECOVER",
+          outcome: "FAILED",
+          streamId: state.registration.id,
+          reason: "IDLE_TIMEOUT",
+        });
+        return realtimeFailure("IDLE_TIMEOUT", "RECOVER", false);
+      }
       if (closed || state.closed) {
         state.freshness = "UNKNOWN";
         return realtimeFailure("CLOSED", "RECOVER");
@@ -806,7 +957,45 @@ export function createRealtimeStreamCoordinator(
     });
   }
 
-  function close(): void {
+  function lifecycleOf(streamId: StreamRegistrationId): RealtimeStreamLifecycle {
+    const state = states.get(streamId);
+    if (!state) return "CLOSED";
+    return state.lifecycle;
+  }
+
+  /**
+   * R-02. `close()` fences immediately but reports honestly: success only when
+   * every retained task actually settled within the drain bound.
+   */
+  async function close(): Promise<RealtimeResult<void>> {
+    fenceAllStates();
+    const retained = [...states.values()].flatMap((state) => [
+      ...state.retainedTasks,
+    ]);
+    if (retained.length === 0) {
+      for (const state of states.values()) state.lifecycle = "CLOSED";
+      return realtimeSuccess(undefined);
+    }
+    let handle: unknown;
+    const drained = await Promise.race([
+      Promise.allSettled(retained).then(() => true),
+      new Promise<false>((resolve) => {
+        handle = scheduleTimeout(
+          () => resolve(false),
+          limits.drainTimeoutMs,
+        );
+      }),
+    ]);
+    clearScheduledTimeout(handle);
+    if (!drained) {
+      // Still DRAINING: the caller must not treat this as quiescence.
+      return realtimeFailure("IDLE_TIMEOUT", "CLOSE", false);
+    }
+    for (const state of states.values()) state.lifecycle = "CLOSED";
+    return realtimeSuccess(undefined);
+  }
+
+  function fenceAllStates(): void {
     if (closed) return;
     closed = true;
     for (const state of states.values()) {
@@ -826,6 +1015,8 @@ export function createRealtimeStreamCoordinator(
       state.eventIds.clear();
       state.sequences.clear();
       state.dedupeBytes = 0;
+      state.lifecycle =
+        state.retainedTasks.size > 0 ? "DRAINING" : "CLOSED";
     }
     observe({
       operation: "CLOSE",
@@ -911,6 +1102,7 @@ export function createRealtimeStreamCoordinator(
     confirmTransportBarrier,
     getResumeState,
     inspect,
+    lifecycle: lifecycleOf,
     close,
   });
 }

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

@@ -214,6 +214,47 @@ describe("live/poll authoritative writer handoff", () => {
     void second;
   });
 
+  it("tracks a retired active writer after handoff queue overflow", async () => {
+    let release: ((result: RealtimeResult<void>) => void) | undefined;
+    const harness = createHarness({
+      limits: {
+        ...limits,
+        maxActiveQueueCount: 2,
+        maxActiveQueueBytes: 10,
+      },
+      apply: vi.fn(async ({ value }) => {
+        if (value === "first") {
+          return await new Promise<RealtimeResult<void>>((resolve) => {
+            release = resolve;
+          });
+        }
+        return realtimeSuccess(undefined);
+      }),
+    });
+    const writer = harness.coordinator.currentWriter()!;
+    const first = writer.write("first", 5);
+    const second = writer.write("second", 5);
+    await flush();
+    await expect(writer.write("overflow", 1)).resolves.toMatchObject({
+      ok: false,
+      error: { kind: "QUEUE_OVERFLOW" },
+    });
+
+    // R-03. The fail-close dropped the active reference, but the writer is
+    // still running, so close() must not claim quiescence.
+    const closing = harness.coordinator.close();
+    await flush();
+    harness.clock.advance(100);
+    await expect(closing).resolves.toMatchObject({
+      ok: false,
+      error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" },
+    });
+
+    release?.(realtimeSuccess(undefined));
+    void first;
+    void second;
+  });
+
   it("fences and aborts live, waits for quiescence, then recovers before activating poll", async () => {
     const liveEffect = deferred<RealtimeResult<void>>();
     const checkpoint = deferred<RealtimeResult<void>>();

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

@@ -29,6 +29,91 @@ import {
 } from "./fixture.ts";
 
 describe("transport-independent realtime stream coordinator", () => {
+  it("keeps the stream DRAINING until a non-cooperative effect settles", async () => {
+    const wedged = deferred<RealtimeResult<void>>();
+    const harness = createHarness({
+      apply: async () => wedged.promise,
+      taskLimits: { effectTimeoutMs: 5, drainTimeoutMs: 5 },
+    });
+    await harness.initialize();
+
+    const applied = await harness.coordinator.accept(
+      harness.event({
+        eventId: "event-00000001",
+        sequence: "1",
+        resumeCursor: "cursor-00000001",
+      }),
+    );
+    // Bounded for the caller, and explicitly non-retryable.
+    expect(applied).toMatchObject({
+      ok: false,
+      error: { kind: "IDLE_TIMEOUT", operation: "APPLY", retryable: false },
+    });
+    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING");
+
+    // DRAINING refuses new admission rather than queueing behind the wedge.
+    await expect(
+      harness.coordinator.accept(
+        harness.event({
+          eventId: "event-00000002",
+          sequence: "2",
+          resumeCursor: "cursor-00000002",
+        }),
+      ),
+    ).resolves.toMatchObject({ ok: false, error: { kind: "CLOSED" } });
+
+    // close() cannot claim quiescence while the task is still running.
+    await expect(harness.coordinator.close()).resolves.toMatchObject({
+      ok: false,
+      error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" },
+    });
+
+    // Only actual settlement ends DRAINING.
+    wedged.resolve(realtimeSuccess(undefined));
+    await expect(harness.coordinator.close()).resolves.toMatchObject({
+      ok: true,
+    });
+    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("CLOSED");
+  });
+
+  it("bounds non-cooperative recovery and rejects its late checkpoint", async () => {
+    const wedged = deferred<RealtimeResult<RealtimeRecoveryCommit>>();
+    let recoveries = 0;
+    const harness = createHarness({
+      recover: async () => {
+        recoveries += 1;
+        return recoveries === 1
+          ? realtimeSuccess(snapshotCommit("0"))
+          : wedged.promise;
+      },
+      taskLimits: { recoveryTimeoutMs: 5, drainTimeoutMs: 5 },
+    });
+    await harness.initialize();
+
+    const recovered = await harness.coordinator.recover(
+      STREAM_ID,
+      "SEQUENCE_GAP",
+    );
+    expect(recovered).toMatchObject({
+      ok: false,
+      error: { kind: "IDLE_TIMEOUT", operation: "RECOVER", retryable: false },
+    });
+    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING");
+
+    const beforeLateCommit = harness.coordinator.getResumeState(STREAM_ID);
+
+    // A late checkpoint from the abandoned attempt cannot commit.
+    wedged.resolve(realtimeSuccess(snapshotCommit("9")));
+    for (let flush = 0; flush < 10; flush += 1) await Promise.resolve();
+    expect(harness.coordinator.getResumeState(STREAM_ID)).toEqual(
+      beforeLateCommit,
+    );
+    // Settlement returns the stream to OPEN, marked STALE for an authoritative
+    // recovery rather than silently trusting the abandoned attempt.
+    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("OPEN");
+    expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe("STALE");
+  });
+
   it("applies one stream sequentially and commits each cursor after its effect", async () => {
     const first = deferred<RealtimeResult<void>>();
     const applied: string[] = [];
@@ -866,6 +951,11 @@ type HarnessOptions = Readonly<{
     request: RealtimeRecoveryRequest,
   ) => Promise<RealtimeResult<RealtimeRecoveryCommit>>;
   observe?: (observation: RealtimeObservation) => void;
+  taskLimits?: Readonly<{
+    effectTimeoutMs?: number;
+    recoveryTimeoutMs?: number;
+    drainTimeoutMs?: number;
+  }>;
 }>;
 
 function createHarness(options: HarnessOptions = {}) {
@@ -911,6 +1001,7 @@ function createHarness(options: HarnessOptions = {}) {
     },
     now: () => 10_000,
     observe: options.observe,
+    ...(options.taskLimits ? { taskLimits: options.taskLimits } : {}),
   });
 
   return {