diff --git a/apps/mobile/src/features/threads/subagent-card-presentation.test.ts b/apps/mobile/src/features/threads/subagent-card-presentation.test.ts index 14a8e6ce99dfe..4bb84596fa26f 100644 --- a/apps/mobile/src/features/threads/subagent-card-presentation.test.ts +++ b/apps/mobile/src/features/threads/subagent-card-presentation.test.ts @@ -1,6 +1,7 @@ import * as DateTime from "effect/DateTime"; import { describe, expect, it } from "vite-plus/test"; import { subagentCardElapsed } from "./subagent-card-presentation"; +import { resolveSubagentRowPresentation, subagentThreadTarget } from "./threadAgentsPresentation"; describe("subagent card", () => { const start = DateTime.makeUnsafe("2026-09-21T12:00:00Z"); @@ -8,6 +9,35 @@ describe("subagent card", () => { const later = DateTime.makeUnsafe("2026-09-21T12:02:00Z"); const done = { status: "completed" as const, startedAt: start, completedAt: end }; + it("keeps a pending remote launch visible without opening a nonexistent thread", () => { + const pending = { + title: null, + prompt: "Audit the timestamps", + status: "running" as const, + result: null, + childThreadId: null, + remoteChild: { environmentId: "environment-box" as never, label: "Box" }, + }; + expect(resolveSubagentRowPresentation(pending, true)).toMatchObject({ + runsOn: "Box", + statusLabel: "Launching", + canOpenThread: false, + }); + expect(subagentThreadTarget(pending, "environment-here" as never, () => true)).toBeNull(); + const launchError = "Launch unconfirmed; work may still be running on Box."; + expect( + resolveSubagentRowPresentation( + { + ...pending, + status: "cancelled", + result: "Cancelled.", + remoteChild: { ...pending.remoteChild, launchError }, + }, + true, + ), + ).toMatchObject({ statusLabel: "Cancelled", detail: launchError, canOpenThread: false }); + }); + it("counts live work but never uses a settled agent's age as its duration", () => { const unfinished = { ...done, completedAt: null }; expect(subagentCardElapsed([unfinished], DateTime.toEpochMillis(later))).toBeNull(); diff --git a/apps/mobile/src/features/threads/thread-subagent-group.tsx b/apps/mobile/src/features/threads/thread-subagent-group.tsx index 734ebf4583e03..319d064aa974a 100644 --- a/apps/mobile/src/features/threads/thread-subagent-group.tsx +++ b/apps/mobile/src/features/threads/thread-subagent-group.tsx @@ -127,6 +127,7 @@ export function ThreadSubagentGroup(props: { completedAt: live?.completedAt ?? item.completedAt, result: live?.result ?? item.result, progress: live?.progress ?? item.progress, + remoteChild: live?.remoteChild ?? item.remoteChild, model: live?.model ?? null, modelSelection: live?.modelSelection, }; @@ -152,7 +153,7 @@ export function ThreadSubagentGroup(props: { {agents.slice(0, 3).map((agent, index) => ( - + ))} {agents.length > 3 ? ( diff --git a/apps/mobile/src/features/threads/threadAgentsPresentation.ts b/apps/mobile/src/features/threads/threadAgentsPresentation.ts index 64c9b232ed424..f72eb7d9d8399 100644 --- a/apps/mobile/src/features/threads/threadAgentsPresentation.ts +++ b/apps/mobile/src/features/threads/threadAgentsPresentation.ts @@ -47,7 +47,9 @@ export function subagentThreadTarget( return { environmentId, threadId: subagent.childThreadId }; } const remote = subagent.remoteChild; - return remote !== undefined && isKnownEnvironment(remote.environmentId) + return remote !== undefined && + remote.threadId !== undefined && + isKnownEnvironment(remote.environmentId) ? { environmentId: remote.environmentId, threadId: remote.threadId } : null; } @@ -99,12 +101,19 @@ export function resolveSubagentRowPresentation( const live = isActiveSubagentStatus(subagent.status); return { title: rowTitle(subagent), - detail: subagentDetailPreview(subagent), - statusLabel: rowStatusLabel(subagent.status), + detail: subagentDetailPreview({ + ...subagent, + result: subagent.remoteChild?.launchError ?? subagent.result, + }), + statusLabel: + live && subagent.remoteChild !== undefined && subagent.remoteChild.threadId === undefined + ? "Launching" + : rowStatusLabel(subagent.status), tone: rowTone(subagent.status), live, canOpenThread: - subagent.childThreadId !== null || (subagent.remoteChild !== undefined && canOpenRemote), + subagent.childThreadId !== null || + (subagent.remoteChild?.threadId !== undefined && canOpenRemote), runsOn: subagent.remoteChild?.label ?? null, }; } diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index e878322472aa4..db65a94954b64 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1971,8 +1971,27 @@ const make = Effect.gen(function* () { ...(task.remoteChild === undefined ? {} : { - remoteChild: task.remoteChild, - link: formatThreadLink(task.remoteChild, task.remoteChild.title), + ...(task.remoteChild.threadId === undefined + ? {} + : { + link: formatThreadLink( + { + environmentId: task.remoteChild.environmentId, + threadId: task.remoteChild.threadId, + }, + task.remoteChild.title, + ), + }), + remoteChild: { + environmentId: task.remoteChild.environmentId, + label: task.remoteChild.label, + ...(task.remoteChild.launchError === undefined + ? {} + : { launchError: task.remoteChild.launchError }), + ...(task.remoteChild.threadId === undefined + ? {} + : { threadId: task.remoteChild.threadId }), + }, }), childRunId: null, childNodeId: task.id, diff --git a/apps/server/src/mcp/toolkits/project/handlers.test.ts b/apps/server/src/mcp/toolkits/project/handlers.test.ts index c95fadd9096e2..931e15e0ae3ae 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.test.ts @@ -61,6 +61,7 @@ const clientLaunchHarness = (input: { readonly hostedAppUrl?: string; /** The orchestrator refuses the new thread's mode, as under a managed ceiling. */ readonly refusedByCeiling?: true; + readonly launchFailure?: "missing" | "lookup"; }) => { const projectId = ProjectId.make("project:client-target"); const modelSelection = { @@ -125,6 +126,18 @@ const clientLaunchHarness = (input: { checkWorktreeBase: input.checkWorktreeBase ?? (() => Effect.void), launch: (launch) => { input.launched.push(launch); + if (input.launchFailure) + return Effect.fail( + new ThreadLaunch.ThreadLaunchError({ + operation: "resolve-project", + commandId: launch.commandId, + projectId: launch.projectId, + cause: + input.launchFailure === "missing" + ? new Project.ProjectNotFoundError({ projectId: launch.projectId }) + : new Error("Project storage unavailable"), + }), + ); if (input.refusedByCeiling) { return Effect.fail( new ThreadLaunch.ThreadLaunchError({ @@ -934,6 +947,31 @@ it.effect("deleting a project through MCP also discards its clone record", () => }), ); +it.effect.each(["missing", "lookup"] as const)( + "classifies %s project launch failures", + (launchFailure) => + Effect.gen(function* () { + const { projectId, dependencies } = clientLaunchHarness({ + runtimeModeCeiling: "full-access", + launched: [], + launchFailure, + }); + const toolkit = yield* ProjectToolkit.pipe( + Effect.provide( + McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer).pipe( + Layer.provide(dependencies), + ), + ), + ); + const result = yield* toolkit + .handle("t3_thread_launch", { projectId, title: "Missing project" }) + .pipe(Stream.unwrap, Stream.runCollect, Effect.provide(dependencies)); + expect(result.at(-1)?.result).toMatchObject({ + code: launchFailure === "missing" ? "invalid_request" : "orchestration_error", + }); + }), +); + it.effect("an agent cannot delete the built-in scratch project, even with force", () => Effect.gen(function* () { const deleted: Array = []; diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index 7a2fb5381227e..c6400b2682790 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.ts @@ -33,6 +33,8 @@ import * as OrchestrationMcp from "../../OrchestratorMcpService.ts"; import { newCommandId, readCaller, resolveProjectId, unavailable } from "../../threadAccess.ts"; import { ProjectToolkit } from "./tools.ts"; +const isProjectNotFoundError = Schema.is(Project.ProjectNotFoundError); + const isAboveCeilingError = Schema.is(OrchestratorRuntimeModeAboveCeilingError); function projectFailure(error: Project.ProjectServiceError) { @@ -303,6 +305,12 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { message: error.message, }); } + if (error._tag === "ThreadLaunchError" && isProjectNotFoundError(error.cause)) { + return new OrchestratorMcpFailure({ + code: "invalid_request", + message: error.cause.message, + }); + } // The administrator's ceiling refuses the mode; say so rather than "unavailable". if (error._tag === "ThreadLaunchError" && isAboveCeilingError(error.cause)) { return new OrchestratorMcpFailure({ diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 53f726bf4628d..bad0e655e2645 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -680,6 +680,8 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { return command.threadId; case "delegated_task.request": case "delegated_task.remote.request": + case "delegated_task.remote.launch": + case "delegated_task.remote.attach": case "delegated_task.remote.complete": case "delegated_task.remote.stop": case "delegated_task.wake-policy": @@ -9964,6 +9966,147 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); }); + /** Launch/attachment need the durable task, not a live parent provider run. */ + const dispatchRemoteDelegatedTaskLaunch = Effect.fn( + "orchestrationV2.dispatch.remoteDelegatedTaskLaunch", + )(function* ( + command: Extract< + OrchestrationV2InternalCommand, + { + readonly type: "delegated_task.remote.launch" | "delegated_task.remote.attach"; + } + >, + events: Ref.Ref>, + ) { + const parentProjection = yield* projectionStore + .getThreadRecords(command.parentThreadId, ["subagents", "turnItems", "nodes"], { + turnItemTypes: ["subagent"], + }) + .pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: command.parentThreadId, cause }), + ), + ); + const task = parentProjection.subagents.find( + (candidate) => + candidate.id === command.taskId && + candidate.origin === "app_owned" && + candidate.remoteChild !== undefined, + ); + if (task?.remoteChild === undefined) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "The remote launch does not belong to this parent task.", + }); + } + const remote = task.remoteChild; + if (command.type === "delegated_task.remote.launch") { + if (remote.threadId !== undefined || remote.launchStarted === true || task.result !== null) + return; + if (remote.launch === undefined) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "The pending remote task has no launch inputs.", + }); + } + } else { + if ( + remote.environmentId !== command.environmentId || + (remote.threadId !== undefined && remote.threadId !== command.threadId) || + (remote.runId !== undefined && remote.runId !== command.runId) || + (remote.threadId !== undefined && remote.runId === undefined && command.runId !== null) + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "The accepted launch does not match this remote task.", + }); + } + if (remote.threadId !== undefined) return; + if (remote.launchStarted !== true) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "The remote task's launch has not been started.", + }); + } + } + const now = yield* DateTime.now; + const remoteChild = + command.type === "delegated_task.remote.launch" + ? { ...remote, launchStarted: true } + : { + ...remote, + threadId: command.threadId, + ...(command.runId === null ? {} : { runId: command.runId }), + }; + const status = + command.type === "delegated_task.remote.attach" && task.result === null + ? "running" + : task.status; + const emitEvent = emit(events, command); + yield* emitEvent({ + type: "subagent.updated", + threadId: command.parentThreadId, + ...(task.runId === null ? {} : { runId: task.runId }), + nodeId: task.id, + driver: task.driver, + occurredAt: now, + payload: { + ...task, + remoteChild, + status, + ...(command.type === "delegated_task.remote.attach" + ? { + providerInstanceId: command.modelSelection.instanceId, + model: command.modelSelection.model, + } + : {}), + updatedAt: now, + }, + }); + const item = parentProjection.turnItems.find( + (candidate) => candidate.type === "subagent" && candidate.subagentId === task.id, + ); + if (item !== undefined && item.type === "subagent") { + yield* emitEvent({ + type: "turn-item.updated", + threadId: command.parentThreadId, + ...(item.runId === null ? {} : { runId: item.runId }), + ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), + driver: task.driver, + occurredAt: now, + payload: { + ...item, + remoteChild, + status: + command.type === "delegated_task.remote.attach" && task.result === null + ? "running" + : item.status, + updatedAt: now, + }, + }); + } + const node = parentProjection.nodes.find((candidate) => candidate.id === task.id); + if ( + node !== undefined && + command.type === "delegated_task.remote.attach" && + task.result === null + ) { + yield* emitEvent({ + type: "node.updated", + threadId: command.parentThreadId, + ...(node.runId === null ? {} : { runId: node.runId }), + nodeId: node.id, + driver: task.driver, + occurredAt: now, + payload: { ...node, status: "running" }, + }); + } + }); + /** * Completes a task that ran in a linked environment: the parent half of * `finalizeAppOwnedSubagent`, with the result its thread there ended with. @@ -10014,12 +10157,39 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if ( command.stoppedThere === "stopped" && (command.stopCount ?? 0) !== (task.remoteChild?.stopCount ?? 0) - ) + ) { + if ( + command.launchError !== undefined && + command.launchError !== task.remoteChild?.launchError + ) { + const now = yield* DateTime.now; + yield* emit( + events, + command, + )({ + type: "subagent.updated", + threadId: command.parentThreadId, + ...(task.runId === null ? {} : { runId: task.runId }), + nodeId: task.id, + driver: task.driver, + occurredAt: now, + payload: { + ...task, + remoteChild: { ...task.remoteChild!, launchError: command.launchError }, + updatedAt: now, + }, + }); + } return; + } const remoteChild = command.stoppedThere === undefined ? task.remoteChild - : { ...task.remoteChild!, stoppedThere: true }; + : { + ...task.remoteChild!, + stoppedThere: true, + ...(command.launchError === undefined ? {} : { launchError: command.launchError }), + }; // The first result stands; a repeated follower report records nothing, which // the dispatcher refuses. A stop there after the result is still recorded, but // a run that ended by itself does not settle a stop the task here still owes. @@ -13707,7 +13877,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const stopDelegatedTasks = command.stopDelegatedTasks === false || !projection.subagents.some( - (task) => task.origin === "app_owned" && task.childThreadId !== null, + (task) => + task.origin === "app_owned" && + (task.childThreadId !== null || task.remoteChild !== undefined), ) ? Effect.void : Ref.update(effects, (existing) => [...existing, delegatedTasksStopEffect(command)]); @@ -15288,6 +15460,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "delegated_task.remote.request": yield* dispatchRemoteDelegatedTaskRequest(command, events); break; + case "delegated_task.remote.launch": + case "delegated_task.remote.attach": + yield* dispatchRemoteDelegatedTaskLaunch(command, events); + break; case "delegated_task.remote.complete": yield* dispatchRemoteDelegatedTaskComplete(command, events); break; @@ -15441,6 +15617,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command.type === "thread.interrupt.unresponsive" || command.type === "thread.stop" || command.type === "subagent.stop" || + command.type === "delegated_task.remote.launch" || + command.type === "delegated_task.remote.attach" || // Closing a share again, or after it ended, is not an error. command.type === "thread.guest-share.close" || (command.type === "thread.delete" && command.onlyIfUnused === true), diff --git a/apps/server/src/orchestration-v2/RemoteDelegatedTask.test.ts b/apps/server/src/orchestration-v2/RemoteDelegatedTask.test.ts index 292882d0f894b..20a6300ed3a5d 100644 --- a/apps/server/src/orchestration-v2/RemoteDelegatedTask.test.ts +++ b/apps/server/src/orchestration-v2/RemoteDelegatedTask.test.ts @@ -290,6 +290,20 @@ it.layer(layerTest)("a delegated task in a linked environment", (it) => { assert.equal(rejected._tag, "OrchestratorDispatchError"); assert.deepEqual(yield* openFor, [{ parentThreadId: parent.threadId, taskId: task.id }]); } + yield* orchestrator.dispatch({ + ...complete, + commandId: CommandId.make("command:stale-stop-warning"), + stoppedThere: "stopped", + stopCount: 1, + launchError: "The remote launch could not be confirmed; work may still be running.", + }); + const uncertain = (yield* orchestrator.getThreadProjection(parent.threadId)).subagents[0]!; + assert.equal( + uncertain.remoteChild?.launchError, + "The remote launch could not be confirmed; work may still be running.", + ); + assert.equal(uncertain.remoteChild?.stoppedThere, false); + assert.deepEqual(yield* openFor, [{ parentThreadId: parent.threadId, taskId: task.id }]); yield* orchestrator.dispatch({ ...complete, commandId: CommandId.make("command:current-stop:2"), @@ -420,4 +434,116 @@ it.layer(layerTest)("a delegated task in a linked environment", (it) => { assert.equal(offered[0]?.threadId, parent.threadId); }), ); + it.effect.each([null, RunId.make("run:accepted")])( + "attaches an uncertain launch after cancellation once, and rejects another destination (run=%s)", + (acceptedRunId) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const parent = yield* seedLiveParent(`pending-attach:${acceptedRunId ?? "none"}`); + const commandId = (name: string) => + CommandId.make(`command:pending-attach:${acceptedRunId ?? "none"}:${name}`); + const pending = { + environmentId: remoteChild.environmentId, + label: remoteChild.label, + launch: { + requestNamespace: "original-session", + clientRequestId: "original-key", + linkIdentity: "original-grant", + projectId: ProjectId.make("project:remote"), + title: "Remote task", + modelSelection, + delegatedFrom: { + environmentId: EnvironmentId.make("environment:source"), + threadId: parent.threadId, + title: "Parent", + }, + }, + }; + const recorded = yield* orchestrator.dispatch({ + type: "delegated_task.remote.request", + commandId: commandId("request"), + parentThreadId: parent.threadId, + parentRunId: parent.runId, + parentNodeId: parent.rootNodeId, + task: "Run there.", + driver: ProviderDriverKind.make("remote"), + modelSelection, + remoteChild: pending, + }); + const taskEvent = recorded.storedEvents.find( + (stored) => stored.event.type === "subagent.updated", + ); + assert.isTrue(taskEvent?.event.type === "subagent.updated"); + if (taskEvent?.event.type !== "subagent.updated") return; + const taskId = taskEvent.event.payload.id; + const attach = { + type: "delegated_task.remote.attach" as const, + commandId: commandId("attach"), + parentThreadId: parent.threadId, + taskId, + environmentId: pending.environmentId, + threadId: remoteChild.threadId, + runId: acceptedRunId, + modelSelection, + }; + assert.equal( + (yield* orchestrator + .dispatch({ ...attach, commandId: commandId("too-early") }) + .pipe(Effect.flip))._tag, + "OrchestratorDispatchError", + ); + yield* orchestrator.dispatch({ + type: "delegated_task.remote.launch", + commandId: commandId("launch"), + parentThreadId: parent.threadId, + taskId, + }); + yield* orchestrator.dispatch({ + type: "delegated_task.remote.complete", + commandId: commandId("cancel"), + parentThreadId: parent.threadId, + taskId, + status: "cancelled", + result: "Cancelled.", + }); + const offers = yield* Offers; + const offersBefore = yield* Ref.get(offers); + yield* orchestrator.dispatch(attach); + const repeated = yield* orchestrator.dispatch({ + ...attach, + commandId: commandId("repeat"), + }); + assert.equal(repeated.storedEvents.length, 0); + for (const changed of [ + { environmentId: EnvironmentId.make("environment:wrong") }, + { threadId: ThreadId.make("thread:wrong") }, + { runId: RunId.make("run:wrong") }, + ]) { + assert.equal( + (yield* orchestrator + .dispatch({ + ...attach, + ...changed, + commandId: commandId(`wrong:${JSON.stringify(changed)}`), + }) + .pipe(Effect.flip))._tag, + "OrchestratorDispatchError", + ); + } + const projection = yield* orchestrator.getThreadProjection(parent.threadId); + const task = projection.subagents.find((candidate) => candidate.id === taskId); + assert.equal(task?.result, "Cancelled."); + assert.equal(task?.status, "cancelled"); + assert.equal(task?.remoteChild?.threadId, remoteChild.threadId); + assert.equal(task?.remoteChild?.runId, attach.runId ?? undefined); + assert.deepEqual(task?.remoteChild?.launch, pending.launch); + assert.equal( + projection.turnItems.find( + (item) => item.type === "subagent" && item.subagentId === taskId, + )?.status, + "cancelled", + ); + assert.deepEqual(yield* Ref.get(offers), offersBefore); + }), + ); }); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index c3a6ad0a0ad03..6226444b8d375 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -72,6 +72,8 @@ import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; import * as RemoteDelegation from "../peer/RemoteDelegation.ts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +const isProjectNotFoundError = Schema.is(ProjectService.ProjectNotFoundError); + const projectId = ProjectId.make("project:launch-test"); const otherProjectId = ProjectId.make("project:launch-other"); const encodeThreadProjection = Schema.encodeEffect(OrchestrationV2ThreadProjectionJson); @@ -347,6 +349,22 @@ function launchInput(input: { }; } +it.effect("retains a typed cause for a nonexistent project on a root launch", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const result = yield* launches + .launch({ + ...launchInput({ command: "command:missing-project", thread: "thread:missing-project" }), + projectId: ProjectId.make("project:missing"), + }) + .pipe(Effect.result); + assert.ok(Result.isFailure(result)); + assert.ok(result.failure._tag === "ThreadLaunchError"); + assert.ok(isProjectNotFoundError(result.failure.cause)); + }).pipe(Effect.provide(harness.layer)); +}); + it.effect.each( (["new", "existing"] as const).flatMap((target) => (["user", "agent"] as const).map((createdBy) => ({ target, createdBy })), diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index d46f5e2d41514..edc67459796fc 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -1176,7 +1176,13 @@ const make = Effect.gen(function* () { Effect.mapError(mapError(input, "resolve-project")), Effect.flatMap( Option.match({ - onNone: () => Effect.fail(mapError(input, "resolve-project")("Project not found.")), + onNone: () => + Effect.fail( + mapError( + input, + "resolve-project", + )(new ProjectService.ProjectNotFoundError({ projectId: input.projectId })), + ), onSome: Effect.succeed, }), ), diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 611d9a3244f24..3a92ec57c1aa3 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -128,6 +128,8 @@ function existingThreadIdsForCommand( case "delegated_task.completion-delivery.acknowledge": case "delegated_task.completion-delivery.dispose": case "delegated_task.remote.request": + case "delegated_task.remote.launch": + case "delegated_task.remote.attach": case "delegated_task.remote.complete": case "delegated_task.remote.stop": return [command.parentThreadId]; diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index fc52191d32b29..155379e2e3efb 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -2,6 +2,7 @@ import { assert, it } from "@effect/vitest"; import { CommandId, ContextTransferId, + EnvironmentId, EventId, MessageId, NodeId, @@ -611,48 +612,69 @@ it.effect.each(["run.interrupt", "thread.stop"] as const)( }).pipe(Effect.provide(layerTest)), ); -it.effect("a repeated Stop that changes nothing still stops the delegated tasks", () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const projections = yield* ProjectionStore.ProjectionStoreV2; - const sql = yield* SqlClient.SqlClient; - const parentThreadId = ThreadId.make("thread:stop-again"); - yield* createWatchingThread(parentThreadId, 1); - yield* send(parentThreadId, "first", "start_immediately"); - yield* delegate(parentThreadId, "stop-again child"); - const now = yield* DateTime.now; - const run = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!; - yield* projections.apply({ - id: EventId.make("event:stop-again:run-completed"), - type: "run.updated", - threadId: parentThreadId, - runId: run.id, - occurredAt: now, - payload: { ...run, status: "completed", completedAt: now }, - }); - yield* orchestrator.dispatch({ - type: "thread.stop", - commandId: CommandId.make("stop-again:first"), - threadId: parentThreadId, - }); +it.effect.each(["local", "remote-pending", "remote-attached"] as const)( + "a repeated Stop that changes nothing still stops %s delegated tasks", + (kind) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const parentThreadId = ThreadId.make("thread:stop-again"); + yield* createWatchingThread(parentThreadId, 1); + yield* send(parentThreadId, "first", "start_immediately"); + const now = yield* DateTime.now; + const run = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!; + if (kind === "local") { + yield* delegate(parentThreadId, "stop-again child"); + } else { + yield* orchestrator.dispatch({ + type: "delegated_task.remote.request", + commandId: CommandId.make("stop-again:remote-delegate"), + parentThreadId, + parentRunId: run.id, + parentNodeId: run.rootNodeId!, + task: "stop-again child", + driver: ProviderDriverKind.make("remote"), + modelSelection: run.modelSelection, + remoteChild: { + environmentId: EnvironmentId.make("environment:box"), + label: "Box", + ...(kind === "remote-attached" ? { threadId: ThreadId.make("thread:on-the-box") } : {}), + }, + completionWake: "always", + }); + } + yield* projections.apply({ + id: EventId.make("event:stop-again:run-completed"), + type: "run.updated", + threadId: parentThreadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status: "completed", completedAt: now }, + }); + yield* orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make("stop-again:first"), + threadId: parentThreadId, + }); - // The first Stop already disposed the delivery and ended the watch, so this one writes no - // events, like Stop on a thread an older server stopped without stopping its children. - const again = yield* orchestrator.dispatch({ - type: "thread.stop", - commandId: CommandId.make("stop-again:second"), - threadId: parentThreadId, - }); - assert.lengthOf(again.storedEvents, 0); - const walks = yield* sql<{ readonly command_id: string }>` + // The first Stop already disposed the delivery and ended the watch, so this one writes no + // events, like Stop on a thread an older server stopped without stopping its children. + const again = yield* orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make("stop-again:second"), + threadId: parentThreadId, + }); + assert.lengthOf(again.storedEvents, 0); + const walks = yield* sql<{ readonly command_id: string }>` SELECT command_id FROM orchestration_v2_effect_outbox WHERE effect_type = 'delegated-tasks.stop' `; - assert.deepEqual( - walks.map((row) => row.command_id), - ["stop-again:first", "stop-again:second"], - ); - }).pipe(Effect.provide(layerTest)), + assert.deepEqual( + walks.map((row) => row.command_id), + ["stop-again:first", "stop-again:second"], + ); + }).pipe(Effect.provide(layerTest)), ); it.effect.each(["thread.stop", "run.interrupt"] as const)( diff --git a/apps/server/src/peer/PeerForwarding.ts b/apps/server/src/peer/PeerForwarding.ts index 080f701c45cb2..abbf6fd6aefa0 100644 --- a/apps/server/src/peer/PeerForwarding.ts +++ b/apps/server/src/peer/PeerForwarding.ts @@ -46,13 +46,14 @@ export class PeerForwarding extends Context.Service< { /** * Runs `tool` in the linked environment `environmentId` with `params`. A - * `t3_thread_wait` goes in pieces short enough to survive any route. + * `t3_thread_wait` returns before MCP client and route timeouts. */ readonly call: ( scope: Scope, tool: T, environmentId: EnvironmentId, params: Tool.Parameters, + options?: { readonly expectedLinkIdentity: string }, ) => Effect.Effect, OrchestratorMcpFailure>; /** `t3_thread_read` there. */ readonly readThread: ( @@ -71,6 +72,7 @@ export class PeerForwarding extends Context.Service< scope: Scope, environmentId: EnvironmentId, input: OrchestratorMcpThreadWaitInput, + options?: { readonly expectedLinkIdentity: string }, ) => Effect.Effect; /** The environments this one links to, as an agent sees them. */ readonly links: ( @@ -155,6 +157,7 @@ const make = Effect.gen(function* () { success: Schema.Top, environmentId: EnvironmentId, params: unknown, + options?: { readonly expectedLinkIdentity: string }, ) { const { caller, limits } = yield* forwardingCaller(scope); const encoded = (yield* Schema.encodeUnknownEffect(parameters as Schema.Encoder)( @@ -188,6 +191,7 @@ const make = Effect.gen(function* () { limits, ...(sender === undefined ? {} : { sender }), success: success as Schema.Decoder, + ...(options === undefined ? {} : options), }); }); @@ -215,7 +219,12 @@ const make = Effect.gen(function* () { input, ); - const waitForThread: PeerForwarding["Service"]["waitForThread"] = (scope, environmentId, input) => + const waitForThread: PeerForwarding["Service"]["waitForThread"] = ( + scope, + environmentId, + input, + options, + ) => forward( scope, "t3_thread_wait", @@ -229,6 +238,7 @@ const make = Effect.gen(function* () { Math.max(1, input.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS), ), }, + options, ); // While the beta is off, agents start nothing over links. Tasks already running there keep @@ -237,7 +247,7 @@ const make = Effect.gen(function* () { Effect.flatMap((enabled) => (enabled ? Effect.void : Effect.fail(linkedEnvironmentsOff))), ); - const call: PeerForwarding["Service"]["call"] = (scope, tool, environmentId, params) => { + const call: PeerForwarding["Service"]["call"] = (scope, tool, environmentId, params, options) => { const forwarded: Effect.Effect< Tool.Success, OrchestratorMcpFailure @@ -246,8 +256,17 @@ const make = Effect.gen(function* () { scope, environmentId, params as OrchestratorMcpThreadWaitInput, + options, ) as Effect.Effect, OrchestratorMcpFailure>) - : forward(scope, tool.name, tool.parametersSchema, tool.successSchema, environmentId, params); + : forward( + scope, + tool.name, + tool.parametersSchema, + tool.successSchema, + environmentId, + params, + options, + ); return Effect.andThen(requireEnabled, forwarded); }; diff --git a/apps/server/src/peer/PeerLinks.ts b/apps/server/src/peer/PeerLinks.ts index 6fddf407cf4d4..4c7167662170d 100644 --- a/apps/server/src/peer/PeerLinks.ts +++ b/apps/server/src/peer/PeerLinks.ts @@ -1,3 +1,5 @@ +// @effect-diagnostics-next-line nodeBuiltinImport:off -- The pinned Effect Crypto service has no digest API. +import * as NodeCrypto from "node:crypto"; import { AuthMcpClientAccess, AuthMcpRegistrationError, @@ -56,6 +58,11 @@ export const probeTimeout = (origin: string) => /** How long a listing waits on a link's addresses; a real call still gives a waking peer `probeTimeout`. */ const LIST_PROBE_TIMEOUT = Duration.seconds(3); +/** Server-only identity of the OAuth grant generation; contains no credential. */ +export interface StoredPeerLink extends PeerLink { + readonly launchIdentity: string; +} + /** Node's TLS verification code in a failed request's cause chain, if any. */ export const certificateErrorCode = (error: unknown) => { let current = error; @@ -75,7 +82,7 @@ export const certificateErrorCode = (error: unknown) => { /** A usable link: where to reach the peer and the token to send there. */ export interface ResolvedPeerLink { - readonly link: PeerLink; + readonly link: StoredPeerLink; /** The first address that answered as this peer. */ readonly url: string; readonly token: string; @@ -100,7 +107,7 @@ export class PeerLinks extends Context.Service< /** The stored link, without asking the peer whether it answers. */ readonly get: ( environmentId: EnvironmentId, - ) => Effect.Effect, PeerLinkError>; + ) => Effect.Effect, PeerLinkError>; /** Forgets the link here. The peer still lists the session until it is revoked there. */ readonly unlink: (environmentId: EnvironmentId) => Effect.Effect; /** An address that answers as the linked peer, with its token, ready to call. */ @@ -239,6 +246,11 @@ const make = Effect.gen(function* () { Effect.orDie, ); + // Relinking replaces the destination OAuth namespace. A pending launch may + // replay only through its original grant, even when its environment is unchanged. + const launchIdentity = (secretName: string) => + Effect.sync(() => NodeCrypto.createHash("sha256").update(secretName).digest("hex")); + // Only the origin `peerOrigin` checked may receive a pairing code or token: // a redirect could forward them anywhere, including plain http. const peerHttpClient = HttpClient.transform(httpClient, (response) => @@ -279,6 +291,7 @@ const make = Effect.gen(function* () { const toLink = Effect.fn("PeerLinks.toLink")(function* (row: PeerLinkRow) { return { environmentId: EnvironmentId.make(row.environment_id), + launchIdentity: yield* launchIdentity(row.secret_name), label: row.label, urls: yield* decodeUrls(row.urls_json), access: yield* decodeAccess(row.access), @@ -288,7 +301,7 @@ const make = Effect.gen(function* () { expiresAt: DateTime.makeUnsafe(row.fenced === 1 ? row.expires_at : row.linked_at), lastReachedAt: row.last_reached_at === null ? null : DateTime.makeUnsafe(row.last_reached_at), lastError: row.last_error, - } satisfies PeerLink; + } satisfies StoredPeerLink; }, Effect.catch(failStorage)); const readRows = sql` @@ -578,7 +591,7 @@ const make = Effect.gen(function* () { }); /** A peer's label follows its machine name, so the link takes the one its own address answers with. */ - const withLabel = (link: PeerLink, label: string) => + const withLabel = (link: Link, label: string) => label === link.label ? Effect.succeed(link) : sql` diff --git a/apps/server/src/peer/PeerMcpClient.test.ts b/apps/server/src/peer/PeerMcpClient.test.ts index 8fd65e7e2d796..93856459926ad 100644 --- a/apps/server/src/peer/PeerMcpClient.test.ts +++ b/apps/server/src/peer/PeerMcpClient.test.ts @@ -16,6 +16,7 @@ import * as Layer from "effect/Layer"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import * as HttpClientError from "effect/http/HttpClientError"; import { Tool, Toolkit } from "effect/ai"; import * as TestClock from "effect/testing/TestClock"; import { HttpClient, HttpClientResponse } from "effect/http"; @@ -60,6 +61,7 @@ const probeThrough = Effect.fn(function* (http: HttpClient.HttpClient) { Effect.succeed({ link: { environmentId: box.environmentId, + launchIdentity: "fixture-grant", label: box.label, urls: ["https://peer.example"], access: "auto", @@ -242,22 +244,58 @@ it.effect("keeps a link healthy when the peer refuses a call", () => ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); -it.effect("renews a missing peer session once and reuses its replacement", () => +it.effect.each([ + { name: "empty result", status: 200, body: "{}", uncertain: false }, + { name: "HTTP refusal", status: 403, body: "", uncertain: false }, + { name: "HTTP failure", status: 503, body: "", uncertain: false }, + { + name: "tool refusal", + status: 200, + body: JSON.stringify({ + result: { + isError: true, + content: [ + { + type: "text", + text: JSON.stringify({ + code: "orchestration_error", + message: "The operation could not be completed.", + }), + }, + ], + }, + }), + uncertain: false, + }, + { + name: "version mismatch", + status: 200, + body: JSON.stringify({ result: { content: [{ type: "text", text: "{}" }] } }), + uncertain: false, + }, + { name: "unreadable answer", status: 200, body: "not JSON", uncertain: true }, + { name: "lost response", status: 200, body: "", uncertain: true }, +])("renews a missing peer session and distinguishes $name from an uncertain call", (answer) => Effect.gen(function* () { const sessionHeaders: Array = []; - const statuses = [200, 202, 404, 200, 202, 200, 200, 200]; + const statuses = [200, 202, 404, 200, 202, 200, 200, answer.status]; const outcomes: Array = []; const http = HttpClient.make((request) => - Effect.sync(() => { + Effect.gen(function* () { const index = sessionHeaders.length; sessionHeaders.push(request.headers["mcp-session-id"]); + if (index === 7 && answer.name === "lost response") { + return yield* new HttpClientError.HttpClientError({ + reason: new HttpClientError.TransportError({ request }), + }); + } const status = statuses[index]; if (status === undefined) throw new Error("Unexpected retry"); return HttpClientResponse.fromWeb( request, new Response( index === 7 - ? "{}" + ? answer.body : index >= 5 ? JSON.stringify({ result: { @@ -289,6 +327,7 @@ it.effect("renews a missing peer session once and reuses its replacement", () => Effect.succeed({ link: { environmentId: box.environmentId, + launchIdentity: "fixture-grant", label: box.label, urls: ["https://peer.example"], access: "auto", @@ -328,7 +367,16 @@ it.effect("renews a missing peer session once and reuses its replacement", () => success: Result, }) .pipe(Effect.flip); - expect(outcomes).toEqual([null, null, empty.message]); + expect(PeerMcpClient.isUncertainCall(empty)).toBe(answer.uncertain); + expect(outcomes).toEqual([ + null, + null, + answer.name === "tool refusal" + ? null + : answer.name === "lost response" + ? "Transport error (POST https://peer.example/mcp)" + : empty.message, + ]); expect(sessionHeaders).toEqual([ undefined, "old", diff --git a/apps/server/src/peer/PeerMcpClient.ts b/apps/server/src/peer/PeerMcpClient.ts index 8c438ad702b53..e3ae268d8aa3d 100644 --- a/apps/server/src/peer/PeerMcpClient.ts +++ b/apps/server/src/peer/PeerMcpClient.ts @@ -19,6 +19,15 @@ import * as HttpClientRequest from "effect/http/HttpClientRequest"; import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; import * as PeerLinks from "./PeerLinks.ts"; +const uncertainCall = Symbol("PeerMcpClient.UncertainCall"); + +/** No usable answer arrived, so a keyed launch may already have been accepted. */ +export class UncertainCall extends OrchestratorMcpFailure { + readonly [uncertainCall] = true; +} + +export const isUncertainCall = (error: OrchestratorMcpFailure) => uncertainCall in error; + /** The only MCP protocol version a T3 Code `/mcp` registers. */ const PROTOCOL_VERSION = "2025-06-18"; @@ -48,6 +57,8 @@ export class PeerMcpClient extends Context.Service< /** The thread here the call is made for, sent as `T3-Sender-Thread`. */ readonly sender?: OrchestrationV2LinkedSender; readonly success: S; + /** Refuse a launch replay through a replacement OAuth grant. */ + readonly expectedLinkIdentity?: string; }) => Effect.Effect; } >()("t3/peer/PeerMcpClient") {} @@ -73,7 +84,7 @@ const decodeFailure = Schema.decodeUnknownOption( /** A peer link problem, as the agent that asked hears it. */ const fromLinkError = (error: PeerLinkError) => - new OrchestratorMcpFailure({ + new (error.reason === "unreachable" ? UncertainCall : OrchestratorMcpFailure)({ code: error.reason === "unknown_link" ? "invalid_request" @@ -90,7 +101,7 @@ const linkedEnvironmentsOffThere = new OrchestratorMcpFailure({ "The linked environment has Linked environments turned off. Turn it on in its Settings → Labs.", }); -const stoppedAnswering = new OrchestratorMcpFailure({ +const stoppedAnswering = new UncertainCall({ code: "orchestration_error", message: "The linked environment stopped answering.", }); @@ -169,6 +180,16 @@ const make = Effect.gen(function* () { const resolved = yield* links .resolve(input.environmentId) .pipe(Effect.mapError(fromLinkError)); + if ( + input.expectedLinkIdentity !== undefined && + resolved.link.launchIdentity !== input.expectedLinkIdentity + ) { + return yield* new OrchestratorMcpFailure({ + code: "capability_denied", + message: + "The destination was linked again. Its original launch identity cannot be replayed through the replacement link.", + }); + } const limitHeader = { [McpInvocationContext.MODE_LIMIT_HEADER]: `${input.limits.runtimeMode}/${input.limits.interactionMode}`, }; @@ -235,7 +256,7 @@ const make = Effect.gen(function* () { ).pipe( Effect.mapError( () => - new OrchestratorMcpFailure({ + new UncertainCall({ code: "orchestration_error", message: "The linked environment answered in an unexpected shape.", }), diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index eeef7bf05a36e..7b1bace2462f3 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -65,6 +65,7 @@ import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts import * as SourceControlRepositoryService from "../sourceControl/SourceControlRepositoryService.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as PeerForwarding from "./PeerForwarding.ts"; +import * as PeerMcpClient from "./PeerMcpClient.ts"; import * as PeerLinks from "./PeerLinks.ts"; import { descriptorOf, layerLinkingEnvironment, linkTo, servePeer } from "./PeerLinks.testkit.ts"; import * as RemoteDelegation from "./RemoteDelegation.ts"; @@ -90,6 +91,10 @@ type LaunchedWith = Pick< /** The thread the box launches for the task, which the test finishes. */ interface BoxThread { readonly launched: Ref.Ref; + readonly launchIds: Ref.Ref>; + readonly holdLaunch: Ref.Ref; + readonly launchAccepted: Deferred.Deferred; + readonly releaseLaunch: Deferred.Deferred; /** The link session that launched it, as the box stamps it. */ readonly linkOrigin: Ref.Ref<{ readonly sessionId: string; readonly label: string } | null>; /** The parent the launch names, so the thread there can link back to it. */ @@ -125,6 +130,7 @@ interface BoxThread { readonly stops: Ref.Ref>; /** The launch starts no run, as one recorded before its run was kept. */ readonly runless: Ref.Ref; + readonly launchError: Ref.Ref; /** Completed on the first interrupt there. */ readonly interruptedOnce: Deferred.Deferred; readonly finished: Deferred.Deferred; @@ -146,6 +152,10 @@ interface BoxThread { const makeBoxThread = Effect.gen(function* () { return { launched: yield* Ref.make(null), + launchIds: yield* Ref.make>([]), + holdLaunch: yield* Ref.make(false), + launchAccepted: yield* Deferred.make(), + releaseLaunch: yield* Deferred.make(), linkOrigin: yield* Ref.make<{ readonly sessionId: string; readonly label: string } | null>( null, ), @@ -168,6 +178,7 @@ const makeBoxThread = Effect.gen(function* () { interrupted: yield* Ref.make(0), stops: yield* Ref.make>([]), runless: yield* Ref.make(false), + launchError: yield* Ref.make(null), interruptedOnce: yield* Deferred.make(), finished: yield* Deferred.make(), laterFinished: yield* Deferred.make(), @@ -502,7 +513,9 @@ const boxToolkitLayer = (thread: BoxThread) => { }); const layerBoxLaunches = Layer.mock(ThreadLaunch.ThreadLaunchService)({ launch: (input) => - Ref.get(thread.runless).pipe( + Ref.get(thread.launchError).pipe( + Effect.flatMap((error) => (error === null ? Effect.void : Effect.fail(error))), + Effect.andThen(Ref.get(thread.runless)), Effect.flatMap((runless) => Ref.set(thread.launched, input.threadId!).pipe( Effect.andThen(Ref.set(thread.linkOrigin, input.linkOrigin ?? null)), @@ -515,6 +528,15 @@ const boxToolkitLayer = (thread: BoxThread) => { interactionMode: input.interactionMode, }), ), + Effect.andThen(Ref.update(thread.launchIds, (ids) => [...ids, input.threadId!])), + Effect.andThen(Deferred.succeed(thread.launchAccepted, undefined)), + Effect.andThen( + Ref.get(thread.holdLaunch).pipe( + Effect.flatMap((held) => + held ? Deferred.await(thread.releaseLaunch) : Effect.void, + ), + ), + ), Effect.andThen(Effect.flatten(Ref.get(thread.afterLaunch))), Effect.as({ threadId: input.threadId, @@ -916,6 +938,105 @@ const taskResult = (laptopEnv: Effect.Success>, ta ), ); +it.effect("returns a definite launch refusal instead of an indefinitely launching task", () => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + yield* Ref.set( + thread.launchError, + new ThreadLaunch.ThreadLaunchError({ + operation: "create-thread", + commandId: CommandId.make("command:refused-launch"), + projectId: boxProject, + cause: new Error("The destination cannot create the thread."), + }), + ); + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const a = yield* makeLaptop(database, offers); + yield* linkTo(a.links, b, "full-access"); + yield* seedParent(a); + const refused = yield* a.call("delegate_task", { + task: "Run the checks there.", + target: { + environmentId: box.environmentId, + projectId: boxProject, + providerInstanceId: instanceId, + model: "gpt-5.4", + }, + mode: "async", + clientRequestId: "refused-launch", + }); + expect(refused.isError).toBe(true); + expect(JSON.stringify(refused.content)).toContain("The operation could not be completed."); + const task = (yield* a.orchestrator.getThreadProjection(parentThreadId)).subagents[0]!; + expect(task.status).toBe("failed"); + expect(task.result).toBe("The operation could not be completed."); + expect(task.completionDelivery?.state).toBe("disposed"); + expect(task.remoteChild?.stoppedThere).toBe(true); + expect(task.remoteChild?.launchError).toBeUndefined(); + expect(yield* Ref.get(thread.launchIds)).toEqual([]); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect("returns the recorded task when an accepted launch loses its response", () => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const linking = yield* layerLinkingEnvironment(laptop).pipe(Layer.build); + const client = Context.get(linking, PeerMcpClient.PeerMcpClient); + let lost = false; + const a = yield* makeLaptop( + database, + offers, + Context.add( + linking, + PeerMcpClient.PeerMcpClient, + PeerMcpClient.PeerMcpClient.of({ + call: (input) => + client.call(input).pipe( + Effect.flatMap((result) => { + if (input.tool !== "t3_thread_launch" || lost) return Effect.succeed(result); + lost = true; + return Effect.fail( + new PeerMcpClient.UncertainCall({ + code: "orchestration_error", + message: "The linked environment stopped answering.", + }), + ); + }), + ), + }), + ), + ); + yield* linkTo(a.links, b, "full-access"); + yield* seedParent(a); + const delegated = yield* delegateToBox(a); + expect(delegated.taskId).toBeDefined(); + const recorded = yield* a.orchestrator.getThreadProjection(parentThreadId); + expect(recorded.subagents).toHaveLength(1); + expect(recorded.subagents[0]?.id).toBe(delegated.taskId); + yield* Ref.set(thread.reply, "Recovered the accepted child."); + yield* Ref.set(thread.status, "completed"); + yield* Deferred.succeed(thread.finished, undefined); + expect(yield* taskResult(a, delegated.taskId)).toMatchObject({ + status: "completed", + result: "Recovered the accepted child.", + }); + expect(new Set(yield* Ref.get(thread.launchIds)).size).toBe(1); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + it.effect("a task delegated to the box wakes the laptop's parent when it ends there", () => Effect.scoped( Effect.gen(function* () { @@ -1242,6 +1363,12 @@ it.effect("a task named only by environment inherits the parent's model, within runtimeMode: "auto-accept-edits", interactionMode: "default", }); + // Inherited modes stay omitted, so the destination can step down to what its provider offers. + const stored = (yield* a.orchestrator.getThreadProjection(parentThreadId)).subagents[0] + ?.remoteChild?.launch; + expect(stored).toBeDefined(); + expect(stored).not.toHaveProperty("runtimeMode"); + expect(stored).not.toHaveProperty("interactionMode"); // Asking for more than the link allows is still refused there. const broader = yield* a.call("delegate_task", { @@ -1251,6 +1378,13 @@ it.effect("a task named only by environment inherits the parent's model, within clientRequestId: "broader", }); expect(broader.isError).toBe(true); + // Nothing was accepted there, so the task says so plainly and owes no wake. + const refused = (yield* a.orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.status === "failed", + ); + expect(refused?.remoteChild?.launchError).toBeUndefined(); + expect(refused?.result).not.toContain("could not be confirmed"); + expect(refused?.completionDelivery?.state).toBe("disposed"); }), ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); @@ -1342,6 +1476,371 @@ it.effect("follows an open task again after the laptop restarts", () => ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); +it.effect( + "recovers an accepted launch after restart without its response or an active parent run", + () => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + yield* Ref.set(thread.holdLaunch, true); + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const linking = yield* layerLinkingEnvironment(laptop).pipe(Layer.build); + const pending = yield* Effect.scoped( + Effect.gen(function* () { + const before = yield* makeLaptop(database, offers, linking); + yield* linkTo(before.links, b, "full-access"); + yield* seedParent(before); + yield* delegateToBox(before).pipe(Effect.forkScoped); + yield* Deferred.await(thread.launchAccepted); + const projection = yield* before.orchestrator.getThreadProjection(parentThreadId); + // B accepted, but its response has not reached A. A must already own the task. + expect(projection.subagents).toHaveLength(1); + const task = projection.subagents[0]!; + expect(task.remoteChild?.threadId).toBeUndefined(); + const run = projection.runs.find((candidate) => candidate.id === parentRunId)!; + const now = yield* DateTime.now; + yield* before.sink.write({ + commandId: CommandId.make("command:restart-during-launch"), + events: [ + { + id: EventId.make("event:restart-during-launch"), + type: "run.updated", + threadId: parentThreadId, + runId: parentRunId, + nodeId: parentRootNode, + driver, + providerInstanceId: instanceId, + occurredAt: now, + payload: { ...run, status: "interrupted", completedAt: now }, + }, + ], + }); + return task; + }), + ); + const acceptedThreadId = yield* Ref.get(thread.launched); + yield* Ref.set(thread.holdLaunch, false); + yield* Deferred.succeed(thread.releaseLaunch, undefined); + yield* Ref.set(thread.reply, "Recovered the original launch."); + yield* Ref.set(thread.status, "completed"); + yield* Deferred.succeed(thread.finished, undefined); + + const after = yield* makeLaptop(database, offers, linking); + yield* after.remote.start(); + const task = yield* taskResult(after, pending.id); + expect(task).toMatchObject({ + id: pending.id, + status: "completed", + result: "Recovered the original launch.", + remoteChild: { threadId: acceptedThreadId, runId: boxRunId }, + }); + expect(new Set(yield* Ref.get(thread.launchIds)).size).toBe(1); + expect( + (yield* after.orchestrator.getThreadProjection(parentThreadId)).subagents, + ).toHaveLength(1); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +/** A committed source intent, before any launch marker or network request. */ +const recordPending = (a: Effect.Success>) => + Effect.gen(function* () { + const link = Option.getOrThrow(yield* a.links.get(box.environmentId)); + const recorded = yield* a.orchestrator.dispatch({ + type: "delegated_task.remote.request", + commandId: CommandId.make("command:pending-launch"), + parentThreadId, + parentRunId, + parentNodeId: parentRootNode, + task: "Run the full suite on the box.", + driver: ProviderDriverKind.make("remote"), + modelSelection, + remoteChild: { + environmentId: box.environmentId, + label: "Box", + launch: { + requestNamespace: parentScope.requestNamespace, + clientRequestId: `delegate:${parentThreadId}:suite-on-box`, + linkIdentity: link.launchIdentity, + projectId: boxProject, + title: "Remote suite", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + delegatedFrom: { + environmentId: laptop.environmentId, + threadId: parentThreadId, + title: "Laptop parent", + }, + }, + }, + completionWake: "always", + }); + const event = recorded.storedEvents.find((stored) => stored.event.type === "subagent.updated"); + if (event?.event.type !== "subagent.updated") return yield* Effect.die("No pending task event"); + return event.event.payload.id; + }); + +/** A specific durable milestone, including attachment and an acknowledged stop. */ +const stoppedTask = (a: Effect.Success>, taskId: NodeId) => + a.orchestrator + .streamStoredEventsFrom({ + threadId: parentThreadId, + afterSequence: 0, + eventType: "subagent.updated", + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.id === taskId && + stored.event.payload.remoteChild?.stoppedThere === true, + ), + Stream.runHead, + ); + +it.effect.each([ + { cancelled: false, stops: 0 }, + { cancelled: true, stops: 0 }, + { cancelled: true, stops: 2 }, +])( + "recovers a never-sent intent after restart (cancelled=$cancelled, stops=$stops)", + ({ cancelled, stops }) => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const linking = yield* layerLinkingEnvironment(laptop).pipe(Layer.build); + const taskId = yield* Effect.scoped( + Effect.gen(function* () { + const before = yield* makeLaptop(database, offers, linking); + yield* linkTo(before.links, b, "full-access"); + yield* seedParent(before); + const id = yield* recordPending(before); + if (cancelled) { + // Commit cancellation while the source is down, before a follower can send. + yield* before.orchestrator.dispatch({ + type: "delegated_task.remote.complete", + commandId: CommandId.make("command:cancel-before-send"), + parentThreadId, + taskId: id, + status: "cancelled", + result: "Cancelled before send.", + }); + for (let generation = 1; generation <= stops; generation += 1) { + yield* before.orchestrator.dispatch({ + type: "delegated_task.remote.stop", + commandId: CommandId.make(`command:stop-before-send:${generation}`), + parentThreadId, + taskId: id, + }); + } + } + return id; + }), + ); + yield* Ref.set(thread.reply, "Recovered the unsent intent."); + yield* Ref.set(thread.status, "completed"); + yield* Deferred.succeed(thread.finished, undefined); + const after = yield* makeLaptop(database, offers, linking); + yield* after.remote.start(); + if (cancelled) { + yield* stoppedTask(after, taskId); + expect(yield* Ref.get(thread.launchIds)).toEqual([]); + expect(yield* taskResult(after, taskId)).toMatchObject({ + status: "cancelled", + result: "Cancelled before send.", + }); + } else { + expect(yield* taskResult(after, taskId)).toMatchObject({ + status: "completed", + result: "Recovered the unsent intent.", + }); + expect(yield* Ref.get(thread.launchIds)).toHaveLength(1); + } + expect(yield* after.projections.getOpenRemoteDelegatedTasks).toEqual([]); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect.each(["lowered parent", "revoked link"] as const)( + "refuses a pending launch after restart with a %s", + (restriction) => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const linking = yield* layerLinkingEnvironment(laptop).pipe(Layer.build); + const taskId = yield* Effect.scoped( + Effect.gen(function* () { + const before = yield* makeLaptop(database, offers, linking); + yield* linkTo(before.links, b, "full-access"); + yield* seedParent(before); + const id = yield* recordPending(before); + if (restriction === "lowered parent") { + yield* before.orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:lower-before-recovery"), + threadId: parentThreadId, + runtimeMode: "approval-required", + }); + } else { + const [session] = yield* b.linkedSessions; + yield* b.auth.revokeSession(session!.sessionId); + } + return id; + }), + ); + const after = yield* makeLaptop(database, offers, linking); + yield* after.remote.start(); + expect(yield* taskResult(after, taskId)).toMatchObject({ status: "failed" }); + expect(yield* Ref.get(thread.launchIds)).toEqual([]); + expect(yield* after.projections.getOpenRemoteDelegatedTasks).toEqual([]); + expect(yield* b.linkedSessions).toHaveLength(restriction === "revoked link" ? 0 : 1); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect.each([ + { cancel: false, restriction: "replaced link" }, + { cancel: true, restriction: "revoked link" }, +] as const)( + "keeps acceptance uncertainty visible with $restriction (cancel=$cancel)", + ({ cancel, restriction }) => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + yield* Ref.set(thread.holdLaunch, true); + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const linking = yield* layerLinkingEnvironment(laptop).pipe(Layer.build); + const taskId = yield* Effect.scoped( + Effect.gen(function* () { + const before = yield* makeLaptop(database, offers, linking); + yield* linkTo(before.links, b, "full-access"); + yield* seedParent(before); + yield* delegateToBox(before).pipe(Effect.forkScoped); + yield* Deferred.await(thread.launchAccepted); + const task = (yield* before.orchestrator.getThreadProjection(parentThreadId)) + .subagents[0]!; + if (cancel) { + expect((yield* before.call("task_cancel", { taskId: task.id })).isError).toBe(false); + } + if (restriction === "revoked link") { + const [session] = yield* b.linkedSessions; + yield* b.auth.revokeSession(session!.sessionId); + } + return task.id; + }), + ); + yield* Ref.set(thread.holdLaunch, false); + yield* Deferred.succeed(thread.releaseLaunch, undefined); + const after = yield* makeLaptop(database, offers, linking); + if (restriction === "replaced link") { + yield* linkTo(after.links, b, "full-access"); + yield* Ref.set(thread.status, "completed"); + yield* Deferred.succeed(thread.finished, undefined); + } + yield* after.remote.start(); + if (cancel) yield* stoppedTask(after, taskId); + yield* taskResult(after, taskId); + const task = (yield* after.orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (entry) => entry.id === taskId, + ); + expect(new Set(yield* Ref.get(thread.launchIds)).size).toBe(1); + expect(task).toMatchObject({ status: cancel ? "cancelled" : "failed" }); + if (cancel) expect(task?.result).toBe("Cancelled."); + expect(task?.remoteChild?.launchError).toContain( + "If it was already accepted, work may still be running on Box.", + ); + // Refusal prevents confirmation/Stop; it does not mean B's accepted work ended. + expect(yield* Ref.get(thread.status)).toBe( + restriction === "replaced link" ? "completed" : "running", + ); + expect(yield* Ref.get(thread.interrupted)).toBe(0); + if (!cancel) expect(yield* Ref.get(thread.launchIds)).toHaveLength(1); + expect(yield* b.linkedSessions).toHaveLength( + restriction === "revoked link" ? 0 : restriction === "replaced link" ? 2 : 1, + ); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect.each(["cancel", "parent Stop"] as const)( + "reconciles %s after acceptance without a launch response", + (action) => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + yield* Ref.set(thread.holdLaunch, true); + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const linking = yield* layerLinkingEnvironment(laptop).pipe(Layer.build); + const taskId = yield* Effect.scoped( + Effect.gen(function* () { + const before = yield* makeLaptop(database, offers, linking); + yield* linkTo(before.links, b, "full-access"); + yield* seedParent(before); + yield* delegateToBox(before).pipe(Effect.forkScoped); + yield* Deferred.await(thread.launchAccepted); + const task = (yield* before.orchestrator.getThreadProjection(parentThreadId)) + .subagents[0]!; + // The accepted child is unreachable until after the source restarts. + yield* Ref.set(b.descriptor, descriptorOf("environment-elsewhere", "Elsewhere")); + if (action === "cancel") { + const result = yield* before.call("task_cancel", { taskId: task.id }); + expect(result.isError).toBe(false); + } else { + yield* before.threads.stopDelegatedTasks({ + threadId: parentThreadId, + commandId: CommandId.make("command:parent-stop-during-launch"), + }); + } + expect(yield* taskResult(before, task.id)).toMatchObject({ + status: action === "cancel" ? "cancelled" : "interrupted", + }); + return task.id; + }), + ); + const acceptedThreadId = yield* Ref.get(thread.launched); + yield* Ref.set(b.descriptor, box); + yield* Ref.set(thread.holdLaunch, false); + yield* Deferred.succeed(thread.releaseLaunch, undefined); + const after = yield* makeLaptop(database, offers, linking); + yield* after.remote.start(); + yield* stoppedTask(after, taskId); + const task = (yield* after.orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (entry) => entry.id === taskId, + ); + expect(task).toMatchObject({ + status: action === "cancel" ? "cancelled" : "interrupted", + remoteChild: { threadId: acceptedThreadId, runId: boxRunId, stoppedThere: true }, + }); + expect(new Set(yield* Ref.get(thread.launchIds)).size).toBe(1); + expect(yield* Ref.get(thread.stops)).toEqual(["thread.stop", "delegated-tasks"]); + expect(yield* after.projections.getOpenRemoteDelegatedTasks).toEqual([]); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + it.effect("fails the task when the box revokes the link, and cancel interrupts it there", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index 932b012969e36..4adc941f82307 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -49,6 +49,7 @@ import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolv import { forkParked } from "../serverActivation.ts"; import * as PeerForwarding from "./PeerForwarding.ts"; import * as PeerLinks from "./PeerLinks.ts"; +import * as PeerMcpClient from "./PeerMcpClient.ts"; /** How long one follow of a remote task waits on its thread there before checking again. */ const FOLLOW_WAIT_MS = 10 * 60 * 1_000; @@ -344,39 +345,42 @@ const make = Effect.gen(function* () { const key = input.clientRequestId ?? (yield* random.uuidV4()); const task = taskPrompt(input); const title = input.title ?? parent.thread.title; - const launched = yield* forwarding.call( - scope, - ProjectToolkit.tools.t3_thread_launch, - target.environmentId, - { - projectId, - title, - modelSelection: { instanceId, model, ...(options === undefined ? {} : { options }) }, - ...(runtimeMode === undefined ? {} : { runtimeMode }), - ...(interactionMode === undefined ? {} : { interactionMode }), - message: task, - clientRequestId: `delegate:${scope.thread.threadId}:${key}`, - // The thread there shows it is this thread's subagent, with a way back. - delegatedFrom: { - environmentId: hereId, - threadId: parent.thread.id, - title: parent.thread.title, - }, - }, - ); const link = yield* links.get(target.environmentId).pipe( Effect.map(Option.getOrUndefined), - Effect.orElseSucceed(() => undefined), + Effect.mapError( + (error) => + new OrchestratorMcpFailure({ code: "orchestration_error", message: error.message }), + ), ); + if (link === undefined) { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "The destination is no longer linked here.", + }); + } const commandId = CommandId.make( `command:mcp:${encodeURIComponent(scope.requestNamespace)}:remote-delegate:${encodeURIComponent(key)}`, ); + const launch = { + requestNamespace: scope.requestNamespace, + clientRequestId: `delegate:${scope.thread.threadId}:${key}`, + linkIdentity: link.launchIdentity, + projectId, + title, + modelSelection: { instanceId, model, ...(options === undefined ? {} : { options }) }, + ...(runtimeMode === undefined ? {} : { runtimeMode }), + ...(interactionMode === undefined ? {} : { interactionMode }), + delegatedFrom: { + environmentId: hereId, + threadId: parent.thread.id, + title: parent.thread.title, + }, + }; const remoteChild: OrchestrationV2RemoteTaskChild = { environmentId: target.environmentId, - threadId: launched.threadId, - label: link?.label ?? target.environmentId, + label: link.label, title, - ...(launched.runId === null ? {} : { runId: launched.runId }), + launch, }; const recorded = yield* threads .dispatch({ @@ -389,7 +393,7 @@ const make = Effect.gen(function* () { ...(input.title === undefined ? {} : { title: input.title }), // The driver runs there; this side only labels the task with it. driver: ProviderDriverKind.make("remote"), - modelSelection: launched.modelSelection, + modelSelection: launch.modelSelection, remoteChild, completionWake: input.mode === "wait" ? "settled_only" : "always", }) @@ -408,14 +412,6 @@ const make = Effect.gen(function* () { message: "Unable to record the delegated task.", }), ), - // Nothing here follows a thread the task was never recorded for, so stop it there. - // A crash between the launch and this record still leaves it running there - // until a retry with the same clientRequestId records it. - Effect.tapError(() => - stopThere(followScope(scope.thread, remoteChild), remoteChild, "stop").pipe( - Effect.ignoreCause({ log: true }), - ), - ), ); const taskEvent = recorded.storedEvents.find( (stored) => stored.event.type === "subagent.updated", @@ -427,7 +423,11 @@ const make = Effect.gen(function* () { yield* threads.getThreadRecords(scope.thread.threadId, ["subagents"]).pipe( Effect.map( ({ subagents }) => - subagents.find((task) => task.remoteChild?.threadId === launched.threadId)?.id, + subagents.find( + (task) => + task.remoteChild?.launch?.requestNamespace === launch.requestNamespace && + task.remoteChild.launch.clientRequestId === launch.clientRequestId, + )?.id, ), Effect.orElseSucceed(() => undefined), ); @@ -437,7 +437,28 @@ const make = Effect.gen(function* () { message: "The delegated task was not recorded.", }); } - yield* followOne(scope.thread.threadId, taskId).pipe(Effect.forkIn(followers)); + yield* launchPending(scope.thread.threadId, taskId).pipe( + Effect.catchIf( + (error) => PeerMcpClient.isUncertainCall(error), + (error) => + Effect.logWarning("remote-delegation.launch-deferred", { + taskId, + code: error.code, + }), + ), + // A refusal reaches the caller as this call's error, so no wake is owed for it. + Effect.tapError(() => + threads + .dispatch({ + type: "delegated_task.completion-delivery.dispose", + commandId: CommandId.make(`command:remote-task-refused:${taskId}`), + parentThreadId: scope.thread.threadId, + taskId, + }) + .pipe(Effect.ignoreCause({ log: true })), + ), + Effect.ensuring(followOne(scope.thread.threadId, taskId).pipe(Effect.forkIn(followers))), + ); return { taskId }; }, ); @@ -539,6 +560,7 @@ const make = Effect.gen(function* () { result: string, stoppedThere: "ran-out" | "stopped" | undefined, stopCount?: number, + launchError?: string, ) { const attempt = yield* random.uuidV4(); yield* threads @@ -550,6 +572,7 @@ const make = Effect.gen(function* () { status, result, ...(stoppedThere === undefined ? {} : { stoppedThere }), + ...(launchError === undefined ? {} : { launchError }), ...(stopCount === undefined ? {} : { stopCount }), }) .pipe( @@ -561,6 +584,141 @@ const make = Effect.gen(function* () { ); }); + /** Replays the stored launch identity; a different follower namespace would launch another child. */ + const launchPending = Effect.fn("RemoteDelegation.launchPending")( + function* (parentThreadId: ThreadId, taskId: NodeId) { + let { thread, task } = yield* readTask(parentThreadId, taskId); + if (task?.remoteChild === undefined || task.remoteChild.threadId !== undefined) return; + if (task.result !== null && task.remoteChild.stoppedThere === true) return; + const launch = task.remoteChild.launch; + if (launch === undefined) { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "The pending remote task has no launch inputs.", + }); + } + // Only a send that may have reached the destination before can leave work there. + const sentBefore = task.remoteChild.launchStarted === true; + if (!sentBefore) { + // The parent's dispatch lock orders this against Stop/cancel. Once committed, + // acceptance is uncertain until the same key returns a destination receipt. + if (task.result !== null) { + yield* complete( + parentThreadId, + taskId, + terminalStatus(task.status), + task.result, + "stopped", + task.remoteChild.stopCount, + ); + return; + } + yield* threads + .dispatch({ + type: "delegated_task.remote.launch", + commandId: CommandId.make(`command:remote-task-launch:${taskId}`), + parentThreadId, + taskId, + }) + .pipe( + Effect.catch((error) => + readTask(parentThreadId, taskId).pipe( + Effect.flatMap(({ task }) => + task?.result != null ? Effect.void : Effect.fail(error), + ), + ), + ), + ); + ({ thread, task } = yield* readTask(parentThreadId, taskId)); + if (task?.remoteChild?.launchStarted !== true) { + if (task?.result != null) + yield* complete( + parentThreadId, + taskId, + terminalStatus(task.status), + task.result, + "stopped", + task.remoteChild?.stopCount, + ); + return; + } + } + const remote = task.remoteChild; + if (remote === undefined) return; + const scope: ThreadScope = { + ...followScope( + { threadId: parentThreadId, providerInstanceId: thread.providerInstanceId }, + remote, + ), + requestNamespace: launch.requestNamespace, + }; + const launched = yield* forwarding + .call( + scope, + ProjectToolkit.tools.t3_thread_launch, + remote.environmentId, + { + projectId: launch.projectId, + title: launch.title, + modelSelection: launch.modelSelection, + ...(launch.runtimeMode === undefined ? {} : { runtimeMode: launch.runtimeMode }), + ...(launch.interactionMode === undefined + ? {} + : { interactionMode: launch.interactionMode }), + delegatedFrom: launch.delegatedFrom, + message: task.prompt, + clientRequestId: launch.clientRequestId, + }, + { expectedLinkIdentity: launch.linkIdentity }, + ) + .pipe( + Effect.catchTags({ + OrchestratorMcpFailure: (error) => { + if ( + isFinal(error) || + isModeRefusal(error) || + error.code === "invalid_request" || + (!sentBefore && !PeerMcpClient.isUncertainCall(error)) + ) { + const uncertain = `${error.message} The remote launch could not be confirmed. If it was already accepted, work may still be running on ${remote.label}.`; + return complete( + parentThreadId, + taskId, + "failed", + sentBefore ? uncertain : error.message, + "stopped", + remote.stopCount, + sentBefore ? uncertain : undefined, + ).pipe(Effect.andThen(Effect.fail(error))); + } + // A prior send may have launched work; a later refusal cannot disprove acceptance. + return Effect.fail( + new PeerMcpClient.UncertainCall({ code: error.code, message: error.message }), + ); + }, + }), + ); + yield* threads.dispatch({ + type: "delegated_task.remote.attach", + commandId: CommandId.make(`command:remote-task-attach:${taskId}:${launched.threadId}`), + parentThreadId, + taskId, + environmentId: remote.environmentId, + threadId: launched.threadId, + runId: launched.runId, + modelSelection: launched.modelSelection, + }); + }, + Effect.mapError((error) => + error._tag === "OrchestratorMcpFailure" + ? error + : new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "Unable to reconcile the remote task's launch.", + }), + ), + ); + /** * How this environment reaches a task's thread there: as the parent * thread, under one namespace per parent, so a stop sent again after a @@ -589,7 +747,7 @@ const make = Effect.gen(function* () { */ const stopThere = ( scope: ThreadScope, - remote: OrchestrationV2RemoteTaskChild, + remote: OrchestrationV2RemoteTaskChild & { readonly threadId: ThreadId }, key: string, reason?: string, ) => @@ -610,7 +768,7 @@ const make = Effect.gen(function* () { */ const lastReply = Effect.fn("RemoteDelegation.lastReply")(function* ( scope: ThreadScope, - remote: OrchestrationV2RemoteTaskChild, + remote: OrchestrationV2RemoteTaskChild & { readonly threadId: ThreadId }, runId: RunId | null, failed = false, ) { @@ -742,10 +900,12 @@ const make = Effect.gen(function* () { parentThreadId: ThreadId, taskId: NodeId, ) { + yield* launchPending(parentThreadId, taskId); const sequence = yield* threads.getThreadEventSequence(parentThreadId); const { thread, task } = yield* readTask(parentThreadId, taskId); if (task?.remoteChild === undefined) return true; - const remote = task.remoteChild; + if (task.remoteChild.threadId === undefined) return task.result !== null; + const remote = { ...task.remoteChild, threadId: task.remoteChild.threadId }; const scope = followScope( { threadId: parentThreadId, providerInstanceId: thread.providerInstanceId }, remote, @@ -923,14 +1083,38 @@ const make = Effect.gen(function* () { }); } const { reason } = input; - const there = followScope(scope.thread, remote); + if (remote.threadId === undefined) { + if (task.result === null) { + // The snapshot may race a committed launch marker. Reconciliation decides + // whether nothing was sent or an accepted launch still owes a stop. + yield* complete( + task.threadId, + task.id, + "cancelled", + reason ?? "Cancelled.", + undefined, + ).pipe( + Effect.mapError( + (error) => + new OrchestratorMcpFailure({ + code: "orchestration_error", + message: `Unable to record the cancellation: ${error.message}`, + }), + ), + ); + } + yield* followOne(task.threadId, task.id).pipe(Effect.forkIn(followers)); + return { taskId: task.id, status: "cancel_requested" } as const; + } + const attached = { ...remote, threadId: remote.threadId }; + const there = followScope(scope.thread, attached); // A later cancel stops what started there since; a retry of this one, nothing new. const key = `cancel:${encodeURIComponent(scope.requestNamespace)}:${ input.clientRequestId ?? (yield* random.uuidV4()) }`; if (task.result !== null) { // Best effort: the result stands whether or not the stop gets there. - yield* stopThere(there, remote, key, reason).pipe( + yield* stopThere(there, attached, key, reason).pipe( Effect.andThen( complete( task.threadId, @@ -953,7 +1137,7 @@ const make = Effect.gen(function* () { } // A link that is gone or a machine that is down still cancels the task // here; its follower keeps trying to stop it there, across restarts. - const unstopped = yield* stopThere(there, remote, key, reason).pipe( + const unstopped = yield* stopThere(there, attached, key, reason).pipe( Effect.as(undefined), Effect.catch((error) => Effect.succeed(error.message)), ); diff --git a/apps/web/src/components/chat/ThreadRelationshipsControl.tsx b/apps/web/src/components/chat/ThreadRelationshipsControl.tsx index 95c6d8f0966dd..10fdac7661376 100644 --- a/apps/web/src/components/chat/ThreadRelationshipsControl.tsx +++ b/apps/web/src/components/chat/ThreadRelationshipsControl.tsx @@ -267,6 +267,7 @@ function RemoteSubagentRow(props: { const navigate = useNavigate(); const environment = useEnvironment(props.agent.remoteChild.environmentId); const { remoteChild } = props.agent; + const remoteThreadId = remoteChild.threadId; const status = props.agent.status; const failed = status === "failed"; const title = relationshipThreadTitle({ @@ -283,13 +284,13 @@ function RemoteSubagentRow(props: { size="sm" variant="ghost" part="row" - aria-disabled={environment === null || undefined} + aria-disabled={environment === null || remoteThreadId === undefined || undefined} onClick={() => { - if (environment === null) return; + if (environment === null || remoteThreadId === undefined) return; void navigate({ to: "/$environmentId/$threadId", params: buildThreadRouteParams( - scopeThreadRef(remoteChild.environmentId, remoteChild.threadId), + scopeThreadRef(remoteChild.environmentId, remoteThreadId), ), }); }} @@ -312,16 +313,24 @@ function RemoteSubagentRow(props: { className={`flex max-w-1/2 shrink-0 gap-1 text-2xs ${failed ? "text-destructive" : "text-muted-foreground"}`} > on {remoteChild.label} - · {threadRelationshipStatusLabel(status)} + + ·{" "} + {remoteThreadId === undefined && props.agent.result === null + ? "Launching" + : threadRelationshipStatusLabel(status)} + diff --git a/apps/web/src/components/chat/V2LifecycleRow.tsx b/apps/web/src/components/chat/V2LifecycleRow.tsx index 1854bc74181d0..adc174a89a5db 100644 --- a/apps/web/src/components/chat/V2LifecycleRow.tsx +++ b/apps/web/src/components/chat/V2LifecycleRow.tsx @@ -412,7 +412,8 @@ function SubagentTimelineLink(props: { environmentThreadDetails.threadAtom(props.parentRef), (thread) => thread?.projection.subagents.find((agent) => agent.id === props.subagentId) ?? null, ); - const remote = props.remote ?? agent?.remoteChild; + const remote = agent?.remoteChild ?? props.remote; + const remoteThreadId = remote?.threadId; const remoteEnvironment = useEnvironment(remote?.environmentId ?? null); const navigate = useNavigate(); const threadId = props.threadId; @@ -420,20 +421,26 @@ function SubagentTimelineLink(props: { const open = threadId !== null ? () => props.onOpenThread(threadId) - : remote !== undefined && remoteEnvironment !== null + : remote !== undefined && remoteThreadId !== undefined && remoteEnvironment !== null ? () => void navigate({ to: "/$environmentId/$threadId", - params: buildThreadRouteParams(scopeThreadRef(remote.environmentId, remote.threadId)), + params: buildThreadRouteParams(scopeThreadRef(remote.environmentId, remoteThreadId)), }) : null; const liveStatus = agent?.status ?? props.status; const status = props.event ? props.event.status : liveStatus; - const statusLabel = props.event?.label ?? subagentStatusVisual(liveStatus).label; + const statusLabel = + props.event?.label ?? + (remote !== undefined && + remoteThreadId === undefined && + !SETTLED_SUBAGENT_STATUSES.has(liveStatus) + ? "Launching" + : subagentStatusVisual(liveStatus).label); const result = (agent?.result ?? props.result)?.trim(); const progress = (agent?.progress ?? props.progress)?.trim(); const settled = SETTLED_SUBAGENT_STATUSES.has(status ?? liveStatus); - const rawDetail = settled ? result || progress : progress || result; + const rawDetail = remote?.launchError ?? (settled ? result || progress : progress || result); const detail = subagentCardDetail(rawDetail); const failed = status === "failed"; const timing = { @@ -527,7 +534,7 @@ function SubagentTimelineLink(props: { model={agent?.model ?? null} modelSelection={agent?.modelSelection} status={liveStatus} - result={agent?.result ?? props.result} + result={remote?.launchError ?? agent?.result ?? props.result} progress={agent?.progress ?? props.progress} /> diff --git a/apps/web/src/state/entities.ts b/apps/web/src/state/entities.ts index 27168e1ce7893..0da67c4944676 100644 --- a/apps/web/src/state/entities.ts +++ b/apps/web/src/state/entities.ts @@ -494,7 +494,7 @@ export function readResponseSubagents( const child = { threadId: childThreadId, title }; byTask.set(taskId, child); local.set(childThreadId, child); - } else if (remote) { + } else if (remote?.threadId !== undefined) { // A task sent to a linked environment has no thread here; its thread lives there. const { threadId, environmentId } = remote; byTask.set(taskId, { threadId, environmentId, title: title ?? remote.title }); diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 44e9aa8290e9a..035018c0359bd 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -833,6 +833,20 @@ export const OrchestrationV2ExecutionNode = Schema.Struct({ }); export type OrchestrationV2ExecutionNode = typeof OrchestrationV2ExecutionNode.Type; +/** Durable launch inputs and the original retry identity. No credentials. */ +const OrchestrationV2RemoteTaskLaunch = Schema.Struct({ + requestNamespace: Schema.String, + clientRequestId: Schema.String, + /** Nonsecret grant-generation identity; relinking must not create a replacement child. */ + linkIdentity: Schema.String, + projectId: ProjectId, + title: Schema.String, + modelSelection: ModelSelection, + runtimeMode: Schema.optional(RuntimeMode), + interactionMode: Schema.optional(ProviderInteractionMode), + delegatedFrom: OrchestrationV2DelegatedFrom, +}); + /** Append-only: phases discovered at runtime join when they start. */ export const OrchestrationV2WorkflowPhase = Schema.Struct({ index: NonNegativeInt, @@ -897,10 +911,16 @@ const DedupedBodyMarker = Schema.optionalKey(Schema.Literal(true)); */ export const OrchestrationV2RemoteTaskChild = Schema.Struct({ environmentId: EnvironmentId, - threadId: ThreadId, + /** Unset until the accepted launch is attached to the durable source task. */ + threadId: Schema.optional(ThreadId), label: Schema.String, /** The title it launched with there; absent on tasks recorded before it was kept. */ title: Schema.optional(Schema.String), + launch: Schema.optional(OrchestrationV2RemoteTaskLaunch), + /** Committed before network I/O: an absent receipt may now mean acceptance is unknown. */ + launchStarted: Schema.optional(Schema.Boolean), + /** Recovery was refused; already-accepted work may still be running there. */ + launchError: Schema.optional(Schema.String), /** The run there that carries the task; absent on tasks recorded before it was kept. */ runId: Schema.optional(RunId), /** @@ -3663,8 +3683,8 @@ const OrchestrationV2InternalCommand = Schema.Union([ /** * Records a delegated task that runs in a linked environment: the task, its * node and its turn item on the parent, as `delegated_task.request` does, - * without a child thread here. `remoteChild` is the thread already launched - * there. The ids derive from `commandId`, so a retry records nothing new. + * without a child thread here. Pending remote work retains its launch inputs + * before it is sent. The ids derive from `commandId`, so a retry records nothing new. */ Schema.Struct({ type: Schema.Literal("delegated_task.remote.request"), @@ -3679,6 +3699,24 @@ const OrchestrationV2InternalCommand = Schema.Union([ remoteChild: OrchestrationV2RemoteTaskChild, completionWake: Schema.optional(Schema.Literals(["always", "settled_only"])), }), + /** Serializes starting a pending launch against cancellation before any network I/O. */ + Schema.Struct({ + type: Schema.Literal("delegated_task.remote.launch"), + commandId: CommandId, + parentThreadId: ThreadId, + taskId: NodeId, + }), + /** Attaches an accepted launch, including after the original parent run has ended. */ + Schema.Struct({ + type: Schema.Literal("delegated_task.remote.attach"), + commandId: CommandId, + parentThreadId: ThreadId, + taskId: NodeId, + environmentId: EnvironmentId, + threadId: ThreadId, + runId: Schema.NullOr(RunId), + modelSelection: ModelSelection, + }), /** * Completes a delegated task that ran in a linked environment, with what its * thread there ended as. The parent wakes exactly as for a child here. A @@ -3699,6 +3737,8 @@ const OrchestrationV2InternalCommand = Schema.Union([ * the stop owed for whatever follows it. */ stoppedThere: Schema.optional(Schema.Literals(["ran-out", "stopped"])), + /** Preserve launch uncertainty even when cancellation already supplied the task result. */ + launchError: Schema.optional(Schema.String), /** The stop generation this report acknowledges; absent for the original unnumbered stop. */ stopCount: Schema.optional(NonNegativeInt), }), diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 84dc4e46c98b2..67553e0bcec6f 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -215,9 +215,14 @@ export const OrchestratorMcpDelegateTaskResult = Schema.Struct({ taskId: NodeId, /** Null for a task that runs in a linked environment; see `remoteChild`. */ childThreadId: Schema.NullOr(ThreadId), - /** The linked environment and thread a remote task runs as. */ + /** The destination environment; threadId is absent until its launch is attached. */ remoteChild: Schema.optional( - Schema.Struct({ environmentId: EnvironmentId, threadId: ThreadId, label: Schema.String }), + Schema.Struct({ + environmentId: EnvironmentId, + threadId: Schema.optional(ThreadId), + label: Schema.String, + launchError: Schema.optional(Schema.String), + }), ), link: Schema.optional(Schema.String), childRunId: Schema.NullOr(RunId),