1import {
2 CommandId,
3 type EnvironmentId,
4 NodeId,
5 OrchestratorMcpCapabilitiesInput,
6 OrchestratorMcpCapabilitiesResult,
7 OrchestratorMcpFailure,
8 type OrchestratorMcpDelegateTaskInput,
9 type OrchestratorMcpTaskCancelResult,
10 type OrchestratorMcpThreadRun,
11 type OrchestratorMcpThreadSendInput,
12 type OrchestratorMcpThreadSendResult,
13 type OrchestratorMcpThreadTimelineItem,
14 type OrchestrationV2RemoteTaskChild,
15 type OrchestrationV2Subagent,
16 type ProjectId,
17 type ProviderInstanceId,
18 ProviderDriverKind,
19 type RunId,
20 type ThreadId,
21} from "@t3tools/contracts";
22import * as Context from "effect/Context";
23import * as SecureRandom from "effect/crypto/SecureRandom";
24import * as DateTime from "effect/DateTime";
25import * as Duration from "effect/Duration";
26import * as Effect from "effect/Effect";
27import * as Layer from "effect/Layer";
28import * as Option from "effect/Option";
29import * as Schedule from "effect/Schedule";
30import * as Scope from "effect/Scope";
31import * as Stream from "effect/Stream";
32import * as Tool from "effect/ai/Tool";
33import { withExecutedModelNote } from "@t3tools/provider-core/server/subagentProjection";
34
35import type * as McpInvocationContext from "../mcp/McpInvocationContext.ts";
36import { ProjectToolkit } from "../mcp/toolkits/project/tools.ts";
37import {
38 resolveInteractionMode,
39 resolveRuntimeMode,
40 taskPrompt,
41} from "../mcp/delegatedTaskModes.ts";
42import { followUpOwnerRun } from "../orchestration-v2/CommandPolicy.ts";
43import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
44import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
45import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
46import * as ProjectService from "../project/ProjectService.ts";
47import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts";
48import { forkParked } from "../serverActivation.ts";
49import * as PeerForwarding from "./PeerForwarding.ts";
50import * as PeerLinks from "./PeerLinks.ts";
51
52/** How long one follow of a remote task waits on its thread there before checking again. */
53const FOLLOW_WAIT_MS = 10 * 60 * 1_000;
54/** A linked environment that stops answering is retried, backing off to this. */
55const FOLLOW_MAX_BACKOFF = Duration.minutes(5);
56/** How a follower retries; tests step it by hand. */
57export const FollowRetrySchedule = Context.Reference<Schedule.Schedule<unknown>>(
58 "t3/peer/RemoteDelegation/FollowRetrySchedule",
59 {
60 defaultValue: () =>
61 Schedule.exponential(Duration.seconds(2)).pipe(
62 Schedule.modifyDelay(({ duration }) =>
63 Effect.succeed(Duration.min(duration, FOLLOW_MAX_BACKOFF)),
64 ),
65 ),
66 },
67);
68
69type ThreadScope = McpInvocationContext.McpThreadInvocationScope;
70
71/** `orchestrator_capabilities` there. Its toolkit imports this module, so it is described here. */
72const RemoteCapabilitiesTool = Tool.make("orchestrator_capabilities", {
73 parameters: OrchestratorMcpCapabilitiesInput,
74 success: OrchestratorMcpCapabilitiesResult,
75 failure: OrchestratorMcpFailure,
76 failureMode: "return",
77});
78
79/**
80 * `delegate_task` to a linked environment. The task's child is an ordinary
81 * thread there, launched through the link; the parent here records the task
82 * like a local one, without a child thread, and a follower completes it when
83 * the thread there ends, so the parent wakes as it would for a local child.
84 */
85export class RemoteDelegation extends Context.Service<
86 RemoteDelegation,
87 {
88 readonly delegate: (
89 scope: ThreadScope,
90 input: OrchestratorMcpDelegateTaskInput & {
91 readonly target: NonNullable<OrchestratorMcpDelegateTaskInput["target"]> & {
92 readonly environmentId: EnvironmentId;
93 };
94 },
95 ) => Effect.Effect<{ readonly taskId: NodeId }, OrchestratorMcpFailure>;
96 /**
97 * A `t3_thread_send` there that started or queued a run on the thread of
98 * one of this thread's tasks that has its result: like a follow-up to a
99 * finished child here, that run is a new task, whose id the send returns.
100 * Any other send is returned as it came.
101 */
102 readonly followUp: (
103 scope: McpInvocationContext.McpInvocationScope,
104 environmentId: EnvironmentId,
105 request: Pick<OrchestratorMcpThreadSendInput, "message" | "clientRequestId">,
106 sent: OrchestratorMcpThreadSendResult,
107 ) => Effect.Effect<OrchestratorMcpThreadSendResult, OrchestratorMcpFailure>;
108 /**
109 * Stops the task's thread there, then completes it here as cancelled. A
110 * task that already has its result keeps it; its thread there is still
111 * stopped, as cancelling a local task stops follow-up work on its child.
112 */
113 readonly cancel: (
114 scope: ThreadScope,
115 task: OrchestrationV2Subagent,
116 input: {
117 readonly reason?: string | undefined;
118 readonly clientRequestId?: string | undefined;
119 },
120 ) => Effect.Effect<OrchestratorMcpTaskCancelResult, OrchestratorMcpFailure>;
121 /**
122 * Follows every open remote task until it ends, including after a
123 * restart, and a finished one again when a Stop here owes it a stop there.
124 */
125 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
126 }
127>()("t3/peer/RemoteDelegation") {}
128
129/** The other side refused because the thread there runs above this caller's modes. */
130const isModeRefusal = (error: OrchestratorMcpFailure) =>
131 error.code === "runtime_mode_escalation_denied" ||
132 error.code === "interaction_mode_escalation_denied";
133
134/**
135 * Refusals that following again cannot change: the link was revoked, expired
136 * or forgotten, or the thread or run there is gone. Anything else (the other
137 * machine asleep, a dropped route) is worth retrying.
138 */
139const isFinal = (error: OrchestratorMcpFailure) =>
140 error.code === "capability_denied" ||
141 error.code === "thread_not_found" ||
142 error.code === "run_not_found";
143
144const make = Effect.gen(function* () {
145 const followRetry = yield* FollowRetrySchedule;
146 const forwarding = yield* PeerForwarding.PeerForwarding;
147 const links = yield* PeerLinks.PeerLinks;
148 const threads = yield* ThreadManagement.ThreadManagementService;
149 const projections = yield* ProjectionStore.ProjectionStoreV2;
150 const projects = yield* ProjectService.ProjectService;
151 const repositoryIdentities = yield* RepositoryIdentityResolver.RepositoryIdentityResolver;
152 const random = yield* SecureRandom.SecureRandom;
153 const hereId = yield* ServerEnvironment.ServerEnvironment.pipe(
154 Effect.flatMap((environment) => environment.getEnvironmentId),
155 );
156 // Followers live as long as this service, so shutting down stops them.
157 const followers = yield* Scope.Scope;
158
159 // One follower runs per open task and holds what it read across a wait of
160 // up to FOLLOW_WAIT_MS, so it reads its own task, never the parent's whole
161 // task list: a parent with thousands of tasks made every restart decode
162 // that list once per open task at the same time and ran out of heap.
163 const readTask = (parentThreadId: ThreadId, taskId: NodeId) =>
164 threads
165 .getThreadRecords(parentThreadId, ["subagents"], { subagentIds: [taskId] })
166 .pipe(Effect.map(({ thread, subagents }) => ({ thread, task: subagents[0] })));
167
168 /** The peer's project with this thread's repository, or the one the caller named. */
169 const resolveRemoteProject = Effect.fn("RemoteDelegation.resolveRemoteProject")(function* (
170 scope: ThreadScope,
171 environmentId: EnvironmentId,
172 parentProjectId: ProjectId,
173 requested: ProjectId | undefined,
174 ) {
175 if (requested !== undefined) return requested;
176 const here = yield* projects.getById(parentProjectId).pipe(
177 Effect.map(Option.getOrUndefined),
178 Effect.orElseSucceed(() => undefined),
179 );
180 // The project's own identity is a one-minute cache that reads blank when
181 // cold, so resolve it here, as the other side's t3_project_list does.
182 const key =
183 here === undefined
184 ? undefined
185 : (yield* repositoryIdentities.resolve(here.workspaceRoot))?.canonicalKey;
186 if (key === undefined) {
187 return yield* new OrchestratorMcpFailure({
188 code: "target_required",
189 message: "Pass target.projectId: this thread's project has no repository to match there.",
190 });
191 }
192 const matches: Array<{ readonly id: ProjectId; readonly title: string }> = [];
193 let cursor: number | null = 0;
194 while (cursor !== null) {
195 const listed: Tool.Success<typeof ProjectToolkit.tools.t3_project_list> =
196 yield* forwarding.call(scope, ProjectToolkit.tools.t3_project_list, environmentId, {
197 cursor,
198 limit: 100,
199 });
200 matches.push(
201 ...listed.projects.filter((project) => project.repositoryIdentity?.canonicalKey === key),
202 );
203 cursor = listed.nextCursor;
204 }
205 if (matches.length === 1) return matches[0]!.id;
206 return yield* new OrchestratorMcpFailure({
207 code: "invalid_request",
208 message:
209 matches.length === 0
210 ? `No project there has this repository (${key}). Pass target.projectId; t3_project_list with this environmentId lists them. To bring the repository there, use t3_project_clone and then t3_project_create with this environmentId.`
211 : `Several projects there have this repository: ${matches.map((project) => `${project.title} (${project.id})`).join(", ")}. Pass target.projectId.`,
212 });
213 });
214
215 /** The instance there of the driver the caller named, preferring the one this thread runs on, as a task here does. */
216 const resolveRemoteInstance = Effect.fn("RemoteDelegation.resolveRemoteInstance")(function* (
217 scope: ThreadScope,
218 environmentId: EnvironmentId,
219 requested: {
220 readonly driverKind: ProviderDriverKind;
221 readonly providerInstanceId: ProviderInstanceId | undefined;
222 readonly inheritedInstanceId: ProviderInstanceId;
223 },
224 ) {
225 const { providers } = yield* forwarding.call(scope, RemoteCapabilitiesTool, environmentId, {});
226 if (requested.providerInstanceId !== undefined) {
227 const named = providers.find(
228 (provider) => provider.providerInstanceId === requested.providerInstanceId,
229 );
230 if (named !== undefined && named.driverKind !== requested.driverKind) {
231 return yield* new OrchestratorMcpFailure({
232 code: "invalid_request",
233 message: `Provider instance ${named.providerInstanceId} there uses driver ${named.driverKind}, not ${requested.driverKind}.`,
234 });
235 }
236 return requested.providerInstanceId;
237 }
238 const candidates = providers.filter(
239 (provider) => provider.driverKind === requested.driverKind && provider.canRunChildTask,
240 );
241 const chosen =
242 candidates.find(
243 (provider) => provider.providerInstanceId === requested.inheritedInstanceId,
244 ) ?? candidates[0];
245 if (chosen === undefined) {
246 return yield* new OrchestratorMcpFailure({
247 code: "provider_unavailable",
248 message: `No instance of driver ${requested.driverKind} there can run a child task. orchestrator_capabilities with this environmentId lists its providers.`,
249 });
250 }
251 return chosen.providerInstanceId;
252 });
253
254 const delegate: RemoteDelegation["Service"]["delegate"] = Effect.fn("RemoteDelegation.delegate")(
255 function* (scope, input) {
256 const parent = yield* threads.getThreadRecords(scope.thread.threadId, ["runs"]).pipe(
257 Effect.mapError(
258 () =>
259 new OrchestratorMcpFailure({
260 code: "thread_not_found",
261 message: "The calling thread was not found.",
262 }),
263 ),
264 );
265 const parentRun = parent.runs
266 .filter(ThreadManagement.isActiveRun)
267 .toSorted((left, right) => right.ordinal - left.ordinal)[0];
268 if (
269 parentRun === undefined ||
270 parentRun.rootNodeId === null ||
271 parentRun.providerInstanceId !== scope.thread.providerInstanceId
272 ) {
273 return yield* new OrchestratorMcpFailure({
274 code: "parent_not_active",
275 message: "Delegated tasks require an active run owned by this MCP provider session.",
276 });
277 }
278 const parentNodeId = parentRun.rootNodeId;
279 const { target } = input;
280 // A worktree there would branch from a commit this checkout has and the
281 // other may not, so a remote task runs in that project's checkout.
282 if (input.isolation === "worktree") {
283 return yield* new OrchestratorMcpFailure({
284 code: "invalid_request",
285 message:
286 'A task in a linked environment runs in that project\'s checkout; isolation "worktree" is not supported there. Omit isolation, or use t3_thread_launch with environmentId and a worktree workspaceStrategy.',
287 });
288 }
289 // The provider and model inherit as they do for a task here. Another
290 // provider there has no model to inherit, so the caller names one.
291 const inherited = parent.thread.modelSelection;
292 const instanceId =
293 target.driverKind === undefined
294 ? (target.providerInstanceId ?? inherited.instanceId)
295 : yield* resolveRemoteInstance(scope, target.environmentId, {
296 driverKind: target.driverKind,
297 providerInstanceId: target.providerInstanceId,
298 inheritedInstanceId: inherited.instanceId,
299 });
300 const model =
301 target.model ?? (instanceId === inherited.instanceId ? inherited.model : undefined);
302 if (model === undefined) {
303 return yield* new OrchestratorMcpFailure({
304 code: "target_required",
305 message: "Pass target.model from orchestrator_capabilities with this environmentId.",
306 });
307 }
308 const options =
309 target.options ??
310 (instanceId === inherited.instanceId && model === inherited.model
311 ? inherited.options
312 : undefined);
313 // Modes asked for are checked against this thread. Omitted ones are left
314 // to the other side, which inherits this thread's modes capped by the
315 // link's access, so a narrower link still takes the task.
316 const runtimeMode =
317 input.runtimeMode === undefined || input.runtimeMode === "inherit"
318 ? undefined
319 : yield* resolveRuntimeMode(parent.thread.runtimeMode, input.runtimeMode);
320 const interactionMode =
321 input.interactionMode === undefined || input.interactionMode === "inherit"
322 ? undefined
323 : yield* resolveInteractionMode(parent.thread.interactionMode, input.interactionMode);
324 const projectId = yield* resolveRemoteProject(
325 scope,
326 target.environmentId,
327 parent.thread.projectId,
328 target.projectId,
329 );
330 // One key for the launch there and the record here, so a retry finds both.
331 // Both are scoped to the provider session, as every key a caller passes is:
332 // the forwarder scopes the launch's, and the record's names it.
333 const key = input.clientRequestId ?? (yield* random.uuidV4());
334 const task = taskPrompt(input);
335 const title = input.title ?? parent.thread.title;
336 const launched = yield* forwarding.call(
337 scope,
338 ProjectToolkit.tools.t3_thread_launch,
339 target.environmentId,
340 {
341 projectId,
342 title,
343 modelSelection: { instanceId, model, ...(options === undefined ? {} : { options }) },
344 ...(runtimeMode === undefined ? {} : { runtimeMode }),
345 ...(interactionMode === undefined ? {} : { interactionMode }),
346 message: task,
347 clientRequestId: `delegate:${scope.thread.threadId}:${key}`,
348 // The thread there shows it is this thread's subagent, with a way back.
349 delegatedFrom: {
350 environmentId: hereId,
351 threadId: parent.thread.id,
352 title: parent.thread.title,
353 },
354 },
355 );
356 const link = yield* links.get(target.environmentId).pipe(
357 Effect.map(Option.getOrUndefined),
358 Effect.orElseSucceed(() => undefined),
359 );
360 const commandId = CommandId.make(
361 `command:mcp:${encodeURIComponent(scope.requestNamespace)}:remote-delegate:${encodeURIComponent(key)}`,
362 );
363 const remoteChild: OrchestrationV2RemoteTaskChild = {
364 environmentId: target.environmentId,
365 threadId: launched.threadId,
366 label: link?.label ?? target.environmentId,
367 title,
368 ...(launched.runId === null ? {} : { runId: launched.runId }),
369 };
370 const recorded = yield* threads
371 .dispatch({
372 type: "delegated_task.remote.request",
373 commandId,
374 parentThreadId: scope.thread.threadId,
375 parentRunId: parentRun.id,
376 parentNodeId,
377 task,
378 ...(input.title === undefined ? {} : { title: input.title }),
379 // The driver runs there; this side only labels the task with it.
380 driver: ProviderDriverKind.make("remote"),
381 modelSelection: launched.modelSelection,
382 remoteChild,
383 completionWake: input.mode === "wait" ? "settled_only" : "always",
384 })
385 .pipe(
386 // The cause goes to the log; the caller learns only that recording failed.
387 Effect.tapError((cause) =>
388 Effect.logWarning("remote-delegation.record-failed", {
389 parentThreadId: scope.thread.threadId,
390 cause,
391 }),
392 ),
393 Effect.mapError(
394 () =>
395 new OrchestratorMcpFailure({
396 code: "orchestration_error",
397 message: "Unable to record the delegated task.",
398 }),
399 ),
400 // Nothing here follows a thread the task was never recorded for, so stop it there.
401 // A crash between the launch and this record still leaves it running there
402 // until a retry with the same clientRequestId records it.
403 Effect.tapError(() =>
404 stopThere(followScope(scope.thread, remoteChild), remoteChild, "stop").pipe(
405 Effect.ignoreCause({ log: true }),
406 ),
407 ),
408 );
409 const taskEvent = recorded.storedEvents.find(
410 (stored) => stored.event.type === "subagent.updated",
411 );
412 const taskId =
413 taskEvent?.event.type === "subagent.updated"
414 ? taskEvent.event.payload.id
415 : // A replayed request recorded the task the first time.
416 yield* threads.getThreadRecords(scope.thread.threadId, ["subagents"]).pipe(
417 Effect.map(
418 ({ subagents }) =>
419 subagents.find((task) => task.remoteChild?.threadId === launched.threadId)?.id,
420 ),
421 Effect.orElseSucceed(() => undefined),
422 );
423 if (taskId === undefined) {
424 return yield* new OrchestratorMcpFailure({
425 code: "orchestration_error",
426 message: "The delegated task was not recorded.",
427 });
428 }
429 yield* followOne(scope.thread.threadId, taskId).pipe(Effect.forkIn(followers));
430 return { taskId };
431 },
432 );
433
434 const followUp: RemoteDelegation["Service"]["followUp"] = Effect.fn("RemoteDelegation.followUp")(
435 function* (scope, environmentId, request, sent) {
436 // A steer joins a run that is already some task's, or no one's.
437 if (scope.thread === undefined || (sent.delivery !== "started" && sent.delivery !== "queued"))
438 return sent;
439 const parentThreadId = scope.thread.threadId;
440 const parent = yield* threads
441 .getThreadRecords(parentThreadId, ["runs", "subagents"])
442 .pipe(Effect.orElseSucceed(() => undefined));
443 const isThere = (task: OrchestrationV2Subagent) =>
444 task.remoteChild?.environmentId === environmentId &&
445 task.remoteChild.threadId === sent.threadId;
446 const finished = parent?.subagents.find((task) => isThere(task) && task.result !== null);
447 // Bound as a follow-up here is: a background caller's foreground run may have ended.
448 const parentRun = parent === undefined ? null : followUpOwnerRun(parent.runs);
449 if (
450 finished?.remoteChild === undefined ||
451 finished.model === null ||
452 parentRun === null ||
453 parentRun.rootNodeId === null
454 )
455 return sent;
456 // The message id there is the send's, so a retry that reached the same send records once.
457 const commandId = CommandId.make(
458 `command:remote-followup:${encodeURIComponent(parentThreadId)}:${encodeURIComponent(environmentId)}:${encodeURIComponent(sent.messageId)}`,
459 );
460 const recorded = yield* threads
461 .dispatch({
462 type: "delegated_task.remote.request",
463 commandId,
464 parentThreadId,
465 parentRunId: parentRun.id,
466 parentNodeId: parentRun.rootNodeId,
467 task: request.message,
468 ...(finished.title === null ? {} : { title: finished.title }),
469 driver: finished.driver,
470 modelSelection: { instanceId: finished.providerInstanceId, model: finished.model },
471 remoteChild: {
472 environmentId,
473 threadId: sent.threadId,
474 label: finished.remoteChild.label,
475 runId: sent.runId,
476 },
477 completionWake: "always",
478 })
479 .pipe(
480 Effect.tapError((cause) =>
481 Effect.logWarning("remote-delegation.followup-record-failed", {
482 parentThreadId,
483 cause,
484 }),
485 ),
486 Effect.mapError(
487 () =>
488 new OrchestratorMcpFailure({
489 code: "orchestration_error",
490 message:
491 request.clientRequestId === undefined
492 ? "The message was sent, but its task was not recorded; read the thread for its answer instead of sending it again."
493 : "The message was sent, but its task was not recorded; send it again with the same clientRequestId to record it.",
494 }),
495 ),
496 );
497 const taskEvent = recorded.storedEvents.find(
498 (stored) => stored.event.type === "subagent.updated",
499 );
500 const taskId =
501 taskEvent?.event.type === "subagent.updated"
502 ? taskEvent.event.payload.id
503 : // A replayed request recorded the task the first time.
504 yield* threads.getThreadRecords(parentThreadId, ["subagents"]).pipe(
505 Effect.map(
506 ({ subagents }) =>
507 subagents.find((task) => isThere(task) && task.remoteChild?.runId === sent.runId)
508 ?.id,
509 ),
510 Effect.orElseSucceed(() => undefined),
511 );
512 if (taskId === undefined) return sent;
513 yield* followOne(parentThreadId, taskId).pipe(Effect.forkIn(followers));
514 return { ...sent, taskId };
515 },
516 );
517
518 /**
519 * Completes `task` here with its thread there's final state. Each attempt is its own
520 * command, so a failed one never blocks the next; a report refused because the task already
521 * has its result (a cancel, a Stop or an earlier report) is done. `stoppedThere` records that
522 * nothing is left to stop there, even on a task that already has its result.
523 */
524 const complete = Effect.fn("RemoteDelegation.complete")(function* (
525 parentThreadId: ThreadId,
526 taskId: NodeId,
527 status: "completed" | "failed" | "cancelled" | "interrupted",
528 result: string,
529 stoppedThere: "ran-out" | "stopped" | undefined,
530 stopCount?: number,
531 ) {
532 const attempt = yield* random.uuidV4();
533 yield* threads
534 .dispatch({
535 type: "delegated_task.remote.complete",
536 commandId: CommandId.make(`command:remote-task-complete:${taskId}:${attempt}`),
537 parentThreadId,
538 taskId,
539 status,
540 result,
541 ...(stoppedThere === undefined ? {} : { stoppedThere }),
542 ...(stopCount === undefined ? {} : { stopCount }),
543 })
544 .pipe(
545 Effect.catch((error) =>
546 readTask(parentThreadId, taskId).pipe(
547 Effect.flatMap(({ task }) => (task?.result != null ? Effect.void : Effect.fail(error))),
548 ),
549 ),
550 );
551 });
552
553 /**
554 * How this environment reaches a task's thread there: as the parent
555 * thread, under one namespace per parent, so a stop sent again after a
556 * restart or a dropped route is recorded there once.
557 */
558 const followScope = (
559 parent: { readonly threadId: ThreadId; readonly providerInstanceId: ProviderInstanceId },
560 remote: OrchestrationV2RemoteTaskChild,
561 ): ThreadScope => ({
562 environmentId: hereId,
563 requestNamespace: `remote-task:${parent.threadId}`,
564 thread: {
565 threadId: parent.threadId,
566 providerSessionId: `remote-task:${remote.threadId}`,
567 providerInstanceId: parent.providerInstanceId,
568 },
569 client: undefined,
570 capabilities: new Set(["orchestration"]),
571 issuedAt: 0,
572 });
573
574 /**
575 * Stops the task's own thread there as its Stop button does: its running turn, queued
576 * messages, pull request watches and delegated tasks. `key` names the stop (the follower
577 * sends one, each cancel its own), and each is recorded there once however often it is sent.
578 */
579 const stopThere = (
580 scope: ThreadScope,
581 remote: OrchestrationV2RemoteTaskChild,
582 key: string,
583 reason?: string,
584 ) =>
585 forwarding
586 .interruptThread(scope, remote.environmentId, {
587 threadId: remote.threadId,
588 stop: true,
589 ...(reason === undefined ? {} : { reason }),
590 clientRequestId: `${key}:${remote.threadId}`,
591 })
592 .pipe(Effect.asVoid);
593
594 /**
595 * The last reply `runId` gave on the task's thread there, which is its
596 * result, read whole however long it is. A failed run prefers its error
597 * from the activity view over an earlier reply, without taking another run's.
598 * Led by a note when its provider reported running another model than it asked for.
599 */
600 const lastReply = Effect.fn("RemoteDelegation.lastReply")(function* (
601 scope: ThreadScope,
602 remote: OrchestrationV2RemoteTaskChild,
603 runId: RunId | null,
604 failed = false,
605 ) {
606 let reply: OrchestratorMcpThreadTimelineItem | undefined;
607 let error: OrchestratorMcpThreadTimelineItem | undefined;
608 let afterPosition: number | undefined;
609 let run: OrchestratorMcpThreadRun | undefined;
610 while (true) {
611 const read = yield* forwarding.readThread(scope, remote.environmentId, {
612 threadId: remote.threadId,
613 view: failed ? "activity" : "messages",
614 limit: 100,
615 runLimit: 50,
616 maxCharsPerItem: 20_000,
617 ...(afterPosition === undefined ? {} : { afterPosition }),
618 });
619 run ??= read.recentRuns.find((candidate) => candidate.runId === runId);
620 reply =
621 read.items.findLast(
622 (item) => item.type === "assistant_message" && (runId === null || item.runId === runId),
623 ) ?? reply;
624 if (failed) {
625 error =
626 read.items.findLast(
627 (item) => item.type === "error" && (runId === null || item.runId === runId),
628 ) ?? error;
629 }
630 if (!read.hasMore || read.nextPosition === null) break;
631 afterPosition = read.nextPosition;
632 }
633 reply = error ?? reply;
634 if (reply === undefined) return undefined;
635 let text = reply.text ?? "";
636 let textOffset = reply.textTruncated ? (reply.nextTextOffset ?? null) : null;
637 while (textOffset !== null) {
638 const rest = yield* forwarding.readThread(scope, remote.environmentId, {
639 threadId: remote.threadId,
640 itemId: reply.itemId,
641 textOffset,
642 maxCharsPerItem: 20_000,
643 });
644 const part = rest.items[0];
645 if (part === undefined) break;
646 text += part.text ?? "";
647 textOffset = part.textTruncated ? (part.nextTextOffset ?? null) : null;
648 }
649 return run === undefined ? text : withExecutedModelNote(text, run.model, run.executedModel);
650 });
651
652 /**
653 * Why following a task can never succeed, if that is what `error` means.
654 * The far side's message already says what was refused and what to do; a
655 * link forgotten here only reads as a bad request, so it is named.
656 */
657 const finalReason = Effect.fn("RemoteDelegation.finalReason")(
658 function* (parentThreadId: ThreadId, taskId: NodeId, error: OrchestratorMcpFailure) {
659 if (isFinal(error)) return error.message;
660 if (error.code !== "invalid_request") return undefined;
661 const { task } = yield* readTask(parentThreadId, taskId);
662 const environmentId = task?.remoteChild?.environmentId;
663 if (environmentId === undefined) return error.message;
664 const link = yield* links.get(environmentId);
665 return Option.isSome(link) ? undefined : error.message;
666 },
667 Effect.orElseSucceed(() => undefined),
668 );
669
670 /** A task's thread there that ended without a reply. */
671 const noReply = "(The thread in the linked environment ended without a reply.)";
672
673 /**
674 * Waits for the task's run there to end, completes the task here, and stops it there if it
675 * ended here first. A revoked, expired or forgotten link, or a thread or run gone there,
676 * fails the task with nothing left to stop; anything else (the other machine asleep, a
677 * dropped route) is retried with back-off, and so is completing.
678 */
679 const follow = (parentThreadId: ThreadId, taskId: NodeId): Effect.Effect<void> =>
680 followOnce(parentThreadId, taskId).pipe(
681 // `false` means it is still running there: follow again straight away.
682 Effect.repeat({ while: (done) => !done }),
683 Effect.catchTags({
684 OrchestratorMcpFailure: Effect.fn("RemoteDelegation.followFailed")(function* (error) {
685 const reason = yield* finalReason(parentThreadId, taskId, error);
686 if (reason === undefined) return yield* error;
687 // Gone there, or no link to reach it by: nothing left to stop there.
688 const { task } = yield* readTask(parentThreadId, taskId);
689 const stopCount = task?.remoteChild?.stopCount;
690 yield* complete(parentThreadId, taskId, "failed", reason, "stopped", stopCount);
691 }),
692 }),
693 Effect.retry(followRetry),
694 Effect.ignoreCause({ log: true }),
695 );
696
697 /**
698 * One wait on the task's thread there; true once the task here is done.
699 * A task completed here first (cancelled, or its parent stopped) is
700 * stopped there, and done once that stop is recorded here.
701 */
702 const followOnce = Effect.fn("RemoteDelegation.followOnce")(function* (
703 parentThreadId: ThreadId,
704 taskId: NodeId,
705 ) {
706 const sequence = yield* threads.getThreadEventSequence(parentThreadId);
707 const { thread, task } = yield* readTask(parentThreadId, taskId);
708 if (task?.remoteChild === undefined) return true;
709 const remote = task.remoteChild;
710 const scope = followScope(
711 { threadId: parentThreadId, providerInstanceId: thread.providerInstanceId },
712 remote,
713 );
714 // Ended here, maybe while this follower backed off from an unreachable
715 // machine or before a restart: it still has to be stopped there. So
716 // does one its parent here was stopped after, for what followed there.
717 if (task.result !== null) {
718 if (
719 remote.stoppedThere === false ||
720 ((task.status === "cancelled" || task.status === "interrupted") &&
721 remote.stoppedThere !== true)
722 ) {
723 const count = remote.stopCount === undefined ? "stop" : `stop-${remote.stopCount}`;
724 // Named for its run too: a follow-up task shares the thread there and counts its own stops.
725 const key = remote.runId === undefined ? count : `${count}:${remote.runId}`;
726 yield* stopThere(scope, remote, key).pipe(
727 // Its thread there now runs above what this side may touch, say
728 // because the parent here was lowered since; no retry changes that.
729 Effect.catchIf(isModeRefusal, (error) =>
730 Effect.logWarning("A linked environment refused to stop a delegated task", {
731 taskId,
732 reason: error.message,
733 }),
734 ),
735 );
736 yield* complete(
737 parentThreadId,
738 taskId,
739 terminalStatus(task.status),
740 task.result,
741 "stopped",
742 remote.stopCount,
743 );
744 return yield* settled(parentThreadId, taskId);
745 }
746 return true;
747 }
748 const endedHere = threads
749 .streamStoredEventsFrom({
750 threadId: parentThreadId,
751 afterSequence: sequence,
752 eventType: "subagent.updated",
753 })
754 .pipe(
755 Stream.filter(
756 (stored) =>
757 stored.event.type === "subagent.updated" &&
758 stored.event.payload.id === taskId &&
759 stored.event.payload.result !== null,
760 ),
761 Stream.runHead,
762 Effect.flatMap((ended) =>
763 Option.isSome(ended) ? Effect.succeed({ endedHere: true } as const) : Effect.never,
764 ),
765 );
766 const waited = yield* forwarding
767 .waitForThread(scope, remote.environmentId, {
768 threadId: remote.threadId,
769 ...(remote.runId === undefined ? {} : { runId: remote.runId }),
770 timeoutMs: FOLLOW_WAIT_MS,
771 // A thread there whose run ended may still be waiting on its own
772 // tasks to wake it for the answer, which a task here waits for too.
773 // An older peer ignores this and reports the run alone.
774 untilDelegatedWorkSettled: true,
775 })
776 .pipe(Effect.raceFirst(endedHere));
777 // The next look reads the result and stops it there.
778 if ("endedHere" in waited) return false;
779 if (waited.timedOut) return false;
780 // A task recorded without its run waits on the thread's latest; a
781 // thread there with none has nothing left to wait for.
782 if (waited.runId === null) {
783 const reply = yield* lastReply(scope, remote, null);
784 yield* complete(
785 parentThreadId,
786 taskId,
787 reply === undefined ? "failed" : "completed",
788 reply ?? "The thread in the linked environment has no run to follow.",
789 "ran-out",
790 );
791 return yield* settled(parentThreadId, taskId);
792 }
793 if (!isTerminal(waited.status)) return false;
794 // A later run's answer there may be another open follow-up's, or its restart, which no run id
795 // here names; concurrent sends may record here out of their order there. Then this task's
796 // answer is its own run's latest restart, the one waited on.
797 const sharedThread =
798 waited.resultRun !== undefined &&
799 waited.resultRun.runId !== waited.runId &&
800 (yield* threads.getThreadRecords(parentThreadId, ["subagents"]).pipe(
801 Effect.map(({ subagents }) => {
802 const recordedAt = subagents.find((task) => task.id === taskId)?.startedAt ?? null;
803 return subagents.some(
804 (other) =>
805 other.id !== taskId &&
806 other.remoteChild?.environmentId === remote.environmentId &&
807 other.remoteChild.threadId === remote.threadId &&
808 // One that settled before this was recorded ran before it there.
809 (recordedAt === null ||
810 other.completedAt === null ||
811 DateTime.isGreaterThan(other.completedAt, recordedAt)),
812 );
813 }),
814 ));
815 const resultRun =
816 waited.resultRun !== undefined && !sharedThread
817 ? waited.resultRun
818 : { runId: waited.runId, status: waited.status };
819 const result = yield* lastReply(scope, remote, resultRun.runId, resultRun.status === "failed");
820 yield* complete(
821 parentThreadId,
822 taskId,
823 terminalStatus(resultRun.status),
824 result ?? noReply,
825 "ran-out",
826 );
827 return yield* settled(parentThreadId, taskId);
828 });
829
830 /**
831 * Whether the task is done here after the follower reported its run ran
832 * out there. A cancel or Stop that landed while the reply was read keeps
833 * its result, and still owes a stop there for what may follow the run;
834 * the next look sends it.
835 */
836 const settled = (parentThreadId: ThreadId, taskId: NodeId) =>
837 readTask(parentThreadId, taskId).pipe(
838 Effect.map(({ task }) => task?.remoteChild?.stoppedThere === true),
839 );
840
841 const following = new Map<NodeId, { again: boolean }>();
842
843 /**
844 * `follow`, at most once per task. Asked again while it runs, say by a
845 * Stop here on a task that already has its result, it looks once more
846 * before it ends, so that stop is not missed.
847 */
848 const followOne = (parentThreadId: ThreadId, taskId: NodeId) =>
849 Effect.suspend(() => {
850 const running = following.get(taskId);
851 if (running !== undefined) {
852 running.again = true;
853 return Effect.void;
854 }
855 const state = { again: false };
856 following.set(taskId, state);
857 return follow(parentThreadId, taskId).pipe(
858 // Checked and released in one step, so a request between them is never lost.
859 Effect.repeat({
860 while: () => {
861 if (state.again) {
862 state.again = false;
863 return true;
864 }
865 following.delete(taskId);
866 return false;
867 },
868 }),
869 Effect.ensuring(
870 Effect.sync(() => {
871 if (following.get(taskId) === state) following.delete(taskId);
872 }),
873 ),
874 );
875 });
876
877 const cancel: RemoteDelegation["Service"]["cancel"] = Effect.fn("RemoteDelegation.cancel")(
878 function* (scope, task, input) {
879 const remote = task.remoteChild;
880 if (remote === undefined) {
881 return yield* new OrchestratorMcpFailure({
882 code: "task_not_found",
883 message: `Delegated task ${task.id} is not remote.`,
884 });
885 }
886 const { reason } = input;
887 const there = followScope(scope.thread, remote);
888 // A later cancel stops what started there since; a retry of this one, nothing new.
889 const key = `cancel:${encodeURIComponent(scope.requestNamespace)}:${
890 input.clientRequestId ?? (yield* random.uuidV4())
891 }`;
892 if (task.result !== null) {
893 // Best effort: the result stands whether or not the stop gets there.
894 yield* stopThere(there, remote, key, reason).pipe(
895 Effect.andThen(
896 complete(
897 task.threadId,
898 task.id,
899 terminalStatus(task.status),
900 task.result,
901 "stopped",
902 remote.stopCount,
903 ),
904 ),
905 Effect.ignoreCause({ log: true }),
906 );
907 return {
908 taskId: task.id,
909 status:
910 task.status === "completed" || task.status === "failed" || task.status === "interrupted"
911 ? task.status
912 : "cancelled",
913 } satisfies OrchestratorMcpTaskCancelResult;
914 }
915 // A link that is gone or a machine that is down still cancels the task
916 // here; its follower keeps trying to stop it there, across restarts.
917 const unstopped = yield* stopThere(there, remote, key, reason).pipe(
918 Effect.as(undefined),
919 Effect.catch((error) => Effect.succeed(error.message)),
920 );
921 const cancelled = reason === undefined ? "Cancelled." : `Cancelled: ${reason}`;
922 yield* complete(
923 task.threadId,
924 task.id,
925 "cancelled",
926 unstopped === undefined
927 ? cancelled
928 : `${cancelled} It could not be stopped in ${remote.label} and may still run there: ${unstopped}`,
929 unstopped === undefined ? "stopped" : undefined,
930 remote.stopCount,
931 ).pipe(
932 Effect.tapError((cause) =>
933 Effect.logWarning("remote-delegation.cancel-record-failed", { taskId: task.id, cause }),
934 ),
935 Effect.mapError(
936 () =>
937 new OrchestratorMcpFailure({
938 code: "orchestration_error",
939 message: "Unable to record the cancellation.",
940 }),
941 ),
942 );
943 return {
944 taskId: task.id,
945 status: "cancel_requested",
946 } satisfies OrchestratorMcpTaskCancelResult;
947 },
948 );
949
950 const start: RemoteDelegation["Service"]["start"] = () =>
951 forkParked(
952 Effect.gen(function* () {
953 // A finished task's follower has ended, so a Stop here that owes its
954 // thread there a stop starts it again. Started before the scan below;
955 // a Stop that slips between the two stays owed for the next start.
956 yield* Stream.runForEach(threads.streamDomainEvents, (event) =>
957 event.type === "subagent.updated" && event.payload.remoteChild?.stoppedThere === false
958 ? followOne(event.threadId, event.payload.id).pipe(Effect.forkIn(followers))
959 : Effect.void,
960 ).pipe(
961 Effect.ignoreCause({ log: true }),
962 Effect.forkIn(followers, { startImmediately: true }),
963 );
964 const open = yield* projections.getOpenRemoteDelegatedTasks;
965 yield* Effect.forEach(
966 open,
967 ({ parentThreadId, taskId }) => followOne(parentThreadId, taskId),
968 {
969 concurrency: "unbounded",
970 discard: true,
971 },
972 );
973 }).pipe(Effect.ignoreCause({ log: true })),
974 );
975
976 return RemoteDelegation.of({ delegate, followUp, cancel, start });
977});
978
979const isTerminal = (status: string) =>
980 status === "completed" ||
981 status === "failed" ||
982 status === "cancelled" ||
983 status === "interrupted" ||
984 status === "rolled_back";
985
986const terminalStatus = (status: string): "completed" | "failed" | "cancelled" | "interrupted" =>
987 status === "completed" || status === "failed" || status === "interrupted" ? status : "cancelled";
988
989export const layer = Layer.effect(RemoteDelegation, make);