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