diff --git a/.changeset/generic-interrupt-resume-hardening.md b/.changeset/generic-interrupt-resume-hardening.md new file mode 100644 index 000000000..425282930 --- /dev/null +++ b/.changeset/generic-interrupt-resume-hardening.md @@ -0,0 +1,9 @@ +--- +'@tanstack/ai': patch +'@tanstack/ai-client': patch +'@tanstack/ai-persistence': patch +--- + +Harden first-party generic interrupt resume. + +Ephemeral continuation now rehydrates an already-parsed display payload instead of running `payloadSchema` again, so transforming schemas keep working. Invalid `expiresAt` values fail closed, binding parse uses one reader, and sequential interrupt-store writes preflight before changing records. diff --git a/packages/ai-client/src/interrupt-manager.ts b/packages/ai-client/src/interrupt-manager.ts index fbfef831c..90144c35c 100644 --- a/packages/ai-client/src/interrupt-manager.ts +++ b/packages/ai-client/src/interrupt-manager.ts @@ -10,6 +10,7 @@ import { hashSchemaInput, isStandardSchema, normalizeApprovalSchema, + readInterruptBinding, wrapGenericInterruptContinuation, } from '@tanstack/ai/client' import type { @@ -197,131 +198,10 @@ function isLegacyInterruptMetadata(interrupt: Interrupt): boolean { ) } -function isBindingBase(value: UnknownObject): boolean { - return ( - // A binding stamped with a version we don't know is another producer's. - // Reject it whole; never read our fields out of it. Missing `v` is read as - // the current version so pre-versioning bindings still resume. - (value['v'] === undefined || value['v'] === INTERRUPT_BINDING_VERSION) && - typeof value['kind'] === 'string' && - typeof value['interruptId'] === 'string' && - typeof value['interruptedRunId'] === 'string' && - typeof value['generation'] === 'number' && - Number.isInteger(value['generation']) && - value['generation'] >= 0 && - (value['expiresAt'] === undefined || - (typeof value['expiresAt'] === 'string' && - Number.isFinite(Date.parse(value['expiresAt'])))) - ) -} - -function readBinding(value: unknown): InterruptBinding | undefined { - if (!isUnknownObject(value) || !isBindingBase(value)) return undefined - const expiresAt = - typeof value['expiresAt'] === 'string' ? value['expiresAt'] : undefined - if (value['kind'] === 'generic') { - if ( - value['responseSchemaHash'] !== undefined && - typeof value['responseSchemaHash'] !== 'string' - ) { - return undefined - } - const firstPartyFields = [ - value['definitionId'], - value['key'], - value['batchIndex'], - value['payloadSchemaHash'], - ] - const hasFirstPartyFields = firstPartyFields.some( - (field) => field !== undefined, - ) - if ( - hasFirstPartyFields && - (typeof value['definitionId'] !== 'string' || - typeof value['key'] !== 'string' || - typeof value['batchIndex'] !== 'number' || - !Number.isInteger(value['batchIndex']) || - value['batchIndex'] < 0 || - (value['payloadSchemaHash'] !== undefined && - typeof value['payloadSchemaHash'] !== 'string')) - ) { - return undefined - } - return { - v: INTERRUPT_BINDING_VERSION, - kind: 'generic', - interruptId: String(value['interruptId']), - interruptedRunId: String(value['interruptedRunId']), - generation: Number(value['generation']), - ...(typeof value['responseSchemaHash'] === 'string' - ? { responseSchemaHash: value['responseSchemaHash'] } - : {}), - ...(expiresAt !== undefined ? { expiresAt } : {}), - ...(hasFirstPartyFields - ? { - definitionId: String(value['definitionId']), - key: String(value['key']), - batchIndex: Number(value['batchIndex']), - ...(typeof value['payloadSchemaHash'] === 'string' - ? { payloadSchemaHash: value['payloadSchemaHash'] } - : {}), - } - : {}), - } - } - if ( - value['kind'] === 'client-tool-execution' && - typeof value['toolName'] === 'string' && - typeof value['toolCallId'] === 'string' && - typeof value['outputSchemaHash'] === 'string' && - typeof value['responseSchemaHash'] === 'string' - ) { - return { - v: INTERRUPT_BINDING_VERSION, - kind: 'client-tool-execution', - interruptId: String(value['interruptId']), - interruptedRunId: String(value['interruptedRunId']), - generation: Number(value['generation']), - toolName: value['toolName'], - toolCallId: value['toolCallId'], - outputSchemaHash: value['outputSchemaHash'], - responseSchemaHash: String(value['responseSchemaHash']), - ...(expiresAt !== undefined ? { expiresAt } : {}), - } - } - if ( - value['kind'] === 'tool-approval' && - typeof value['toolName'] === 'string' && - typeof value['toolCallId'] === 'string' && - typeof value['inputSchemaHash'] === 'string' && - typeof value['approvalSchemaHash'] === 'string' && - typeof value['responseSchemaHash'] === 'string' && - 'originalArgs' in value - ) { - return { - v: INTERRUPT_BINDING_VERSION, - kind: 'tool-approval', - interruptId: String(value['interruptId']), - interruptedRunId: String(value['interruptedRunId']), - generation: Number(value['generation']), - toolName: value['toolName'], - toolCallId: value['toolCallId'], - originalArgs: value['originalArgs'], - inputSchemaHash: value['inputSchemaHash'], - approvalSchemaHash: value['approvalSchemaHash'], - responseSchemaHash: String(value['responseSchemaHash']), - ...(expiresAt !== undefined ? { expiresAt } : {}), - } - } - return undefined -} - function getDescriptorBinding( interrupt: Interrupt, ): InterruptBinding | undefined { - const candidate: unknown = - interrupt.metadata?.[INTERRUPT_BINDING_METADATA_KEY] - return readBinding(candidate) + return readInterruptBinding(interrupt) } function hasReservedFirstPartyBindingMarker(interrupt: Interrupt): boolean { diff --git a/packages/ai-persistence/src/middleware.ts b/packages/ai-persistence/src/middleware.ts index a0cb19c2b..dfad7ea81 100644 --- a/packages/ai-persistence/src/middleware.ts +++ b/packages/ai-persistence/src/middleware.ts @@ -486,13 +486,31 @@ async function applyPendingResumes( await interrupts.commitBatch(entries) return } - for (const interrupt of pending) { - const entry = resumeByInterruptId.get(interrupt.interruptId) - if (!entry) continue + const ids = new Set() + for (const entry of entries) { + if (ids.has(entry.interruptId)) { + throw new Error( + `Interrupt batch contains duplicate id: ${entry.interruptId}.`, + ) + } + ids.add(entry.interruptId) + const existing = await interrupts.get(entry.interruptId) + if (!existing) { + throw new Error( + `Interrupt batch references missing id: ${entry.interruptId}.`, + ) + } + if (existing.status !== 'pending') { + throw new Error( + `Interrupt batch references non-pending id: ${entry.interruptId}.`, + ) + } + } + for (const entry of entries) { if (entry.status === 'resolved') { - await interrupts.resolve(interrupt.interruptId, entry.payload) + await interrupts.resolve(entry.interruptId, entry.response) } else { - await interrupts.cancel(interrupt.interruptId) + await interrupts.cancel(entry.interruptId) } } } diff --git a/packages/ai-persistence/tests/interrupts.test.ts b/packages/ai-persistence/tests/interrupts.test.ts index a123e23cf..7424f682c 100644 --- a/packages/ai-persistence/tests/interrupts.test.ts +++ b/packages/ai-persistence/tests/interrupts.test.ts @@ -1235,6 +1235,104 @@ describe('interrupt persistence', () => { ).toBe('resolved') }) + it('rejects a persisted generic interrupt whose definition hash drifted', async () => { + const persistence = memoryPersistence() + const review = defineInterrupt({ + id: 'persisted-review', + payloadSchema: transformedDisplaySchema, + responseSchema: coercedCountSchema, + }) + const first = mockAdapter([[runStarted(), runFinished('r1')]]) + await collect( + chat({ + adapter: first.adapter, + interrupts: [review], + messages: [{ role: 'user', content: 'hi' }], + runId: 'r1', + threadId: 't1', + middleware: [ + defineChatMiddleware({ + onInterruptBoundary(ctx) { + if (ctx.phase !== 'afterModel') return + return { + interrupts: [ + review.interrupt({ + key: 'one', + payload: 'Review this plan', + reason: 'review', + message: 'Review this plan', + }), + ], + } + }, + }), + withPersistence(persistence), + ], + }) as AsyncIterable, + ) + const interruptId = ( + await persistence.stores.interrupts!.listPending('t1') + )[0]?.interruptId + expect(interruptId).toBeDefined() + if (!interruptId) throw new Error('Expected a persisted generic interrupt') + + const driftedResponseSchema = { + '~standard': { + version: 1, + vendor: 'test', + validate(value: unknown) { + return value && + typeof value === 'object' && + !Array.isArray(value) && + 'approved' in value && + typeof value.approved === 'boolean' + ? { value: { approved: value.approved } } + : { issues: [{ message: 'approved is required' }] } + }, + jsonSchema: { + input() { + return { + type: 'object', + required: ['approved'], + properties: { approved: { type: 'boolean' } }, + } + }, + }, + }, + } as const + const drifted = defineInterrupt({ + id: 'persisted-review', + payloadSchema: transformedDisplaySchema, + responseSchema: driftedResponseSchema, + }) + const resumed = mockAdapter([[runStarted(), text('SHOULD NOT RUN')]]) + const chunks = await collect( + chat({ + adapter: resumed.adapter, + interrupts: [drifted], + messages: [], + runId: 'r1', + threadId: 't1', + resume: [ + { + interruptId, + status: 'resolved', + payload: { approved: true }, + }, + ], + middleware: [withPersistence(persistence)], + }) as AsyncIterable, + ) + + expect(resumed.calls).toHaveLength(0) + expect(chunks.some((chunk) => chunk.type === EventType.RUN_ERROR)).toBe( + true, + ) + expect( + (await persistence.stores.interrupts!.get(interruptId))?.status, + ).toBe('pending') + }) + it('resumes a registered generic record without blocking on a foreign persisted interrupt', async () => { const persistence = memoryPersistence() const review = defineInterrupt({ diff --git a/packages/ai/src/activities/chat/index.ts b/packages/ai/src/activities/chat/index.ts index c7eb73a2f..552424dac 100644 --- a/packages/ai/src/activities/chat/index.ts +++ b/packages/ai/src/activities/chat/index.ts @@ -14,6 +14,7 @@ import { EventType } from '../../types' import { INTERRUPT_BINDING_METADATA_KEY, InterruptResumeValidationError, + readInterruptBinding, readUnopenedInterruptBinding, validateInterruptResumeBatch, } from '../../interrupt-resume' @@ -21,6 +22,7 @@ import { INTERRUPT_BINDING_VERSION } from '../../interrupts' import { INTERRUPT_PAYLOAD_METADATA_KEY, createInterruptBinding, + rehydrateInterruptRequest, } from '../../interrupt-definition' import { readGenericInterruptContinuation } from '../../generic-interrupt-continuation' import type { @@ -216,118 +218,11 @@ function normalizePublicInterruptBinding( value: unknown, expectedInterruptId: string, ): InterruptBinding | undefined { - if (value === null || typeof value !== 'object' || Array.isArray(value)) { - return undefined - } - const binding: Record = Object.fromEntries( - Object.entries(value), - ) - if ( - binding.interruptId !== expectedInterruptId || - // A binding version we don't recognise belongs to another producer. Drop - // it rather than reading our fields out of it. - (binding.v !== undefined && binding.v !== INTERRUPT_BINDING_VERSION) || - typeof binding.interruptedRunId !== 'string' || - typeof binding.generation !== 'number' || - !Number.isInteger(binding.generation) || - binding.generation < 0 || - (binding.expiresAt !== undefined && typeof binding.expiresAt !== 'string') - ) { - return undefined - } - const base = { - v: INTERRUPT_BINDING_VERSION, - interruptId: binding.interruptId, - interruptedRunId: binding.interruptedRunId, - generation: binding.generation, - ...(typeof binding.responseSchemaHash === 'string' - ? { responseSchemaHash: binding.responseSchemaHash } - : {}), - ...(typeof binding.expiresAt === 'string' - ? { expiresAt: binding.expiresAt } - : {}), - } - if (binding.kind === 'generic') { - if ( - binding.responseSchemaHash !== undefined && - typeof binding.responseSchemaHash !== 'string' - ) { - return undefined - } - const hasFirstPartyFields = [ - binding.definitionId, - binding.key, - binding.batchIndex, - binding.payloadSchemaHash, - ].some((field) => field !== undefined) - if ( - hasFirstPartyFields && - (typeof binding.definitionId !== 'string' || - typeof binding.key !== 'string' || - typeof binding.batchIndex !== 'number' || - !Number.isInteger(binding.batchIndex) || - binding.batchIndex < 0 || - (binding.payloadSchemaHash !== undefined && - typeof binding.payloadSchemaHash !== 'string')) - ) { - return undefined - } - if ( - typeof binding.definitionId === 'string' && - typeof binding.key === 'string' && - typeof binding.batchIndex === 'number' - ) { - return { - kind: binding.kind, - ...base, - definitionId: binding.definitionId, - key: binding.key, - batchIndex: binding.batchIndex, - ...(typeof binding.payloadSchemaHash === 'string' - ? { payloadSchemaHash: binding.payloadSchemaHash } - : {}), - } - } - return { kind: binding.kind, ...base } - } - if ( - typeof binding.responseSchemaHash !== 'string' || - typeof binding.toolName !== 'string' || - typeof binding.toolCallId !== 'string' - ) { - return undefined - } - if ( - binding.kind === 'client-tool-execution' && - typeof binding.outputSchemaHash === 'string' - ) { - return { - kind: binding.kind, - ...base, - responseSchemaHash: binding.responseSchemaHash, - toolName: binding.toolName, - toolCallId: binding.toolCallId, - outputSchemaHash: binding.outputSchemaHash, - } - } - if ( - binding.kind === 'tool-approval' && - Object.prototype.hasOwnProperty.call(binding, 'originalArgs') && - typeof binding.inputSchemaHash === 'string' && - typeof binding.approvalSchemaHash === 'string' - ) { - return { - kind: binding.kind, - ...base, - responseSchemaHash: binding.responseSchemaHash, - toolName: binding.toolName, - toolCallId: binding.toolCallId, - originalArgs: binding.originalArgs, - inputSchemaHash: binding.inputSchemaHash, - approvalSchemaHash: binding.approvalSchemaHash, - } - } - return undefined + return readInterruptBinding({ + id: expectedInterruptId, + reason: '', + metadata: { [INTERRUPT_BINDING_METADATA_KEY]: value }, + }) } // The leaf context-inference primitives (KnownContext, MergeContext, @@ -4239,19 +4134,17 @@ class TextEngine< InterruptDefinition > try { - request = Reflect.apply(definition.interrupt, definition, [ - { - key: entry.key, - reason: entry.reason, - message: entry.message, - ...(typeof entry.expiresAt === 'string' - ? { expiresAt: entry.expiresAt } - : {}), - ...(Object.prototype.hasOwnProperty.call(entry, 'payload') - ? { payload: entry.payload } - : {}), - }, - ]) + request = rehydrateInterruptRequest(definition, { + key: entry.key, + reason: entry.reason, + message: entry.message, + ...(typeof entry.expiresAt === 'string' + ? { expiresAt: entry.expiresAt } + : {}), + ...(Object.prototype.hasOwnProperty.call(entry, 'payload') + ? { payload: entry.payload } + : {}), + }) } catch (error) { return fail( `Generic interrupt continuation ${id} is invalid: ${ diff --git a/packages/ai/src/interrupt-definition.ts b/packages/ai/src/interrupt-definition.ts index 48c23ef38..a39608b4a 100644 --- a/packages/ai/src/interrupt-definition.ts +++ b/packages/ai/src/interrupt-definition.ts @@ -11,11 +11,11 @@ import { isStandardSchema, isStandardJSONSchema, } from './activities/chat/tools/schema-converter' +import { INTERRUPT_BINDING_VERSION } from './interrupts' export const INTERRUPT_PAYLOAD_METADATA_KEY = 'tanstack:interruptPayload' as const export const INTERRUPT_BINDING_KIND = 'generic' as const -export const INTERRUPT_BINDING_VERSION = 1 as const type PortableSchema = | StandardJSONSchemaV1 @@ -381,8 +381,8 @@ function validateNonEmptyString(value: unknown, label: string): string { } function validateExpiresAt(value: unknown): string { - if (typeof value !== 'string') { - throw new TypeError('Interrupt expiresAt must be a string.') + if (typeof value !== 'string' || !Number.isFinite(Date.parse(value))) { + throw new TypeError('Interrupt expiresAt must be a valid date string.') } return value } diff --git a/packages/ai/src/interrupt-resume.ts b/packages/ai/src/interrupt-resume.ts index 6694e452f..6202c3d1b 100644 --- a/packages/ai/src/interrupt-resume.ts +++ b/packages/ai/src/interrupt-resume.ts @@ -343,19 +343,29 @@ export async function validateInterruptResumeBatch( ), ) } - if ( - binding.expiresAt !== undefined && - Date.parse(binding.expiresAt) <= (input.now ?? Date.now()) - ) { - errors.push( - interruptItemError( - input, - record.interruptId, - 'expired', - `Interrupt ${record.interruptId} has expired.`, - { source: 'server' }, - ), - ) + if (binding.expiresAt !== undefined) { + const expiresAt = Date.parse(binding.expiresAt) + if (!Number.isFinite(expiresAt)) { + errors.push( + interruptItemError( + input, + record.interruptId, + 'invalid-payload', + `Interrupt ${record.interruptId} has an invalid expiresAt.`, + { source: 'server' }, + ), + ) + } else if (expiresAt <= (input.now ?? Date.now())) { + errors.push( + interruptItemError( + input, + record.interruptId, + 'expired', + `Interrupt ${record.interruptId} has expired.`, + { source: 'server' }, + ), + ) + } } const responseSchema = validateDescriptorSchema( @@ -684,23 +694,42 @@ export async function validateInterruptResumeBatch( if (!entry) continue const binding = record.binding if (binding.kind === 'generic') { - const parsed = - entry.status === 'resolved' && record.genericRequest !== undefined - ? await parseSchemaValue( - record.genericRequest.definition.responseSchema, - entry.payload, - ) - : undefined - genericInterrupts.set( - record.interruptId, - entry.status === 'resolved' - ? { - interruptId: record.interruptId, - status: 'resolved', - payload: parsed?.success ? parsed.data : entry.payload, - } - : { interruptId: record.interruptId, status: 'cancelled' }, + if (entry.status !== 'resolved') { + genericInterrupts.set(record.interruptId, { + interruptId: record.interruptId, + status: 'cancelled', + }) + continue + } + if (record.genericRequest === undefined) { + genericInterrupts.set(record.interruptId, { + interruptId: record.interruptId, + status: 'resolved', + payload: entry.payload, + }) + continue + } + const parsed = await parseSchemaValue( + record.genericRequest.definition.responseSchema, + entry.payload, ) + if (!parsed.success) { + return { + errors: [ + interruptItemError( + input, + record.interruptId, + 'invalid-payload', + `Interrupt ${record.interruptId} payload is invalid.`, + ), + ], + } + } + genericInterrupts.set(record.interruptId, { + interruptId: record.interruptId, + status: 'resolved', + payload: parsed.data, + }) continue } if (entry.status === 'cancelled') { @@ -787,6 +816,9 @@ export function readUnopenedInterruptBinding( const responseSchemaHash = stringField(raw, 'responseSchemaHash') const expiresAt = stringField(raw, 'expiresAt') if (!interruptId || responseSchemaHash === '') return undefined + if (expiresAt !== undefined && !Number.isFinite(Date.parse(expiresAt))) { + return undefined + } const v = INTERRUPT_BINDING_VERSION if (kind === 'generic') { const definitionId = stringField(raw, 'definitionId') diff --git a/packages/ai/tests/chat.test.ts b/packages/ai/tests/chat.test.ts index d3ba020ba..4ab1220e3 100644 --- a/packages/ai/tests/chat.test.ts +++ b/packages/ai/tests/chat.test.ts @@ -3984,7 +3984,7 @@ describe('chat()', () => { id: 'review-plan', responseSchema: z.object({ approved: z.boolean() }), }) - const { adapter } = createMockAdapter({ + const { adapter, calls } = createMockAdapter({ iterations: [ [ ev.runStarted(), @@ -4020,6 +4020,15 @@ describe('chat()', () => { }) as AsyncIterable, ) + expect(calls).toHaveLength(1) + expect( + chunks.some( + (chunk) => + chunk.type === EventType.TEXT_MESSAGE_CONTENT && + 'delta' in chunk && + chunk.delta === 'Plan', + ), + ).toBe(true) const terminal = expectSingleRunFinished(chunks) expect(terminal.outcome).toMatchObject({ type: 'interrupt', @@ -4040,6 +4049,185 @@ describe('chat()', () => { ).toBe(false) }) + it('rehydrates a transformed display payload on ephemeral resume', async () => { + const review = defineInterrupt({ + id: 'review-plan', + payloadSchema: z.string().transform((value) => value.length), + responseSchema: z.object({ approved: z.boolean() }), + }) + const observed: Array = [] + const middleware = defineChatMiddleware({ + onInterruptBoundary(ctx) { + if (ctx.parentRunId) return + if (ctx.phase !== 'afterModel') return + return { + interrupts: [ + review.interrupt({ + key: 'turn-1', + reason: 'review', + message: 'Review the plan', + payload: 'hello', + }), + ], + } + }, + onInterruptResolution(_ctx, resolutions) { + observed.push( + ...resolutions + .for(review) + .map((resolution) => resolution.request.payload), + ) + return { toolResume: 'continue' } + }, + }) + const { adapter, calls } = createMockAdapter({ + iterations: [ + [ + ev.runStarted(), + ev.textStart(), + ev.textContent('Plan'), + ev.textEnd(), + ev.runFinished('stop'), + ], + [ + ev.runStarted(), + ev.textStart(), + ev.textContent('Continued'), + ev.textEnd(), + ev.runFinished('stop'), + ], + ], + }) + + const first = await collectChunks( + chat({ + adapter, + interrupts: [review], + middleware: [middleware], + messages: [{ role: 'user', content: 'Make a plan' }], + threadId: 'thread-transform', + runId: 'run-transform', + }) as AsyncIterable, + ) + const interrupt = expectSingleRunFinished(first).outcome + if (interrupt?.type !== 'interrupt') { + throw new Error('Expected afterModel interrupt') + } + const paused = interrupt.interrupts[0] + if (!paused) throw new Error('Expected interrupt id') + const continuation = genericInterruptContinuationFromDescriptor(paused) + if (!continuation) { + throw new Error('Expected generic continuation metadata') + } + + const resume = await collectChunks( + chat({ + adapter, + interrupts: [review], + middleware: [middleware], + messages: [{ role: 'user', content: 'Make a plan' }], + threadId: 'thread-transform', + runId: 'run-transform-resume', + parentRunId: 'run-transform', + resume: [ + { + interruptId: paused.id, + status: 'resolved', + payload: { approved: true }, + metadata: wrapGenericInterruptContinuation(continuation), + }, + ], + }) as AsyncIterable, + ) + + expect(calls).toHaveLength(2) + expect(observed).toEqual([5]) + expect( + resume.some((chunk) => chunk.type === EventType.RUN_FINISHED), + ).toBe(true) + }) + + it('rejects an ephemeral continuation whose schema hash drifted', async () => { + const review = defineInterrupt({ + id: 'review-plan', + responseSchema: z.object({ approved: z.boolean() }), + }) + const { adapter, calls } = createMockAdapter({ + iterations: [ + [ + ev.runStarted(), + ev.textStart(), + ev.textContent('Plan'), + ev.textEnd(), + ev.runFinished('stop'), + ], + ], + }) + const middleware = defineChatMiddleware({ + onInterruptBoundary(ctx) { + if (ctx.phase !== 'afterModel') return + return { + interrupts: [ + review.interrupt({ + key: 'turn-1', + reason: 'review', + message: 'Review the plan', + }), + ], + } + }, + }) + + const first = await collectChunks( + chat({ + adapter, + interrupts: [review], + middleware: [middleware], + messages: [{ role: 'user', content: 'Make a plan' }], + threadId: 'thread-stale', + runId: 'run-stale', + }) as AsyncIterable, + ) + const interrupt = expectSingleRunFinished(first).outcome + if (interrupt?.type !== 'interrupt') { + throw new Error('Expected afterModel interrupt') + } + const paused = interrupt.interrupts[0] + if (!paused) throw new Error('Expected interrupt id') + const continuation = genericInterruptContinuationFromDescriptor(paused) + if (!continuation) { + throw new Error('Expected generic continuation metadata') + } + + const resume = await collectChunks( + chat({ + adapter, + interrupts: [review], + middleware: [middleware], + messages: [{ role: 'user', content: 'Make a plan' }], + threadId: 'thread-stale', + runId: 'run-stale-resume', + parentRunId: 'run-stale', + resume: [ + { + interruptId: paused.id, + status: 'resolved', + payload: { approved: true }, + metadata: wrapGenericInterruptContinuation({ + ...continuation, + responseSchemaHash: 'sha256:drifted', + }), + }, + ], + }) as AsyncIterable, + ) + + expect(calls).toHaveLength(1) + expect(resume.some((chunk) => chunk.type === EventType.RUN_ERROR)).toBe( + true, + ) + }) + it('starts a synthetic run before a beforeModel interrupt', async () => { const review = defineInterrupt({ id: 'before-model-lifecycle', diff --git a/packages/ai/tests/interrupt-resume.test.ts b/packages/ai/tests/interrupt-resume.test.ts index 7b7cf91bd..625c68759 100644 --- a/packages/ai/tests/interrupt-resume.test.ts +++ b/packages/ai/tests/interrupt-resume.test.ts @@ -129,6 +129,24 @@ describe('validateInterruptResumeBatch', () => { expect(result.errors.some((error) => error.code === 'expired')).toBe(true) }) + it('rejects an unparseable expiresAt', async () => { + const fixture = approvalFixture({ + expiresAt: 'not-a-date', + }) + const result = await validateInterruptResumeBatch( + baseInput(pendingOf(fixture), [ + { + interruptId: fixture.binding.interruptId, + status: 'resolved', + payload: { approved: true, payload: { note: 'ok' } }, + }, + ]), + ) + expect( + result.errors.some((error) => error.code === 'invalid-payload'), + ).toBe(true) + }) + it('rejects stale correlation metadata', async () => { const fixture = approvalFixture({ interruptedRunId: 'other-run' }) const result = await validateInterruptResumeBatch( @@ -352,6 +370,56 @@ describe('validateInterruptResumeBatch', () => { expect(result.resumeToolState?.clientToolResults?.size).toBe(0) }) + it('rejects an invalid first-party generic answer', async () => { + const review = defineInterrupt({ + id: 'review-plan', + responseSchema: z.object({ + approved: z.boolean(), + note: z.string(), + }), + }) + const request = review.interrupt({ + key: 'one', + reason: 'review', + message: 'Review', + }) + const binding: Extract = { + v: INTERRUPT_BINDING_VERSION, + kind: 'generic', + interruptId: 'generic-1', + interruptedRunId: 'run-1', + generation: 0, + definitionId: 'review-plan', + key: 'one', + batchIndex: 0, + } + const result = await validateInterruptResumeBatch({ + threadId: 'thread-1', + interruptedRunId: 'run-1', + generation: 0, + pending: [ + { + interruptId: binding.interruptId, + payload: request, + binding, + genericRequest: request, + }, + ], + resume: [ + { + interruptId: binding.interruptId, + status: 'resolved', + payload: { approved: 'yes' }, + }, + ], + tools: [transfer], + }) + expect( + result.errors.some((error) => error.code === 'invalid-payload'), + ).toBe(true) + expect(result.resumeToolState).toBeUndefined() + }) + it('still requires a client-tool resume when the batch has no generic interrupt', async () => { const renderDef = toolDefinition({ name: 'render_review', diff --git a/packages/ai/tests/interrupts.test.ts b/packages/ai/tests/interrupts.test.ts index 081128856..a915356e5 100644 --- a/packages/ai/tests/interrupts.test.ts +++ b/packages/ai/tests/interrupts.test.ts @@ -101,6 +101,18 @@ describe('first-party interrupt definitions', () => { ]), ).toThrow(/Interrupt input field id is not allowed/) }) + + it('rejects a non-date expiresAt', () => { + expect(() => + approval.interrupt({ + key: 'payment-1', + payload: { amount: 10 }, + reason: 'tool_call', + message: 'Approve payment?', + expiresAt: 'not-a-date', + }), + ).toThrow(/expiresAt must be a valid date string/) + }) }) describe('AG-UI interrupt protocol types', () => { @@ -347,4 +359,20 @@ describe('interrupt binding seam', () => { expect(readInterruptBinding(legacy)).toEqual(openedBinding) }) + + it('rejects a binding with an unparseable expiresAt', () => { + const invalid: Interrupt = { + id: 'pause-1', + reason: 'confirmation', + metadata: { + [INTERRUPT_BINDING_METADATA_KEY]: { + ...openedBinding, + expiresAt: 'not-a-date', + }, + }, + } + + expect(readUnopenedInterruptBinding(invalid)).toBeUndefined() + expect(readInterruptBinding(invalid)).toBeUndefined() + }) })