feat(schedule): send messages later on existing threads

kun/src/server/routes/scheduled-send-admission.test.ts

@@ -0,0 +1,76 @@
+import { describe, expect, it, vi } from 'vitest'
+import { InMemoryEventBus } from '../../adapters/in-memory-event-bus.js'
+import { InMemorySessionStore } from '../../adapters/in-memory-session-store.js'
+import { InMemoryThreadStore } from '../../adapters/in-memory-thread-store.js'
+import type { StartTurnRequest } from '../../contracts/turns.js'
+import { createThreadRecord } from '../../domain/thread.js'
+import { ContextCompactor } from '../../loop/context-compactor.js'
+import { InflightTracker } from '../../loop/inflight-tracker.js'
+import { SteeringQueue } from '../../loop/steering-queue.js'
+import { SequentialIdGenerator } from '../../ports/id-generator.js'
+import { RuntimeEventRecorder } from '../../services/runtime-event-recorder.js'
+import { TurnService } from '../../services/turn-service.js'
+import type { JsonResponse } from '../response.js'
+import { startTurn } from './turns.js'
+
+describe('scheduled send turn admission', () => {
+  it('admits repeated wakeups with the same key exactly once', async () => {
+    const threadStore = new InMemoryThreadStore()
+    const sessionStore = new InMemorySessionStore()
+    const eventBus = new InMemoryEventBus()
+    const nowIso = () => '2026-08-30T10:00:00.000Z'
+    const turns = new TurnService({
+      threadStore,
+      sessionStore,
+      events: new RuntimeEventRecorder({
+        eventBus,
+        sessionStore,
+        allocateSeq: (threadId) => eventBus.allocateSeq(threadId),
+        nowIso
+      }),
+      inflight: new InflightTracker(),
+      steering: new SteeringQueue(),
+      compactor: new ContextCompactor(),
+      ids: new SequentialIdGenerator(),
+      nowIso
+    })
+    const threadId = 'thread-scheduled-send'
+    await threadStore.upsert(createThreadRecord({
+      id: threadId,
+      title: 'Existing conversation',
+      workspace: '/tmp/workspace',
+      model: 'model-a',
+      providerId: 'provider-a',
+      accountId: 'account-a'
+    }))
+    const requestBody: StartTurnRequest = {
+      prompt: 'Continue the existing investigation',
+      clientRequestId: 'scheduled-send:task-1:dispatch-1',
+      providerId: 'provider-a',
+      accountId: 'account-a',
+      model: 'model-a',
+      reasoningEffort: 'high'
+    }
+    const request = () => new Request(`http://kun.local/v1/threads/${threadId}/turns`, {
+      method: 'POST',
+      headers: { 'content-type': 'application/json' },
+      body: JSON.stringify(requestBody)
+    })
+    const onStarted = vi.fn()
+
+    const [first, duplicate] = await Promise.all([
+      startTurn(turns, threadId, request(), onStarted),
+      startTurn(turns, threadId, request(), onStarted)
+    ]) as JsonResponse[]
+
+    expect(first.status).toBe(202)
+    expect(duplicate.status).toBe(202)
+    expect(JSON.parse(duplicate.body)).toEqual(JSON.parse(first.body))
+    expect(onStarted).toHaveBeenCalledTimes(1)
+    expect((await threadStore.get(threadId))?.turns).toHaveLength(1)
+    expect((await sessionStore.loadItems(threadId)).filter((item) => item.kind === 'user_message'))
+      .toHaveLength(1)
+    expect((await sessionStore.loadEventsSince(threadId, 0)).filter((event) => event.kind === 'turn_started'))
+      .toHaveLength(1)
+  })
+})

src/main/ipc/app-ipc-schemas/system.ts

@@ -106,37 +106,72 @@ export const clawTaskFromTextPayloadSchema = z
   })
   .strict()
 
+const scheduleDefinitionSchema = z.object({
+  kind: z.enum(['manual', 'interval', 'daily', 'at']),
+  everyMinutes: z.number().int().min(1).max(10_080).optional(),
+  timeOfDay: z.string().regex(/^([01]\d|2[0-3]):[0-5]\d$/).optional(),
+  atTime: z.string().datetime().optional(),
+  timeZone: z.string().trim().min(1).max(128).refine(isValidTimeZone, 'Invalid IANA time zone.').optional()
+}).strict()
+
 export const scheduleTaskCreatePayloadSchema = z
   .object({
     title: z.string().trim().min(1).max(200),
     prompt: z.string().min(1).max(500_000),
     workspaceRoot: defaultPathSchema,
-    sourcePlanId: z.string().trim().min(1).max(MAX_ID_LENGTH),
+    sourcePlanId: z.string().trim().min(1).max(MAX_ID_LENGTH).optional(),
     sourceThreadId: z.string().trim().min(1).max(MAX_ID_LENGTH).optional(),
     providerId: z.string().trim().min(1).max(128),
+    accountId: z.string().trim().min(1).max(MAX_ID_LENGTH).optional(),
+    attachmentIds: z.array(z.string().trim().min(1).max(MAX_ID_LENGTH)).max(8)
+      .refine((ids) => new Set(ids).size === ids.length, 'attachmentIds must not contain duplicates')
+      .optional(),
+    enabled: z.boolean().optional(),
+    clawChannelId: z.string().trim().max(MAX_ID_LENGTH).optional(),
+    priority: z.number().int().min(0).max(100).optional(),
+    dependsOn: z.array(z.string().trim().min(1).max(MAX_ID_LENGTH)).max(100).optional(),
+    useWorktree: z.boolean().optional(),
     model: modelIdSchema,
     reasoningEffort: scheduleReasoningEffortSchema,
     mode: z.enum(['agent', 'plan']),
     orchestration: z.enum(['direct', 'graph']),
-    schedule: z.object({
-      kind: z.literal('at'),
-      atTime: z.string().datetime().refine((value) => Date.parse(value) > Date.now(), 'Execution time must be in the future.'),
-      timeZone: z.string().trim().min(1).max(128).refine(isValidTimeZone, 'Invalid IANA time zone.')
-    }).strict()
+    schedule: scheduleDefinitionSchema
+  })
+  .superRefine((value, context) => {
+    if (!value.sourcePlanId && !value.sourceThreadId) {
+      context.addIssue({ code: z.ZodIssueCode.custom, message: 'sourcePlanId or sourceThreadId is required' })
+    }
+    if (value.schedule.kind === 'at' && (!value.schedule.atTime || Date.parse(value.schedule.atTime) <= Date.now())) {
+      context.addIssue({ code: z.ZodIssueCode.custom, path: ['schedule', 'atTime'], message: 'Execution time must be in the future.' })
+    }
+    if (value.sourceThreadId && !value.sourcePlanId && value.mode !== 'agent') {
+      context.addIssue({ code: z.ZodIssueCode.custom, path: ['mode'], message: 'Scheduled sends use agent mode' })
+    }
   })
   .strict()
 
 export const scheduleTaskUpdatePayloadSchema = z
   .object({
     taskId: z.string().trim().min(1).max(MAX_ID_LENGTH),
-    providerId: z.string().trim().min(1).max(128),
-    model: modelIdSchema,
-    reasoningEffort: scheduleReasoningEffortSchema,
-    schedule: z.object({
-      kind: z.literal('at'),
-      atTime: z.string().datetime().refine((value) => Date.parse(value) > Date.now(), 'Execution time must be in the future.'),
-      timeZone: z.string().trim().min(1).max(128).refine(isValidTimeZone, 'Invalid IANA time zone.')
-    }).strict()
+    title: z.string().trim().min(1).max(200).optional(),
+    prompt: z.string().min(1).max(500_000).optional(),
+    workspaceRoot: defaultPathSchema.optional(),
+    enabled: z.boolean().optional(),
+    clawChannelId: z.string().trim().max(MAX_ID_LENGTH).optional(),
+    providerId: z.string().trim().min(1).max(128).optional(),
+    model: modelIdSchema.optional(),
+    reasoningEffort: scheduleReasoningEffortSchema.optional(),
+    mode: z.enum(['agent', 'plan']).optional(),
+    orchestration: z.enum(['direct', 'graph']).optional(),
+    priority: z.number().int().min(0).max(100).optional(),
+    dependsOn: z.array(z.string().trim().min(1).max(MAX_ID_LENGTH)).max(100).optional(),
+    useWorktree: z.boolean().optional(),
+    schedule: scheduleDefinitionSchema.partial().optional()
+  })
+  .superRefine((value, context) => {
+    if (value.schedule?.kind === 'at' && value.schedule.atTime && Date.parse(value.schedule.atTime) <= Date.now()) {
+      context.addIssue({ code: z.ZodIssueCode.custom, path: ['schedule', 'atTime'], message: 'Execution time must be in the future.' })
+    }
   })
   .strict()
 

src/main/ipc/register-app-runtime-ipc-handlers.ts

@@ -147,8 +147,9 @@ export function registerAppRuntimeIpcHandlers(options: RegisterAppIpcHandlersOpt
     }
   )
 
-  ipcMain.handle('schedule:task:create', async (_, payload: unknown): Promise<ScheduleTaskMutationResult> => {
+  ipcMain.handle('schedule:task:create', async (event, payload: unknown): Promise<ScheduleTaskMutationResult> => {
     try {
+      assertTrustedWorkbenchSender(event, getMainWindow)
       const input = parseIpcPayload('schedule:task:create', scheduleTaskCreatePayloadSchema, payload) as ScheduleTaskCreateInput
       const scheduleRuntime = getScheduleRuntime()
       if (!scheduleRuntime) return { ok: false, message: 'Schedule runtime is not initialized.' }
@@ -159,25 +160,23 @@ export function registerAppRuntimeIpcHandlers(options: RegisterAppIpcHandlersOpt
     }
   })
 
-  ipcMain.handle('schedule:task:update', async (_, payload: unknown): Promise<ScheduleTaskMutationResult> => {
+  ipcMain.handle('schedule:task:update', async (event, payload: unknown): Promise<ScheduleTaskMutationResult> => {
     try {
+      assertTrustedWorkbenchSender(event, getMainWindow)
       const input = parseIpcPayload('schedule:task:update', scheduleTaskUpdatePayloadSchema, payload) as ScheduleTaskUpdateInput
       const scheduleRuntime = getScheduleRuntime()
       if (!scheduleRuntime) return { ok: false, message: 'Schedule runtime is not initialized.' }
-      const task = await scheduleRuntime.updateTaskById(input.taskId, {
-        providerId: input.providerId,
-        model: input.model,
-        reasoningEffort: input.reasoningEffort,
-        schedule: input.schedule
-      })
+      const { taskId, ...patch } = input
+      const task = await scheduleRuntime.updateTaskById(input.taskId, patch)
       return task ? { ok: true, task } : { ok: false, message: 'Scheduled task was not found.' }
     } catch (error) {
       return { ok: false, message: error instanceof Error ? error.message : String(error) }
     }
   })
 
-  ipcMain.handle('schedule:task:delete', async (_, taskId: unknown): Promise<ScheduleTaskDeleteResult> => {
+  ipcMain.handle('schedule:task:delete', async (event, taskId: unknown): Promise<ScheduleTaskDeleteResult> => {
     try {
+      assertTrustedWorkbenchSender(event, getMainWindow)
       const normalizedTaskId = parseIpcPayload('schedule:task:delete', streamIdSchema, taskId)
       const scheduleRuntime = getScheduleRuntime()
       if (!scheduleRuntime) return { ok: false, message: 'Schedule runtime is not initialized.' }

src/main/ipc/scheduled-send-ipc.test.ts

@@ -0,0 +1,68 @@
+import { describe, expect, it } from 'vitest'
+import { scheduleTaskCreatePayloadSchema } from './app-ipc-schemas'
+
+describe('scheduled send IPC contract', () => {
+  const future = '2099-01-01T10:00:00.000Z'
+
+  function payload(overrides: Record<string, unknown> = {}): Record<string, unknown> {
+    return {
+      title: 'Scheduled send',
+      prompt: 'Continue the existing investigation',
+      workspaceRoot: '/tmp/project',
+      sourceThreadId: 'thread-existing',
+      providerId: 'provider-a',
+      accountId: 'account-a',
+      model: 'model-a',
+      reasoningEffort: 'high',
+      mode: 'agent',
+      orchestration: 'direct',
+      attachmentIds: ['attachment-a', 'attachment-b'],
+      schedule: {
+        kind: 'at',
+        atTime: future,
+        timeZone: 'Asia/Taipei'
+      },
+      ...overrides
+    }
+  }
+
+  it('accepts a bounded existing-thread snapshot without a plan binding', () => {
+    const parsed = scheduleTaskCreatePayloadSchema.parse(payload())
+
+    expect(parsed).toMatchObject({
+      sourceThreadId: 'thread-existing',
+      providerId: 'provider-a',
+      accountId: 'account-a',
+      model: 'model-a',
+      attachmentIds: ['attachment-a', 'attachment-b']
+    })
+    expect(parsed).not.toHaveProperty('sourcePlanId')
+  })
+
+  it('requires either an existing thread or a plan owner', () => {
+    const withoutOwner = payload()
+    delete withoutOwner.sourceThreadId
+
+    expect(() => scheduleTaskCreatePayloadSchema.parse(withoutOwner)).toThrow()
+  })
+
+  it('bounds and deduplicates the frozen attachment snapshot', () => {
+    const maximum = Array.from({ length: 8 }, (_, index) => `attachment-${index}`)
+
+    expect(scheduleTaskCreatePayloadSchema.parse(payload({ attachmentIds: maximum })).attachmentIds)
+      .toHaveLength(8)
+    expect(() => scheduleTaskCreatePayloadSchema.parse(payload({
+      attachmentIds: [...maximum, 'attachment-over-limit']
+    }))).toThrow()
+    expect(() => scheduleTaskCreatePayloadSchema.parse(payload({
+      attachmentIds: ['attachment-a', 'attachment-a']
+    }))).toThrow()
+  })
+
+  it('rejects unbounded routing snapshot identifiers', () => {
+    expect(() => scheduleTaskCreatePayloadSchema.parse(payload({ accountId: 'a'.repeat(257) })))
+      .toThrow()
+    expect(() => scheduleTaskCreatePayloadSchema.parse(payload({ attachmentIds: ['a'.repeat(257)] })))
+      .toThrow()
+  })
+})

src/main/schedule-runtime-helpers.ts

@@ -18,6 +18,7 @@ import type {
   ScheduledTaskV1
 } from '../shared/app-settings'
 import type { JsonSettingsStore } from './settings-store'
+import { parseRuntimeErrorBody } from '../shared/runtime-error'
 
 export type RuntimeRequestResult = { ok: boolean; status: number; body: string }
 
@@ -80,6 +81,8 @@ export type ThreadDetailJson = {
 
 export type RunPromptOptions = {
   prompt: string
+  /** Existing-thread scheduled sends must preserve the exact user draft. */
+  preservePrompt?: boolean
   title: string
   workspaceRoot: string
   /** Existing thread that should receive the scheduled turn instead of creating a new one. */
@@ -88,6 +91,9 @@ export type RunPromptOptions = {
   /** Optional provider id; routed via Kun's MultiProviderModelClient. */
   providerId?: string
   reasoningEffort: ScheduleReasoningEffort
+  accountId?: string
+  attachmentIds?: string[]
+  clientRequestId?: string
   mode: ScheduleRunMode
   orchestration?: 'direct' | 'graph'
   clawChannel?: ClawImChannelV1 | null
@@ -126,6 +132,21 @@ export function runtimeErrorMessage(result: RuntimeRequestResult, fallback: stri
   return result.body.trim() || fallback
 }
 
+function runtimeErrorResult(
+  result: RuntimeRequestResult,
+  fallback: string
+): Extract<ScheduleRunResult, { ok: false }> {
+  const parsed = parseRuntimeErrorBody(result.body, fallback)
+  return {
+    ok: false,
+    message: parsed.message,
+    status: result.status,
+    ...(parsed.code !== 'unknown'
+      ? { code: parsed.code }
+      : result.status === 0 ? { code: 'fetch_failed' } : {})
+  }
+}
+
 export function isRunningStatus(status: string | undefined): boolean {
   return status === 'queued' || status === 'in_progress' || status === 'started' || status === 'running'
 }
@@ -363,6 +384,9 @@ export type RunPromptViaRuntimeOptions = {
    */
   providerId?: string
   reasoningEffort: ScheduleReasoningEffort | ''
+  accountId?: string
+  attachmentIds?: string[]
+  clientRequestId?: string
   mode: ScheduleRunMode
   orchestration?: 'direct' | 'graph'
   waitForResult: boolean
@@ -402,7 +426,7 @@ export async function runPromptViaRuntime(
         ...(options.title.trim() ? { title: options.title.trim() } : {})
       })
     })
-    if (!create.ok) return { ok: false, message: runtimeErrorMessage(create, 'Failed to create thread.') }
+    if (!create.ok) return runtimeErrorResult(create, 'Failed to create thread.')
     const thread = JSON.parse(create.body) as ThreadRecordJson
     threadId = thread.id
   }
@@ -419,6 +443,9 @@ export async function runPromptViaRuntime(
   if (model) turnBody.model = model
   if (providerId) turnBody.providerId = providerId
   if (options.reasoningEffort) turnBody.reasoningEffort = options.reasoningEffort
+  if (options.accountId?.trim()) turnBody.accountId = options.accountId.trim()
+  if (options.attachmentIds?.length) turnBody.attachmentIds = options.attachmentIds
+  if (options.clientRequestId?.trim()) turnBody.clientRequestId = options.clientRequestId.trim()
   const turn = await deps.runtimeRequest(
     settings,
     `/v1/threads/${encodeURIComponent(threadId)}/turns`,
@@ -428,7 +455,9 @@ export async function runPromptViaRuntime(
       ...(options.signal ? { signal: options.signal } : {})
     }
   )
-  if (!turn.ok) return { ok: false, message: runtimeErrorMessage(turn, 'Failed to start turn.') }
+  if (!turn.ok) {
+    return runtimeErrorResult(turn, 'Failed to start turn.')
+  }
 
   const parsedTurn = parseJsonObject(turn.body)
   const turnId = asString(nestedRecord(parsedTurn?.turn).id) || asString(parsedTurn?.turnId)