diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index 79c5b5d880..43ba8cd05c 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -821,14 +821,21 @@ function listProjectionsFor( } } -/** Projection baseline for a detached history tail without Agent activation. */ +/** + * Projection baseline for a detached history tail without Agent activation. + * The cache owns the cold-read recipe (seed from the cached checkpoint, + * skip the cached prefix's applies, write the refreshed checkpoint back); + * without the cache the registry restores from init over the full log. + */ function detachedProjectionsFor( ctx: Context, + meta: SessionHeader, events: readonly SessionEvent[], ): SessionProjectionsBlock | undefined { const registry = ctx.get('sessionProjections') if (registry === undefined) return undefined - return registry.restore({}, events, 0).snapshot + const cache = ctx.get('sessionProjectionCache') + return cache?.coldSnapshot(meta, events) ?? registry.restore({}, events, 0).snapshot } /** @@ -1520,7 +1527,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro includeProjections: boolean, ): { events: SessionEvent[]; projections?: SessionProjectionsBlock } { if (source.kind === 'detached') { - const projections = includeProjections ? detachedProjectionsFor(ctx, source.events) : undefined + const projections = includeProjections ? detachedProjectionsFor(ctx, source.header, source.events) : undefined return { events: source.events, ...projections === undefined ? {} : { projections } } } const events = [...source.session.events] @@ -2625,7 +2632,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro header = inspected.meta events = inspected.events projections = beforeSeq === undefined - ? subagentHistoryProjections(ctx, childSessionId, () => detachedProjectionsFor(ctx, inspected.events)) + ? subagentHistoryProjections(ctx, childSessionId, () => detachedProjectionsFor(ctx, inspected.meta, inspected.events)) : undefined } catch (error: unknown) { if (signal?.aborted) { diff --git a/packages/host/apiproxy/tests/api-proxy-cold.spec.ts b/packages/host/apiproxy/tests/api-proxy-cold.spec.ts index 3cc780d300..1b4fead899 100644 --- a/packages/host/apiproxy/tests/api-proxy-cold.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-cold.spec.ts @@ -15,6 +15,7 @@ import { TypertLookupFailure } from '@deepseek-ai/dsh-typert-protocol' import TypertRegistry from '@deepseek-ai/dsh-typert-registry' import { createUserMessage, MessageId } from '@deepseek-ai/dsh-llm' import type { Agent } from '@deepseek-ai/dsh-agent' +import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection' import UserQuestionService from '@deepseek-ai/dsh-user-questions' import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session' import { @@ -103,6 +104,9 @@ describe('sessions.list cold merge', () => { } return undefined }, + // The detached cold-history path calls the cache's coldSnapshot; this + // listing harness never detaches, so it is inert here. + coldSnapshot: () => undefined, } as never) const api = createApiProxy(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' }) @@ -540,6 +544,45 @@ describe('subagent ownership fence', () => { expect(inspect).toHaveBeenCalledTimes(3) }) + it('hands the detached cold history to the cache, which owns the seeded fold and write-back', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + await ctx.plugin(AgentRegistry) + await ctx.plugin(UserQuestionService) + await ctx.plugin(SessionProjectionRegistry) + const sessionId = sid('seeded-cold') + const meta = header(sessionId, 1) + const events = [ + { type: 'user/message', seq: 0, time: 1, data: { content: [{ type: 'text', text: 'm0' }], source: { kind: 'user' } } }, + { type: 'user/message', seq: 1, time: 2, data: { content: [{ type: 'text', text: 'm1' }], source: { kind: 'user' } } }, + { type: 'user/message', seq: 2, time: 3, data: { content: [{ type: 'text', text: 'm2' }], source: { kind: 'user' } } }, + ] as SessionEvent[] + const inspect = vi.fn(() => Promise.resolve({ meta, events })) + ctx.provide('sessionPersistence', { + list: () => Promise.resolve([meta]), + inspect, + locate: () => undefined, + } as never) + // The cache owns the cold-read recipe (seed, prefix-skip, write-back); + // the carrier's only job is to hand it the stored header and the full log. + const coldSnapshot = vi.fn((_meta: SessionHeader, _events: readonly SessionEvent[]) => + ({ asOfSeq: 2, values: { 'test/last-user': { text: 'm2' } } })) + ctx.provide('sessionProjectionCache', { + cachedSnapshot: () => undefined, + coldSnapshot, + } as never) + const api = createApiProxy(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' }) + + const history = await api.sessions.history(request({ sessionId })) + expect(history.result.ok).toBe(true) + if (!history.result.ok) throw new Error('unreachable') + expect(history.result.value.projections?.values['test/last-user']).toEqual({ text: 'm2' }) + // The full log (all events) was handed to the cache's cold snapshot. + expect(coldSnapshot).toHaveBeenCalledTimes(1) + expect(coldSnapshot.mock.calls[0]?.[0]?.id).toBe(sessionId) + expect(coldSnapshot.mock.calls[0]?.[1]?.map(event => event.seq)).toEqual([0, 1, 2]) + }) + it('no longer treats a descriptor-only cold child without origin as subagent-owned', async () => { const ctx = new Context() await ctx.plugin(SessionStore) diff --git a/packages/session/session-projection-cache/src/index.ts b/packages/session/session-projection-cache/src/index.ts index 35b0701492..ecf5982030 100644 --- a/packages/session/session-projection-cache/src/index.ts +++ b/packages/session/session-projection-cache/src/index.ts @@ -131,6 +131,28 @@ export class SessionProjectionCache extends Service { return { asOfSeq, values } } + /** + * Cold-read one session's projections from its complete log. Each unit is + * seeded from the identity-checked cached rows — the registry skips `apply` + * for the already-folded prefix (events at or below the row's `seq`) — and + * the refreshed checkpoint is written back (fail-soft, fire-and-forget), so + * the first cold read creates the cache row and later ones seed from it. + * The caller supplies the complete log in seq order: this service never + * consults the persistence layer. + * @param meta - the stored session header (identity witness). + * @param events - the session's complete log, in seq order. + * @returns the projection cut at the log end. + */ + coldSnapshot(meta: SessionHeader, events: readonly SessionEvent[]): ProjectionSnapshot { + const restored = this.ctx.sessionProjections.restore(this.recordFor(meta.id, identityOf(meta))?.rows ?? {}, events, 0) + // Refresh the row so the next cold read seeds from it; fail-soft and + // fire-and-forget — a failed write-back only costs a longer tail replay. + void this.put(meta.id, identityOf(meta), restored.checkpoint).catch((error: unknown) => { + this.ctx.logger.warn(`session projection cache: cold-read write-back for "${meta.id}" failed (cache stays stale): ${String(error)}`) + }) + return restored.snapshot + } + /** * Durably checkpoint one live session NOW (both mandatory points call * this; tests and carriers may too). The registry cut is snapshotted at diff --git a/packages/session/session-projection-cache/tests/cache.spec.ts b/packages/session/session-projection-cache/tests/cache.spec.ts index c91ad1d438..347b2a6ce3 100644 --- a/packages/session/session-projection-cache/tests/cache.spec.ts +++ b/packages/session/session-projection-cache/tests/cache.spec.ts @@ -35,6 +35,7 @@ declare module '@deepseek-ai/dsh-session-projection/types' { interface SessionProjectionStateMap { 'cache-test/marks': MarksState 'cache-test/marks2': Map + 'cache-test/count': number } interface SessionProjectionMap { 'cache-test/marks': { marks: string[] } @@ -311,3 +312,66 @@ describe('SessionProjectionCache listing read', () => { expect(cache.cachedSnapshot(headerOf(SessionId('malformed')))).toBeUndefined() }) }) + +describe('SessionProjectionCache cold-read seeding', () => { + it('coldSnapshot traverses the full log but applies only the events after each cached watermark', async () => { + const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-')) + roots.push(root) + // A cached row covering the prefix through seq 2 (three applies folded). + await seedRecord(root, 'cold-snap', { + 'cache-test/count': { ver: 1, seq: 2, val: 3 }, + }, { createdAt: 9 }) + const { cache, ctx } = await harness({ root }) + const apply = vi.fn((_state: number, _event: SessionEvent) => 1) + ctx.sessionProjections.register({ + key: 'cache-test/count', + stateSchema: z.number().int().nonnegative(), + init: () => 0, + apply, + stateVersion: 1, + } satisfies ProjectionDefinition<'cache-test/count', number>) + const meta = headerOf(SessionId('cold-snap'), 9) + const events = Array.from({ length: 5 }, (_, seq) => ({ + type: 'cache-test/mark', seq, time: seq, data: { marks: [`m${seq}`] }, + })) as SessionEvent[] + const snapshot = cache.coldSnapshot(meta, events) + // The full log was traversed, but the fold applied only seqs 3 and 4. + expect(apply).toHaveBeenCalledTimes(2) + expect(apply.mock.calls.map(call => call[1].seq)).toEqual([3, 4]) + expect(snapshot.asOfSeq).toBe(4) + // Host-only unit: folded but not served; the refreshed row is written + // back (fail-soft, fire-and-forget) once the write lands. + expect(Object.keys(snapshot.values)).not.toContain('cache-test/count') + await settle() + expect((await storedRows(root, meta.id))?.['cache-test/count']?.seq).toBe(4) + // No cached row yet: the first cold read folds from init over the full + // log and creates the cache row (the `?? {}` seed path). + const fresh = headerOf(SessionId('cold-fresh'), 10) + cache.coldSnapshot(fresh, events) + expect(apply).toHaveBeenCalledTimes(7) // 2 tail + 5 full + await settle() + expect((await storedRows(root, fresh.id))?.['cache-test/count']?.seq).toBe(4) + }) + + it('coldSnapshot write-back is fail-soft: a failed durable write logs and never throws', async () => { + const root = await mkdtemp(join(tmpdir(), 'dsh-projcache-')) + roots.push(root) + const ctx = new Context() + contexts.push(ctx) + await ctx.plugin(Storage) + await ctx.plugin({ name: storageJsonName, inject: storageJsonInject, apply: storageJsonApply, Config: storageJsonConfig }, { root }) + await ctx.plugin({ name: storageDomainName, inject: storageDomainInject, apply: storageDomainApply, Config: storageDomainConfig }, { backend: 'json' }) + await ctx.plugin(SessionStore) + await ctx.plugin(SessionProjectionRegistry) + ctx.sessionProjections.register(marksUnit()) + await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 }) + const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {}) + // A directory where the record document must land makes the write-back + // fail; the cold read itself still succeeds and never throws. + const meta = headerOf(SessionId('cold-fail')) + await mkdir(recordPath(root, meta.id), { recursive: true }) + expect(ctx.sessionProjectionCache.coldSnapshot(meta, [])).toBeDefined() + await settle() + expect(warn).toHaveBeenCalledWith(expect.stringContaining('cold-read write-back for "cold-fail" failed')) + }) +}) diff --git a/packages/session/session-projection/src/index.ts b/packages/session/session-projection/src/index.ts index e457370794..be306821e4 100644 --- a/packages/session/session-projection/src/index.ts +++ b/packages/session/session-projection/src/index.ts @@ -440,10 +440,12 @@ export class SessionProjectionRegistry extends Service { ) } let state = usable ? def.stateSchema.parse(row.val) : def.init() + // The events are seq-ordered: slice off the already-folded prefix + // (events at or below the seed watermark) and fold only the tail. const from = usable ? row.seq : baseSeq - 1 - for (const event of events) { - if (event.seq > from) state = def.apply(state, event) - } + const tailStart = events.findIndex(event => event.seq > from) + const tail = tailStart === -1 ? [] : events.slice(tailStart) + for (const event of tail) state = def.apply(state, event) if (def.wire !== undefined) values[def.key] = def.wire.viewSchema.parse(def.wire.view(state)) refreshed[def.key] = { ver: def.stateVersion, seq: endSeq, val: state } }