fix(schedule): stop cancelled mutations

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) {

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<T>(
+  agent: Agent,
+  signal: AbortSignal,
+  task: () => Promise<T>,
+): Promise<T | InternalScheduleError> {
+  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<ScheduleListValue> {
         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 {

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<ToolExecutionResult> {
   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<void>((resolve) => {
+      markOwnerStarted = resolve
+    })
+    const owner = runScheduleTransaction(test.agent, async () => {
+      markOwnerStarted?.()
+      await new Promise<void>((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')