diff --git a/docs/module-graph.md b/docs/module-graph.md index 729745f7ad..255202bbae 100644 --- a/docs/module-graph.md +++ b/docs/module-graph.md @@ -211,6 +211,7 @@ flowchart TD end subgraph group_session_projection["packages/session-projection"] pkg_session_projection["session-projection"] + pkg_session_projection_cache["session-projection-cache"] end subgraph group_storage["packages/storage"] pkg_storage["storage"] @@ -502,6 +503,11 @@ flowchart TD pkg_pty --> pkg_invariants pkg_scripts --> pkg_app_boot pkg_scripts --> pkg_invariants + pkg_session_projection_cache --> pkg_invariants + pkg_session_projection_cache --> pkg_session + pkg_session_projection_cache --> pkg_session_persistence + pkg_session_projection_cache --> pkg_session_projection + pkg_session_projection_cache --> pkg_storage_domain pkg_tasks --> pkg_agent pkg_tasks --> pkg_brand pkg_tasks --> pkg_invariants @@ -1003,6 +1009,7 @@ flowchart TD | [`time-context`](../packages/context/time-context) | `context` | [`agent`](../packages/core/agent), [`invariants`](../packages/support/invariants), [`session`](../packages/core/session) | | [`pty`](../packages/pty/pty) | `pty` | [`agent`](../packages/core/agent), [`brand`](../packages/util/brand), [`invariants`](../packages/support/invariants) | | [`scripts`](../packages/sdk/scripts) | `sdk` | [`app-boot`](../packages/ui/app-boot), [`invariants`](../packages/support/invariants) | +| [`session-projection-cache`](../packages/session-projection/session-projection-cache) | `session-projection` | [`invariants`](../packages/support/invariants), [`session`](../packages/core/session), [`session-persistence`](../packages/session-persistence/session-persistence), [`session-projection`](../packages/session-projection/session-projection), [`storage-domain`](../packages/storage/storage-domain) | | [`tasks`](../packages/tasks/tasks) | `tasks` | [`agent`](../packages/core/agent), [`brand`](../packages/util/brand), [`invariants`](../packages/support/invariants), [`session`](../packages/core/session) | | [`session-telemetry`](../packages/telemetry/session-telemetry) | `telemetry` | [`agent`](../packages/core/agent), [`invariants`](../packages/support/invariants), [`session`](../packages/core/session) | | [`workflow`](../packages/workflow/workflow) | `workflow` | [`agent`](../packages/core/agent), [`brand`](../packages/util/brand), [`invariants`](../packages/support/invariants), [`llm`](../packages/llm/llm), [`session`](../packages/core/session) | diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 966060dfbb..bca32438f4 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -161,6 +161,13 @@ class ControlledBackend implements PersistenceBackend { repairAttempts = 0 beforeAppend?: (attempt: number) => Promise beforeLoadStored?: (attempt: number, signal?: AbortSignal) => Promise + /** When set, the declared seek hook delegates here so readFrom exercises it; unset throws (tests set it first). */ + seekHook?: (id: SessionId, fromSeq: number, signal?: AbortSignal) => Promise | undefined> + + loadStoredFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise | undefined> { + if (this.seekHook === undefined) throw new Error('seekHook not configured for this test') + return this.seekHook(id, fromSeq, signal) + } async loadStored(id: SessionId, signal?: AbortSignal): Promise | undefined> { await this.beforeLoadStored?.(++this.loadAttempts, signal) @@ -457,6 +464,58 @@ describe('PersistenceCoordinator observation cancellation', () => { } }) + 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) + const backend = new ControlledBackend() + const id = SessionId('seek-read-from') + const log = oneTurnLog() + backend.store.set(id, { meta: meta(id), events: log }) + let coordinator!: PersistenceCoordinator + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + + try { + // Happy path through the hook: only the suffix comes back, detached. + backend.seekHook = async (hookId, fromSeq) => { + const entry = backend.store.get(hookId) + if (entry === undefined) return undefined + return { meta: structuredClone(entry.meta), events: entry.events.filter(e => e.seq >= fromSeq) } + } + const suffix = await coordinator.readFrom(id, 3) + expect(suffix.events).toEqual(log.slice(3)) + // The hook's undefined is the seam's not-found. + await expect(coordinator.readFrom(SessionId('missing-seek'), 0)).rejects.toThrow('not found') + + // A hook failure with no cancellation in play propagates as-is. + const hookFailure = new Error('seek backend exploded') + backend.seekHook = () => Promise.reject(hookFailure) + await expect(coordinator.readFrom(id, 0)).rejects.toBe(hookFailure) + + // A hook failure after cancellation surfaces the caller's abort reason, + // not the backend's internal teardown error. The abort fires only once + // the hook is provably entered, so the failure exercises the catch (not + // the pre-invocation throwIfAborted). + const controller = new AbortController() + const reason = new Error('read-from cancelled mid-hook') + let hookEntered = false + backend.seekHook = async (_hookId, _fromSeq, signal) => { + hookEntered = true + await new Promise((resolve) => { signal?.addEventListener('abort', () => { resolve() }, { once: true }) }) + throw new Error('backend teardown after abort') + } + const pending = coordinator.readFrom(id, 0, controller.signal) + const observed = pending.catch((error: unknown) => error) + await vi.waitFor(() => { expect(hookEntered).toBe(true) }) + controller.abort(reason) + expect(await observed).toBe(reason) + } finally { + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + it('rejects a cancelled inspect while an in-flight retirement drain is still pending', async () => { const ctx = new Context() await ctx.plugin(SessionStore) @@ -541,6 +600,66 @@ describe('PersistenceCoordinator retirement', () => { } }) + it('a superseded retirement leaves the successor lifecycle\'s pending drain in place', 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 internals = coordinator as unknown as CoordinatorInternals + const readGate = Promise.withResolvers() + + try { + const id = SessionId('superseded-retirement') + // First lifecycle: unmaterialized (zero events), so a same-id successor + // may legally reclaim the abandoned id later. + let first!: Session + const firstFiber = await ctx.plugin(Object.assign((inner: Context) => { + first = inner.sessions.create(id) + }, { 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 + } + const parked = coordinator.inspect(id).catch((error: unknown) => error) + await vi.waitFor(() => { expect(backend.loadAttempts).toBe(1) }) + + // First retirement queues behind the gate and stays pending. + await firstFiber.dispose() + await vi.waitFor(() => { expect(internals.retirements.has(id)).toBe(true) }) + const firstRetirement = internals.retirements.get(id) + + // Successor lifecycle retires while the first drain is still in flight: + // retire() replaces the map entry synchronously. + const secondFiber = await ctx.plugin(Object.assign((inner: Context) => { + inner.sessions.create(id) + }, { inject: ['sessions'] })) + await secondFiber.dispose() + await vi.waitFor(() => { + expect(internals.retirements.get(id)).not.toBe(firstRetirement) + }) + + // Release the chain: the first drain settles and its forget() must not + // delete the successor's entry (exact-entry guard); the successor's own + // forget() then clears the map. + readGate.resolve(true) + expect(await parked).toBeInstanceOf(Error) // the parked inspect (not found) is observed + await firstRetirement + await vi.waitFor(() => { expect(internals.retirements.has(id)).toBe(false) }) + } finally { + readGate.resolve(true) + await backendFiber.dispose() + await ctx.fiber.dispose() + } + }) + it('a replacement queued before retirement cleanup still collides with the live owner', async () => { const ctx = new Context() await ctx.plugin(SessionStore) diff --git a/packages/session-projection/session-projection-cache/package.json b/packages/session-projection/session-projection-cache/package.json index 5080713412..d06a49def0 100644 --- a/packages/session-projection/session-projection-cache/package.json +++ b/packages/session-projection/session-projection-cache/package.json @@ -21,7 +21,6 @@ "files": [ "lib/index.js", "lib/invariant.js", - "lib/types/**/*.js", "lib/types/**/*.d.ts", "lib/types/**/*.d.ts.map", "src" @@ -46,7 +45,6 @@ "@deepseek-ai/dsh-session-projection": "workspace:^", "@deepseek-ai/dsh-storage": "workspace:^", "@deepseek-ai/dsh-storage-domain": "workspace:^", - "@deepseek-ai/dsh-storage-json": "workspace:^", "cordis": "^4.0.0-rc.7" } } diff --git a/packages/session-projection/session-projection-cache/src/index.ts b/packages/session-projection/session-projection-cache/src/index.ts index 7e217e314e..c598bd1312 100644 --- a/packages/session-projection/session-projection-cache/src/index.ts +++ b/packages/session-projection/session-projection-cache/src/index.ts @@ -186,9 +186,10 @@ export class SessionProjectionCache extends Service { restored = this.ctx.sessionProjections.restore(cached, tail.events, floor) } catch { // The recoverable restore failures: an unrelated record, or a row - // overreaching the stored log end (or predating the floor). All - // resolve identically — discard the cache and refold the full log. - const whole = floor === 0 && related ? tail : await persistence.readFrom(id, 0, signal) + // overreaching the stored log end (or predating the floor). Both imply + // floor > 0 (baseSeq-0 restores never throw and an unrelated record + // still carried a usable watermark), so the full log is a fresh read. + const whole = await persistence.readFrom(id, 0, signal) restored = this.ctx.sessionProjections.restore({}, whole.events, 0) } await this.putSoft(id, identityOf(tail.meta), restored.checkpoint, 'cold-read write-back') @@ -237,11 +238,12 @@ export class SessionProjectionCache extends Service { }, 'sessionProjectionCache.timers') } - /** One fail-soft durable checkpoint: skip when clean, log on failure. */ + /** + * One fail-soft durable checkpoint. Every caller has work by construction: + * the throttle triggers only fire dirty (markClean clears the timer with + * the counter) and the two mandatory points write unconditionally. + */ private async flushSoft(session: Session, trigger: string): Promise { - const state = this.dirty.get(session) - const mandatory = trigger === 'turn/end' || trigger === 'detach' - if (!mandatory && (state === undefined || state.pending === 0)) return try { await this.write(session) } catch (error) { diff --git a/packages/session-projection/session-projection-cache/tests/cache.spec.ts b/packages/session-projection/session-projection-cache/tests/cache.spec.ts index 94406b1803..e572f9a378 100644 --- a/packages/session-projection/session-projection-cache/tests/cache.spec.ts +++ b/packages/session-projection/session-projection-cache/tests/cache.spec.ts @@ -58,7 +58,8 @@ function fakePersistence(logs: Map) { } /** Header shape for cachedSnapshot calls (fake logs stamp createdAt 0, no cwd). */ -const headerOf = (id: SessionId, createdAt = 0) => ({ version: 0, id, createdAt }) +const headerOf = (id: SessionId, createdAt = 0, cwd?: string) => + ({ version: 0, id, createdAt, ...cwd === undefined ? {} : { cwd } }) interface HarnessOptions { pool?: MemoryMediaPool @@ -166,6 +167,39 @@ describe('SessionProjectionCache write policy', () => { expect(storedRows(pool, session.id)?.['cache-test/marks']?.state).toEqual({ marks: ['slow'] }) }) + it('write() on a never-dirty session checkpoints directly and rejects a non-JSON unit state', async () => { + const { ctx, pool } = await harness() + // Never dirtied: no events — write() still lands the init-derived cut. + const clean = ctx.sessions.create(SessionId('clean-write')) + await ctx.sessionProjectionCache.write(clean) + expect(storedRows(pool, clean.id)?.['cache-test/marks']).toEqual({ stateVersion: 1, observedSeq: -1, state: null }) + // A unit whose state violates the plain-JSON contract fails the write loud. + ctx.sessionProjections.register({ + key: 'cache-test/marks2' as never, + schema: { parse: (value: unknown) => value } as never, + init: () => new Map(), + apply: (state: unknown) => state, + view: () => null as never, + stateVersion: 1, + } as never) + await expect(ctx.sessionProjectionCache.write(clean)).rejects.toThrow('not losslessly JSON-serializable') + }) + + it('plugin disposal clears armed interval timers and leaves cleaned sessions alone', async () => { + vi.useFakeTimers() + const { ctx, pool, fiber } = await harness({ config: { writeEveryEvents: 100, writeIntervalMs: 5000 } }) + const armed = ctx.sessions.create(SessionId('armed')) + const cleaned = ctx.sessions.create(SessionId('cleaned')) + mark(armed, ['pending']) // timer armed, no write yet + mark(cleaned, ['done']) + endTurn(cleaned) // mandatory write; markClean leaves {pending: 0, timer: undefined} in the map + await vi.runAllTicks() + await fiber.dispose() + // The armed timer died with the plugin: advancing time writes nothing. + await vi.advanceTimersByTimeAsync(10_000) + expect(storedRows(pool, armed.id)).toBeUndefined() + }) + it('contains a durable write failure: logs a warning, event path unharmed, next write self-heals', async () => { const { ctx, pool } = await harness() const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {}) @@ -281,6 +315,41 @@ describe('SessionProjectionCache cold read', () => { expect(storedRecord(samePool, SessionId('reborn'))?.identity).toEqual({ createdAt: 0 }) }) + it('cachedSnapshot returns undefined when every stored row is version-mismatched', async () => { + const pool = new MemoryMediaPool() + seedRow(pool, 'all-stale', { stateVersion: 99, observedSeq: 4, state: { marks: ['old'] } }) + const { cache } = await harness({ pool }) + expect(cache.cachedSnapshot(headerOf(SessionId('all-stale')))).toBeUndefined() + }) + + it('binds identity on cwd too: a matching cwd serves, a moved session does not', async () => { + const pool = new MemoryMediaPool() + seedRow(pool, 'homed', { stateVersion: 1, observedSeq: 2, state: { marks: ['w'] } }, { createdAt: 0, cwd: '/work' }) + const { cache } = await harness({ pool }) + const id = SessionId('homed') + expect(cache.cachedSnapshot(headerOf(id, 0, '/work'))?.values['cache-test/marks']).toEqual({ marks: ['w'] }) + expect(cache.cachedSnapshot(headerOf(id, 0, '/elsewhere'))).toBeUndefined() + expect(cache.cachedSnapshot(headerOf(id, 0))).toBeUndefined() + }) + + it('dates an empty stored log at -1 in the zero-units topology', async () => { + const pool = new MemoryMediaPool() + const logs = new Map([['empty', [] as SessionEvent[]]]) + const ctx = new Context() + contexts.push(ctx) + await ctx.plugin(Storage) + ctx.storage.backend.register('memory', new MemoryStorageBackend(pool)) + const facility = new DomainFacility(ctx, { backend: 'memory', routes: {} }) + ctx.storage.mount('domain', facility) + ctx.provide('storageDomain', facility) + await ctx.plugin(SessionStore) + await ctx.plugin(SessionProjectionRegistry) + ctx.provide('sessionPersistence', fakePersistence(logs) as never) + await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 }) + await expect(ctx.sessionProjectionCache.coldSnapshot(SessionId('empty'))) + .resolves.toEqual({ asOfSeq: -1, values: {} }) + }) + it('cachedSnapshot serves identity-matching rows with the cut watermark and refuses unrelated ones', async () => { const pool = new MemoryMediaPool() seedRow(pool, 'listed', { stateVersion: 1, observedSeq: 4, state: { marks: ['t'] } }) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 408a990c35..6aa2b57331 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -3493,9 +3493,6 @@ importers: '@deepseek-ai/dsh-storage-domain': specifier: workspace:^ version: link:../../storage/storage-domain - '@deepseek-ai/dsh-storage-json': - specifier: workspace:^ - version: link:../../storage/storage-json cordis: specifier: ^4.0.0-rc.7 version: 4.0.0-rc.7(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@1.0.0-rc.5)