Open History to see the last commit
1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { assert, it } from "@effect/vitest";
4import {
5 CommandId,
6 EnvironmentId,
7 MessageId,
8 ProjectId,
9 ProviderDriverKind,
10 ProviderInstanceId,
11 ProviderThreadId,
12 ProviderTurnId,
13 ThreadId,
14 type OrchestrationV2DomainEvent,
15} from "@t3tools/contracts";
16import * as DateTime from "effect/DateTime";
17import * as Effect from "effect/Effect";
18import * as Fiber from "effect/Fiber";
19import * as Layer from "effect/Layer";
20import * as Queue from "effect/Queue";
21import * as Stream from "effect/Stream";
22import * as TestClock from "effect/testing/TestClock";
23
24import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts";
25import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts";
26import * as EffectWorker from "../orchestration-v2/EffectWorker.ts";
27import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
28import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
29import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
30import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
31import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
32import * as UsageLimitRecoveryWorker from "../orchestration-v2/UsageLimitRecoveryWorker.ts";
33import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
34import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
35import * as ServerSettings from "../serverSettings.ts";
36import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts";
37import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
38import * as SqlitePersistence from "../persistence/Sqlite.ts";
39import * as ProjectService from "../project/ProjectService.ts";
40import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
41import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
42import * as Scheduler from "../scheduling/Scheduler.ts";
43import * as SecretRequests from "../secrets/SecretRequests.ts";
44import * as RemoteDelegation from "../peer/RemoteDelegation.ts";
45import type { McpInvocationScope } from "./McpInvocationContext.ts";
46import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";
47
48const driver = ProviderDriverKind.make("claudeAgent");
49const instanceId = ProviderInstanceId.make("claudeAgent");
50const modelSelection = { instanceId, model: "claude-test" };
51
52// A delegated child that fails on a usage limit is armed to continue at the reset.
53// Once its parent cancels the task, nothing is waiting for that continuation.
54it.effect.each([
55 { cancel: "after its limit", cancelStatus: "failed", resumed: false },
56 { cancel: "as its limit lands", cancelStatus: "cancel_requested", resumed: false },
57 { cancel: "never", cancelStatus: null, resumed: true },
58] as const)(
59 "a delegated child whose task is cancelled $cancel resumes at the reset: $resumed",
60 ({ cancel, cancelStatus, resumed }) =>
61 Effect.gen(function* () {
62 const cwd = yield* checkpointWorkspace("task-cancel-usage-limit");
63 // Set while the provider's usage-limit result is already on its way.
64 let limitInFlight: Effect.Effect<void> = Effect.void;
65 const turns: Array<{
66 readonly turn: ProviderAdapter.ProviderAdapterV2TurnInput;
67 readonly events: Queue.Queue<ProviderAdapter.ProviderAdapterV2Event>;
68 }> = [];
69 const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = {
70 instanceId,
71 driver,
72 getCapabilities: () => Effect.succeed(ClaudeProviderCapabilitiesV2),
73 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
74 openSession: (input) =>
75 Effect.gen(function* () {
76 const now = yield* DateTime.now;
77 const events = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
78 return {
79 instanceId,
80 driver,
81 providerSessionId: input.providerSessionId,
82 providerSession: {
83 id: input.providerSessionId,
84 driver,
85 providerInstanceId: instanceId,
86 status: "ready",
87 cwd,
88 model: modelSelection.model,
89 capabilities: ClaudeProviderCapabilitiesV2,
90 createdAt: now,
91 updatedAt: now,
92 lastError: null,
93 },
94 events: Stream.fromQueue(events),
95 ensureThread: ({ threadId }) =>
96 Effect.succeed({
97 id: ProviderThreadId.make(`provider-thread:${threadId}`),
98 driver,
99 providerInstanceId: instanceId,
100 providerSessionId: input.providerSessionId,
101 appThreadId: threadId,
102 ownerNodeId: null,
103 nativeThreadRef: { driver, nativeId: `native:${threadId}`, strength: "strong" },
104 nativeConversationHeadRef: null,
105 status: "idle",
106 firstRunOrdinal: null,
107 lastRunOrdinal: null,
108 handoffIds: [],
109 forkedFrom: null,
110 createdAt: now,
111 updatedAt: now,
112 }),
113 resumeThread: ({ providerThread }) => Effect.succeed(providerThread),
114 startTurn: (turn) =>
115 Effect.gen(function* () {
116 turns.push({ turn, events });
117 yield* Queue.offer(events, {
118 type: "provider_turn.updated",
119 driver,
120 providerTurn: {
121 id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`),
122 providerThreadId: turn.providerThread.id,
123 nodeId: turn.rootNodeId,
124 runAttemptId: turn.attemptId,
125 nativeTurnRef: null,
126 ordinal: turn.providerTurnOrdinal,
127 status: "running",
128 startedAt: now,
129 completedAt: null,
130 },
131 });
132 }),
133 steerTurn: () => Effect.die("unused"),
134 interruptTurn: () => Effect.suspend(() => limitInFlight),
135 respondToRuntimeRequest: () => Effect.die("unused"),
136 readThreadSnapshot: () => Effect.die("unused"),
137 rollbackThread: () => Effect.die("unused"),
138 forkThread: () => Effect.die("unused"),
139 };
140 }),
141 };
142 const layerAdapters = ProviderAdapterRegistry.layerSingle(adapter);
143 const layerOrchestrator = ProviderReplayHarness.layerWithRegistry(
144 { name: "task-cancel-usage-limit" },
145 layerAdapters,
146 { runEffectWorker: false },
147 );
148 const layerOrchestration = Layer.merge(
149 layerOrchestrator,
150 ThreadManagementService.layer.pipe(
151 Layer.provide(layerOrchestrator),
152 Layer.provide(ServerSettings.layerTest()),
153 Layer.provide(
154 CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
155 ),
156 ),
157 );
158 const sweeps = new Map<string, Effect.Effect<void>>();
159 const layerRecovery = UsageLimitRecoveryWorker.layer.pipe(
160 Layer.provide(
161 Layer.mergeAll(
162 Layer.succeed(Scheduler.Scheduler, {
163 register: <E, R>(name: string, work: Effect.Effect<void, E, R>) =>
164 Effect.gen(function* () {
165 const context = yield* Effect.context<R>();
166 sweeps.set(name, work.pipe(Effect.provideContext(context), Effect.orDie));
167 }),
168 }),
169 ProjectionStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
170 ServerSettings.layerTest({ autoResumeLimitedThreads: true }),
171 ),
172 ),
173 );
174 const layerTest = Layer.merge(
175 OrchestratorMcpService.layer.pipe(
176 Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
177 Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
178 ),
179 layerRecovery,
180 ).pipe(
181 Layer.provideMerge(layerOrchestration),
182 Layer.provide(
183 Layer.mergeAll(
184 layerAdapters,
185 Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }),
186 Layer.mock(ProjectService.ProjectService)({}),
187 Layer.mock(SecretRequests.SecretRequests)({}),
188 Layer.mock(RemoteDelegation.RemoteDelegation)({}),
189 ServerSettings.layerTest(),
190 Layer.mock(ScheduledTaskService.ScheduledTaskService)({}),
191 NodeServices.layer,
192 NodeCrypto.layer,
193 ),
194 ),
195 );
196
197 yield* Effect.gen(function* () {
198 const orchestrator = yield* Orchestrator.OrchestratorV2;
199 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
200 const service = yield* OrchestratorMcpService.OrchestratorMcpService;
201 const sweep = Effect.andThen(
202 Effect.suspend(() => sweeps.get("usage-limit-recovery")!),
203 worker.drain(),
204 );
205 const parentThreadId = ThreadId.make("thread:task-cancel-limit-parent");
206 const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) =>
207 orchestrator.streamDomainEvents.pipe(
208 Stream.filter(predicate),
209 Stream.take(1),
210 Stream.runDrain,
211 Effect.forkScoped,
212 );
213 const turnRunning = () =>
214 watch(
215 (event) => event.type === "provider-turn.updated" && event.payload.status === "running",
216 );
217 yield* orchestrator.dispatch({
218 type: "thread.create",
219 commandId: CommandId.make("create-parent"),
220 threadId: parentThreadId,
221 projectId: ProjectId.make("project:task-cancel-limit"),
222 title: "Parent",
223 modelSelection,
224 runtimeMode: "full-access",
225 interactionMode: "default",
226 branch: null,
227 worktreePath: cwd,
228 createdBy: "user",
229 creationSource: "web",
230 });
231 const parentRunning = yield* turnRunning();
232 yield* orchestrator.dispatch({
233 type: "message.dispatch",
234 commandId: CommandId.make("start-parent"),
235 threadId: parentThreadId,
236 messageId: MessageId.make("message:start-parent"),
237 text: "Delegate a review",
238 attachments: [],
239 dispatchMode: { type: "start_immediately" },
240 createdBy: "user",
241 creationSource: "web",
242 });
243 yield* worker.drain();
244 yield* Fiber.join(parentRunning);
245
246 const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!;
247 const childRunning = yield* turnRunning();
248 yield* orchestrator.dispatch({
249 type: "delegated_task.request",
250 commandId: CommandId.make("delegate"),
251 parentThreadId,
252 parentRunId: parentRun.id,
253 parentNodeId: parentRun.rootNodeId!,
254 task: "Review the change",
255 modelSelection,
256 runtimeMode: "full-access",
257 interactionMode: "default",
258 completionWake: "always",
259 createdBy: "agent",
260 creationSource: "mcp",
261 });
262 yield* worker.drain();
263 yield* Fiber.join(childRunning);
264 const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!;
265 const childThreadId = task.childThreadId!;
266 const child = turns.find((entry) => entry.turn.threadId === childThreadId)!;
267 const childTurn = (yield* orchestrator.getThreadProjection(childThreadId))
268 .providerTurns[0]!;
269
270 const scope: McpInvocationScope = {
271 environmentId: EnvironmentId.make("environment:task-cancel-limit"),
272 requestNamespace: "provider-session:task-cancel-limit",
273 thread: {
274 threadId: parentThreadId,
275 providerSessionId: "provider-session:task-cancel-limit",
276 providerInstanceId: instanceId,
277 },
278 client: undefined,
279 capabilities: new Set(["orchestration"]),
280 issuedAt: 1,
281 };
282 const cancelTask = Effect.gen(function* () {
283 const result = yield* service.cancelTask(scope, {
284 taskId: task.id,
285 clientRequestId: "cancel-task",
286 });
287 yield* worker.drain();
288 return result.status;
289 });
290
291 const now = yield* DateTime.now;
292 const childFailed = yield* watch(
293 (event) =>
294 event.type === "run.updated" &&
295 event.payload.id === child.turn.runId &&
296 event.payload.status === "failed",
297 );
298 const hitLimit = Effect.forEach(
299 [
300 {
301 type: "provider_turn.updated",
302 driver,
303 providerTurn: { ...childTurn, status: "failed", completedAt: now },
304 },
305 {
306 type: "turn.terminal",
307 driver,
308 providerThreadId: childTurn.providerThreadId,
309 providerTurnId: childTurn.id,
310 runOrdinal: child.turn.runOrdinal,
311 failureItemOrdinal: 1,
312 status: "failed",
313 failure: {
314 class: "usage_limit",
315 message: "Claude usage limit reached.",
316 code: null,
317 retryable: null,
318 resetAt: DateTime.formatIso(DateTime.add(now, { minutes: 1 })),
319 },
320 threadDisposition: "reusable",
321 },
322 ] satisfies ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>,
323 (event) => Queue.offer(child.events, event),
324 { discard: true },
325 );
326 let cancelled: string | null = null;
327 if (cancel === "as its limit lands") {
328 limitInFlight = hitLimit;
329 cancelled = yield* cancelTask;
330 } else {
331 yield* hitLimit;
332 }
333 yield* Fiber.join(childFailed);
334 yield* worker.drain();
335
336 // The sweep runs every few seconds, so it arms the child before a parent reacts.
337 yield* sweep;
338 if (cancel === "after its limit") cancelled = yield* cancelTask;
339 assert.equal(cancelled, cancelStatus);
340
341 yield* TestClock.adjust("1 minute");
342 yield* sweep;
343 const childAfter = yield* orchestrator.getThreadProjection(childThreadId);
344 assert.deepEqual(
345 childAfter.messages.flatMap((message) => (message.role === "user" ? [message.text] : [])),
346 resumed ? ["Review the change", "Continue where you left off."] : ["Review the change"],
347 );
348 }).pipe(Effect.provide(layerTest));
349 }),
350);