From ef502c3c95b5e1cf702bbb2d144d6dc843c1b64a Mon Sep 17 00:00:00 2001 From: spoukyii <61633921+spoukyii@users.noreply.github.com> Date: Sat, 24 Oct 2026 17:06:33 -0700 Subject: [PATCH 1/3] test(server): a cancelled delegated child's usage-limit recovery Drives task_cancel on a real delegated child that fails on a usage limit, then runs the recovery sweep on both sides of the reset. Cancelling after the failure and never cancelling already behave; cancelling while the limit result is in flight still resumes the child, and fails here. --- .../TaskCancelUsageLimit.integration.test.ts | 351 ++++++++++++++++++ 1 file changed, 351 insertions(+) create mode 100644 apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts diff --git a/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts b/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts new file mode 100644 index 0000000000000..f55985ae32584 --- /dev/null +++ b/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts @@ -0,0 +1,351 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { + CommandId, + EnvironmentId, + MessageId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ProviderThreadId, + ProviderTurnId, + ThreadId, + type OrchestrationV2DomainEvent, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; + +import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts"; +import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts"; +import * as EffectWorker from "../orchestration-v2/EffectWorker.ts"; +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; +import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; +import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; +import * as UsageLimitRecoveryWorker from "../orchestration-v2/UsageLimitRecoveryWorker.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; +import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts"; +import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; +import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; +import * as Scheduler from "../scheduling/Scheduler.ts"; +import * as SecretRequests from "../secrets/SecretRequests.ts"; +import type { McpInvocationScope } from "./McpInvocationContext.ts"; +import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; + +const driver = ProviderDriverKind.make("claudeAgent"); +const instanceId = ProviderInstanceId.make("claudeAgent"); +const modelSelection = { instanceId, model: "claude-test" }; + +// A delegated child that fails on a usage limit is armed to continue at the reset. +// Once its parent cancels the task, nothing is waiting for that continuation. +it.effect.each([ + { cancel: "after its limit", cancelStatus: "failed", resumed: false }, + { cancel: "as its limit lands", cancelStatus: "cancel_requested", resumed: false }, + { cancel: "never", cancelStatus: null, resumed: true }, +] as const)( + "a delegated child whose task is cancelled $cancel resumes at the reset: $resumed", + ({ cancel, cancelStatus, resumed }) => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("task-cancel-usage-limit"); + // Set while the provider's usage-limit result is already on its way. + let limitInFlight: Effect.Effect = Effect.void; + const turns: Array<{ + readonly turn: ProviderAdapter.ProviderAdapterV2TurnInput; + readonly events: Queue.Queue; + }> = []; + const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = { + instanceId, + driver, + getCapabilities: () => Effect.succeed(ClaudeProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + openSession: (input) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const events = yield* Queue.unbounded(); + return { + instanceId, + driver, + providerSessionId: input.providerSessionId, + providerSession: { + id: input.providerSessionId, + driver, + providerInstanceId: instanceId, + status: "ready", + cwd, + model: modelSelection.model, + capabilities: ClaudeProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.fromQueue(events), + ensureThread: ({ threadId }) => + Effect.succeed({ + id: ProviderThreadId.make(`provider-thread:${threadId}`), + driver, + providerInstanceId: instanceId, + providerSessionId: input.providerSessionId, + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: { driver, nativeId: `native:${threadId}`, strength: "strong" }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }), + resumeThread: ({ providerThread }) => Effect.succeed(providerThread), + startTurn: (turn) => + Effect.gen(function* () { + turns.push({ turn, events }); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), + providerThreadId: turn.providerThread.id, + nodeId: turn.rootNodeId, + runAttemptId: turn.attemptId, + nativeTurnRef: null, + ordinal: turn.providerTurnOrdinal, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + }), + steerTurn: () => Effect.die("unused"), + interruptTurn: () => Effect.suspend(() => limitInFlight), + respondToRuntimeRequest: () => Effect.die("unused"), + readThreadSnapshot: () => Effect.die("unused"), + rollbackThread: () => Effect.die("unused"), + forkThread: () => Effect.die("unused"), + }; + }), + }; + const layerAdapters = ProviderAdapterRegistry.layerSingle(adapter); + const layerOrchestrator = ProviderReplayHarness.layerWithRegistry( + { name: "task-cancel-usage-limit" }, + layerAdapters, + { runEffectWorker: false }, + ); + const layerOrchestration = Layer.merge( + layerOrchestrator, + ThreadManagementService.layer.pipe( + Layer.provide(layerOrchestrator), + Layer.provide(ServerSettings.layerTest()), + Layer.provide( + CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)), + ), + ), + ); + const sweeps = new Map>(); + const layerRecovery = UsageLimitRecoveryWorker.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed(Scheduler.Scheduler, { + register: (name: string, work: Effect.Effect) => + Effect.gen(function* () { + const context = yield* Effect.context(); + sweeps.set(name, work.pipe(Effect.provideContext(context), Effect.orDie)); + }), + }), + ProjectionStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)), + ServerSettings.layerTest({ autoResumeLimitedThreads: true }), + ), + ), + ); + const layerTest = Layer.merge( + OrchestratorMcpService.layer.pipe( + Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})), + Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), + ), + layerRecovery, + ).pipe( + Layer.provideMerge(layerOrchestration), + Layer.provide( + Layer.mergeAll( + layerAdapters, + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + ServerSettings.layerTest(), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + NodeServices.layer, + ), + ), + ); + + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const sweep = Effect.andThen( + Effect.suspend(() => sweeps.get("usage-limit-recovery")!), + worker.drain(), + ); + const parentThreadId = ThreadId.make("thread:task-cancel-limit-parent"); + const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const turnRunning = () => + watch( + (event) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-parent"), + threadId: parentThreadId, + projectId: ProjectId.make("project:task-cancel-limit"), + title: "Parent", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const parentRunning = yield* turnRunning(); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("start-parent"), + threadId: parentThreadId, + messageId: MessageId.make("message:start-parent"), + text: "Delegate a review", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(parentRunning); + + const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!; + const childRunning = yield* turnRunning(); + yield* orchestrator.dispatch({ + type: "delegated_task.request", + commandId: CommandId.make("delegate"), + parentThreadId, + parentRunId: parentRun.id, + parentNodeId: parentRun.rootNodeId!, + task: "Review the change", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + completionWake: "always", + createdBy: "agent", + creationSource: "mcp", + }); + yield* worker.drain(); + yield* Fiber.join(childRunning); + const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!; + const childThreadId = task.childThreadId!; + const child = turns.find((entry) => entry.turn.threadId === childThreadId)!; + const childTurn = (yield* orchestrator.getThreadProjection(childThreadId)) + .providerTurns[0]!; + + const scope: McpInvocationScope = { + environmentId: EnvironmentId.make("environment:task-cancel-limit"), + requestNamespace: "provider-session:task-cancel-limit", + thread: { + threadId: parentThreadId, + providerSessionId: "provider-session:task-cancel-limit", + providerInstanceId: instanceId, + }, + client: undefined, + capabilities: new Set(["orchestration"]), + issuedAt: 1, + }; + const cancelTask = Effect.gen(function* () { + const result = yield* service.cancelTask(scope, { + taskId: task.id, + clientRequestId: "cancel-task", + }); + yield* worker.drain(); + return result.status; + }); + + const now = yield* DateTime.now; + const childFailed = yield* watch( + (event) => + event.type === "run.updated" && + event.payload.id === child.turn.runId && + event.payload.status === "failed", + ); + const hitLimit = Effect.forEach( + [ + { + type: "provider_turn.updated", + driver, + providerTurn: { ...childTurn, status: "failed", completedAt: now }, + }, + { + type: "turn.terminal", + driver, + providerThreadId: childTurn.providerThreadId, + providerTurnId: childTurn.id, + runOrdinal: child.turn.runOrdinal, + failureItemOrdinal: 1, + status: "failed", + failure: { + class: "usage_limit", + message: "Claude usage limit reached.", + code: null, + retryable: null, + resetAt: DateTime.formatIso(DateTime.add(now, { minutes: 1 })), + }, + threadDisposition: "reusable", + }, + ] satisfies ReadonlyArray, + (event) => Queue.offer(child.events, event), + { discard: true }, + ); + let cancelled: string | null = null; + if (cancel === "as its limit lands") { + limitInFlight = hitLimit; + cancelled = yield* cancelTask; + } else { + yield* hitLimit; + } + yield* Fiber.join(childFailed); + yield* worker.drain(); + + // The sweep runs every few seconds, so it arms the child before a parent reacts. + yield* sweep; + if (cancel === "after its limit") cancelled = yield* cancelTask; + assert.equal(cancelled, cancelStatus); + + yield* TestClock.adjust("1 minute"); + yield* sweep; + const childAfter = yield* orchestrator.getThreadProjection(childThreadId); + assert.deepEqual( + childAfter.messages.flatMap((message) => + message.role === "user" ? [message.text] : [], + ), + resumed ? ["Review the change", "Continue where you left off."] : ["Review the change"], + ); + }).pipe(Effect.provide(layerTest)); + }), + ), +); From f81e9424f59dd1975f2fba0ab57d9e5fd6f89a55 Mon Sep 17 00:00:00 2001 From: spoukyii <61633921+spoukyii@users.noreply.github.com> Date: Sat, 24 Oct 2026 17:09:07 -0700 Subject: [PATCH 2/3] fix(server): a task cancelled as its usage limit lands no longer resumes at the reset task_cancel tells agents its child "stops like a user Stop: the running turn is interrupted, other queued turns are held, pull request watches end, and the tasks it delegated stop too. The task's own queued turn is cancelled, including when held in a stopped queue. This includes later child-thread runs, even after the task is terminal." The usage-limit resume is such a later run, and a user Stop keeps it from starting by disarming the failure it finds (#21114). A limit result already in flight when the cancel's Stop interrupts the run lands after it, so Stop had nothing to disarm, the recovery sweep armed the failure, and the child started its brief again at the reset, beside the retry its parent had already run elsewhere. A top-level Stop raced the same way. The sweep's automatic arm now records the recovery disarmed when Stop reached the run, the record Stop writes for a failure already there. Only the automatic arm changes: a thread nobody stopped resumes as before, and resuming a stopped thread by choice still works. The decision sits in the arm, not in task_cancel or the recovery query. task_cancel cannot disarm a run that has not failed yet, and has no reset time to key a record by. A disposed completion delivery is no cancel marker: a parent's Stop, an archive or a closed delivery cohort disposes it too, and a top-level Stop has none. A cancel after the child's run failed already keeps the promise: #22653 stops the child on every cancel and #21114 disarms the failure. Bug: a delegated child whose task was cancelled never auto-resumes after a usage limit --- apps/server/src/orchestration-v2/Orchestrator.ts | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 351e804ac631a..98acabec91a85 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -12261,7 +12261,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio thread.limitRecovery.resetAt === input.limitRecovery.resetAt ) return; - yield* dispatchWithReceiptEffect({ type: "thread.metadata.update", ...input }); + const command = { type: "thread.metadata.update", ...input } as const; + // A Stop that reached the run before its limit failure landed had nothing to disarm. + const stopped = yield* stopReachedRun(command, input.threadId, input.limitRecovery.runId); + yield* dispatchWithReceiptEffect( + stopped + ? { + ...command, + limitRecovery: { ...input.limitRecovery, autoResume: false, snooze: false }, + } + : command, + ); }), ), searchThreadStream: (input) => From c6c0058ddc21353b53e153fc3054541cf638ccfb Mon Sep 17 00:00:00 2001 From: maria-rcks <254055478+maria-rcks@users.noreply.github.com> Date: Fri, 30 Oct 2026 20:22:48 +0000 Subject: [PATCH 3/3] test(server): supply remote delegation in cancellation fixture --- apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts b/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts index f55985ae32584..e4176219244cf 100644 --- a/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts +++ b/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts @@ -40,6 +40,7 @@ import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; import * as Scheduler from "../scheduling/Scheduler.ts"; import * as SecretRequests from "../secrets/SecretRequests.ts"; +import * as RemoteDelegation from "../peer/RemoteDelegation.ts"; import type { McpInvocationScope } from "./McpInvocationContext.ts"; import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; @@ -184,6 +185,7 @@ it.effect.each([ Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), Layer.mock(ProjectService.ProjectService)({}), Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), ServerSettings.layerTest(), Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), NodeServices.layer,