From 45fa58992207812d240c9e430dc5891716283751 Mon Sep 17 00:00:00 2001 From: marcel Date: Fri, 6 Nov 2026 18:49:56 +0100 Subject: [PATCH 1/2] fix(server): retain scheduled dispatch refusal reasons --- .../ScheduledTaskService.test.ts | 122 +++++++++++++++++- .../scheduledTasks/ScheduledTaskService.ts | 13 +- .../ScheduledTaskService.webhook.test.ts | 4 + 3 files changed, 135 insertions(+), 4 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index 71454e55a6540..624abb9a321ee 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -4,6 +4,7 @@ import * as NodeUtil from "node:util"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { assert, it, vi } from "@effect/vitest"; import { + CommandId, ProjectId, ProviderInstanceId, ScheduledTaskError, @@ -16,6 +17,7 @@ import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; import * as SqlClient from "effect/sql/SqlClient"; +import { SqlError, classifySqliteError } from "effect/sql/SqlError"; import * as Fiber from "effect/Fiber"; import * as Stream from "effect/Stream"; @@ -26,6 +28,7 @@ import * as TestClock from "effect/testing/TestClock"; import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; +import { OrchestratorDispatchError } from "../orchestration-v2/Orchestrator.ts"; import * as SecretRequests from "../secrets/SecretRequests.ts"; import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; @@ -490,6 +493,8 @@ it.effect("a fresh-session task launches one thread, then starts each later run const launches = yield* Ref.make(0); const sends = yield* Ref.make>([]); const launched = ThreadId.make("thread:fresh-session-task"); + const refusal = + "The current session is still running background agents or commands, and a fresh session would end them. Wait for them to finish, or press Stop, then send the message again."; const layer = ScheduledTaskService.layer.pipe( Layer.provide( Layer.mergeAll( @@ -503,13 +508,25 @@ it.effect("a fresh-session task launches one thread, then starts each later run Layer.mock(ThreadManagementService.ThreadManagementService)({ getThreadShell: () => Effect.succeed({ deletedAt: null } as never), sendToThread: (input) => - Ref.update(sends, (all) => [ + Ref.updateAndGet(sends, (all) => [ ...all, { threadId: input.threadId, ...(input.freshContext === undefined ? {} : { freshContext: input.freshContext }), }, - ]).pipe(Effect.as({} as never)), + ]).pipe( + Effect.flatMap((all) => + all.length === 1 + ? Effect.fail( + new OrchestratorDispatchError({ + commandId: input.commandId, + commandType: "message.dispatch", + cause: refusal, + }), + ) + : Effect.succeed({} as never), + ), + ), }), Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ @@ -550,10 +567,28 @@ it.effect("a fresh-session task launches one thread, then starts each later run const legacy = yield* service.upsert({ ...input, threadId: null }); assert.equal(legacy.task.threadId, launched); assert.equal(legacy.task.threadMode, "fresh_session"); - yield* service.runNow({ id: input.id }); + const refused = yield* service.runNow({ id: input.id }); + assert.equal(refused.threadId, null); + assert.equal(refused.task.threadId, launched); + assert.equal(refused.task.lastRunStatus, "failed"); + assert.equal(refused.task.runCount, 2); + assert.equal(refused.task.nextRunAt, task.nextRunAt); + assert.equal(refused.task.lastRunError, refusal); + const sql = yield* SqlClient.SqlClient; + const [stored] = + yield* sql`SELECT last_run_error FROM scheduled_tasks WHERE task_id = ${input.id}`; + assert.equal(stored?.last_run_error, refusal); + const reread = yield* ScheduledTaskService.ScheduledTaskService.use((fresh) => + fresh.list(), + ).pipe(Effect.provide(Layer.fresh(layer))); + assert.deepEqual(reread.tasks[0], refused.task); assert.equal(yield* Ref.get(launches), 1); assert.deepEqual(yield* Ref.get(sends), [{ threadId: launched, freshContext: true }]); + const succeeded = yield* service.runNow({ id: input.id }); + assert.equal(succeeded.task.lastRunStatus, "succeeded"); + assert.equal(succeeded.task.lastRunError, null); + assert.equal((yield* service.list()).tasks[0]?.lastRunError, null); // Its thread stays in the old project, so moving the task leaves it for a new one. const moved = yield* service.upsert({ ...input, @@ -568,6 +603,87 @@ it.effect("a fresh-session task launches one thread, then starts each later run }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); +it.effect("dispatch diagnostics keep opaque failures private and stack-free", () => + Effect.gen(function* () { + const commandId = CommandId.make("command:opaque-failure"); + const wrapped = (cause?: unknown) => + new OrchestratorDispatchError({ commandId, commandType: "message.dispatch", cause }); + const cyclic: { cause?: unknown; message: string } = { message: "private-nested-secret" }; + cyclic.cause = cyclic; + const sqlError = new SqlError({ + reason: classifySqliteError(Object.assign(new Error("private-query-data"), { errcode: 13 })), + }); + const failures = [ + [ + Effect.die(new Error("plain defect", { cause: new Error("private-nested-secret") })), + "plain defect", + ], + [Effect.die("plain string"), "plain string"], + [Effect.interrupt, "All fibers interrupted without error"], + [ + Effect.fail( + new ScheduledTaskError({ + message: "safe task failure", + cause: new Error("private-nested-secret"), + }), + ), + "safe task failure", + ], + ...[ + undefined, + "", + " ", + wrapped().message, + cyclic, + sqlError, + new Error("private-nested-secret"), + ].map((cause) => [Effect.fail(wrapped(cause)), wrapped().message] as const), + ] as const; + const failure = yield* Ref.make>(Effect.die("unconfigured")); + const layer = ScheduledTaskService.layer.pipe( + Layer.provide( + Layer.mergeAll( + ManagedPolicy.layerTest(), + Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), + Layer.mock(ThreadManagementService.ThreadManagementService)({ + sendToThread: () => Ref.get(failure).pipe(Effect.flatten, Effect.orDie), + }), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: "/projects", + isScratchProject: () => Effect.succeed(false), + }), + NodeCrypto.layer, + Layer.succeed(Scheduler.Scheduler, { register: () => Effect.void }), + ), + ), + ); + yield* Effect.gen(function* () { + const service = yield* ScheduledTaskService.ScheduledTaskService; + const { task } = yield* service.upsert({ + id: ScheduledTaskId.make("task:opaque-failure"), + title: "Opaque failures", + prompt: "Check progress", + enabled: true, + schedule: { type: "interval", everyMs: 60_000 }, + projectId: ProjectId.make("project:test"), + threadId: ThreadId.make("thread:opaque-failure"), + threadMode: "fresh_session", + workspaceStrategy: { type: "root" }, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }); + for (const [effect, expected] of failures) { + yield* Ref.set(failure, effect); + const result = yield* service.runNow({ id: task.id }); + assert.equal(result.task.lastRunError, expected); + assert.equal((yield* service.list()).tasks[0]?.lastRunError, expected); + } + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); + it.effect("a first run whose thread was deleted before it bound removes the task", () => Effect.gen(function* () { const layer = ScheduledTaskService.layer.pipe( diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 9dd04b41c9275..bd668ff3a0e85 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -53,6 +53,8 @@ import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts"; import * as Metrics from "../observability/Metrics.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; +import { OrchestratorDispatchError } from "../orchestration-v2/Orchestrator.ts"; +import { userFacingDispatchErrorMessage } from "../orchestration-v2/UserFacingErrors.ts"; import * as SecretRequests from "../secrets/SecretRequests.ts"; import * as ManagedPolicy from "../policy/ManagedPolicy.ts"; import * as DueHint from "../scheduling/DueHint.ts"; @@ -78,6 +80,7 @@ import { constantTimeEquals, verifyWebhookSignature } from "./webhookVerificatio export const WEBHOOK_ROUTE_PREFIX = "/api/hooks"; const isScheduledTaskError = Schema.is(ScheduledTaskError); +const isOrchestratorDispatchError = Schema.is(OrchestratorDispatchError); const PROMPT_TOO_LONG = "The filled-in prompt is too long."; const WEBHOOK_DELIVERY_RETENTION = 50; const WEBHOOK_DEDUPE_DELIVERY_RETENTION = 20; @@ -356,6 +359,14 @@ function errorMessage(error: unknown): string { return failure instanceof Error ? failure.message : String(failure); } +function dispatchErrorMessage(cause: Cause.Cause): string { + const failure = Cause.squash(cause); + // Direct orchestration refusals are user-facing; nested defects can contain private data. + return isOrchestratorDispatchError(failure) && typeof failure.cause === "string" + ? (userFacingDispatchErrorMessage(failure.cause) ?? errorMessage(failure)) + : errorMessage(failure); +} + function webhookPath(taskId: string, token: string): string { return `${WEBHOOK_ROUTE_PREFIX}/${encodeURIComponent(taskId)}/${token}`; } @@ -1012,7 +1023,7 @@ export const layer = Layer.effect( if (webhook !== undefined && runSucceeded) yield* Ref.set(webhook.dispatched, true); const completedAt = yield* localNow; const lastRunStatus = runSucceeded ? ("succeeded" as const) : ("failed" as const); - const lastRunError = runSucceeded ? null : errorMessage(result.cause); + const lastRunError = runSucceeded ? null : dispatchErrorMessage(result.cause); const threadId = runSucceeded ? result.value : null; // One transaction, so an edit cannot land between reading the schedule and // writing the next run computed from it. The origin is read first: it reads diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index e06ed817ac4f4..c1b28e234f280 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -1535,6 +1535,10 @@ it.effect("keeps replay keys when dispatch succeeds but completion bookkeeping f ); yield* Queue.take(launches); assert.equal((yield* sql`SELECT 1 FROM scheduled_task_webhook_replay_keys`).length, 1); + assert.equal( + (yield* service.list()).tasks[0]?.lastRunError, + "Could not record schedule task run.", + ); yield* sql`DROP TRIGGER fail_completion`; const retry = yield* service.triggerWebhook(requestFor(task)); assert.equal(retry._tag === "accepted" && retry.outcome, "duplicate"); From 1ea9ec2f12f8c4084c4b41cf04ca188f821d9079 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 15 Nov 2026 03:49:13 +0000 Subject: [PATCH 2/2] test(server): cut scheduled dispatch refusal tests to one table Replace the extended fresh-session and opaque-failure tests with one case table: a direct string refusal is recorded, while blank and non-string causes keep the generic message. Drop the unrelated webhook bookkeeping assertion. Co-Authored-By: Claude Opus 5.5 --- .../ScheduledTaskService.test.ts | 105 +++++------------- .../ScheduledTaskService.webhook.test.ts | 4 - 2 files changed, 26 insertions(+), 83 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index 231f21a440781..1335e369484a5 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -20,7 +20,6 @@ import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; import * as Option from "effect/Option"; import * as SqlClient from "effect/sql/SqlClient"; -import { SqlError, classifySqliteError } from "effect/sql/SqlError"; import * as Fiber from "effect/Fiber"; import * as Stream from "effect/Stream"; @@ -588,8 +587,6 @@ it.effect("a fresh-session task launches one thread, then starts each later run const launches = yield* Ref.make(0); const sends = yield* Ref.make>([]); const launched = ThreadId.make("thread:fresh-session-task"); - const refusal = - "The current session is still running background agents or commands, and a fresh session would end them. Wait for them to finish, or press Stop, then send the message again."; const layer = ScheduledTaskService.layer.pipe( Layer.provide( Layer.mergeAll( @@ -603,25 +600,13 @@ it.effect("a fresh-session task launches one thread, then starts each later run Layer.mock(ThreadManagementService.ThreadManagementService)({ getThreadShell: () => Effect.succeed({ deletedAt: null } as never), sendToThread: (input) => - Ref.updateAndGet(sends, (all) => [ + Ref.update(sends, (all) => [ ...all, { threadId: input.threadId, ...(input.freshContext === undefined ? {} : { freshContext: input.freshContext }), }, - ]).pipe( - Effect.flatMap((all) => - all.length === 1 - ? Effect.fail( - new OrchestratorDispatchError({ - commandId: input.commandId, - commandType: "message.dispatch", - cause: refusal, - }), - ) - : Effect.succeed({} as never), - ), - ), + ]).pipe(Effect.as({} as never)), }), Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ @@ -662,28 +647,10 @@ it.effect("a fresh-session task launches one thread, then starts each later run const legacy = yield* service.upsert({ ...input, threadId: null }); assert.equal(legacy.task.threadId, launched); assert.equal(legacy.task.threadMode, "fresh_session"); - const refused = yield* service.runNow({ id: input.id }); - assert.equal(refused.threadId, null); - assert.equal(refused.task.threadId, launched); - assert.equal(refused.task.lastRunStatus, "failed"); - assert.equal(refused.task.runCount, 2); - assert.equal(refused.task.nextRunAt, task.nextRunAt); - assert.equal(refused.task.lastRunError, refusal); - const sql = yield* SqlClient.SqlClient; - const [stored] = - yield* sql`SELECT last_run_error FROM scheduled_tasks WHERE task_id = ${input.id}`; - assert.equal(stored?.last_run_error, refusal); - const reread = yield* ScheduledTaskService.ScheduledTaskService.use((fresh) => - fresh.list(), - ).pipe(Effect.provide(Layer.fresh(layer))); - assert.deepEqual(reread.tasks[0], refused.task); + yield* service.runNow({ id: input.id }); assert.equal(yield* Ref.get(launches), 1); assert.deepEqual(yield* Ref.get(sends), [{ threadId: launched, freshContext: true }]); - const succeeded = yield* service.runNow({ id: input.id }); - assert.equal(succeeded.task.lastRunStatus, "succeeded"); - assert.equal(succeeded.task.lastRunError, null); - assert.equal((yield* service.list()).tasks[0]?.lastRunError, null); // Its thread stays in the old project, so moving the task leaves it for a new one. const moved = yield* service.upsert({ ...input, @@ -698,50 +665,31 @@ it.effect("a fresh-session task launches one thread, then starts each later run }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); -it.effect("dispatch diagnostics keep opaque failures private and stack-free", () => +it.effect("a refused dispatch records its refusal, and opaque causes stay generic", () => Effect.gen(function* () { - const commandId = CommandId.make("command:opaque-failure"); - const wrapped = (cause?: unknown) => - new OrchestratorDispatchError({ commandId, commandType: "message.dispatch", cause }); - const cyclic: { cause?: unknown; message: string } = { message: "private-nested-secret" }; - cyclic.cause = cyclic; - const sqlError = new SqlError({ - reason: classifySqliteError(Object.assign(new Error("private-query-data"), { errcode: 13 })), - }); - const failures = [ - [ - Effect.die(new Error("plain defect", { cause: new Error("private-nested-secret") })), - "plain defect", - ], - [Effect.die("plain string"), "plain string"], - [Effect.interrupt, "All fibers interrupted without error"], - [ - Effect.fail( - new ScheduledTaskError({ - message: "safe task failure", - cause: new Error("private-nested-secret"), - }), - ), - "safe task failure", - ], - ...[ - undefined, - "", - " ", - wrapped().message, - cyclic, - sqlError, - new Error("private-nested-secret"), - ].map((cause) => [Effect.fail(wrapped(cause)), wrapped().message] as const), + const refusal = "Thread thread:refused was stopped."; + const refused = (cause: unknown) => + new OrchestratorDispatchError({ + commandId: CommandId.make("command:refused"), + commandType: "message.dispatch", + cause, + }); + const generic = refused(undefined).message; + const cases = [ + [refusal, refusal], + [" ", generic], + [new Error("private-nested-secret"), generic], ] as const; - const failure = yield* Ref.make>(Effect.die("unconfigured")); + const failure = yield* Ref.make(undefined); const layer = ScheduledTaskService.layer.pipe( Layer.provide( Layer.mergeAll( ManagedPolicy.layerTest(), Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), Layer.mock(ThreadManagementService.ThreadManagementService)({ - sendToThread: () => Ref.get(failure).pipe(Effect.flatten, Effect.orDie), + getThreadShell: () => Effect.succeed(threadShell(ProjectId.make("project:test"))), + sendToThread: () => + Ref.get(failure).pipe(Effect.flatMap((cause) => Effect.fail(refused(cause)))), }), Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ @@ -756,24 +704,23 @@ it.effect("dispatch diagnostics keep opaque failures private and stack-free", () yield* Effect.gen(function* () { const service = yield* ScheduledTaskService.ScheduledTaskService; const { task } = yield* service.upsert({ - id: ScheduledTaskId.make("task:opaque-failure"), - title: "Opaque failures", + id: ScheduledTaskId.make("task:refused"), + title: "Refused dispatch", prompt: "Check progress", enabled: true, schedule: { type: "interval", everyMs: 60_000 }, projectId: ProjectId.make("project:test"), - threadId: ThreadId.make("thread:opaque-failure"), - threadMode: "fresh_session", + threadId: ThreadId.make("thread:refused"), workspaceStrategy: { type: "root" }, modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, runtimeMode: "full-access", interactionMode: "default", }); - for (const [effect, expected] of failures) { - yield* Ref.set(failure, effect); + for (const [cause, expected] of cases) { + yield* Ref.set(failure, cause); const result = yield* service.runNow({ id: task.id }); + assert.equal(result.task.lastRunStatus, "failed"); assert.equal(result.task.lastRunError, expected); - assert.equal((yield* service.list()).tasks[0]?.lastRunError, expected); } }).pipe(Effect.provide(layer)); }).pipe(Effect.provide(SqlitePersistence.layerMemory)), diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index c1b28e234f280..e06ed817ac4f4 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -1535,10 +1535,6 @@ it.effect("keeps replay keys when dispatch succeeds but completion bookkeeping f ); yield* Queue.take(launches); assert.equal((yield* sql`SELECT 1 FROM scheduled_task_webhook_replay_keys`).length, 1); - assert.equal( - (yield* service.list()).tasks[0]?.lastRunError, - "Could not record schedule task run.", - ); yield* sql`DROP TRIGGER fail_completion`; const retry = yield* service.triggerWebhook(requestFor(task)); assert.equal(retry._tag === "accepted" && retry.outcome, "duplicate");