From beb9b2a7fa8765d4c5cff03505fd003dad7600f4 Mon Sep 17 00:00:00 2001 From: imccyu Date: Thu, 6 Aug 2026 01:12:17 +0800 Subject: [PATCH] test: close session preparation CI gaps --- .../cordis-inspect-jsdoc/session.jsonl | 2 +- packages/core/agent-loop/tests/resume.spec.ts | 45 +++- packages/core/session/tests/session.spec.ts | 9 + .../session-persistence-sqlite/src/index.ts | 2 +- .../tests/sqlite.spec.ts | 17 ++ .../tests/persistence.spec.ts | 204 +++++++++++++++++- .../tests/preparations.spec.ts | 16 +- .../subagent/tests/continuation.spec.ts | 11 + .../subagent/tests/list-children.spec.ts | 11 + 9 files changed, 305 insertions(+), 12 deletions(-) diff --git a/examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/session.jsonl b/examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/session.jsonl index e82f4f04d8..f18e7f5954 100644 --- a/examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/session.jsonl +++ b/examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/session.jsonl @@ -15,7 +15,7 @@ {"type":"assistant/chunk","seq":13,"time":1785730459883,"data":{"turn":1,"step":1,"chunk":{"type":"finish","reason":{"kind":"tool-calls"}}}} {"type":"assistant/message","seq":14,"time":1785730459883,"data":{"turn":1,"step":1,"message":{"role":"assistant","content":[{"type":"tool-call","id":"inspect-tools-api","name":"cordis_inspect","arguments":"{\"what\":\"api\",\"name\":\"tools\"}"}],"source":{"kind":"model","provider":"deepseek-official","model":"deepseek-v4-flash"},"id":"6b62bed7-113a-4d2e-a6aa-b935a1063ee2"},"usage":{"inputTokens":3,"outputTokens":3}},"sourceEventSeqs":[9,10,11,12,13],"surfaceOp":"append"} {"type":"tool/call","seq":15,"time":1785730459883,"data":{"turn":1,"step":1,"callId":"inspect-tools-api","name":"cordis_inspect","arguments":"{\"what\":\"api\",\"name\":\"tools\"}"}} -{"type":"tool/result","seq":16,"time":1785730459904,"data":{"turn":1,"step":1,"message":{"source":{"kind":"tool","callId":"inspect-tools-api"},"content":[{"type":"tool-result","toolCallId":"inspect-tools-api","content":[{"type":"text","text":"## api\n- tools — Tool registry and execution pipeline.\n /**\n * Register globally or in the calling agent scope. Scoped tools shadow\n * globals; duplicates within one layer and the reserved `run_code` name fail.\n * @param definition - tool schema, execution, and optional finalization/presentation callbacks.\n * @returns the exact disposer that unregisters the tool.\n */\n register(definition: ToolDefinition): () => void\n /**\n * Restrict global tools for the calling agent scope. Empty filters, unknown\n * names, scope-local names, and reserved transport names fail. Restrictions\n * intersect; scoped registrations remain visible.\n * @param filter - global-surface mask: `allow` (keep only) and/or `deny` (remove).\n * @returns the exact disposer that lifts this restriction.\n */\n restrict(filter: ToolRestriction): () => void\n /**\n * Register a monotonic guard after the extensible `tools/pre-execute`\n * waterfall. A plain-context guard applies globally; one registered through\n * `agent.ctx` applies only to that agent. Any matching guard may deny by\n * returning a reason, while no guard can force-allow a call another guard\n * denied. The exact effect disposer is returned for ordered ownership and\n * HMR cleanup.\n * @param guard - synchronous check; a returned string denies the execution.\n * @returns the exact disposer that unregisters the guard.\n */\n guard(guard: ToolGuard): () => void\n /**\n * Look up a tool as one scope sees it (scoped\n * shadows global; a restricted-away global reads as absent). Presenters pass\n * the calling agent so the rendered card matches the definition that\n * actually executed.\n * @param name - the tool name as registered.\n * @param scope - the viewing scope (the agent); omitted = the global view.\n * @returns the definition the scope resolves, or undefined when none is visible.\n */\n get(name: string, scope?: ScopeKey): ToolDefinition | undefined\n /**\n * Project visible definitions onto the allowlisted model-facing schema fields,\n * excluding execution and presentation callbacks.\n * @param scope - the viewing scope (the agent); omitted = the global view.\n * @returns one deep-cloned schema per visible tool.\n */\n schemas(scope?: ScopeKey): ToolSchema[]\n /**\n * Classify a pending call through the caller's visible tool definition. Only\n * an exact `true` is parallel; unknown, hidden, undeclared, invalid, or\n * throwing classifiers are exclusive.\n * @param exec - call name, parsed arguments, and optional agent scope.\n * @returns the fail-closed scheduling mode.\n */\n executionMode(exec: ToolExecutionInput): ToolExecutionMode\n /**\n * Execute through pre-policy, guards, around-dispatch, post-policy,\n * definition-owned content finalization, and final notification. Tool and\n * listener failures resolve as materialized error results; an invisible tool\n * reports `UNKNOWN_TOOL`. The returned outcome is the same lossless, frozen\n * snapshot final observers receive. Cancellation\n * arriving after entry and before final result materialization skips a\n * not-yet-started body with `ABORTED_BEFORE_DISPATCH` or replaces a\n * successful started outcome with `ABORTED`; already-started work is still\n * drained and may retain a tool-owned structured error.\n * @param exec - the typed same-process call input. The registry assigns its\n * correlation token before policy begins.\n * @returns the materialized final result.\n */\n async execute(exec: ToolExecutionInput): Promise\ntype shapes (referenced by the signatures above — read these before assuming a field is a string):\n export interface Agent {\n readonly id: SessionId;\n readonly options: AgentOptions;\n readonly session: Session;\n readonly inbox: Inbox;\n readonly status: AgentStatus;\n readonly ctx: Context;\n cancel(cause: AgentCancelCause, options?: CancelOptions): void;\n whenIdle(): Promise;\n runMaintenance(task: (signal: AbortSignal) => Promise): Promise;\n send(message: UserMessage, target: InboxTarget, wakeup: boolean): void;\n followup(message: UserMessage): void;\n steer(message: UserMessage): void;\n inject(message: UserMessage): void;\n }\n export type AgentCancelCause = {\n readonly kind: 'user';\n } | {\n readonly kind: 'parent';\n } | {\n readonly kind: 'hook';\n readonly reason: string;\n } | {\n readonly kind: 'disposed';\n };\n export interface AgentOptions {\n provider?: string;\n model?: string;\n maxTokens?: number;\n }\n export type AgentStatus = 'idle' | 'running';\n export interface AssistantMessage extends Message {\n readonly role: 'assistant';\n readonly source: ModelMessageSource;\n }\n export interface AssistantProvenance {\n provider: string;\n model: string;\n replayState?: unknown;\n }\n export type Branded = string & {\n readonly [BRAND]: B;\n };\n export type CallId = Branded<'CallId'>;\n export interface CancelOptions {\n keepInbox?: boolean | undefined;\n }\n export interface ContentBlockMap {\n 'text': TextBlock;\n 'reasoning': ReasoningBlock;\n 'tool-call': ToolCallBlock;\n 'tool-result': ToolResultBlock;\n }\n export type ContentBlockType = keyof ContentBlockMap;\n export interface DiffCallView {\n card: 'diff';\n title: string;\n diffs: FileDiff[];\n locations?: FileLocation[];\n }\n export interface DiffResultView {\n card: 'diff';\n title?: string;\n diffs: FileDiff[];\n }\n export interface EpochHeader {\n config: LlmCallConfig;\n adapterDefaults?: LlmCallConfigAdapterDefaults;\n system?: string;\n tools?: ToolSchema[];\n }\n export interface FileDiff {\n path: string;\n oldText: string | null;\n newText: string;\n }\n export interface FileLocation {\n path: string;\n line?: number;\n }\n export type FinishReason = FinishReasonMap[keyof FinishReasonMap];\n export interface FinishReasonMap {\n 'stop': {\n kind: 'stop';\n };\n 'tool-calls': {\n kind: 'tool-calls';\n };\n 'max-tokens': {\n kind: 'max-tokens';\n };\n 'aborted': {\n kind: 'aborted';\n failure: LlmFailure;\n };\n 'error': {\n kind: 'error';\n failure: LlmFailure;\n };\n }\n export interface GenericCallView {\n card: 'generic';\n title: string;\n kind?: ToolCallKind;\n rawInput?: unknown;\n content?: ContentBlock[];\n locations?: FileLocation[];\n }\n export interface GenericResultView {\n card: 'generic';\n title?: string;\n content?: ContentBlock[];\n }\n export class Inbox {\n constructor(private readonly session: Session, private readonly notifications: InboxNotifications);\n get nextTurn(): readonly UserMessage[];\n get nextStep(): readonly UserMessage[];\n get hasPending(): boolean;\n clear(): void;\n claim(target: InboxTarget, turn: number): UserMessage[];\n append(target: InboxTarget, message: UserMessage): void;\n prepend(target: InboxTarget, message: UserMessage): void;\n replace(messageId: MessageId, newMessage: UserMessage): boolean;\n remove(messageId: MessageId): boolean;\n splice(target: InboxTarget, start: number, deleteCount: number, inserted: UserMessage[]): UserMessage[];\n }\n export interface InboxNotifications {\n inserted(message: UserMessage): void;\n discarded(message: UserMessage): void;\n claimed(message: UserMessage, turn: number): void;\n }\n export type InboxTarget = 'next-turn' | 'next-step';\n export interface JsonSchemaNode {\n type?: JsonSchemaType;\n oneOf?: JsonSchemaNode[];\n properties?: Record;\n required?: string[];\n additionalProperties?: boolean;\n items?: JsonSchemaNode;\n enum?: JsonSchemaScalar[];\n const?: JsonSchemaScalar;\n description?: string;\n title?: string;\n default?: JsonValue;\n examples?: JsonValue;\n }\n export type JsonSchemaScalar = string | number | boolean | null;\n export type JsonSchemaType = 'object' | 'array' | 'string' | 'number' | 'integer' | 'boolean' | 'null';\n export type JsonValue = null | boolean | number | string | JsonValue[] | {\n [key: string]: JsonValue;\n };\n export interface LlmCallConfig {\n provider: string;\n model: string;\n reasoningEffort?: ReasoningEffortId;\n temperature?: number;\n maxTokens?: number;\n stop?: string[];\n }\n export interface LlmCallConfigAdapterDefaults {\n reasoningEffort?: true;\n maxTokens?: true;\n }\n export interface LlmFailure {\n readonly message: string;\n readonly code: string;\n readonly status?: number;\n readonly providerRetryAfterMs?: number;\n readonly requestId?: ProviderRequestId;\n }\n export interface Message {\n readonly id: MessageId;\n readonly role: 'system' | 'user' | 'assistant';\n readonly content: ContentBlock[];\n readonly source: MessageSource;\n }\n export type MessageId = Branded<'MessageId'>;\n export type MessageSource = MessageSourceMap[keyof MessageSourceMap];\n export interface MessageSourceMap {\n user: {\n kind: 'user';\n };\n plugin: {\n kind: 'plugin';\n plugin: string;\n };\n model: ModelMessageSource;\n tool: ToolMessageSource;\n }\n export interface ModelMessageSource extends AssistantProvenance {\n kind: 'model';\n }\n export type ProviderRequestId = Branded<'ProviderRequestId'>;\n export interface ReadFileLine {\n number: number;\n text: string;\n }\n export interface ReadResultView {\n card: 'read';\n title?: string;\n path: string;\n offset: number;\n lines: ReadFileLine[];\n totalLines: number;\n lang?: string;\n content?: ContentBlock[];\n }\n export interface ReasoningBlock {\n type: 'reasoning';\n text: string;\n }\n export type ReasoningEffortId = Branded<'ReasoningEffortId'>;\n export interface RequestContext {\n provider: string;\n model: string;\n contextWindow?: number;\n }\n export type RequestHeaderReason = 'initial' | 'resume' | 'change';\n export type ScopeKey = object;\n export interface SearchFileMatches {\n path: string;\n matches: SearchLineMatch[];\n }\n export interface SearchLineMatch {\n lineNumber: number;\n line: string;\n }\n export interface SearchMatchesResultView {\n card: 'search';\n shape: 'matches';\n title?: string;\n files: SearchFileMatches[];\n truncated: boolean;\n total: number;\n }\n export interface SearchPathsResultView {\n card: 'search';\n shape: 'paths';\n title?: string;\n paths: string[];\n truncated: boolean;\n total: number;\n }\n export type SearchResultView = SearchMatchesResultView | SearchPathsResultView;\n export class Session {\n get surface(): SessionSurface;\n readonly header: SessionHeader;\n get id(): SessionId;\n readonly firstLiveSeq: number;\n static create(id: SessionId, seed?: readonly SessionEvent[], header?: SessionHeader): Session;\n get events(): readonly SessionEvent[];\n get seq(): number;\n append(type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [\n opts: SurfaceIntent\n ] : [\n ]): SessionEvent;\n requestHeader(): EpochHeader | undefined;\n requestContext(): RequestContext | undefined;\n deriveMessages(): Message[];\n deriveEventMessage(event: SessionEvent): Message | null;\n }\n export type SessionEvent = {\n [K in SessionEventType]: {\n type: K;\n seq: number;\n time: number;\n data: SessionEventMap[K];\n } & (K extends SurfaceEventType ? {\n sourceEventSeqs?: number[];\n surfaceOp?: SurfaceOp;\n } : object);\n }[T];\n export interface SessionEventMap {\n 'turn/start': {\n turn: number;\n };\n 'turn/end': {\n turn: number;\n reason: TurnEndReason;\n };\n 'step/start': {\n turn: number;\n step: number;\n };\n 'step/end': {\n turn: number;\n step: number;\n };\n 'user/message': UserMessage;\n 'assistant/chunk': {\n turn: number;\n step: number;\n chunk: StreamChunk;\n };\n 'assistant/message': {\n turn: number;\n step: number;\n message: AssistantMessage;\n usage?: TokenUsage;\n };\n 'tool/call': {\n turn: number;\n step: number;\n callId: CallId;\n name: string;\n arguments: string;\n };\n 'tool/result': {\n turn: number;\n step: number;\n message: ToolResultMessage;\n error?: {\n name: string;\n code: string;\n };\n meta?: JsonValue;\n };\n 'todo/write': {\n todos: TodoItem[];\n };\n 'request/header': {\n header: EpochHeader;\n reason: RequestHeaderReason;\n };\n 'request/context': RequestContext;\n 'session/end-seed': Record;\n }\n export type SessionEventType = keyof SessionEventMap;\n export interface SessionHeader {\n readonly version: number;\n readonly id: SessionId;\n readonly createdAt: number;\n readonly cwd?: string;\n readonly parentSession?: SessionId;\n readonly seedLength?: number;\n readonly origin?: 'subagent';\n readonly delegationDepth?: number;\n }\n export type SessionId = Branded<'SessionId'>;\n export interface SessionSurface {\n readonly nodes: readonly number[];\n readonly replaceGeneration: number;\n }\n export type StreamChunk = {\n type: 'block-start';\n index: number;\n blockType: ContentBlockType;\n } | {\n type: 'text-delta';\n index: number;\n text: string;\n } | {\n type: 'reasoning-delta';\n index: number;\n text: string;\n } | {\n type: 'tool-call-delta';\n index: number;\n id: CallId;\n name?: string;\n argumentsDelta: string;\n } | {\n type: 'block-end';\n index: number;\n block: ContentBlock;\n } | {\n type: 'usage';\n usage: TokenUsage;\n } | {\n type: 'finish';\n reason: FinishReason;\n replayState?: unknown;\n };\n export type SurfaceEventType = 'user/message' | 'assistant/message' | 'tool/result';\n export interface SurfaceIntent {\n surfaceOp: SurfaceOp;\n sourceEventSeqs?: number[];\n }\n export type SurfaceOp = 'append' | {\n op: 'replace';\n start: number;\n end: number;\n };\n export interface TerminalCallView {\n card: 'terminal';\n title: string;\n description?: string;\n cwd?: string;\n }\n export interface TerminalResultView {\n card: 'terminal';\n title?: string;\n output?: string;\n exitCode?: number;\n signal?: string;\n }\n export interface TodoItem {\n content: string;\n status: 'pending' | 'in_progress' | 'completed';\n }\n export interface TokenUsage {\n inputTokens: number;\n outputTokens: number;\n cacheReadTokens?: number;\n cacheWriteTokens?: number;\n reasoningTokens?: number;\n }\n export interface ToolCallBlock {\n type: 'tool-call';\n id: CallId;\n name: string;\n arguments: string;\n }\n export type ToolCallKind = 'read' | 'edit' | 'delete' | 'move' | 'search' | 'execute' | 'fetch' | 'other';\n export type ToolCallView = GenericCallView | TerminalCallView | DiffCallView;\n export interface ToolDefinition extends ToolSchema {\n readonly output: ToolOutputDefinition;\n execute(args: unknown, exec: ToolRunContext): Promise;\n finalizeContent?(exec: Readonly, result: Readonly): ContentBlock[] | undefined;\n timeoutMs?: number;\n isConcurrencySafe?(args: unknown): boolean;\n presentCall?(args: unknown): ToolCallView | undefined;\n presentResult?(args: unknown, result: ToolResult): ToolResultView | undefined;\n }\n export interface ToolErrorInfo {\n name: string;\n code: string;\n }\n export interface ToolExecution extends ToolExecutionInput {\n readonly token: ToolExecutionToken;\n }\n export interface ToolExecutionFailure {\n readonly isError: true;\n readonly error: ToolFailure;\n readonly value?: never;\n readonly content: ContentBlock[];\n readonly meta?: JsonValue;\n readonly additionalContexts?: UserMessage[];\n readonly concludesTurn?: never;\n }\n export interface ToolExecutionInput {\n readonly callId: CallId;\n readonly name: string;\n readonly arguments: unknown;\n readonly agent?: Agent;\n readonly parent?: ToolExecutionToken;\n readonly signal: AbortSignal;\n }\n export type ToolExecutionMode = {\n kind: 'parallel';\n } | {\n kind: 'exclusive';\n };\n export type ToolExecutionResult = ToolExecutionSuccess | ToolExecutionFailure;\n export interface ToolExecutionSuccess {\n readonly isError: false;\n readonly value: JsonValue;\n readonly content: ContentBlock[];\n readonly error?: never;\n readonly meta?: JsonValue;\n readonly additionalContexts?: UserMessage[];\n readonly concludesTurn?: true;\n }\n export type ToolExecutionToken = symbol & {\n readonly [toolExecutionTokenBrand]: true;\n };\n export interface ToolFailure {\n message: string;\n info?: ToolErrorInfo;\n }\n export type ToolGuard = (execution: Readonly) => string | undefined;\n export interface ToolMessageSource {\n kind: 'tool';\n callId: CallId;\n }\n export interface ToolOutputDefinition {\n readonly schema: JsonSchemaNode;\n render(args: unknown, value: JsonValue): ContentBlock[];\n presentationMeta?(args: unknown, value: JsonValue): JsonValue;\n }\n export interface ToolRestriction {\n readonly allow?: readonly string[];\n readonly deny?: readonly string[];\n }\n export interface ToolResult {\n content: ContentBlock[];\n isError: boolean;\n meta?: JsonValue;\n }\n export interface ToolResultBlock {\n type: 'tool-result';\n toolCallId: CallId;\n content: ContentBlock[];\n isError?: boolean;\n }\n export interface ToolResultMessage extends Message {\n readonly role: 'user';\n readonly content: [\n ToolResultBlock\n ];\n readonly source: ToolMessageSource;\n }\n export type ToolResultView = GenericResultView | TerminalResultView | DiffResultView | SearchResultView | ReadResultView | WebResultView;\n export interface ToolRunContext extends ToolExecution {\n deferContext(context: UserMessage): void;\n concludeTurn(): void;\n }\n export interface ToolSchema {\n name: string;\n description: string;\n parameters: Record;\n }\n export type TurnEndCancelCause = AgentCancelCause | {\n readonly kind: 'legacy';\n };\n export type TurnEndReason = TurnEndReasonMap[keyof TurnEndReasonMap];\n export interface TurnEndReasonMap {\n completed: {\n kind: 'completed';\n };\n aborted: {\n kind: 'aborted';\n reason: TurnEndCancelCause;\n };\n blocked: {\n kind: 'blocked';\n };\n error: {\n kind: 'error';\n error: LlmFailure;\n };\n 'max-tokens': {\n kind: 'max-tokens';\n };\n interrupted: {\n kind: 'interrupted';\n };\n }\n export interface UserMessage extends Message {\n readonly role: 'user';\n }\n export interface WebFetchResultView {\n card: 'web';\n kind: 'fetch';\n title?: string;\n url: string;\n statusCode: number;\n truncated: boolean;\n }\n export type WebResultView = WebSearchResultView | WebFetchResultView;\n export interface WebSearchResultView {\n card: 'web';\n kind: 'search';\n title?: string;\n sources: WebSource[];\n answer?: string;\n truncated: boolean;\n }\n export interface WebSource {\n url: string;\n title?: string;\n snippet?: string;\n publishedAt?: string;\n }"}],"isError":false}],"role":"user","id":"847bf2e6-59da-4621-946d-06932a78f0ce"}},"sourceEventSeqs":[15],"surfaceOp":"append"} +{"type":"tool/result","seq":16,"time":1785730459904,"data":{"turn":1,"step":1,"message":{"source":{"kind":"tool","callId":"inspect-tools-api"},"content":[{"type":"tool-result","toolCallId":"inspect-tools-api","content":[{"type":"text","text":"## api\n- tools — Tool registry and execution pipeline.\n /**\n * Register globally or in the calling agent scope. Scoped tools shadow\n * globals; duplicates within one layer and the reserved `run_code` name fail.\n * @param definition - tool schema, execution, and optional finalization/presentation callbacks.\n * @returns the exact disposer that unregisters the tool.\n */\n register(definition: ToolDefinition): () => void\n /**\n * Restrict global tools for the calling agent scope. Empty filters, unknown\n * names, scope-local names, and reserved transport names fail. Restrictions\n * intersect; scoped registrations remain visible.\n * @param filter - global-surface mask: `allow` (keep only) and/or `deny` (remove).\n * @returns the exact disposer that lifts this restriction.\n */\n restrict(filter: ToolRestriction): () => void\n /**\n * Register a monotonic guard after the extensible `tools/pre-execute`\n * waterfall. A plain-context guard applies globally; one registered through\n * `agent.ctx` applies only to that agent. Any matching guard may deny by\n * returning a reason, while no guard can force-allow a call another guard\n * denied. The exact effect disposer is returned for ordered ownership and\n * HMR cleanup.\n * @param guard - synchronous check; a returned string denies the execution.\n * @returns the exact disposer that unregisters the guard.\n */\n guard(guard: ToolGuard): () => void\n /**\n * Look up a tool as one scope sees it (scoped\n * shadows global; a restricted-away global reads as absent). Presenters pass\n * the calling agent so the rendered card matches the definition that\n * actually executed.\n * @param name - the tool name as registered.\n * @param scope - the viewing scope (the agent); omitted = the global view.\n * @returns the definition the scope resolves, or undefined when none is visible.\n */\n get(name: string, scope?: ScopeKey): ToolDefinition | undefined\n /**\n * Project visible definitions onto the allowlisted model-facing schema fields,\n * excluding execution and presentation callbacks.\n * @param scope - the viewing scope (the agent); omitted = the global view.\n * @returns one deep-cloned schema per visible tool.\n */\n schemas(scope?: ScopeKey): ToolSchema[]\n /**\n * Classify a pending call through the caller's visible tool definition. Only\n * an exact `true` is parallel; unknown, hidden, undeclared, invalid, or\n * throwing classifiers are exclusive.\n * @param exec - call name, parsed arguments, and optional agent scope.\n * @returns the fail-closed scheduling mode.\n */\n executionMode(exec: ToolExecutionInput): ToolExecutionMode\n /**\n * Execute through pre-policy, guards, around-dispatch, post-policy,\n * definition-owned content finalization, and final notification. Tool and\n * listener failures resolve as materialized error results; an invisible tool\n * reports `UNKNOWN_TOOL`. The returned outcome is the same lossless, frozen\n * snapshot final observers receive. Cancellation\n * arriving after entry and before final result materialization skips a\n * not-yet-started body with `ABORTED_BEFORE_DISPATCH` or replaces a\n * successful started outcome with `ABORTED`; already-started work is still\n * drained and may retain a tool-owned structured error.\n * @param exec - the typed same-process call input. The registry assigns its\n * correlation token before policy begins.\n * @returns the materialized final result.\n */\n async execute(exec: ToolExecutionInput): Promise\ntype shapes (referenced by the signatures above — read these before assuming a field is a string):\n export interface Agent {\n readonly id: SessionId;\n readonly options: AgentOptions;\n readonly session: Session;\n readonly inbox: Inbox;\n readonly status: AgentStatus;\n readonly ctx: Context;\n cancel(cause: AgentCancelCause, options?: CancelOptions): void;\n whenIdle(): Promise;\n runMaintenance(task: (signal: AbortSignal) => Promise): Promise;\n send(message: UserMessage, target: InboxTarget, wakeup: boolean): void;\n followup(message: UserMessage): void;\n steer(message: UserMessage): void;\n inject(message: UserMessage): void;\n }\n export type AgentCancelCause = {\n readonly kind: 'user';\n } | {\n readonly kind: 'parent';\n } | {\n readonly kind: 'hook';\n readonly reason: string;\n } | {\n readonly kind: 'disposed';\n };\n export interface AgentOptions {\n provider?: string;\n model?: string;\n maxTokens?: number;\n }\n export type AgentStatus = 'idle' | 'running';\n export interface AssistantMessage extends Message {\n readonly role: 'assistant';\n readonly source: ModelMessageSource;\n }\n export interface AssistantProvenance {\n provider: string;\n model: string;\n replayState?: unknown;\n }\n export type Branded = string & {\n readonly [BRAND]: B;\n };\n export type CallId = Branded<'CallId'>;\n export interface CancelOptions {\n keepInbox?: boolean | undefined;\n }\n export interface ContentBlockMap {\n 'text': TextBlock;\n 'reasoning': ReasoningBlock;\n 'tool-call': ToolCallBlock;\n 'tool-result': ToolResultBlock;\n }\n export type ContentBlockType = keyof ContentBlockMap;\n export interface DiffCallView {\n card: 'diff';\n title: string;\n diffs: FileDiff[];\n locations?: FileLocation[];\n }\n export interface DiffResultView {\n card: 'diff';\n title?: string;\n diffs: FileDiff[];\n }\n export interface EpochHeader {\n config: LlmCallConfig;\n adapterDefaults?: LlmCallConfigAdapterDefaults;\n system?: string;\n tools?: ToolSchema[];\n }\n export interface FileDiff {\n path: string;\n oldText: string | null;\n newText: string;\n }\n export interface FileLocation {\n path: string;\n line?: number;\n }\n export type FinishReason = FinishReasonMap[keyof FinishReasonMap];\n export interface FinishReasonMap {\n 'stop': {\n kind: 'stop';\n };\n 'tool-calls': {\n kind: 'tool-calls';\n };\n 'max-tokens': {\n kind: 'max-tokens';\n };\n 'aborted': {\n kind: 'aborted';\n failure: LlmFailure;\n };\n 'error': {\n kind: 'error';\n failure: LlmFailure;\n };\n }\n export interface GenericCallView {\n card: 'generic';\n title: string;\n kind?: ToolCallKind;\n rawInput?: unknown;\n content?: ContentBlock[];\n locations?: FileLocation[];\n }\n export interface GenericResultView {\n card: 'generic';\n title?: string;\n content?: ContentBlock[];\n }\n export class Inbox {\n constructor(private readonly session: Session, private readonly notifications: InboxNotifications);\n get nextTurn(): readonly UserMessage[];\n get nextStep(): readonly UserMessage[];\n get hasPending(): boolean;\n clear(): void;\n claim(target: InboxTarget, turn: number): UserMessage[];\n append(target: InboxTarget, message: UserMessage): void;\n prepend(target: InboxTarget, message: UserMessage): void;\n replace(messageId: MessageId, newMessage: UserMessage): boolean;\n remove(messageId: MessageId): boolean;\n splice(target: InboxTarget, start: number, deleteCount: number, inserted: UserMessage[]): UserMessage[];\n }\n export interface InboxNotifications {\n inserted(message: UserMessage): void;\n discarded(message: UserMessage): void;\n claimed(message: UserMessage, turn: number): void;\n }\n export type InboxTarget = 'next-turn' | 'next-step';\n export interface JsonSchemaNode {\n type?: JsonSchemaType;\n oneOf?: JsonSchemaNode[];\n properties?: Record;\n required?: string[];\n additionalProperties?: boolean;\n items?: JsonSchemaNode;\n enum?: JsonSchemaScalar[];\n const?: JsonSchemaScalar;\n description?: string;\n title?: string;\n default?: JsonValue;\n examples?: JsonValue;\n }\n export type JsonSchemaScalar = string | number | boolean | null;\n export type JsonSchemaType = 'object' | 'array' | 'string' | 'number' | 'integer' | 'boolean' | 'null';\n export type JsonValue = null | boolean | number | string | JsonValue[] | {\n [key: string]: JsonValue;\n };\n export interface LlmCallConfig {\n provider: string;\n model: string;\n reasoningEffort?: ReasoningEffortId;\n temperature?: number;\n maxTokens?: number;\n stop?: string[];\n }\n export interface LlmCallConfigAdapterDefaults {\n reasoningEffort?: true;\n maxTokens?: true;\n }\n export interface LlmFailure {\n readonly message: string;\n readonly code: string;\n readonly status?: number;\n readonly providerRetryAfterMs?: number;\n readonly requestId?: ProviderRequestId;\n }\n export interface Message {\n readonly id: MessageId;\n readonly role: 'system' | 'user' | 'assistant';\n readonly content: ContentBlock[];\n readonly source: MessageSource;\n }\n export type MessageId = Branded<'MessageId'>;\n export type MessageSource = MessageSourceMap[keyof MessageSourceMap];\n export interface MessageSourceMap {\n user: {\n kind: 'user';\n };\n plugin: {\n kind: 'plugin';\n plugin: string;\n };\n model: ModelMessageSource;\n tool: ToolMessageSource;\n }\n export interface ModelMessageSource extends AssistantProvenance {\n kind: 'model';\n }\n export type ProviderRequestId = Branded<'ProviderRequestId'>;\n export interface ReadFileLine {\n number: number;\n text: string;\n }\n export interface ReadResultView {\n card: 'read';\n title?: string;\n path: string;\n offset: number;\n lines: ReadFileLine[];\n totalLines: number;\n lang?: string;\n content?: ContentBlock[];\n }\n export interface ReasoningBlock {\n type: 'reasoning';\n text: string;\n }\n export type ReasoningEffortId = Branded<'ReasoningEffortId'>;\n export interface RequestContext {\n provider: string;\n model: string;\n contextWindow?: number;\n }\n export type RequestHeaderReason = 'initial' | 'resume' | 'change';\n export type ScopeKey = object;\n export interface SearchFileMatches {\n path: string;\n matches: SearchLineMatch[];\n }\n export interface SearchLineMatch {\n lineNumber: number;\n line: string;\n }\n export interface SearchMatchesResultView {\n card: 'search';\n shape: 'matches';\n title?: string;\n files: SearchFileMatches[];\n truncated: boolean;\n total: number;\n }\n export interface SearchPathsResultView {\n card: 'search';\n shape: 'paths';\n title?: string;\n paths: string[];\n truncated: boolean;\n total: number;\n }\n export type SearchResultView = SearchMatchesResultView | SearchPathsResultView;\n export class Session {\n get surface(): SessionSurface;\n readonly header: SessionHeader;\n get id(): SessionId;\n readonly firstLiveSeq: number;\n static create(id: SessionId, seed?: readonly SessionEvent[], header?: SessionHeader): Session;\n static fromRestore(id: SessionId, seed: readonly SessionEvent[], header: SessionHeader): Session;\n get events(): readonly SessionEvent[];\n get seq(): number;\n append(type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [\n opts: SurfaceIntent\n ] : [\n ]): SessionEvent;\n requestHeader(): EpochHeader | undefined;\n requestContext(): RequestContext | undefined;\n deriveMessages(): Message[];\n deriveEventMessage(event: SessionEvent): Message | null;\n }\n export type SessionEvent = {\n [K in SessionEventType]: {\n type: K;\n seq: number;\n time: number;\n data: SessionEventMap[K];\n } & (K extends SurfaceEventType ? {\n sourceEventSeqs?: number[];\n surfaceOp?: SurfaceOp;\n } : object);\n }[T];\n export interface SessionEventMap {\n 'turn/start': {\n turn: number;\n };\n 'turn/end': {\n turn: number;\n reason: TurnEndReason;\n };\n 'step/start': {\n turn: number;\n step: number;\n };\n 'step/end': {\n turn: number;\n step: number;\n };\n 'user/message': UserMessage;\n 'assistant/chunk': {\n turn: number;\n step: number;\n chunk: StreamChunk;\n };\n 'assistant/message': {\n turn: number;\n step: number;\n message: AssistantMessage;\n usage?: TokenUsage;\n };\n 'tool/call': {\n turn: number;\n step: number;\n callId: CallId;\n name: string;\n arguments: string;\n };\n 'tool/result': {\n turn: number;\n step: number;\n message: ToolResultMessage;\n error?: {\n name: string;\n code: string;\n };\n meta?: JsonValue;\n };\n 'todo/write': {\n todos: TodoItem[];\n };\n 'request/header': {\n header: EpochHeader;\n reason: RequestHeaderReason;\n };\n 'request/context': RequestContext;\n 'session/end-seed': Record;\n }\n export type SessionEventType = keyof SessionEventMap;\n export interface SessionHeader {\n readonly version: number;\n readonly id: SessionId;\n readonly createdAt: number;\n readonly cwd?: string;\n readonly parentSession?: SessionId;\n readonly seedLength?: number;\n readonly origin?: 'subagent';\n readonly delegationDepth?: number;\n }\n export type SessionId = Branded<'SessionId'>;\n export interface SessionSurface {\n readonly nodes: readonly number[];\n readonly replaceGeneration: number;\n }\n export type StreamChunk = {\n type: 'block-start';\n index: number;\n blockType: ContentBlockType;\n } | {\n type: 'text-delta';\n index: number;\n text: string;\n } | {\n type: 'reasoning-delta';\n index: number;\n text: string;\n } | {\n type: 'tool-call-delta';\n index: number;\n id: CallId;\n name?: string;\n argumentsDelta: string;\n } | {\n type: 'block-end';\n index: number;\n block: ContentBlock;\n } | {\n type: 'usage';\n usage: TokenUsage;\n } | {\n type: 'finish';\n reason: FinishReason;\n replayState?: unknown;\n };\n export type SurfaceEventType = 'user/message' | 'assistant/message' | 'tool/result';\n export interface SurfaceIntent {\n surfaceOp: SurfaceOp;\n sourceEventSeqs?: number[];\n }\n export type SurfaceOp = 'append' | {\n op: 'replace';\n start: number;\n end: number;\n };\n export interface TerminalCallView {\n card: 'terminal';\n title: string;\n description?: string;\n cwd?: string;\n }\n export interface TerminalResultView {\n card: 'terminal';\n title?: string;\n output?: string;\n exitCode?: number;\n signal?: string;\n }\n export interface TodoItem {\n content: string;\n status: 'pending' | 'in_progress' | 'completed';\n }\n export interface TokenUsage {\n inputTokens: number;\n outputTokens: number;\n cacheReadTokens?: number;\n cacheWriteTokens?: number;\n reasoningTokens?: number;\n }\n export interface ToolCallBlock {\n type: 'tool-call';\n id: CallId;\n name: string;\n arguments: string;\n }\n export type ToolCallKind = 'read' | 'edit' | 'delete' | 'move' | 'search' | 'execute' | 'fetch' | 'other';\n export type ToolCallView = GenericCallView | TerminalCallView | DiffCallView;\n export interface ToolDefinition extends ToolSchema {\n readonly output: ToolOutputDefinition;\n execute(args: unknown, exec: ToolRunContext): Promise;\n finalizeContent?(exec: Readonly, result: Readonly): ContentBlock[] | undefined;\n timeoutMs?: number;\n isConcurrencySafe?(args: unknown): boolean;\n presentCall?(args: unknown): ToolCallView | undefined;\n presentResult?(args: unknown, result: ToolResult): ToolResultView | undefined;\n }\n export interface ToolErrorInfo {\n name: string;\n code: string;\n }\n export interface ToolExecution extends ToolExecutionInput {\n readonly token: ToolExecutionToken;\n }\n export interface ToolExecutionFailure {\n readonly isError: true;\n readonly error: ToolFailure;\n readonly value?: never;\n readonly content: ContentBlock[];\n readonly meta?: JsonValue;\n readonly additionalContexts?: UserMessage[];\n readonly concludesTurn?: never;\n }\n export interface ToolExecutionInput {\n readonly callId: CallId;\n readonly name: string;\n readonly arguments: unknown;\n readonly agent?: Agent;\n readonly parent?: ToolExecutionToken;\n readonly signal: AbortSignal;\n }\n export type ToolExecutionMode = {\n kind: 'parallel';\n } | {\n kind: 'exclusive';\n };\n export type ToolExecutionResult = ToolExecutionSuccess | ToolExecutionFailure;\n export interface ToolExecutionSuccess {\n readonly isError: false;\n readonly value: JsonValue;\n readonly content: ContentBlock[];\n readonly error?: never;\n readonly meta?: JsonValue;\n readonly additionalContexts?: UserMessage[];\n readonly concludesTurn?: true;\n }\n export type ToolExecutionToken = symbol & {\n readonly [toolExecutionTokenBrand]: true;\n };\n export interface ToolFailure {\n message: string;\n info?: ToolErrorInfo;\n }\n export type ToolGuard = (execution: Readonly) => string | undefined;\n export interface ToolMessageSource {\n kind: 'tool';\n callId: CallId;\n }\n export interface ToolOutputDefinition {\n readonly schema: JsonSchemaNode;\n render(args: unknown, value: JsonValue): ContentBlock[];\n presentationMeta?(args: unknown, value: JsonValue): JsonValue;\n }\n export interface ToolRestriction {\n readonly allow?: readonly string[];\n readonly deny?: readonly string[];\n }\n export interface ToolResult {\n content: ContentBlock[];\n isError: boolean;\n meta?: JsonValue;\n }\n export interface ToolResultBlock {\n type: 'tool-result';\n toolCallId: CallId;\n content: ContentBlock[];\n isError?: boolean;\n }\n export interface ToolResultMessage extends Message {\n readonly role: 'user';\n readonly content: [\n ToolResultBlock\n ];\n readonly source: ToolMessageSource;\n }\n export type ToolResultView = GenericResultView | TerminalResultView | DiffResultView | SearchResultView | ReadResultView | WebResultView;\n export interface ToolRunContext extends ToolExecution {\n deferContext(context: UserMessage): void;\n concludeTurn(): void;\n }\n export interface ToolSchema {\n name: string;\n description: string;\n parameters: Record;\n }\n export type TurnEndCancelCause = AgentCancelCause | {\n readonly kind: 'legacy';\n };\n export type TurnEndReason = TurnEndReasonMap[keyof TurnEndReasonMap];\n export interface TurnEndReasonMap {\n completed: {\n kind: 'completed';\n };\n aborted: {\n kind: 'aborted';\n reason: TurnEndCancelCause;\n };\n blocked: {\n kind: 'blocked';\n };\n error: {\n kind: 'error';\n error: LlmFailure;\n };\n 'max-tokens': {\n kind: 'max-tokens';\n };\n interrupted: {\n kind: 'interrupted';\n };\n }\n export interface UserMessage extends Message {\n readonly role: 'user';\n }\n export interface WebFetchResultView {\n card: 'web';\n kind: 'fetch';\n title?: string;\n url: string;\n statusCode: number;\n truncated: boolean;\n }\n export type WebResultView = WebSearchResultView | WebFetchResultView;\n export interface WebSearchResultView {\n card: 'web';\n kind: 'search';\n title?: string;\n sources: WebSource[];\n answer?: string;\n truncated: boolean;\n }\n export interface WebSource {\n url: string;\n title?: string;\n snippet?: string;\n publishedAt?: string;\n }"}],"isError":false}],"role":"user","id":"847bf2e6-59da-4621-946d-06932a78f0ce"}},"sourceEventSeqs":[15],"surfaceOp":"append"} {"type":"step/end","seq":17,"time":1785730459904,"data":{"turn":1,"step":1}} {"type":"step/start","seq":18,"time":1785730459916,"data":{"turn":1,"step":2}} {"type":"assistant/chunk","seq":19,"time":1784449176734,"data":{"turn":1,"step":2,"chunk":{"type":"block-start","index":0,"blockType":"tool-call"}}} diff --git a/packages/core/agent-loop/tests/resume.spec.ts b/packages/core/agent-loop/tests/resume.spec.ts index 58c26fa2c1..62f9e9059e 100644 --- a/packages/core/agent-loop/tests/resume.spec.ts +++ b/packages/core/agent-loop/tests/resume.spec.ts @@ -1,5 +1,5 @@ import { createUserMessage } from '@deepseek-ai/dsh-llm' -import { afterEach, describe, expect, it } from 'vitest' +import { afterEach, describe, expect, it, vi } from 'vitest' import { Context } from 'cordis' import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' @@ -747,6 +747,24 @@ describe('creation and resume cancellation edges', () => { await ctx.fiber.dispose() }) + it('rejects when setup synchronously aborts its caller signal', async () => { + const { ctx } = await persistentHarness(new MockAdapter([])) + const controller = new AbortController() + + const creating = ctx.agents.create({ + sessionId: SessionId('setup-synchronous-abort'), + agentOptions: { provider: 'mock', model: 'mock' }, + signal: controller.signal, + setup() { + controller.abort(new Error('setup synchronously cancelled')) + }, + }) + + await expect(promptly(creating)).rejects.toThrow('setup synchronously cancelled') + expect(ctx.agents.get(SessionId('setup-synchronous-abort'))).toBeUndefined() + await ctx.fiber.dispose() + }) + it('resume with a pre-aborted caller signal rejects out of the load race', async () => { const sessionId = SessionId('resume-pre-aborted') const root = await persistSession(sessionId) @@ -760,6 +778,31 @@ describe('creation and resume cancellation edges', () => { signal: controller.signal, }))).rejects.toThrow('resume abandoned') + const stringReason = new AbortController() + stringReason.abort('resume string reason') + await expect(promptly(ctx.agents.resume({ + resumeSessionId: sessionId, + agentOptions: { provider: 'mock', model: 'mock' }, + signal: stringReason.signal, + }))).rejects.toThrow(/creation aborted/) + + expect(ctx.agents.get(sessionId)).toBeUndefined() + await ctx.fiber.dispose() + }) + + it('releases a restored preparation if the loop becomes inactive before setup', async () => { + const sessionId = SessionId('resume-loop-inactive-after-prepare') + const root = await persistSession(sessionId) + const ctx = await mountPersistentHarness(root, new MockAdapter([])) + const loop = ctx.agentLoop as unknown as { + ownership: { isActive: () => boolean } + } + vi.spyOn(loop.ownership, 'isActive').mockReturnValueOnce(false) + + await expect(ctx.agents.resume({ + resumeSessionId: sessionId, + agentOptions: { provider: 'mock', model: 'mock' }, + })).rejects.toThrow('agent loop is not active') expect(ctx.agents.get(sessionId)).toBeUndefined() await ctx.fiber.dispose() }) diff --git a/packages/core/session/tests/session.spec.ts b/packages/core/session/tests/session.spec.ts index 7258726ad1..0b4ef1afe6 100644 --- a/packages/core/session/tests/session.spec.ts +++ b/packages/core/session/tests/session.spec.ts @@ -998,6 +998,15 @@ describe('Session', () => { expect(() => Session.create(SessionId('header-invalid'), undefined, new ExoticHeader())) .toThrow(/not losslessly JSON-serializable/) + expect(() => Session.fromRestore(SessionId('header-invalid'), [], new ExoticHeader())) + .toThrow(/not a plain JSON record/) + for (const header of [null, 1, []]) { + expect(() => Session.fromRestore( + SessionId('header-invalid'), + [], + header as unknown as SessionHeader, + )).toThrow(/not a plain JSON record/) + } expect(() => Session.create(SessionId('header-invalid'), undefined, { version: SESSION_FORMAT_VERSION, id: SessionId('header-invalid'), diff --git a/packages/session-persistence/session-persistence-sqlite/src/index.ts b/packages/session-persistence/session-persistence-sqlite/src/index.ts index c171f5820c..875926cdec 100644 --- a/packages/session-persistence/session-persistence-sqlite/src/index.ts +++ b/packages/session-persistence/session-persistence-sqlite/src/index.ts @@ -109,7 +109,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers // 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, { - preparedSessionCacheSize: config.preparedSessionCacheSize ?? DEFAULT_PREPARED_SESSION_CACHE_SIZE, + preparedSessionCacheSize: (config as Required).preparedSessionCacheSize, }) } 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 ba57c6a1ef..ca8c1cd76a 100644 --- a/packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts +++ b/packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts @@ -641,6 +641,23 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => { }) describe('SessionPersistenceSqlite: edge cases', () => { + it('uses the configured preparation cache through the public service', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const fiber = await ctx.plugin(SessionPersistenceSqlite, { + path: ':memory:', + preparedSessionCacheSize: 1, + }) + const m = meta('sqlite-preparation-cache') + await ctx.sessionPersistence.create(m) + await ctx.sessionPersistence.append(m.id, oneTurnLog()) + + const preparation = await ctx.sessionPersistence.prepare(m.id) + expect(preparation.session.header).toEqual(m) + preparation[Symbol.dispose]() + await fiber.dispose() + }) + it('rejects and closes a current-schema database with an invalid store identity', async () => { const path = await freshDbPath() const db = openDatabase(path, 'wal') diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 58b74df505..65d957311d 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -1,7 +1,7 @@ import { describe, expect, it, vi } from 'vitest' import { Context } from 'cordis' -import SessionStore, { SessionId, isJsonValue } from '@deepseek-ai/dsh-session' -import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session' +import SessionStore, { Session, SessionId, isJsonValue } from '@deepseek-ai/dsh-session' +import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session' import { SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator, type PersistenceBackend, type SessionPersistenceSnapshot, type StoredPrefix, @@ -371,6 +371,162 @@ describe('PersistenceCoordinator stored identity', () => { }) describe('PersistenceCoordinator session preparations', () => { + it.each([0, 1.5])('rejects invalid preparation cache capacity %s', (capacity) => { + const ctx = new Context() + const backend = new ControlledBackend() + + expect(() => new PersistenceCoordinator(ctx, backend, { + preparedSessionCacheSize: capacity, + })).toThrow(/positive safe integer/) + }) + + it('retries invalidated prepare and load reservations', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const prepareId = SessionId('prepare-reservation-retry') + const loadId = SessionId('load-reservation-retry') + backend.store.set(prepareId, { meta: meta(prepareId), events: oneTurnLog() }) + backend.store.set(loadId, { meta: meta(loadId), events: oneTurnLog() }) + let coordinator!: PersistenceCoordinator + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + const preparations = (coordinator as unknown as { + preparations: { reserve: (...args: unknown[]) => Promise } + }).preparations + const reserve = vi.spyOn(preparations, 'reserve') + + try { + reserve.mockResolvedValueOnce(undefined) + const preparation = await coordinator.prepare(prepareId) + preparation[Symbol.dispose]() + + reserve.mockResolvedValueOnce(undefined) + await expect(coordinator.load(loadId)).resolves.toMatchObject({ meta: { id: loadId } }) + } finally { + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + + it('prefers a session that becomes live across preparation reads', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const prepareId = SessionId('prepare-became-live') + const loadId = SessionId('load-became-live') + const inspectId = SessionId('inspect-became-live') + const failedInspectId = SessionId('failed-inspect-became-live') + for (const id of [prepareId, loadId, inspectId]) { + 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'] })) + + try { + const prepareLive = Session.create(prepareId, oneTurnLog(), meta(prepareId)) + const prepareGet = vi.spyOn(ctx.sessions, 'get') + .mockReturnValueOnce(undefined) + .mockReturnValueOnce(prepareLive) + await expect(coordinator.prepare(prepareId)).rejects.toThrow(/while it is live/) + prepareGet.mockRestore() + + const loadLive = Session.create(loadId, oneTurnLog(), meta(loadId)) + const loadGet = vi.spyOn(ctx.sessions, 'get') + .mockReturnValueOnce(undefined) + .mockReturnValueOnce(loadLive) + await expect(coordinator.load(loadId)).resolves.toMatchObject({ meta: { id: loadId } }) + loadGet.mockRestore() + + const inspectLive = Session.create(inspectId, oneTurnLog(), meta(inspectId)) + const inspectGet = vi.spyOn(ctx.sessions, 'get') + .mockReturnValueOnce(undefined) + .mockReturnValueOnce(inspectLive) + await expect(coordinator.inspect(inspectId)).resolves.toMatchObject({ meta: { id: inspectId } }) + inspectGet.mockRestore() + + const failedInspectLive = Session.create(failedInspectId, oneTurnLog(), meta(failedInspectId)) + backend.beforeLoadStored = () => Promise.reject(new Error('load failed')) + const failedInspectGet = vi.spyOn(ctx.sessions, 'get') + .mockReturnValueOnce(undefined) + .mockReturnValueOnce(failedInspectLive) + await expect(coordinator.inspect(failedInspectId)) + .resolves.toMatchObject({ meta: { id: failedInspectId } }) + failedInspectGet.mockRestore() + } finally { + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + + it('rejects a prepared commit when durable state already has a live owner', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('prepared-commit-live-owner') + 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'] })) + const owner = Session.create(id, oneTurnLog(), meta(id)) + const states = (coordinator as unknown as { + states: Map + }).states + states.set(id, { + meta: owner.header, + cursor: oneTurnLog().length, + materialized: true, + owner, + }) + + try { + await expect(coordinator.prepare(id)).rejects.toThrow(/live persistence owner/) + } finally { + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + + it('rejects publication after a preparation state no longer matches', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('prepared-publication-mismatch') + 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'] })) + const preparation = await coordinator.prepare(id) + const preparations = (coordinator as unknown as { + preparations: { + reservationFor: (session: Session) => { state: { cursor: number } } | undefined + } + }).preparations + const reservation = preparations.reservationFor(preparation.session) + if (reservation === undefined) throw new Error('test preparation must stay reserved') + reservation.state.cursor += 1 + const detach = ctx.sessions.enter(preparation.session) + + try { + expect(() => { ctx.sessions.announce(preparation.session) }).toThrow(/no longer matches/) + } finally { + detach() + preparation[Symbol.dispose]() + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + it('reuses the exact Session from inspect through repeated unpublished prepare calls', async () => { const ctx = new Context() await ctx.plugin(SessionStore) @@ -1112,6 +1268,50 @@ describe('PersistenceCoordinator retirement', () => { }) describe('SessionPersistence service registration', () => { + it('provides a cancellation-aware default preparation for simple backends', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const fiber = await ctx.plugin(MemoryPersistence) + const m = meta('default-preparation') + await ctx.sessionPersistence.create(m) + await ctx.sessionPersistence.append(m.id, oneTurnLog()) + const defaultPrepare = SessionPersistence.prototype.prepare.bind(ctx.sessionPersistence) + + const preparation = await defaultPrepare(m.id) + expect(preparation.session.header).toEqual(m) + preparation[Symbol.dispose]() + + const preAborted = new AbortController() + const preAbortReason = new Error('pre-aborted preparation') + preAborted.abort(preAbortReason) + await expect(defaultPrepare(m.id, preAborted.signal)) + .rejects.toBe(preAbortReason) + + const postAborted = new AbortController() + const postAbortReason = new Error('post-load preparation abort') + const originalLoad = ctx.sessionPersistence.load.bind(ctx.sessionPersistence) + ctx.sessionPersistence.load = async (id) => { + const loaded = await originalLoad(id) + postAborted.abort(postAbortReason) + return loaded + } + await expect(defaultPrepare(m.id, postAborted.signal)) + .rejects.toBe(postAbortReason) + + await fiber.dispose() + }) + + it('requires SessionStore for the default preparation', async () => { + const id = SessionId('default-preparation-without-store') + const persistence = { + ctx: new Context(), + load: () => Promise.resolve({ meta: meta(id), events: oneTurnLog() }), + } as unknown as SessionPersistence + + await expect(SessionPersistence.prototype.prepare.call(persistence, id)) + .rejects.toThrow(/SessionStore is not configured/) + }) + it('registers as ctx.sessionPersistence and is removed on fiber dispose (HMR safety)', async () => { const ctx = new Context() await ctx.plugin(SessionStore) diff --git a/packages/session-persistence/session-persistence/tests/preparations.spec.ts b/packages/session-persistence/session-persistence/tests/preparations.spec.ts index dee919c798..8f3ace9aa8 100644 --- a/packages/session-persistence/session-persistence/tests/preparations.spec.ts +++ b/packages/session-persistence/session-persistence/tests/preparations.spec.ts @@ -98,7 +98,7 @@ describe('SessionPreparations reservation', () => { expect(first).toBeDefined() expect(preparations.reservationFor(source.session)).toBe(first) expect(() => preparations.reservationFor(Session.create(id))).toThrow(/cannot publish/) - expect(() => preparations.assertWritable(id)).toThrow(/is reserved/) + expect(() => { preparations.assertWritable(id) }).toThrow(/is reserved/) let secondSettled = false const secondPromise = preparations.reserve(id, () => Promise.resolve(prepared('unused')), committed) @@ -116,10 +116,10 @@ describe('SessionPreparations reservation', () => { expect(second?.source).toBe(source) preparations.attach(second!) expect(preparations.reservationFor(source.session)).toBeUndefined() - expect(() => preparations.attach(second!)).toThrow(/no longer reserved/) + expect(() => { preparations.attach(second!) }).toThrow(/no longer reserved/) preparations.discard(second!) preparations.release(second!, true) - expect(() => preparations.assertWritable(id)).not.toThrow() + expect(() => { preparations.assertWritable(id) }).not.toThrow() }) it('supports abortable reservation waits without cancelling the held reservation', async () => { @@ -152,7 +152,7 @@ describe('SessionPreparations reservation', () => { return commitGate.promise }) await commitStarted.promise - expect(() => preparations.assertWritable(id)).toThrow(/is reserved/) + expect(() => { preparations.assertWritable(id) }).toThrow(/is reserved/) const second = preparations.reserve(id, () => Promise.resolve(prepared('unused')), committed) commitGate.reject(failure) @@ -168,7 +168,7 @@ describe('SessionPreparations reservation', () => { const controller = new AbortController() const reason = new Error('cancel after commit') - await expect(preparations.reserve(id, () => Promise.resolve(source), async value => { + await expect(preparations.reserve(id, () => Promise.resolve(source), async (value) => { controller.abort(reason) return { source: value, state: value.label } }, controller.signal)).rejects.toBe(reason) @@ -185,7 +185,7 @@ describe('SessionPreparations reservation', () => { const commitGate = Promise.withResolvers() const controller = new AbortController() const reason = new Error('cancel invalidated commit') - const reservation = preparations.reserve(id, () => Promise.resolve(source), async value => { + const reservation = preparations.reserve(id, () => Promise.resolve(source), async (value) => { commitStarted.resolve(undefined) await commitGate.promise return { source: value, state: value.label } @@ -235,7 +235,9 @@ describe('observeQueuedAbort', () => { const signal = new AbortController().signal await expect(observeQueuedAbort(Promise.resolve('value'), signal)).resolves.toBe('value') const failure = { kind: 'failed' } - await expect(observeQueuedAbort(Promise.reject(failure), signal)).rejects.toBe(failure) + const rejected = Promise.withResolvers() + rejected.reject(failure) + await expect(observeQueuedAbort(rejected.promise, signal)).rejects.toBe(failure) }) it('rejects promptly with an exact abort reason and ignores later settlement', async () => { diff --git a/packages/subagent/subagent/tests/continuation.spec.ts b/packages/subagent/subagent/tests/continuation.spec.ts index de521bfbd1..38cb358fc5 100644 --- a/packages/subagent/subagent/tests/continuation.spec.ts +++ b/packages/subagent/subagent/tests/continuation.spec.ts @@ -568,6 +568,17 @@ describe('SubagentService.followup residency routing', () => { .rejects.toMatchObject({ code: 'NOT_RESUMABLE' }) }) + it('preserves a SubagentError raised while cold-materializing a child', async () => { + const { ctx, parent } = await setup([textResponse('first')]) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await waitNoActivation(ctx, started.childId) + const failure = new SubagentError('materialization denied', 'UNAUTHORIZED') + ctx.agents.resume = () => Promise.reject(failure) + + await expect(followup(ctx, parent, started.childId, message('continue'))) + .rejects.toBe(failure) + }) + it('cold-resumes a delivery that lost the race with final disposal', async () => { const { ctx, parent } = await setup([textResponse('first'), textResponse('after the race')]) const started = await ctx.subagents.startContinuable(startSpec(parent)) diff --git a/packages/subagent/subagent/tests/list-children.spec.ts b/packages/subagent/subagent/tests/list-children.spec.ts index a6fefc0a74..07098d0387 100644 --- a/packages/subagent/subagent/tests/list-children.spec.ts +++ b/packages/subagent/subagent/tests/list-children.spec.ts @@ -363,6 +363,17 @@ describe('SubagentService.listChildren', () => { expect(entries).toEqual([{ kind: 'diagnostic', id: childId, reason: 'unavailable' }]) }) + it('maps an invalid child surface to corrupt', async () => { + const { ctx, parent } = await setup([textResponse('done')]) + const childId = await startChild(ctx, parent, 'invalid surface') + const query = ctx.get('sessionQuery')! + query.listEvents = () => + Promise.reject(new SessionQueryError('invalid surface', 'SESSION_QUERY_INVALID_SURFACE')) + + const entries = await ctx.subagents.listChildren(parent.id) + expect(entries).toEqual([{ kind: 'diagnostic', id: childId, reason: 'corrupt' }]) + }) + it('diagnoses a read whose header no longer names this parent as corrupt', async () => { const { ctx, parent } = await setup([textResponse('done')]) const childId = await startChild(ctx, parent, 'reparented child')