diff --git a/packages/core/agent-loop/src/index.ts b/packages/core/agent-loop/src/index.ts index 530fbc0a3c..3f77973d92 100644 --- a/packages/core/agent-loop/src/index.ts +++ b/packages/core/agent-loop/src/index.ts @@ -20,7 +20,7 @@ import type { SessionStartSource, } from '@deepseek-ai/dsh-agent' import { errorChain } from '@deepseek-ai/dsh-llm' -import { SessionId } from '@deepseek-ai/dsh-session' +import { SessionId, SessionPreparation } from '@deepseek-ai/dsh-session' import type { Session, SessionHeader } from '@deepseek-ai/dsh-session' import type {} from '@deepseek-ai/dsh-system-prompt' import type {} from '@deepseek-ai/dsh-tools' @@ -104,6 +104,30 @@ async function raceAbort(operation: PromiseLike | T, signal: AbortSignal, } } +/** Start an abortable operation and release a value that arrives after cancellation. */ +async function raceAbortCall( + operation: () => PromiseLike | T, + signal: AbortSignal, + id: SessionId, + releaseAbandoned?: (value: T) => void, +): Promise { + if (signal.aborted) { + throw signal.reason instanceof Error + ? signal.reason + : new Error(`agent "${id}" creation aborted`, { cause: signal.reason }) + } + const pending = Promise.resolve().then(operation) + try { + return await raceAbort(pending, signal, id) + } catch (error: unknown) { + // oxlint-disable-next-line typescript/no-unnecessary-condition -- the signal can abort while the operation is awaited. + if (signal.aborted && releaseAbandoned !== undefined) { + void pending.then(releaseAbandoned, () => undefined) + } + throw error + } +} + /** Resolve the deployment-wide scheduler cap at the owning config boundary. */ function resolveMaxParallelToolCalls(value: number | undefined): number { const maxParallelToolCalls = value ?? DEFAULT_MAX_PARALLEL_TOOL_CALLS @@ -524,8 +548,8 @@ export class AgentLoop extends Service implements AgentFactory { * @returns the published running agent. */ create(id: SessionId, options: AgentOptions = {}, meta: Pick = {}): Agent { - const session = this.runtime.ctx.sessions.prepare(id, { meta }) - const prepared = this.prepare(this.ctx, id, options, session) + using preparation = SessionPreparation.create(this.runtime.ctx.sessions.prepare(id, { meta })) + const prepared = this.prepare(this.ctx, id, options, preparation.session) try { return prepared.publish('startup').agent } catch (error: unknown) { @@ -541,14 +565,14 @@ export class AgentLoop extends Service implements AgentFactory { * @returns the published handle. */ async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise { - const session = this.runtime.ctx.sessions.prepare(options.sessionId, { + const preparation = SessionPreparation.create(this.runtime.ctx.sessions.prepare(options.sessionId, { ...options.seed === undefined ? {} : { seed: options.seed }, ...options.meta === undefined ? {} : { meta: options.meta }, - }) + })) const published = this.setupAndPublish( ownerCtx, options.sessionId, - session, + preparation, options.agentOptions ?? {}, options.setup, options.signal, @@ -562,12 +586,14 @@ export class AgentLoop extends Service implements AgentFactory { private async setupAndPublish( ownerCtx: Context, id: SessionId, - session: Session, + preparation: SessionPreparation, agentOptions: AgentOptions, setup: AgentSetup | undefined, signal: AbortSignal | undefined, source: SessionStartSource, ): Promise { + using ownedPreparation = preparation + const session = ownedPreparation.session const prepared = this.prepare(ownerCtx, id, agentOptions, session, signal) try { const setupCommit = await raceAbort(setup?.(prepared.agent.ctx), prepared.signal, id) @@ -613,26 +639,31 @@ export class AgentLoop extends Service implements AgentFactory { ownerAbort.signal, this.ownership.signal, ]) - let loaded: Awaited> + let preparation: SessionPreparation | undefined try { - loaded = await raceAbort(persistence.load(id), fused, id) + try { + preparation = await raceAbortCall( + () => persistence.prepare(id, fused), + fused, + id, + (abandoned) => { abandoned[Symbol.dispose]() }, + ) + } finally { + await unfollowOwner() + } + ownerCtx.fiber.assertActive() + if (!this.ownership.isActive()) throw new Error('agent loop is not active') + return await this.setupAndPublish( + ownerCtx, + id, + preparation, + options.agentOptions ?? {}, + options.setup, + options.signal, + 'resume', + ) } finally { - await unfollowOwner() - } - ownerCtx.fiber.assertActive() - if (!this.ownership.isActive()) throw new Error('agent loop is not active') - const session = this.runtime.ctx.sessions.prepare(id, { - seed: loaded.events, - meta: loaded.meta, - }) - const prepared = this.prepare(ownerCtx, id, options.agentOptions ?? {}, session, options.signal) - try { - const setupCommit = await raceAbort(options.setup?.(prepared.agent.ctx), prepared.signal, id) - setupCommit?.commit() - return prepared.publish('resume') - } catch (error: unknown) { - await prepared.dispose() - throw error + preparation?.[Symbol.dispose]() } })() this.ownership.trackWrapper(published) diff --git a/packages/core/agent/src/index.ts b/packages/core/agent/src/index.ts index d1cce08d2f..0a16a2bf53 100644 --- a/packages/core/agent/src/index.ts +++ b/packages/core/agent/src/index.ts @@ -187,8 +187,8 @@ export interface AgentFactory { */ createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise /** - * Load a persisted session and resume an agent on it. Async because it awaits - * both `ctx.sessionPersistence.load` and the optional unpublished setup + * Prepare a persisted session and resume an agent on it. Async because it awaits + * both `ctx.sessionPersistence.prepare` and the optional unpublished setup * transaction; must be called after that service exists (consumers inject * `sessionPersistence`). Publication follows the same setup-commit and * ordered boundary as {@link createAgent}. diff --git a/packages/core/session/src/index.ts b/packages/core/session/src/index.ts index 0ae2a75869..ed53fcba29 100644 --- a/packages/core/session/src/index.ts +++ b/packages/core/session/src/index.ts @@ -13,13 +13,15 @@ import { scopeOf, scopeTarget } from '@deepseek-ai/dsh-scope' import type { Scoped } from '@deepseek-ai/dsh-scope' import type { Message } from '@deepseek-ai/dsh-llm' import { SESSION_FORMAT_VERSION, SessionId } from './types.ts' -import type { CreateSessionOptions, EpochHeader, RequestContext, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceIntent, SurfaceEventType } from './types.ts' +import type { CreateSessionOptions, EpochHeader, PrepareSessionOptions, RequestContext, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceIntent, SurfaceEventType } from './types.ts' import { snapshotJsonValue } from './json.ts' import { SurfaceManager } from './surface.ts' import type { SessionSurface } from './surface.ts' import { foldRequestHeader } from './request-header.ts' export * from './types.ts' +export { SessionPreparation } from './preparation.ts' +export type { SessionPreparationOptions } from './preparation.ts' export type { AssistantMessage, ToolResultMessage, UserMessage } from '@deepseek-ai/dsh-llm' export { isJsonValue, snapshotJsonValue } from './json.ts' export type { JsonValue } from './json.ts' @@ -143,6 +145,17 @@ function validateSessionHeader(id: SessionId, input: unknown): SessionHeader { return deepFreeze(record as unknown as SessionHeader) } +/** Validate and freeze one exclusively owned persistence header in place. */ +function validateRestoredSessionHeader(id: SessionId, input: unknown): SessionHeader { + if (input !== null && typeof input === 'object' && !Array.isArray(input)) { + const prototype = Reflect.getPrototypeOf(input) + if (prototype !== Object.prototype && prototype !== null) { + throw new Error('session header is not a plain JSON record') + } + } + return validateSessionHeader(id, input) +} + /** Detach, validate, and freeze the creation metadata published by a session. */ function snapshotSessionHeader(id: SessionId, source?: SessionHeader): SessionHeader { const input: unknown = source === undefined @@ -442,7 +455,28 @@ export class Session { return new Session(id, seed, header) } - private constructor(id: SessionId, seed?: readonly SessionEvent[], header?: SessionHeader) { + /** + * Restore a detached session by taking ownership of fresh persistence values. + * Storage shape, event envelopes, sequence continuity, surface transitions, + * and header fields are validated before the graphs are frozen in place. + * @param id - restored session identity. + * @param seed - fresh detached events whose ownership is transferred. + * @param header - fresh detached metadata whose ownership is transferred. + * @returns a restored detached session. + */ + static fromRestore(id: SessionId, seed: readonly SessionEvent[], header: SessionHeader): Session { + return new Session(id, seed, header, 'restore') + } + + private constructor( + id: SessionId, + seed?: readonly SessionEvent[], + header?: SessionHeader, + mode: 'snapshot' | 'restore' = 'snapshot', + ) { + const restoredHeader = mode === 'restore' + ? validateRestoredSessionHeader(id, header) + : undefined if (seed !== undefined) { // Validate the seed to the SAME invariants `append` enforces, so a // replay/fork (`ctx.sessions.create(id, { seed })`) cannot construct a @@ -454,7 +488,7 @@ export class Session { for (const [index, source] of seed.entries()) { // The seed is a persistence/replay boundary: validate and detach the // complete event in one lossless-JSON pass. - const snapshot = snapshotJsonValue(source) + const snapshot = mode === 'restore' ? source : snapshotJsonValue(source) if (snapshot === undefined) { throw new Error(`seed event at index ${index} is not losslessly JSON-serializable`) } @@ -475,7 +509,7 @@ export class Session { } } this.firstLiveSeq = this.log.length - this.header = snapshotSessionHeader(id, header) + this.header = restoredHeader ?? snapshotSessionHeader(id, header) // Appended here so the marker is already in `events` when a backend // captures the creation seed: no load-time write. Re-marking is skipped // because a cold session is resumed on first touch, so repeatedly opening @@ -822,7 +856,7 @@ export class SessionStore extends Service { * lossless-JSON record with valid scalar fields, or `meta.cwd` is a * non-absolute path. */ - prepare(id?: SessionId, options?: CreateSessionOptions): Session { + prepare(id?: SessionId, options?: PrepareSessionOptions): Session { let sessionId: SessionId if (id === undefined) { do sessionId = SessionId(`session-${++this.counter}`) @@ -831,6 +865,9 @@ export class SessionStore extends Service { sessionId = SessionId(id) } if (this.store.has(sessionId)) throw new Error(`session "${sessionId}" already exists`) + if (options?.seedSource === 'persistence') { + return Session.fromRestore(sessionId, options.seed, options.meta) + } const seed = options?.seed const meta = options?.meta const header: SessionHeader = { diff --git a/packages/core/session/src/preparation.ts b/packages/core/session/src/preparation.ts new file mode 100644 index 0000000000..ee8fe53747 --- /dev/null +++ b/packages/core/session/src/preparation.ts @@ -0,0 +1,49 @@ +/** + * Ownership of one unpublished Session before registry publication. + * @module @deepseek-ai/dsh-session/preparation + */ + +import type { Session } from './index.ts' + +/** Options for a preparation whose provider retains unpublished state. */ +export interface SessionPreparationOptions { + /** Release provider-owned state when the Session was not published. */ + readonly release?: () => void +} + +/** + * One exact unpublished Session and the provider state that keeps it usable. + * Disposal is synchronous and idempotent. Providers decide whether release + * returns the Session to a cache or discards it; publication may consume that + * state before disposal, making the callback a no-op. + */ +export class SessionPreparation implements Disposable { + private released = false + + /** The exact Session to use for setup and publication. */ + readonly session: Session + + private constructor( + session: Session, + private readonly options: SessionPreparationOptions, + ) { + this.session = session + } + + /** + * Wrap an unpublished Session in one preparation lifetime. + * @param session - exact unpublished Session. + * @param options - optional provider release behavior. + * @returns a preparation disposed after publication or rollback. + */ + static create(session: Session, options?: SessionPreparationOptions): SessionPreparation { + return new SessionPreparation(session, options ?? {}) + } + + /** Release provider state once when this preparation leaves its caller. */ + [Symbol.dispose](): void { + if (this.released) return + this.released = true + this.options.release?.() + } +} diff --git a/packages/core/session/src/types.ts b/packages/core/session/src/types.ts index 4e7026e0a3..854e28d2e7 100644 --- a/packages/core/session/src/types.ts +++ b/packages/core/session/src/types.ts @@ -93,6 +93,24 @@ export interface CreateSessionOptions { } } +/** + * Fresh storage values transferred to {@link SessionStore.prepare} without a + * second serialization copy. Callers retain no mutable aliases. + */ +export interface RestoredSessionOptions { + /** Fresh detached storage events to validate and freeze in place. */ + readonly seed: SessionEvent[] + /** Fresh detached storage metadata to validate and freeze in place. */ + readonly meta: SessionHeader + /** Select the persistence ownership-transfer path. */ + readonly seedSource: 'persistence' +} + +/** Inputs accepted while constructing an unpublished Session. */ +export type PrepareSessionOptions = + | (CreateSessionOptions & { readonly seedSource?: undefined }) + | RestoredSessionOptions + /** Why an active agent driver was cancelled. */ export type AgentCancelCause = | { readonly kind: 'user' } diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index a61926fa85..838e4f3a92 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -972,7 +972,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro if (meta === undefined || meta.cwd === undefined) throw new SessionNotFound(`session "${sessionId}" not found`) const inspected = await persistence.inspect(sessionId) if (inspected.meta.cwd === undefined) throw new SessionNotFound(`session "${sessionId}" not found`) - return inspected + return { meta: inspected.meta, events: [...inspected.events] } } /** diff --git a/packages/session-persistence/session-persistence-jsonl/src/index.ts b/packages/session-persistence/session-persistence-jsonl/src/index.ts index 4967010902..c153325139 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/index.ts @@ -13,11 +13,11 @@ import { open, mkdir, readFile, readdir, realpath, link, rm, stat, truncate } fr import { dirname, join, resolve } from 'node:path' import { randomBytes } from 'node:crypto' import { - SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator, + DEFAULT_PREPARED_SESSION_CACHE_SIZE, SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator, type PersistenceBackend, type SessionLocation, type SessionPersistenceSnapshot, - type StoredPrefix, + type SessionInspection, type StoredPrefix, } from '@deepseek-ai/dsh-session-persistence' -import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session' +import type { SessionEvent, SessionId, SessionHeader, SessionPreparation } from '@deepseek-ai/dsh-session' import { encodeSegment, eventLines, logPath, logSuffix, parseHeaderMeta, projectDir, scanLog, sessionDir, toHeaderLine, type JsonlCompression, @@ -56,6 +56,8 @@ export interface Config { packChunks?: boolean /** Physical encoding; defaults to checksummed Zstandard frames. */ compression?: JsonlCompression + /** Maximum cold Session preparations retained for history-to-resume reuse. */ + preparedSessionCacheSize?: number } /** Opaque coordinator token for replacing bytes recovered from a torn frame. */ @@ -82,6 +84,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi root: z.string().required(), packChunks: z.boolean().default(DEFAULT_PACK_CHUNKS), compression: JsonlCompressionSchema, + preparedSessionCacheSize: z.number().step(1).min(1).default(DEFAULT_PREPARED_SESSION_CACHE_SIZE), }) /** @@ -105,7 +108,9 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi this.packChunks = config.packChunks ?? DEFAULT_PACK_CHUNKS this.compression = config.compression ?? DEFAULT_COMPRESSION this.assertUsableRoot() - this.coordinator = new PersistenceCoordinator(this.ctx, this) + this.coordinator = new PersistenceCoordinator(this.ctx, this, { + preparedSessionCacheSize: config.preparedSessionCacheSize ?? DEFAULT_PREPARED_SESSION_CACHE_SIZE, + }) } // Each backend keeps the typed service surface beside its storage hooks; @@ -126,11 +131,15 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi return this.coordinator.append(id, events) } - load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + override prepare(id: SessionId, signal?: AbortSignal): Promise { + return this.coordinator.prepare(id, signal) + } + + load(id: SessionId): Promise { return this.coordinator.load(id) } - inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + inspect(id: SessionId, signal?: AbortSignal): Promise { return this.coordinator.inspect(id, signal) } diff --git a/packages/session-persistence/session-persistence-sqlite/src/index.ts b/packages/session-persistence/session-persistence-sqlite/src/index.ts index 8472b9836c..c171f5820c 100644 --- a/packages/session-persistence/session-persistence-sqlite/src/index.ts +++ b/packages/session-persistence/session-persistence-sqlite/src/index.ts @@ -14,11 +14,11 @@ import { DatabaseSync } from 'node:sqlite' import { mkdir, open } from 'node:fs/promises' import { dirname, resolve } from 'node:path' import { - SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator, + DEFAULT_PREPARED_SESSION_CACHE_SIZE, SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator, type PersistenceBackend, type SessionLocation, type SessionPersistenceSnapshot, - type StoredPrefix, type StoredSuffix, + type SessionInspection, type StoredPrefix, type StoredSuffix, } from '@deepseek-ai/dsh-session-persistence' -import type { SessionEvent, SurfaceEventType, SessionId, SessionHeader } from '@deepseek-ai/dsh-session' +import type { SessionEvent, SurfaceEventType, SessionId, SessionHeader, SessionPreparation } from '@deepseek-ai/dsh-session' import { type JournalMode, openDatabase, rowToMeta, scanRows, type EventRow, type SessionRow, } from './schema.ts' @@ -73,6 +73,8 @@ export interface Config { * (network mounts). See {@link JournalMode}. */ journalMode?: JournalMode + /** Maximum cold Session preparations retained for history-to-resume reuse. */ + preparedSessionCacheSize?: number } /** @@ -86,6 +88,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers static Config: z = z.object({ path: z.string().required(), journalMode: z.union(['wal', 'delete', 'truncate', 'persist'] as const).default('wal'), + preparedSessionCacheSize: z.number().step(1).min(1).default(DEFAULT_PREPARED_SESSION_CACHE_SIZE), }) /** @@ -105,7 +108,9 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers // Open asynchronously so directory creation does not block plugin apply; // every storage hook awaits the same readiness promise. this.ready = this.openDb(config.path, (config as Required).journalMode) - this.coordinator = new PersistenceCoordinator(this.ctx, this) + this.coordinator = new PersistenceCoordinator(this.ctx, this, { + preparedSessionCacheSize: config.preparedSessionCacheSize ?? DEFAULT_PREPARED_SESSION_CACHE_SIZE, + }) } private async openDb(path: string, journalMode: JournalMode): Promise { @@ -153,11 +158,15 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers return this.coordinator.append(id, events) } - load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + override prepare(id: SessionId, signal?: AbortSignal): Promise { + return this.coordinator.prepare(id, signal) + } + + load(id: SessionId): Promise { return this.coordinator.load(id) } - inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + inspect(id: SessionId, signal?: AbortSignal): Promise { return this.coordinator.inspect(id, signal) } diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index 4c6d940888..370f419f6c 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -7,12 +7,26 @@ import { Context } from 'cordis' import { + adoptSessionEvent, interruptedTurnClosers, SESSION_FORMAT_VERSION, + SessionPreparation, snapshotJsonValue, snapshotSessionEvent, } from '@deepseek-ai/dsh-session' import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session' +import type { SessionInspection } from './index.ts' +import { SessionPreparations } from './preparations.ts' +import type { SessionPreparationReservation } from './preparations.ts' + +/** Default number of detached session preparations retained by a coordinator. */ +export const DEFAULT_PREPARED_SESSION_CACHE_SIZE = 5 + +/** Coordinator policy supplied by a concrete persistence backend. */ +export interface PersistenceCoordinatorOptions { + /** Maximum completed unpublished preparations retained for reuse. */ + readonly preparedSessionCacheSize: number +} /** * A stored session's header, valid contiguous event prefix, and optional opaque @@ -55,7 +69,9 @@ export interface PersistenceBackend { * `undefined` if no stored artifact exists. Returned metadata must identify * `id` before repair or state publication. Used by resume/load, live adoption, * and — via `!== undefined` — the create-collision probe. The returned - * `tornMarker` is present iff there is a torn tail to truncate. + * `tornMarker` is present iff there is a torn tail to truncate. Every header + * and event graph must be fresh, mutually unaliased, and unretained by the + * backend because preparation freezes and publishes them in place. * @param id - persisted session id to resolve. * @param signal - optional cancellation for backend read work. */ @@ -138,6 +154,16 @@ interface LiveSessionState { flush: Promise | undefined } +/** One validated cold source and the exact unpublished Session built from it. */ +interface PreparedSessionSource { + readonly inspection: SessionInspection + readonly session: Session + /** Session length after constructor-owned seed markers were appended. */ + readonly sessionLength: number + readonly tornMarker: TornMarker | undefined + readonly closers: readonly SessionEvent[] +} + /** Collect the rejection reasons from a set of promises (none-throwing). */ async function settledErrors(promises: Iterable>): Promise { const settled = await Promise.allSettled([...promises]) @@ -443,6 +469,19 @@ function snapshotStoredEvents(events: readonly SessionEvent[], id: SessionId): S }) } +/** Upgrade and validate an exclusively owned backend result without copying it. */ +function adoptStoredEvents(events: SessionEvent[], id: SessionId): SessionEvent[] { + assertSupportedEvents(events, id) + const messageIds = new Map() + for (const [index, event] of events.entries()) { + const adopted = adoptSessionEvent(migrateLegacyMessageEvent(event, id, messageIds)) + events[index] = adopted + const messageId = eventMessageId(adopted) + if (messageId !== undefined) messageIds.set(adopted.seq, messageId) + } + return events +} + /** * Owns the backend-agnostic session write-path orchestration. A backend * constructs one (`new PersistenceCoordinator(ctx, this)`), implements @@ -463,15 +502,26 @@ export class PersistenceCoordinator { private live = new Map() /** Exact disposed lifecycles whose eager tail is still draining. */ private retirements = new Map>() - /** Cold loads currently reserving an id across backend reads and repair writes. */ - private coldLoads = new Set() + /** Shared cold reads, unpublished reservations, and completed LRU entries. */ + private readonly preparations: SessionPreparations, SessionState> /** * Per-session serialization: every operation chains onto the prior one for the * same id, so writes for one session never interleave. Keyed by session id. */ private chains = new Map>() - constructor(private ctx: Context, private backend: PersistenceBackend) { + constructor( + private ctx: Context, + private backend: PersistenceBackend, + options: PersistenceCoordinatorOptions = { + preparedSessionCacheSize: DEFAULT_PREPARED_SESSION_CACHE_SIZE, + }, + ) { + if (!Number.isSafeInteger(options.preparedSessionCacheSize) + || options.preparedSessionCacheSize < 1) { + throw new TypeError('preparedSessionCacheSize must be a positive safe integer') + } + this.preparations = new SessionPreparations(options.preparedSessionCacheSize) this.installWritePath() } @@ -495,7 +545,7 @@ export class PersistenceCoordinator { private async createCore(meta: SessionHeader): Promise { // Do NOT clobber an existing session: the SessionId IS the identity. - if (this.states.has(meta.id)) { + if (this.states.has(meta.id) || this.preparations.has(meta.id)) { throw new Error(`session "${meta.id}" already exists in this backend`) } // A persisted artifact under this id (in ANY scope) blocks creation: load/ @@ -537,8 +587,9 @@ export class PersistenceCoordinator { // this same backend will refuse to load. assertSupportedEvents(events, id) if (events.length === 0) return + this.preparations.assertWritable(id) let state = this.states.get(id) - if (state === undefined) state = await this.adopt(id) // calls loadCore, not load + if (state === undefined) state = await this.adopt(id) // Contiguity contract: each event's seq must continue the stored log. for (const [i, event] of events.entries()) { @@ -552,68 +603,93 @@ export class PersistenceCoordinator { // cursor as soon as it commits (uniform across backends). state.materialized = true state.cursor += events.length + this.preparations.invalidate(id) } /** - * Reload a session: its {@link SessionHeader} plus the event log up to the last - * durable checkpoint, with any interrupted final turn durably closed (synthetic - * boundary events) during load. - * @param id - the persisted session to reload. - * @returns the header plus the event log, ending on a balanced `turn/end`. + * Prepare and reserve the exact unpublished Session used by resume. + * @param id - persisted session to prepare. + * @param signal - optional cancellation for reading and repair. + * @returns an owned preparation released after publication or rollback. */ - async load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - await this.retirements.get(id) - const selected = await this.serialize(id, async () => { - const live = this.ctx.sessions.get(id) - if (live !== undefined) return { live } - this.coldLoads.add(id) - try { - return { loaded: await this.loadCore(id) } - } finally { - this.coldLoads.delete(id) + async prepare(id: SessionId, signal?: AbortSignal): Promise { + for (;;) { + await this.waitForRetirement(id, signal) + if (this.ctx.sessions.get(id) !== undefined) { + throw new Error(`cannot prepare session "${id}" while it is live`) } - }) - return 'loaded' in selected ? selected.loaded : this.loadLiveSnapshot(selected.live) + const reservation = await this.preparations.reserve( + id, + () => this.serialize(id, () => this.prepareCore(id, signal), signal), + source => this.serialize(id, () => this.commitPrepared(source), signal), + signal, + ) + if (reservation === undefined) continue + if (this.ctx.sessions.get(id) !== undefined) { + this.preparations.release(reservation, false) + throw new Error(`cannot prepare session "${id}" while it is live`) + } + return SessionPreparation.create(reservation.source.session, { + release: () => { + this.preparations.release( + reservation, + reservation.state.owner === undefined + && reservation.source.session.events.length === reservation.source.sessionLength, + ) + }, + }) + } } /** - * Read a detached valid stored prefix without recovery mutations or - * coordinator-state publication. - * @param id - persisted session to inspect. - * @param signal - optional cancellation for queued and backend read work. - * @returns stored header and events before any synthetic recovery closers. + * Commit recovery and return its immutable logical view without publication. + * @param id - persisted session to load. + * @returns prepared header and balanced events. */ - inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - // Waiting for an in-flight retirement drain must honor cancellation too: a - // slow drain would otherwise pin a cancelled inspect until it finishes, - // past the documented boundary. serialize() already races the signal for - // the queued read; do the same for the retirement wait. - const retired = Promise.resolve(this.retirements.get(id)) - const waited = signal === undefined ? retired : observeQueuedAbort(retired, signal, () => false) - return waited.then(() => this.serialize(id, () => this.inspectCore(id, signal), signal)) + async load(id: SessionId): Promise { + for (;;) { + await this.waitForRetirement(id) + const live = this.ctx.sessions.get(id) + if (live !== undefined) return this.loadLiveSnapshot(live) + const reservation = await this.preparations.reserve( + id, + () => this.serialize(id, () => this.prepareCore(id)), + source => this.serialize(id, () => this.commitPrepared(source)), + ) + if (reservation === undefined) continue + const attached = this.ctx.sessions.get(id) + if (attached !== undefined) { + this.preparations.discard(reservation) + return this.loadLiveSnapshot(attached) + } + this.preparations.discard(reservation) + return reservation.source.inspection + } } - private async inspectCore( - id: SessionId, - signal?: AbortSignal, - ): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - signal?.throwIfAborted() - let stored: StoredPrefix | undefined + /** + * Inspect a logical session without publishing it or committing recovery. + * @param id - persisted session to inspect. + * @param signal - optional cancellation for preparation work. + * @returns immutable prepared metadata and balanced events. + */ + async inspect(id: SessionId, signal?: AbortSignal): Promise { + await this.waitForRetirement(id, signal) + const live = this.ctx.sessions.get(id) + if (live !== undefined) return this.inspectLive(live) try { - stored = await this.backend.loadStored(id, signal) + const source = await this.preparations.inspect( + id, + () => this.serialize(id, () => this.prepareCore(id, signal), signal), + signal, + ) + const attached = this.ctx.sessions.get(id) + return attached === undefined ? source.inspection : this.inspectLive(attached) } catch (error: unknown) { - if (signal?.aborted) signal.throwIfAborted() + const attached = this.ctx.sessions.get(id) + if (attached !== undefined) return this.inspectLive(attached) throw error } - signal?.throwIfAborted() - if (stored === undefined) throw new Error(`session "${id}" not found`) - this.assertStoredId(id, stored.meta) - this.assertVersion(stored.meta) - const events = snapshotStoredEvents(stored.events, id) - return { - meta: structuredClone(stored.meta), - events, - } } /** @@ -660,45 +736,124 @@ export class PersistenceCoordinator { } return { meta: structuredClone(suffix.meta), events: snapshotStoredEvents(suffix.events, id) } } - const whole = await this.inspectCore(id, signal) + const whole = await this.readStoredPrefix(id, signal) // Sequential fallback: contiguous seqs from 0 make the suffix an index slice. return { meta: whole.meta, events: whole.events.slice(fromSeq) } } - private async loadCore(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - const stored = await this.backend.loadStored(id) + /** Read one detached physical prefix without logical recovery or caching. */ + private async readStoredPrefix( + id: SessionId, + signal?: AbortSignal, + ): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + signal?.throwIfAborted() + const stored = await this.backend.loadStored(id, signal) + signal?.throwIfAborted() + if (stored === undefined) throw new Error(`session "${id}" not found`) + this.assertStoredId(id, stored.meta) + this.assertVersion(stored.meta) + return { + meta: structuredClone(stored.meta), + events: snapshotStoredEvents(stored.events, id), + } + } + + /** Read, repair in memory, validate, and freeze one cold source once. */ + private async prepareCore( + id: SessionId, + signal?: AbortSignal, + ): Promise> { + signal?.throwIfAborted() + let stored: StoredPrefix | undefined + try { + stored = await this.backend.loadStored(id, signal) + } catch (error: unknown) { + if (signal?.aborted) signal.throwIfAborted() + throw error + } + signal?.throwIfAborted() if (stored === undefined) throw new Error(`session "${id}" not found`) const { meta, events, tornMarker } = stored this.assertStoredId(id, meta) this.assertVersion(meta) - const storedEvents = snapshotStoredEvents(events, id) + const storedEvents = adoptStoredEvents(events, id) // Preserve complete interrupted events and synthesize only missing closers. - const closers = interruptedTurnClosers(storedEvents).map(snapshotSessionEvent) + const closers = interruptedTurnClosers(storedEvents).map(adoptSessionEvent) const balanced = [...storedEvents, ...closers] - - // Repair storage before publishing coordinator state. - if (tornMarker !== undefined || closers.length > 0) { - await this.backend.commitRepair(meta, tornMarker, closers) + const session = this.ctx.sessions.prepare(id, { + seed: balanced, + meta, + seedSource: 'persistence', + }) + const inspection: SessionInspection = Object.freeze({ + meta: session.header, + events: Object.freeze(balanced), + }) + return { + inspection, + session, + sessionLength: session.events.length, + tornMarker, + closers, } - // Keep coordinator metadata detached from the returned record. - this.states.set(id, { meta: { ...meta }, cursor: balanced.length, materialized: true }) - return { meta: structuredClone(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 events = session.events.map(snapshotSessionEvent) + /** Commit one prepared repair and establish its ownerless durable cursor. */ + private async commitPrepared( + source: PreparedSessionSource, + ): Promise<{ source: PreparedSessionSource; state: SessionState }> { + const id = source.inspection.meta.id + const cursor = source.inspection.events.length + const existing = this.states.get(id) + if (existing?.owner !== undefined) { + throw new Error(`session "${id}" already has a live persistence owner`) + } + if (source.tornMarker !== undefined || source.closers.length > 0) { + await this.backend.commitRepair(source.inspection.meta, source.tornMarker, source.closers) + } + const state = existing ?? { + meta: source.inspection.meta, + cursor, + materialized: true, + } + state.meta = source.inspection.meta + state.cursor = cursor + state.materialized = true + this.states.set(id, state) + return { + source: source.tornMarker === undefined && source.closers.length === 0 + ? source + : { ...source, tornMarker: undefined, closers: [] }, + state, + } + } + + /** Return one durable immutable view of an already-live Session. */ + private async loadLiveSnapshot(session: Session): Promise { + const events = session.events await this.flush(session) const state = this.states.get(session.id) /* v8 ignore next -- successful flush always publishes this live session's durable state */ if (state === undefined) throw new Error(`session "${session.id}" lost persistence state during load`) - const meta = structuredClone(state.meta) 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 } + return Object.freeze({ meta: state.meta, events }) + } + + /** Borrow one immutable view from an already-live Session. */ + private inspectLive(session: Session): SessionInspection { + return Object.freeze({ meta: session.header, events: session.events }) + } + + /** Await one retiring lifecycle with caller cancellation. */ + private waitForRetirement(id: SessionId, signal?: AbortSignal): Promise { + const retired = Promise.resolve(this.retirements.get(id)) + return signal === undefined + ? retired + : observeQueuedAbort(retired, signal, () => false) } // Listing is a direct backend read and needs no coordinator state. @@ -738,13 +893,10 @@ export class PersistenceCoordinator { /** Build a state for a session discovered in storage but not yet in memory. */ private async adopt(id: SessionId): Promise { - // loadCore (NOT load) — adopt runs inside an already-serialized op, so - // re-entering the chain via the public load() would deadlock. - await this.loadCore(id) - const state = this.states.get(id) - /* v8 ignore next -- loadCore always sets the state for the id */ - if (!state) throw new Error(`failed to adopt session "${id}"`) - return state + // This runs inside the id's serialization chain, so it uses core helpers + // instead of re-entering through public prepare/load methods. + const source = this.preparations.takeReady(id) ?? await this.prepareCore(id) + return (await this.commitPrepared(source)).state } private assertVersion(meta: SessionHeader): void { @@ -795,9 +947,6 @@ export class PersistenceCoordinator { // Capture the header on creation and persist a fork's seed once. ctx.on('session/created', (session) => { - if (this.coldLoads.has(session.id)) { - throw new Error(`cannot publish session "${session.id}" while its persisted history is loading`) - } void this.initFor(session) }) @@ -847,6 +996,12 @@ export class PersistenceCoordinator { private initFor(session: Session): LiveSessionState { const existing = this.live.get(session) if (existing) return existing + const reservation = this.preparations.reservationFor(session) + if (reservation !== undefined) { + const restored = this.attachPrepared(session, reservation) + this.live.set(session, restored) + return restored + } const seed = session.events.map(e => structuredClone(e)) const live: LiveSessionState = { pending: [], init: Promise.resolve(), flush: undefined } this.live.set(session, live) @@ -855,6 +1010,28 @@ export class PersistenceCoordinator { return live } + /** Bind one exact prepared Session and persist only its unpublished suffix. */ + private attachPrepared( + session: Session, + reservation: SessionPreparationReservation, SessionState>, + ): LiveSessionState { + const { source, state } = reservation + if (source.session !== session || state.owner !== undefined + || state.cursor !== source.inspection.events.length + || session.firstLiveSeq !== state.cursor) { + throw new Error(`session "${session.id}" preparation no longer matches its persistence state`) + } + const suffix = session.events.slice(state.cursor).map(event => structuredClone(event)) + this.preparations.attach(reservation) + state.owner = session + const live: LiveSessionState = { pending: [], init: Promise.resolve(), flush: undefined } + if (suffix.length > 0) { + live.init = this.serialize(session.id, () => this.appendCore(session.id, suffix)) + live.init.catch(() => { /* observed by flush/dispose through the controller */ }) + } + return live + } + /** * Whether a live session's `seed` reproduces the first `cursor` persisted * events. A `cursor` of 0 (nothing persisted yet) trivially matches. Used when @@ -922,7 +1099,7 @@ export class PersistenceCoordinator { // cwd mismatch before repair or state publication. const live = await this.backend.loadStored(id) if (live !== undefined) { - // Do NOT route through loadCore(): that crash-repairs open turns as + // Do NOT route through cold preparation: that crash-repairs open turns as // interrupted, which is wrong for HMR while the live Session is still the // authority and may append the real step/turn end later. await this.adoptLivePrefix(session, seed, live) diff --git a/packages/session-persistence/session-persistence/src/index.ts b/packages/session-persistence/session-persistence/src/index.ts index 5d4f5b616e..35aa8bed2d 100644 --- a/packages/session-persistence/session-persistence/src/index.ts +++ b/packages/session-persistence/session-persistence/src/index.ts @@ -6,6 +6,7 @@ */ import { Context, Service } from 'cordis' +import { SessionPreparation } from '@deepseek-ai/dsh-session' import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session' import type { SessionPersistenceRevision } from './revision.ts' @@ -21,9 +22,22 @@ export interface SessionPersistenceSnapshot { revision: SessionPersistenceRevision } +/** Immutable logical session prepared from persistence or a live owner. */ +export interface SessionInspection { + /** Validated immutable session metadata. */ + readonly meta: SessionHeader + /** Validated contiguous logical event log. */ + readonly events: readonly SessionEvent[] +} + // The backend-agnostic write-path orchestration first-party backends compose. -export { PersistenceCoordinator } from './coordinator.ts' -export type { PersistenceBackend, StoredPrefix, StoredSuffix } from './coordinator.ts' +export { DEFAULT_PREPARED_SESSION_CACHE_SIZE, PersistenceCoordinator } from './coordinator.ts' +export type { + PersistenceBackend, + PersistenceCoordinatorOptions, + StoredPrefix, + StoredSuffix, +} from './coordinator.ts' declare module 'cordis' { interface Context { @@ -83,46 +97,64 @@ export abstract class SessionPersistence extends Service { 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. Implementations - * MUST NOT crash-repair an identity still bound to a live Session: a balanced - * live log may return with its stored header as a durable snapshot, while an - * open live turn rejects. - * A coordinator-backed cold load reserves the identity across storage awaits, - * so concurrent publication of a same-id live Session rejects. - * Returned events are detached, and every identified message is deeply - * frozen. Coordinator-backed implementations upgrade supported pre-identity - * message events before validation; other malformed messages reject before - * any stored event is returned. + * Prepare the exact unpublished Session used by resume. Implementations may + * reuse object graphs retained by an earlier {@link inspect}; disposal + * releases an unpublished reservation. + * @param id - persisted session to prepare. + * @param signal - optional cancellation for preparation work. + * @returns one owned unpublished Session preparation. + */ + async prepare(id: SessionId, signal?: AbortSignal): Promise { + signal?.throwIfAborted() + const loaded = await this.load(id) + signal?.throwIfAborted() + const sessions = this.ctx.get('sessions') + if (sessions === undefined) { + throw new Error('cannot prepare a session: SessionStore is not configured') + } + return SessionPreparation.create(sessions.prepare(id, { + seed: loaded.events.map(event => structuredClone(event)), + meta: structuredClone(loaded.meta), + seedSource: 'persistence', + })) + } + + /** + * Load an immutable balanced logical view and commit any required cold + * recovery. 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. 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. Returned values may be + * shared with immutable live or prepared state and must not be mutated. * @param id - the persisted session to reload. * @returns the header and a log ending on a balanced `turn/end`. */ - abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> + abstract load(id: SessionId): Promise /** - * Inspect a header and its valid contiguous stored prefix without repairing - * a torn tail, closing an interrupted turn, or publishing coordinator state. - * This read is serialized with writes for the same id and returns detached - * values with upgraded, deeply frozen identified messages, so observers - * cannot mutate message identity/content or backend-owned state. Other - * malformed messages reject. + * Inspect an immutable balanced logical session without committing recovery + * or publishing it. A complete interrupted turn receives synthetic closers + * in memory and a torn physical tail remains untouched. Coordinator-backed + * implementations retain the exact unpublished Session for bounded reuse by + * a later {@link prepare}; callers borrow only its immutable header and log. * @param id - the persisted session to inspect. * @param signal - optional cancellation for queued and backend read work. - * @returns the header and valid stored event prefix exactly as observed. + * @returns the validated header and balanced logical event log. */ - abstract inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> + abstract inspect(id: SessionId, signal?: AbortSignal): Promise /** * Read the stored events from `fromSeq` onward — the read-from-seq * primitive for read models that resume from a watermark (e.g. a persisted - * projection cache folding only the tail past its checkpoint). Like - * {@link inspect} it is non-mutating and detached: no torn-tail truncation, - * no synthetic closers, no coordinator-state publication; only events from - * the valid contiguous stored prefix are returned, so a torn fragment never - * reaches the caller. `fromSeq` at or beyond the stored prefix returns an - * empty event list (never an error). Backends whose medium can seek by seq + * projection cache folding only the tail past its checkpoint). Unlike + * {@link inspect}, it is a detached physical suffix read: no preparation + * cache, torn-tail truncation, synthetic closers, or coordinator-state + * publication. Only events from the valid contiguous stored prefix are + * returned, so a torn fragment never reaches the caller. `fromSeq` at or + * beyond the stored prefix returns an empty event list (never an error). + * Backends whose medium can seek by seq * (SQLite) read only the suffix; sequential media (JSONL, both encodings) * still parse the whole artifact and skip forward — the primitive bounds * what is RETURNED and refolded, not every backend's physical read. diff --git a/packages/session-persistence/session-persistence/src/preparations.ts b/packages/session-persistence/session-persistence/src/preparations.ts new file mode 100644 index 0000000000..1df7c7d79b --- /dev/null +++ b/packages/session-persistence/session-persistence/src/preparations.ts @@ -0,0 +1,308 @@ +/** + * Bounded sharing and exclusive reservation of unpublished Sessions. + * @module @deepseek-ai/dsh-session-persistence/preparations + */ + +import type { Session, SessionId } from '@deepseek-ai/dsh-session' + +interface PreparedSource { + readonly session: Session +} + +type PreparationPhase = 'loading' | 'ready' | 'committing' | 'reserved' + +interface PreparationEntry { + readonly id: SessionId + readonly result: Promise + phase: PreparationPhase + source?: Source + reservation?: SessionPreparationReservation + reservationSettled?: Promise + settleReservation?: () => void +} + +/** One exclusively held prepared source and its committed persistence state. */ +export interface SessionPreparationReservation { + readonly entry: PreparationEntry + readonly source: Source + readonly state: CommitState +} + +/** Per-coordinator cold-read sharing, exclusive reservation, and ready-entry LRU. */ +export class SessionPreparations { + private readonly entries = new Map>() + + constructor(private readonly capacity: number) {} + + /** + * Whether this pool currently knows about an unpublished identity. + * @param id - session identity. + * @returns whether an entry exists for the identity. + */ + has(id: SessionId): boolean { + return this.entries.has(id) + } + + /** + * Observe one prepared source, sharing an in-flight read for the same id. + * @param id - session identity. + * @param load - cold loader used when no entry exists. + * @param signal - optional cancellation signal while waiting. + * @returns the shared prepared source. + */ + async inspect( + id: SessionId, + load: () => Promise, + signal?: AbortSignal, + ): Promise { + const { entry, created } = this.entryFor(id, load) + const loaded = signal === undefined || created + ? await entry.result + : await observeQueuedAbort(entry.result, signal) + const source = entry.source ?? loaded + if (this.entries.get(id) === entry && entry.phase === 'ready') this.touch(entry) + return source + } + + /** + * Reserve one ready source after committing its pending durable repair. + * @param id - session identity. + * @param load - cold loader used when no entry exists. + * @param commit - durable repair and cursor-state commit. + * @param signal - optional cancellation signal while waiting. + * @returns the exclusive reservation, or undefined if its entry was invalidated. + */ + async reserve( + id: SessionId, + load: () => Promise, + commit: (source: Source) => Promise<{ source: Source; state: CommitState }>, + signal?: AbortSignal, + ): Promise | undefined> { + const { entry, created } = this.entryFor(id, load) + const loaded = signal === undefined || created + ? await entry.result + : await observeQueuedAbort(entry.result, signal) + while (this.entries.get(id) === entry && entry.phase !== 'ready') { + const settled = entry.reservationSettled + /* v8 ignore next -- committing/reserved transitions install this waiter synchronously. */ + if (settled === undefined) throw new Error(`session "${id}" preparation lost its reservation waiter`) + if (signal === undefined) await settled + else await observeQueuedAbort(settled, signal) + } + if (this.entries.get(id) !== entry) return undefined + const source = entry.source ?? loaded + const reservationSettled = Promise.withResolvers() + entry.phase = 'committing' + entry.reservationSettled = reservationSettled.promise + entry.settleReservation = reservationSettled.resolve + let committed: { source: Source; state: CommitState } + try { + committed = await commit(source) + } catch (error: unknown) { + this.remove(entry) + throw error + } + entry.source = committed.source + try { + signal?.throwIfAborted() + } catch (error: unknown) { + this.makeReady(entry) + throw error + } + const reservation: SessionPreparationReservation = { + entry, + source: committed.source, + state: committed.state, + } + entry.phase = 'reserved' + entry.reservation = reservation + return reservation + } + + /** + * Return the exact reservation for Session publication, rejecting aliases. + * @param session - exact Session candidate for publication. + * @returns its reservation, or undefined when no preparation exists. + */ + reservationFor(session: Session): SessionPreparationReservation | undefined { + const entry = this.entries.get(session.id) + if (entry === undefined) return undefined + if (entry.phase === 'reserved' + && entry.source?.session === session + && entry.reservation !== undefined) { + return entry.reservation + } + throw new Error(`cannot publish session "${session.id}" while a persisted preparation exists`) + } + + /** + * Consume a reservation after its exact Session has attached. + * @param reservation - reservation to consume. + */ + attach(reservation: SessionPreparationReservation): void { + const { entry } = reservation + if (this.entries.get(entry.id) !== entry || entry.reservation !== reservation) { + throw new Error(`session "${entry.id}" preparation is no longer reserved`) + } + this.remove(entry) + } + + /** + * Consume a reservation whose caller only needs the committed inspection. + * @param reservation - reservation to consume. + */ + discard(reservation: SessionPreparationReservation): void { + const { entry } = reservation + if (this.entries.get(entry.id) !== entry || entry.reservation !== reservation) return + this.remove(entry) + } + + /** + * Return a reusable unpublished reservation to the ready LRU. + * @param reservation - reservation to release. + * @param reusable - whether the source remains valid for reuse. + */ + release( + reservation: SessionPreparationReservation, + reusable: boolean, + ): void { + const { entry } = reservation + if (this.entries.get(entry.id) !== entry + || entry.reservation !== reservation + || entry.phase !== 'reserved') return + if (!reusable) { + this.remove(entry) + return + } + delete entry.reservation + this.makeReady(entry) + } + + /** + * Discard a prepared view after the durable log changes. + * @param id - changed session identity. + */ + invalidate(id: SessionId): void { + const entry = this.entries.get(id) + if (entry !== undefined) this.remove(entry) + } + + /** + * Reject writes while an unpublished Session exclusively reserves the id. + * @param id - session identity to check. + */ + assertWritable(id: SessionId): void { + const phase = this.entries.get(id)?.phase + if (phase === 'committing' || phase === 'reserved') { + throw new Error(`cannot append session "${id}" while its persisted preparation is reserved`) + } + } + + /** + * 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. + */ + takeReady(id: SessionId): Source | undefined { + const entry = this.entries.get(id) + if (entry === undefined) return undefined + if (entry.phase !== 'ready' || entry.source === undefined) { + throw new Error(`cannot adopt session "${id}" while its preparation is pending`) + } + this.remove(entry) + return entry.source + } + + private entryFor( + id: SessionId, + load: () => Promise, + ): { entry: PreparationEntry; created: boolean } { + const existing = this.entries.get(id) + if (existing !== undefined) return { entry: existing, created: false } + const result = Promise.resolve().then(load) + const entry: PreparationEntry = { id, result, phase: 'loading' } + this.entries.set(id, entry) + void result.then((source) => { + if (this.entries.get(id) !== entry) return + entry.source = source + entry.phase = 'ready' + }, () => { + this.remove(entry) + }) + return { entry, created: true } + } + + private makeReady(entry: PreparationEntry): void { + if (this.entries.get(entry.id) !== entry) return + entry.phase = 'ready' + const settle = entry.settleReservation + delete entry.reservationSettled + delete entry.settleReservation + settle?.() + this.touch(entry) + } + + private remove(entry: PreparationEntry): void { + if (this.entries.get(entry.id) !== entry) return + this.entries.delete(entry.id) + const settle = entry.settleReservation + delete entry.reservationSettled + delete entry.settleReservation + settle?.() + } + + private touch(entry: PreparationEntry): void { + if (this.entries.get(entry.id) !== entry || entry.phase !== 'ready') return + this.entries.delete(entry.id) + this.entries.set(entry.id, entry) + let readyCount = 0 + for (const candidate of this.entries.values()) { + if (candidate.phase === 'ready') readyCount += 1 + } + if (readyCount <= this.capacity) return + for (const [id, candidate] of this.entries) { + if (candidate.phase !== 'ready') continue + this.entries.delete(id) + readyCount -= 1 + if (readyCount <= this.capacity) break + } + } +} + +/** Give a queued observer a prompt cancellation view without cancelling shared work. */ +function observeQueuedAbort(operation: Promise, signal: AbortSignal): Promise { + return new Promise((resolve, reject) => { + let settled = false + const finish = (callback: () => void): void => { + if (settled) return + settled = true + signal.removeEventListener('abort', onAbort) + callback() + } + const onAbort = (): void => { + finish(() => { + try { + signal.throwIfAborted() + } catch (reason: unknown) { + rejectPreparationObservation(reject, reason) + return + } + /* v8 ignore next -- a native AbortSignal emits abort only after becoming aborted. */ + reject(new Error('preparation observation abort event lacked an aborted signal')) + }) + } + signal.addEventListener('abort', onAbort, { once: true }) + operation.then( + (value) => { finish(() => { resolve(value) }) }, + (reason: unknown) => { + finish(() => { rejectPreparationObservation(reject, reason) }) + }, + ) + if (signal.aborted) onAbort() + }) +} + +/** Preserve an exact loader or AbortSignal reason, including legacy non-Error values. */ +function rejectPreparationObservation(reject: (reason?: unknown) => void, reason: unknown): void { + reject(reason) +} diff --git a/packages/subagent/subagent/src/continuation.ts b/packages/subagent/subagent/src/continuation.ts index 2abc52bb29..869656a550 100644 --- a/packages/subagent/subagent/src/continuation.ts +++ b/packages/subagent/subagent/src/continuation.ts @@ -688,7 +688,7 @@ export class SubagentContinuationManager { } /** - * Cold-resume a persisted child: load and authorize its Session, fold the + * Cold-resume a persisted child: inspect and authorize its Session, fold the * generic descriptor, create the Activation through `ctx.agents.resume()`, * and submit the waiting turn. This never dispatches through a subagent * provider — the persisted Session already holds the initial prefix and the @@ -701,13 +701,12 @@ export class SubagentContinuationManager { options: SubagentFollowupOptions, ): Promise { const persistence = this.requirePersistence() - let loaded: Awaited> + let loaded: Awaited> try { - loaded = await persistence.load(childId) + loaded = await persistence.inspect(childId, options.signal) } catch (error: unknown) { throw new SubagentError(`subagent "${childId}" is unavailable`, 'NOT_RESUMABLE', { cause: error }) } - // The persistence seam takes no signal; recheck before any child work. options.signal.throwIfAborted() this.assertAdmitting(parent) // Authorize the persisted header before folding: only the durable child's @@ -724,17 +723,24 @@ export class SubagentContinuationManager { 'NOT_RESUMABLE', ) } - const activation = await this.materialize({ - childId, - provider: descriptor.provider, - parent, - agentOptions: { - ...descriptor.agentProvider !== undefined ? { provider: descriptor.agentProvider } : {}, - ...descriptor.agentModel !== undefined ? { model: descriptor.agentModel } : {}, - }, - composition: { persona: descriptor.persona, toolFilter: descriptor.toolFilter }, - signal: options.signal, - }) + let activation: Activation + try { + activation = await this.materialize({ + childId, + provider: descriptor.provider, + parent, + agentOptions: { + ...descriptor.agentProvider !== undefined ? { provider: descriptor.agentProvider } : {}, + ...descriptor.agentModel !== undefined ? { model: descriptor.agentModel } : {}, + }, + composition: { persona: descriptor.persona, toolFilter: descriptor.toolFilter }, + signal: options.signal, + }) + } catch (error: unknown) { + options.signal.throwIfAborted() + if (error instanceof SubagentError) throw error + throw new SubagentError(`subagent "${childId}" is unavailable`, 'NOT_RESUMABLE', { cause: error }) + } return this.submitMaterialized(activation, content, options.source, parent, options.signal) }