1import { assert, it } from "@effect/vitest";
2import {
3 CommandId,
4 MessageId,
5 EventId,
6 NodeId,
7 type OrchestrationV2DomainEvent,
8 PlanId,
9 ProjectId,
10 ProviderDriverKind,
11 ProviderInstanceId,
12 ProviderSessionId,
13 ProviderThreadId,
14 ProviderTurnId,
15 RunAttemptId,
16 RunId,
17 RuntimeRequestId,
18 ThreadId,
19 TurnItemId,
20} from "@t3tools/contracts";
21import * as DateTime from "effect/DateTime";
22import * as Effect from "effect/Effect";
23import * as Layer from "effect/Layer";
24import * as SqlClient from "effect/sql/SqlClient";
25import * as SqlitePersistence from "../persistence/Sqlite.ts";
26import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
27import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
28import * as DelegatedTaskQuestionReactor from "./DelegatedTaskQuestionReactor.ts";
29import * as EventSink from "./EventSink.ts";
30import * as Orchestrator from "./Orchestrator.ts";
31import * as ProjectionStore from "./ProjectionStore.ts";
32import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
33import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
34import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
35
36const instanceId = ProviderInstanceId.make("codex");
37const modelSelection = { instanceId, model: "gpt-5.1-codex" };
38const adapter = {
39 instanceId,
40 driver: ProviderDriverKind.make("codex"),
41 getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
42 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }),
43 openSession: () => Effect.die("No provider process needed for metadata controls"),
44} as ProviderAdapter.ProviderAdapterV2["Service"];
45const layerDatabase = SqlitePersistence.layerMemory;
46const layerTest = Layer.mergeAll(
47 layerDatabase,
48 OrchestrationEventStore.layer.pipe(Layer.provide(layerDatabase)),
49 ProjectionStore.layer.pipe(Layer.provide(layerDatabase)),
50 ProviderReplayHarness.layerWithRegistry(
51 { name: "control-reads" },
52 ProviderAdapterRegistry.layerFromAdapters([adapter]),
53 { databaseLayer: layerDatabase, runEffectWorker: false },
54 ),
55);
56
57it.effect("cancels only the automatic settlement that observed newly opened resources", () =>
58 Effect.gen(function* () {
59 const orchestrator = yield* Orchestrator.OrchestratorV2;
60 const projections = yield* ProjectionStore.ProjectionStoreV2;
61 for (const action of ["none", "manual-settle", "revive"] as const) {
62 const threadId = ThreadId.make(`thread:auto-settle-resources:${action}`);
63 yield* orchestrator.dispatch({
64 type: "thread.create",
65 commandId: CommandId.make(`create-auto-settle-resources:${action}`),
66 threadId,
67 projectId: ProjectId.make("project:auto-settle-resources"),
68 title: "Resource settlement race",
69 modelSelection,
70 runtimeMode: "full-access",
71 interactionMode: "default",
72 branch: null,
73 worktreePath: null,
74 createdBy: "user",
75 creationSource: "web",
76 });
77 const before = yield* projections.getThread(threadId);
78 const automatic = yield* orchestrator.dispatch({
79 type: "thread.auto-settle",
80 commandId: CommandId.make(`server:auto-settle:${threadId}:1`),
81 threadId,
82 snapshotAt: before.updatedAt,
83 });
84 const settledAt = (yield* projections.getThread(threadId)).settledAt;
85 if (action !== "none") {
86 yield* orchestrator.dispatch(
87 action === "manual-settle"
88 ? {
89 type: "thread.settle",
90 commandId: CommandId.make(`manual-settle-resources:${action}`),
91 threadId,
92 }
93 : {
94 type: "thread.unsettle",
95 commandId: CommandId.make(`revive-resources:${action}`),
96 threadId,
97 reason: "user",
98 },
99 );
100 }
101 const result = yield* orchestrator
102 .dispatch({
103 type: "thread.unsettle",
104 commandId: CommandId.make(`server:auto-settle-resources:${threadId}:1`),
105 threadId,
106 reason: "user",
107 snapshotSequence: automatic.sequence,
108 })
109 .pipe(Effect.exit);
110
111 assert.equal(result._tag, action === "none" ? "Success" : "Failure");
112 const after = yield* projections.getThread(threadId);
113 assert.equal(after.settledOverride, action === "manual-settle" ? "settled" : "active");
114 if (action === "manual-settle") assert.deepStrictEqual(after.settledAt, settledAt);
115 }
116 }).pipe(Effect.provide(layerTest)),
117);
118
119it.effect(
120 "dispatches metadata, queue resume and request controls without hydrating unrelated history",
121 () =>
122 Effect.gen(function* () {
123 const orchestrator = yield* Orchestrator.OrchestratorV2;
124 const projections = yield* ProjectionStore.ProjectionStoreV2;
125 const sql = yield* SqlClient.SqlClient;
126 const threadId = ThreadId.make("thread:control-dispatch");
127 const now = yield* DateTime.now;
128 yield* orchestrator.dispatch({
129 type: "thread.create",
130 commandId: CommandId.make("create-control"),
131 threadId,
132 projectId: ProjectId.make("project:control-dispatch"),
133 title: "Before",
134 modelSelection,
135 runtimeMode: "full-access",
136 interactionMode: "default",
137 branch: null,
138 worktreePath: null,
139 createdBy: "user",
140 creationSource: "web",
141 });
142 for (const enabled of [false, true]) {
143 yield* orchestrator.dispatch({
144 type: "thread.auto-settle.set",
145 commandId: CommandId.make(`auto-settle-${enabled}`),
146 threadId,
147 enabled,
148 });
149 const updated = yield* projections.getThreadProjection(threadId);
150 assert.equal(updated.thread.autoSettleDisabledAt == null, enabled);
151 const shell = yield* projections.getThreadShell(threadId);
152 assert.ok(shell);
153 assert.equal(shell.autoSettleDisabledAt == null, enabled);
154 }
155 yield* sql`INSERT INTO orchestration_v2_projection_messages
156 (message_id, thread_id, run_id, node_id, role, streaming, created_at, updated_at, payload_json)
157 VALUES ('obsolete', ${threadId}, NULL, NULL, 'assistant', 0, ${DateTime.formatIso(now)}, ${DateTime.formatIso(now)}, '{"obsolete":true}')`;
158 assert.equal((yield* Effect.exit(projections.getThreadProjection(threadId)))._tag, "Failure");
159 yield* orchestrator.dispatch({
160 type: "queue.resume",
161 commandId: CommandId.make("resume-empty-queue"),
162 threadId,
163 });
164 yield* orchestrator.dispatch({
165 type: "thread.metadata.update",
166 commandId: CommandId.make("rename-control"),
167 threadId,
168 title: "After",
169 });
170 yield* orchestrator.dispatch({
171 type: "thread.runtime-mode.set",
172 commandId: CommandId.make("mode-control"),
173 threadId,
174 runtimeMode: "approval-required",
175 });
176 yield* orchestrator.dispatch({
177 type: "thread.model-selection.set",
178 commandId: CommandId.make("model-control"),
179 threadId,
180 modelSelection: { ...modelSelection, model: "gpt-6" },
181 });
182 const sessionId = ProviderSessionId.make("session:control-dispatch");
183 yield* projections.apply({
184 id: EventId.make("attach-control"),
185 type: "provider-session.attached",
186 threadId,
187 occurredAt: now,
188 payload: {
189 id: sessionId,
190 driver: adapter.driver,
191 providerInstanceId: instanceId,
192 status: "ready",
193 cwd: "/repo",
194 model: "gpt-6",
195 capabilities: CodexProviderCapabilitiesV2,
196 createdAt: now,
197 updatedAt: now,
198 lastError: null,
199 },
200 });
201 for (const mode of ["live", "message"] as const) {
202 const requestId = RuntimeRequestId.make(`request:${mode}`);
203 yield* projections.apply({
204 id: EventId.make(`request:${mode}`),
205 type: "runtime-request.updated",
206 threadId,
207 occurredAt: now,
208 payload: {
209 id: requestId,
210 nodeId: NodeId.make(`node:${mode}`),
211 providerTurnId: null,
212 nativeRequestRef: null,
213 kind: "user_input",
214 status: "pending",
215 responseCapability:
216 mode === "live"
217 ? { type: "live", providerSessionId: sessionId }
218 : { type: "message" },
219 createdAt: now,
220 resolvedAt: null,
221 },
222 });
223 const nodeId = NodeId.make(`node:${mode}`);
224 yield* projections.apply({
225 id: EventId.make(`node:${mode}`),
226 type: "node.updated",
227 threadId,
228 occurredAt: now,
229 payload: {
230 id: nodeId,
231 threadId,
232 runId: null,
233 parentNodeId: null,
234 rootNodeId: nodeId,
235 kind: "user_input_request",
236 status: "waiting",
237 countsForRun: false,
238 providerThreadId: null,
239 providerTurnId: null,
240 nativeItemRef: null,
241 runtimeRequestId: requestId,
242 checkpointScopeId: null,
243 startedAt: now,
244 completedAt: null,
245 },
246 });
247 yield* projections.apply({
248 id: EventId.make(`item:${mode}`),
249 type: "turn-item.updated",
250 threadId,
251 occurredAt: now,
252 payload: {
253 id: TurnItemId.make(`item:${mode}`),
254 threadId,
255 runId: null,
256 nodeId,
257 providerThreadId: null,
258 providerTurnId: null,
259 nativeItemRef: null,
260 parentItemId: null,
261 ordinal: mode === "live" ? 1 : 2,
262 status: "waiting",
263 title: null,
264 startedAt: now,
265 completedAt: null,
266 updatedAt: now,
267 type: "user_input_request",
268 requestId,
269 questions: [],
270 },
271 });
272 yield* orchestrator.dispatch(
273 mode === "live"
274 ? {
275 type: "runtime-request.respond",
276 commandId: CommandId.make(`respond:${mode}`),
277 threadId,
278 requestId,
279 decision: "accept",
280 }
281 : {
282 type: "thread.user-input.dismiss",
283 commandId: CommandId.make(`respond:${mode}`),
284 threadId,
285 requestId,
286 },
287 );
288 assert.equal(
289 (yield* projections.getRuntimeRequest(threadId, requestId))?.status,
290 "resolved",
291 );
292 const response = yield* projections.getRuntimeResponseContext(threadId, requestId);
293 assert.equal(response.node?.status, mode === "live" ? "completed" : "cancelled");
294 assert.equal(response.item?.status, mode === "live" ? "completed" : "cancelled");
295 }
296 yield* orchestrator.dispatch({
297 type: "thread.metadata.update",
298 commandId: CommandId.make("workspace-control"),
299 threadId,
300 worktreePath: "/new-repo",
301 });
302 const thread = yield* projections.getThread(threadId);
303 assert.equal(thread.title, "After");
304 assert.equal(thread.modelSelection.model, "gpt-6");
305 assert.equal(thread.runtimeMode, "approval-required");
306 assert.deepEqual(
307 (yield* projections.getThreadProviderContext(threadId)).providerSessions,
308 [],
309 );
310 yield* sql`INSERT INTO orchestration_v2_projection_turn_items
311 (turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
312 type, status, ordinal, updated_at, payload_json)
313 VALUES ('obsolete-output', ${threadId}, NULL, NULL, NULL, NULL,
314 'command_execution', 'completed', 900, ${DateTime.formatIso(now)}, '{"obsolete":true}')`;
315 yield* sql`INSERT INTO orchestration_v2_projection_plans
316 (plan_id, thread_id, run_id, node_id, kind, status, payload_json)
317 VALUES ('obsolete-plan', ${threadId}, NULL, 'old-node', 'proposed', 'completed', '{"obsolete":true}')`;
318 yield* sql`INSERT INTO orchestration_v2_projection_context_handoffs
319 (context_handoff_id, thread_id, target_run_id, to_provider_thread_id, strategy, status, updated_at, payload_json)
320 VALUES ('obsolete-handoff', ${threadId}, 'old-run', 'old-provider-thread', 'full_thread_summary', 'ready', ${DateTime.formatIso(now)}, '{"obsolete":true}')`;
321 yield* orchestrator.dispatch({
322 type: "message.dispatch",
323 commandId: CommandId.make("dispatch-with-old-history"),
324 threadId,
325 messageId: MessageId.make("fresh-input"),
326 text: "Continue",
327 attachments: [],
328 dispatchMode: { type: "defer_start" },
329 createdBy: "user",
330 creationSource: "web",
331 });
332 const fresh = yield* projections.getThreadRecords(threadId, ["turnItems"], {
333 turnItemTypes: ["user_message"],
334 });
335 assert.isAbove(fresh.turnItems.at(-1)!.ordinal, 900);
336 yield* orchestrator.dispatch({
337 type: "thread.archive",
338 commandId: CommandId.make("archive-with-old-history"),
339 threadId,
340 });
341 yield* orchestrator.dispatch({
342 type: "thread.delete",
343 commandId: CommandId.make("delete-with-old-history"),
344 threadId,
345 });
346 assert.isNotNull((yield* projections.getThread(threadId)).deletedAt);
347 }).pipe(Effect.provide(layerTest)),
348);
349
350it.effect("implements a proposed plan that the command projection leaves out", () =>
351 Effect.gen(function* () {
352 const orchestrator = yield* Orchestrator.OrchestratorV2;
353 const projections = yield* ProjectionStore.ProjectionStoreV2;
354 const threadId = ThreadId.make("thread:implement-plan");
355 const planId = PlanId.make("plan:implement-plan");
356 const now = yield* DateTime.now;
357 yield* orchestrator.dispatch({
358 type: "thread.create",
359 commandId: CommandId.make("create-implement-plan"),
360 threadId,
361 projectId: ProjectId.make("project:implement-plan"),
362 title: "Plan",
363 modelSelection,
364 runtimeMode: "full-access",
365 interactionMode: "plan",
366 branch: null,
367 worktreePath: null,
368 createdBy: "user",
369 creationSource: "web",
370 });
371 yield* projections.apply({
372 id: EventId.make("plan:implement-plan"),
373 type: "plan.updated",
374 threadId,
375 occurredAt: now,
376 payload: {
377 id: planId,
378 threadId,
379 runId: null,
380 nodeId: NodeId.make("node:implement-plan"),
381 kind: "proposed_plan",
382 status: "active",
383 markdown: "# Plan\n\n1. Do the thing.",
384 },
385 });
386
387 yield* orchestrator.dispatch({
388 type: "message.dispatch",
389 commandId: CommandId.make("implement-plan"),
390 threadId,
391 messageId: MessageId.make("implement-plan-input"),
392 text: "Implement the plan.",
393 attachments: [],
394 sourcePlanRef: { threadId, planId },
395 dispatchMode: { type: "defer_start" },
396 createdBy: "user",
397 creationSource: "web",
398 });
399
400 assert.equal((yield* projections.getPlan(threadId, planId))?.status, "completed");
401 }).pipe(Effect.provide(layerTest)),
402);
403
404// Stop's settle follow-up runs after the provider interrupt returns, possibly
405// long after the Stop (retries) or again (an effect replayed after a crash).
406// A later run's background work is not that Stop's to end.
407it.effect("settles only the stopped run's background work, once", () =>
408 Effect.gen(function* () {
409 const orchestrator = yield* Orchestrator.OrchestratorV2;
410 const projections = yield* ProjectionStore.ProjectionStoreV2;
411 const threadId = ThreadId.make("thread:settle-binding");
412 const providerThreadId = ProviderThreadId.make("provider-thread:settle-binding");
413 const now = yield* DateTime.now;
414 yield* orchestrator.dispatch({
415 type: "thread.create",
416 commandId: CommandId.make("create-settle-binding"),
417 threadId,
418 projectId: ProjectId.make("project:settle-binding"),
419 title: "Settle binding",
420 modelSelection,
421 runtimeMode: "full-access",
422 interactionMode: "default",
423 branch: null,
424 worktreePath: null,
425 createdBy: "user",
426 creationSource: "web",
427 });
428 yield* projections.apply({
429 id: EventId.make("settle-binding:provider-thread"),
430 type: "provider-thread.updated",
431 threadId,
432 occurredAt: now,
433 payload: {
434 id: providerThreadId,
435 driver: adapter.driver,
436 providerInstanceId: instanceId,
437 providerSessionId: null,
438 appThreadId: threadId,
439 ownerNodeId: null,
440 nativeThreadRef: null,
441 nativeConversationHeadRef: null,
442 status: "idle",
443 firstRunOrdinal: 1,
444 lastRunOrdinal: 2,
445 handoffIds: [],
446 forkedFrom: null,
447 createdAt: now,
448 updatedAt: now,
449 },
450 });
451 const commandItem = (ordinal: number) => TurnItemId.make(`turn-item:settle-binding:${ordinal}`);
452 for (const ordinal of [1, 2]) {
453 const runId = RunId.make(`run:settle-binding:${ordinal}`);
454 const attemptId = RunAttemptId.make(`attempt:settle-binding:${ordinal}`);
455 const nodeId = NodeId.make(`node:settle-binding:${ordinal}`);
456 const providerTurnId = ProviderTurnId.make(`provider-turn:settle-binding:${ordinal}`);
457 yield* projections.apply({
458 id: EventId.make(`settle-binding:run:${ordinal}`),
459 type: "run.created",
460 threadId,
461 occurredAt: now,
462 payload: {
463 id: runId,
464 threadId,
465 ordinal,
466 providerInstanceId: instanceId,
467 modelSelection,
468 providerThreadId,
469 userMessageId: MessageId.make(`message:settle-binding:${ordinal}`),
470 rootNodeId: nodeId,
471 activeAttemptId: attemptId,
472 status: "completed",
473 requestedAt: now,
474 startedAt: now,
475 completedAt: now,
476 checkpointId: null,
477 contextHandoffId: null,
478 },
479 });
480 yield* projections.apply({
481 id: EventId.make(`settle-binding:attempt:${ordinal}`),
482 type: "run-attempt.created",
483 threadId,
484 occurredAt: now,
485 payload: {
486 id: attemptId,
487 runId,
488 attemptOrdinal: 1,
489 rootNodeId: nodeId,
490 providerInstanceId: instanceId,
491 providerThreadId,
492 providerTurnId,
493 reason: "initial",
494 status: "completed",
495 startedAt: now,
496 completedAt: now,
497 },
498 });
499 yield* projections.apply({
500 id: EventId.make(`settle-binding:turn:${ordinal}`),
501 type: "provider-turn.updated",
502 threadId,
503 occurredAt: now,
504 payload: {
505 id: providerTurnId,
506 providerThreadId,
507 nodeId,
508 runAttemptId: attemptId,
509 nativeTurnRef: null,
510 ordinal,
511 status: "completed",
512 startedAt: now,
513 completedAt: now,
514 },
515 });
516 yield* projections.apply({
517 id: EventId.make(`settle-binding:item:${ordinal}`),
518 type: "turn-item.updated",
519 threadId,
520 runId,
521 occurredAt: now,
522 payload: {
523 id: commandItem(ordinal),
524 threadId,
525 runId,
526 nodeId,
527 providerThreadId,
528 providerTurnId,
529 nativeItemRef: null,
530 parentItemId: null,
531 ordinal: ordinal * 10,
532 status: "running",
533 title: `Background command ${ordinal}`,
534 startedAt: now,
535 completedAt: null,
536 updatedAt: now,
537 type: "command_execution",
538 input: `sleep ${ordinal}`,
539 },
540 });
541 }
542 const itemStatuses = Effect.map(projections.getThreadProjection(threadId), (projection) =>
543 projection.turnItems
544 .flatMap((item) => (item.type === "command_execution" ? [`${item.id}:${item.status}`] : []))
545 .toSorted(),
546 );
547 // The settle that followed a Stop of run 1's turn, dispatched only after
548 // run 2 had settled with work of its own.
549 const settle = {
550 type: "thread.background-work.settle",
551 commandId: CommandId.make("stop-run-1:background-work-settled"),
552 threadId,
553 providerThreadId,
554 providerTurnId: ProviderTurnId.make("provider-turn:settle-binding:1"),
555 } as const;
556 yield* orchestrator.dispatch(settle);
557 assert.deepEqual(yield* itemStatuses, [
558 `${commandItem(1)}:interrupted`,
559 `${commandItem(2)}:running`,
560 ]);
561
562 // A settle that found nothing to end replays as a no-op, even after work
563 // it would match appears: its receipt is recorded with no events.
564 const emptySettle = {
565 ...settle,
566 commandId: CommandId.make("stop-run-1-again:background-work-settled"),
567 };
568 const first = yield* orchestrator.dispatch(emptySettle);
569 assert.lengthOf(first.storedEvents, 0);
570 yield* projections.apply({
571 id: EventId.make("settle-binding:item:late"),
572 type: "turn-item.updated",
573 threadId,
574 runId: RunId.make("run:settle-binding:1"),
575 occurredAt: now,
576 payload: {
577 id: commandItem(3),
578 threadId,
579 runId: RunId.make("run:settle-binding:1"),
580 nodeId: NodeId.make("node:settle-binding:1"),
581 providerThreadId,
582 providerTurnId: ProviderTurnId.make("provider-turn:settle-binding:1"),
583 nativeItemRef: null,
584 parentItemId: null,
585 ordinal: 30,
586 status: "running",
587 title: "Late background command",
588 startedAt: now,
589 completedAt: null,
590 updatedAt: now,
591 type: "command_execution",
592 input: "sleep 3",
593 },
594 });
595 const replayed = yield* orchestrator.dispatch(emptySettle);
596 assert.lengthOf(replayed.storedEvents, 0);
597 assert.deepEqual(yield* itemStatuses, [
598 `${commandItem(1)}:interrupted`,
599 `${commandItem(2)}:running`,
600 `${commandItem(3)}:running`,
601 ]);
602 }).pipe(Effect.provide(layerTest)),
603);
604
605it.effect("keeps delegated child pull-request links independent of the parent", () =>
606 Effect.gen(function* () {
607 const orchestrator = yield* Orchestrator.OrchestratorV2;
608 const projections = yield* ProjectionStore.ProjectionStoreV2;
609 const parentThreadId = ThreadId.make("thread:parent-pr");
610 const projectId = ProjectId.make("project:parent-pr");
611 const parentPullRequest = {
612 projectId,
613 repository: "pingdotgg/t3code",
614 number: 123,
615 url: "https://github.com/pingdotgg/t3code/pull/123",
616 };
617 yield* orchestrator.dispatch({
618 type: "thread.create",
619 commandId: CommandId.make("create-parent-pr"),
620 threadId: parentThreadId,
621 projectId,
622 title: "Parent with a linked PR",
623 modelSelection,
624 runtimeMode: "full-access",
625 interactionMode: "default",
626 branch: "feature/parent-pr",
627 worktreePath: "/repo-worktree",
628 createdBy: "user",
629 creationSource: "web",
630 });
631 yield* orchestrator.dispatch({
632 type: "thread.metadata.update",
633 commandId: CommandId.make("link-parent-pr"),
634 threadId: parentThreadId,
635 linkedPullRequest: parentPullRequest,
636 });
637 yield* orchestrator.dispatch({
638 type: "message.dispatch",
639 commandId: CommandId.make("start-parent-pr"),
640 threadId: parentThreadId,
641 messageId: MessageId.make("message:parent-pr"),
642 text: "Delegate a review",
643 attachments: [],
644 dispatchMode: { type: "start_immediately" },
645 createdBy: "user",
646 creationSource: "web",
647 });
648 const parent = yield* projections.getThreadProjection(parentThreadId);
649 const parentRun = parent.runs[0]!;
650 yield* orchestrator.dispatch({
651 type: "delegated_task.request",
652 commandId: CommandId.make("delegate-parent-pr"),
653 parentThreadId,
654 parentRunId: parentRun.id,
655 parentNodeId: parentRun.rootNodeId!,
656 task: "Review the changes",
657 modelSelection,
658 runtimeMode: "full-access",
659 interactionMode: "default",
660 createdBy: "agent",
661 creationSource: "mcp",
662 });
663 const updatedParent = yield* projections.getThreadProjection(parentThreadId);
664 const childThreadId = updatedParent.subagents[0]!.childThreadId!;
665 const child = yield* projections.getThreadProjection(childThreadId);
666 assert.isNull(child.thread.linkedPullRequest);
667 assert.deepEqual(child.thread.pullRequests, []);
668 assert.equal(child.thread.branch, parent.thread.branch);
669 assert.equal(child.thread.worktreePath, parent.thread.worktreePath);
670 assert.equal(child.thread.lineage.parentThreadId, parentThreadId);
671
672 const childPullRequest = {
673 ...parentPullRequest,
674 number: 456,
675 url: "https://github.com/pingdotgg/t3code/pull/456",
676 };
677 yield* orchestrator.dispatch({
678 type: "thread.metadata.update",
679 commandId: CommandId.make("link-child-pr"),
680 threadId: childThreadId,
681 linkedPullRequest: childPullRequest,
682 });
683 const linkedChild = yield* projections.getThreadProjection(childThreadId);
684 assert.deepEqual(linkedChild.thread.linkedPullRequest, childPullRequest);
685 assert.deepEqual(
686 linkedChild.thread.pullRequests?.map((link) => link.number),
687 [456],
688 );
689 const parentAfterChildLink = yield* projections.getThreadProjection(parentThreadId);
690 assert.deepEqual(parentAfterChildLink.thread.linkedPullRequest, parentPullRequest);
691 assert.deepEqual(parentAfterChildLink.thread.pullRequests, parent.thread.pullRequests);
692 }).pipe(Effect.provide(layerTest)),
693);
694
695it.effect("wakes the parent once when its delegated task asks a question", () =>
696 Effect.gen(function* () {
697 const orchestrator = yield* Orchestrator.OrchestratorV2;
698 const projections = yield* ProjectionStore.ProjectionStoreV2;
699 const reactor = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor;
700 const parentThreadId = ThreadId.make("thread:question-parent");
701 yield* orchestrator.dispatch({
702 type: "thread.create",
703 commandId: CommandId.make("create-question-parent"),
704 threadId: parentThreadId,
705 projectId: ProjectId.make("project:question"),
706 title: "Parent",
707 modelSelection,
708 runtimeMode: "full-access",
709 interactionMode: "default",
710 branch: null,
711 worktreePath: null,
712 createdBy: "user",
713 creationSource: "web",
714 });
715 yield* orchestrator.dispatch({
716 type: "message.dispatch",
717 commandId: CommandId.make("start-question-parent"),
718 threadId: parentThreadId,
719 messageId: MessageId.make("message:question-parent"),
720 text: "Delegate the research",
721 attachments: [],
722 dispatchMode: { type: "start_immediately" },
723 createdBy: "user",
724 creationSource: "web",
725 });
726 const parentRun = (yield* projections.getThreadProjection(parentThreadId)).runs[0]!;
727 yield* orchestrator.dispatch({
728 type: "delegated_task.request",
729 commandId: CommandId.make("delegate-question-task"),
730 parentThreadId,
731 parentRunId: parentRun.id,
732 parentNodeId: parentRun.rootNodeId!,
733 task: "Research the sources",
734 completionWake: "settled_only",
735 modelSelection,
736 runtimeMode: "full-access",
737 interactionMode: "default",
738 createdBy: "agent",
739 creationSource: "mcp",
740 });
741 const task = (yield* projections.getThreadProjection(parentThreadId)).subagents[0]!;
742 const childRun = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs[0]!;
743 const now = yield* DateTime.now;
744 const requestEvent = (
745 eventId: string,
746 requestId: string,
747 responseCapability: "live" | "message" = "message",
748 kind: "user_input" | "command" = "user_input",
749 run = childRun,
750 ) =>
751 ({
752 id: EventId.make(eventId),
753 type: "runtime-request.updated",
754 threadId: task.childThreadId!,
755 runId: run.id,
756 occurredAt: now,
757 payload: {
758 id: RuntimeRequestId.make(requestId),
759 nodeId: run.rootNodeId!,
760 providerTurnId: null,
761 nativeRequestRef: null,
762 kind,
763 status: "pending",
764 responseCapability:
765 responseCapability === "live"
766 ? { type: "live", providerSessionId: ProviderSessionId.make("session:question") }
767 : { type: "message" },
768 createdAt: now,
769 resolvedAt: null,
770 },
771 }) satisfies OrchestrationV2DomainEvent;
772 const question = requestEvent("event:question", "request:question", "live");
773 const asyncQuestion = requestEvent("event:async-question", "request:async-question");
774 const answered = requestEvent("event:answered", "request:answered", "live");
775 const approval = requestEvent("event:approval", "request:approval", "message", "command");
776 const eventSink = yield* EventSink.EventSinkV2;
777 yield* eventSink.write({
778 events: [question, asyncQuestion, answered, approval],
779 });
780
781 yield* reactor.handle(question);
782 assert.isFalse(
783 (yield* projections.getThreadProjection(parentThreadId)).messages.some(
784 (message) => message.notification !== undefined,
785 ),
786 );
787 yield* reactor.handle(asyncQuestion);
788 assert.equal(
789 (yield* projections.getThreadProjection(parentThreadId)).messages.filter(
790 (message) => message.notification !== undefined,
791 ).length,
792 1,
793 );
794 yield* reactor.handle(asyncQuestion);
795 yield* reactor.handle(answered);
796 yield* eventSink.write({
797 events: [
798 {
799 ...answered,
800 id: EventId.make("event:answered-resolved"),
801 payload: { ...answered.payload, status: "resolved", resolvedAt: now },
802 },
803 ],
804 });
805 yield* reactor.handle(answered);
806 yield* reactor.handle(approval);
807
808 const wakes = (yield* projections.getThreadProjection(parentThreadId)).messages.filter(
809 (message) => message.notification !== undefined,
810 );
811 assert.include(wakes[0]?.text, "asked a question and is still working");
812 assert.deepEqual(
813 wakes.map((message) => message.notification),
814 [
815 {
816 source: {
817 kind: "delegated_task",
818 taskIds: [task.id],
819 childThreadId: task.childThreadId!,
820 },
821 outcome: "updated",
822 summary: 'Delegated task "Research the sources" has a question',
823 },
824 ],
825 );
826
827 const wakeRun = (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.find(
828 (run) => run.userMessageId === wakes[0]!.id,
829 )!;
830 assert.equal(wakeRun.status, "queued");
831 const resolved = {
832 ...asyncQuestion,
833 id: EventId.make("event:question-resolved"),
834 payload: { ...asyncQuestion.payload, status: "resolved" as const, resolvedAt: now },
835 };
836 yield* eventSink.write({ events: [resolved] });
837 yield* reactor.handle(resolved);
838 assert.equal(
839 (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.find(
840 (run) => run.id === wakeRun.id,
841 )?.status,
842 "cancelled",
843 );
844 const parentCompleted = {
845 id: EventId.make("event:question-parent-completed"),
846 type: "run.updated",
847 threadId: parentThreadId,
848 runId: parentRun.id,
849 occurredAt: now,
850 payload: { ...parentRun, status: "completed", completedAt: now },
851 } satisfies OrchestrationV2DomainEvent;
852 for (const status of [
853 "preparing",
854 "starting",
855 "running",
856 "queued",
857 "cancelled",
858 "rolled_back",
859 "interrupted",
860 ] as const) {
861 yield* reactor.handle({ ...parentCompleted, payload: { ...parentRun, status } });
862 }
863 assert.equal(
864 (yield* projections.getThreadRecords(parentThreadId, ["messages"])).messages.length,
865 2,
866 );
867 yield* eventSink.write({ events: [parentCompleted] });
868 for (const state of ["disposed", "pending"] as const) {
869 yield* eventSink.write({
870 events: [
871 {
872 id: EventId.make(`event:question-task-${state}`),
873 type: "subagent.updated",
874 threadId: parentThreadId,
875 occurredAt: now,
876 payload: { ...task, completionDelivery: { state, observedByRunId: null } },
877 },
878 ],
879 });
880 yield* reactor.handle(parentCompleted);
881 if (state === "disposed") {
882 assert.equal(
883 (yield* projections.getThreadRecords(parentThreadId, ["messages"])).messages.length,
884 2,
885 );
886 }
887 }
888 yield* reactor.handle(parentCompleted);
889 const caughtUp = (yield* projections.getThreadProjection(parentThreadId)).messages.filter(
890 (message) => message.notification !== undefined,
891 );
892 assert.equal(caughtUp.length, 2);
893 assert.equal(
894 caughtUp.filter((message) => message.text.includes(`requestId ${question.payload.id}.`))
895 .length,
896 1,
897 );
898 assert.include(caughtUp[1]?.text, "is blocked on a question");
899 assert.isFalse(
900 caughtUp.some((message) => message.text.includes(`requestId ${answered.payload.id}.`)),
901 );
902 yield* reactor.handle(question);
903 assert.equal(
904 (yield* projections.getThreadProjection(parentThreadId)).messages.filter(
905 (message) => message.notification !== undefined,
906 ).length,
907 2,
908 );
909 const terminalQuestion = requestEvent(
910 "event:terminal-question",
911 "request:terminal-question",
912 "live",
913 );
914 yield* eventSink.write({ events: [terminalQuestion] });
915 yield* reactor.handle(terminalQuestion);
916 assert.equal(
917 (yield* projections.getThreadProjection(parentThreadId)).messages.filter(
918 (message) => message.notification !== undefined,
919 ).length,
920 3,
921 );
922 yield* eventSink.write({
923 events: [
924 {
925 id: EventId.make("event:question-parent-running-again"),
926 type: "run.updated",
927 threadId: parentThreadId,
928 runId: parentRun.id,
929 occurredAt: now,
930 payload: { ...parentRun, status: "running", completedAt: null },
931 },
932 ],
933 });
934 yield* orchestrator.dispatch({
935 type: "delegated_task.wake-policy",
936 commandId: CommandId.make("terminal-question-wait-ended"),
937 parentThreadId,
938 taskId: task.id,
939 completionWake: "always",
940 });
941 const upgradedQuestion = requestEvent(
942 "event:upgraded-question",
943 "request:upgraded-question",
944 "live",
945 );
946 yield* eventSink.write({ events: [upgradedQuestion] });
947 yield* reactor.handle(upgradedQuestion);
948 assert.equal(
949 (yield* projections.getThreadProjection(parentThreadId)).messages.filter(
950 (message) => message.notification !== undefined,
951 ).length,
952 4,
953 );
954 const waitDelivered = requestEvent("event:wait-delivered", "request:wait-delivered", "live");
955 const waitDeliveredParentCompleted = {
956 ...parentCompleted,
957 id: EventId.make("event:wait-delivered-parent-completed"),
958 };
959 yield* eventSink.write({ events: [waitDelivered, waitDeliveredParentCompleted] });
960 yield* reactor.handle(waitDeliveredParentCompleted);
961 assert.isFalse(
962 (yield* projections.getThreadProjection(parentThreadId)).messages.some((message) =>
963 message.text.includes(`requestId ${waitDelivered.payload.id}.`),
964 ),
965 );
966 yield* eventSink.write({
967 commandId: CommandId.make("command:runtime-reconcile:question-followup"),
968 events: [
969 {
970 id: EventId.make("event:question-child-completed"),
971 type: "run.updated",
972 threadId: task.childThreadId!,
973 occurredAt: now,
974 payload: { ...childRun, status: "completed", completedAt: now },
975 },
976 {
977 id: EventId.make("event:question-task-completed"),
978 type: "subagent.updated",
979 threadId: parentThreadId,
980 occurredAt: now,
981 payload: {
982 ...task,
983 status: "completed",
984 result: "Sources researched.",
985 completedAt: now,
986 },
987 },
988 ],
989 });
990 yield* orchestrator.dispatch({
991 type: "message.dispatch",
992 commandId: CommandId.make("question-followup"),
993 threadId: task.childThreadId!,
994 messageId: MessageId.make("message:question-followup"),
995 text: "Check the sources again",
996 attachments: [],
997 dispatchMode: { type: "start_immediately" },
998 createdBy: "user",
999 creationSource: "web",
1000 });
1001 const followup = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs.find(
1002 (run) => run.userMessageId === "message:question-followup",
1003 )!;
1004 assert.isDefined(followup.delegatedTaskId);
1005 assert.notEqual(followup.delegatedTaskId, task.id);
1006 const followupQuestion = requestEvent(
1007 "event:followup-question",
1008 "request:followup-question",
1009 "live",
1010 "user_input",
1011 followup,
1012 );
1013 yield* eventSink.write({ events: [followupQuestion] });
1014 yield* reactor.handle(followupQuestion);
1015 const followupWake = (yield* projections.getThreadRecords(parentThreadId, [
1016 "messages",
1017 ])).messages.find((message) =>
1018 message.text.includes(`requestId ${followupQuestion.payload.id}`),
1019 )!;
1020 assert.deepEqual(followupWake.notification?.source, {
1021 kind: "delegated_task",
1022 taskIds: [followup.delegatedTaskId],
1023 childThreadId: task.childThreadId!,
1024 });
1025 }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))),
1026);
1027
1028it.effect("guarded archive preserves a resumed thread and rejects an earlier settlement", () =>
1029 Effect.gen(function* () {
1030 const orchestrator = yield* Orchestrator.OrchestratorV2;
1031 const projections = yield* ProjectionStore.ProjectionStoreV2;
1032 const threadId = ThreadId.make("thread:archive-race");
1033 yield* orchestrator.dispatch({
1034 type: "thread.create",
1035 commandId: CommandId.make("create-archive-race"),
1036 threadId,
1037 projectId: ProjectId.make("project:archive-race"),
1038 title: "Work",
1039 modelSelection,
1040 runtimeMode: "full-access",
1041 interactionMode: "default",
1042 branch: null,
1043 worktreePath: null,
1044 createdBy: "user",
1045 creationSource: "web",
1046 });
1047 yield* orchestrator.dispatch({
1048 type: "thread.settle",
1049 commandId: CommandId.make("settle-archive-race"),
1050 threadId,
1051 });
1052 const { snapshotSequence } = yield* orchestrator.getShellSnapshot();
1053 const originalSettledAt = (yield* projections.getThread(threadId)).settledAt;
1054 yield* orchestrator.dispatch({
1055 type: "thread.settle",
1056 commandId: CommandId.make("duplicate-settle-archive-race"),
1057 threadId,
1058 });
1059 yield* orchestrator.dispatch({
1060 type: "thread.unsettle",
1061 commandId: CommandId.make("resume-archive-race"),
1062 threadId,
1063 reason: "user",
1064 });
1065 const resumed = yield* orchestrator
1066 .dispatch({
1067 type: "thread.archive",
1068 commandId: CommandId.make("archive-resumed"),
1069 threadId,
1070 snapshotSequence,
1071 })
1072 .pipe(Effect.flip);
1073 assert.include(resumed, {
1074 _tag: "OrchestratorDispatchError",
1075 reason: "archive-condition-changed",
1076 });
1077 assert.isNull((yield* projections.getThread(threadId)).archivedAt);
1078
1079 yield* orchestrator.dispatch({
1080 type: "thread.settle",
1081 commandId: CommandId.make("resettle-archive-race"),
1082 threadId,
1083 });
1084 assert.deepEqual((yield* projections.getThread(threadId)).settledAt, originalSettledAt);
1085 const stale = yield* orchestrator
1086 .dispatch({
1087 type: "thread.archive",
1088 commandId: CommandId.make("archive-previous-settlement"),
1089 threadId,
1090 snapshotSequence,
1091 })
1092 .pipe(Effect.flip);
1093 assert.include(stale, {
1094 _tag: "OrchestratorDispatchError",
1095 reason: "archive-condition-changed",
1096 });
1097 assert.isNull((yield* projections.getThread(threadId)).archivedAt);
1098
1099 const beforeRename = yield* orchestrator.getShellSnapshot();
1100 yield* orchestrator.dispatch({
1101 type: "thread.metadata.update",
1102 commandId: CommandId.make("rename-archive-race"),
1103 threadId,
1104 title: "Changed",
1105 });
1106 const renamed = yield* orchestrator
1107 .dispatch({
1108 type: "thread.archive",
1109 commandId: CommandId.make("archive-before-rename"),
1110 threadId,
1111 snapshotSequence: beforeRename.snapshotSequence,
1112 })
1113 .pipe(Effect.flip);
1114 assert.include(renamed, {
1115 _tag: "OrchestratorDispatchError",
1116 reason: "archive-condition-changed",
1117 });
1118 const current = yield* orchestrator.getShellSnapshot();
1119 yield* orchestrator.dispatch({
1120 type: "thread.create",
1121 commandId: CommandId.make("create-unrelated-archive-race"),
1122 threadId: ThreadId.make("thread:unrelated-archive-race"),
1123 projectId: ProjectId.make("project:archive-race"),
1124 title: "Unrelated",
1125 modelSelection,
1126 runtimeMode: "full-access",
1127 interactionMode: "default",
1128 branch: null,
1129 worktreePath: null,
1130 createdBy: "user",
1131 creationSource: "web",
1132 });
1133 yield* orchestrator.dispatch({
1134 type: "thread.archive",
1135 commandId: CommandId.make("archive-current-settlement"),
1136 threadId,
1137 snapshotSequence: current.snapshotSequence,
1138 });
1139 assert.isNotNull((yield* projections.getThread(threadId)).archivedAt);
1140 }).pipe(Effect.provide(layerTest)),
1141);
1142
1143it.effect("guarded archive retains threads with active, queued, or pending request work", () =>
1144 Effect.gen(function* () {
1145 const orchestrator = yield* Orchestrator.OrchestratorV2;
1146 const projections = yield* ProjectionStore.ProjectionStoreV2;
1147 const now = yield* DateTime.now;
1148 for (const status of [
1149 "queued",
1150 "preparing",
1151 "starting",
1152 "running",
1153 "waiting",
1154 "request",
1155 ] as const) {
1156 const threadId = ThreadId.make(`thread:archive-blocked:${status}`);
1157 yield* orchestrator.dispatch({
1158 type: "thread.create",
1159 commandId: CommandId.make(`create-archive-blocked:${status}`),
1160 threadId,
1161 projectId: ProjectId.make("project:archive-blocked"),
1162 title: "Work",
1163 modelSelection,
1164 runtimeMode: "full-access",
1165 interactionMode: "default",
1166 branch: null,
1167 worktreePath: null,
1168 createdBy: "user",
1169 creationSource: "web",
1170 });
1171 yield* orchestrator.dispatch({
1172 type: "thread.settle",
1173 commandId: CommandId.make(`settle-archive-blocked:${status}`),
1174 threadId,
1175 });
1176 if (status === "request") {
1177 yield* projections.apply({
1178 id: EventId.make("archive-blocked:request"),
1179 type: "runtime-request.updated",
1180 threadId,
1181 occurredAt: now,
1182 payload: {
1183 id: RuntimeRequestId.make("archive-blocked:request"),
1184 nodeId: NodeId.make("archive-blocked:request"),
1185 providerTurnId: null,
1186 nativeRequestRef: null,
1187 kind: "user_input",
1188 status: "pending",
1189 responseCapability: { type: "message" },
1190 createdAt: now,
1191 resolvedAt: null,
1192 },
1193 });
1194 } else {
1195 yield* projections.apply({
1196 id: EventId.make(`archive-blocked:run:${status}`),
1197 type: "run.created",
1198 threadId,
1199 occurredAt: now,
1200 payload: {
1201 id: RunId.make(`archive-blocked:run:${status}`),
1202 threadId,
1203 ordinal: 1,
1204 providerInstanceId: instanceId,
1205 modelSelection,
1206 providerThreadId: null,
1207 userMessageId: MessageId.make(`archive-blocked:message:${status}`),
1208 rootNodeId: null,
1209 activeAttemptId: null,
1210 status,
1211 requestedAt: now,
1212 startedAt: null,
1213 completedAt: null,
1214 checkpointId: null,
1215 contextHandoffId: null,
1216 },
1217 });
1218 }
1219 const snapshot = yield* orchestrator.getShellSnapshot();
1220 const exit = yield* orchestrator
1221 .dispatch({
1222 type: "thread.archive",
1223 commandId: CommandId.make(`archive-blocked:${status}`),
1224 threadId,
1225 snapshotSequence: snapshot.snapshotSequence,
1226 })
1227 .pipe(Effect.flip);
1228 assert.include(exit, {
1229 _tag: "OrchestratorDispatchError",
1230 reason: "archive-condition-changed",
1231 });
1232 assert.isNull((yield* projections.getThread(threadId)).archivedAt);
1233 }
1234 }).pipe(Effect.provide(layerTest)),
1235);