1import { assert, it } from "@effect/vitest";
2import {
3 CommandId,
4 EnvironmentId,
5 EventId,
6 MessageId,
7 NodeId,
8 ProjectId,
9 ProviderDriverKind,
10 ProviderInstanceId,
11 ProviderThreadId,
12 RunId,
13 ThreadId,
14} from "@t3tools/contracts";
15import * as Context from "effect/Context";
16import * as DateTime from "effect/DateTime";
17import * as Effect from "effect/Effect";
18import * as Layer from "effect/Layer";
19import * as Ref from "effect/Ref";
20
21import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
22import * as EventSink from "./EventSink.ts";
23import * as Orchestrator from "./Orchestrator.ts";
24import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
25import * as ProjectionStore from "./ProjectionStore.ts";
26import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
27import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
28import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
29import * as SqlitePersistence from "../persistence/Sqlite.ts";
30
31// A delegated task whose child runs in a linked environment is recorded on
32// its parent without a child thread here, and completing it wakes the parent
33// the way a local child's result does.
34
35const instanceId = ProviderInstanceId.make("codex");
36const driver = ProviderDriverKind.make("codex");
37const modelSelection = { instanceId, model: "gpt-5.4" };
38const adapter = {
39 instanceId,
40 driver,
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 delegated task records"),
44} as ProviderAdapter.ProviderAdapterV2["Service"];
45
46/** Every parent wake the orchestrator offers, for the tests to read. */
47class Offers extends Context.Service<
48 Offers,
49 Ref.Ref<ReadonlyArray<ProviderContinuationRequests.ProviderContinuationRequest>>
50>()("t3/orchestration-v2/RemoteDelegatedTask.test/Offers") {}
51const layerOffers = Layer.effect(
52 Offers,
53 Ref.make<ReadonlyArray<ProviderContinuationRequests.ProviderContinuationRequest>>([]),
54);
55const layerDatabase = SqlitePersistence.layerMemory;
56const layerTest = Layer.mergeAll(
57 layerDatabase,
58 ProjectionStore.layer.pipe(Layer.provide(layerDatabase)),
59 ProviderReplayHarness.layerWithRegistry(
60 { name: "remote-delegated-task" },
61 ProviderAdapterRegistry.layerFromAdapters([adapter]),
62 { databaseLayer: layerDatabase, runEffectWorker: false },
63 ),
64).pipe(
65 Layer.provide(
66 Layer.effect(
67 ProviderContinuationRequests.ProviderContinuationRequests,
68 Offers.pipe(
69 Effect.map((offers) => ({
70 offer: (request: ProviderContinuationRequests.ProviderContinuationRequest) =>
71 Ref.update(offers, (seen) => [...seen, request]),
72 take: Effect.never,
73 })),
74 ),
75 ),
76 ),
77 Layer.provideMerge(layerOffers),
78);
79
80/** A parent thread whose run is live, as when its agent calls delegate_task. */
81const seedLiveParent = (name: string) =>
82 Effect.gen(function* () {
83 const orchestrator = yield* Orchestrator.OrchestratorV2;
84 const sink = yield* EventSink.EventSinkV2;
85 const now = yield* DateTime.now;
86 const threadId = ThreadId.make(`thread:${name}`);
87 const runId = RunId.make(`run:${name}`);
88 const rootNodeId = NodeId.make(`node:${name}:root`);
89 const providerThreadId = ProviderThreadId.make(`provider-thread:${name}`);
90 yield* orchestrator.dispatch({
91 type: "thread.create",
92 createdBy: "user",
93 creationSource: "web",
94 commandId: CommandId.make(`command:create:${name}`),
95 threadId,
96 projectId: ProjectId.make("project:remote-delegated-task"),
97 title: "Parent",
98 modelSelection,
99 runtimeMode: "full-access",
100 interactionMode: "default",
101 branch: null,
102 worktreePath: null,
103 });
104 yield* sink.write({
105 commandId: CommandId.make(`command:seed:${name}`),
106 events: [
107 {
108 id: EventId.make(`event:seed-provider-thread:${name}`),
109 type: "provider-thread.updated",
110 threadId,
111 driver,
112 providerInstanceId: instanceId,
113 occurredAt: now,
114 payload: {
115 id: providerThreadId,
116 driver,
117 providerInstanceId: instanceId,
118 providerSessionId: null,
119 appThreadId: threadId,
120 ownerNodeId: rootNodeId,
121 nativeThreadRef: { driver, nativeId: `native:${name}`, strength: "strong" },
122 nativeConversationHeadRef: null,
123 status: "active",
124 firstRunOrdinal: 1,
125 lastRunOrdinal: 1,
126 handoffIds: [],
127 forkedFrom: null,
128 createdAt: now,
129 updatedAt: now,
130 },
131 },
132 {
133 id: EventId.make(`event:seed-node:${name}`),
134 type: "node.updated",
135 threadId,
136 runId,
137 nodeId: rootNodeId,
138 driver,
139 providerInstanceId: instanceId,
140 occurredAt: now,
141 payload: {
142 id: rootNodeId,
143 threadId,
144 runId,
145 parentNodeId: null,
146 rootNodeId,
147 kind: "root_turn",
148 status: "running",
149 countsForRun: true,
150 providerThreadId,
151 providerTurnId: null,
152 nativeItemRef: null,
153 runtimeRequestId: null,
154 checkpointScopeId: null,
155 startedAt: now,
156 completedAt: null,
157 },
158 },
159 {
160 id: EventId.make(`event:seed-run:${name}`),
161 type: "run.updated",
162 threadId,
163 runId,
164 nodeId: rootNodeId,
165 providerInstanceId: instanceId,
166 occurredAt: now,
167 payload: {
168 id: runId,
169 threadId,
170 ordinal: 1,
171 providerInstanceId: instanceId,
172 modelSelection,
173 providerThreadId,
174 userMessageId: MessageId.make(`message:seed-user:${name}`),
175 rootNodeId,
176 activeAttemptId: null,
177 status: "running",
178 requestedAt: now,
179 startedAt: now,
180 completedAt: null,
181 checkpointId: null,
182 contextHandoffId: null,
183 },
184 },
185 ],
186 });
187 return { threadId, runId, rootNodeId };
188 });
189
190const remoteChild = {
191 environmentId: EnvironmentId.make("environment:box"),
192 threadId: ThreadId.make("thread:on-the-box"),
193 label: "Box",
194};
195
196it.layer(layerTest)("a delegated task in a linked environment", (it) => {
197 it.effect("is recorded on its parent without a child thread here, once", () =>
198 Effect.gen(function* () {
199 const orchestrator = yield* Orchestrator.OrchestratorV2;
200 const parent = yield* seedLiveParent("records");
201 const request = {
202 type: "delegated_task.remote.request" as const,
203 commandId: CommandId.make("command:remote-delegate:records"),
204 parentThreadId: parent.threadId,
205 parentRunId: parent.runId,
206 parentNodeId: parent.rootNodeId,
207 task: "Run the suite there.",
208 driver: ProviderDriverKind.make("remote"),
209 modelSelection,
210 remoteChild,
211 completionWake: "always" as const,
212 };
213 yield* orchestrator.dispatch(request);
214 yield* orchestrator.dispatch(request);
215 const projection = yield* orchestrator.getThreadProjection(parent.threadId);
216 assert.equal(projection.subagents.length, 1);
217 const [task] = projection.subagents;
218 assert.equal(task?.childThreadId, null);
219 assert.deepEqual(task?.remoteChild, remoteChild);
220 assert.equal(task?.status, "running");
221 const item = projection.turnItems.find((candidate) => candidate.type === "subagent");
222 assert.deepEqual(item?.type === "subagent" ? item.remoteChild : undefined, remoteChild);
223 // No thread was created here for the child.
224 assert.equal(yield* orchestrator.getThreadShell(remoteChild.threadId), null);
225 const other = yield* seedLiveParent("records-no-task");
226 const index = yield* (yield* ProjectionStore.ProjectionStoreV2).listActiveShellIndex();
227 assert.isTrue(index.find((entry) => entry.id === parent.threadId)?.hasRemoteTasks);
228 assert.isFalse(index.find((entry) => entry.id === other.threadId)?.hasRemoteTasks);
229 // Nothing here can cancel the task there, so it must not block archiving.
230 yield* orchestrator.dispatch({
231 type: "thread.archive",
232 commandId: CommandId.make("command:archive:records"),
233 threadId: parent.threadId,
234 });
235 assert.isNotNull((yield* orchestrator.getThreadShell(parent.threadId))?.archivedAt);
236 }),
237 );
238
239 it.effect("keeps a newer stop recoverable after an older acknowledgement", () =>
240 Effect.gen(function* () {
241 const orchestrator = yield* Orchestrator.OrchestratorV2;
242 const projections = yield* ProjectionStore.ProjectionStoreV2;
243 const parent = yield* seedLiveParent("stop-generation");
244 yield* orchestrator.dispatch({
245 type: "delegated_task.remote.request",
246 commandId: CommandId.make("command:remote-delegate:stop-generation"),
247 parentThreadId: parent.threadId,
248 parentRunId: parent.runId,
249 parentNodeId: parent.rootNodeId,
250 task: "Run a check.",
251 driver: ProviderDriverKind.make("remote"),
252 modelSelection,
253 remoteChild,
254 });
255 const task = (yield* orchestrator.getThreadProjection(parent.threadId)).subagents[0]!;
256 const openFor = projections.getOpenRemoteDelegatedTasks.pipe(
257 Effect.map((open) => open.filter((entry) => entry.parentThreadId === parent.threadId)),
258 );
259 const complete = {
260 type: "delegated_task.remote.complete" as const,
261 parentThreadId: parent.threadId,
262 taskId: task.id,
263 status: "completed" as const,
264 result: "Original result.",
265 };
266 yield* orchestrator.dispatch({
267 ...complete,
268 commandId: CommandId.make("command:result:stop-generation"),
269 stoppedThere: "ran-out",
270 });
271 for (const count of [1, 2]) {
272 yield* orchestrator.dispatch({
273 type: "delegated_task.remote.stop",
274 parentThreadId: parent.threadId,
275 taskId: task.id,
276 commandId: CommandId.make(`command:stop-generation:${count}`),
277 });
278 }
279 // Reports from before generation tracking cannot clear a later numbered Stop either.
280 for (const stopCount of [undefined, 1]) {
281 const rejected = yield* orchestrator
282 .dispatch({
283 ...complete,
284 commandId: CommandId.make(`command:stale-stop:${stopCount}`),
285 stoppedThere: "stopped",
286 ...(stopCount === undefined ? {} : { stopCount }),
287 })
288 .pipe(Effect.flip);
289 assert.equal(rejected._tag, "OrchestratorDispatchError");
290 assert.deepEqual(yield* openFor, [{ parentThreadId: parent.threadId, taskId: task.id }]);
291 }
292 yield* orchestrator.dispatch({
293 ...complete,
294 commandId: CommandId.make("command:current-stop:2"),
295 stoppedThere: "stopped",
296 stopCount: 2,
297 });
298 const settled = (yield* orchestrator.getThreadProjection(parent.threadId)).subagents[0]!;
299 assert.equal(settled.result, "Original result.");
300 assert.equal(settled.status, "completed");
301 assert.equal(settled.remoteChild?.stoppedThere, true);
302 assert.deepEqual(yield* openFor, []);
303 }),
304 );
305
306 it.effect("owes a stop there for a Stop here that loses the race to its result", () =>
307 Effect.gen(function* () {
308 const orchestrator = yield* Orchestrator.OrchestratorV2;
309 const projections = yield* ProjectionStore.ProjectionStoreV2;
310 const parent = yield* seedLiveParent("stop-after-result");
311 yield* orchestrator.dispatch({
312 type: "delegated_task.remote.request",
313 commandId: CommandId.make("command:remote-delegate:stop-after-result"),
314 parentThreadId: parent.threadId,
315 parentRunId: parent.runId,
316 parentNodeId: parent.rootNodeId,
317 task: "Run a check.",
318 driver: ProviderDriverKind.make("remote"),
319 modelSelection,
320 remoteChild,
321 });
322 const task = (yield* orchestrator.getThreadProjection(parent.threadId)).subagents[0]!;
323 const complete = {
324 type: "delegated_task.remote.complete" as const,
325 parentThreadId: parent.threadId,
326 taskId: task.id,
327 };
328 yield* orchestrator.dispatch({
329 ...complete,
330 commandId: CommandId.make("command:result:stop-after-result"),
331 status: "completed",
332 result: "Done there.",
333 stoppedThere: "ran-out",
334 });
335 // The parent's Stop read the task before its result landed.
336 yield* orchestrator.dispatch({
337 ...complete,
338 commandId: CommandId.make("command:parent-stop:stop-after-result"),
339 status: "interrupted",
340 result: "Stopped with its parent.",
341 });
342
343 const stopped = (yield* orchestrator.getThreadProjection(parent.threadId)).subagents[0]!;
344 assert.equal(stopped.status, "completed");
345 assert.equal(stopped.result, "Done there.");
346 assert.equal(stopped.remoteChild?.stoppedThere, false);
347 assert.equal(stopped.remoteChild?.stopCount, 1);
348 const open = yield* projections.getOpenRemoteDelegatedTasks;
349 assert.deepEqual(
350 open.filter((entry) => entry.parentThreadId === parent.threadId),
351 [{ parentThreadId: parent.threadId, taskId: task.id }],
352 );
353 }),
354 );
355
356 it.effect("wakes its parent with the result when it completes there", () =>
357 Effect.gen(function* () {
358 const orchestrator = yield* Orchestrator.OrchestratorV2;
359 const parent = yield* seedLiveParent("completes");
360 const recorded = yield* orchestrator.dispatch({
361 type: "delegated_task.remote.request",
362 commandId: CommandId.make("command:remote-delegate:completes"),
363 parentThreadId: parent.threadId,
364 parentRunId: parent.runId,
365 parentNodeId: parent.rootNodeId,
366 task: "Run the suite there.",
367 driver: ProviderDriverKind.make("remote"),
368 modelSelection,
369 remoteChild,
370 completionWake: "always",
371 });
372 const taskEvent = recorded.storedEvents.find(
373 (stored) => stored.event.type === "subagent.updated",
374 );
375 const taskId =
376 taskEvent?.event.type === "subagent.updated" ? taskEvent.event.payload.id : undefined;
377 assert.isDefined(taskId);
378 // Until it completes, it is open, and a restarted follower picks it up.
379 const projections = yield* ProjectionStore.ProjectionStoreV2;
380 const openFor = projections.getOpenRemoteDelegatedTasks.pipe(
381 Effect.map((open) => open.filter((task) => task.parentThreadId === parent.threadId)),
382 );
383 assert.deepEqual(yield* openFor, [{ parentThreadId: parent.threadId, taskId: taskId! }]);
384 const offers = yield* Offers;
385 yield* Ref.set(offers, []);
386 const complete = (result: string) =>
387 orchestrator.dispatch({
388 type: "delegated_task.remote.complete",
389 commandId: CommandId.make(`command:remote-complete:${result}`),
390 parentThreadId: parent.threadId,
391 taskId: taskId!,
392 status: "completed",
393 result,
394 stoppedThere: "ran-out",
395 });
396 yield* complete("All 412 tests pass on the box.");
397 // A second report is refused, so the first result stands.
398 const late = yield* complete("A late, different answer.").pipe(Effect.flip);
399 assert.equal(late._tag, "OrchestratorDispatchError");
400
401 const projection = yield* orchestrator.getThreadProjection(parent.threadId);
402 const task = projection.subagents.find((candidate) => candidate.id === taskId);
403 assert.equal(task?.status, "completed");
404 assert.equal(task?.result, "All 412 tests pass on the box.");
405 // The live parent run claimed it, so task_status acknowledges it as for a local child.
406 assert.equal(task?.completionDelivery?.state, "claimed");
407 // Nothing is left for the follower to pick up after a restart.
408 assert.deepEqual(yield* openFor, []);
409 const item = projection.turnItems.find(
410 (candidate) => candidate.type === "subagent" && candidate.subagentId === taskId,
411 );
412 assert.equal(item?.status, "completed");
413 // The parent's live run is offered the delivery, as for a local child.
414 const delivery = projection.runs.find((run) => run.id === parent.runId)?.delegatedCompletion
415 ?.delivery;
416 assert.deepEqual(delivery?.taskIds, [taskId]);
417 const offered = yield* Ref.get(offers);
418 assert.equal(offered.length, 1);
419 assert.equal(offered[0]?.threadId, parent.threadId);
420 }),
421 );
422});