From 23aca3de941352713d916a346158a5e067e3a264 Mon Sep 17 00:00:00 2001 From: luoye520ww <100058663+luoye520ww@users.noreply.github.com> Date: Mon, 31 Aug 2026 00:23:56 +0800 Subject: [PATCH 1/9] feat(schedule): send messages later on existing threads --- .../routes/scheduled-send-admission.test.ts | 76 ++ src/main/ipc/app-ipc-schemas/system.ts | 63 +- .../ipc/register-app-runtime-ipc-handlers.ts | 17 +- src/main/ipc/scheduled-send-ipc.test.ts | 68 ++ src/main/schedule-runtime-helpers.ts | 33 +- src/main/schedule-runtime-queue.ts | 217 ++++-- .../schedule-runtime.scheduled-send.test.ts | 677 ++++++++++++++++++ src/main/schedule-runtime.ts | 115 ++- .../FloatingComposer.scheduled-send.test.ts | 297 ++++++++ .../src/components/chat/FloatingComposer.tsx | 31 +- .../chat/FloatingComposerSurfaceView.tsx | 16 +- .../components/chat/ScheduledSendDialog.tsx | 118 +++ .../src/components/chat/use-scheduled-send.ts | 64 ++ .../components/schedule/ScheduleTasksView.tsx | 78 +- src/shared/app-settings-schedule.ts | 30 +- src/shared/app-settings-types-kun-services.ts | 54 +- src/shared/app-settings-types-product.ts | 2 +- 17 files changed, 1768 insertions(+), 188 deletions(-) create mode 100644 kun/src/server/routes/scheduled-send-admission.test.ts create mode 100644 src/main/ipc/scheduled-send-ipc.test.ts create mode 100644 src/main/schedule-runtime.scheduled-send.test.ts create mode 100644 src/renderer/src/components/chat/FloatingComposer.scheduled-send.test.ts create mode 100644 src/renderer/src/components/chat/ScheduledSendDialog.tsx create mode 100644 src/renderer/src/components/chat/use-scheduled-send.ts diff --git a/kun/src/server/routes/scheduled-send-admission.test.ts b/kun/src/server/routes/scheduled-send-admission.test.ts new file mode 100644 index 000000000..17580c2bc --- /dev/null +++ b/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) + }) +}) diff --git a/src/main/ipc/app-ipc-schemas/system.ts b/src/main/ipc/app-ipc-schemas/system.ts index 4e53dd906..780b01433 100644 --- a/src/main/ipc/app-ipc-schemas/system.ts +++ b/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() diff --git a/src/main/ipc/register-app-runtime-ipc-handlers.ts b/src/main/ipc/register-app-runtime-ipc-handlers.ts index b030fcd56..ca2910020 100644 --- a/src/main/ipc/register-app-runtime-ipc-handlers.ts +++ b/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 => { + ipcMain.handle('schedule:task:create', async (event, payload: unknown): Promise => { 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 => { + ipcMain.handle('schedule:task:update', async (event, payload: unknown): Promise => { 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 => { + ipcMain.handle('schedule:task:delete', async (event, taskId: unknown): Promise => { 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.' } diff --git a/src/main/ipc/scheduled-send-ipc.test.ts b/src/main/ipc/scheduled-send-ipc.test.ts new file mode 100644 index 000000000..0f59a041c --- /dev/null +++ b/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 = {}): Record { + 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() + }) +}) diff --git a/src/main/schedule-runtime-helpers.ts b/src/main/schedule-runtime-helpers.ts index aeac8778f..063f878f3 100644 --- a/src/main/schedule-runtime-helpers.ts +++ b/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 { + 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) diff --git a/src/main/schedule-runtime-queue.ts b/src/main/schedule-runtime-queue.ts index 9038c6bf3..c032ba729 100644 --- a/src/main/schedule-runtime-queue.ts +++ b/src/main/schedule-runtime-queue.ts @@ -25,17 +25,15 @@ import { findAvailablePoolIndex, releaseWorktree } from './services/worktree-service' - const MAX_CONCURRENT_BACKGROUND_TASKS = 3 - +const DEFAULT_SCHEDULED_SEND_MAX_ATTEMPTS = 3 +const SCHEDULED_SEND_RETRY_DELAY_MS = 1_000 export function scheduledThreadTitle(title: string): string { const trimmed = title.trim() const prefix = '[Scheduled task]' const suffix = Array.from(trimmed).slice(0, 4).join('') return suffix ? `${prefix} ${suffix}` : prefix } - - export class ScheduleExecutionQueue { private runningTaskIds = new Set() private queuedTaskIds = new Set() @@ -48,20 +46,21 @@ export class ScheduleExecutionQueue { private drainingQueue = false private readonly stopController = new AbortController() private readonly activeTasks = new Set>() + private wakeTimer: ReturnType | null = null + private wakeAt = 0 + private readonly cancelledTaskIds = new Set() + private readonly admittedTaskIds = new Set() private stopped = false - constructor( private readonly deps: ScheduleRuntimeDeps, private readonly onSettingsUpdated: (settings: AppSettingsV1) => void ) {} - private async loadSettings(): Promise { const settings = await this.deps.store.load() return this.deps.withModelCredentials ? this.deps.withModelCredentials(settings) : settings } - private resolveScheduleModelConfig( settings: AppSettingsV1, input: { @@ -72,32 +71,43 @@ export class ScheduleExecutionQueue { ): ScheduleModelConfig { return resolveScheduleModelConfig(settings, input, settings.schedule.providerId?.trim() || '') } - runningIds(): string[] { return [...this.runningTaskIds] } - /** @internal Preserves the runtime's legacy characterization seam. */ runningSet(): Set { return this.runningTaskIds } - queuedIds(): string[] { return [...this.queuedTaskIds] } - hasRunning(taskId: string): boolean { return this.runningTaskIds.has(taskId) } - hasQueued(taskId: string): boolean { return this.queuedTaskIds.has(taskId) } - + hasAdmitted(taskId: string): boolean { + return this.admittedTaskIds.has(taskId) + } + markQueuedScheduled(taskId: string): void { + if (this.queuedTaskIds.has(taskId)) this.queuedTaskModes.set(taskId, true) + } + cancelTask(taskId: string): void { + this.cancelledTaskIds.add(taskId) + const wasQueued = this.queuedTaskIds.delete(taskId) + this.queuedTaskModes.delete(taskId) + if (wasQueued) { + this.resolveTaskCompletion(taskId, { ok: false, message: 'Scheduled task was cancelled.' }) + void this.drainQueue() + } + } async stop(): Promise { if (this.stopped) return this.stopped = true this.stopController.abort() + if (this.wakeTimer) clearTimeout(this.wakeTimer) + this.wakeTimer = null this.queuedTaskIds.clear() this.queuedTaskModes.clear() for (const taskId of [...this.taskCompletions.keys()]) { @@ -106,8 +116,8 @@ export class ScheduleExecutionQueue { await Promise.allSettled([...this.activeTasks]) await Promise.allSettled([...this.worktreeLeases.keys()].map((taskId) => this.releaseTaskWorktree(taskId))) this.runningTaskIds.clear() + this.admittedTaskIds.clear() } - async runTask(taskId: string): Promise { if (this.stopped) return { ok: false, message: 'Schedule runtime stopped.' } const settings = await this.loadSettings() @@ -125,27 +135,15 @@ export class ScheduleExecutionQueue { if (hasTaskDependencyCycle(task.id, settings.schedule.tasks)) { return { ok: false, message: 'Task dependencies contain a cycle.' } } - // Always go through the queue+drain path. The earlier two-step check — - // `size < MAX` followed by an awaited store.load() — let two concurrent - // IPC callers both pass the cap check before either of them had - // incremented runningTaskIds, briefly running 4+ tasks at once. drainQueue - // owns the cap synchronously (it is serialized via drainingQueue), so - // routing every immediate-run through it eliminates the race. const dependenciesReady = dependencies.every((dependency) => dependency?.lastStatus === 'success') const completion = this.createTaskCompletion(task.id) await this.enqueueTask(task, false) if (!dependenciesReady) { - // Dependency tasks have not all finished yet — the queue will pick this - // task up later when they complete. Return the queued ack now; the - // completion deferred stays parked. return { ok: true, threadId: '', queued: true, message: 'Task queued.' } } return completion } - private createTaskCompletion(taskId: string): Promise { - // If a completion is already parked for this task (e.g. someone else is - // about to run it), return that one. Otherwise create a fresh deferred. const existing = this.taskCompletions.get(taskId) if (existing) { return new Promise((resolve, reject) => { @@ -170,15 +168,12 @@ export class ScheduleExecutionQueue { this.taskCompletions.set(taskId, { resolve: resolveFn, reject: rejectFn }) return promise } - private resolveTaskCompletion(taskId: string, value: ScheduleRunResult): void { const deferred = this.taskCompletions.get(taskId) if (!deferred) return this.taskCompletions.delete(taskId) deferred.resolve(value) } - - async ensureNextRuns(_settings: AppSettingsV1): Promise { if (this.stopped) return const now = new Date() @@ -194,6 +189,10 @@ export class ScheduleExecutionQueue { return task } const wasInterrupted = wasRunning || wasQueued + if (wasInterrupted && task.scheduledSend?.kind === 'thread-send' && task.enabled && task.schedule.kind !== 'manual' && task.scheduledSend.attemptCount < task.scheduledSend.maxAttempts) { + changed = true; this.queuedTaskIds.add(task.id); this.queuedTaskModes.set(task.id, true) + return { ...task, lastStatus: 'queued' as const, lastMessage: 'Resuming scheduled send after interruption.', nextRunAt: now.toISOString(), updatedAt: now.toISOString() } + } if (!task.enabled || task.schedule.kind === 'manual' || this.runningTaskIds.has(task.id)) { if (!wasInterrupted) return task changed = true @@ -225,7 +224,6 @@ export class ScheduleExecutionQueue { }) this.onSettingsUpdated(saved) } - private async updateTask( taskId: string, updater: (task: ScheduledTaskV1, settings: AppSettingsV1) => ScheduledTaskV1 @@ -239,7 +237,6 @@ export class ScheduleExecutionQueue { this.onSettingsUpdated(saved) return saved } - async enqueueTask(task: ScheduledTaskV1, scheduled: boolean): Promise { if (this.stopped) return this.queuedTaskIds.add(task.id) @@ -252,7 +249,6 @@ export class ScheduleExecutionQueue { })) void this.drainQueue() } - async drainQueue(): Promise { if (this.drainingQueue || this.stopped) return this.drainingQueue = true @@ -267,7 +263,17 @@ export class ScheduleExecutionQueue { .filter((task) => this.queuedTaskIds.has(task.id)) .sort((left, right) => (right.priority ?? 0) - (left.priority ?? 0) || left.createdAt.localeCompare(right.createdAt)) let next: ScheduledTaskV1 | undefined + const claimedThreadIds = new Set() for (const task of queued) { + const scheduledSendThreadId = task.scheduledSend?.kind === 'thread-send' ? task.sourceThreadId?.trim() || '' : '' + if (scheduledSendThreadId) { + const hasEarlierQueuedTask = queued.some((candidate) => candidate.id !== task.id && this.queuedTaskIds.has(candidate.id) && candidate.scheduledSend?.kind === 'thread-send' && candidate.sourceThreadId?.trim() === scheduledSendThreadId && (candidate.createdAt < task.createdAt || (candidate.createdAt === task.createdAt && candidate.id < task.id))) + if (hasEarlierQueuedTask) continue + } + if (task.scheduledSend?.kind === 'thread-send' && task.sourceThreadId) { + const threadHasRunningSend = settings.schedule.tasks.some((candidate) => candidate.id !== task.id && this.runningTaskIds.has(candidate.id) && candidate.scheduledSend?.kind === 'thread-send' && candidate.sourceThreadId === task.sourceThreadId) + if (threadHasRunningSend) continue + } if (!task.enabled) { this.queuedTaskIds.delete(task.id) this.queuedTaskModes.delete(task.id) @@ -312,21 +318,25 @@ export class ScheduleExecutionQueue { this.resolveTaskCompletion(task.id, { ok: false, message: cycleMessage }) continue } + if (scheduledSendThreadId) { + if (claimedThreadIds.has(scheduledSendThreadId)) continue + claimedThreadIds.add(scheduledSendThreadId) + } + if (task.scheduledSend?.kind === 'thread-send' && (this.queuedTaskModes.get(task.id) === true || task.scheduledSend.attemptCount > 0) && task.nextRunAt && Date.parse(task.nextRunAt) > Date.now()) continue if (dependencies.every((dependency) => dependency?.lastStatus === 'success')) { next = task break } } - if (!next) break + if (!next) { + const nextWakeAt = queued.filter((task) => task.scheduledSend?.kind === 'thread-send' && task.nextRunAt).map((task) => Date.parse(task.nextRunAt)).filter((at) => Number.isFinite(at) && at > Date.now()).sort((left, right) => left - right)[0] + if (nextWakeAt) this.scheduleDrainAt(nextWakeAt) + break + } const scheduled = this.queuedTaskModes.get(next.id) ?? false const dequeued = next this.queuedTaskIds.delete(dequeued.id) this.queuedTaskModes.delete(dequeued.id) - // Synchronously reserve the running slot BEFORE awaiting anything so - // the next iteration of this drain loop (and a re-entrant drainQueue - // call) sees the updated size. runTaskInternal also defends against - // double-running, but reserving here is what makes the size check at - // the top of the loop correct under back-to-back drains. this.runningTaskIds.add(dequeued.id) const task = this.runTaskInternal(dequeued, scheduled, { slotReserved: true }) this.trackTask(task) @@ -346,7 +356,6 @@ export class ScheduleExecutionQueue { this.drainingQueue = false } } - async runTaskInternal( task: ScheduledTaskV1, scheduled: boolean, @@ -365,31 +374,42 @@ export class ScheduleExecutionQueue { if (slotReserved) this.runningTaskIds.delete(task.id) return { ok: false, message: 'Task prompt is empty.' } } - if (!slotReserved) this.runningTaskIds.add(task.id) + const scheduledSendAttempt = task.scheduledSend?.kind === 'thread-send' + ? task.scheduledSend.attemptCount + 1 + : 0 + if ( + task.scheduledSend?.kind === 'thread-send' && + scheduledSendAttempt > task.scheduledSend.maxAttempts + ) { + if (slotReserved) this.runningTaskIds.delete(task.id) + return { ok: false, message: 'Scheduled send retry limit reached.' } + } await this.updateTask(task.id, (current) => ({ ...current, lastStatus: 'running', lastMessage: 'Running', nextRunAt: '', + ...(current.scheduledSend?.kind === 'thread-send' + ? { scheduledSend: { ...current.scheduledSend, attemptCount: scheduledSendAttempt, reconciliationPending: true } } + : {}), updatedAt: new Date().toISOString() })) - try { const settings = await this.loadSettings() + const persistedTask = settings.schedule.tasks.find((candidate) => candidate.id === task.id) + if (!persistedTask || this.cancelledTaskIds.has(task.id)) { this.runningTaskIds.delete(task.id); return { ok: false, message: 'Scheduled task was removed before admission.' } } + if (task.scheduledSend?.kind === 'thread-send' && (!task.sourceThreadId?.trim() || !task.scheduledSend.clientRequestId.trim() || !task.providerId?.trim() || !task.model.trim())) { + this.runningTaskIds.delete(task.id) + await this.updateTask(task.id, (current) => ({ ...current, enabled: false, lastStatus: 'error', lastMessage: 'Scheduled send snapshot is invalid; no message was sent.', updatedAt: new Date().toISOString() })) + return { ok: false, message: 'Scheduled send snapshot is invalid.' } + } const clawChannel = this.resolveTaskClawChannel(settings, task) let workspaceRoot = this.resolveTaskWorkspaceRoot(settings, task, clawChannel) if (task.useWorktree) { const projectPath = workspaceRoot const poolIndex = await findAvailablePoolIndex({ projectPath }) if (poolIndex === null) { - // No slot is currently available. If other worktree tasks are - // running, one of them will release a slot soon — re-enqueue this - // task so drainQueue picks it up once a slot frees. If nothing else - // is running, every slot is permanently in a state findAvailable... - // can't recover from (e.g. dirty from a non-scheduled lease); fall - // through to the existing error path so the user sees a clear - // failure instead of an unbounded re-queue loop. const hasOtherWorktreeTasks = [...this.runningTaskIds].some((id) => { if (id === task.id) return false return this.worktreeLeases.has(id) @@ -404,8 +424,6 @@ export class ScheduleExecutionQueue { })) this.queuedTaskIds.add(task.id) this.queuedTaskModes.set(task.id, scheduled) - // Defer the drain so the currently-running worktree task gets a - // chance to release before this one is re-picked. setTimeout(() => { void this.drainQueue() }, 250).unref?.() return { ok: true, threadId: '', queued: true, message: 'Task re-queued: no worktree slot available.' } } @@ -415,19 +433,29 @@ export class ScheduleExecutionQueue { workspaceRoot = worktree.path this.worktreeLeases.set(task.id, { projectPath, poolIndex }) } - const modelConfig = this.resolveScheduleModelConfig(settings, { - providerId: task.providerId, - model: task.model, - reasoningEffort: task.reasoningEffort - }) + const modelConfig = task.scheduledSend?.kind === 'thread-send' + ? { providerId: task.providerId?.trim() ?? '', model: task.model, reasoningEffort: task.reasoningEffort } + : this.resolveScheduleModelConfig(settings, { + providerId: task.providerId, + model: task.model, + reasoningEffort: task.reasoningEffort + }) const result = await this.runPrompt(settings, { prompt: task.prompt, + preservePrompt: task.scheduledSend?.kind === 'thread-send', title: scheduledThreadTitle(task.title), workspaceRoot, ...(task.sourceThreadId ? { threadId: task.sourceThreadId } : {}), model: modelConfig.model, ...(modelConfig.providerId ? { providerId: modelConfig.providerId } : {}), reasoningEffort: modelConfig.reasoningEffort, + ...(task.scheduledSend?.kind === 'thread-send' + ? { + accountId: task.scheduledSend.accountId, + attachmentIds: task.scheduledSend.attachmentIds, + clientRequestId: task.scheduledSend.clientRequestId + } + : {}), mode: task.mode, orchestration: task.orchestration ?? 'direct', clawChannel, @@ -437,10 +465,20 @@ export class ScheduleExecutionQueue { }) if (this.stopped) return { ok: false, message: 'Schedule runtime stopped.' } if (!result.ok) { + if (task.scheduledSend?.kind === 'thread-send' && isRetryableScheduledSendResult(result, scheduledSendAttempt, task.scheduledSend.maxAttempts)) { + const retryAt = new Date(Date.now() + retryDelayMs(scheduledSendAttempt)) + this.runningTaskIds.delete(task.id) + await this.updateTask(task.id, (current) => ({ ...current, lastStatus: 'queued', lastMessage: `Retrying scheduled send (${scheduledSendAttempt}/${current.scheduledSend?.maxAttempts ?? DEFAULT_SCHEDULED_SEND_MAX_ATTEMPTS}).`, nextRunAt: retryAt.toISOString(), updatedAt: new Date().toISOString() })) + this.queuedTaskIds.add(task.id) + this.queuedTaskModes.set(task.id, scheduled) + this.scheduleDrainAt(retryAt.getTime()) + return { ok: true, threadId: '', queued: true, message: 'Scheduled send queued for retry.' } + } const finishedAt = new Date() await this.updateTask(task.id, (current) => ({ ...current, ...(current.schedule.kind === 'at' ? { enabled: false } : {}), + ...(current.scheduledSend?.kind === 'thread-send' ? { scheduledSend: { ...current.scheduledSend, reconciliationPending: false } } : {}), lastRunAt: finishedAt.toISOString(), nextRunAt: current.schedule.kind === 'at' ? '' : computeScheduleNextRunAt(current, finishedAt), lastStatus: 'error', @@ -448,12 +486,13 @@ export class ScheduleExecutionQueue { updatedAt: finishedAt.toISOString() })) this.runningTaskIds.delete(task.id) + this.admittedTaskIds.delete(task.id) await this.releaseTaskWorktree(task.id) void this.drainQueue() return result } - const startedAt = new Date() + this.admittedTaskIds.add(task.id) await this.updateTask(task.id, (current) => ({ ...current, lastRunAt: startedAt.toISOString(), @@ -461,6 +500,7 @@ export class ScheduleExecutionQueue { lastStatus: 'running', lastMessage: result.message ?? 'Started', lastThreadId: result.threadId, + ...(current.scheduledSend?.kind === 'thread-send' ? { scheduledSend: { ...current.scheduledSend, reconciliationPending: false } } : {}), updatedAt: startedAt.toISOString() })) this.trackTask(Promise.resolve(this.monitorTaskTurn(task.id, result.threadId, result.turnId ?? ''))) @@ -468,22 +508,32 @@ export class ScheduleExecutionQueue { } catch (error) { if (this.stopped) return { ok: false, message: 'Schedule runtime stopped.' } const message = error instanceof Error ? error.message : String(error) + if (task.scheduledSend?.kind === 'thread-send' && scheduledSendAttempt < task.scheduledSend.maxAttempts && isRetryableScheduledSendError(error)) { + const delay = retryDelayMs(scheduledSendAttempt) + this.runningTaskIds.delete(task.id) + await this.updateTask(task.id, (current) => ({ ...current, lastStatus: 'queued', lastMessage: `Retrying scheduled send (${scheduledSendAttempt}/${current.scheduledSend?.maxAttempts ?? DEFAULT_SCHEDULED_SEND_MAX_ATTEMPTS}).`, nextRunAt: new Date(Date.now() + delay).toISOString(), updatedAt: new Date().toISOString() })) + this.queuedTaskIds.add(task.id) + this.queuedTaskModes.set(task.id, scheduled) + this.scheduleDrainAt(Date.now() + delay) + return { ok: true, threadId: '', queued: true, message: 'Scheduled send queued for retry.' } + } const finishedAt = new Date() await this.updateTask(task.id, (current) => ({ ...current, lastRunAt: finishedAt.toISOString(), + ...(current.scheduledSend?.kind === 'thread-send' ? { scheduledSend: { ...current.scheduledSend, reconciliationPending: false } } : {}), nextRunAt: computeScheduleNextRunAt(current, finishedAt), lastStatus: 'error', lastMessage: message, updatedAt: finishedAt.toISOString() })) this.runningTaskIds.delete(task.id) + this.admittedTaskIds.delete(task.id) await this.releaseTaskWorktree(task.id) void this.drainQueue() return { ok: false, message } } } - async monitorTaskTurn(taskId: string, threadId: string, turnId: string): Promise { try { const settings = await this.loadSettings() @@ -523,11 +573,11 @@ export class ScheduleExecutionQueue { this.deps.logError('schedule-task', 'Scheduled task failed', { message, taskId, threadId }) } finally { this.runningTaskIds.delete(taskId) + this.admittedTaskIds.delete(taskId) await this.releaseTaskWorktree(taskId) void this.drainQueue() } } - private async releaseTaskWorktree(taskId: string): Promise { const lease = this.worktreeLeases.get(taskId) if (!lease) return @@ -539,9 +589,10 @@ export class ScheduleExecutionQueue { }) }) } - runPrompt(settings: AppSettingsV1, options: RunPromptOptions): Promise { - const prompt = options.clawChannel + const prompt = options.preservePrompt + ? options.prompt + : options.clawChannel ? buildClawRuntimePrompt(settings, options.prompt, { channel: options.clawChannel }) : buildScheduleRuntimePrompt(settings, options.prompt) return runPromptViaRuntime(this.deps, settings, { @@ -552,6 +603,9 @@ export class ScheduleExecutionQueue { model: options.model, ...(options.providerId ? { providerId: options.providerId } : {}), reasoningEffort: options.reasoningEffort, + ...(options.accountId ? { accountId: options.accountId } : {}), + ...(options.attachmentIds?.length ? { attachmentIds: options.attachmentIds } : {}), + ...(options.clientRequestId ? { clientRequestId: options.clientRequestId } : {}), mode: options.mode, orchestration: options.orchestration ?? 'direct', waitForResult: options.waitForResult, @@ -559,7 +613,6 @@ export class ScheduleExecutionQueue { ...(options.signal ? { signal: options.signal } : {}) }) } - waitForAssistantText( settings: AppSettingsV1, threadId: string, @@ -571,7 +624,6 @@ export class ScheduleExecutionQueue { void workspaceRoot return waitForAssistantTextViaRuntime(this.deps, settings, threadId, turnId, timeoutMs, signal) } - private trackTask(task: Promise): Promise { this.activeTasks.add(task) void task.then( @@ -580,25 +632,32 @@ export class ScheduleExecutionQueue { ) return task } - + private scheduleDrainAt(at: number): void { + if (this.stopped) return + if (this.wakeTimer && this.wakeAt <= at) return + if (this.wakeTimer) clearTimeout(this.wakeTimer) + this.wakeAt = at + this.wakeTimer = setTimeout(() => { + this.wakeTimer = null + this.wakeAt = 0 + void this.drainQueue() + }, Math.max(0, at - Date.now())) + this.wakeTimer.unref?.() + } resolveDefaultWorkspaceRoot(settings: AppSettingsV1): string { return settings.schedule.defaultWorkspaceRoot.trim() || settings.workspaceRoot } - resolveClawChannel(settings: AppSettingsV1, channelId: string | null | undefined): ClawImChannelV1 | null { const id = channelId?.trim() if (!id) return null return settings.claw.channels.find((channel) => channel.id === id) ?? null } - private resolveTaskClawChannel(settings: AppSettingsV1, task: ScheduledTaskV1): ClawImChannelV1 | null { return this.resolveClawChannel(settings, task.clawChannelId) } - resolveClawChannelWorkspaceRoot(settings: AppSettingsV1, channel: ClawImChannelV1): string { return channel.workspaceRoot.trim() || settings.claw.im.workspaceRoot.trim() || this.resolveDefaultWorkspaceRoot(settings) } - private resolveTaskWorkspaceRoot( settings: AppSettingsV1, task: ScheduledTaskV1, @@ -607,9 +666,23 @@ export class ScheduleExecutionQueue { return task.workspaceRoot.trim() || (channel ? this.resolveClawChannelWorkspaceRoot(settings, channel) : this.resolveDefaultWorkspaceRoot(settings)) } - } - +function retryDelayMs(attempt: number): number { return Math.min(SCHEDULED_SEND_RETRY_DELAY_MS * 2 ** Math.max(0, attempt - 1), 30_000) } +function isRetryableScheduledSendResult( + result: Extract, + attempt: number, + maxAttempts: number +): boolean { + if (attempt >= maxAttempts) return false + if (result.status === 409 && /thread_busy|active turn|thread already/i.test(result.message)) return true + if (result.status === 408 || result.status === 425 || result.status === 429) return true + if (result.status === 0 && (result.code === 'fetch_failed' || result.code === 'runtime_offline' || result.code === 'runtime_request_failed')) return true + return typeof result.status === 'number' && result.status >= 500 +} +function isRetryableScheduledSendError(error: unknown): boolean { + if (!(error instanceof Error)) return false + return /fetch failed|econnrefused|econnreset|socket|timed out|timeout|network|connect/.test(`${error.name} ${error.message}`.toLowerCase()) +} export function hasTaskDependencyCycle(taskId: string, tasks: readonly ScheduledTaskV1[]): boolean { const dependencies = new Map(tasks.map((task) => [task.id, task.dependsOn ?? []])) const visiting = new Set() @@ -621,9 +694,7 @@ export function hasTaskDependencyCycle(taskId: string, tasks: readonly Scheduled for (const dependency of dependencies.get(id) ?? []) { if (visit(dependency)) return true } - visiting.delete(id) - visited.add(id) + visiting.delete(id); visited.add(id) return false } - return visit(taskId) -} + return visit(taskId) } diff --git a/src/main/schedule-runtime.scheduled-send.test.ts b/src/main/schedule-runtime.scheduled-send.test.ts new file mode 100644 index 000000000..5e473eb31 --- /dev/null +++ b/src/main/schedule-runtime.scheduled-send.test.ts @@ -0,0 +1,677 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + defaultClawSettings, + defaultDesignSettings, + defaultKeyboardShortcuts, + defaultKunRuntimeSettings, + defaultModelProviderSettings, + defaultScheduleSettings, + defaultTerminalSettings, + defaultWorkflowSettings, + defaultWriteSettings, + mergeScheduleSettings, + type AppSettingsPatch, + type AppSettingsV1, + type ScheduledTaskV1 +} from '../shared/app-settings' +import { ScheduleRuntime } from './schedule-runtime' +let workspaceRoot = '' +function scheduledSendTask(overrides: Partial = {}): ScheduledTaskV1 { + const base = { + id: 'scheduled-send-1', + title: 'Continue investigation', + enabled: true, + prompt: 'Continue the existing investigation', + workspaceRoot, + sourceThreadId: 'thread-existing', + clawChannelId: '', + providerId: 'deepseek', + model: 'deepseek-v4-flash', + reasoningEffort: 'high', + mode: 'agent', + orchestration: 'direct', + priority: 0, + dependsOn: [], + useWorktree: false, + scheduledSend: { + kind: 'thread-send', + clientRequestId: 'scheduled-send:scheduled-send-1', + accountId: 'account-a', + attachmentIds: ['attachment-a', 'attachment-b'], + attemptCount: 0, + maxAttempts: 3 + }, + schedule: { kind: 'manual', everyMinutes: 60, timeOfDay: '09:00', atTime: '' }, + createdAt: '2026-08-30T00:00:00.000Z', + updatedAt: '2026-08-30T00:00:00.000Z', + lastRunAt: '', + nextRunAt: '', + lastStatus: 'idle', + lastMessage: '', + lastThreadId: '' + } as ScheduledTaskV1 + return { + ...base, + ...overrides, + scheduledSend: overrides.scheduledSend ?? base.scheduledSend, + schedule: overrides.schedule ?? base.schedule + } +} +function settingsWith(taskOrTasks: ScheduledTaskV1 | ScheduledTaskV1[]): AppSettingsV1 { + const tasks = Array.isArray(taskOrTasks) ? taskOrTasks : [taskOrTasks] + return { + version: 1, + locale: 'en', + theme: 'system', + uiFontScale: 0.82, + chatContentMaxWidthPx: 896, + composerSendKey: 'enter', + provider: defaultModelProviderSettings(), + agents: { kun: { ...defaultKunRuntimeSettings(), apiKey: 'test-key' } }, + workspaceRoot, + conversationWorkspaceRoot: '~/Documents/Kun', + log: { enabled: true, retentionDays: 7 }, + checkpointCleanup: { createEnabled: false, enabled: false, intervalDays: 3 }, + notifications: { turnComplete: true }, + appBehavior: { openAtLogin: false, startMinimized: false, closeToTray: false }, + keyboardShortcuts: defaultKeyboardShortcuts(), + write: defaultWriteSettings(), + claw: defaultClawSettings(), + schedule: mergeScheduleSettings(defaultScheduleSettings(), { enabled: true, tasks }), + workflow: defaultWorkflowSettings(), + design: defaultDesignSettings(), + terminal: defaultTerminalSettings(), + guiUpdate: { channel: 'stable' }, + codePromptPrefix: '', + chatWelcomeMessage: '', + codeAgentPresets: [], + disabledSkillIds: [] + } +} +function createStore(initial: AppSettingsV1) { + let current = initial + return { + read: () => current, + load: vi.fn(async () => current), + patch: vi.fn(async (partial: AppSettingsPatch) => { + current = { + ...current, + schedule: mergeScheduleSettings(current.schedule, partial.schedule) + } + return current + }), + update: vi.fn(async ( + mutation: (settings: AppSettingsV1) => AppSettingsV1 | Promise + ) => { + current = await mutation(current) + return current + }) + } +} +describe('ScheduleRuntime existing-thread scheduled send', () => { + beforeEach(() => { + workspaceRoot = mkdtempSync(join(tmpdir(), 'kun-scheduled-send-')) + }) + afterEach(() => { + vi.useRealTimers() + rmSync(workspaceRoot, { recursive: true, force: true }) + workspaceRoot = '' + }) + it('posts the frozen request snapshot to the existing thread with a retry-stable admission key', async () => { + const task = scheduledSendTask() + const settings = settingsWith(task) + const requests: Array> = [] + const runtimeRequest = vi.fn(async ( + _settings: AppSettingsV1, + path: string, + init: { method?: string; body?: string } + ) => { + expect(path).toBe('/v1/threads/thread-existing/turns') + expect(init.method).toBe('POST') + requests.push(JSON.parse(init.body ?? '{}')) + return requests.length === 1 + ? { ok: false, status: 400, body: 'invalid request' } + : { ok: true, status: 202, body: JSON.stringify({ turnId: 'turn-existing' }) } + }) + const runtime = new ScheduleRuntime({ + store: createStore(settings) as never, + runtimeRequest: runtimeRequest as never, + logError: vi.fn() + }) + ;(runtime as unknown as { monitorTaskTurn: () => void }).monitorTaskTurn = vi.fn() + await expect(runtime.runTask(task.id)).resolves.toMatchObject({ ok: false }) + await new Promise((resolve) => setTimeout(resolve, 0)) + await expect(runtime.runTask(task.id)).resolves.toMatchObject({ + ok: true, + threadId: 'thread-existing', + turnId: 'turn-existing' + }) + expect(requests).toHaveLength(2) + expect(requests[0]).toMatchObject({ + prompt: expect.stringContaining('Continue the existing investigation'), + providerId: 'deepseek', + accountId: 'account-a', + model: 'deepseek-v4-flash', + reasoningEffort: 'high', + attachmentIds: ['attachment-a', 'attachment-b'] + }) + expect(requests[0]?.clientRequestId).toEqual(expect.any(String)) + expect(requests[1]?.clientRequestId).toBe(requests[0]?.clientRequestId) + expect(runtimeRequest.mock.calls.some(([, path]) => path === '/v1/threads')).toBe(false) + }) + it('automatically retries a transient busy response with the same admission key', async () => { + vi.useFakeTimers() + const task = scheduledSendTask() + const store = createStore(settingsWith(task)) + const requests: Array> = [] + const runtimeRequest = vi.fn(async ( + _settings: AppSettingsV1, + path: string, + init: { body?: string } + ) => { + expect(path).toBe('/v1/threads/thread-existing/turns') + requests.push(JSON.parse(init.body ?? '{}')) + return requests.length === 1 + ? { + ok: false, + status: 409, + body: JSON.stringify({ code: 'thread_busy', message: 'thread already has an active turn' }) + } + : { ok: true, status: 202, body: JSON.stringify({ turnId: 'turn-retried' }) } + }) + const runtime = new ScheduleRuntime({ + store: store as never, + runtimeRequest: runtimeRequest as never, + logError: vi.fn() + }) + ;(runtime as unknown as { waitForAssistantText: () => Promise }).waitForAssistantText = + vi.fn(async () => 'retried response') + await expect(runtime.runTask(task.id)).resolves.toMatchObject({ ok: true, queued: true }) + expect(requests).toHaveLength(1) + await vi.advanceTimersByTimeAsync(1_000) + await vi.waitFor(() => expect(requests).toHaveLength(2)) + expect(requests[1]?.clientRequestId).toBe(requests[0]?.clientRequestId) + expect(store.load).toHaveBeenCalled() + await vi.waitFor(async () => { + const persisted = (await store.load()).schedule.tasks[0] + expect(persisted.scheduledSend?.attemptCount).toBe(2) + expect(persisted.lastStatus).toBe('success') + }) + }) + it('runs two sends for the same thread FIFO and never overlaps their admissions', async () => { + const first = scheduledSendTask({ + id: 'send-first', + prompt: 'first', + createdAt: '2026-08-30T00:00:00.000Z', + scheduledSend: { + kind: 'thread-send', + clientRequestId: 'scheduled-send:send-first', + accountId: 'account-a', + attachmentIds: [], + attemptCount: 0, + maxAttempts: 3 + } + }) + const second = scheduledSendTask({ + id: 'send-second', + prompt: 'second', + createdAt: '2026-08-30T00:00:01.000Z', + scheduledSend: { + kind: 'thread-send', + clientRequestId: 'scheduled-send:send-second', + accountId: 'account-a', + attachmentIds: [], + attemptCount: 0, + maxAttempts: 3 + } + }) + let releaseFirst!: () => void + const firstAdmission = new Promise((resolve) => { releaseFirst = resolve }) + const prompts: string[] = [] + let activeAdmissions = 0 + let maxActiveAdmissions = 0 + const runtimeRequest = vi.fn(async ( + _settings: AppSettingsV1, + path: string, + init: { body?: string } + ) => { + expect(path).toBe('/v1/threads/thread-existing/turns') + const body = JSON.parse(init.body ?? '{}') as { prompt?: string } + prompts.push(body.prompt ?? '') + activeAdmissions += 1 + maxActiveAdmissions = Math.max(maxActiveAdmissions, activeAdmissions) + if (body.prompt === 'first') await firstAdmission + activeAdmissions -= 1 + return { ok: true, status: 202, body: JSON.stringify({ turnId: `turn-${body.prompt}` }) } + }) + const runtime = new ScheduleRuntime({ + store: createStore(settingsWith([first, second])) as never, + runtimeRequest: runtimeRequest as never, + logError: vi.fn() + }) + ;(runtime as unknown as { waitForAssistantText: () => Promise }).waitForAssistantText = + vi.fn(async () => 'done') + const firstResult = runtime.runTask(first.id) + const secondResult = runtime.runTask(second.id) + await vi.waitFor(() => expect(prompts).toEqual(['first'])) + releaseFirst() + await expect(firstResult).resolves.toMatchObject({ ok: true, turnId: 'turn-first' }) + await expect(secondResult).resolves.toMatchObject({ ok: true, turnId: 'turn-second' }) + expect(prompts).toEqual(['first', 'second']) + expect(maxActiveAdmissions).toBe(1) + }) + it('keeps same-thread FIFO order even when the newer send has higher priority', async () => { + const first = scheduledSendTask({ + id: 'send-fifo-first', + prompt: 'fifo first', + priority: 0, + createdAt: '2026-08-30T00:00:00.000Z' + }) + const second = scheduledSendTask({ + id: 'send-fifo-second', + prompt: 'fifo second', + priority: 100, + createdAt: '2026-08-30T00:00:01.000Z', + scheduledSend: { + kind: 'thread-send', + clientRequestId: 'scheduled-send:send-fifo-second', + accountId: '', + attachmentIds: [], + attemptCount: 0, + maxAttempts: 3 + } + }) + const prompts: string[] = [] + const runtime = new ScheduleRuntime({ + store: createStore(settingsWith([first, second])) as never, + runtimeRequest: vi.fn(async ( + _settings: AppSettingsV1, + _path: string, + init: { body?: string } + ) => { + const body = JSON.parse(init.body ?? '{}') as { prompt?: string } + prompts.push(body.prompt ?? '') + return { ok: true, status: 202, body: JSON.stringify({ turnId: `turn-${prompts.length}` }) } + }) as never, + logError: vi.fn() + }) + ;(runtime as unknown as { waitForAssistantText: () => Promise }).waitForAssistantText = + vi.fn(async () => 'done') + const queue = (runtime as unknown as { + queue: { enqueueTask: (task: ScheduledTaskV1, scheduled: boolean) => Promise } + }).queue + await Promise.all([ + queue.enqueueTask(first, false), + queue.enqueueTask(second, false) + ]) + await vi.waitFor(() => expect(prompts).toHaveLength(2)) + expect(prompts).toEqual(['fifo first', 'fifo second']) + }) + it('does not immediately run a queued task after it is edited to a future time', async () => { + vi.useFakeTimers() + vi.setSystemTime('2026-08-30T00:00:00.000Z') + const active = scheduledSendTask({ id: 'send-edit-active', prompt: 'edit active' }) + const rescheduled = scheduledSendTask({ + id: 'send-edit-future', + prompt: 'edit future', + scheduledSend: { + kind: 'thread-send', + clientRequestId: 'scheduled-send:send-edit-future', + accountId: '', + attachmentIds: [], + attemptCount: 0, + maxAttempts: 3 + } + }) + let releaseActive!: () => void + const activeAdmission = new Promise((resolve) => { releaseActive = resolve }) + const prompts: string[] = [] + const runtime = new ScheduleRuntime({ + store: createStore(settingsWith([active, rescheduled])) as never, + runtimeRequest: vi.fn(async ( + _settings: AppSettingsV1, + _path: string, + init: { body?: string } + ) => { + const body = JSON.parse(init.body ?? '{}') as { prompt?: string } + prompts.push(body.prompt ?? '') + if (body.prompt === 'edit active') await activeAdmission + return { ok: true, status: 202, body: JSON.stringify({ turnId: `turn-${body.prompt}` }) } + }) as never, + logError: vi.fn() + }) + ;(runtime as unknown as { waitForAssistantText: () => Promise }).waitForAssistantText = + vi.fn(async () => 'done') + const activeResult = runtime.runTask(active.id) + void runtime.runTask(rescheduled.id) + await vi.waitFor(() => expect(prompts).toEqual(['edit active'])) + await runtime.updateTaskById(rescheduled.id, { + schedule: { + kind: 'at', + atTime: '2026-08-30T00:01:00.000Z' + } + }) + releaseActive() + await expect(activeResult).resolves.toMatchObject({ ok: true }) + await vi.advanceTimersByTimeAsync(1_000) + expect(prompts).toEqual(['edit active']) + await runtime.stop() + }) + it('removes a cancelled same-thread queued send without a ghost admission', async () => { + const first = scheduledSendTask({ id: 'send-active', prompt: 'active' }) + const queued = scheduledSendTask({ + id: 'send-cancelled', + prompt: 'cancelled', + scheduledSend: { + kind: 'thread-send', + clientRequestId: 'scheduled-send:send-cancelled', + accountId: '', + attachmentIds: [], + attemptCount: 0, + maxAttempts: 3 + } + }) + let releaseActive!: () => void + const activeAdmission = new Promise((resolve) => { releaseActive = resolve }) + const prompts: string[] = [] + const runtimeRequest = vi.fn(async ( + _settings: AppSettingsV1, + _path: string, + init: { body?: string } + ) => { + const body = JSON.parse(init.body ?? '{}') as { prompt?: string } + prompts.push(body.prompt ?? '') + if (body.prompt === 'active') await activeAdmission + return { ok: true, status: 202, body: JSON.stringify({ turnId: `turn-${body.prompt}` }) } + }) + const runtime = new ScheduleRuntime({ + store: createStore(settingsWith([first, queued])) as never, + runtimeRequest: runtimeRequest as never, + logError: vi.fn() + }) + ;(runtime as unknown as { waitForAssistantText: () => Promise }).waitForAssistantText = + vi.fn(async () => 'done') + const activeResult = runtime.runTask(first.id) + const queuedResult = runtime.runTask(queued.id) + await vi.waitFor(() => expect(prompts).toEqual(['active'])) + await expect(runtime.deleteTaskById(queued.id)).resolves.toBe(true) + await expect(queuedResult).resolves.toEqual({ + ok: false, + message: 'Scheduled task was cancelled.' + }) + await expect(runtime.status()).resolves.toMatchObject({ queuedTaskIds: [] }) + releaseActive() + await expect(activeResult).resolves.toMatchObject({ ok: true }) + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(prompts).toEqual(['active']) + }) + it('does not admit a task deleted after dequeue but before the runtime POST', async () => { + const task = scheduledSendTask({ id: 'send-delete-race', prompt: 'delete race' }) + const store = createStore(settingsWith(task)) + let releaseLoad!: () => void + let signalRunningLoad!: () => void + const runningLoad = new Promise((resolve) => { signalRunningLoad = resolve }) + const loadGate = new Promise((resolve) => { releaseLoad = resolve }) + let gated = false + store.load.mockImplementation(async () => { + const current = store.read() + if ( + !gated && + current.schedule.tasks.some((candidate) => + candidate.id === task.id && candidate.lastStatus === 'running' + ) + ) { + gated = true + signalRunningLoad() + await loadGate + } + return store.read() + }) + const runtimeRequest = vi.fn(async () => ({ + ok: true, + status: 202, + body: JSON.stringify({ turnId: 'turn-should-not-exist' }) + })) + const runtime = new ScheduleRuntime({ + store: store as never, + runtimeRequest: runtimeRequest as never, + logError: vi.fn() + }) + ;(runtime as unknown as { waitForAssistantText: () => Promise }).waitForAssistantText = + vi.fn(async () => 'done') + const result = runtime.runTask(task.id) + await runningLoad + await expect(runtime.deleteTaskById(task.id)).resolves.toBe(true) + releaseLoad() + await expect(result).resolves.toMatchObject({ ok: false }) + expect(runtimeRequest).not.toHaveBeenCalled() + await expect(runtime.status()).resolves.toMatchObject({ + runningTaskIds: [], + queuedTaskIds: [] + }) + await runtime.stop() + }) + it('retries a status-0 fetch failure and preserves the original admission key', async () => { + vi.useFakeTimers() + const task = scheduledSendTask({ id: 'send-network-retry' }) + const bodies: Array> = [] + const runtimeRequest = vi.fn(async ( + _settings: AppSettingsV1, + _path: string, + init: { body?: string } + ) => { + bodies.push(JSON.parse(init.body ?? '{}')) + return bodies.length === 1 + ? { ok: false, status: 0, body: 'fetch failed: ECONNRESET' } + : { ok: true, status: 202, body: JSON.stringify({ turnId: 'turn-network-retry' }) } + }) + const runtime = new ScheduleRuntime({ + store: createStore(settingsWith(task)) as never, + runtimeRequest: runtimeRequest as never, + logError: vi.fn() + }) + ;(runtime as unknown as { waitForAssistantText: () => Promise }).waitForAssistantText = + vi.fn(async () => 'done') + await expect(runtime.runTask(task.id)).resolves.toMatchObject({ ok: true, queued: true }) + await vi.advanceTimersByTimeAsync(1_000) + await vi.waitFor(() => expect(bodies).toHaveLength(2)) + expect(bodies[1]?.clientRequestId).toBe(bodies[0]?.clientRequestId) + }) + it('keeps independent retry wakeups ordered by their earliest deadline', async () => { + vi.useFakeTimers() + vi.setSystemTime('2026-08-30T00:00:00.000Z') + const early = scheduledSendTask({ + id: 'send-wake-early', + sourceThreadId: 'thread-early', + scheduledSend: { + kind: 'thread-send', + clientRequestId: 'scheduled-send:send-wake-early', + accountId: '', + attachmentIds: [], + attemptCount: 0, + maxAttempts: 3 + } + }) + const later = scheduledSendTask({ + id: 'send-wake-later', + sourceThreadId: 'thread-later', + scheduledSend: { + kind: 'thread-send', + clientRequestId: 'scheduled-send:send-wake-later', + accountId: '', + attachmentIds: [], + attemptCount: 1, + maxAttempts: 3 + } + }) + const counts = new Map() + const runtimeRequest = vi.fn(async ( + _settings: AppSettingsV1, + _path: string, + init: { body?: string } + ) => { + const body = JSON.parse(init.body ?? '{}') as { clientRequestId?: string } + const key = body.clientRequestId ?? '' + const count = (counts.get(key) ?? 0) + 1 + counts.set(key, count) + return count === 1 + ? { ok: false, status: 503, body: 'temporarily unavailable' } + : { ok: true, status: 202, body: JSON.stringify({ turnId: `turn-${key}` }) } + }) + const runtime = new ScheduleRuntime({ + store: createStore(settingsWith([early, later])) as never, + runtimeRequest: runtimeRequest as never, + logError: vi.fn() + }) + ;(runtime as unknown as { waitForAssistantText: () => Promise }).waitForAssistantText = + vi.fn(async () => 'done') + await Promise.all([runtime.runTask(early.id), runtime.runTask(later.id)]) + expect(runtimeRequest).toHaveBeenCalledTimes(2) + await vi.advanceTimersByTimeAsync(999) + expect(runtimeRequest).toHaveBeenCalledTimes(2) + await vi.advanceTimersByTimeAsync(1) + await vi.waitFor(() => expect(runtimeRequest).toHaveBeenCalledTimes(3)) + expect(counts.get('scheduled-send:send-wake-early')).toBe(2) + expect(counts.get('scheduled-send:send-wake-later')).toBe(1) + await vi.advanceTimersByTimeAsync(999) + expect(runtimeRequest).toHaveBeenCalledTimes(3) + await vi.advanceTimersByTimeAsync(1) + await vi.waitFor(() => expect(runtimeRequest).toHaveBeenCalledTimes(4)) + expect(counts.get('scheduled-send:send-wake-later')).toBe(2) + }) + it.each([ + ['missing provider', 400, JSON.stringify({ code: 'provider_not_found', message: 'provider was removed' })], + ['missing account', 400, JSON.stringify({ code: 'account_not_found', message: 'account was removed' })] + ])('treats %s as terminal without changing the frozen route', async (_label, status, responseBody) => { + vi.useFakeTimers() + const task = scheduledSendTask({ providerId: 'removed-provider' }) + const bodies: Array> = [] + const runtimeRequest = vi.fn(async ( + _settings: AppSettingsV1, + path: string, + init: { body?: string } + ) => { + expect(path).toBe('/v1/threads/thread-existing/turns') + bodies.push(JSON.parse(init.body ?? '{}')) + return { ok: false, status, body: responseBody } + }) + const runtime = new ScheduleRuntime({ + store: createStore(settingsWith(task)) as never, + runtimeRequest: runtimeRequest as never, + logError: vi.fn() + }) + await expect(runtime.runTask(task.id)).resolves.toMatchObject({ ok: false }) + await vi.advanceTimersByTimeAsync(30_000) + expect(bodies).toHaveLength(1) + expect(bodies[0]).toMatchObject({ + providerId: 'removed-provider', + accountId: 'account-a', + model: 'deepseek-v4-flash' + }) + expect(runtimeRequest.mock.calls.some(([, path]) => path === '/v1/threads')).toBe(false) + }) + it.each([ + ['sourceThreadId', { sourceThreadId: '' }], + ['clientRequestId', { + scheduledSend: { + kind: 'thread-send', + clientRequestId: '', + accountId: '', + attachmentIds: [], + attemptCount: 0, + maxAttempts: 3 + } + }] + ])('fails closed when persisted scheduled send %s is empty', async (_field, overrides) => { + const task = scheduledSendTask(overrides as Partial) + const runtimeRequest = vi.fn(async () => ({ ok: false, status: 500, body: 'unexpected request' })) + const runtime = new ScheduleRuntime({ + store: createStore(settingsWith(task)) as never, + runtimeRequest: runtimeRequest as never, + logError: vi.fn() + }) + await expect(runtime.runTask(task.id)).resolves.toMatchObject({ ok: false }) + expect(runtimeRequest).not.toHaveBeenCalled() + }) + it('recovers interrupted running and queued scheduled sends once after restart', async () => { + const atSchedule = { + kind: 'at' as const, + everyMinutes: 60, + timeOfDay: '09:00', + atTime: '2026-08-30T00:00:00.000Z' + } + const interrupted = scheduledSendTask({ + id: 'send-interrupted', + prompt: 'interrupted', + lastStatus: 'running', + schedule: atSchedule + }) + const queued = scheduledSendTask({ + id: 'send-persisted-queued', + prompt: 'persisted queued', + lastStatus: 'queued', + schedule: atSchedule, + scheduledSend: { + kind: 'thread-send', + clientRequestId: 'scheduled-send:send-persisted-queued', + accountId: '', + attachmentIds: [], + attemptCount: 1, + maxAttempts: 3 + } + }) + const prompts: string[] = [] + const settings = settingsWith([interrupted, queued]) + const runtimeRequest = vi.fn(async ( + _settings: AppSettingsV1, + _path: string, + init: { body?: string } + ) => { + const body = JSON.parse(init.body ?? '{}') as { prompt?: string } + prompts.push(body.prompt ?? '') + return { ok: true, status: 202, body: JSON.stringify({ turnId: `turn-${prompts.length}` }) } + }) + const runtime = new ScheduleRuntime({ + store: createStore(settings) as never, + runtimeRequest: runtimeRequest as never, + logError: vi.fn() + }) + ;(runtime as unknown as { waitForAssistantText: () => Promise }).waitForAssistantText = + vi.fn(async () => 'done') + await (runtime as unknown as { + ensureNextRuns: (value: AppSettingsV1) => Promise + }).ensureNextRuns(settings) + await (runtime as unknown as { + queue: { drainQueue: () => Promise } + }).queue.drainQueue() + await vi.waitFor(() => expect(prompts).toHaveLength(2)) + expect(prompts).toEqual(['interrupted', 'persisted queued']) + expect(runtimeRequest).toHaveBeenCalledTimes(2) + }) + it.each([ + [409, JSON.stringify({ code: 'conflict', message: 'thread is archived: thread-existing' })], + [404, JSON.stringify({ code: 'not_found', message: 'thread not found: thread-existing' })] + ])('does not create a replacement thread when the bound thread fails with %s', async (status, body) => { + vi.useFakeTimers() + const task = scheduledSendTask() + const runtimeRequest = vi.fn(async ( + _settings: AppSettingsV1, + path: string + ) => { + expect(path).toBe('/v1/threads/thread-existing/turns') + return { ok: false, status, body } + }) + const runtime = new ScheduleRuntime({ + store: createStore(settingsWith(task)) as never, + runtimeRequest: runtimeRequest as never, + logError: vi.fn() + }) + await expect(runtime.runTask(task.id)).resolves.toMatchObject({ ok: false }) + await vi.advanceTimersByTimeAsync(30_000) + expect(runtimeRequest).toHaveBeenCalledTimes(1) + expect(runtimeRequest.mock.calls.some(([, path]) => path === '/v1/threads')).toBe(false) + }) +}) diff --git a/src/main/schedule-runtime.ts b/src/main/schedule-runtime.ts index a953ac059..faa08387c 100644 --- a/src/main/schedule-runtime.ts +++ b/src/main/schedule-runtime.ts @@ -15,7 +15,9 @@ import { DEFAULT_SCHEDULE_MODEL, DEFAULT_SCHEDULE_REASONING_EFFORT, buildClawRuntimePrompt, - buildScheduleRuntimePrompt + buildScheduleRuntimePrompt, + getModelProviderSettings, + normalizeScheduleReasoningEffort } from '../shared/app-settings' import { buildScheduledTaskFromDetectedRequest, @@ -61,6 +63,24 @@ export { hasTaskDependencyCycle, scheduledThreadTitle } from './schedule-runtime-queue' + +function resolveExactScheduledSendModelConfig( + settings: AppSettingsV1, + providerId: string, + model: string, + reasoningEffort: ScheduleReasoningEffort | undefined +): ScheduleModelConfig { + const provider = getModelProviderSettings(settings).providers.find((candidate) => candidate.id === providerId) + if (!provider) throw new Error(`Selected provider is no longer available: ${providerId || '(empty)'}`) + const exactModel = provider.models.find((candidate) => candidate.trim().toLowerCase() === model.trim().toLowerCase()) + if (!exactModel) throw new Error(`Selected model is no longer available for ${providerId}: ${model || '(empty)'}`) + return { + providerId: provider.id, + model: exactModel, + reasoningEffort: normalizeScheduleReasoningEffort(reasoningEffort) + } +} + export class ScheduleRuntime { private readonly deps: ScheduleRuntimeDeps private scheduler: ReturnType | null = null @@ -261,6 +281,8 @@ export class ScheduleRuntime { sourcePlanId?: string sourceThreadId?: string providerId?: string + accountId?: string + attachmentIds?: string[] model?: string reasoningEffort?: ScheduleReasoningEffort mode?: ScheduleRunMode @@ -271,22 +293,41 @@ export class ScheduleRuntime { }): Promise { const settings = await this.loadSettings() const clawChannel = this.queue.resolveClawChannel(settings, input.clawChannelId) - const modelConfig = this.resolveScheduleModelConfig(settings, { - providerId: input.providerId ?? settings.schedule.providerId, - model: input.model?.trim() || clawChannel?.model.trim() || settings.schedule.model || DEFAULT_SCHEDULE_MODEL, - reasoningEffort: input.reasoningEffort ?? DEFAULT_SCHEDULE_REASONING_EFFORT - }) + const requestedProviderId = input.providerId?.trim() || settings.schedule.providerId?.trim() || '' + const requestedModel = input.model?.trim() || clawChannel?.model.trim() || settings.schedule.model || DEFAULT_SCHEDULE_MODEL + const modelConfig = input.sourceThreadId?.trim() && !input.sourcePlanId?.trim() + ? resolveExactScheduledSendModelConfig(settings, requestedProviderId, requestedModel, input.reasoningEffort) + : this.resolveScheduleModelConfig(settings, { + providerId: requestedProviderId, + model: requestedModel, + reasoningEffort: input.reasoningEffort ?? DEFAULT_SCHEDULE_REASONING_EFFORT + }) const now = new Date().toISOString() + const taskId = randomUUID() + const sourcePlanId = input.sourcePlanId?.trim() || '' + const sourceThreadId = input.sourceThreadId?.trim() || '' const task: ScheduledTaskV1 = { - id: randomUUID(), + id: taskId, title: input.title.trim() || 'New scheduled task', enabled: input.enabled !== false, prompt: input.prompt, workspaceRoot: input.workspaceRoot?.trim() || (clawChannel ? this.queue.resolveClawChannelWorkspaceRoot(settings, clawChannel) : this.queue.resolveDefaultWorkspaceRoot(settings)), - sourcePlanId: input.sourcePlanId?.trim() || '', - sourceThreadId: input.sourceThreadId?.trim() || '', + sourcePlanId, + sourceThreadId, + ...(!sourcePlanId && sourceThreadId + ? { + scheduledSend: { + kind: 'thread-send' as const, + clientRequestId: `scheduled-send:${taskId}`, + accountId: input.accountId?.trim() || '', + attachmentIds: [...new Set(input.attachmentIds ?? [])].slice(0, 8), + attemptCount: 0, + maxAttempts: 3 + } + } + : {}), clawChannelId: clawChannel?.id ?? '', providerId: modelConfig.providerId, model: modelConfig.model, @@ -320,38 +361,56 @@ export class ScheduleRuntime { taskId: string, patch: Omit, 'schedule'> & { schedule?: Partial } ): Promise { - const settings = await this.loadSettings() - const task = settings.schedule.tasks.find((item) => item.id === taskId) - if (!task) return null const now = new Date().toISOString() const shouldRecomputeNextRun = Object.prototype.hasOwnProperty.call(patch, 'enabled') || patch.schedule !== undefined - const nextTask: ScheduledTaskV1 = { - ...task, - ...patch, - schedule: patch.schedule ? { ...task.schedule, ...patch.schedule } : task.schedule, - ...(shouldRecomputeNextRun ? { nextRunAt: '' } : {}), - updatedAt: now - } - const saved = await this.deps.store.patch({ + let nextTask: ScheduledTaskV1 | null = null + const saved = await this.deps.store.update((current) => ({ + ...current, schedule: { - tasks: settings.schedule.tasks.map((item) => (item.id === taskId ? nextTask : item)) + ...current.schedule, + tasks: current.schedule.tasks.map((item) => { + if (item.id !== taskId) return item + nextTask = { + ...item, + ...patch, + schedule: patch.schedule ? { ...item.schedule, ...patch.schedule } : item.schedule, + ...(shouldRecomputeNextRun ? { nextRunAt: '' } : {}), + updatedAt: now + } + return nextTask + }) } - }) + })) + if (!nextTask) return null + const updatedTask = nextTask as ScheduledTaskV1 + if (updatedTask.scheduledSend?.kind === 'thread-send' && patch.schedule) { + this.queue.markQueuedScheduled(taskId) + } this.sync(saved) if (shouldRecomputeNextRun) await this.queue.ensureNextRuns(await this.loadSettings()) const latest = await this.loadSettings() - return latest.schedule.tasks.find((item) => item.id === taskId) ?? nextTask + return latest.schedule.tasks.find((item) => item.id === taskId) ?? updatedTask } async deleteTaskById(taskId: string): Promise { - const settings = await this.loadSettings() - if (!settings.schedule.tasks.some((item) => item.id === taskId)) return false - const saved = await this.deps.store.patch({ - schedule: { - tasks: settings.schedule.tasks.filter((item) => item.id !== taskId) + if (this.queue.hasRunning(taskId) && this.queue.hasAdmitted(taskId)) { + throw new Error('A running scheduled task cannot be deleted after admission. Stop the turn first.') + } + this.queue.cancelTask(taskId) + let found = false + const saved = await this.deps.store.update((current) => { + found = current.schedule.tasks.some((item) => item.id === taskId) + if (!found) return current + return { + ...current, + schedule: { + ...current.schedule, + tasks: current.schedule.tasks.filter((item) => item.id !== taskId) + } } }) + if (!found) return false this.sync(saved) return saved.schedule.tasks.every((item) => item.id !== taskId) } diff --git a/src/renderer/src/components/chat/FloatingComposer.scheduled-send.test.ts b/src/renderer/src/components/chat/FloatingComposer.scheduled-send.test.ts new file mode 100644 index 000000000..6f1c0892a --- /dev/null +++ b/src/renderer/src/components/chat/FloatingComposer.scheduled-send.test.ts @@ -0,0 +1,297 @@ +/** @vitest-environment jsdom */ +import { act, createElement } from 'react' +import { createRoot, type Root } from 'react-dom/client' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { ModelProviderModelGroup } from '@shared/kun-gui-api' +import { useChatStore } from '../../store/chat-store' +import i18n from '../../i18n' +import { FloatingComposer } from './FloatingComposer' +import { ScheduledSendDialog } from './ScheduledSendDialog' + +const MODEL_GROUPS: ModelProviderModelGroup[] = [{ + providerId: 'provider-a', + accountId: 'account-a', + label: 'Provider A', + modelIds: ['model-a'] +}] + +function deferred(): { + promise: Promise + resolve: (value: T) => void +} { + let resolve!: (value: T) => void + const promise = new Promise((next) => { resolve = next }) + return { promise, resolve } +} + +function setReactActEnvironment(value: boolean): void { + ;(globalThis as typeof globalThis & { IS_REACT_ACT_ENVIRONMENT: boolean }) + .IS_REACT_ACT_ENVIRONMENT = value +} + +async function changeControlValue( + control: HTMLInputElement | HTMLSelectElement, + value: string +): Promise { + const prototype = control instanceof HTMLSelectElement + ? HTMLSelectElement.prototype + : HTMLInputElement.prototype + Object.getOwnPropertyDescriptor(prototype, 'value')?.set?.call(control, value) + await act(async () => { + control.dispatchEvent(new Event('change', { bubbles: true })) + }) +} + +function composerProps(overrides: Record = {}) { + return { + input: 'Follow up using the current context', + setInput: vi.fn(), + mode: 'agent' as const, + setMode: vi.fn(), + busy: false, + runtimeReady: true, + hasActiveThread: true, + workspaceRootOverride: '/workspace/project', + composerModel: 'model-a', + composerProviderId: 'provider-a', + composerModelGroups: MODEL_GROUPS, + composerPickList: ['model-a'], + composerReasoningEffort: 'high', + onComposerModelChange: vi.fn(), + queuedMessages: [] as [], + onRemoveQueuedMessage: vi.fn(), + onSend: vi.fn(), + onInterrupt: vi.fn(), + attachmentUploadEnabled: true, + attachments: [{ id: 'attachment-a', kind: 'document' as const, name: 'notes.txt' }], + onRemoveAttachment: vi.fn(), + ...overrides + } +} + +describe('FloatingComposer scheduled send', () => { + let container: HTMLDivElement + let root: Root + let createScheduleTask: ReturnType + + beforeEach(() => { + setReactActEnvironment(true) + vi.useFakeTimers() + vi.setSystemTime(new Date('2030-05-10T08:00:00.000Z')) + container = document.createElement('div') + document.body.append(container) + root = createRoot(container) + createScheduleTask = vi.fn() + Object.defineProperty(window, 'kunGui', { + configurable: true, + value: { + createScheduleTask, + getSettings: vi.fn().mockRejectedValue(new Error('settings not needed by this test')), + runtimeRequest: vi.fn().mockResolvedValue({ ok: true, status: 200, body: '{"sessions":[],"running":0}' }) + } + }) + useChatStore.setState({ + activeThreadId: 'thread-a', + activeThreadGoal: null, + activeThreadTodos: null, + blocks: [], + route: 'chat', + workspaceRoot: '/workspace/project', + threads: [{ + id: 'thread-a', + title: 'Existing Thread', + updatedAt: '2030-05-10T07:00:00.000Z', + model: 'model-a', + mode: 'agent', + workspace: '/workspace/project' + }] + }) + }) + + afterEach(async () => { + await act(async () => root.unmount()) + container.remove() + Reflect.deleteProperty(window, 'kunGui') + vi.useRealTimers() + setReactActEnvironment(false) + }) + + it('opens the dialog and freezes the current Thread selection in a future task payload', async () => { + createScheduleTask.mockResolvedValue({ ok: true, task: { id: 'scheduled-a' } }) + const props = composerProps() + await act(async () => root.render(createElement(FloatingComposer, props))) + + const trigger = container.querySelector('.ds-composer-scheduled-send-action') + expect(trigger).not.toBeNull() + expect(trigger?.getAttribute('aria-label')).toBeTruthy() + + await act(async () => trigger?.click()) + expect(container.querySelector('[role="dialog"]')).not.toBeNull() + expect(container.querySelector('[data-scheduled-send-date]')).not.toBeNull() + expect(container.querySelector('[data-scheduled-send-time]')).not.toBeNull() + + const confirm = container.querySelector('[data-scheduled-send-confirm]') + expect(confirm?.disabled).toBe(false) + await act(async () => confirm?.click()) + + expect(createScheduleTask).toHaveBeenCalledOnce() + const payload = createScheduleTask.mock.calls[0]?.[0] + expect(new Date(payload.schedule.atTime).getTime()).toBeGreaterThan(Date.now()) + expect(payload).toMatchObject({ + prompt: 'Follow up using the current context', + workspaceRoot: '/workspace/project', + sourceThreadId: 'thread-a', + providerId: 'provider-a', + accountId: 'account-a', + model: 'model-a', + reasoningEffort: 'high', + attachmentIds: ['attachment-a'], + schedule: { kind: 'at' } + }) + expect(props.onSend).not.toHaveBeenCalled() + expect(props.setInput).toHaveBeenCalledWith('') + expect(props.onRemoveAttachment).toHaveBeenCalledWith('attachment-a') + expect(container.querySelector('[role="dialog"]')).toBeNull() + }) + + it('uses the shared localized DST error instead of exposing the converter message', async () => { + await act(async () => root.render(createElement(ScheduledSendDialog, { + submitting: false, + error: '', + onClose: vi.fn(), + onSubmit: vi.fn() + }))) + await changeControlValue( + container.querySelector('[data-scheduled-send-date]')!, + '2030-03-10' + ) + await changeControlValue( + container.querySelector('[data-scheduled-send-time]')!, + '02:30' + ) + await changeControlValue( + container.querySelector('[data-scheduled-send-time-zone]')!, + 'America/New_York' + ) + + expect(container.querySelector('[role="alert"]')?.textContent).toBe( + i18n.t('planScheduleBuildErrorNonexistentTime', { ns: 'common' }) + ) + expect(container.querySelector('[data-scheduled-send-confirm]')?.disabled).toBe(true) + }) + + it('keeps the draft and attachments visible when task creation fails', async () => { + createScheduleTask.mockResolvedValue({ ok: false, message: 'schedule store unavailable' }) + const props = composerProps() + await act(async () => root.render(createElement(FloatingComposer, props))) + await act(async () => { + container.querySelector('.ds-composer-scheduled-send-action')?.click() + }) + await act(async () => { + container.querySelector('[data-scheduled-send-confirm]')?.click() + }) + + expect(container.querySelector('[role="dialog"]')).not.toBeNull() + expect(container.querySelector('[role="alert"]')?.textContent).toContain('schedule store unavailable') + expect(container.querySelector('textarea')?.value).toBe('Follow up using the current context') + expect(container.textContent).toContain('notes.txt') + expect(props.setInput).not.toHaveBeenCalled() + expect(props.onRemoveAttachment).not.toHaveBeenCalled() + expect(props.onSend).not.toHaveBeenCalled() + }) + + it('traps focus, closes on Escape, and returns focus to the opener', async () => { + const props = composerProps() + await act(async () => root.render(createElement(FloatingComposer, props))) + const trigger = container.querySelector('.ds-composer-scheduled-send-action')! + trigger.focus() + await act(async () => trigger.click()) + + const dialog = container.querySelector('[role="dialog"]')! + const date = container.querySelector('[data-scheduled-send-date]')! + const close = dialog.querySelector('button[aria-label]')! + const confirm = dialog.querySelector('[data-scheduled-send-confirm]')! + expect(document.activeElement).toBe(date) + + close.focus() + await act(async () => { + close.dispatchEvent(new KeyboardEvent('keydown', { key: 'Tab', shiftKey: true, bubbles: true })) + }) + expect(document.activeElement).toBe(confirm) + await act(async () => { + confirm.dispatchEvent(new KeyboardEvent('keydown', { key: 'Tab', bubbles: true })) + }) + expect(document.activeElement).toBe(close) + + await act(async () => { + dialog.dispatchEvent(new KeyboardEvent('keydown', { key: 'Escape', bubbles: true })) + }) + expect(container.querySelector('[role="dialog"]')).toBeNull() + expect(document.activeElement).toBe(trigger) + }) + + it('locks dismissal while submitting and does not clear a newer draft or Thread', async () => { + const pending = deferred<{ ok: true; task: { id: string } }>() + createScheduleTask.mockReturnValue(pending.promise) + const original = composerProps() + await act(async () => root.render(createElement(FloatingComposer, original))) + await act(async () => { + container.querySelector('.ds-composer-scheduled-send-action')?.click() + }) + await act(async () => { + container.querySelector('[data-scheduled-send-confirm]')?.click() + }) + + const dialog = container.querySelector('[role="dialog"]')! + expect(dialog.getAttribute('aria-busy')).toBe('true') + expect(Array.from(dialog.querySelectorAll('button, input, select')).every((control) => ( + (control as HTMLButtonElement | HTMLInputElement | HTMLSelectElement).disabled + ))).toBe(true) + await act(async () => { + dialog.dispatchEvent(new KeyboardEvent('keydown', { key: 'Escape', bubbles: true })) + }) + expect(container.querySelector('[role="dialog"]')).not.toBeNull() + + useChatStore.setState({ + activeThreadId: 'thread-b', + threads: [{ + id: 'thread-b', + title: 'New Thread', + updatedAt: '2030-05-10T07:30:00.000Z', + model: 'model-b', + mode: 'agent', + workspace: '/workspace/other' + }] + }) + const newer = composerProps({ + input: 'A newer draft', + workspaceRootOverride: '/workspace/other', + composerModel: 'model-b', + composerProviderId: 'provider-b', + composerModelGroups: [{ + providerId: 'provider-b', + accountId: 'account-b', + label: 'Provider B', + modelIds: ['model-b'] + }], + attachments: [{ id: 'attachment-b', kind: 'document' as const, name: 'new.txt' }] + }) + await act(async () => root.render(createElement(FloatingComposer, newer))) + await act(async () => pending.resolve({ ok: true, task: { id: 'scheduled-a' } })) + + expect(createScheduleTask).toHaveBeenCalledWith(expect.objectContaining({ + sourceThreadId: 'thread-a', + prompt: 'Follow up using the current context', + workspaceRoot: '/workspace/project', + providerId: 'provider-a', + accountId: 'account-a', + model: 'model-a', + attachmentIds: ['attachment-a'] + })) + expect(original.setInput).not.toHaveBeenCalled() + expect(original.onRemoveAttachment).not.toHaveBeenCalled() + expect(newer.setInput).not.toHaveBeenCalled() + expect(newer.onRemoveAttachment).not.toHaveBeenCalled() + expect(container.querySelector('[role="dialog"]')).toBeNull() + }) +}) diff --git a/src/renderer/src/components/chat/FloatingComposer.tsx b/src/renderer/src/components/chat/FloatingComposer.tsx index 29d336283..72d2bf3e9 100644 --- a/src/renderer/src/components/chat/FloatingComposer.tsx +++ b/src/renderer/src/components/chat/FloatingComposer.tsx @@ -13,6 +13,7 @@ import { } from 'react' import { BarChart3, + CalendarClock, FileText, Folder, ImagePlus, @@ -141,9 +142,10 @@ import { useGoalElapsedLabel } from './use-goal-elapsed' import type { FloatingComposerRenderContext } from './floating-composer-view-context' import { FloatingComposerStackView } from './FloatingComposerStackView' import { FloatingComposerSurfaceView } from './FloatingComposerSurfaceView' +import { ScheduledSendDialog } from './ScheduledSendDialog' +import { useScheduledSend } from './use-scheduled-send' import { FloatingComposerTaskProfile } from './FloatingComposerTaskProfile' export * from './floating-composer-public' - export function FloatingComposer({ variant = 'default', workspaceRootOverride, @@ -345,7 +347,6 @@ export function FloatingComposer({ activeClawChannel?.conversations.length || activeClawChannel?.remoteSession?.chatId?.trim() ) - const canEditComposer = !disabled && !hydratingActiveThread && (route === 'claw' ? clawHasInboundConversation : true) const canCompose = !disabled && !hydratingActiveThread && runtimeReady && ( route === 'claw' @@ -360,6 +361,7 @@ export function FloatingComposer({ (attachmentUploadEnabled && attachments.length > 0) || (fileReferenceEnabled && fileReferences.length > 0) ) + const canScheduleSend = canSend && fileReferences.length === 0 && route === 'chat' && Boolean(activeThreadId) && Boolean(effectiveWorkspaceRoot) && !pendingUserInputBlock const canPickAttachment = canCompose && attachmentUploadEnabled && !attachmentUploadBusy const canPickFileReference = canCompose && fileReferenceEnabled && Boolean(effectiveWorkspaceRoot) && Boolean(onOpenFileReferencePicker) const canPickDesignReference = canCompose && fileReferenceEnabled && Boolean(onOpenDesignReferencePicker) @@ -409,6 +411,7 @@ export function FloatingComposer({ const [goalInputMode, setGoalInputMode] = useState(false) const [promptOptimizationBusy, setPromptOptimizationBusy] = useState(false) const [promptOptimizationError, setPromptOptimizationError] = useState(null) + const scheduledSend = useScheduledSend({ canScheduleSend, activeThreadId, activeThreadTitle: activeThread?.title, input, workspaceRoot: effectiveWorkspaceRoot, composerProviderId, composerModel, composerModelGroups, composerReasoningEffort, orchestration, attachmentIds: attachments.map((attachment) => attachment.id), setInput, onRemoveAttachment }) const onDismissPromptOptimizationError = useCallback((): void => { setPromptOptimizationError(null) }, []) @@ -518,7 +521,6 @@ export function FloatingComposer({ && !composerMenuOpen && !goalPanelOpen && !pendingUserInputBlock - const parsedGoalCommand = parseGoalCommand(input) const goalPanelDraftObjective = getGoalPanelDraftObjective(input, goalPanelOpen) const canSetGoalPanelDraft = @@ -572,14 +574,11 @@ export function FloatingComposer({ goalPanelOpen, composerMenuOpen }) - useEffect(() => { if (slashQuery != null || goalPanelOpen) setComposerMenuOpen(false) }, [goalPanelOpen, slashQuery]) - useEffect(() => { if (!composerMenuOpen && !goalPanelOpen) return - const onPointerDown = (event: PointerEvent): void => { const target = event.target if (!(target instanceof Node)) return @@ -589,13 +588,11 @@ export function FloatingComposer({ setComposerMenuOpen(false) setGoalPanelOpen(false) } - const onKeyDown = (event: KeyboardEvent): void => { if (event.key !== 'Escape') return setComposerMenuOpen(false) setGoalPanelOpen(false) } - window.addEventListener('pointerdown', onPointerDown) window.addEventListener('keydown', onKeyDown) return () => { @@ -603,7 +600,6 @@ export function FloatingComposer({ window.removeEventListener('keydown', onKeyDown) } }, [composerMenuOpen, goalPanelOpen]) - const actionContext: FloatingComposerRenderContext = { activeThreadId, archiveThread, buildResearchPrompt, canAcceptComposerFileDrop, canAddFileReference, canEditComposer, canOpenComposerMenu, canOpenGoalPanel, @@ -621,10 +617,9 @@ export function FloatingComposer({ route, routeComposerFileDrop, runtimeReady, setActiveThreadGoal, setActiveThreadGoalStatus, setComposerMenuOpen, setGoalInputMode, setGoalPanelOpen, setInput, setMode, setPromptOptimizationBusy, setPromptOptimizationError, slashCommandMenu, slashCommands, - t, userInput + t, userInput, canScheduleSend, onOpenScheduledSend: scheduledSend.openScheduledSend } const composerActions = useFloatingComposerActions(actionContext) - const renderContext: FloatingComposerRenderContext = { ...actionContext, ...composerActions, @@ -633,7 +628,7 @@ export function FloatingComposer({ FloatingComposerActionMenu, Folder, GitBranchPicker, ImagePlus, ListTodo, Loader2, Mic, Monitor, Paperclip, PauseCircle, Pencil, PlayCircle, Plus, Puzzle, Send, Share2, Sparkles, - Square, Target, Trash2, TypeIcon, VoiceRecordingStrip, WorkspaceProjectPicker, X, activeThreadGoal, + Square, Target, Trash2, TypeIcon, VoiceRecordingStrip, WorkspaceProjectPicker, X, CalendarClock, activeThreadGoal, activeThreadId, activeThreadTodos, attachmentUploadBusy, attachmentUploadEnabled, attachmentUploadError, attachments, busy, canChangeModel, canCompose, canEditComposer, canOpenComposerMenu, canOpenGoalPanel, canOptimizePrompt, canPickAttachment, canPickDesignReference, canPickFileReference, canPickLocalFileReference, canSetGoalPanelDraft, canToggleGraphMode, canTogglePlanMode, canToggleWorktreeMode, clearActiveThreadGoal, compact, composerFastMode, @@ -650,9 +645,9 @@ export function FloatingComposer({ route, runningGraphTurn, runtimeReady, setActiveThreadGoalStatus, setGoalInputMode, setGoalPanelOpen, setInput, showComposerMenuButton, showCodeExecutionControls, showExecutionSettingsPicker, showGoalFloater, showGoalMenuOption, showGraphMenuOption, showGraphProgress, showPlanMenuOption, showProviderInModelLabel, showTodoProgress, showToolbarStartControls, showUsageHistoryFooter, showVoiceDictation, showWorkspaceControls, side, slashCommandMenu, slashQuery, stretchModelPicker, t, threadUsage, primaryActionKind, - taskSurface, taskSurfaceLocked, emptyTaskLayout, onTaskSurfaceChange, onNewRequirement, threadUsageState, timingThreadUsage, useWorktreePool, userInput, worktreeBranch + taskSurface, taskSurfaceLocked, emptyTaskLayout, onTaskSurfaceChange, onNewRequirement, threadUsageState, timingThreadUsage, useWorktreePool, userInput, worktreeBranch, + canScheduleSend, onOpenScheduledSend: scheduledSend.openScheduledSend } - return (
+ {scheduledSend.open ? ( + + ) : null}
) diff --git a/src/renderer/src/components/chat/FloatingComposerSurfaceView.tsx b/src/renderer/src/components/chat/FloatingComposerSurfaceView.tsx index 915633cd4..fc1091870 100644 --- a/src/renderer/src/components/chat/FloatingComposerSurfaceView.tsx +++ b/src/renderer/src/components/chat/FloatingComposerSurfaceView.tsx @@ -16,10 +16,10 @@ export function FloatingComposerSurfaceView({ FileText, FloatingComposerAgentPicker, FloatingComposerAttachments, FloatingComposerContextCapacity, FloatingComposerExecutionPicker, FloatingComposerModelPicker, FloatingComposerTaskProfile, - Folder, GitBranchPicker, ListTodo, Loader2, Mic, Plus, Send, Share2, Sparkles, + Folder, GitBranchPicker, ListTodo, Loader2, Mic, Plus, Send, Share2, Sparkles, CalendarClock, Square, Target, VoiceRecordingStrip, WorkspaceProjectPicker, X, activeThreadGoal, activeThreadId, attachmentUploadEnabled, attachmentUploadError, attachments, busy, - canChangeModel, canCompose, canEditComposer, canOpenComposerMenu, canOptimizePrompt, + canChangeModel, canCompose, canEditComposer, canOpenComposerMenu, canOptimizePrompt, canScheduleSend, onOpenScheduledSend, canToggleWorktreeMode, compact, composerFastMode, composerMenuButtonRef, composerMenuOpen, composerShellRef, composerModel, composerModelGroups, composerPickList, composerProviderId, composerReasoningEffort, contextChips, designTaskProfile, designProfileLocked, dictation, draft, effectiveWorkspaceRoot, @@ -436,6 +436,18 @@ export function FloatingComposerSurfaceView({ )} ) : null} + {!side ? ( + + ) : null} + +
+ + + +
+ {instant.ok ?

{formatInTimeZone(instant.iso, timeZone, locale)} · {relativeScheduleLabel(instant.iso, Date.now(), locale)}

:

{instantError(instant, t)}

} + {error ?

{error}

: null} +

{t('planScheduleBuildRunningNotice')}

+
+ + + ) +} diff --git a/src/renderer/src/components/chat/use-scheduled-send.ts b/src/renderer/src/components/chat/use-scheduled-send.ts new file mode 100644 index 000000000..e8de995bf --- /dev/null +++ b/src/renderer/src/components/chat/use-scheduled-send.ts @@ -0,0 +1,64 @@ +import { useCallback, useRef, useState } from 'react' +import { normalizeScheduleReasoningEffort } from '@shared/app-settings' +import { accountIdForComposerSelection, providerIdForComposerModel } from '../../store/chat-store-helpers' +import type { ScheduledSendDraft } from './ScheduledSendDialog' + +type Snapshot = { + threadId: string + threadTitle: string + prompt: string + workspaceRoot: string + providerId: string + accountId: string + model: string + reasoningEffort: ReturnType + orchestration: 'direct' | 'graph' + attachmentIds: string[] +} + +type Props = { + canScheduleSend: boolean + activeThreadId: string | null + activeThreadTitle?: string + input: string + workspaceRoot: string + composerProviderId?: string | null + composerModel: string + composerModelGroups: Parameters[0] + composerReasoningEffort: Parameters[0] + orchestration: 'direct' | 'graph' + attachmentIds: string[] + setInput: (value: string) => void + onRemoveAttachment?: (id: string) => void +} + +export function useScheduledSend(props: Props) { + const [open, setOpen] = useState(false) + const [snapshot, setSnapshot] = useState(null) + const [submitting, setSubmitting] = useState(false) + const [error, setError] = useState('') + const submittingRef = useRef(false) + const currentRef = useRef({ threadId: props.activeThreadId, prompt: props.input, attachmentIds: props.attachmentIds }) + currentRef.current = { threadId: props.activeThreadId, prompt: props.input, attachmentIds: props.attachmentIds } + const openScheduledSend = useCallback(() => { + if (!props.canScheduleSend || !props.activeThreadId) return + const providerId = props.composerProviderId?.trim() || providerIdForComposerModel(props.composerModelGroups, props.composerModel) + setSnapshot({ threadId: props.activeThreadId, threadTitle: props.activeThreadTitle?.trim() || 'Thread', prompt: props.input, workspaceRoot: props.workspaceRoot, providerId, accountId: accountIdForComposerSelection(props.composerModelGroups, providerId, props.composerModel), model: props.composerModel, reasoningEffort: normalizeScheduleReasoningEffort(props.composerReasoningEffort), orchestration: props.orchestration, attachmentIds: [...props.attachmentIds] }) + setError(''); setOpen(true) + }, [props]) + const submit = useCallback(async ({ atTime, timeZone }: ScheduledSendDraft) => { + const item = snapshot + if (!item || submittingRef.current) return + submittingRef.current = true; setSubmitting(true); setError('') + try { + const result = await window.kunGui.createScheduleTask({ title: `Scheduled send: ${item.threadTitle}`.slice(0, 200), prompt: item.prompt, workspaceRoot: item.workspaceRoot, sourceThreadId: item.threadId, providerId: item.providerId, ...(item.accountId ? { accountId: item.accountId } : {}), model: item.model, reasoningEffort: item.reasoningEffort, mode: 'agent', orchestration: item.orchestration, attachmentIds: item.attachmentIds, schedule: { kind: 'at', atTime, timeZone } }) + if (!result.ok) throw new Error(result.message) + const current = currentRef.current + const sameAttachments = current.attachmentIds.length === item.attachmentIds.length && current.attachmentIds.every((id, i) => id === item.attachmentIds[i]) + if (current.threadId === item.threadId && current.prompt === item.prompt && sameAttachments) { props.setInput(''); item.attachmentIds.forEach((id) => props.onRemoveAttachment?.(id)) } + setOpen(false); setSnapshot(null) + } catch (cause) { setError(cause instanceof Error ? cause.message : String(cause)) } + finally { submittingRef.current = false; setSubmitting(false) } + }, [snapshot, props]) + return { open, snapshot, submitting, error, openScheduledSend, submit, close: () => { if (!submitting) { setOpen(false); setSnapshot(null) } } } +} diff --git a/src/renderer/src/components/schedule/ScheduleTasksView.tsx b/src/renderer/src/components/schedule/ScheduleTasksView.tsx index c2a4b23b7..62a776071 100644 --- a/src/renderer/src/components/schedule/ScheduleTasksView.tsx +++ b/src/renderer/src/components/schedule/ScheduleTasksView.tsx @@ -51,14 +51,12 @@ import { SidebarTitlebarToggleButton } from '../sidebar/SidebarPrimitives' import { ScheduleDefaultsDialog } from './ScheduleDefaultsDialog' import { createScheduleRefreshCoordinator } from './schedule-refresh-coordinator' import { SessionDaemonsView } from './SessionDaemonsView' - type Props = { leftSidebarCollapsed: boolean onToggleLeftSidebar: () => void onOpenThread?: (threadId: string) => void onConnectWeixin?: () => void } - export { dateTimeLocalValueFromIso, filterScheduledTasks, @@ -131,7 +129,6 @@ export function ScheduleTasksView({ const [settingsDialogOpen, setSettingsDialogOpen] = useState(false) const [expandedResultTaskIds, setExpandedResultTaskIds] = useState>(() => new Set()) const refreshCoordinator = useRef(createScheduleRefreshCoordinator()).current - const load = useCallback(async (): Promise => { const ticket = refreshCoordinator.beginRefresh() if (ticket === null) return @@ -153,7 +150,6 @@ export function ScheduleTasksView({ if (refreshCoordinator.isCurrent(ticket)) setLoading(false) } }, [refreshCoordinator]) - useEffect(() => { void load() const id = window.setInterval(() => void load(), 5_000) @@ -162,7 +158,6 @@ export function ScheduleTasksView({ refreshCoordinator.invalidate() } }, [load, refreshCoordinator]) - const schedule = settings ? normalizeScheduleSettings(settings.schedule) : null const tasks = schedule?.tasks ?? EMPTY_SCHEDULE_TASKS const clawChannels = settings?.claw.channels ?? [] @@ -173,7 +168,6 @@ export function ScheduleTasksView({ const runningTaskIds = useMemo(() => new Set(status?.runningTaskIds ?? []), [status]) const queuedTaskIds = useMemo(() => new Set(status?.queuedTaskIds ?? []), [status]) const visibleTasks = useMemo(() => filterScheduledTasks(tasks, filter), [filter, tasks]) - const persistSchedule = async ( patch: Parameters[1] ): Promise => { @@ -198,13 +192,11 @@ export function ScheduleTasksView({ refreshCoordinator.endMutation() } } - const resolveDialogWorkspaceRoot = useCallback((workspaceRoot?: string): string => { const explicit = workspaceRoot?.trim() || '' if (explicit) return explicit return schedule?.defaultWorkspaceRoot.trim() || settings?.workspaceRoot.trim() || '' }, [schedule?.defaultWorkspaceRoot, settings?.workspaceRoot]) - const openCreateDialog = (): void => { const workspaceRoot = resolveDialogWorkspaceRoot() const selection = settings @@ -223,7 +215,6 @@ export function ScheduleTasksView({ }) }) setDialogError(null) } - const openEditDialog = (task: ScheduledTaskV1): void => { const selection = resolveScheduleModelSelection(modelProviders, task.providerId, task.model) const selectedProvider = modelProviders.find((provider) => provider.providerId === selection.providerId) ?? null @@ -242,7 +233,6 @@ export function ScheduleTasksView({ }) setDialogError(null) } - const pickDialogWorkspace = async (): Promise => { if (!dialog) return try { @@ -259,11 +249,9 @@ export function ScheduleTasksView({ setDialogError(formatWorkspacePickerError(error)) } } - const onDraftChangeInDialog = (patch: Partial): void => { setDialog((current) => current ? { ...current, draft: { ...current.draft, ...patch } } : current) } - const saveDialog = async (): Promise => { if (!dialog || !schedule || !settings) return const validation = validateScheduledTaskDraft(dialog.draft, t) @@ -300,21 +288,60 @@ export function ScheduleTasksView({ nextRunAt: '' } if (dialog.mode === 'create') { - await persistSchedule({ - enabled: true, - tasks: [...schedule.tasks, { ...task, createdAt: now }] + if (typeof window.kunGui?.createScheduleTask !== 'function') throw new Error('Schedule task creation is unavailable.') + const result = await window.kunGui.createScheduleTask({ + title: task.title, + prompt: task.prompt, + workspaceRoot: task.workspaceRoot, + ...(task.sourcePlanId ? { sourcePlanId: task.sourcePlanId } : {}), + ...(task.sourceThreadId ? { sourceThreadId: task.sourceThreadId } : {}), + providerId: task.providerId ?? '', + model: task.model, + reasoningEffort: task.reasoningEffort, + mode: task.mode, + orchestration: task.orchestration ?? 'direct', + clawChannelId: task.clawChannelId, + enabled: task.enabled, + priority: task.priority, + dependsOn: task.dependsOn, + useWorktree: task.useWorktree, + schedule: task.schedule }) + if (!result.ok) throw new Error(result.message) + await load() } else { - await persistSchedule({ - tasks: schedule.tasks.map((item) => item.id === dialog.taskId ? task : item) + if (typeof window.kunGui?.updateScheduleTask !== 'function') throw new Error('Schedule task updates are unavailable.') + const result = await window.kunGui.updateScheduleTask({ + taskId: dialog.taskId, + title: task.title, + prompt: task.prompt, + workspaceRoot: task.workspaceRoot, + enabled: task.enabled, + clawChannelId: task.clawChannelId, + providerId: task.providerId, + model: task.model, + reasoningEffort: task.reasoningEffort, + mode: task.mode, + orchestration: task.orchestration, + priority: task.priority, + dependsOn: task.dependsOn, + useWorktree: task.useWorktree, + schedule: task.schedule }) + if (!result.ok) throw new Error(result.message) + await load() } setDialog(null) setDialogError(null) } - const updateTask = async (taskId: string, patch: Partial): Promise => { if (!schedule) return + if (typeof window.kunGui?.updateScheduleTask === 'function') { + const result = await window.kunGui.updateScheduleTask({ taskId, ...patch }) + if (!result.ok) throw new Error(result.message) + await load() + return + } const now = nowIso() await persistSchedule({ tasks: schedule.tasks.map((task) => @@ -330,13 +357,14 @@ export function ScheduleTasksView({ ) }) } - const deleteTask = async (taskId: string): Promise => { if (!schedule) return if (!(await confirmDialog(t('scheduleDeleteConfirm')))) return - await persistSchedule({ tasks: schedule.tasks.filter((task) => task.id !== taskId) }) + if (typeof window.kunGui?.deleteScheduleTask !== 'function') throw new Error('Schedule task deletion is unavailable.') + const result = await window.kunGui.deleteScheduleTask(taskId) + if (!result.ok) throw new Error(result.message) + await load() } - const runTask = async (taskId: string): Promise => { if (typeof window.kunGui?.runScheduleTask !== 'function') return const result = await window.kunGui.runScheduleTask(taskId) @@ -346,11 +374,9 @@ export function ScheduleTasksView({ } await load() } - const toggleKeepAwake = async (value: boolean): Promise => { await persistSchedule({ keepAwake: value }) } - const toggleResultPreview = (taskId: string): void => { setExpandedResultTaskIds((current) => { const next = new Set(current) @@ -362,7 +388,6 @@ export function ScheduleTasksView({ return next }) } - return (
@@ -385,7 +410,6 @@ export function ScheduleTasksView({
-
-
@@ -481,7 +504,6 @@ export function ScheduleTasksView({
- {loading ? (
{t('loading')}
) : error ? ( @@ -629,7 +651,6 @@ export function ScheduleTasksView({ )}
- {dialog ? ( ) : null} - {settingsDialogOpen && schedule ? ( & { kind?: ScheduleKind } } export type ScheduleTaskDeleteResult = @@ -460,6 +488,8 @@ export type ScheduledTaskV1 = { sourcePlanId?: string /** Existing GUI thread reused by plan-scheduled builds. */ sourceThreadId?: string + /** Frozen delivery metadata for ordinary sends on an existing Thread. */ + scheduledSend?: ScheduledThreadSendV1 /** Optional Claw IM channel whose persona/defaults should drive this scheduled task. */ clawChannelId: string /** Selected model provider for this scheduled task. Empty means the current/default runtime provider. */ diff --git a/src/shared/app-settings-types-product.ts b/src/shared/app-settings-types-product.ts index b964f3b22..a264571d2 100644 --- a/src/shared/app-settings-types-product.ts +++ b/src/shared/app-settings-types-product.ts @@ -469,7 +469,7 @@ export type ClawRunResult = /** The task was accepted by the background queue but has not started. */ queued?: boolean } - | { ok: false; message: string } + | { ok: false; message: string; status?: number; code?: string } export type ScheduleRunResult = ClawRunResult From 17f55c866ff048f36bca46e62de0d3d58013dc53 Mon Sep 17 00:00:00 2001 From: luoye520ww <100058663+luoye520ww@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:09:21 +0800 Subject: [PATCH 2/9] fix(schedule): keep scheduled send snapshots immutable --- .../schedule-runtime.scheduled-send.test.ts | 31 +++++++++++++++++++ src/main/schedule-runtime.ts | 18 +++++++++++ 2 files changed, 49 insertions(+) diff --git a/src/main/schedule-runtime.scheduled-send.test.ts b/src/main/schedule-runtime.scheduled-send.test.ts index 5e473eb31..03612fadd 100644 --- a/src/main/schedule-runtime.scheduled-send.test.ts +++ b/src/main/schedule-runtime.scheduled-send.test.ts @@ -360,6 +360,37 @@ describe('ScheduleRuntime existing-thread scheduled send', () => { expect(prompts).toEqual(['edit active']) await runtime.stop() }) + + it('keeps the existing-thread send snapshot immutable when a generic editor changes payload fields', async () => { + const task = scheduledSendTask() + const runtime = new ScheduleRuntime({ + store: createStore(settingsWith(task)) as never, + runtimeRequest: vi.fn() as never, + logError: vi.fn() + }) + + await expect(runtime.updateTaskById(task.id, { prompt: 'mutated after scheduling' })).rejects.toThrow( + 'Scheduled send snapshot fields are immutable' + ) + await expect(runtime.updateTaskById(task.id, { providerId: 'other-provider' })).rejects.toThrow( + 'Scheduled send snapshot fields are immutable' + ) + await expect(runtime.updateTaskById(task.id, { model: 'other-model' })).rejects.toThrow( + 'Scheduled send snapshot fields are immutable' + ) + await expect(runtime.updateTaskById(task.id, { + scheduledSend: { ...task.scheduledSend!, attachmentIds: ['different-attachment'] } + })).rejects.toThrow('Scheduled send snapshot fields are immutable') + + const persisted = await runtime.listTasks() + expect(persisted[0]).toMatchObject({ + prompt: task.prompt, + providerId: task.providerId, + model: task.model, + scheduledSend: task.scheduledSend + }) + await runtime.stop() + }) it('removes a cancelled same-thread queued send without a ghost admission', async () => { const first = scheduledSendTask({ id: 'send-active', prompt: 'active' }) const queued = scheduledSendTask({ diff --git a/src/main/schedule-runtime.ts b/src/main/schedule-runtime.ts index faa08387c..def0b9c22 100644 --- a/src/main/schedule-runtime.ts +++ b/src/main/schedule-runtime.ts @@ -371,6 +371,24 @@ export class ScheduleRuntime { ...current.schedule, tasks: current.schedule.tasks.map((item) => { if (item.id !== taskId) return item + // A send bound to an existing thread is a durable snapshot. Keep + // its delivery target and request payload immutable after admission + // so editing a generic scheduled-task form cannot silently change + // what will be sent later. + if (item.scheduledSend?.kind === 'thread-send') { + const immutableChanged = + (Object.prototype.hasOwnProperty.call(patch, 'prompt') && patch.prompt !== item.prompt) || + (Object.prototype.hasOwnProperty.call(patch, 'workspaceRoot') && patch.workspaceRoot !== item.workspaceRoot) || + (Object.prototype.hasOwnProperty.call(patch, 'sourceThreadId') && patch.sourceThreadId !== item.sourceThreadId) || + (Object.prototype.hasOwnProperty.call(patch, 'providerId') && patch.providerId !== item.providerId) || + (Object.prototype.hasOwnProperty.call(patch, 'model') && patch.model !== item.model) || + (Object.prototype.hasOwnProperty.call(patch, 'reasoningEffort') && patch.reasoningEffort !== item.reasoningEffort) || + (Object.prototype.hasOwnProperty.call(patch, 'scheduledSend') && + JSON.stringify(patch.scheduledSend) !== JSON.stringify(item.scheduledSend)) + if (immutableChanged) { + throw new Error('Scheduled send snapshot fields are immutable; cancel it and create a new send.') + } + } nextTask = { ...item, ...patch, From 03012c212aaf434339fb1ecbd6c610a44a132cef Mon Sep 17 00:00:00 2001 From: luoye520ww <100058663+luoye520ww@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:14:54 +0800 Subject: [PATCH 3/9] test(schedule): satisfy file line guard --- .../schedule-runtime.scheduled-send.test.ts | 32 ++++--------------- 1 file changed, 7 insertions(+), 25 deletions(-) diff --git a/src/main/schedule-runtime.scheduled-send.test.ts b/src/main/schedule-runtime.scheduled-send.test.ts index 03612fadd..39f6b537e 100644 --- a/src/main/schedule-runtime.scheduled-send.test.ts +++ b/src/main/schedule-runtime.scheduled-send.test.ts @@ -363,32 +363,14 @@ describe('ScheduleRuntime existing-thread scheduled send', () => { it('keeps the existing-thread send snapshot immutable when a generic editor changes payload fields', async () => { const task = scheduledSendTask() - const runtime = new ScheduleRuntime({ - store: createStore(settingsWith(task)) as never, - runtimeRequest: vi.fn() as never, - logError: vi.fn() - }) - - await expect(runtime.updateTaskById(task.id, { prompt: 'mutated after scheduling' })).rejects.toThrow( - 'Scheduled send snapshot fields are immutable' - ) - await expect(runtime.updateTaskById(task.id, { providerId: 'other-provider' })).rejects.toThrow( - 'Scheduled send snapshot fields are immutable' - ) - await expect(runtime.updateTaskById(task.id, { model: 'other-model' })).rejects.toThrow( - 'Scheduled send snapshot fields are immutable' - ) - await expect(runtime.updateTaskById(task.id, { - scheduledSend: { ...task.scheduledSend!, attachmentIds: ['different-attachment'] } - })).rejects.toThrow('Scheduled send snapshot fields are immutable') - + const runtime = new ScheduleRuntime({ store: createStore(settingsWith(task)) as never, runtimeRequest: vi.fn() as never, logError: vi.fn() }) + const immutable = 'Scheduled send snapshot fields are immutable' + await expect(runtime.updateTaskById(task.id, { prompt: 'mutated after scheduling' })).rejects.toThrow(immutable) + await expect(runtime.updateTaskById(task.id, { providerId: 'other-provider' })).rejects.toThrow(immutable) + await expect(runtime.updateTaskById(task.id, { model: 'other-model' })).rejects.toThrow(immutable) + await expect(runtime.updateTaskById(task.id, { scheduledSend: { ...task.scheduledSend!, attachmentIds: ['different-attachment'] } })).rejects.toThrow(immutable) const persisted = await runtime.listTasks() - expect(persisted[0]).toMatchObject({ - prompt: task.prompt, - providerId: task.providerId, - model: task.model, - scheduledSend: task.scheduledSend - }) + expect(persisted[0]).toMatchObject({ prompt: task.prompt, providerId: task.providerId, model: task.model, scheduledSend: task.scheduledSend }) await runtime.stop() }) it('removes a cancelled same-thread queued send without a ghost admission', async () => { From 7b8eee49f072305390e1ccf32a0ab6dbd7fc64b1 Mon Sep 17 00:00:00 2001 From: luoye520ww <100058663+luoye520ww@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:17:53 +0800 Subject: [PATCH 4/9] fix(schedule): guard send admission and retry exhaustion --- src/main/schedule-runtime.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/main/schedule-runtime.ts b/src/main/schedule-runtime.ts index def0b9c22..de7147d3b 100644 --- a/src/main/schedule-runtime.ts +++ b/src/main/schedule-runtime.ts @@ -261,6 +261,9 @@ export class ScheduleRuntime { } async createTask(task: ScheduledTaskV1): Promise { + if (task.scheduledSend?.kind === 'thread-send' && task.schedule.kind !== 'at') { + throw new Error('Scheduled sends to an existing thread must use a one-time at schedule.') + } const saved = await this.deps.store.update((current) => ({ ...current, schedule: { @@ -306,6 +309,9 @@ export class ScheduleRuntime { const taskId = randomUUID() const sourcePlanId = input.sourcePlanId?.trim() || '' const sourceThreadId = input.sourceThreadId?.trim() || '' + if (sourceThreadId && !sourcePlanId && input.schedule.kind !== 'at') { + throw new Error('Scheduled sends to an existing thread must use a one-time at schedule.') + } const task: ScheduledTaskV1 = { id: taskId, title: input.title.trim() || 'New scheduled task', From 16a188fb2e5a83c4df71b3b7f09f72d81afef9e0 Mon Sep 17 00:00:00 2001 From: luoye520ww <100058663+luoye520ww@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:19:19 +0800 Subject: [PATCH 5/9] fix(schedule): close scheduled send edge cases --- src/main/schedule-runtime-queue.ts | 17 ++++++++++++++++- .../schedule-runtime.scheduled-send.test.ts | 10 ++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/src/main/schedule-runtime-queue.ts b/src/main/schedule-runtime-queue.ts index c032ba729..5a72cdca8 100644 --- a/src/main/schedule-runtime-queue.ts +++ b/src/main/schedule-runtime-queue.ts @@ -383,6 +383,14 @@ export class ScheduleExecutionQueue { scheduledSendAttempt > task.scheduledSend.maxAttempts ) { if (slotReserved) this.runningTaskIds.delete(task.id) + await this.updateTask(task.id, (current) => ({ + ...current, + enabled: current.schedule.kind === 'at' ? false : current.enabled, + nextRunAt: current.schedule.kind === 'at' ? '' : current.nextRunAt, + lastStatus: 'error', + lastMessage: 'Scheduled send retry limit reached.', + updatedAt: new Date().toISOString() + })) return { ok: false, message: 'Scheduled send retry limit reached.' } } await this.updateTask(task.id, (current) => ({ @@ -399,6 +407,11 @@ export class ScheduleExecutionQueue { const settings = await this.loadSettings() const persistedTask = settings.schedule.tasks.find((candidate) => candidate.id === task.id) if (!persistedTask || this.cancelledTaskIds.has(task.id)) { this.runningTaskIds.delete(task.id); return { ok: false, message: 'Scheduled task was removed before admission.' } } + if (task.scheduledSend?.kind === 'thread-send' && !persistedTask.enabled) { + this.runningTaskIds.delete(task.id) + await this.updateTask(task.id, (current) => ({ ...current, lastStatus: 'idle', lastMessage: 'Scheduled send was paused before admission.', updatedAt: new Date().toISOString() })) + return { ok: false, message: 'Scheduled send was paused before admission.' } + } if (task.scheduledSend?.kind === 'thread-send' && (!task.sourceThreadId?.trim() || !task.scheduledSend.clientRequestId.trim() || !task.providerId?.trim() || !task.model.trim())) { this.runningTaskIds.delete(task.id) await this.updateTask(task.id, (current) => ({ ...current, enabled: false, lastStatus: 'error', lastMessage: 'Scheduled send snapshot is invalid; no message was sent.', updatedAt: new Date().toISOString() })) @@ -522,7 +535,9 @@ export class ScheduleExecutionQueue { ...current, lastRunAt: finishedAt.toISOString(), ...(current.scheduledSend?.kind === 'thread-send' ? { scheduledSend: { ...current.scheduledSend, reconciliationPending: false } } : {}), - nextRunAt: computeScheduleNextRunAt(current, finishedAt), + ...(current.schedule.kind === 'at' + ? { enabled: false, nextRunAt: '' } + : { nextRunAt: computeScheduleNextRunAt(current, finishedAt) }), lastStatus: 'error', lastMessage: message, updatedAt: finishedAt.toISOString() diff --git a/src/main/schedule-runtime.scheduled-send.test.ts b/src/main/schedule-runtime.scheduled-send.test.ts index 39f6b537e..770f5a724 100644 --- a/src/main/schedule-runtime.scheduled-send.test.ts +++ b/src/main/schedule-runtime.scheduled-send.test.ts @@ -373,6 +373,16 @@ describe('ScheduleRuntime existing-thread scheduled send', () => { expect(persisted[0]).toMatchObject({ prompt: task.prompt, providerId: task.providerId, model: task.model, scheduledSend: task.scheduledSend }) await runtime.stop() }) + it('rejects recurring sends and persists retry exhaustion as terminal', async () => { + const recurring = scheduledSendTask({ schedule: { kind: 'interval', everyMinutes: 5, timeOfDay: '09:00', atTime: '' } }) + const runtime = new ScheduleRuntime({ store: createStore(settingsWith(recurring)) as never, runtimeRequest: vi.fn() as never, logError: vi.fn() }) + await expect(runtime.createTask(recurring)).rejects.toThrow('must use a one-time at schedule') + const exhausted = scheduledSendTask({ id: 'exhausted', scheduledSend: { ...recurring.scheduledSend!, attemptCount: 3 }, schedule: { kind: 'at', everyMinutes: 60, timeOfDay: '09:00', atTime: '2026-08-30T00:00:00.000Z' } }) + const terminal = new ScheduleRuntime({ store: createStore(settingsWith(exhausted)) as never, runtimeRequest: vi.fn() as never, logError: vi.fn() }) + await expect(terminal.runTask(exhausted.id)).resolves.toMatchObject({ ok: false }) + expect((await terminal.listTasks())[0]).toMatchObject({ enabled: false, nextRunAt: '', lastStatus: 'error' }) + await runtime.stop(); await terminal.stop() + }) it('removes a cancelled same-thread queued send without a ghost admission', async () => { const first = scheduledSendTask({ id: 'send-active', prompt: 'active' }) const queued = scheduledSendTask({ From efdfaa12125797c754bf591e7621950a49109dcc Mon Sep 17 00:00:00 2001 From: luoye520ww <100058663+luoye520ww@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:21:29 +0800 Subject: [PATCH 6/9] fix(schedule): validate frozen send execution fields --- src/main/schedule-runtime.ts | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/src/main/schedule-runtime.ts b/src/main/schedule-runtime.ts index de7147d3b..bf3f31d3d 100644 --- a/src/main/schedule-runtime.ts +++ b/src/main/schedule-runtime.ts @@ -294,6 +294,9 @@ export class ScheduleRuntime { enabled?: boolean schedule: Partial & { kind: ScheduledTaskV1['schedule']['kind'] } }): Promise { + if (input.schedule.kind === 'at' && !input.schedule.atTime?.trim()) { + throw new Error('An at schedule requires atTime.') + } const settings = await this.loadSettings() const clawChannel = this.queue.resolveClawChannel(settings, input.clawChannelId) const requestedProviderId = input.providerId?.trim() || settings.schedule.providerId?.trim() || '' @@ -367,6 +370,9 @@ export class ScheduleRuntime { taskId: string, patch: Omit, 'schedule'> & { schedule?: Partial } ): Promise { + if (patch.schedule?.kind === 'at' && !patch.schedule.atTime?.trim()) { + throw new Error('An at schedule requires atTime.') + } const now = new Date().toISOString() const shouldRecomputeNextRun = Object.prototype.hasOwnProperty.call(patch, 'enabled') || patch.schedule !== undefined @@ -389,6 +395,11 @@ export class ScheduleRuntime { (Object.prototype.hasOwnProperty.call(patch, 'providerId') && patch.providerId !== item.providerId) || (Object.prototype.hasOwnProperty.call(patch, 'model') && patch.model !== item.model) || (Object.prototype.hasOwnProperty.call(patch, 'reasoningEffort') && patch.reasoningEffort !== item.reasoningEffort) || + (Object.prototype.hasOwnProperty.call(patch, 'mode') && patch.mode !== item.mode) || + (Object.prototype.hasOwnProperty.call(patch, 'orchestration') && patch.orchestration !== item.orchestration) || + (Object.prototype.hasOwnProperty.call(patch, 'useWorktree') && patch.useWorktree !== item.useWorktree) || + (Object.prototype.hasOwnProperty.call(patch, 'dependsOn') && JSON.stringify(patch.dependsOn) !== JSON.stringify(item.dependsOn)) || + (Object.prototype.hasOwnProperty.call(patch, 'clawChannelId') && patch.clawChannelId !== item.clawChannelId) || (Object.prototype.hasOwnProperty.call(patch, 'scheduledSend') && JSON.stringify(patch.scheduledSend) !== JSON.stringify(item.scheduledSend)) if (immutableChanged) { From 5a61870bec599f0cf97987ce190d3d5bd27f26c1 Mon Sep 17 00:00:00 2001 From: luoye520ww <100058663+luoye520ww@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:24:17 +0800 Subject: [PATCH 7/9] fix(schedule): honor cancellation during turn settlement --- src/main/schedule-runtime-queue.ts | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/main/schedule-runtime-queue.ts b/src/main/schedule-runtime-queue.ts index 5a72cdca8..71fc17aab 100644 --- a/src/main/schedule-runtime-queue.ts +++ b/src/main/schedule-runtime-queue.ts @@ -565,8 +565,7 @@ export class ScheduleExecutionQueue { const finishedAt = new Date() await this.updateTask(taskId, (current) => ({ ...current, - ...(current.schedule.kind === 'at' ? { enabled: false } : {}), - nextRunAt: current.schedule.kind === 'at' ? '' : computeScheduleNextRunAt(current, finishedAt), + ...(current.schedule.kind === 'at' || !current.enabled ? { enabled: false, nextRunAt: '' } : { nextRunAt: computeScheduleNextRunAt(current, finishedAt) }), lastStatus: 'success', lastMessage: summarizeTaskResult(text), lastThreadId: threadId, @@ -578,8 +577,7 @@ export class ScheduleExecutionQueue { const finishedAt = new Date() await this.updateTask(taskId, (current) => ({ ...current, - ...(current.schedule.kind === 'at' ? { enabled: false } : {}), - nextRunAt: current.schedule.kind === 'at' ? '' : computeScheduleNextRunAt(current, finishedAt), + ...(current.schedule.kind === 'at' || !current.enabled ? { enabled: false, nextRunAt: '' } : { nextRunAt: computeScheduleNextRunAt(current, finishedAt) }), lastStatus: 'error', lastMessage: message, lastThreadId: threadId || current.lastThreadId, From 2b0d8f91d9fe0e25eb8f4601e67c3f2d322893be Mon Sep 17 00:00:00 2001 From: luoye520ww <100058663+luoye520ww@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:24:53 +0800 Subject: [PATCH 8/9] fix(schedule): recheck cancellation before admission --- src/main/schedule-runtime-queue.ts | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/main/schedule-runtime-queue.ts b/src/main/schedule-runtime-queue.ts index 71fc17aab..07c01508f 100644 --- a/src/main/schedule-runtime-queue.ts +++ b/src/main/schedule-runtime-queue.ts @@ -453,6 +453,13 @@ export class ScheduleExecutionQueue { model: task.model, reasoningEffort: task.reasoningEffort }) + const beforeAdmission = await this.loadSettings() + const latestTask = beforeAdmission.schedule.tasks.find((candidate) => candidate.id === task.id) + if (!latestTask || this.cancelledTaskIds.has(task.id) || (task.scheduledSend?.kind === 'thread-send' && !latestTask.enabled)) { + this.runningTaskIds.delete(task.id) + await this.releaseTaskWorktree(task.id) + return { ok: false, message: 'Scheduled send was cancelled before admission.' } + } const result = await this.runPrompt(settings, { prompt: task.prompt, preservePrompt: task.scheduledSend?.kind === 'thread-send', From 4a8ab97006b617226e252e24ca11b30842e62d90 Mon Sep 17 00:00:00 2001 From: luoye520ww <100058663+luoye520ww@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:33:58 +0800 Subject: [PATCH 9/9] fix(schedule): satisfy queue file line gate --- src/main/schedule-runtime-queue.ts | 33 ++++++------------------------ 1 file changed, 6 insertions(+), 27 deletions(-) diff --git a/src/main/schedule-runtime-queue.ts b/src/main/schedule-runtime-queue.ts index 07c01508f..d6c917a17 100644 --- a/src/main/schedule-runtime-queue.ts +++ b/src/main/schedule-runtime-queue.ts @@ -378,19 +378,9 @@ export class ScheduleExecutionQueue { const scheduledSendAttempt = task.scheduledSend?.kind === 'thread-send' ? task.scheduledSend.attemptCount + 1 : 0 - if ( - task.scheduledSend?.kind === 'thread-send' && - scheduledSendAttempt > task.scheduledSend.maxAttempts - ) { + if (task.scheduledSend?.kind === 'thread-send' && scheduledSendAttempt > task.scheduledSend.maxAttempts) { if (slotReserved) this.runningTaskIds.delete(task.id) - await this.updateTask(task.id, (current) => ({ - ...current, - enabled: current.schedule.kind === 'at' ? false : current.enabled, - nextRunAt: current.schedule.kind === 'at' ? '' : current.nextRunAt, - lastStatus: 'error', - lastMessage: 'Scheduled send retry limit reached.', - updatedAt: new Date().toISOString() - })) + await this.updateTask(task.id, (current) => ({ ...current, enabled: current.schedule.kind === 'at' ? false : current.enabled, nextRunAt: current.schedule.kind === 'at' ? '' : current.nextRunAt, lastStatus: 'error', lastMessage: 'Scheduled send retry limit reached.', updatedAt: new Date().toISOString() })) return { ok: false, message: 'Scheduled send retry limit reached.' } } await this.updateTask(task.id, (current) => ({ @@ -407,11 +397,7 @@ export class ScheduleExecutionQueue { const settings = await this.loadSettings() const persistedTask = settings.schedule.tasks.find((candidate) => candidate.id === task.id) if (!persistedTask || this.cancelledTaskIds.has(task.id)) { this.runningTaskIds.delete(task.id); return { ok: false, message: 'Scheduled task was removed before admission.' } } - if (task.scheduledSend?.kind === 'thread-send' && !persistedTask.enabled) { - this.runningTaskIds.delete(task.id) - await this.updateTask(task.id, (current) => ({ ...current, lastStatus: 'idle', lastMessage: 'Scheduled send was paused before admission.', updatedAt: new Date().toISOString() })) - return { ok: false, message: 'Scheduled send was paused before admission.' } - } + if (task.scheduledSend?.kind === 'thread-send' && !persistedTask.enabled) { this.runningTaskIds.delete(task.id); await this.updateTask(task.id, (current) => ({ ...current, lastStatus: 'idle', lastMessage: 'Scheduled send was paused before admission.', updatedAt: new Date().toISOString() })); return { ok: false, message: 'Scheduled send was paused before admission.' } } if (task.scheduledSend?.kind === 'thread-send' && (!task.sourceThreadId?.trim() || !task.scheduledSend.clientRequestId.trim() || !task.providerId?.trim() || !task.model.trim())) { this.runningTaskIds.delete(task.id) await this.updateTask(task.id, (current) => ({ ...current, enabled: false, lastStatus: 'error', lastMessage: 'Scheduled send snapshot is invalid; no message was sent.', updatedAt: new Date().toISOString() })) @@ -453,13 +439,8 @@ export class ScheduleExecutionQueue { model: task.model, reasoningEffort: task.reasoningEffort }) - const beforeAdmission = await this.loadSettings() - const latestTask = beforeAdmission.schedule.tasks.find((candidate) => candidate.id === task.id) - if (!latestTask || this.cancelledTaskIds.has(task.id) || (task.scheduledSend?.kind === 'thread-send' && !latestTask.enabled)) { - this.runningTaskIds.delete(task.id) - await this.releaseTaskWorktree(task.id) - return { ok: false, message: 'Scheduled send was cancelled before admission.' } - } + const latestTask = (await this.loadSettings()).schedule.tasks.find((candidate) => candidate.id === task.id) + if (!latestTask || this.cancelledTaskIds.has(task.id) || (task.scheduledSend?.kind === 'thread-send' && !latestTask.enabled)) { this.runningTaskIds.delete(task.id); await this.releaseTaskWorktree(task.id); return { ok: false, message: 'Scheduled send was cancelled before admission.' } } const result = await this.runPrompt(settings, { prompt: task.prompt, preservePrompt: task.scheduledSend?.kind === 'thread-send', @@ -542,9 +523,7 @@ export class ScheduleExecutionQueue { ...current, lastRunAt: finishedAt.toISOString(), ...(current.scheduledSend?.kind === 'thread-send' ? { scheduledSend: { ...current.scheduledSend, reconciliationPending: false } } : {}), - ...(current.schedule.kind === 'at' - ? { enabled: false, nextRunAt: '' } - : { nextRunAt: computeScheduleNextRunAt(current, finishedAt) }), + ...(current.schedule.kind === 'at' ? { enabled: false, nextRunAt: '' } : { nextRunAt: computeScheduleNextRunAt(current, finishedAt) }), lastStatus: 'error', lastMessage: message, updatedAt: finishedAt.toISOString()