diff --git a/packages/client/runtime/src/client/sessions/session.ts b/packages/client/runtime/src/client/sessions/session.ts index 366c25e617..d91a2adb7d 100644 --- a/packages/client/runtime/src/client/sessions/session.ts +++ b/packages/client/runtime/src/client/sessions/session.ts @@ -943,6 +943,8 @@ export class Session implements SessionFace { retryGap = hasGap && repairedTail !== null && (previousTail === null || repairedTail > previousTail) } else { + // Keep buffered events for the next live frame or reconnect; retrying + // immediately would spin against the same unavailable history endpoint. this.mergeWindow() } } catch (error) { diff --git a/packages/schedule/tool-schedule/src/tools.ts b/packages/schedule/tool-schedule/src/tools.ts index 1f952e5a56..77e2099f6e 100644 --- a/packages/schedule/tool-schedule/src/tools.ts +++ b/packages/schedule/tool-schedule/src/tools.ts @@ -25,6 +25,7 @@ import type { ScheduleCreateValue, ScheduleDeleteValue, ScheduleId as ScheduleIdType, + InternalScheduleError, ScheduleListValue, SchedulePersistenceOperation, ScheduleToolError, @@ -134,10 +135,27 @@ function present(title: string, kind: 'read' | 'other', rawInput?: unknown): Gen } /** Stable error for failures not safe to expose. */ -function internalError(): ScheduleToolError { +function internalError(): InternalScheduleError { return { code: 'internal_error', message: 'The schedule operation failed.' } } +/** Placeholder the registry replaces with its canonical ABORTED result after body quiescence. */ +function cancellationPlaceholder(signal: AbortSignal): InternalScheduleError | undefined { + return signal.aborted ? internalError() : undefined +} + +/** Serialize one operation, stopping a body whose caller cancelled before its FIFO turn. */ +function runCancellableScheduleTransaction( + agent: Agent, + signal: AbortSignal, + task: () => Promise, +): Promise { + return runScheduleTransaction(agent, async () => { + const cancelled = cancellationPlaceholder(signal) + return cancelled ?? task() + }) +} + /** Stable durable-log failure. */ function corruptLogError(): ScheduleToolError { return { code: 'corrupt_schedule_log', message: 'The session schedule log is corrupt.' } @@ -256,7 +274,7 @@ export function registerScheduleTools( if (exec.agent !== agent) return internalError() const invalid = validateCreateArgs(args) if (invalid !== undefined) return invalid - return runScheduleTransaction(agent, async () => { + return runCancellableScheduleTransaction(agent, exec.signal, async () => { const uncertain = await preflight(rootCtx, agent, 'create') if (uncertain !== undefined) return uncertain notifyDurableChange() @@ -269,6 +287,8 @@ export function registerScheduleTools( } catch (error: unknown) { return error instanceof ScheduleInputError ? inputError(error) : internalError() } + const cancelledBeforeAppend = cancellationPlaceholder(exec.signal) + if (cancelledBeforeAppend !== undefined) return cancelledBeforeAppend try { agent.session.append('schedule/change', { version: 1, @@ -294,7 +314,7 @@ export function registerScheduleTools( output: { schema: LIST_OUTPUT_SCHEMA, render: renderValue }, async execute(_args, exec): Promise { if (exec.agent !== agent) return internalError() - return runScheduleTransaction(agent, async () => { + return runCancellableScheduleTransaction(agent, exec.signal, async () => { const uncertain = await preflight(rootCtx, agent, 'list') if (uncertain !== undefined) return uncertain notifyDurableChange() @@ -320,7 +340,7 @@ export function registerScheduleTools( } const id = ScheduleId(args.id) if (exec.agent !== agent) return internalError() - return runScheduleTransaction(agent, async () => { + return runCancellableScheduleTransaction(agent, exec.signal, async () => { const uncertain = await preflight(rootCtx, agent, 'delete', id) if (uncertain !== undefined) return uncertain notifyDurableChange() @@ -329,6 +349,8 @@ export function registerScheduleTools( if (!folded.active.some(record => record.id === id)) { return { id, deleted: false, code: 'schedule_not_found' } } + const cancelledBeforeAppend = cancellationPlaceholder(exec.signal) + if (cancelledBeforeAppend !== undefined) return cancelledBeforeAppend try { agent.session.append('schedule/change', { version: 1, operation: 'delete', id }) } catch { diff --git a/packages/schedule/tool-schedule/tests/tools.spec.ts b/packages/schedule/tool-schedule/tests/tools.spec.ts index 8490e379ab..8c9809731a 100644 --- a/packages/schedule/tool-schedule/tests/tools.spec.ts +++ b/packages/schedule/tool-schedule/tests/tools.spec.ts @@ -9,6 +9,7 @@ import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRegistry from '@deepseek-ai/dsh-tools' import type { ToolExecutionResult } from '@deepseek-ai/dsh-tools' import { registerScheduleTools } from '../src/tools.ts' +import { runScheduleTransaction } from '../src/transaction.ts' const signal = new AbortController().signal const contexts: Context[] = [] @@ -69,9 +70,10 @@ async function execute( name: string, args: unknown, agent: Agent = test.agent, + executionSignal: AbortSignal = signal, ): Promise { return test.ctx.agents.withInitiator(agent, () => test.ctx.tools.execute({ - signal, + signal: executionSignal, callId: CallId(`call-${Math.random()}`), name, arguments: args, @@ -310,6 +312,87 @@ describe('Schedule persistence failure boundaries', () => { expect(test.flushes.count).toBe(3) }) + it('does not persist a create cancelled while it waits in the Schedule FIFO', async () => { + const test = await harness() + let releaseOwner: (() => void) | undefined + let markOwnerStarted: (() => void) | undefined + const ownerStarted = new Promise((resolve) => { + markOwnerStarted = resolve + }) + const owner = runScheduleTransaction(test.agent, async () => { + markOwnerStarted?.() + await new Promise((resolve) => { releaseOwner = resolve }) + }) + await ownerStarted + + const controller = new AbortController() + const creating = execute(test, 'schedule_create', { + prompt: 'cancelled before its turn', after_seconds: 1, + }, test.agent, controller.signal) + await Promise.resolve() + controller.abort() + if (releaseOwner === undefined) throw new Error('missing owner transaction release') + releaseOwner() + await owner + + await expect(creating).resolves.toMatchObject({ + isError: true, + error: { info: { name: 'AbortError', code: 'ABORTED' } }, + }) + expect(test.flushes.count).toBe(0) + expect(test.agent.session.events.filter(event => event.type === 'schedule/change')).toEqual([]) + }) + + it('does not persist a create cancelled during its first preflight', async () => { + const test = await harness() + let releaseCreate: (() => void) | undefined + const blockedCreate = new Promise<'resolve'>((resolve) => { + releaseCreate = () => { resolve('resolve') } + }) + test.flushes.outcomes.push(blockedCreate) + const controller = new AbortController() + const creating = execute(test, 'schedule_create', { + prompt: 'cancelled during preflight', after_seconds: 1, + }, test.agent, controller.signal) + await vi.waitFor(() => { expect(test.flushes.count).toBe(1) }) + controller.abort() + if (releaseCreate === undefined) throw new Error('missing create preflight release') + releaseCreate() + + await expect(creating).resolves.toMatchObject({ + isError: true, + error: { info: { name: 'AbortError', code: 'ABORTED' } }, + }) + expect(test.flushes.count).toBe(1) + expect(test.agent.session.events.filter(event => event.type === 'schedule/change')).toEqual([]) + }) + + it('does not persist a delete cancelled during its first preflight', async () => { + const test = await harness() + await execute(test, 'schedule_create', { prompt: 'keep me', after_seconds: 60 }) + let releaseDelete: (() => void) | undefined + const blockedDelete = new Promise<'resolve'>((resolve) => { + releaseDelete = () => { resolve('resolve') } + }) + test.flushes.outcomes.push(blockedDelete) + const controller = new AbortController() + const deleting = execute(test, 'schedule_delete', { id: 'schedule-1' }, test.agent, controller.signal) + await vi.waitFor(() => { expect(test.flushes.count).toBe(3) }) + controller.abort() + if (releaseDelete === undefined) throw new Error('missing delete preflight release') + releaseDelete() + + await expect(deleting).resolves.toMatchObject({ + isError: true, + error: { info: { name: 'AbortError', code: 'ABORTED' } }, + }) + expect(test.flushes.count).toBe(3) + expect(test.agent.session.events.filter(event => event.type === 'schedule/change')) + .toHaveLength(1) + expect(value(await execute(test, 'schedule_list', {}))) + .toEqual([expect.objectContaining({ id: 'schedule-1' })]) + }) + it('returns uncertainty before create or delete reads when their preflight rejects', async () => { const createTest = await harness() createTest.flushes.outcomes.push('reject')