1import {
2 CheckpointId,
3 CheckpointRef,
4 CheckpointScopeId,
5 CommandId,
6 EnvironmentId,
7 MessageId,
8 NodeId,
9 ORCHESTRATION_V2_WS_METHODS,
10 ProjectId,
11 ProviderInstanceId,
12 ProviderDriverKind,
13 RunId,
14 ThreadId,
15 TurnItemId,
16 WS_METHODS,
17 type OrchestrationV2Command,
18 type OrchestrationV2ThreadLaunchInput,
19 type OrchestrationV2ThreadProjection,
20 type ProjectMutation,
21} from "@t3tools/contracts";
22import { describe, expect, it } from "@effect/vitest";
23import * as Cause from "effect/Cause";
24import * as SecureRandom from "effect/crypto/SecureRandom";
25import * as Deferred from "effect/Deferred";
26import * as Effect from "effect/Effect";
27import * as Exit from "effect/Exit";
28import * as Fiber from "effect/Fiber";
29import * as Layer from "effect/Layer";
30import * as Option from "effect/Option";
31import * as SubscriptionRef from "effect/SubscriptionRef";
32import * as TestClock from "effect/testing/TestClock";
33import { RpcClientError } from "effect/rpc";
34
35import {
36 AVAILABLE_CONNECTION_STATE,
37 PrimaryConnectionTarget,
38 type PreparedConnection,
39 type SupervisorConnectionState,
40} from "../connection/model.ts";
41import * as EnvironmentSupervisor from "../connection/supervisor.ts";
42import * as RpcSession from "../rpc/session.ts";
43import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
44import { v2Now, v2Projection, v2ThreadId } from "../state/orchestrationV2TestFixtures.ts";
45import {
46 archiveThread,
47 interruptThreadTurn,
48 revertThreadCheckpoint,
49 resumeThreadQueue,
50 startThreadTurn,
51} from "./commands.ts";
52
53const layerTestRandom = Layer.succeed(
54 SecureRandom.SecureRandom,
55 SecureRandom.make((size) => new Uint8Array(size)),
56);
57
58const TARGET = new PrimaryConnectionTarget({
59 environmentId: EnvironmentId.make("environment-1"),
60 label: "Test environment",
61 httpBaseUrl: "https://environment.example.test",
62 wsBaseUrl: "wss://environment.example.test",
63});
64
65const CONNECTED_STATE: SupervisorConnectionState = {
66 ...AVAILABLE_CONNECTION_STATE,
67 desired: true,
68 phase: "connected",
69};
70
71const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(function* (input: {
72 readonly commands: OrchestrationV2Command[];
73 readonly projects: ProjectMutation[];
74 readonly launches?: OrchestrationV2ThreadLaunchInput[];
75 readonly projection?: OrchestrationV2ThreadProjection;
76 readonly projectionRequests?: ThreadId[];
77 readonly advertiseThreadStop?: boolean;
78}) {
79 const client = {
80 [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) =>
81 Effect.sync(() => {
82 input.commands.push(command);
83 return { sequence: input.commands.length };
84 }),
85 [ORCHESTRATION_V2_WS_METHODS.getThreadProjection]: (requestInput: {
86 readonly threadId: ThreadId;
87 }) =>
88 Effect.sync(() => {
89 input.projectionRequests?.push(requestInput.threadId);
90 return input.projection ?? v2Projection;
91 }),
92 [ORCHESTRATION_V2_WS_METHODS.launchThread]: (launchInput: OrchestrationV2ThreadLaunchInput) =>
93 Effect.sync(() => {
94 input.launches?.push(launchInput);
95 return {
96 threadId: launchInput.threadId ?? v2ThreadId,
97 projection: input.projection ?? v2Projection,
98 resumed: false,
99 };
100 }),
101 [WS_METHODS.projectsMutate]: (mutation: ProjectMutation) =>
102 Effect.sync(() => {
103 input.projects.push(mutation);
104 return {
105 id: mutation.projectId,
106 title: mutation.type === "project.create" ? mutation.title : "Project",
107 workspaceRoot:
108 mutation.type === "project.create" ? mutation.workspaceRoot : "/workspace/project",
109 repositoryIdentity: null,
110 faviconPath: null,
111 defaultModelSelection: null,
112 scripts: [],
113 createdAt: "2026-06-06T00:00:00.000Z",
114 updatedAt: "2026-06-06T00:00:00.000Z",
115 deletedAt: null,
116 };
117 }),
118 } as unknown as WsRpcProtocolClient;
119 const session: RpcSession.RpcSession = {
120 client,
121 initialConfig: Effect.succeed({
122 environment: {
123 capabilities: {
124 repositoryIdentity: true,
125 ...(input.advertiseThreadStop === true ? { threadStop: true } : {}),
126 },
127 },
128 } as never),
129 subscribeServerConfig: (input = {}) => client.subscribeServerConfig(input),
130 ready: Effect.void,
131 probe: Effect.void,
132 inboundFrames: Effect.succeed(0),
133 closed: Effect.never,
134 };
135 return EnvironmentSupervisor.EnvironmentSupervisor.of({
136 target: TARGET,
137 state: yield* SubscriptionRef.make(CONNECTED_STATE),
138 session: yield* SubscriptionRef.make(Option.some(session)),
139 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
140 connect: Effect.void,
141 disconnect: Effect.void,
142 retryNow: Effect.void,
143 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
144});
145
146describe("V2 environment commands", () => {
147 it.effect("distinguishes sending a selected queued message from resuming queue order", () =>
148 Effect.gen(function* () {
149 const commands: OrchestrationV2Command[] = [];
150 const supervisor = yield* makeSupervisor({ commands, projects: [] });
151 const threadId = ThreadId.make("thread-1");
152 yield* resumeThreadQueue({
153 commandId: CommandId.make("send-selected"),
154 threadId,
155 runId: RunId.make("selected"),
156 }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
157 yield* resumeThreadQueue({
158 commandId: CommandId.make("resume-queue"),
159 threadId,
160 }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
161
162 expect(commands).toEqual([
163 { type: "queued-run.start", commandId: "send-selected", threadId, runId: "selected" },
164 { type: "queue.resume", commandId: "resume-queue", threadId },
165 ]);
166 }).pipe(Effect.provide(layerTestRandom)),
167 );
168
169 it.effect("preserves caller command ids for idempotent V2 commands", () =>
170 Effect.gen(function* () {
171 const commands: OrchestrationV2Command[] = [];
172 const supervisor = yield* makeSupervisor({ commands, projects: [] });
173
174 yield* archiveThread({
175 commandId: CommandId.make("queued-command"),
176 threadId: ThreadId.make("thread-1"),
177 }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
178
179 expect(commands).toEqual([
180 { type: "thread.archive", commandId: "queued-command", threadId: "thread-1" },
181 ]);
182 }).pipe(Effect.provide(layerTestRandom)),
183 );
184
185 it.effect("resolves run ordinal zero to the persisted thread-start checkpoint", () =>
186 Effect.gen(function* () {
187 const scopeId = CheckpointScopeId.make("checkpoint-scope-root");
188 const checkpointId = CheckpointId.make("checkpoint-thread-start");
189 const projection: OrchestrationV2ThreadProjection = {
190 ...v2Projection,
191 checkpoints: [
192 {
193 id: checkpointId,
194 threadId: v2ThreadId,
195 scopeId,
196 runId: null,
197 nodeId: NodeId.make("node-run-1"),
198 parentCheckpointId: null,
199 ordinalWithinScope: 0,
200 appRunOrdinal: null,
201 ref: CheckpointRef.make("refs/t3/thread-start"),
202 status: "ready",
203 files: [],
204 capturedAt: v2Now,
205 },
206 ],
207 };
208 const commands: OrchestrationV2Command[] = [];
209 const supervisor = yield* makeSupervisor({ commands, projects: [], projection });
210
211 yield* revertThreadCheckpoint({
212 commandId: CommandId.make("rollback-thread-start"),
213 threadId: v2ThreadId,
214 turnCount: 0,
215 }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
216
217 expect(commands).toEqual([
218 {
219 type: "checkpoint.rollback",
220 commandId: "rollback-thread-start",
221 threadId: v2ThreadId,
222 scopeId,
223 checkpointId,
224 },
225 ]);
226 }).pipe(Effect.provide(layerTestRandom)),
227 );
228
229 it.effect("preserves an existing worktree and branch during first-message launch", () =>
230 Effect.gen(function* () {
231 const launches: OrchestrationV2ThreadLaunchInput[] = [];
232 const supervisor = yield* makeSupervisor({ commands: [], projects: [], launches });
233
234 yield* startThreadTurn({
235 commandId: CommandId.make("launch-existing-worktree"),
236 threadId: v2ThreadId,
237 message: {
238 messageId: MessageId.make("message-existing-worktree"),
239 role: "user",
240 text: "Continue here",
241 attachments: [],
242 },
243 runtimeMode: "full-access",
244 interactionMode: "default",
245 titleSeed: "Continue here",
246 bootstrap: {
247 createThread: {
248 projectId: ProjectId.make("project-1"),
249 title: "Thread",
250 modelSelection: v2Projection.thread.modelSelection,
251 runtimeMode: "full-access",
252 interactionMode: "default",
253 branch: "feature",
254 worktreePath: "/workspace/project-worktrees/feature",
255 createdAt: "2026-06-20T00:00:00.000Z",
256 },
257 },
258 }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
259
260 expect(launches[0]).toMatchObject({
261 threadId: v2ThreadId,
262 title: "Continue here",
263 generateTitle: true,
264 workspaceStrategy: {
265 type: "existing_worktree",
266 worktreePath: "/workspace/project-worktrees/feature",
267 branch: "feature",
268 },
269 });
270 }).pipe(Effect.provide(layerTestRandom)),
271 );
272
273 it.effect("uses server-resolved delivery intent without fetching the full projection", () =>
274 Effect.gen(function* () {
275 const commands: OrchestrationV2Command[] = [];
276 const projectionRequests: ThreadId[] = [];
277 const supervisor = yield* makeSupervisor({ commands, projects: [], projectionRequests });
278
279 for (const [mode, expectedType] of [
280 ["queue", "queue_after_active"],
281 ["auto", "start_immediately"],
282 ["steer", "start_immediately"],
283 ["restart", "start_immediately"],
284 ] as const) {
285 yield* startThreadTurn({
286 commandId: CommandId.make(`command-${mode}`),
287 threadId: v2ThreadId,
288 message: {
289 messageId: MessageId.make(`message-${mode}`),
290 role: "user",
291 text: mode,
292 attachments: [],
293 },
294 runtimeMode: "full-access",
295 interactionMode: "default",
296 dispatchMode: mode,
297 }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
298
299 expect(commands.at(-1)).toMatchObject({
300 type: "message.dispatch",
301 dispatchMode: { type: expectedType },
302 ...(mode === "queue" ? {} : { deliveryIntent: mode }),
303 });
304 }
305 expect(projectionRequests).toEqual([]);
306 }).pipe(Effect.provide(layerTestRandom)),
307 );
308
309 // A dead socket reaches the caller as a protocol error, or as an interrupt when
310 // the supervisor closes the session before the protocol reports the error.
311 it.effect.each([
312 [
313 "protocol error",
314 new RpcClientError.RpcClientError({
315 reason: new RpcClientError.RpcClientDefect({
316 message: "socket closed",
317 cause: new Error("socket closed"),
318 }),
319 }),
320 ],
321 ["session close", Effect.failCause(Cause.interrupt())],
322 ] as const)("retries a message dispatch lost to a %s with the same command id", ([, loss]) =>
323 Effect.gen(function* () {
324 const commands: OrchestrationV2Command[] = [];
325 const supervisor = yield* makeSupervisor({ commands, projects: [] });
326 const healthySession = Option.getOrThrow(yield* SubscriptionRef.get(supervisor.session));
327 const firstAttemptLost = yield* Deferred.make<void>();
328 const lostClient = {
329 ...healthySession.client,
330 [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) =>
331 Effect.gen(function* () {
332 commands.push(command);
333 yield* SubscriptionRef.set(supervisor.session, Option.none());
334 yield* Deferred.succeed(firstAttemptLost, undefined);
335 return yield* loss;
336 }),
337 } as unknown as WsRpcProtocolClient;
338 yield* SubscriptionRef.set(
339 supervisor.session,
340 Option.some({ ...healthySession, client: lostClient }),
341 );
342
343 const sendFiber = yield* startThreadTurn({
344 commandId: CommandId.make("command-lost-ack"),
345 threadId: v2ThreadId,
346 message: {
347 messageId: MessageId.make("message-lost-ack"),
348 role: "user",
349 text: "hello",
350 attachments: [],
351 },
352 runtimeMode: "full-access",
353 interactionMode: "default",
354 }).pipe(
355 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
356 Effect.forkChild,
357 );
358 yield* Deferred.await(firstAttemptLost);
359 yield* SubscriptionRef.set(supervisor.session, Option.some(healthySession));
360 const exit = yield* Fiber.await(sendFiber);
361
362 expect(commands.map((command) => [command.type, command.commandId])).toEqual([
363 ["message.dispatch", "command-lost-ack"],
364 ["message.dispatch", "command-lost-ack"],
365 ]);
366 expect(Exit.isSuccess(exit)).toBe(true);
367 }).pipe(Effect.provide(layerTestRandom)),
368 );
369
370 it.effect("stops resending a message whose dispatch keeps losing the socket", () =>
371 Effect.gen(function* () {
372 const commands: OrchestrationV2Command[] = [];
373 const supervisor = yield* makeSupervisor({ commands, projects: [] });
374 const healthySession = Option.getOrThrow(yield* SubscriptionRef.get(supervisor.session));
375 const dyingClient: WsRpcProtocolClient = {
376 ...healthySession.client,
377 [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) =>
378 Effect.gen(function* () {
379 commands.push(command);
380 yield* SubscriptionRef.set(
381 supervisor.session,
382 Option.some({ ...healthySession, client: dyingClient }),
383 );
384 return yield* new RpcClientError.RpcClientError({
385 reason: new RpcClientError.RpcClientDefect({
386 message: "socket closed",
387 cause: new Error("socket closed"),
388 }),
389 });
390 }),
391 } as unknown as WsRpcProtocolClient;
392 yield* SubscriptionRef.set(
393 supervisor.session,
394 Option.some({ ...healthySession, client: dyingClient }),
395 );
396
397 const error = yield* startThreadTurn({
398 commandId: CommandId.make("command-socket-killer"),
399 threadId: v2ThreadId,
400 message: {
401 messageId: MessageId.make("message-socket-killer"),
402 role: "user",
403 text: "hello",
404 attachments: [],
405 },
406 runtimeMode: "full-access",
407 interactionMode: "default",
408 }).pipe(
409 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
410 Effect.flip,
411 );
412
413 expect(commands.map((command) => command.commandId)).toEqual([
414 "command-socket-killer",
415 "command-socket-killer",
416 "command-socket-killer",
417 ]);
418 expect(error._tag).toBe("RpcClientError");
419 }).pipe(Effect.provide(layerTestRandom)),
420 );
421
422 it.effect("fails a message dispatch the host never answers on a live socket", () =>
423 Effect.gen(function* () {
424 const commands: OrchestrationV2Command[] = [];
425 const supervisor = yield* makeSupervisor({ commands, projects: [] });
426 const healthySession = Option.getOrThrow(yield* SubscriptionRef.get(supervisor.session));
427 const silentClient = {
428 ...healthySession.client,
429 [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) =>
430 Effect.sync(() => commands.push(command)).pipe(Effect.andThen(Effect.never)),
431 } as unknown as WsRpcProtocolClient;
432 yield* SubscriptionRef.set(
433 supervisor.session,
434 Option.some({ ...healthySession, client: silentClient }),
435 );
436
437 const sendFiber = yield* startThreadTurn({
438 commandId: CommandId.make("command-unanswered"),
439 threadId: v2ThreadId,
440 message: {
441 messageId: MessageId.make("message-unanswered"),
442 role: "user",
443 text: "hello",
444 attachments: [],
445 },
446 runtimeMode: "full-access",
447 interactionMode: "default",
448 }).pipe(
449 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
450 Effect.flip,
451 Effect.forkChild,
452 );
453 yield* TestClock.adjust("1 minute");
454
455 expect(sendFiber.pollUnsafe()).toBeDefined();
456 expect((yield* Fiber.join(sendFiber))._tag).toBe("EnvironmentRpcUnavailableError");
457 expect(commands.map((command) => command.commandId)).toEqual(["command-unanswered"]);
458 }).pipe(Effect.provide(layerTestRandom)),
459 );
460
461 it.effect("holds a lost message through reconnect backoff until disconnect", () =>
462 Effect.gen(function* () {
463 const commands: OrchestrationV2Command[] = [];
464 const supervisor = yield* makeSupervisor({ commands, projects: [] });
465 const healthySession = Option.getOrThrow(yield* SubscriptionRef.get(supervisor.session));
466 const lostClient = {
467 ...healthySession.client,
468 [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) =>
469 Effect.gen(function* () {
470 commands.push(command);
471 yield* SubscriptionRef.set(supervisor.session, Option.none());
472 return yield* new RpcClientError.RpcClientError({
473 reason: new RpcClientError.RpcClientDefect({
474 message: "socket closed",
475 cause: new Error("socket closed"),
476 }),
477 });
478 }),
479 } as unknown as WsRpcProtocolClient;
480 const send = (resendAfterSocketLoss: boolean) =>
481 Effect.gen(function* () {
482 yield* SubscriptionRef.set(
483 supervisor.session,
484 Option.some({ ...healthySession, client: lostClient }),
485 );
486 return yield* startThreadTurn({
487 commandId: CommandId.make(`command-host-gone-${resendAfterSocketLoss}`),
488 threadId: v2ThreadId,
489 message: {
490 messageId: MessageId.make(`message-host-gone-${resendAfterSocketLoss}`),
491 role: "user",
492 text: "hello",
493 attachments: [],
494 },
495 runtimeMode: "full-access",
496 interactionMode: "default",
497 resendAfterSocketLoss,
498 }).pipe(
499 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
500 Effect.flip,
501 Effect.forkChild,
502 );
503 });
504
505 const optedOut = yield* send(false);
506 expect((yield* Fiber.join(optedOut))._tag).toBe("RpcClientError");
507
508 const waiting = yield* send(true);
509 yield* SubscriptionRef.set(supervisor.state, {
510 ...CONNECTED_STATE,
511 phase: "backoff" as const,
512 });
513 yield* TestClock.adjust("2 minutes");
514 expect(waiting.pollUnsafe()).toBeUndefined();
515 yield* SubscriptionRef.set(supervisor.state, AVAILABLE_CONNECTION_STATE);
516 expect((yield* Fiber.join(waiting))._tag).toBe("RpcClientError");
517 expect(commands.map((command) => command.commandId)).toEqual([
518 "command-host-gone-false",
519 "command-host-gone-true",
520 ]);
521 }).pipe(Effect.provide(layerTestRandom)),
522 );
523
524 it.effect("Stop with no run left ends the thread's pull request watches", () =>
525 Effect.gen(function* () {
526 const link = {
527 host: "github.com",
528 repository: "pingdotgg/t3code",
529 url: "https://github.com/pingdotgg/t3code/pull/7",
530 source: "agent" as const,
531 linkedAt: "2026-10-05T00:00:00.000Z",
532 snapshot: null,
533 stack: null,
534 };
535 const watch = {
536 startedAt: "2026-10-05T00:00:00.000Z",
537 headSha: null,
538 failedChecks: [],
539 passed: false,
540 passedChecks: [],
541 remarksThrough: "2026-10-05T00:00:00.000Z",
542 remarkIds: [],
543 conflicting: false,
544 wakes: 0,
545 };
546 const projection: OrchestrationV2ThreadProjection = {
547 ...v2Projection,
548 thread: {
549 ...v2Projection.thread,
550 pullRequests: [
551 { ...link, number: 7, watch },
552 { ...link, number: 8 },
553 { ...link, number: 9, source: "stack-dismissed", watch },
554 ],
555 },
556 };
557 const commands: OrchestrationV2Command[] = [];
558 const supervisor = yield* makeSupervisor({ commands, projects: [], projection });
559
560 yield* interruptThreadTurn({ threadId: v2ThreadId }).pipe(
561 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
562 );
563
564 expect(commands).toEqual([
565 {
566 type: "thread.pull-request.watch",
567 commandId: expect.any(String),
568 threadId: v2ThreadId,
569 host: "github.com",
570 repository: "pingdotgg/t3code",
571 number: 7,
572 watching: false,
573 },
574 ]);
575 }).pipe(Effect.provide(layerTestRandom)),
576 );
577
578 it.effect.each(["waiting", "completed", "rolled_back"] as const)(
579 "dispatches Stop for %s runs with background commands except rollback",
580 (status) =>
581 Effect.gen(function* () {
582 const waitingRunId = RunId.make("run-waiting");
583 const projection: OrchestrationV2ThreadProjection = {
584 ...v2Projection,
585 runs: [
586 {
587 id: waitingRunId,
588 threadId: v2ThreadId,
589 ordinal: 1,
590 providerInstanceId: v2Projection.thread.providerInstanceId,
591 modelSelection: v2Projection.thread.modelSelection,
592 providerThreadId: null,
593 userMessageId: MessageId.make("message-waiting"),
594 rootNodeId: null,
595 activeAttemptId: null,
596 status,
597 requestedAt: v2Now,
598 startedAt: v2Now,
599 completedAt: null,
600 checkpointId: null,
601 contextHandoffId: null,
602 },
603 ],
604 turnItems: [
605 {
606 id: TurnItemId.make("background-command"),
607 threadId: v2ThreadId,
608 runId: waitingRunId,
609 nodeId: null,
610 providerThreadId: null,
611 providerTurnId: null,
612 nativeItemRef: null,
613 parentItemId: null,
614 ordinal: 1,
615 status: "running",
616 title: null,
617 startedAt: v2Now,
618 completedAt: null,
619 updatedAt: v2Now,
620 type: "command_execution",
621 input: "vp run dev",
622 },
623 ],
624 };
625 const commands: OrchestrationV2Command[] = [];
626 const supervisor = yield* makeSupervisor({ commands, projects: [], projection });
627
628 const result = yield* interruptThreadTurn({ threadId: v2ThreadId }).pipe(
629 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
630 );
631
632 expect(result).toEqual({ sequence: status === "rolled_back" ? 0 : 1 });
633 expect(commands).toEqual(
634 status === "rolled_back"
635 ? []
636 : [
637 {
638 type: "run.interrupt",
639 commandId: expect.any(String),
640 threadId: v2ThreadId,
641 runId: waitingRunId,
642 holdQueue: true,
643 },
644 ],
645 );
646 }).pipe(Effect.provide(layerTestRandom)),
647 );
648
649 it.effect(
650 "a native subagent stop targets its owning run rather than holding the parent queue",
651 () =>
652 Effect.gen(function* () {
653 const subagentId = NodeId.make("native-subagent");
654 const runId = RunId.make("settled-owner-run");
655 const commands: OrchestrationV2Command[] = [];
656 const supervisor = yield* makeSupervisor({
657 commands,
658 projects: [],
659 projection: {
660 ...v2Projection,
661 subagents: [
662 {
663 id: subagentId,
664 threadId: v2ThreadId,
665 runId,
666 parentNodeId: NodeId.make("owner-node"),
667 origin: "provider_native",
668 createdBy: "agent",
669 driver: ProviderDriverKind.make("claudeAgent"),
670 providerInstanceId: ProviderInstanceId.make("claudeAgent"),
671 providerThreadId: null,
672 childThreadId: ThreadId.make("native-child-thread"),
673 nativeTaskRef: null,
674 prompt: "Work",
675 title: null,
676 model: null,
677 result: null,
678 status: "running",
679 startedAt: v2Now,
680 completedAt: null,
681 updatedAt: v2Now,
682 },
683 ],
684 },
685 });
686 yield* interruptThreadTurn({
687 commandId: CommandId.make("native-stop"),
688 threadId: v2ThreadId,
689 subagentId,
690 }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
691 expect(commands).toEqual([
692 {
693 type: "subagent.stop",
694 commandId: "native-stop",
695 threadId: v2ThreadId,
696 runId,
697 subagentId,
698 },
699 ]);
700 }).pipe(Effect.provide(layerTestRandom)),
701 );
702
703 it.effect(
704 "an app-owned subagent stop asks the parent to clear an orphan, then stops the child",
705 () =>
706 Effect.gen(function* () {
707 const subagentId = NodeId.make("app-owned-subagent");
708 const runId = RunId.make("parent-run");
709 const childThreadId = ThreadId.make("app-owned-child-thread");
710 const commands: OrchestrationV2Command[] = [];
711 const supervisor = yield* makeSupervisor({
712 commands,
713 projects: [],
714 advertiseThreadStop: true,
715 projection: {
716 ...v2Projection,
717 subagents: [
718 {
719 id: subagentId,
720 threadId: v2ThreadId,
721 runId,
722 parentNodeId: NodeId.make("owner-node"),
723 origin: "app_owned",
724 createdBy: "agent",
725 driver: ProviderDriverKind.make("codex"),
726 providerInstanceId: ProviderInstanceId.make("codex"),
727 providerThreadId: null,
728 childThreadId,
729 nativeTaskRef: null,
730 prompt: "Work",
731 title: null,
732 model: null,
733 result: null,
734 status: "running",
735 startedAt: v2Now,
736 completedAt: null,
737 updatedAt: v2Now,
738 },
739 ],
740 },
741 });
742 yield* interruptThreadTurn({
743 commandId: CommandId.make("lineage-stop"),
744 threadId: v2ThreadId,
745 subagentId,
746 // A parent run must not be interrupted on the child thread.
747 runId,
748 }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
749 expect(commands).toEqual([
750 {
751 type: "subagent.stop",
752 commandId: expect.any(String),
753 threadId: v2ThreadId,
754 runId,
755 subagentId,
756 },
757 { type: "thread.stop", commandId: "lineage-stop", threadId: childThreadId },
758 ]);
759 }).pipe(Effect.provide(layerTestRandom)),
760 );
761
762 it.effect("stops a thread without relying on background work in the history window", () =>
763 Effect.gen(function* () {
764 const commands: OrchestrationV2Command[] = [];
765 const projectionRequests: ThreadId[] = [];
766 const supervisor = yield* makeSupervisor({
767 commands,
768 projects: [],
769 projectionRequests,
770 advertiseThreadStop: true,
771 });
772
773 const result = yield* interruptThreadTurn({
774 commandId: CommandId.make("stop-outside-history-window"),
775 threadId: v2ThreadId,
776 }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
777
778 expect(result).toEqual({ sequence: 1 });
779 expect(projectionRequests).toEqual([]);
780 expect(commands).toEqual([
781 {
782 type: "thread.stop",
783 commandId: "stop-outside-history-window",
784 threadId: v2ThreadId,
785 },
786 ]);
787 }).pipe(Effect.provide(layerTestRandom)),
788 );
789});