1import { assert, it } from "@effect/vitest";
2import {
3 CommandId,
4 EventId,
5 MessageId,
6 NodeId,
7 ProjectId,
8 ProviderDriverKind,
9 ProviderInstanceId,
10 ProviderSessionId,
11 ProviderTurnId,
12 RuntimeRequestId,
13 ThreadId,
14 TurnItemId,
15} from "@t3tools/contracts";
16import * as DateTime from "effect/DateTime";
17import * as Effect from "effect/Effect";
18import * as Exit from "effect/Exit";
19import * as Layer from "effect/Layer";
20import * as Schema from "effect/Schema";
21import * as SqlClient from "effect/sql/SqlClient";
22import * as Stream from "effect/Stream";
23import * as EventStore from "./EventStore.ts";
24import * as TestClock from "effect/testing/TestClock";
25
26import * as SqlitePersistence from "../persistence/Sqlite.ts";
27import * as ServerSettings from "../serverSettings.ts";
28import { CLAUDE_PROVIDER } from "./Adapters/ClaudeAdapterV2.ts";
29import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
30import * as CommandReceiptStore from "./CommandReceiptStore.ts";
31import { OrchestrationEffectRequestV2 } from "./EffectOutbox.ts";
32import * as Orchestrator from "./Orchestrator.ts";
33import * as EffectWorker from "./EffectWorker.ts";
34import * as EffectOutbox from "./EffectOutbox.ts";
35import * as ProjectionStore from "./ProjectionStore.ts";
36import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
37import * as ThreadManagementService from "./ThreadManagementService.ts";
38import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
39import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
40
41const instanceId = ProviderInstanceId.make("codex");
42const modelSelection = { instanceId, model: "gpt-5.1-codex" };
43const adapter = {
44 instanceId,
45 driver: ProviderDriverKind.make("codex"),
46 getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
47 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }),
48 openSession: () => Effect.die("Runs here never reach a provider"),
49} as ProviderAdapter.ProviderAdapterV2["Service"];
50const layerDatabase = SqlitePersistence.layerMemory;
51// No effect worker: runs stay unstarted, so Stop ends them without a provider.
52const layerTest = ThreadManagementService.layer.pipe(
53 Layer.provideMerge(
54 Layer.mergeAll(
55 layerDatabase,
56 ProjectionStore.layer.pipe(Layer.provide(layerDatabase)),
57 EventStore.layer.pipe(Layer.provide(layerDatabase)),
58 EffectOutbox.layer.pipe(Layer.provide(layerDatabase)),
59 ProviderReplayHarness.layerWithRegistry(
60 { name: "thread-stop" },
61 ProviderAdapterRegistry.layerFromAdapters([adapter]),
62 { databaseLayer: layerDatabase, runEffectWorker: false },
63 ),
64 ),
65 ),
66 Layer.provideMerge(ServerSettings.layerTest()),
67 Layer.provideMerge(CommandReceiptStore.layer.pipe(Layer.provide(layerDatabase))),
68);
69
70const encodeEffectRequest = Schema.encodeSync(Schema.fromJsonString(OrchestrationEffectRequestV2));
71
72const pullRequest = (number: number) => ({
73 host: "github.com",
74 repository: "pingdotgg/t3code",
75 number,
76});
77
78const createWatchingThread = (threadId: ThreadId, number: number) =>
79 Effect.gen(function* () {
80 const orchestrator = yield* Orchestrator.OrchestratorV2;
81 yield* orchestrator.dispatch({
82 type: "thread.create",
83 commandId: CommandId.make(`create:${threadId}`),
84 threadId,
85 projectId: ProjectId.make("project:thread-stop"),
86 title: threadId,
87 modelSelection,
88 runtimeMode: "full-access",
89 interactionMode: "default",
90 branch: null,
91 worktreePath: null,
92 createdBy: "user",
93 creationSource: "web",
94 });
95 yield* watch(threadId, number);
96 });
97
98const watch = (threadId: ThreadId, number: number) =>
99 Effect.gen(function* () {
100 const orchestrator = yield* Orchestrator.OrchestratorV2;
101 yield* orchestrator.dispatch({
102 type: "thread.pull-request.watch",
103 commandId: CommandId.make(`watch:${threadId}:${number}`),
104 threadId,
105 ...pullRequest(number),
106 watching: true,
107 link: { url: `https://github.com/pingdotgg/t3code/pull/${number}`, source: "agent" },
108 });
109 });
110
111const send = (threadId: ThreadId, text: string, type: "start_immediately" | "queue_after_active") =>
112 Effect.gen(function* () {
113 const orchestrator = yield* Orchestrator.OrchestratorV2;
114 yield* orchestrator.dispatch({
115 type: "message.dispatch",
116 commandId: CommandId.make(`send:${threadId}:${text}`),
117 threadId,
118 messageId: MessageId.make(`message:${threadId}:${text}`),
119 text,
120 attachments: [],
121 dispatchMode: { type },
122 createdBy: "user",
123 creationSource: "web",
124 });
125 });
126
127/** Delegates `task` from the parent's latest run and returns the child thread. */
128const delegate = (parentThreadId: ThreadId, task: string) =>
129 Effect.gen(function* () {
130 const orchestrator = yield* Orchestrator.OrchestratorV2;
131 const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.at(-1)!;
132 yield* orchestrator.dispatch({
133 type: "delegated_task.request",
134 commandId: CommandId.make(`delegate:${task}`),
135 parentThreadId,
136 parentRunId: parentRun.id,
137 parentNodeId: parentRun.rootNodeId!,
138 task,
139 modelSelection,
140 runtimeMode: "full-access",
141 interactionMode: "default",
142 completionWake: "always",
143 createdBy: "agent",
144 creationSource: "mcp",
145 });
146 const projection = yield* orchestrator.getThreadProjection(parentThreadId);
147 return projection.subagents.find((candidate) => candidate.prompt === task)!.childThreadId!;
148 });
149
150const threadState = (threadId: ThreadId) =>
151 Effect.gen(function* () {
152 const orchestrator = yield* Orchestrator.OrchestratorV2;
153 const projection = yield* orchestrator.getThreadProjection(threadId);
154 return {
155 runs: projection.runs.map((run) => `${run.status}${run.queueHeld === true ? ":held" : ""}`),
156 watched: (projection.thread.pullRequests ?? [])
157 .filter((link) => link.watch !== undefined)
158 .map((link) => link.number),
159 };
160 });
161
162it.effect("Stop ends watches, holds queues, and stops the delegated tasks under the thread", () =>
163 Effect.gen(function* () {
164 const orchestrator = yield* Orchestrator.OrchestratorV2;
165 const projections = yield* ProjectionStore.ProjectionStoreV2;
166 const threads = yield* ThreadManagementService.ThreadManagementService;
167 const sql = yield* SqlClient.SqlClient;
168 const parentThreadId = ThreadId.make("thread:stop-parent");
169 yield* createWatchingThread(parentThreadId, 1);
170 yield* send(parentThreadId, "first", "start_immediately");
171 const childThreadId = yield* delegate(parentThreadId, "child task");
172 // The parent owns its pull requests; a delegated task cannot watch one.
173 assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(childThreadId, 2))));
174 const grandchildThreadId = yield* delegate(childThreadId, "grandchild task");
175 yield* send(childThreadId, "child follow-up", "queue_after_active");
176
177 // The run that delegated the child ends before the run the user stops, so
178 // only a Stop that covers earlier runs keeps the stopped child from waking it.
179 const now = yield* DateTime.now;
180 const firstRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!;
181 yield* projections.apply({
182 id: EventId.make("event:stop-parent:first-run-completed"),
183 type: "run.updated",
184 threadId: parentThreadId,
185 runId: firstRun.id,
186 occurredAt: now,
187 payload: { ...firstRun, status: "completed", completedAt: now },
188 });
189 yield* send(parentThreadId, "second", "start_immediately");
190 const secondRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.at(-1)!;
191
192 const stopCommandId = CommandId.make("stop-parent");
193 yield* orchestrator.dispatch({
194 type: "run.interrupt",
195 commandId: stopCommandId,
196 threadId: parentThreadId,
197 runId: secondRun.id,
198 holdQueue: true,
199 });
200
201 assert.deepEqual(yield* threadState(parentThreadId), {
202 runs: ["completed", "interrupted"],
203 watched: [],
204 });
205 const parent = yield* orchestrator.getThreadProjection(parentThreadId);
206 assert.equal(parent.subagents[0]?.completionDelivery?.state, "disposed");
207 const effects = yield* sql<{ readonly effect_type: string }>`
208 SELECT effect_type FROM orchestration_v2_effect_outbox WHERE command_id = ${stopCommandId}
209 `;
210 assert.include(
211 effects.map((row) => row.effect_type),
212 "delegated-tasks.stop",
213 );
214
215 // What the delegated-tasks.stop effect runs once the Stop commits.
216 yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId });
217 assert.deepEqual(yield* threadState(childThreadId), {
218 runs: ["interrupted", "queued:held"],
219 watched: [],
220 });
221 assert.deepEqual((yield* threadState(grandchildThreadId)).runs, ["interrupted"]);
222
223 // A retried effect stops nothing twice.
224 yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId });
225 assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted", "queued:held"]);
226 }).pipe(Effect.provide(layerTest)),
227);
228
229it.effect.each(["thread.stop", "run.interrupt"] as const)(
230 "%s cancels buffered wakes and preserves durable queued messages",
231 (type) =>
232 Effect.gen(function* () {
233 const orchestrator = yield* Orchestrator.OrchestratorV2;
234 const threadId = ThreadId.make(`thread:stop-buffered:${type}`);
235 yield* createWatchingThread(threadId, 5);
236 yield* send(threadId, "active", "start_immediately");
237 yield* send(threadId, "user follow-up", "queue_after_active");
238 for (const creationSource of ["provider", "server"] as const) {
239 yield* orchestrator.dispatch({
240 type: "message.dispatch",
241 commandId: CommandId.make(`notification:${type}:${creationSource}`),
242 threadId,
243 messageId: MessageId.make(`notification:${type}:${creationSource}`),
244 text: "Background activity updated",
245 attachments: [],
246 createdBy: "agent",
247 creationSource,
248 notification: {
249 source: { kind: "background_task" },
250 outcome: "updated",
251 summary: "Background activity updated",
252 },
253 dispatchMode: { type: "queue_after_active" },
254 });
255 }
256 const before = yield* orchestrator.getThreadProjection(threadId);
257 const buffered = before.runs.find((run) =>
258 before.messages.some(
259 (message) => message.id === run.userMessageId && message.creationSource === "provider",
260 ),
261 )!;
262 const stop = { commandId: CommandId.make(`stop:${type}`), threadId };
263 yield* orchestrator.dispatch(
264 type === "run.interrupt"
265 ? { ...stop, type, runId: before.runs[0]!.id, holdQueue: true }
266 : { ...stop, type },
267 );
268 const stopped = yield* orchestrator.getThreadProjection(threadId);
269 assert.equal(stopped.runs.find((run) => run.id === buffered.id)?.status, "cancelled");
270 assert.equal(
271 stopped.attempts.find((attempt) => attempt.runId === buffered.id)?.status,
272 "cancelled",
273 );
274 assert.equal(
275 stopped.nodes.find((node) => node.id === buffered.rootNodeId)?.status,
276 "cancelled",
277 );
278 assert.deepEqual(
279 stopped.runs.filter((run) => run.status === "queued").map((run) => run.queueHeld),
280 [true, true],
281 );
282 }).pipe(Effect.provide(layerTest)),
283);
284
285it.effect(
286 "thread.stop ends an idle thread's watches and accepts a thread with nothing to stop",
287 () =>
288 Effect.gen(function* () {
289 const orchestrator = yield* Orchestrator.OrchestratorV2;
290 const threadId = ThreadId.make("thread:stop-idle");
291 yield* createWatchingThread(threadId, 3);
292
293 yield* orchestrator.dispatch({
294 type: "thread.stop",
295 commandId: CommandId.make("stop-idle"),
296 threadId,
297 });
298 assert.deepEqual(yield* threadState(threadId), { runs: [], watched: [] });
299
300 const again = yield* orchestrator.dispatch({
301 type: "thread.stop",
302 commandId: CommandId.make("stop-idle-again"),
303 threadId,
304 });
305 assert.lengthOf(again.storedEvents, 0);
306 }).pipe(Effect.provide(layerTest)),
307);
308
309it.effect("a run Stop reached cannot delegate or start a watch, even after it ends", () =>
310 Effect.gen(function* () {
311 const orchestrator = yield* Orchestrator.OrchestratorV2;
312 const projections = yield* ProjectionStore.ProjectionStoreV2;
313 const threadId = ThreadId.make("thread:stop-barrier");
314 yield* createWatchingThread(threadId, 4);
315 yield* send(threadId, "work", "start_immediately");
316 const now = yield* DateTime.now;
317 const run = { ...(yield* orchestrator.getThreadProjection(threadId)).runs[0]!, startedAt: now };
318 yield* projections.apply({
319 id: EventId.make("event:stop-barrier:running"),
320 type: "run.updated",
321 threadId,
322 runId: run.id,
323 occurredAt: now,
324 payload: { ...run, status: "running" },
325 });
326 // A queued message the user cancelled does not count as the latest run.
327 yield* send(threadId, "queued", "queue_after_active");
328 const queued = (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)!;
329 yield* orchestrator.dispatch({
330 type: "queued-run.cancel",
331 commandId: CommandId.make("cancel-queued"),
332 threadId,
333 runId: queued.id,
334 });
335 // Stop reached the run, but its provider has not stopped it yet.
336 yield* projections.apply({
337 id: EventId.make("event:stop-barrier:interrupt-request"),
338 type: "turn-item.updated",
339 threadId,
340 runId: run.id,
341 occurredAt: now,
342 payload: {
343 id: TurnItemId.make("turn-item:stop-barrier:interrupt-request"),
344 threadId,
345 runId: run.id,
346 nodeId: run.rootNodeId!,
347 providerThreadId: run.providerThreadId!,
348 providerTurnId: null,
349 nativeItemRef: null,
350 parentItemId: null,
351 ordinal: 1_000,
352 status: "completed",
353 title: "Interrupt requested",
354 startedAt: now,
355 completedAt: now,
356 updatedAt: now,
357 type: "run_interrupt_request",
358 message: "Stop",
359 },
360 });
361
362 assert.isTrue(Exit.isFailure(yield* Effect.exit(delegate(threadId, "late task"))));
363 assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(threadId, 5))));
364
365 // A slow watch_pull_request call can land after the stopped run ended.
366 yield* projections.apply({
367 id: EventId.make("event:stop-barrier:interrupted"),
368 type: "run.updated",
369 threadId,
370 runId: run.id,
371 occurredAt: now,
372 payload: { ...run, status: "interrupted", completedAt: now },
373 });
374 assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(threadId, 5))));
375 assert.deepEqual(yield* threadState(threadId), {
376 runs: ["interrupted", "cancelled"],
377 watched: [4],
378 });
379
380 // The user can still stop and restart a watch by hand.
381 for (const watching of [false, true]) {
382 yield* orchestrator.dispatch({
383 type: "thread.pull-request.watch",
384 commandId: CommandId.make(`manual-watch:${watching}`),
385 threadId,
386 ...pullRequest(4),
387 watching,
388 });
389 }
390 assert.deepEqual((yield* threadState(threadId)).watched, [4]);
391 }).pipe(Effect.provide(layerTest)),
392);
393
394it.effect("thread.stop keeps a restart continuation of the stopped run from starting", () =>
395 Effect.gen(function* () {
396 const orchestrator = yield* Orchestrator.OrchestratorV2;
397 const projections = yield* ProjectionStore.ProjectionStoreV2;
398 const sql = yield* SqlClient.SqlClient;
399 const threadId = ThreadId.make("thread:stop-restart");
400 yield* createWatchingThread(threadId, 8);
401 yield* send(threadId, "work", "start_immediately");
402 const now = yield* DateTime.now;
403 const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
404 // A server restart cut the run before its provider started; its continuation is pending.
405 yield* projections.apply({
406 id: EventId.make("event:stop-restart:cancelled"),
407 type: "run.updated",
408 threadId,
409 runId: run.id,
410 occurredAt: now,
411 payload: { ...run, status: "cancelled", startedAt: null, completedAt: now },
412 });
413 const at = DateTime.formatIso(now);
414 yield* sql`
415 INSERT INTO orchestration_v2_effect_outbox (
416 effect_id, command_id, thread_id, effect_type, payload_json, status,
417 attempt_count, available_at, created_at, updated_at
418 ) VALUES (
419 ${`effect:restart-continuation:${run.id}`}, 'command:restart', ${threadId},
420 'provider-runtime.continue',
421 ${encodeEffectRequest({ type: "provider-runtime.continue", sourceRunId: run.id })},
422 'pending', 0, ${at}, ${at}, ${at}
423 )
424 `;
425
426 yield* orchestrator.dispatch({
427 type: "thread.stop",
428 commandId: CommandId.make("stop-restart"),
429 threadId,
430 });
431 yield* orchestrator.dispatch({
432 type: "message.dispatch",
433 commandId: CommandId.make("restart-continuation"),
434 threadId,
435 messageId: MessageId.make("message:restart-continuation"),
436 text: "Continue.",
437 attachments: [],
438 dispatchMode: { type: "start_immediately" },
439 createdBy: "agent",
440 creationSource: "server",
441 restartContinuationOfRunId: run.id,
442 });
443 assert.deepEqual(yield* threadState(threadId), { runs: ["cancelled"], watched: [] });
444 }).pipe(Effect.provide(layerTest)),
445);
446
447it.effect.each(["thread.stop", "run.interrupt"] as const)(
448 "a successful %s blocks a newer restart continuation when completion timestamps tie",
449 (stopType) =>
450 Effect.gen(function* () {
451 const orchestrator = yield* Orchestrator.OrchestratorV2;
452 const projections = yield* ProjectionStore.ProjectionStoreV2;
453 const sql = yield* SqlClient.SqlClient;
454 const threadId = ThreadId.make("thread:stop-resumed-queue");
455 yield* createWatchingThread(threadId, 12);
456 yield* send(threadId, "first", "start_immediately");
457 yield* send(threadId, "queued", "queue_after_active");
458 const now = yield* DateTime.now;
459 const [first, queued] = (yield* orchestrator.getThreadProjection(threadId)).runs;
460 assert.isDefined(first);
461 assert.isDefined(queued);
462 for (const run of [first!, queued!]) {
463 yield* projections.apply({
464 id: EventId.make(`event:stop-resumed-queue:hold:${run.id}`),
465 type: "run.updated",
466 threadId,
467 runId: run.id,
468 occurredAt: now,
469 payload:
470 run.id === first!.id
471 ? { ...run, status: "completed", completedAt: now }
472 : { ...run, queueHeld: true },
473 });
474 }
475 // A later turn runs ahead of the held queue, then recovery cancels it.
476 yield* send(threadId, "restart source", "start_immediately");
477 const source = (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)!;
478 yield* projections.apply({
479 id: EventId.make("event:stop-resumed-queue:restart-cancelled"),
480 type: "run.updated",
481 threadId,
482 runId: source.id,
483 occurredAt: now,
484 payload: { ...source, status: "cancelled", startedAt: now, completedAt: now },
485 });
486 const at = DateTime.formatIso(now);
487 yield* sql`
488 INSERT INTO orchestration_v2_effect_outbox (
489 effect_id, command_id, thread_id, effect_type, payload_json, status,
490 attempt_count, available_at, created_at, updated_at
491 ) VALUES (
492 ${`effect:restart-continuation:${source.id}`}, 'command:restart', ${threadId},
493 'provider-runtime.continue',
494 ${encodeEffectRequest({ type: "provider-runtime.continue", sourceRunId: source.id })},
495 'pending', 0, ${at}, ${at}, ${at}
496 )
497 `;
498 yield* orchestrator.dispatch({
499 type: "queue.resume",
500 commandId: CommandId.make("resume-older-queue"),
501 threadId,
502 });
503 assert.equal(
504 (yield* orchestrator.getThreadProjection(threadId)).runs.find(
505 (run) => run.id === queued!.id,
506 )?.status,
507 "starting",
508 );
509 yield* orchestrator.dispatch(
510 stopType === "thread.stop"
511 ? { type: "thread.stop", commandId: CommandId.make("stop-resumed-queue"), threadId }
512 : {
513 type: "run.interrupt",
514 commandId: CommandId.make("stop-resumed-queue"),
515 threadId,
516 runId: queued!.id,
517 holdQueue: true,
518 },
519 );
520 const stopped = yield* orchestrator.getThreadProjection(threadId);
521 const interrupted = stopped.runs.find((run) => run.id === queued!.id)!;
522 assert.equal(interrupted.status, "interrupted");
523 assert.equal(DateTime.toEpochMillis(interrupted.completedAt!), DateTime.toEpochMillis(now));
524 yield* orchestrator.dispatch({
525 type: "message.dispatch",
526 commandId: CommandId.make("restart-after-stop"),
527 threadId,
528 messageId: MessageId.make("message:restart-after-stop"),
529 text: "Continue.",
530 attachments: [],
531 dispatchMode: { type: "start_immediately" },
532 createdBy: "agent",
533 creationSource: "server",
534 restartContinuationOfRunId: source.id,
535 });
536 assert.deepEqual(yield* threadState(threadId), {
537 runs: ["completed", "interrupted", "cancelled"],
538 watched: [],
539 });
540 }).pipe(Effect.provide(layerTest.pipe(Layer.provideMerge(TestClock.layer())))),
541);
542
543it.effect.each([
544 { ending: "cancelled", question: "resolved" },
545 { ending: "interrupted", question: "resolved" },
546 { ending: "completed", question: "pending" },
547 { ending: "failed", question: "pending" },
548] as const)(
549 "a $ending delegated task leaves its async question $question",
550 ({ ending, question }) =>
551 Effect.gen(function* () {
552 const orchestrator = yield* Orchestrator.OrchestratorV2;
553 const projections = yield* ProjectionStore.ProjectionStoreV2;
554 const parentThreadId = ThreadId.make(`thread:stop-question-parent:${ending}`);
555 yield* createWatchingThread(parentThreadId, 20);
556 yield* send(parentThreadId, "first", "start_immediately");
557 const childThreadId = yield* delegate(parentThreadId, `question task ${ending}`);
558 const childRun = (yield* orchestrator.getThreadProjection(childThreadId)).runs[0]!;
559 assert.equal(
560 (yield* orchestrator.getThreadProjection(childThreadId)).nodes.find(
561 (node) => node.id === childRun.rootNodeId,
562 )?.runId,
563 childRun.id,
564 );
565 const now = yield* DateTime.now;
566 yield* projections.apply({
567 id: EventId.make("event:stop-question:request"),
568 type: "runtime-request.updated",
569 threadId: childThreadId,
570 runId: childRun.id,
571 occurredAt: now,
572 payload: {
573 id: RuntimeRequestId.make("request:stop-question"),
574 nodeId: childRun.rootNodeId!,
575 providerTurnId: null,
576 nativeRequestRef: null,
577 kind: "user_input",
578 status: "pending",
579 responseCapability: { type: "message" },
580 createdAt: now,
581 resolvedAt: null,
582 },
583 });
584 const request = Effect.map(
585 orchestrator.getThreadProjection(childThreadId),
586 (projection) => projection.runtimeRequests[0],
587 );
588
589 // While the task runs, its question stays open.
590 yield* orchestrator.recoverDelegatedTask(childThreadId, childRun.id);
591 assert.equal((yield* request)?.status, "pending");
592 if (ending === "interrupted") {
593 // What task_cancel dispatches.
594 yield* orchestrator.dispatch({
595 type: "thread.stop",
596 commandId: CommandId.make("stop-question-child"),
597 threadId: childThreadId,
598 });
599 } else {
600 yield* projections.apply({
601 id: EventId.make(`event:stop-question:${ending}`),
602 type: "run.updated",
603 threadId: childThreadId,
604 runId: childRun.id,
605 occurredAt: now,
606 payload: { ...childRun, status: ending, startedAt: now, completedAt: now },
607 });
608 }
609 yield* orchestrator.recoverDelegatedTask(childThreadId, childRun.id);
610 assert.equal((yield* request)?.status, question);
611 assert.equal((yield* request)?.decision, question === "resolved" ? "cancel" : undefined);
612 }).pipe(Effect.provide(layerTest)),
613);
614
615it.effect("a delegated task that cannot be stopped fails the walk after its siblings stop", () =>
616 Effect.gen(function* () {
617 const orchestrator = yield* Orchestrator.OrchestratorV2;
618 const projections = yield* ProjectionStore.ProjectionStoreV2;
619 const threads = yield* ThreadManagementService.ThreadManagementService;
620 const parentThreadId = ThreadId.make("thread:stop-partial");
621 yield* createWatchingThread(parentThreadId, 6);
622 yield* send(parentThreadId, "work", "start_immediately");
623 const childThreadId = yield* delegate(parentThreadId, "stoppable task");
624 const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!;
625 // A task whose thread is gone, listed before the one that can stop.
626 yield* projections.apply({
627 id: EventId.make("event:stop-partial:missing-task"),
628 type: "subagent.updated",
629 threadId: parentThreadId,
630 runId: task.runId!,
631 nodeId: NodeId.make("node:stop-partial:missing"),
632 providerInstanceId: instanceId,
633 occurredAt: yield* DateTime.now,
634 payload: {
635 ...task,
636 id: NodeId.make("node:stop-partial:missing"),
637 childThreadId: ThreadId.make("thread:stop-partial:missing"),
638 startedAt: DateTime.subtract(task.startedAt!, { hours: 1 }),
639 },
640 });
641
642 const walked = yield* Effect.exit(
643 threads.stopDelegatedTasks({
644 threadId: parentThreadId,
645 commandId: CommandId.make("stop-partial"),
646 }),
647 );
648 assert.isTrue(Exit.isFailure(walked));
649 assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted"]);
650 }).pipe(Effect.provide(layerTest)),
651);
652
653it.effect("thread.stop on a finished thread refuses a late agent watch", () =>
654 Effect.gen(function* () {
655 const orchestrator = yield* Orchestrator.OrchestratorV2;
656 const projections = yield* ProjectionStore.ProjectionStoreV2;
657 const threadId = ThreadId.make("thread:stop-finished");
658 yield* createWatchingThread(threadId, 9);
659 yield* send(threadId, "work", "start_immediately");
660 const now = yield* DateTime.now;
661 const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
662 yield* projections.apply({
663 id: EventId.make("event:stop-finished:completed"),
664 type: "run.updated",
665 threadId,
666 runId: run.id,
667 occurredAt: now,
668 payload: { ...run, status: "completed", startedAt: now, completedAt: now },
669 });
670
671 yield* orchestrator.dispatch({
672 type: "thread.stop",
673 commandId: CommandId.make("stop-finished"),
674 threadId,
675 });
676 assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(threadId, 10))));
677 assert.deepEqual(yield* threadState(threadId), { runs: ["completed"], watched: [] });
678 }).pipe(Effect.provide(layerTest)),
679);
680
681it.effect("thread.stop marks a turn it cannot interrupt so a late agent watch is refused", () =>
682 Effect.gen(function* () {
683 const orchestrator = yield* Orchestrator.OrchestratorV2;
684 const projections = yield* ProjectionStore.ProjectionStoreV2;
685 const threadId = ThreadId.make("thread:stop-lost-session");
686 yield* createWatchingThread(threadId, 11);
687 yield* send(threadId, "work", "start_immediately");
688 const now = yield* DateTime.now;
689 const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
690 // The turn is running, but its provider session is gone, so the interrupt fails.
691 yield* projections.apply({
692 id: EventId.make("event:stop-lost-session:running"),
693 type: "run.updated",
694 threadId,
695 runId: run.id,
696 occurredAt: now,
697 payload: { ...run, status: "running", startedAt: now },
698 });
699 yield* projections.apply({
700 id: EventId.make("event:stop-lost-session:turn"),
701 type: "provider-turn.updated",
702 threadId,
703 occurredAt: now,
704 payload: {
705 id: ProviderTurnId.make("provider-turn:stop-lost-session"),
706 providerThreadId: run.providerThreadId!,
707 nodeId: run.rootNodeId!,
708 runAttemptId: run.activeAttemptId!,
709 nativeTurnRef: null,
710 ordinal: 1,
711 status: "running",
712 startedAt: now,
713 completedAt: null,
714 },
715 });
716
717 yield* orchestrator.dispatch({
718 type: "thread.stop",
719 commandId: CommandId.make("stop-lost-session"),
720 threadId,
721 });
722 assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(threadId, 12))));
723 assert.deepEqual(yield* threadState(threadId), { runs: ["running"], watched: [] });
724 }).pipe(Effect.provide(layerTest)),
725);
726
727it.effect.each(["missing session", "completed", "failed", "cancelled"] as const)(
728 "a native child Stop only settles a confirmed %s outcome and preserves its owner and sibling",
729 (providerState) =>
730 Effect.gen(function* () {
731 const orchestrator = yield* Orchestrator.OrchestratorV2;
732 const projections = yield* ProjectionStore.ProjectionStoreV2;
733 const sql = yield* SqlClient.SqlClient;
734 const threadId = ThreadId.make("thread:native-child-stop");
735 yield* createWatchingThread(threadId, 20);
736 yield* send(threadId, "owner", "start_immediately");
737 yield* delegate(threadId, "native-child-fixture");
738 const before = yield* orchestrator.getThreadProjection(threadId);
739 const run = before.runs[0]!;
740 const now = yield* DateTime.now;
741 const subagent = {
742 ...before.subagents[0]!,
743 origin: "provider_native" as const,
744 driver: CLAUDE_PROVIDER,
745 providerInstanceId: ProviderInstanceId.make("claudeAgent"),
746 nativeTaskRef: {
747 driver: CLAUDE_PROVIDER,
748 nativeId: "claude-task",
749 strength: "strong" as const,
750 },
751 status: "running" as const,
752 startedAt: now,
753 };
754 yield* projections.apply({
755 id: EventId.make("event:native-child-stop:agent"),
756 type: "subagent.updated",
757 threadId,
758 runId: run.id,
759 nodeId: subagent.id,
760 occurredAt: now,
761 payload: subagent,
762 });
763 const sibling = {
764 ...subagent,
765 id: NodeId.make("node:native-stop-sibling"),
766 childThreadId: null,
767 };
768 yield* projections.apply({
769 id: EventId.make("event:native-child-stop:sibling"),
770 type: "subagent.updated",
771 threadId,
772 runId: run.id,
773 nodeId: sibling.id,
774 occurredAt: now,
775 payload: sibling,
776 });
777 const item = before.turnItems.find(
778 (item) => item.type === "subagent" && item.subagentId === subagent.id,
779 )!;
780 yield* projections.apply({
781 id: EventId.make("event:native-child-stop:item"),
782 type: "turn-item.updated",
783 threadId,
784 runId: run.id,
785 occurredAt: now,
786 payload: { ...item, status: "running", completedAt: null },
787 });
788 const childRoot = {
789 ...before.nodes.find((node) => node.id === run.rootNodeId)!,
790 id: NodeId.make("node:native-stop-child-root"),
791 threadId: subagent.childThreadId!,
792 runId: null,
793 parentNodeId: null,
794 status: "running" as const,
795 countsForRun: false,
796 completedAt: null,
797 };
798 yield* projections.apply({
799 id: EventId.make("event:native-child-stop:child-root"),
800 type: "node.updated",
801 threadId: childRoot.threadId,
802 nodeId: childRoot.id,
803 occurredAt: now,
804 payload: childRoot,
805 });
806 const providerThread = before.providerThreads.find(
807 (thread) => thread.id === run.providerThreadId,
808 )!;
809 yield* projections.apply({
810 id: EventId.make("event:native-child-stop:provider-thread"),
811 type: "provider-thread.updated",
812 threadId,
813 occurredAt: now,
814 payload: {
815 ...providerThread,
816 driver: CLAUDE_PROVIDER,
817 providerInstanceId: ProviderInstanceId.make("claudeAgent"),
818 providerSessionId: ProviderSessionId.make("session:native-child-stop"),
819 },
820 });
821 const providerTurnId = ProviderTurnId.make("turn:native-child-stop");
822 yield* projections.apply({
823 id: EventId.make("event:native-child-stop:provider-turn"),
824 type: "provider-turn.updated",
825 threadId,
826 occurredAt: now,
827 payload: {
828 id: providerTurnId,
829 providerThreadId: providerThread.id,
830 nodeId: run.rootNodeId!,
831 runAttemptId: run.activeAttemptId!,
832 nativeTurnRef: null,
833 ordinal: 1,
834 status: "completed",
835 startedAt: now,
836 completedAt: now,
837 },
838 });
839 // The owner has finished but its child and queued prompt remain.
840 yield* projections.apply({
841 id: EventId.make("event:native-child-stop:owner-completed"),
842 type: "run.updated",
843 threadId,
844 runId: run.id,
845 occurredAt: now,
846 payload: { ...run, status: "completed", startedAt: now, completedAt: now },
847 });
848 yield* send(threadId, "queued", "queue_after_active");
849 const stateBefore = yield* threadState(threadId);
850 const eventStore = yield* EventStore.EventStoreV2;
851 const afterSequence = yield* eventStore.latestSequence({ threadId });
852 const confirmedStatus =
853 providerState === "completed" || providerState === "failed" || providerState === "cancelled"
854 ? providerState
855 : null;
856 const confirmedOutcome =
857 confirmedStatus === null
858 ? undefined
859 : { status: confirmedStatus, completedAt: now, result: "Actual provider outcome" };
860 const commandId = CommandId.make("stop-native-child");
861 yield* orchestrator.dispatch({
862 type: "subagent.stop",
863 commandId,
864 threadId,
865 runId: run.id,
866 subagentId: subagent.id,
867 });
868 assert.deepEqual(yield* threadState(threadId), stateBefore);
869 const effects = yield* sql<{ payload_json: string }>`
870 SELECT payload_json FROM orchestration_v2_effect_outbox WHERE command_id = ${commandId}
871 `;
872 assert.equal(effects.length, 1);
873 assert.deepEqual(JSON.parse(effects[0]!.payload_json), {
874 type: "provider-turn.interrupt",
875 providerSessionId: "session:native-child-stop",
876 providerThreadId: providerThread.id,
877 providerTurnId,
878 subagent: { id: subagent.id, nativeTaskId: "claude-task" },
879 });
880 assert.equal(
881 (yield* orchestrator.getThreadProjection(threadId)).subagents[0]?.status,
882 "running",
883 );
884 const unrelated = yield* Effect.exit(
885 orchestrator.dispatch({
886 type: "subagent.stop",
887 commandId: CommandId.make("stop-unknown-child"),
888 threadId,
889 runId: run.id,
890 subagentId: NodeId.make("unknown-child"),
891 }),
892 );
893 assert.isTrue(Exit.isFailure(unrelated));
894 // The provider outcome predates the delayed projection repair.
895 yield* TestClock.adjust("1 second");
896 // Only execute Stop's durable effect, with no terminal event ingestion.
897 yield* sql`DELETE FROM orchestration_v2_effect_outbox WHERE command_id != ${commandId}`;
898 if (providerState === "missing session") {
899 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
900 yield* worker.drain();
901 } else {
902 yield* orchestrator.dispatch({
903 type: "thread.background-work.settle",
904 commandId: CommandId.make("repair-native-child"),
905 threadId,
906 providerThreadId: providerThread.id,
907 providerTurnId,
908 subagent: { id: subagent.id, stoppedAt: now, outcome: confirmedOutcome! },
909 });
910 }
911 const settled = yield* orchestrator.getThreadProjection(threadId);
912 const cancelled = settled.subagents.find((task) => task.id === subagent.id)!;
913 if (confirmedOutcome === undefined) {
914 const events = yield* eventStore.read({ threadId, afterSequence }).pipe(Stream.runCollect);
915 assert.isFalse(
916 events.some(
917 ({ event }) =>
918 (event.type === "subagent.updated" ||
919 event.type === "node.updated" ||
920 event.type === "turn-item.updated") &&
921 ["completed", "failed", "cancelled", "interrupted"].includes(event.payload.status),
922 ),
923 );
924 assert.equal(cancelled.status, "running");
925 assert.equal(
926 (yield* orchestrator.getThreadProjection(childRoot.threadId)).nodes.find(
927 (node) => node.id === childRoot.id,
928 )?.status,
929 "running",
930 );
931 assert.deepEqual(yield* threadState(threadId), stateBefore);
932 return;
933 }
934
935 // Cached clients resume from this sequence, so all lifecycle repairs
936 // must be stored in the parent stream, not just applied to projections.
937 const repaired = yield* eventStore.read({ threadId, afterSequence }).pipe(Stream.runCollect);
938 const terminal = repaired.flatMap(({ event }) =>
939 (event.type === "subagent.updated" ||
940 event.type === "node.updated" ||
941 event.type === "turn-item.updated") &&
942 event.payload.status === confirmedStatus
943 ? [event]
944 : [],
945 );
946 assert.deepEqual(terminal.map((event) => event.type).sort(), [
947 "node.updated",
948 "subagent.updated",
949 "turn-item.updated",
950 ]);
951 for (const event of terminal) {
952 assert.equal(event.nodeId, subagent.id);
953 assert.equal(
954 DateTime.toEpochMillis(event.payload.completedAt!),
955 DateTime.toEpochMillis(confirmedOutcome!.completedAt),
956 );
957 if (event.type === "subagent.updated" || event.type === "turn-item.updated") {
958 assert.isDefined(event.payload.updatedAt);
959 }
960 }
961
962 assert.equal(cancelled.status, confirmedStatus);
963 assert.isNotNull(cancelled.completedAt);
964 assert.equal(
965 DateTime.toEpochMillis(cancelled.completedAt!),
966 DateTime.toEpochMillis(confirmedOutcome!.completedAt),
967 );
968 assert.equal(cancelled.result, confirmedOutcome!.result);
969 assert.equal(
970 settled.turnItems.find((candidate) => candidate.id === item.id)?.status,
971 confirmedStatus,
972 );
973 const child = yield* orchestrator.getThreadProjection(childRoot.threadId);
974 assert.equal(child.nodes.find((node) => node.id === childRoot.id)?.status, confirmedStatus);
975 assert.isNotNull(child.nodes.find((node) => node.id === childRoot.id)?.completedAt);
976 assert.equal(settled.subagents.find((task) => task.id === sibling.id)?.status, "running");
977 assert.deepEqual(yield* threadState(threadId), stateBefore);
978 // An old Stop follow-up cannot cancel a genuine subsequent resume.
979 yield* TestClock.adjust("1 second");
980 const resumedAt = yield* DateTime.now;
981 yield* projections.apply({
982 id: EventId.make("event:native-child-stop:resume"),
983 type: "subagent.updated",
984 threadId,
985 runId: run.id,
986 nodeId: subagent.id,
987 occurredAt: resumedAt,
988 payload: { ...cancelled, status: "running", startedAt: resumedAt, completedAt: null },
989 });
990 yield* orchestrator.dispatch({
991 type: "thread.background-work.settle",
992 commandId: CommandId.make("old-stop-after-resume"),
993 threadId,
994 providerThreadId: providerThread.id,
995 providerTurnId,
996 subagent: { id: subagent.id, stoppedAt: now, outcome: confirmedOutcome },
997 });
998 assert.equal(
999 (yield* orchestrator.getThreadProjection(threadId)).subagents.find(
1000 (task) => task.id === subagent.id,
1001 )?.status,
1002 "running",
1003 );
1004 }).pipe(Effect.provide(layerTest)),
1005);