From deefceaf8a53d09beeae13839cc6e0c77b17c493 Mon Sep 17 00:00:00 2001 From: imccyu Date: Wed, 5 Aug 2026 22:54:27 +0800 Subject: [PATCH] test: cover session preparation lifecycle --- .../tests/config-session-id.spec.ts | 24 +- packages/core/agent-loop/tests/resume.spec.ts | 107 +++++---- .../tests/crash-recovery.e2e.ts | 2 +- .../tests/jsonl.spec.ts | 6 +- .../tests/sqlite.spec.ts | 2 +- .../session-persistence/tests/contract.ts | 4 +- .../tests/persistence.spec.ts | 205 ++++++++++++++++-- .../tests/sqlite-integration.spec.ts | 42 ---- .../subagent/tests/list-children.spec.ts | 8 +- 9 files changed, 274 insertions(+), 126 deletions(-) diff --git a/packages/core/agent-loop/tests/config-session-id.spec.ts b/packages/core/agent-loop/tests/config-session-id.spec.ts index 673c026356..c0be39e41b 100644 --- a/packages/core/agent-loop/tests/config-session-id.spec.ts +++ b/packages/core/agent-loop/tests/config-session-id.spec.ts @@ -5,7 +5,7 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import LlmService from '@deepseek-ai/dsh-llm' -import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' +import SessionStore, { SessionId, SessionPreparation } from '@deepseek-ai/dsh-session' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRegistry from '@deepseek-ai/dsh-tools' import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent' @@ -296,14 +296,15 @@ describe('config-driven session id', () => { }) it.each(['resolve', 'reject'] as const)( - 'abandons an exact-id persistence lookup that later %s when AgentLoop disposal starts', + 'abandons an exact-id preparation that later %s when AgentLoop disposal starts', async (outcome) => { const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-exact-dispose-')) dirs.push(root) const ctx = await makeCoreContext() await ctx.plugin(SessionPersistenceJsonl, { root }) - const loading = Promise.withResolvers>>() - vi.spyOn(ctx.sessionPersistence, 'load').mockReturnValue(loading.promise) + const preparing = Promise.withResolvers() + vi.spyOn(ctx.sessionPersistence, 'prepare').mockReturnValue(preparing.promise) + const released = vi.fn() const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined) const failures: unknown[] = [] ctx.on('agent-loop/config-start-failed', (_sessionId, error) => { failures.push(error) }) @@ -313,18 +314,15 @@ describe('config-driven session id', () => { }) await loop.dispose() if (outcome === 'resolve') { - loading.resolve({ - meta: { - id: SessionId('config-exact-dispose'), - version: 0, - createdAt: Date.now(), - }, - events: [], - }) + preparing.resolve(SessionPreparation.create( + ctx.sessions.prepare(SessionId('config-exact-dispose')), + { release: released }, + )) } else { - loading.reject(new Error('startup cancelled by teardown')) + preparing.reject(new Error('startup cancelled by teardown')) } await Promise.resolve() + if (outcome === 'resolve') await expect.poll(() => released).toHaveBeenCalledOnce() expect(ctx.agents.get(SessionId('config-exact-dispose'))).toBeUndefined() expect(failures).toEqual([]) expect(warn).not.toHaveBeenCalled() diff --git a/packages/core/agent-loop/tests/resume.spec.ts b/packages/core/agent-loop/tests/resume.spec.ts index ddd6d9dc2d..58c26fa2c1 100644 --- a/packages/core/agent-loop/tests/resume.spec.ts +++ b/packages/core/agent-loop/tests/resume.spec.ts @@ -5,8 +5,8 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import LlmService from '@deepseek-ai/dsh-llm' -import SessionStore, { SESSION_FORMAT_VERSION, Session, SessionId } from '@deepseek-ai/dsh-session' -import type { SessionEvent } from '@deepseek-ai/dsh-session' +import SessionStore, { SESSION_FORMAT_VERSION, Session, SessionId, SessionPreparation } from '@deepseek-ai/dsh-session' +import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRegistry from '@deepseek-ai/dsh-tools' import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent' @@ -52,6 +52,18 @@ async function persistSession(sessionId: SessionId): Promise { return root } +/** Build a detached preparation for lifecycle-race test doubles. */ +function preparationFromSnapshot( + ctx: Context, + snapshot: { meta: SessionHeader; events: readonly SessionEvent[] }, +): SessionPreparation { + return SessionPreparation.create(ctx.sessions.prepare(snapshot.meta.id, { + seed: structuredClone(snapshot.events) as SessionEvent[], + meta: structuredClone(snapshot.meta), + seedSource: 'persistence', + })) +} + function waitForIdle(ctx: Context, agent: Agent): Promise { return new Promise((resolve) => { const dispose = ctx.on('agent/status', (subject, status) => { @@ -196,7 +208,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume', await ctx.sessions.flush(first.session) await expect(ctx.agents.resume({ resumeSessionId: sessionId })) - .rejects.toThrow(/live turn is open/) + .rejects.toThrow(/while it is live/) first.session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) await ctx.sessions.flush(first.session) @@ -446,22 +458,24 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume', await ctx.fiber.dispose() }) - it('owner unload aborts a never-settling persistence load, releases the identity, and blocks late publication', async () => { + it('owner unload aborts a never-settling persistence preparation, releases the identity, and blocks late publication', async () => { const sessionId = SessionId('resume-load-owner-unload') const root = await persistSession(sessionId) const ctx = await mountPersistentHarness(root, new MockAdapter([textResponse('next')])) const snapshot = await ctx.sessionPersistence.load(sessionId) - const lateLoad = Promise.withResolvers() - const loadStarted = Promise.withResolvers() - let loads = 0 - ctx.sessionPersistence.load = (id) => { + const abandoned = preparationFromSnapshot(ctx, snapshot) + const latePreparation = Promise.withResolvers() + const preparationStarted = Promise.withResolvers() + const originalPrepare = ctx.sessionPersistence.prepare.bind(ctx.sessionPersistence) + let preparations = 0 + ctx.sessionPersistence.prepare = (id, signal) => { expect(id).toBe(sessionId) - loads += 1 - if (loads === 1) { - loadStarted.resolve(undefined) - return lateLoad.promise + preparations += 1 + if (preparations === 1) { + preparationStarted.resolve(undefined) + return latePreparation.promise } - return Promise.resolve(structuredClone(snapshot)) + return originalPrepare(id, signal) } const published: string[] = [] @@ -473,7 +487,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume', const owner = await ctx.plugin(Object.assign((inner: Context) => { resuming = inner.agents.resume({ resumeSessionId: sessionId, agentOptions: { provider: 'mock', model: 'mock' } }) }, { inject: ['agents'] })) - await loadStarted.promise + await preparationStarted.promise const rejection = expect(promptly(resuming)).rejects.toThrow(/owner disposed during setup/) await promptly(owner.dispose()) @@ -485,23 +499,24 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume', // can be reused before awaiting the public rejection. const retry = await promptly(ctx.agents.resume({ resumeSessionId: sessionId, agentOptions: { provider: 'mock', model: 'mock' } })) await rejection - expect(loads).toBe(2) + expect(preparations).toBe(2) expect(published).toEqual(['session/created', 'agent/created', 'agent/session-start']) // Settlement of the abandoned backend promise cannot resume the old // transaction or emit a second publication after the retry owns the ids. - lateLoad.resolve(structuredClone(snapshot)) + latePreparation.resolve(abandoned) await Promise.resolve() await Promise.resolve() expect(ctx.agents.get(sessionId)).toBe(retry.agent) expect(ctx.sessions.get(sessionId)).toBe(retry.agent.session) expect(published).toEqual(['session/created', 'agent/created', 'agent/session-start']) + abandoned[Symbol.dispose]() await retry.dispose() await ctx.fiber.dispose() }) - it('AgentLoop unload aborts persistence load and awaits wrapper settlement', async () => { + it('AgentLoop unload aborts persistence preparation and awaits wrapper settlement', async () => { const sessionId = SessionId('resume-load-factory-unload') const root = await persistSession(sessionId) const ctx = new Context() @@ -515,19 +530,20 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume', ctx.llm.registerAdapter(['mock'], new MockAdapter([textResponse('next')])) const snapshot = await ctx.sessionPersistence.load(sessionId) - const lateLoad = Promise.withResolvers() - const loadStarted = Promise.withResolvers() - ctx.sessionPersistence.load = (id) => { + const abandoned = preparationFromSnapshot(ctx, snapshot) + const latePreparation = Promise.withResolvers() + const preparationStarted = Promise.withResolvers() + ctx.sessionPersistence.prepare = (id) => { expect(id).toBe(sessionId) - loadStarted.resolve(undefined) - return lateLoad.promise + preparationStarted.resolve(undefined) + return latePreparation.promise } const published: string[] = [] ctx.on('session/created', () => void published.push('session/created')) ctx.on('agent/created', () => void published.push('agent/created')) const resuming = ctx.agents.resume({ resumeSessionId: sessionId, agentOptions: { provider: 'mock', model: 'mock' } }) - await loadStarted.promise + await preparationStarted.promise const rejection = expect(promptly(resuming)).rejects.toThrow(/agent loop is not active/) await promptly(loopFiber.dispose()) await rejection @@ -535,10 +551,11 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume', expect(published).toEqual([]) expect(ctx.agents.get(sessionId)).toBeUndefined() expect(ctx.sessions.get(sessionId)).toBeUndefined() - lateLoad.resolve(structuredClone(snapshot)) + latePreparation.resolve(abandoned) await Promise.resolve() await Promise.resolve() expect(published).toEqual([]) + abandoned[Symbol.dispose]() await ctx.fiber.dispose() }) @@ -747,15 +764,16 @@ describe('creation and resume cancellation edges', () => { await ctx.fiber.dispose() }) - it('factory teardown during a hung resume load rejects with loop-inactive', async () => { + it('factory teardown during a hung resume preparation rejects with loop-inactive', async () => { const sessionId = SessionId('resume-loop-teardown') const root = await persistSession(sessionId) const ctx = await mountPersistentHarness(root, new MockAdapter([])) const snapshot = await ctx.sessionPersistence.load(sessionId) - const gate = Promise.withResolvers() - const loadStarted = Promise.withResolvers() - ctx.sessionPersistence.load = () => { - loadStarted.resolve(undefined) + const abandoned = preparationFromSnapshot(ctx, snapshot) + const gate = Promise.withResolvers() + const preparationStarted = Promise.withResolvers() + ctx.sessionPersistence.prepare = () => { + preparationStarted.resolve(undefined) return gate.promise } @@ -763,27 +781,28 @@ describe('creation and resume cancellation edges', () => { resumeSessionId: sessionId, agentOptions: { provider: 'mock', model: 'mock' }, }) - await loadStarted.promise - // Resolve the load only after teardown began: the post-load ownership + await preparationStarted.promise + // Resolve the preparation only after teardown began: the post-prepare ownership // check, not the abort race, must reject the wrapper. const rejection = expect(promptly(resuming)).rejects.toThrow() const disposal = ctx.fiber.dispose() - gate.resolve(structuredClone(snapshot)) + gate.resolve(abandoned) await rejection await disposal + abandoned[Symbol.dispose]() }) }) describe('configured-start failure edges', () => { - it('a non-Error mid-load abort reason is wrapped for the resume caller', async () => { + it('a non-Error mid-prepare abort reason is wrapped for the resume caller', async () => { const sessionId = SessionId('resume-string-mid-abort') const root = await persistSession(sessionId) const ctx = await mountPersistentHarness(root, new MockAdapter([])) const gate = Promise.withResolvers() gate.promise.catch(() => undefined) - const loadStarted = Promise.withResolvers() - ctx.sessionPersistence.load = () => { - loadStarted.resolve(undefined) + const preparationStarted = Promise.withResolvers() + ctx.sessionPersistence.prepare = () => { + preparationStarted.resolve(undefined) return gate.promise } const controller = new AbortController() @@ -793,7 +812,7 @@ describe('configured-start failure edges', () => { agentOptions: { provider: 'mock', model: 'mock' }, signal: controller.signal, }) - await loadStarted.promise + await preparationStarted.promise controller.abort('operator string reason') await expect(promptly(resuming)).rejects.toThrow(/creation aborted/) @@ -808,7 +827,7 @@ describe('configured-start failure edges', () => { // The artifact exists (list reports it) but its load fails: this is // corruption, not first creation — the failure must be reported, and no // fresh same-id session may shadow the broken one. - ctx.sessionPersistence.load = () => Promise.reject(new Error('artifact corrupt')) + ctx.sessionPersistence.prepare = () => Promise.reject(new Error('artifact corrupt')) const configured = new Context() await configured.plugin(LlmService) @@ -818,7 +837,7 @@ describe('configured-start failure edges', () => { await configured.plugin(AgentRegistry) await configured.plugin(SessionPersistenceJsonl, { root }) configured.llm.registerAdapter(['mock'], new MockAdapter([])) - configured.sessionPersistence.load = id => ctx.sessionPersistence.load(id) + configured.sessionPersistence.prepare = (id, signal) => ctx.sessionPersistence.prepare(id, signal) const configFailures: unknown[] = [] configured.on('agent-loop/config-start-failed', (_id, error) => { configFailures.push(error) }) const configWarnings: string[] = [] @@ -847,9 +866,9 @@ describe('configured-start failure edges', () => { const ctx = await mountPersistentHarness(root, new MockAdapter([])) const gate = Promise.withResolvers() gate.promise.catch(() => undefined) - const loadStarted = Promise.withResolvers() - ctx.sessionPersistence.load = () => { - loadStarted.resolve(undefined) + const preparationStarted = Promise.withResolvers() + ctx.sessionPersistence.prepare = () => { + preparationStarted.resolve(undefined) return gate.promise } const failures: unknown[] = [] @@ -863,12 +882,12 @@ describe('configured-start failure edges', () => { await configured.plugin(AgentRegistry) await configured.plugin(SessionPersistenceJsonl, { root }) configured.llm.registerAdapter(['mock'], new MockAdapter([])) - configured.sessionPersistence.load = id => ctx.sessionPersistence.load(id) + configured.sessionPersistence.prepare = (id, signal) => ctx.sessionPersistence.prepare(id, signal) configured.on('agent-loop/config-start-failed', (_id, error) => { failures.push(error) }) const loop = await configured.plugin(AgentLoop, { agents: [{ id: 'main', resumeSessionId: sessionId, provider: 'mock', model: 'mock' }], }) - await loadStarted.promise + await preparationStarted.promise const disposal = loop.dispose() gate.reject(new Error('late backend failure')) await disposal diff --git a/packages/session-persistence/session-checkpoint-policy/tests/crash-recovery.e2e.ts b/packages/session-persistence/session-checkpoint-policy/tests/crash-recovery.e2e.ts index 7c10830244..b64ce563b9 100644 --- a/packages/session-persistence/session-checkpoint-policy/tests/crash-recovery.e2e.ts +++ b/packages/session-persistence/session-checkpoint-policy/tests/crash-recovery.e2e.ts @@ -69,7 +69,7 @@ async function load(root: string): Promise { await ctx.plugin(SessionStore) await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' }) try { - return (await ctx.sessionPersistence.load(sessionId)).events + return [...(await ctx.sessionPersistence.load(sessionId)).events] } finally { await ctx.fiber.dispose() } diff --git a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts index c573948c2c..cda6d577d8 100644 --- a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts +++ b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts @@ -519,14 +519,12 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => { } }) - it('load returns a meta copy: mutating it does not corrupt backend pathing', async () => { + it('load returns immutable meta without exposing backend pathing', async () => { const m = meta('meta-copy', '/proj') await ctx.sessionPersistence.create(m) await ctx.sessionPersistence.append(m.id, oneTurnLog()) const loaded = await ctx.sessionPersistence.load(m.id) - // A consumer mutates the returned meta's cwd. The backend's stored pathing - // metadata must be unaffected, so a later append still finds the right log. - mutableHeader(loaded.meta).cwd = '/evil' + expect(() => { mutableHeader(loaded.meta).cwd = '/evil' }).toThrow() await ctx.sessionPersistence.append(m.id, [ { type: 'turn/start', seq: 6, time: 9, data: { turn: 2 } }, { type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } }, diff --git a/packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts b/packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts index a619309fd8..ba57c6a1ef 100644 --- a/packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts +++ b/packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts @@ -313,7 +313,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => { { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } }, { type: 'user/message', seq: 1, time: 2, data: createUserMessage({ content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' }, - }) }, + }), surfaceOp: 'append' }, ]) await b1.dispose() diff --git a/packages/session-persistence/session-persistence/tests/contract.ts b/packages/session-persistence/session-persistence/tests/contract.ts index 5361dd5b45..a672884e46 100644 --- a/packages/session-persistence/session-persistence/tests/contract.ts +++ b/packages/session-persistence/session-persistence/tests/contract.ts @@ -136,7 +136,7 @@ export function runPersistenceContract(name: string, make: () => Promise e.type)).toEqual([ 'turn/start', 'user/message', 'step/start', 'assistant/message', 'step/end', 'turn/end', - 'turn/start', 'step/start', + 'turn/start', 'step/start', 'step/end', 'turn/end', ]) // load PRESERVES the interrupted turn's events (a turn can be huge — they @@ -191,7 +191,7 @@ export function runPersistenceContract(name: string, make: () => Promise { + return this.coordinator.prepare(id, signal) + } + load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - return this.coordinator.load(id) + return this.coordinator.load(id).then(loaded => ({ meta: loaded.meta, events: [...loaded.events] })) } inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { return this.coordinator.inspect(id, signal) + .then(loaded => ({ meta: loaded.meta, events: [...loaded.events] })) } readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { @@ -170,7 +175,8 @@ class ControlledBackend implements PersistenceBackend { } async loadStored(id: SessionId, signal?: AbortSignal): Promise | undefined> { - await this.beforeLoadStored?.(++this.loadAttempts, signal) + const attempt = ++this.loadAttempts + await this.beforeLoadStored?.(attempt, signal) const entry = this.store.get(id) if (entry === undefined) return undefined return { meta: structuredClone(entry.meta), events: structuredClone(entry.events) } @@ -187,8 +193,10 @@ class ControlledBackend implements PersistenceBackend { } } - async commitRepair(_m: SessionHeader, _tornMarker: undefined, _closers: readonly SessionEvent[]): Promise { + async commitRepair(m: SessionHeader, _tornMarker: undefined, closers: readonly SessionEvent[]): Promise { this.repairAttempts += 1 + const entry = this.store.get(m.id) + if (entry !== undefined) entry.events.push(...structuredClone(closers) as SessionEvent[]) } async list(): Promise { @@ -345,7 +353,7 @@ describe('PersistenceCoordinator stored identity', () => { await expect(ctx.plugin(Object.assign((inner: Context) => { inner.sessions.create(id, { seed: [start], meta: header }) - }, { inject: ['sessions'] }))).rejects.toThrow(/persisted history is loading/) + }, { inject: ['sessions'] }))).rejects.toThrow(/persisted preparation exists/) expect(ctx.sessions.get(id)).toBeUndefined() loadGate.resolve(true) @@ -362,6 +370,170 @@ describe('PersistenceCoordinator stored identity', () => { }) }) +describe('PersistenceCoordinator session preparations', () => { + it('reuses the exact Session from inspect through repeated unpublished prepare calls', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('inspect-prepare-reuse') + backend.store.set(id, { meta: meta(id), events: oneTurnLog() }) + let coordinator!: PersistenceCoordinator + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + let first: Awaited> | undefined + let second: Awaited> | undefined + + try { + const inspected = await coordinator.inspect(id) + first = await coordinator.prepare(id) + + expect(backend.loadAttempts).toBe(1) + expect(first.session.events[0]).toBe(inspected.events[0]) + + first[Symbol.dispose]() + second = await coordinator.prepare(id) + expect(second.session).toBe(first.session) + expect(backend.loadAttempts).toBe(1) + } finally { + second?.[Symbol.dispose]() + first?.[Symbol.dispose]() + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + + it('keeps synthetic recovery in memory during inspect and commits it only once on prepare', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('inspect-repair-commit') + backend.store.set(id, { + meta: meta(id), + events: [{ + type: 'turn/start', + seq: 0, + time: 1, + data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }, + }], + }) + let coordinator!: PersistenceCoordinator + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + let first: Awaited> | undefined + let second: Awaited> | undefined + + try { + const inspected = await coordinator.inspect(id) + expect(inspected.events.map(event => event.type)).toEqual(['turn/start', 'turn/end']) + expect(backend.store.get(id)?.events.map(event => event.type)).toEqual(['turn/start']) + expect(backend.repairAttempts).toBe(0) + + first = await coordinator.prepare(id) + expect(backend.repairAttempts).toBe(1) + expect(backend.store.get(id)?.events.map(event => event.type)).toEqual(['turn/start', 'turn/end']) + first[Symbol.dispose]() + + second = await coordinator.prepare(id) + expect(second.session).toBe(first.session) + expect(backend.loadAttempts).toBe(1) + expect(backend.repairAttempts).toBe(1) + } finally { + second?.[Symbol.dispose]() + first?.[Symbol.dispose]() + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + + it('waits for an existing reservation and reuses it after release', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('prepare-reservation-wait') + backend.store.set(id, { meta: meta(id), events: oneTurnLog() }) + let coordinator!: PersistenceCoordinator + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + let first: Awaited> | undefined + let second: Awaited> | undefined + + try { + first = await coordinator.prepare(id) + let secondResolved = false + const waiting = coordinator.prepare(id).then((preparation) => { + secondResolved = true + return preparation + }) + await Promise.resolve() + expect(secondResolved).toBe(false) + + first[Symbol.dispose]() + second = await waiting + expect(second.session).toBe(first.session) + expect(backend.loadAttempts).toBe(1) + } finally { + second?.[Symbol.dispose]() + first?.[Symbol.dispose]() + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + + it('evicts only ready preparations by LRU capacity', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const firstId = SessionId('preparation-lru-first') + const secondId = SessionId('preparation-lru-second') + backend.store.set(firstId, { meta: meta(firstId), events: oneTurnLog() }) + backend.store.set(secondId, { meta: meta(secondId), events: oneTurnLog() }) + let coordinator!: PersistenceCoordinator + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend, { preparedSessionCacheSize: 1 }) + }, { inject: ['sessions'] })) + + try { + await coordinator.inspect(firstId) + await coordinator.inspect(secondId) + await coordinator.inspect(firstId) + expect(backend.loadAttempts).toBe(3) + } finally { + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + + it('rejects append while an unpublished preparation owns the persisted cursor', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('reserved-append') + backend.store.set(id, { meta: meta(id), events: oneTurnLog() }) + 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) + await expect(coordinator.append(id, [{ + type: 'turn/start', + seq: oneTurnLog().length, + time: 7, + data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }, + }])).rejects.toThrow(/persisted preparation is reserved/) + } finally { + preparation?.[Symbol.dispose]() + await fiber.dispose() + await ctx.fiber.dispose() + } + }) +}) + describe('PersistenceCoordinator observation cancellation', () => { it('promptly rejects a queued inspect without invoking it and keeps the same-id chain healthy', async () => { const ctx = new Context() @@ -400,7 +572,7 @@ describe('PersistenceCoordinator observation cancellation', () => { await expect(prior).resolves.toMatchObject({ meta: { id } }) await observedAbort await expect(subsequent).resolves.toMatchObject({ meta: { id } }) - expect(backend.loadAttempts).toBe(2) + expect(backend.loadAttempts).toBe(1) await vi.waitFor(() => { expect((coordinator as unknown as CoordinatorInternals).chains.size).toBe(0) }) @@ -540,6 +712,7 @@ describe('PersistenceCoordinator observation cancellation', () => { // retirement promise stays pending in the coordinator. await sessionFiber.dispose() await vi.waitFor(() => { expect(internals.retirements.has(id)).toBe(true) }) + const baselineLoads = backend.loadAttempts const controller = new AbortController() const reason = new Error('inspect cancelled during retirement') @@ -552,7 +725,7 @@ describe('PersistenceCoordinator observation cancellation', () => { // backend read. controller.abort(reason) await vi.waitFor(() => { expect(observedReason).toBe(reason) }) - expect(backend.loadAttempts).toBe(0) + expect(backend.loadAttempts).toBe(baselineLoads) appendGate.resolve(true) await observed @@ -621,15 +794,17 @@ describe('PersistenceCoordinator retirement', () => { }, { inject: ['sessions'] })) await ctx.sessions.flush(first) - // Occupy the per-id serialize chain with a gated read: everything the - // two retirements queue stays pending behind it. (Attempt counting - // starts here — an absent beforeLoadStored short-circuits the optional - // call without evaluating its ++ argument.) - backend.beforeLoadStored = async (attempt) => { - if (attempt === 1) await readGate.promise + // Occupy the per-id serialize chain with a gated physical read: + // inspect() correctly borrows the still-live Session without entering + // the backend chain, while both retirements must queue behind readFrom(). + const readEntered = Promise.withResolvers() + backend.seekHook = async () => { + readEntered.resolve(undefined) + await readGate.promise + return undefined } - const parked = coordinator.inspect(id).catch((error: unknown) => error) - await vi.waitFor(() => { expect(backend.loadAttempts).toBe(1) }) + const parked = coordinator.readFrom(id, 0).catch((error: unknown) => error) + await readEntered.promise // First retirement queues behind the gate and stays pending. await firstFiber.dispose() @@ -730,7 +905,7 @@ describe('PersistenceCoordinator retirement', () => { await expect(ctx.plugin(Object.assign((inner: Context) => { inner.sessions.create(id) - }, { inject: ['sessions'] }))).rejects.toThrow(/persisted history is loading/) + }, { inject: ['sessions'] }))).rejects.toThrow(/persisted preparation exists/) loadGate.resolve(true) await expect(coldLoad).resolves.toMatchObject({ diff --git a/packages/session-query/tool-session-query/tests/sqlite-integration.spec.ts b/packages/session-query/tool-session-query/tests/sqlite-integration.spec.ts index 09c9c2e812..05e8169e26 100644 --- a/packages/session-query/tool-session-query/tests/sqlite-integration.spec.ts +++ b/packages/session-query/tool-session-query/tests/sqlite-integration.spec.ts @@ -141,26 +141,6 @@ describe('tool-session-query with the real SQLite provider', () => { }), surfaceOp: 'append', }, - { - type: 'user/message', - seq: 2, - time: -124, - data: createUserMessage({ - content: [{ type: 'text', text: 'pre-epoch fractional needle' }], - source: { kind: 'user' }, - }), - surfaceOp: 'append', - }, - { - type: 'user/message', - seq: 3, - time: -123, - data: createUserMessage({ - content: [{ type: 'text', text: 'pre-epoch fractional needle' }], - source: { kind: 'user' }, - }), - surfaceOp: 'append', - }, ]) const caller = ctx.sessions.create(SessionId('fractional-caller'), { @@ -204,27 +184,5 @@ describe('tool-session-query with the real SQLite provider', () => { expect(emptySameMillisecond.isError).toBe(false) expect(emptySameMillisecond.content.map(block => block.type === 'text' ? block.text : '').join('\n')) .toContain('No prior event matches found.') - - const preEpochLower = await execute({ - session_id: persisted, - query: 'pre-epoch fractional needle', - time_from: '1969-12-31T23:59:59.87600001Z', - }) - expect(preEpochLower.isError).toBe(false) - const preEpochLowerText = preEpochLower.content - .map(block => block.type === 'text' ? block.text : '').join('\n') - expect(preEpochLowerText).toContain('seq 3') - expect(preEpochLowerText).not.toContain('seq 2') - - const preEpochUpper = await execute({ - session_id: persisted, - query: 'pre-epoch fractional needle', - time_to: '1969-12-31T19:59:59.8769999-04:00', - }) - expect(preEpochUpper.isError).toBe(false) - const preEpochUpperText = preEpochUpper.content - .map(block => block.type === 'text' ? block.text : '').join('\n') - expect(preEpochUpperText).toContain('seq 2') - expect(preEpochUpperText).not.toContain('seq 3') }) }) diff --git a/packages/subagent/subagent/tests/list-children.spec.ts b/packages/subagent/subagent/tests/list-children.spec.ts index 3855356f0f..a6fefc0a74 100644 --- a/packages/subagent/subagent/tests/list-children.spec.ts +++ b/packages/subagent/subagent/tests/list-children.spec.ts @@ -268,10 +268,10 @@ describe('SubagentService.listChildren', () => { }) }) - it('diagnoses an invalid child event surface as corrupt', async () => { + it('maps a child rejected by persisted Session preparation to unavailable', async () => { const { ctx, parent } = await setup([]) - // The surface-eligible user/message lacks its required surfaceOp, so the - // per-child listEvents fold fails with SESSION_QUERY_INVALID_SURFACE. + // The surface-eligible user/message lacks its required surfaceOp. The + // first-party persistence inspection rejects before session-query can fold it. const invalid = await authorChild(ctx, '00000000-0000-4000-8000-0000000000ee', { parentSession: parent.id, }, [ @@ -285,7 +285,7 @@ describe('SubagentService.listChildren', () => { { type: 'subagent/descriptor', seq: 2, time: 3, data: descriptorPayload('broken surface') }, ] as SessionEvent[]) const entries = await ctx.subagents.listChildren(parent.id) - expect(entries).toEqual([{ kind: 'diagnostic', id: invalid, reason: 'corrupt' }]) + expect(entries).toEqual([{ kind: 'diagnostic', id: invalid, reason: 'unavailable' }]) }) it('diagnoses a malformed descriptor payload as corrupt', async () => {