diff --git a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.i18n.yaml b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.i18n.yaml index 2e5125fa4c..09e63a6a1c 100644 --- a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.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 .agents/notes/implemented/architecture/2026-08-05-session-preparation.md -2026-08-05-session-preparation.md: d6b6cc190779279d1cc008bca890f3818736e920 -2026-08-05-session-preparation.zh.md: 191ad84db0e7bba38149c90cd91b410187f08f20 +2026-08-05-session-preparation.md: 69d39f552ed3041403a24b5aefb435e4e721b09c +2026-08-05-session-preparation.zh.md: a0ca27eb63552566c918c299bd5fba976687812c diff --git a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.md b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.md index d6b6cc1907..69d39f552e 100644 --- a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.md +++ b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.md @@ -22,9 +22,9 @@ This refines the publication boundary from the [Agent lifecycle and ownership de A coordinator-backed persistence implementation loads one cold source into a prepared Session. The backend transfers fresh, mutually unaliased metadata and events together with the source-qualified revision that identifies those exact values; the Session restore path validates and freezes the graphs in place instead of cloning them. The coordinator computes interrupted-turn closers and constructs the exact unpublished Session once. Its immutable header and balanced logical event log form the `SessionInspection` borrowed by readers, while the revision remains internal to persistence. -`inspect(id, signal?)` does not mutate storage. Synthetic closers exist only in the prepared in-memory view, and a torn physical tail remains untouched. Same-id callers share an in-flight cold read. Once ready, the preparation may remain in a per-coordinator LRU whose capacity defaults to five and is configurable by first-party backends. Before reusing a retained source, the coordinator reads that id's current revision; a mismatch evicts the source and repeats the cold materialization. +`inspect(id, signal?)` does not mutate storage. Synthetic closers exist only in the prepared in-memory view, and a torn physical tail remains untouched. Same-id callers share an in-flight cold read. Once ready, the preparation may remain in a per-coordinator LRU whose capacity defaults to five and is configurable by first-party backends. Before reusing a retained source, the coordinator reads that id's current revision; a mismatch evicts a ready source and repeats the cold materialization. A source already committing or reserved for resume remains exclusively owned, so concurrent inspection borrows that immutable view until publication or release. -`prepare(id, signal?)` exclusively reserves the prepared Session. It confirms the retained revision before committing any torn-tail and interrupted-turn repair, establishes the durable cursor, then returns a disposable preparation. A stale source is discarded and reloaded instead of being repaired or published. Another same-id preparation waits until the reservation is published or released. Publication accepts only the exact reserved Session and attaches the committed cursor without rebuilding its history. Failed setup or cancellation returns an unchanged unpublished Session to the LRU; mutation or attachment consumes the reservation. +`prepare(id, signal?)` exclusively reserves the prepared Session. It confirms the retained revision before committing any torn-tail and interrupted-turn repair, establishes the durable cursor, then returns a disposable preparation. A stale source is discarded and reloaded instead of being repaired or published. A successful repair also discards the pre-repair source and materializes the committed log again before reservation, so a newer revision is never associated with an older event graph. Another same-id preparation waits until the reservation is published or released. Publication accepts only the exact reserved Session and attaches the committed cursor without rebuilding its history. Failed setup or cancellation returns an unchanged unpublished Session to the LRU; mutation or attachment consumes the reservation. The legacy `load(id)` API uses the same preparation and repair machinery, then discards its reservation and returns the immutable logical view. It remains a compatibility API, not the history-to-resume reuse path. This lifecycle extends the [shared persistence coordinator](2026-06-18-shared-persistence-write-coordinator.md) while preserving the storage and recovery rules owned by the [session persistence decision](2026-06-14-session-persistence.md). @@ -32,7 +32,7 @@ The legacy `load(id)` API uses the same preparation and repair machinery, then d History reads use `inspect()`, so repeated pages borrow the same immutable prepared state without activating an Agent. A later resume uses `prepare()` and receives the exact Session retained by inspection; it does not read, decompress, parse, clone, validate, or freeze the complete log again. -If the durable log changes after inspection, its revision changes. The next history read or resume discards the retained Session and materializes the new log, so an old event graph cannot be associated with a newer snapshot revision. +If the durable log changes after inspection, its revision changes. The next history read or resume discards a retained ready Session and materializes the new log, so an old event graph cannot be associated with a newer snapshot revision. A source already claimed by an in-flight resume is not evicted: its exclusive owner keeps it through publication or release, and concurrent history may borrow the same immutable view. Cold continuable-subagent access follows the same path. Descriptor authorization first inspects the child, then `ctx.agents.resume()` reserves and publishes the retained Session. This preserves the lifecycle and authorization rules in the [continuable subagent conversation decision](../feature/2026-07-28-continuable-subagent-conversations.md) while removing its duplicate cold read. diff --git a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.zh.md b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.zh.md index 191ad84db0..a0ca27eb63 100644 --- a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.zh.md +++ b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.zh.md @@ -22,9 +22,9 @@ agent loop(智能体循环)通过同一条设置与发布流水线消费这 使用协调器的持久化实现会将一个冷源加载为准备完成的 Session。后端转移新鲜、彼此无别名的元数据和事件,以及标识这些精确值的来源限定 revision;Session 恢复路径直接验证并冻结这些对象图,不再复制。协调器计算中断轮次的 closer,并且只构造一次精确的未发布 Session。其不可变 header 与平衡逻辑事件日志构成读取方借用的 `SessionInspection`,revision 则保留在持久化内部。 -`inspect(id, signal?)` 不修改存储。合成 closer 只存在于准备完成的内存视图中,撕裂的物理尾部保持不变。同 id 调用方共享进行中的冷读。准备完成后,该对象可以进入每个协调器自己的 LRU;第一方后端可配置容量,默认保留五个。协调器复用保留源之前会读取该 id 的当前 revision;如果不匹配,就淘汰旧源并重新完成冷实体化。 +`inspect(id, signal?)` 不修改存储。合成 closer 只存在于准备完成的内存视图中,撕裂的物理尾部保持不变。同 id 调用方共享进行中的冷读。准备完成后,该对象可以进入每个协调器自己的 LRU;第一方后端可配置容量,默认保留五个。协调器复用保留源之前会读取该 id 的当前 revision;如果不匹配,就淘汰处于就绪阶段的源并重新完成冷实体化。已经进入提交或为恢复而预留的源仍由其所有者独占,因此并发检查会借用该不可变视图,直至发布或释放。 -`prepare(id, signal?)` 独占预留准备完成的 Session。它先确认保留的 revision,再提交撕裂尾部和中断轮次修复、建立持久游标,最后返回可 dispose 的准备对象。过期源会被丢弃并重新读取,不会参与修复或发布。同 id 的另一个准备请求会等待当前预留发布或释放。发布只接受精确的预留 Session,并直接附接已提交游标,无需重建历史。设置失败或取消时,未发生变化的未发布 Session 会返回 LRU;发生变更或完成附接后,系统会消费该预留。 +`prepare(id, signal?)` 独占预留准备完成的 Session。它先确认保留的 revision,再提交撕裂尾部和中断轮次修复、建立持久游标,最后返回可 dispose 的准备对象。陈旧源会被丢弃并重新读取,不会参与修复或发布。修复成功后也会丢弃修复前的源,并在预留前重新实体化已提交日志,以免把较新的 revision 关联到较旧的事件对象图。同 id 的另一个准备请求会等待当前预留发布或释放。发布只接受精确的预留 Session,并直接附接已提交游标,无需重建历史。设置失败或取消时,未发生变化的未发布 Session 会返回 LRU;发生变更或完成附接后,系统会消费该预留。 存量 `load(id)` API 使用相同的准备和修复机制,随后丢弃其预留并返回不可变逻辑视图。它保留为兼容 API,不承担历史到恢复的复用路径。该生命周期扩展了[共享持久化协调器](2026-06-18-shared-persistence-write-coordinator.md),同时继续遵循[会话持久化决策](2026-06-14-session-persistence.md)所规定的存储与恢复规则。 @@ -32,7 +32,7 @@ agent loop(智能体循环)通过同一条设置与发布流水线消费这 历史读取使用 `inspect()`,因此重复分页可以借用同一份不可变准备状态,而不会激活 agent。后续恢复调用 `prepare()`,直接取得检查阶段保留的精确 Session;系统不会再次完整读取、解压、解析、复制、验证或冻结日志。 -如果持久日志在检查后发生变化,其 revision 也会变化。下一次历史读取或恢复会丢弃保留的 Session,并实体化新日志,因此旧事件对象图不会被关联到较新的快照 revision。 +如果持久日志在检查后发生变化,其 revision 也会变化。下一次历史读取或恢复会丢弃保留且处于就绪阶段的 Session,并实体化新日志,因此旧事件对象图不会被关联到较新的快照 revision。已经由进行中恢复操作取得的源不会被淘汰:其独占所有者会持有它直至发布或释放,并发历史读取可以借用同一个不可变视图。 冷 continuable subagent 访问沿用同一路径。系统先检查子会话并完成 descriptor 授权,再由 `ctx.agents.resume()` 预留并发布保留的 Session。这样既遵循 [continuable subagent 会话决策](../feature/2026-07-28-continuable-subagent-conversations.md)中的生命周期与授权规则,也消除了重复冷读。 diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index db1322a275..3e5fa77f67 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -1208,9 +1208,10 @@ abstract load(id: SessionId): Promise * Session instead yields its current immutable snapshot, which may contain an * open turn and its `session/end-seed` boundary. Coordinator-backed * implementations retain the exact cold unpublished Session for bounded - * reuse by a later {@link prepare}, reloading it when its durable revision - * changes; callers borrow only its immutable header and log. Continuous - * external writers may delay revision convergence. + * reuse by a later {@link prepare}. A stale ready source is reloaded; a source + * already committing or reserved for resume remains exclusive, and inspection + * may borrow its immutable view. Callers borrow only the immutable header and + * log. Continuous external writers may delay revision convergence. * @param id - the persisted session to inspect. * @param signal - optional cancellation for queued and backend read work. * @returns the validated header and current logical event log. @@ -1706,7 +1707,7 @@ fork(source: SessionForkSource, boundary?: number, childSessionId?: SessionId): Types: [CreateSessionOptions](../core-data-structures/persistence.md) · [PrepareSessionOptions](../core-data-structures/persistence.md) · [Session](../core-data-structures/session.md) · [SessionId](../core-data-structures/core.md) -Source: [`packages/core/session/src/index.ts:801`](../../packages/core/session/src/index.ts) +Source: [`packages/core/session/src/index.ts:803`](../../packages/core/session/src/index.ts) ## `ctx.sessionTitle` — `SessionTitleService` diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index d00d545068..40b354fce3 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -590,7 +590,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ }, { signature: 'abstract inspect(id: SessionId, signal?: AbortSignal): Promise', - jsDoc: '/**\n * Inspect an immutable logical session without committing recovery or\n * publishing it. A cold complete interrupted turn receives synthetic closers\n * in memory and a torn physical tail remains untouched. An already-live\n * Session instead yields its current immutable snapshot, which may contain an\n * open turn and its `session/end-seed` boundary. Coordinator-backed\n * implementations retain the exact cold unpublished Session for bounded\n * reuse by a later {@link prepare}, reloading it when its durable revision\n * changes; callers borrow only its immutable header and log. Continuous\n * external writers may delay revision convergence.\n * @param id - the persisted session to inspect.\n * @param signal - optional cancellation for queued and backend read work.\n * @returns the validated header and current logical event log.\n */', + jsDoc: '/**\n * Inspect an immutable logical session without committing recovery or\n * publishing it. A cold complete interrupted turn receives synthetic closers\n * in memory and a torn physical tail remains untouched. An already-live\n * Session instead yields its current immutable snapshot, which may contain an\n * open turn and its `session/end-seed` boundary. Coordinator-backed\n * implementations retain the exact cold unpublished Session for bounded\n * reuse by a later {@link prepare}. A stale ready source is reloaded; a source\n * already committing or reserved for resume remains exclusive, and inspection\n * may borrow its immutable view. Callers borrow only the immutable header and\n * log. Continuous external writers may delay revision convergence.\n * @param id - the persisted session to inspect.\n * @param signal - optional cancellation for queued and backend read work.\n * @returns the validated header and current logical event log.\n */', }, { signature: 'abstract readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }>', diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index 876fd5f818..1ff5d0d3eb 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -704,9 +704,10 @@ export class PersistenceCoordinator { /** * Inspect a logical session without publishing it or committing recovery. - * Retained cold state is reloaded after its durable revision changes. Revision - * retries converge once the log is stable for one read/check round trip; - * continuous external writers may delay completion. + * A stale ready source is reloaded. A source already committing or reserved + * for resume remains exclusive, and inspection may borrow its immutable view. + * Revision retries converge once the log is stable for one read/check round + * trip; continuous external writers may delay completion. * @param id - persisted session to inspect. * @param signal - optional cancellation for preparation work. * @returns immutable prepared metadata and events; a live view may have an open turn. @@ -737,6 +738,7 @@ export class PersistenceCoordinator { return source.inspection } } catch (error: unknown) { + signal?.throwIfAborted() const attached = this.ctx.sessions.get(id) if (attached !== undefined) return this.inspectLive(attached) throw error @@ -859,14 +861,11 @@ export class PersistenceCoordinator { throw new Error(`session "${id}" already has a live persistence owner`) } if (!await this.isPreparedSourceCurrent(source)) return undefined - let committedSource = source if (source.tornMarker !== undefined || source.closers.length > 0) { await this.backend.commitRepair(source.inspection.meta, source.tornMarker, source.closers) - const revision = await this.backend.readStoredRevision(id) - if (revision === undefined) { - throw new Error(`session "${id}" disappeared after persistence repair`) - } - committedSource = { ...source, revision, tornMarker: undefined, closers: [] } + // The repair changed the durable revision. Reload the exact committed + // graph instead of associating the old in-memory view with a newer revision. + return undefined } const state = existing ?? { meta: source.inspection.meta, @@ -878,7 +877,7 @@ export class PersistenceCoordinator { state.materialized = true this.states.set(id, state) return { - source: committedSource, + source, state, } } diff --git a/packages/session-persistence/session-persistence/src/index.ts b/packages/session-persistence/session-persistence/src/index.ts index c7955c34e3..94596855df 100644 --- a/packages/session-persistence/session-persistence/src/index.ts +++ b/packages/session-persistence/session-persistence/src/index.ts @@ -147,9 +147,10 @@ export abstract class SessionPersistence extends Service { * Session instead yields its current immutable snapshot, which may contain an * open turn and its `session/end-seed` boundary. Coordinator-backed * implementations retain the exact cold unpublished Session for bounded - * reuse by a later {@link prepare}, reloading it when its durable revision - * changes; callers borrow only its immutable header and log. Continuous - * external writers may delay revision convergence. + * reuse by a later {@link prepare}. A stale ready source is reloaded; a source + * already committing or reserved for resume remains exclusive, and inspection + * may borrow its immutable view. Callers borrow only the immutable header and + * log. Continuous external writers may delay revision convergence. * @param id - the persisted session to inspect. * @param signal - optional cancellation for queued and backend read work. * @returns the validated header and current logical event log. diff --git a/packages/session-persistence/session-persistence/src/preparations.ts b/packages/session-persistence/session-persistence/src/preparations.ts index 31f6fb3d31..2a685f71f9 100644 --- a/packages/session-persistence/session-persistence/src/preparations.ts +++ b/packages/session-persistence/session-persistence/src/preparations.ts @@ -184,11 +184,10 @@ export class SessionPreparations { /** * Discard a prepared view after the durable log changes. * @param id - changed session identity. - * @param expected - when supplied, invalidate only that exact source. */ - invalidate(id: SessionId, expected?: Source): void { + invalidate(id: SessionId): void { const entry = this.entries.get(id) - if (entry !== undefined && (expected === undefined || entry.source === expected)) this.remove(entry) + if (entry !== undefined) this.remove(entry) } /** @@ -219,7 +218,7 @@ export class SessionPreparations { /** * Remove a completed entry for an already-serialized append adoption. * @param id - adopted session identity. - * @returns the prepared source, or undefined when no entry exists. + * @returns the prepared source, or undefined when no ready entry exists. */ takeReady(id: SessionId): Source | undefined { const entry = this.entries.get(id) diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 3d365cd59d..8fac982591 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -875,7 +875,7 @@ describe('PersistenceCoordinator session preparations', () => { second = await coordinator.prepare(id) expect(second.session).toBe(first.session) - expect(backend.loadAttempts).toBe(1) + expect(backend.loadAttempts).toBe(2) expect(backend.repairAttempts).toBe(1) } finally { second?.[Symbol.dispose]() @@ -885,7 +885,52 @@ describe('PersistenceCoordinator session preparations', () => { } }) - it('rejects preparation when storage disappears after repair', async () => { + it('reloads the committed graph when another writer appends after repair', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('repair-external-append') + backend.store.set(id, { + meta: meta(id), + events: [{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } }], + }) + const commitRepair = backend.commitRepair.bind(backend) + vi.spyOn(backend, 'commitRepair').mockImplementation(async (header, tornMarker, closers) => { + await commitRepair(header, tornMarker, closers) + const entry = backend.store.get(id) + if (entry === undefined) throw new Error('test repair must keep storage materialized') + const seq = entry.events.length + entry.events.push( + { type: 'turn/start', seq, time: 3, data: { turn: 2 } }, + { type: 'turn/end', seq: seq + 1, time: 4, data: { turn: 2, reason: { kind: 'completed' } } }, + ) + }) + let coordinator!: PersistenceCoordinator + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + let preparation: Awaited> | undefined + + try { + preparation = await coordinator.prepare(id) + + expect(preparation.session.events.map(event => event.type)).toEqual([ + 'turn/start', + 'turn/end', + 'turn/start', + 'turn/end', + 'session/end-seed', + ]) + expect(backend.loadAttempts).toBe(2) + expect(backend.repairAttempts).toBe(1) + } finally { + preparation?.[Symbol.dispose]() + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + + it('rejects preparation when storage disappears during the post-repair reload', async () => { const ctx = new Context() await ctx.plugin(SessionStore) const backend = new ControlledBackend() @@ -894,24 +939,20 @@ describe('PersistenceCoordinator session preparations', () => { meta: meta(id), events: [{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } }], }) + const commitRepair = backend.commitRepair.bind(backend) + vi.spyOn(backend, 'commitRepair').mockImplementation(async (header, tornMarker, closers) => { + await commitRepair(header, tornMarker, closers) + backend.store.delete(id) + }) let coordinator!: PersistenceCoordinator const fiber = await ctx.plugin(Object.assign((inner: Context) => { coordinator = new PersistenceCoordinator(inner, backend) }, { inject: ['sessions'] })) try { - await coordinator.inspect(id) - const readStoredRevision = backend.readStoredRevision.bind(backend) - let revisionReads = 0 - vi.spyOn(backend, 'readStoredRevision').mockImplementation((sessionId, signal) => { - revisionReads += 1 - if (revisionReads === 2) return Promise.resolve(undefined) - return readStoredRevision(sessionId, signal) - }) - - await expect(coordinator.prepare(id)).rejects.toThrow(/disappeared after persistence repair/) + await expect(coordinator.prepare(id)).rejects.toThrow(/not found/) expect(backend.repairAttempts).toBe(1) - expect(revisionReads).toBe(2) + expect(backend.loadAttempts).toBe(2) } finally { await fiber.dispose() await ctx.fiber.dispose() @@ -1089,6 +1130,36 @@ describe('PersistenceCoordinator observation cancellation', () => { } }) + it('preserves inspect cancellation when the session concurrently becomes live', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('cancelled-inspect-became-live') + backend.store.set(id, { meta: meta(id), events: oneTurnLog() }) + const controller = new AbortController() + const reason = new Error('inspect cancelled while publishing') + backend.beforeLoadStored = async () => { + controller.abort(reason) + throw new Error('load stopped after cancellation') + } + let coordinator!: PersistenceCoordinator + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + const live = Session.create(id, oneTurnLog(), meta(id)) + const get = vi.spyOn(ctx.sessions, 'get') + .mockReturnValueOnce(undefined) + .mockReturnValueOnce(live) + + try { + await expect(coordinator.inspect(id, controller.signal)).rejects.toBe(reason) + } finally { + get.mockRestore() + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + it('readFrom via the seek hook: serves the suffix, maps undefined to not-found, and relays hook failures by abort state', async () => { const ctx = new Context() await ctx.plugin(SessionStore) diff --git a/packages/session-persistence/session-persistence/tests/preparations.spec.ts b/packages/session-persistence/session-persistence/tests/preparations.spec.ts index b59a1590d5..5e12f29a79 100644 --- a/packages/session-persistence/session-persistence/tests/preparations.spec.ts +++ b/packages/session-persistence/session-persistence/tests/preparations.spec.ts @@ -34,8 +34,6 @@ describe('SessionPreparations inspection', () => { await expect(preparations.inspect(id, load)).resolves.toBe(source) expect(load).toHaveBeenCalledOnce() - preparations.invalidate(id, prepared('different-source')) - expect(preparations.has(id)).toBe(true) preparations.invalidate(id) preparations.invalidate(id) expect(preparations.has(id)).toBe(false)