diff --git a/.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.i18n.yaml b/.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.i18n.yaml index 117e2b7202..4c0f54d546 100644 --- a/.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.i18n.yaml +++ b/.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write -2026-07-23-collapse-persistence-flush-state.md: 69403fe3c2ee556cb10593fd43857e0d844242df -2026-07-23-collapse-persistence-flush-state.zh.md: 0b38ee26b9e6273672cbc218826c0dc399158a71 +2026-07-23-collapse-persistence-flush-state.md: 9a2de00b2ad0c2417b6cdcba9cbc4f020b93037d +2026-07-23-collapse-persistence-flush-state.zh.md: f9152989fef79efbe0afc27d256d8d3418ce84ba diff --git a/.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.md b/.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.md index 69403fe3c2..9a2de00b2a 100644 --- a/.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.md +++ b/.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.md @@ -16,6 +16,8 @@ Each live `Session` has one controller containing `pending`, `init`, and the opt Initialization now enters the existing per-id operation chain once and calls the unserialized core operations while it owns that turn. The chain remains separate from the live controller because detached public `create`/`append`/`load` calls can race without a `Session` object and still require identity-level serialization. +Crash repair is cold-only. For a live identity, `load(id)` snapshots the authoritative in-memory header and events before awaiting their flush; it returns that durable snapshot when balanced and rejects an open turn without reading or repairing storage. A cold identity follows the stored-prefix repair path. HMR adoption remains separate through `loadLive` and truncates torn storage without closing the authoritative live turn. + The live-controller map is also the retirement registry. Successful retirement drains and removes its controller; failed retirement leaves it in the map. Backend teardown stops event admission, flushes every controller still present, awaits remaining per-id operations, and closes the backend. No separate retirement set is needed to rediscover unfinished work. ## Alternatives considered @@ -26,11 +28,15 @@ The live-controller map is also the retirement registry. Successful retirement d **Latch the first eager error permanently.** This makes every later flush deterministic, but prevents the existing teardown retry from recovering a transient storage failure. Retaining the batch without latching the error preserves both observability and retry. +**Reject every live load.** This is safe but removes established balanced live snapshots used by persistence consumers and tests. Snapshot-before-flush gives the call a stable linearization point: successful flush proves exactly that snapshot is durable, while the live path never invokes crash repair. + ## Verification - A focused coordinator test gates the first append, admits another event during that write, and observes an automatic second durable batch without calling `session/flush`. - The shared coordinator contract still covers live adoption, collisions, crash repair, and session/backend disposal over the in-memory, JSONL, and SQLite backends. - Failure and teardown tests keep rejected batches pending, retry them before close, and prove an in-flight controller delays backend close. +- The shared backend contract persists an open live turn, proves `load` rejects without writing synthetic closers, completes and retires the owner, then reloads the exact completed turn. +- An AgentLoop regression races `resume()` against a live open turn and proves the original agent can still durably complete it without an injected `interrupted` boundary. ## Consequences diff --git a/.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.zh.md b/.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.zh.md index 0b38ee26b9..f9152989fe 100644 --- a/.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.zh.md +++ b/.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.zh.md @@ -16,6 +16,8 @@ Status: implemented 初始化只进入现有的按 id 操作链一次,并在占有该轮执行权时调用未串行化的核心操作。该操作链与活跃控制器保持分离,因为公共 `create`、`append`、`load` 调用即使没有 `Session` 对象仍可能发生竞态,依然需要按标识串行执行。 +崩溃修复仅适用于冷态标识。对于活跃标识,`load(id)` 会在等待刷新完成前,先对内存中的权威标头和事件生成快照;若快照闭合,则返回这个已持久化的快照;若轮次仍打开,则在不读取或修复存储的情况下拒绝该次加载。冷态标识沿用已存储前缀的修复路径。HMR 接管仍通过 `loadLive` 独立处理,会截断撕裂的存储,但不会闭合权威的活跃轮次。 + 活跃控制器映射同时也是退役注册表。退役成功时,系统排空并移除其控制器;退役失败时,控制器保留在映射中。后端资源销毁会停止接纳事件,刷新所有仍存在的控制器,等待其余按 id 操作完成,然后关闭后端。无需另设退役集合来重新发现未完成的工作。 ## 备选方案 @@ -26,11 +28,15 @@ Status: implemented **永久锁存首次即时写入错误。** 这会让后续每次刷新都得到确定的结果,却会阻止现有的资源销毁重试从暂时性存储故障中恢复。保留批次但不锁存错误,可以同时保留可观测性和重试能力。 +**拒绝对所有活跃会话的加载。** 这样做很安全,但会让持久化消费方和测试无法再使用既有的闭合活跃会话快照。先生成快照再刷新,为调用提供了稳定的线性化点:刷新成功即可证明正是该快照已持久化,而活跃路径绝不调用崩溃修复。 + ## 验证 - 一个针对协调器的测试会阻塞第一次追加,在该次写入期间接纳另一个事件,并在不调用 `session/flush` 的情况下观测到自动执行的第二个持久批次。 - 共享协调器契约仍覆盖内存、JSONL 和 SQLite 后端上的活跃会话接管、冲突、崩溃修复,以及会话和后端的资源释放。 - 失败和资源销毁测试会让写入失败的批次保持待处理,在关闭前重试这些批次,并证明尚在执行的控制器会延迟后端关闭。 +- 共享后端契约会持久化一个仍打开的活跃轮次,证明 `load` 会拒绝且不会写入合成闭合事件,随后完成该轮次并让其所有者退役,最后重新加载完全相同的已完成轮次。 +- AgentLoop 回归测试让 `resume()` 与一个仍打开的活跃轮次发生竞态,并证明原有的 agent(智能体)仍能完成该轮次并将其持久化,其间不会注入 `interrupted` 边界。 ## 后果 diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index 67b4ecbd9e..0e3e773006 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -917,7 +917,9 @@ abstract append(id: SessionId, events: readonly SessionEvent[]): Promise * Load a header and balanced contiguous log. A complete interrupted final * turn is preserved and durably closed with missing tool errors plus any open * step and turn boundaries; only a torn final record is discarded. Unknown - * versions and corruption in the committed prefix reject. + * versions and corruption in the committed prefix reject. Implementations + * MUST NOT crash-repair an identity still bound to a live Session: a balanced + * live log may return as a durable snapshot, while an open live turn rejects. * @param id - the persisted session to reload. * @returns the header and a log ending on a balanced `turn/end`. */ diff --git a/docs/core-data-structures/persistence.md b/docs/core-data-structures/persistence.md index 9327c2e4dd..03350623f8 100644 --- a/docs/core-data-structures/persistence.md +++ b/docs/core-data-structures/persistence.md @@ -12,6 +12,8 @@ The seam is a textbook [capability seam](../../.agents/notes/implemented/archite A backend that reloads a log crashed mid-turn finds an open `turn/start` with no `turn/end`. It does **not** truncate — a single turn can be huge in a long-horizon task (many steps, large tool output), and those events were durably appended before the crash. Instead it closes the orphaned turn with a synthetic `turn/end { reason: { kind: 'interrupted' } }`, keeping the log balanced and the turn-enclosure invariant intact. `interrupted` is the one `TurnEndReason` no loop emits (see [session.md](session.md#why-a-turn-ended-turnendreasonmap)). +Repair applies only to cold sessions. For a live id, `SessionPersistence.load(id)` snapshots the in-memory log, waits until that snapshot is durable, and returns it only when balanced; an open live turn rejects rather than receiving synthetic interruption boundaries. HMR also adopts a live prefix without closing its active turn. + ## `SessionLocation` — optional per-session artifact target `SessionPersistence.locate(meta)` synchronously resolves a backend-owned independent artifact without reading, creating, or flushing it. JSONL returns its absolute target path; SQLite returns `undefined` because sessions share one database. A returned path can therefore name a file that does not yet exist or lacks the current unflushed turn; it is a location hint, not authorization or a freshness guarantee. diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index 8c39e81b7d..c04fcb7891 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -462,7 +462,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ }, { signature: 'abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }>', - jsDoc: '/**\n * Load a header and balanced contiguous log. A complete interrupted final\n * turn is preserved and durably closed with missing tool errors plus any open\n * step and turn boundaries; only a torn final record is discarded. Unknown\n * versions and corruption in the committed prefix reject.\n * @param id - the persisted session to reload.\n * @returns the header and a log ending on a balanced `turn/end`.\n */', + jsDoc: '/**\n * Load a header and balanced contiguous log. A complete interrupted final\n * turn is preserved and durably closed with missing tool errors plus any open\n * step and turn boundaries; only a torn final record is discarded. Unknown\n * versions and corruption in the committed prefix reject. Implementations\n * MUST NOT crash-repair an identity still bound to a live Session: a balanced\n * live log may return as a durable snapshot, while an open live turn rejects.\n * @param id - the persisted session to reload.\n * @returns the header and a log ending on a balanced `turn/end`.\n */', }, { signature: 'abstract list(): Promise', diff --git a/packages/core/agent-loop/tests/resume.spec.ts b/packages/core/agent-loop/tests/resume.spec.ts index 5c64bb92f9..fdf5c39514 100644 --- a/packages/core/agent-loop/tests/resume.spec.ts +++ b/packages/core/agent-loop/tests/resume.spec.ts @@ -111,6 +111,27 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume', await ctx.fiber.dispose() }) + it('resume cannot crash-repair a turn owned by a live agent', async () => { + const { ctx } = await persistentHarness(new MockAdapter([textResponse('unused')])) + const sessionId = SessionId('live-resume-race') + const first = (await ctx.agents.create({ sessionId })).agent + first.session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + await ctx.sessions.flush(first.session) + + await expect(ctx.agents.resume({ resumeSessionId: sessionId })) + .rejects.toThrow(/live turn is open/) + + first.session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + await ctx.sessions.flush(first.session) + const loaded = await ctx.sessionPersistence.load(sessionId) + expect(loaded.events.map(event => event.type)).toEqual(['turn/start', 'turn/end']) + expect(loaded.events.at(-1)).toMatchObject({ + type: 'turn/end', + data: { reason: { kind: 'completed' } }, + }) + await ctx.fiber.dispose() + }) + it('createAgent works without meta (no cwd)', async () => { const adapter = new MockAdapter([textResponse('hi')]) const { ctx } = await persistentHarness(adapter) diff --git a/packages/session-persistence/session-persistence/README.md b/packages/session-persistence/session-persistence/README.md index 32097a7108..a09dbcb426 100644 --- a/packages/session-persistence/session-persistence/README.md +++ b/packages/session-persistence/session-persistence/README.md @@ -11,7 +11,7 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l | `locate(meta): SessionLocation \| undefined` | Resolve an absolute per-session artifact target without I/O or materialization. Backends without an independent local artifact return `undefined`. | | `create(meta): Promise` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). | | `append(id, events): Promise` | Durably persist a batch. Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. | -| `load(id): Promise<{ meta; events }>` | Reload meta + log. Preserves an interrupted (unclosed) final turn and closes it with synthetic closers — an error `tool/result` per unanswered `tool-call`, then `step/end?`+`turn/end {interrupted}` (a turn can be huge — never truncated); only a torn tail fragment is dropped. Events contiguous (`events[i].seq === i`); rejects a committed-region gap/parse error or unknown `version`. | +| `load(id): Promise<{ meta; events }>` | Return a flushed balanced snapshot for a live session, rejecting while its turn is open; cold load preserves an interrupted final turn and closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption and unknown `version` reject. | | `list(): Promise` | Lightweight listing from metadata, no full-log parse. A zero-event lazily-materialized session is absent from `list`. | ## Invariants every backend must honor @@ -27,6 +27,8 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l Each `session/event` copies its event into the session controller and starts an eager drain without blocking the producer. Concurrent notifications share the current drain; events admitted during a write remain pending and trigger the next batch. `session/flush` is an observation barrier that waits until the controller has no current or pending batch. An eager failure is logged and retains the batch; the next explicit flush or backend teardown retries it and surfaces failure to its caller. +Crash repair is cold-only. For a live id, `load(id)` snapshots the authoritative in-memory log, waits for that snapshot to become durable, and returns it only when balanced; an open live turn rejects instead of receiving synthetic interruption closers. A cold id follows storage repair normally. HMR adoption likewise uses the separate `loadLive` hook and never closes the active turn. + When a live session emits `session/disposed`, the coordinator waits for its controller, serializes a final drain, then releases state owned by that exact `Session` object. Failed retirement leaves the controller in the live-session map, so backend teardown can retry it. Backend teardown stops event admission first, flushes every remaining controller, awaits per-id operations, and only then closes the storage handle. The side-effect-free `locate` query remains backend-owned because it describes storage topology rather than write orchestration. diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index e1b604e807..602ac43f72 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -256,6 +256,8 @@ export class PersistenceCoordinator { * @returns the header plus the event log, ending on a balanced `turn/end`. */ load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + const live = this.ctx.sessions.get(id) + if (live !== undefined) return this.loadLiveSnapshot(live) return this.serialize(id, () => this.loadCore(id)) } @@ -279,6 +281,18 @@ export class PersistenceCoordinator { return { meta, events: balanced } } + /** Return a durable balanced live snapshot without applying cold crash repair. */ + private async loadLiveSnapshot(session: Session): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + const meta = structuredClone(session.header) + const events = session.events.map(event => structuredClone(event)) + await this.flush(session) + if (events.length === 0) throw new Error(`session "${session.id}" not found`) + if (interruptedTurnClosers(events).length > 0) { + throw new Error(`cannot load session "${session.id}" while its live turn is open; use the live Session or wait for the turn to close`) + } + return { meta, events } + } + // Listing is a direct backend read and needs no coordinator state. // --- per-id serialization + adoption helpers --- diff --git a/packages/session-persistence/session-persistence/src/index.ts b/packages/session-persistence/session-persistence/src/index.ts index eec0292a10..8490b133bd 100644 --- a/packages/session-persistence/session-persistence/src/index.ts +++ b/packages/session-persistence/session-persistence/src/index.ts @@ -74,9 +74,11 @@ export abstract class SessionPersistence extends Service { /** * Load a header and balanced contiguous log. A complete interrupted final - * turn is preserved and durably closed with missing tool errors plus any open - * step and turn boundaries; only a torn final record is discarded. Unknown - * versions and corruption in the committed prefix reject. + * turn is preserved and durably closed with missing tool errors plus any open + * step and turn boundaries; only a torn final record is discarded. Unknown + * versions and corruption in the committed prefix reject. Implementations + * MUST NOT crash-repair an identity still bound to a live Session: a balanced + * live log may return as a durable snapshot, while an open live turn rejects. * @param id - the persisted session to reload. * @returns the header and a log ending on a balanced `turn/end`. */ diff --git a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts index 620d069d32..a4d6b16072 100644 --- a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts +++ b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts @@ -88,6 +88,51 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< } }) + it('rejects crash-repair load while a live session owns the persisted prefix', async () => { + const fix = await makeFixture() + const { ctx, fiber } = await freshCtx(fix) + let session!: Session + const sessionFiber = await ctx.plugin(Object.assign((inner: Context) => { + session = inner.sessions.create(SessionId('live-load'), { meta: { cwd: WORK } }) + }, { inject: ['sessions'] })) + try { + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + await ctx.sessions.flush(session) + + await expect(ctx.sessionPersistence.load(session.id)) + .rejects.toThrow(`cannot load session "${session.id}" while its live turn is open`) + + send(session, oneTurnLog().slice(1)) + await ctx.sessions.flush(session) + await sessionFiber.dispose() + + await vi.waitFor(async () => { + const loaded = await ctx.sessionPersistence.load(session.id) + expect(loaded.events.map(event => event.type)).toEqual(oneTurnLog().map(event => event.type)) + expect(loaded.events.at(-1)).toMatchObject({ + type: 'turn/end', + data: { reason: { kind: 'completed' } }, + }) + }) + } finally { + await sessionFiber.dispose() + await fiber.dispose() + await fix.cleanup() + } + }) + + it('does not load an unmaterialized empty live session', async () => { + const fix = await makeFixture() + const { ctx, fiber } = await freshCtx(fix) + try { + const session = ctx.sessions.create(SessionId('empty-live'), { meta: { cwd: WORK } }) + await expect(ctx.sessionPersistence.load(session.id)).rejects.toThrow(/not found/) + } finally { + await fiber.dispose() + await fix.cleanup() + } + }) + it('round-trips the seed boundary (seedLength) through persistence', async () => { // A forked child records how many leading events were inherited via the seed; the // boundary must survive a reload (so a resume/replay can tell the inherited prefix from diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 4b0a93134f..7c15d52a4a 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -245,7 +245,6 @@ describe('PersistenceCoordinator retirement', () => { const backendFiber = await ctx.plugin(Object.assign((inner: Context) => { coordinator = new PersistenceCoordinator(inner, backend) }, { inject: ['sessions'] })) - const loadGate = Promise.withResolvers() try { const id = SessionId('retiring-lazy-owner') @@ -254,52 +253,42 @@ describe('PersistenceCoordinator retirement', () => { first = inner.sessions.create(id) }, { inject: ['sessions'] })) await ctx.sessions.flush(first) - - const baselineLoads = backend.loadAttempts - backend.beforeLoadStored = async () => { await loadGate.promise } - const blockingLoad = coordinator.load(id) - await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 1) }) await firstFiber.dispose() + const internals = coordinator as unknown as CoordinatorInternals + await vi.waitFor(() => { expect(internals.states.has(id)).toBe(false) }) let reuse!: Session await ctx.plugin(Object.assign((inner: Context) => { reuse = inner.sessions.create(id) }, { inject: ['sessions'] })) - loadGate.resolve(true) - await expect(blockingLoad).rejects.toThrow(/not found/) await expect(ctx.sessions.flush(reuse)).resolves.toBeUndefined() } finally { - loadGate.resolve(true) await backendFiber.dispose() await ctx.fiber.dispose() } }) - it('a retiring owner with buffered events still rejects same-id reuse', async () => { + it('a replacement queued before retirement cleanup still collides with the live owner', async () => { const ctx = new Context() await ctx.plugin(SessionStore) const backend = new ControlledBackend() - let coordinator!: PersistenceCoordinator const backendFiber = await ctx.plugin(Object.assign((inner: Context) => { - coordinator = new PersistenceCoordinator(inner, backend) + new PersistenceCoordinator(inner, backend) }, { inject: ['sessions'] })) - const loadGate = Promise.withResolvers() + const appendGate = Promise.withResolvers() try { - const id = SessionId('retiring-buffered-owner') + const id = SessionId('retiring-live-owner') let first!: Session const firstFiber = await ctx.plugin(Object.assign((inner: Context) => { first = inner.sessions.create(id) }, { inject: ['sessions'] })) await ctx.sessions.flush(first) + backend.beforeAppend = async () => { await appendGate.promise } first.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) first.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) - - const baselineLoads = backend.loadAttempts - backend.beforeLoadStored = async () => { await loadGate.promise } - const blockingLoad = coordinator.load(id) - await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 1) }) + await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) }) await firstFiber.dispose() let reuse!: Session @@ -308,14 +297,56 @@ describe('PersistenceCoordinator retirement', () => { }, { inject: ['sessions'] })) const reuseFlush = ctx.sessions.flush(reuse) - loadGate.resolve(true) - await expect(blockingLoad).rejects.toThrow(/not found/) + appendGate.resolve(true) + await expect(reuseFlush).rejects.toThrow(/bound to a different live session/) + expect(backend.store.get(id)?.events.map(event => event.seq)).toEqual([0, 1]) + } finally { + appendGate.resolve(true) + await backendFiber.dispose() + await ctx.fiber.dispose() + } + }) + + it('a racing cold load survives retirement cleanup and rejects same-id reuse', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + let coordinator!: PersistenceCoordinator + const backendFiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + const appendGate = Promise.withResolvers() + + try { + const id = SessionId('retiring-buffered-owner') + let first!: Session + const firstFiber = await ctx.plugin(Object.assign((inner: Context) => { + first = inner.sessions.create(id) + }, { inject: ['sessions'] })) + await ctx.sessions.flush(first) + backend.beforeAppend = async () => { await appendGate.promise } + first.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + first.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) }) + await firstFiber.dispose() + const coldLoad = coordinator.load(id) + + let reuse!: Session + await ctx.plugin(Object.assign((inner: Context) => { + reuse = inner.sessions.create(id) + }, { inject: ['sessions'] })) + const reuseFlush = ctx.sessions.flush(reuse) + + appendGate.resolve(true) + await expect(coldLoad).resolves.toMatchObject({ + events: [{ seq: 0 }, { seq: 1 }], + }) await expect(reuseFlush).rejects.toThrow(/id collision/) await vi.waitFor(() => { expect(backend.store.get(id)?.events.map(event => event.seq)).toEqual([0, 1]) }) } finally { - loadGate.resolve(true) + appendGate.resolve(true) await backendFiber.dispose() await ctx.fiber.dispose() }