diff --git a/apps/mobile/src/features/threads/PendingApprovalCard.tsx b/apps/mobile/src/features/threads/PendingApprovalCard.tsx index b5a9d4ed27eb..7cf5df2e2c2e 100644 --- a/apps/mobile/src/features/threads/PendingApprovalCard.tsx +++ b/apps/mobile/src/features/threads/PendingApprovalCard.tsx @@ -45,6 +45,11 @@ export function PendingApprovalCard(props: PendingApprovalCardProps) { {props.approval.appName ?? props.approval.requestKind} + {props.approval.remoteEnvironmentLabel ? ( + + Delegated task on {props.approval.remoteEnvironmentLabel} + + ) : null} {props.approval.detail ? ( {props.approval.detail} diff --git a/apps/mobile/src/features/threads/PendingUserInputCard.tsx b/apps/mobile/src/features/threads/PendingUserInputCard.tsx index d097416b2ef2..c7ae353e1177 100644 --- a/apps/mobile/src/features/threads/PendingUserInputCard.tsx +++ b/apps/mobile/src/features/threads/PendingUserInputCard.tsx @@ -213,6 +213,11 @@ export function PendingUserInputCard(props: PendingUserInputCardProps) { run to continue. ) : null} + {props.pendingUserInput.remoteEnvironmentLabel ? ( + + Delegated task on {props.pendingUserInput.remoteEnvironmentLabel} + + ) : null} {props.pendingUserInput.questions.map((question) => { const draft = props.drafts[question.id]; return ( @@ -271,6 +276,7 @@ export function PendingUserInputCard(props: PendingUserInputCardProps) { question={question} questions={props.pendingUserInput.questions} disabled={responseDisabled} + attachable={props.pendingUserInput.remoteTaskId === undefined} value={draft?.customAnswer ?? ""} onChangeText={(value) => props.onChangeCustomAnswer(props.pendingUserInput.requestId, question.id, value) diff --git a/apps/mobile/src/features/threads/QuestionAttachments.tsx b/apps/mobile/src/features/threads/QuestionAttachments.tsx index 724a607f57d5..9aded5be9322 100644 --- a/apps/mobile/src/features/threads/QuestionAttachments.tsx +++ b/apps/mobile/src/features/threads/QuestionAttachments.tsx @@ -35,6 +35,7 @@ export function QuestionAttachments(props: { question: UserInputQuestion; questions: ReadonlyArray; disabled: boolean; + attachable: boolean; value: string; onChangeText: (value: string) => void; onInputFocusChange?: ((focused: boolean) => void) | undefined; @@ -86,6 +87,10 @@ export function QuestionAttachments(props: { const paste = useNativePaste((uris) => { const scope = pickerScope.current; if (!selectedThread || props.disabled || !configs.get(selectedThread.environmentId)) return; + if (!props.attachable) { + Dialog.alert("Could not paste image", "This question cannot accept attachments."); + return; + } const key = questionAttachmentDraftKey( selectedThread.environmentId, selectedThread.id, @@ -117,7 +122,7 @@ export function QuestionAttachments(props: { if (!selectedThread || props.question.allowCustomAnswer === false) return null; const { environmentId, id: threadId } = selectedThread; const capabilities = configs.get(environmentId)?.environment.capabilities; - const canAttach = capabilities !== undefined; + const canAttach = props.attachable && capabilities !== undefined; const key = questionAttachmentDraftKey( environmentId, threadId, diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index a23f06abef24..300fb5fd7150 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -53,6 +53,7 @@ import { } from "@t3tools/shared/codexArtifactTemplates"; import { declinePendingApprovals, + redirectableApprovals, sendApprovalFeedback, type ApprovalResponder, type ThreadUserInputQuestion, @@ -1080,12 +1081,15 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { : []; })(); + const redirectedApprovals = redirectableApprovals(props.activePendingApprovals); const currentPendingApprovals = () => props.selectedThreadDetailRef === null ? [] - : (appAtomRegistry.get( - environmentThreadDetails.pendingRequestsAtom(props.selectedThreadDetailRef), - )?.approvals ?? []); + : redirectableApprovals( + appAtomRegistry.get( + environmentThreadDetails.pendingRequestsAtom(props.selectedThreadDetailRef), + )?.approvals ?? [], + ); const handleSendMessage = async ( followUp?: ActiveTurnComposerAction, stillSendable?: () => boolean, @@ -1097,7 +1101,7 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { ); // Runs only once the send path has validated the draft and is about to deliver it, so an // empty submit, a queued-message edit, or `/feedback` leaves approvals alone. - const approvalsToDecline = props.activePendingApprovals; + const approvalsToDecline = redirectedApprovals; const declineBeforeDispatch = approvalsToDecline.length > 0 && props.activePendingUserInput === null ? async (text: string, hasAttachmentsOrContext: boolean) => { @@ -1577,8 +1581,8 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { reportedModelSelection={reportedModelSelection} editorRef={composerEditorRef} placeholder={ - props.activePendingApproval - ? props.activePendingApproval.supportsFeedback + redirectedApprovals[0] + ? redirectedApprovals[0].supportsFeedback ? "Send feedback to decline this request" : "Send a message to decline and redirect the agent" : "Ask anything" @@ -1609,8 +1613,7 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { interactionLocked={props.draftMode?.interactionLocked} draftKey={props.composerDraftKey ?? undefined} followUpBehavior={ - props.activePendingApprovals.length > 0 && - props.activePendingUserInput === null + redirectedApprovals.length > 0 && props.activePendingUserInput === null ? "steer" : props.followUpBehavior } diff --git a/apps/server/src/mcp/McpInvocationContext.ts b/apps/server/src/mcp/McpInvocationContext.ts index 26795c51696b..1a94adf75557 100644 --- a/apps/server/src/mcp/McpInvocationContext.ts +++ b/apps/server/src/mcp/McpInvocationContext.ts @@ -127,6 +127,16 @@ export const hiddenWhileLinksOff = (): boolean => { return scope?.linkedEnvironments !== false; }; +/** Another T3 Code environment's peer link calling on its own behalf, not an agent here. */ +export const isLinkSession = (scope: McpInvocationScope | undefined): boolean => + scope?.thread === undefined && scope?.client?.linked === true; + +/** `McpSchema.EnabledWhen` for tools only a peer link's own environment calls. */ +export const visibleToLinkSessions = (): boolean => { + const fiber = Fiber.getCurrent(); + return isLinkSession(fiber && Context.getOrUndefined(fiber.context, McpInvocationContext)); +}; + export type McpCapabilityError = C extends "preview" ? PreviewAutomationUnavailableError : McpCapabilityUnavailableError; diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 3681c4578909..b373afa98815 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -8,6 +8,7 @@ import { NodeId, ORCHESTRATOR_MCP_RUNNING_TASK_NEXT, TurnItemId, + type OrchestrationV2RemoteTaskRequest, type OrchestrationV2Run, type OrchestrationV2Subagent, type OrchestrationV2ThreadProjection, @@ -63,6 +64,7 @@ import { type ServerProviderModel, ThreadId, isOrchestrationV2WorkActive, + unansweredRemoteTaskRequests, } from "@t3tools/contracts"; import { getProviderOptionCurrentValue } from "@t3tools/shared/model"; import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError"; @@ -131,6 +133,10 @@ const TASK_WAKE_EVENTS = [ { thread: "child", eventType: "provider-thread.updated" }, { thread: "child", eventType: "runtime-request.updated" }, ] as const; +const REMOTE_TASK_WAKE_EVENTS = [ + { thread: "parent", eventType: "subagent.updated" }, + { thread: "parent", eventType: "thread.metadata-updated" }, +] as const; const SECRET_REQUEST_POLL_INTERVAL_MS = 500; const DEFAULT_THREAD_LIST_LIMIT = 50; const DEFAULT_THREAD_READ_LIMIT = 50; @@ -510,6 +516,23 @@ function pendingTaskApproval( return approval; } +/** The oldest approval a remote task's thread there waits on that the user has not answered here. */ +function remoteTaskApproval( + task: OrchestrationV2Subagent, + unanswered: ReadonlyArray, +): OrchestratorMcpDelegateTaskResult["pendingApproval"] { + for (const request of unanswered.toSorted( + (left, right) => + DateTime.toEpochMillis(left.createdAt) - DateTime.toEpochMillis(right.createdAt), + )) { + const runId = request.runId ?? task.remoteChild?.runId; + if (request.kind !== "user_input" && runId !== undefined) { + return { requestId: request.requestId, kind: request.kind, runId }; + } + } + return null; +} + function directAppOwnedChildTasks( parent: Pick, target: Pick, @@ -1411,7 +1434,15 @@ const make = Effect.gen(function* () { ); } if (task.childThreadId === null) { - return yield* readRemoteTask(scope, task, waitTimedOut, acknowledgeTerminal); + return yield* readRemoteTask( + scope, + task, + unansweredRemoteTaskRequests(parentProjection.thread).filter( + (request) => request.taskId === task.id, + ), + waitTimedOut, + acknowledgeTerminal, + ); } const childControls = yield* threadManagement .getThreadRecords( @@ -1607,8 +1638,8 @@ const make = Effect.gen(function* () { Math.max(1, requestedTimeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS), ); const waited = yield* waitForTask(scope, taskId, timeoutMs); - if (Option.isSome(waited)) return waited.value; - // As for a local task: once the blocking wait ends, a later result wakes the parent. + if (Option.isSome(waited) && isTerminalTaskStatus(waited.value.status)) return waited.value; + // As for a local task: once the blocking wait ends without a result, a later one wakes the parent. yield* threadManagement .dispatch({ type: "delegated_task.wake-policy", @@ -1622,7 +1653,7 @@ const make = Effect.gen(function* () { completionWake: "always", }) .pipe(Effect.ignoreCause({ log: true })); - return yield* readTask(scope, taskId, true, true); + return yield* readTask(scope, taskId, Option.isNone(waited), true); }); /** @@ -1632,6 +1663,7 @@ const make = Effect.gen(function* () { const readRemoteTask = Effect.fn("OrchestratorMcpService.readRemoteTask")(function* ( scope: McpThreadInvocationScope, task: OrchestrationV2Subagent, + unanswered: ReadonlyArray, waitTimedOut: boolean, acknowledgeTerminal: boolean, ): Effect.fn.Return { @@ -1698,9 +1730,8 @@ const make = Effect.gen(function* () { : null, latestTerminalSummary: terminal ? task.result : null, latestTerminalResultContextTransferId: null, - // Its questions and approvals are answered in the environment it runs in. - hasPendingQuestion: false, - pendingApproval: null, + hasPendingQuestion: !terminal && unanswered.some((request) => request.kind === "user_input"), + pendingApproval: terminal ? null : remoteTaskApproval(task, unanswered), waitTimedOut, } satisfies OrchestratorMcpDelegateTaskResult; }); @@ -1721,12 +1752,10 @@ const make = Effect.gen(function* () { return Option.some(initial); } // One stream per event type, so transcript events never fill a buffer. - // A remote task has no child here; its completion lands on the parent. + // A remote task has no child here; its completion lands on the parent, and so + // does the mirror of what its thread there waits on. const childThreadId = initial.childThreadId; - const wakeEvents = - childThreadId === null - ? TASK_WAKE_EVENTS.filter(({ thread }) => thread === "parent") - : TASK_WAKE_EVENTS; + const wakeEvents = childThreadId === null ? REMOTE_TASK_WAKE_EVENTS : TASK_WAKE_EVENTS; return yield* Stream.mergeAll( wakeEvents.map(({ thread, eventType }) => threadManagement.streamStoredEventsFrom({ diff --git a/apps/server/src/mcp/linkOrigin.ts b/apps/server/src/mcp/linkOrigin.ts index 7373fd51d2bd..ae799cdb1fb8 100644 --- a/apps/server/src/mcp/linkOrigin.ts +++ b/apps/server/src/mcp/linkOrigin.ts @@ -34,6 +34,10 @@ const fencingLinkOrigin = (caller: Pick) => { return origin !== undefined && linkOriginIsFenced(origin) ? origin : undefined; }; +/** Whether a link paired below full access limits what `caller` may change. */ +export const callerIsFenced = (caller: Pick) => + fencingLinkOrigin(caller) !== undefined; + const outsideLink = (origin: OrchestrationV2LinkOrigin, message: string) => new OrchestratorMcpFailure({ code: "capability_denied", diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index 8ad9afa43519..c50c3fd2243a 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -23,6 +23,7 @@ import { import * as ThreadSearch from "../../../orchestration-v2/ThreadSearch.ts"; import * as ScheduledTasks from "../../../scheduledTasks/ScheduledTaskService.ts"; import { queuedRunsInDeliveryOrder } from "../../../orchestration-v2/QueuedRunOrder.ts"; +import { respondToLinkedRequest, watchLinkedRequests } from "./linkRequests.ts"; import { ThreadToolkit } from "./tools.ts"; function queueEntry( @@ -256,6 +257,23 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { }), ), ), + t3_link_requests: McpToolAccess.either( + (input) => input.action === "watch", + McpToolAccess.reads((input) => watchLinkedRequests(input)), + McpToolAccess.writesThreads( + (input) => [input.threadId], + (input) => + Effect.gen(function* () { + return yield* respondToLinkedRequest({ + threadId: input.threadId, + requestId: yield* required(input.requestId, "requestId"), + decision: input.decision, + answers: input.answers, + clientRequestId: yield* required(input.clientRequestId, "clientRequestId"), + }); + }), + ), + ), t3_queue: McpToolAccess.either( (input) => input.action === "list" || input.action === "read", McpToolAccess.reads((input) => diff --git a/apps/server/src/mcp/toolkits/thread/linkRequests.test.ts b/apps/server/src/mcp/toolkits/thread/linkRequests.test.ts new file mode 100644 index 000000000000..7c40e46c5c3c --- /dev/null +++ b/apps/server/src/mcp/toolkits/thread/linkRequests.test.ts @@ -0,0 +1,401 @@ +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import { expect, it } from "@effect/vitest"; +import { + EnvironmentId, + NodeId, + type OrchestrationV2ServerCommand, + type OrchestrationV2RuntimeRequest, + type OrchestrationV2ThreadShell, + type OrchestrationV2TurnItem, + ProviderInstanceId, + ProviderSessionId, + RunId, + RuntimeRequestId, + ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import { McpSchema, McpServer } from "effect/ai"; + +import * as PeerForwarding from "../../../peer/PeerForwarding.ts"; +import * as PeerLinkRequests from "../../../peer/PeerLinkRequests.ts"; +import * as PeerLinks from "../../../peer/PeerLinks.ts"; +import * as ThreadManagement from "../../../orchestration-v2/ThreadManagementService.ts"; +import * as ThreadSearch from "../../../orchestration-v2/ThreadSearch.ts"; +import * as ScheduledTaskService from "../../../scheduledTasks/ScheduledTaskService.ts"; +import * as McpHttpServer from "../../McpHttpServer.ts"; +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import { idleThreadProjection, liveThreadShell } from "../../McpToolAccess.testkit.ts"; + +const threadId = ThreadId.make("thread:delegated-here"); +const runId = RunId.make("run:delegated-here"); +const approvalId = RuntimeRequestId.make("request:approval"); +const questionId = RuntimeRequestId.make("request:question"); +const linkSession = "session:laptop-link"; + +const shell = ( + linkOrigin?: OrchestrationV2ThreadShell["linkOrigin"], +): OrchestrationV2ThreadShell => ({ + ...liveThreadShell(threadId, { runtimeMode: "approval-required" }), + ...(linkOrigin === undefined ? {} : { linkOrigin }), +}); + +const request = ( + id: RuntimeRequestId, + kind: OrchestrationV2RuntimeRequest["kind"], + status: OrchestrationV2RuntimeRequest["status"] = "pending", +): OrchestrationV2RuntimeRequest => ({ + id, + nodeId: NodeId.make("node:root"), + providerTurnId: null, + nativeRequestRef: null, + kind, + status, + responseCapability: { type: "live", providerSessionId: ProviderSessionId.make("session") }, + createdAt: liveThreadShell(threadId).createdAt, + resolvedAt: null, +}); + +const itemBase = { + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 0, + status: "waiting" as const, + title: null, + startedAt: liveThreadShell(threadId).createdAt, + completedAt: null, + updatedAt: liveThreadShell(threadId).createdAt, +}; +const approvalItem: OrchestrationV2TurnItem = { + ...itemBase, + id: TurnItemId.make("item:approval"), + type: "approval_request", + requestId: approvalId, + requestKind: "command", + prompt: "rm -rf build", + reason: "Clean the build", + options: [ + { decision: "accept", label: "Allow" }, + { decision: "acceptAlways", label: "Always allow" }, + ], +}; +const questionItem: OrchestrationV2TurnItem = { + ...itemBase, + id: TurnItemId.make("item:question"), + type: "user_input_request", + requestId: questionId, + questions: [{ id: "size", header: "Size", question: "Which size?", options: [] }], +}; + +const scopeOf = ( + client: McpInvocationContext.McpClientCaller | undefined, + thread?: McpInvocationContext.McpThreadCaller, +): McpInvocationContext.McpInvocationScope => ({ + environmentId: EnvironmentId.make("environment:box"), + requestNamespace: client?.sessionId ?? "agent", + thread, + client, + capabilities: new Set(["orchestration"]), + issuedAt: 0, +}); +const link = (access: McpInvocationContext.McpClientCaller["access"]) => + scopeOf({ sessionId: linkSession, label: "Laptop", access, linked: true }); +const outsideClient = scopeOf({ sessionId: "session:claude", label: "Claude", access: "auto" }); +const agent = scopeOf(undefined, { + threadId: ThreadId.make("thread:agent"), + providerSessionId: "provider:agent", + providerInstanceId: ProviderInstanceId.make("codex"), +}); + +const mcpClient = McpSchema.McpServerClient.of({ + clientId: 1, + protocolVersion: "2025-06-18", + clientCapabilities: {}, + clientInfo: { name: "link-requests", version: "1" }, + initializePayload: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name: "link-requests", version: "1" }, + }, + getClient: Effect.die("unused"), +}); + +const call = (scope: McpInvocationContext.McpInvocationScope, args: Record) => + McpServer.McpServer.pipe( + Effect.flatMap((server) => server.callTool({ name: "t3_link_requests", arguments: args })), + Effect.map((result) => { + const text = result.content[0]?.type === "text" ? result.content[0].text : ""; + return { failed: result.isError === true, body: JSON.parse(text) }; + }), + Effect.provideService(McpInvocationContext.McpInvocationContext, scope), + Effect.provideService(McpSchema.McpServerClient, mcpClient), + ); + +type StoredEvents = ReturnType< + ThreadManagement.ThreadManagementService["Service"]["streamStoredEventsFrom"] +>; + +const layerBox = (input: { + readonly shell: OrchestrationV2ThreadShell; + readonly requests: Ref.Ref>; + readonly items: Ref.Ref>; + readonly dispatched: Ref.Ref>; + readonly requestRecorded: StoredEvents; + readonly itemRecorded: StoredEvents; +}) => + McpHttpServer.layerThreadToolkit.pipe( + Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})), + Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})), + Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), + Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), + Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provideMerge(McpServer.McpServer.layer), + Layer.provide(NodeCrypto.layer), + Layer.provide( + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: (id) => Effect.succeed(id === threadId ? input.shell : liveThreadShell(id)), + getThreadEventSequence: () => Effect.succeed(0), + getProjectThreadRecords: () => + Effect.all([Ref.get(input.requests), Ref.get(input.items)]).pipe( + Effect.map(([runtimeRequests, turnItems]) => ({ + ...idleThreadProjection(input.shell), + runtimeRequests, + turnItems, + })), + ), + streamStoredEventsFrom: (query) => + query?.eventType === "turn-item.updated" ? input.itemRecorded : input.requestRecorded, + dispatch: (command) => + Ref.update(input.dispatched, (seen) => [...seen, command]).pipe( + Effect.as({ sequence: 7, storedEvents: [] } as never), + ), + }), + ), + ); + +const box = Effect.fn("box")(function* ( + options: { + readonly linkOrigin?: OrchestrationV2ThreadShell["linkOrigin"]; + readonly requests?: ReadonlyArray; + readonly items?: ReadonlyArray; + readonly requestRecorded?: ( + requests: Ref.Ref>, + ) => Stream.Stream; + readonly itemRecorded?: OrchestrationV2TurnItem; + } = {}, +) { + const requests = yield* Ref.make(options.requests ?? []); + const items = yield* Ref.make(options.items ?? [approvalItem, questionItem]); + const dispatched = yield* Ref.make>([]); + const { itemRecorded } = options; + const layer = layerBox({ + shell: shell(options.linkOrigin ?? { sessionId: linkSession, label: "Laptop", access: "auto" }), + requests, + items, + dispatched, + requestRecorded: (options.requestRecorded?.(requests) ?? Stream.never) as StoredEvents, + itemRecorded: + itemRecorded === undefined + ? Stream.never + : (Stream.fromEffect( + Ref.update(items, (current) => [...current, itemRecorded]).pipe( + Effect.as({ + sequence: 1, + commandId: null, + event: { type: "turn-item.updated", payload: itemRecorded }, + }), + ), + ) as never), + }); + return { requests, dispatched, layer }; +}); + +const watch = (layer: Layer.Layer, args: Record = {}) => + call(link("auto"), { action: "watch", threadId, ...args }).pipe(Effect.provide(layer)); + +it.effect( + "a link watching a thread reads its live questions and approvals with their details", + () => + Effect.gen(function* () { + const { layer } = yield* box({ + requests: [ + request(approvalId, "command"), + request(questionId, "user_input"), + request(RuntimeRequestId.make("request:answered"), "command", "resolved"), + request(RuntimeRequestId.make("request:auth"), "auth_refresh"), + ], + }); + const { failed, body } = yield* call(link("auto"), { action: "watch", threadId }).pipe( + Effect.provide(layer), + ); + expect(failed).toBe(false); + expect(body.requests).toEqual([ + { + requestId: approvalId, + kind: "command", + createdAt: "2026-01-01T00:00:00.000Z", + runId, + detail: "rm -rf build", + reason: "Clean the build", + options: [{ decision: "accept", label: "Allow" }], + }, + { + requestId: questionId, + kind: "user_input", + createdAt: "2026-01-01T00:00:00.000Z", + runId, + questions: questionItem.type === "user_input_request" ? questionItem.questions : [], + }, + ]); + }), +); + +it.effect("a watch reports a request recorded while it waits", () => + Effect.gen(function* () { + const { layer } = yield* box({ + requests: [request(approvalId, "command")], + requestRecorded: (requests) => + Stream.fromEffect( + Ref.set(requests, [request(approvalId, "command"), request(questionId, "user_input")]), + ), + }); + const known = (yield* watch(layer)).body.version; + const { body } = yield* watch(layer, { knownVersion: known }); + expect(body.requests.map((entry: { requestId: string }) => entry.requestId)).toEqual([ + approvalId, + questionId, + ]); + }), +); + +it.effect("a watch reports a question whose details land after its request", () => + Effect.gen(function* () { + const { layer } = yield* box({ + requests: [request(questionId, "user_input")], + items: [], + itemRecorded: questionItem, + }); + const before = (yield* watch(layer)).body; + expect(before.requests).toEqual([]); + const { body } = yield* watch(layer, { knownVersion: before.version }); + expect(body.requests.map((entry: { requestId: string }) => entry.requestId)).toEqual([ + questionId, + ]); + }), +); + +it.effect("a watch with nothing new answers with what the link knows once it times out", () => + Effect.gen(function* () { + const { layer } = yield* box({ requests: [request(approvalId, "command")] }); + const known = (yield* watch(layer)).body.version; + const watching = yield* watch(layer, { knownVersion: known, timeoutMs: 1_000 }).pipe( + Effect.forkChild, + ); + yield* TestClock.adjust(1_000); + const { body } = yield* Fiber.join(watching); + expect(body).toMatchObject({ version: known }); + expect(body.requests.map((entry: { requestId: string }) => entry.requestId)).toEqual([ + approvalId, + ]); + }), +); + +it.effect.each([ + ["an agent here", agent], + ["an outside MCP client", outsideClient], +] as const)("%s does not find the link tool", ([, scope]) => + Effect.gen(function* () { + const { layer } = yield* box({ requests: [request(approvalId, "command")] }); + const refused = yield* call(scope, { action: "watch", threadId }).pipe( + Effect.provide(layer), + Effect.flip, + ); + expect(refused._tag).toBe("InvalidParams"); + }), +); + +it.effect("a read-only link cannot answer", () => + Effect.gen(function* () { + const { dispatched, layer } = yield* box({ requests: [request(approvalId, "command")] }); + const { failed, body } = yield* call(link("read-only"), { + action: "respond", + threadId, + requestId: approvalId, + decision: "accept", + clientRequestId: "answer-1", + }).pipe(Effect.provide(layer)); + expect(failed).toBe(true); + expect(body.code).toBe("capability_denied"); + expect(yield* Ref.get(dispatched)).toEqual([]); + }), +); + +it.effect("a link paired below full access answers only threads it started", () => + Effect.gen(function* () { + const { dispatched, layer } = yield* box({ + linkOrigin: { sessionId: "session:other-link", label: "Other", access: "auto" }, + requests: [request(approvalId, "command")], + }); + const { failed, body } = yield* call(link("auto"), { + action: "respond", + threadId, + requestId: approvalId, + decision: "accept", + clientRequestId: "answer-1", + }).pipe(Effect.provide(layer)); + expect(failed).toBe(true); + expect(body.code).toBe("capability_denied"); + expect(yield* Ref.get(dispatched)).toEqual([]); + }), +); + +it.effect("a link paired below full access cannot accept a request for every later session", () => + Effect.gen(function* () { + const { dispatched, layer } = yield* box({ requests: [request(approvalId, "command")] }); + const { failed, body } = yield* call(link("auto"), { + action: "respond", + threadId, + requestId: approvalId, + decision: "acceptAlways", + clientRequestId: "answer-1", + }).pipe(Effect.provide(layer)); + expect(failed).toBe(true); + expect(body.code).toBe("capability_denied"); + expect(yield* Ref.get(dispatched)).toEqual([]); + }), +); + +it.effect("a link answers an approval on a thread it started, once per answer", () => + Effect.gen(function* () { + const { dispatched, layer } = yield* box({ requests: [request(approvalId, "command")] }); + const answer = call(link("approval-required"), { + action: "respond", + threadId, + requestId: approvalId, + decision: "accept", + clientRequestId: "answer-1", + }).pipe(Effect.provide(layer)); + expect((yield* answer).failed).toBe(false); + expect((yield* answer).failed).toBe(false); + const commands = yield* Ref.get(dispatched); + expect(commands).toHaveLength(2); + expect(commands[0]).toMatchObject({ + type: "runtime-request.respond", + threadId, + requestId: approvalId, + decision: "accept", + }); + // The same answer resent is the same command, which the box records once. + expect(commands[1]?.commandId).toBe(commands[0]?.commandId); + }), +); diff --git a/apps/server/src/mcp/toolkits/thread/linkRequests.ts b/apps/server/src/mcp/toolkits/thread/linkRequests.ts new file mode 100644 index 000000000000..a08a855df396 --- /dev/null +++ b/apps/server/src/mcp/toolkits/thread/linkRequests.ts @@ -0,0 +1,187 @@ +import { + CommandId, + type OrchestrationV2LinkedRequest, + type OrchestrationV2ThreadProjection, + OrchestratorMcpFailure, + type ProviderApprovalDecision, + type ProviderUserInputAnswers, + type RuntimeRequestId, + type ThreadId, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Hash from "effect/Hash"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; + +import { assertNotLinked, callerIsFenced } from "../../linkOrigin.ts"; +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import { dispatchFailure, loadCaller, threadNotFound, unavailable } from "../../threadAccess.ts"; +import { FORWARDED_WAIT_CHUNK_MS } from "./tools.ts"; + +const requireLinkSession = McpInvocationContext.McpInvocationContext.pipe( + Effect.flatMap((scope) => + McpInvocationContext.isLinkSession(scope) + ? Effect.void + : Effect.fail( + new OrchestratorMcpFailure({ + code: "capability_denied", + message: "Only a linked T3 Code environment answers here for its user.", + }), + ), + ), +); + +/** + * The questions and approvals `projection` waits on that an answer can still reach, without + * the choices a fenced link may not make. + */ +const linkedRequestsOf = ( + projection: Pick, + fenced: boolean, +): ReadonlyArray => + projection.runtimeRequests.flatMap((request): ReadonlyArray => { + if (request.status !== "pending" || request.responseCapability.type !== "live") return []; + const common = { requestId: request.id, createdAt: request.createdAt }; + if (request.kind === "user_input") { + const item = projection.turnItems.findLast( + (candidate) => + candidate.type === "user_input_request" && candidate.requestId === request.id, + ); + if (item?.type !== "user_input_request") return []; + return [ + { + ...common, + kind: "user_input", + ...(item.runId === null ? {} : { runId: item.runId }), + questions: item.questions, + }, + ]; + } + if (request.kind === "auth_refresh" || request.kind === "dynamic_tool_call") return []; + const item = projection.turnItems.findLast( + (candidate) => candidate.type === "approval_request" && candidate.requestId === request.id, + ); + const approval = item?.type === "approval_request" ? item : undefined; + const options = fenced + ? approval?.options?.filter((option) => option.decision !== "acceptAlways") + : approval?.options; + return [ + { + ...common, + kind: request.kind, + ...(approval?.runId == null ? {} : { runId: approval.runId }), + ...(approval?.prompt === undefined ? {} : { detail: approval.prompt }), + ...(approval?.reason === undefined ? {} : { reason: approval.reason }), + ...(options === undefined ? {} : { options }), + }, + ]; + }); + +const linkedThread = Effect.fn("mcp.linkedThread")(function* (threadId: ThreadId) { + yield* requireLinkSession; + const caller = yield* loadCaller(); + const shell = yield* caller.threads.getThreadShell(threadId).pipe(Effect.mapError(unavailable)); + if (shell === null || shell.deletedAt !== null) { + return yield* threadNotFound(caller.threads, threadId, "The thread was not found."); + } + const readRequests = caller.threads + .getProjectThreadRecords( + { projectId: shell.projectId, threadId }, + ["runtimeRequests", "turnItems"], + { + turnItemTypes: ["user_input_request", "approval_request"], + }, + ) + .pipe( + Effect.map((projection) => { + const requests = linkedRequestsOf(projection, callerIsFenced(caller)); + return { requests, version: String(Hash.string(JSON.stringify(requests))) }; + }), + Effect.mapError(unavailable), + ); + return { caller, readRequests }; +}); + +/** + * The thread's live requests once their `version` differs from `knownVersion`, or as they are + * when `timeoutMs` passes. One wait answers within a forwarded call's idle limit. + */ +export const watchLinkedRequests = Effect.fn("mcp.watchLinkedRequests")(function* (input: { + readonly threadId: ThreadId; + readonly knownVersion?: string | undefined; + readonly timeoutMs?: number | undefined; +}) { + const { caller, readRequests } = yield* linkedThread(input.threadId); + const differs = (read: { readonly version: string }) => read.version !== input.knownVersion; + // Taken before the read, so a request recorded between the two still wakes the wait. + const afterSequence = yield* caller.threads + .getThreadEventSequence(input.threadId) + .pipe(Effect.mapError(unavailable)); + const current = yield* readRequests; + if (differs(current)) return current; + const updates = (eventType: "runtime-request.updated" | "turn-item.updated") => + caller.threads.streamStoredEventsFrom({ threadId: input.threadId, afterSequence, eventType }); + const changed = yield* Stream.merge( + updates("runtime-request.updated"), + // A request's question or approval details can land after the request itself. + updates("turn-item.updated").pipe( + Stream.filter( + ({ event }) => + event.type === "turn-item.updated" && + (event.payload.type === "approval_request" || + event.payload.type === "user_input_request"), + ), + ), + ).pipe( + Stream.mapError(unavailable), + Stream.mapEffect(() => readRequests), + Stream.filter(differs), + Stream.runHead, + Effect.timeoutOption(input.timeoutMs ?? FORWARDED_WAIT_CHUNK_MS), + Effect.map(Option.flatten), + ); + return Option.getOrElse(changed, () => current); +}); + +/** + * Answers a request for the user of the environment the link belongs to. The caller's + * declaration has already held the link to its access and fence for this thread; a link + * paired below full access may not also accept the request for every later session here. + */ +export const respondToLinkedRequest = Effect.fn("mcp.respondToLinkedRequest")(function* (input: { + readonly threadId: ThreadId; + readonly requestId: RuntimeRequestId; + readonly decision?: ProviderApprovalDecision | undefined; + readonly answers?: ProviderUserInputAnswers | undefined; + readonly clientRequestId: string; +}) { + const { caller } = yield* linkedThread(input.threadId); + if (input.decision === "acceptAlways") { + yield* assertNotLinked(caller, "accept a request for every later session here"); + } + const answer = + input.decision !== undefined && input.answers === undefined + ? { decision: input.decision } + : input.answers !== undefined && input.decision === undefined + ? { answers: input.answers } + : undefined; + if (answer === undefined) { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "Answer an approval with `decision` or a question with `answers`.", + }); + } + const result = yield* caller.threads + .dispatch({ + type: "runtime-request.respond", + // A repeat replays the first answer's receipt, so a resent answer is answered once. + commandId: CommandId.make( + `mcp:link-respond:${encodeURIComponent(caller.scope.requestNamespace)}:${encodeURIComponent(input.clientRequestId)}`, + ), + threadId: input.threadId, + requestId: input.requestId, + ...answer, + }) + .pipe(Effect.mapError(dispatchFailure)); + return { sequence: result.sequence }; +}); diff --git a/apps/server/src/mcp/toolkits/thread/tools.ts b/apps/server/src/mcp/toolkits/thread/tools.ts index 4b566c44a69d..42b06f06d327 100644 --- a/apps/server/src/mcp/toolkits/thread/tools.ts +++ b/apps/server/src/mcp/toolkits/thread/tools.ts @@ -12,6 +12,9 @@ import { ProviderInteractionMode, RuntimeRequestId, ProviderUserInputAnswers, + ProviderApprovalDecision, + OrchestrationV2LinkedRequestJson, + OrchestratorMcpClientRequestId, IsoDateTime, OrchestratorMcpFailure, OrchestrationV2DispatchCommandResult, @@ -22,7 +25,7 @@ import { } from "@t3tools/contracts"; import * as Crypto from "effect/Crypto"; import * as Schema from "effect/Schema"; -import { Tool, Toolkit } from "effect/ai"; +import { McpSchema, Tool, Toolkit } from "effect/ai"; import * as ThreadSearch from "../../../orchestration-v2/ThreadSearch.ts"; import * as ScheduledTaskService from "../../../scheduledTasks/ScheduledTaskService.ts"; @@ -143,6 +146,45 @@ const PendingRequestTool = Tool.make("t3_pending_request", { .annotate(Tool.Destructive, true) .annotate(Tool.OpenWorld, true); +/** + * A peer's `/mcp` holds a wait open for its whole timeout without sending + * anything, and T3 Connect's edge drops a request idle for about 100 s, so a + * forwarded wait goes out in pieces no longer than this. + */ +export const FORWARDED_WAIT_CHUNK_MS = 50_000; + +/** + * How the environment that delegated a task here mirrors its thread's questions and approvals + * and answers them for its user. Only that environment's link calls it, never an agent: an + * agent may not approve, and the link's access and fence still hold here. + */ +export const LinkRequestsTool = Tool.make("t3_link_requests", { + ...commandTool, + description: + "For linked T3 Code environments only. watch: the thread's live questions and approvals, once their `version` differs from `knownVersion` or `timeoutMs` passes. respond: answer one.", + parameters: Schema.Struct({ + action: Schema.Literals(["watch", "respond"]), + threadId: ThreadId, + knownVersion: Schema.optional(Schema.String), + timeoutMs: Schema.optional( + Schema.Int.check(Schema.isBetween({ minimum: 1, maximum: FORWARDED_WAIT_CHUNK_MS })), + ), + requestId: Schema.optional(RuntimeRequestId), + decision: Schema.optional(ProviderApprovalDecision), + answers: Schema.optional(ProviderUserInputAnswers), + clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), + }), + success: Schema.Union([ + Schema.Struct({ + requests: Schema.Array(OrchestrationV2LinkedRequestJson), + version: Schema.String, + }), + OrchestrationV2DispatchCommandResult, + ]), +}) + .annotate(Tool.Destructive, true) + .annotate(McpSchema.EnabledWhen, McpInvocationContext.visibleToLinkSessions); + const threadConfiguration = Schema.Struct({ threadId: ThreadId, modelSelection: ModelSelection, @@ -243,5 +285,6 @@ export const ThreadToolkit = Toolkit.make( ThreadTransferTool, ThreadOrganizeTool, PendingRequestTool, + LinkRequestsTool, QueueTool, ); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index fa447ef16553..020c80d3e377 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -48,6 +48,7 @@ import { type OrchestrationV2ProviderThread, type OrchestrationV2ProviderTurn, type OrchestrationV2RemoteTaskChild, + type OrchestrationV2RemoteTaskRequest, type OrchestrationV2Run, type OrchestrationV2RunAttempt, type OrchestrationV2ThreadShell, @@ -70,6 +71,7 @@ import { ThreadId, TurnItemId, TrimmedNonEmptyString, + unansweredRemoteTaskRequests, } from "@t3tools/contracts"; import { modelSelectionsEqual } from "@t3tools/shared/model"; import { recordThreadOpen } from "@t3tools/shared/paletteSearch"; @@ -422,6 +424,14 @@ export class OrchestratorV2 extends Context.Service + value !== null && typeof value === "object" && !Array.isArray(value) + ? Object.fromEntries( + Object.entries(value).toSorted(([left], [right]) => left.localeCompare(right)), + ) + : value; + /** Run ordinals can skip values: a queued run moves past newer runs when it starts. */ function nextRunOrdinal(projection: Pick): number { return projection.runs.reduce((max, run) => Math.max(max, run.ordinal), 0) + 1; @@ -516,6 +526,8 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { case "delegated_task.remote.request": case "delegated_task.remote.complete": case "delegated_task.remote.stop": + case "delegated_task.remote.requests": + case "delegated_task.remote.respond.fail": case "delegated_task.wake-policy": case "delegated_task.completion-delivery.acknowledge": case "delegated_task.completion-delivery.dispose": @@ -3041,9 +3053,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const pendingRequests = projection.runtimeRequests.filter( (request) => request.status === "pending", ); - const blockingRequestExists = pendingRequests.some( - (request) => request.kind !== "user_input" || request.responseCapability.type !== "message", - ); + const blockingRequestExists = + pendingRequests.some( + (request) => + request.kind !== "user_input" || request.responseCapability.type !== "message", + ) || unansweredRemoteTaskRequests(projection.thread).length > 0; const blocker = blockingRequestExists ? "is waiting on an approval or question. Answer it before settling." : runningRunExists @@ -3151,7 +3165,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio failure?.class !== "usage_limit" || failure.resetAt !== command.limitRecovery.resetAt || resetMs <= DateTime.toEpochMillis(run.completedAt ?? run.requestedAt) || - projection.runtimeRequests.some((request) => request.status === "pending") + projection.runtimeRequests.some((request) => request.status === "pending") || + unansweredRemoteTaskRequests(projection.thread).length > 0 ) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, @@ -3176,7 +3191,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Thread ${command.threadId} snooze wake time ${command.snoozedUntil} is not in the future.`, }); } - if (projection.runtimeRequests.some((request) => request.status === "pending")) { + if ( + projection.runtimeRequests.some((request) => request.status === "pending") || + unansweredRemoteTaskRequests(projection.thread).length > 0 + ) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, @@ -7838,6 +7856,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } as OrchestrationV2TurnItem, }); } + yield* emitRemoteTaskRequests(events, command, parentProjection.thread, task.id, []); }); /** @@ -7878,6 +7897,219 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* emitRemoteStopOwed(events, command, task, task.remoteChild); }); + /** + * Replaces what the parent mirrors of one remote task's requests. A request + * answered here keeps its answer for as long as the thread there still waits on it. + */ + const emitRemoteTaskRequests = Effect.fn("orchestrationV2.dispatch.remoteTaskRequests")( + function* ( + events: Ref.Ref>, + command: OrchestrationV2InternalCommand | OrchestrationV2Command, + thread: OrchestrationV2AppThread, + taskId: NodeId, + requests: ReadonlyArray, + ) { + const { remoteTaskRequests: current = [], ...rest } = thread; + const next = [...current.filter((request) => request.taskId !== taskId), ...requests]; + const fingerprint = (entries: ReadonlyArray) => + JSON.stringify(entries, sortObjectKeys); + const shown = current.filter((request) => request.taskId === taskId); + if (fingerprint(requests) === fingerprint(shown)) return; + const now = yield* DateTime.now; + yield* emit( + events, + command, + )({ + type: "thread.metadata-updated", + threadId: thread.id, + occurredAt: now, + payload: { + ...rest, + ...(next.length === 0 ? {} : { remoteTaskRequests: next }), + updatedAt: now, + }, + }); + }, + ); + + const dispatchRemoteDelegatedTaskRequests = Effect.fn( + "orchestrationV2.dispatch.remoteDelegatedTaskRequests", + )(function* ( + command: Extract< + OrchestrationV2InternalCommand, + { readonly type: "delegated_task.remote.requests" } + >, + events: Ref.Ref>, + ) { + const { thread, subagents } = yield* projectionStore + .getThreadRecords(command.parentThreadId, ["subagents"]) + .pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: command.parentThreadId, cause }), + ), + ); + const task = subagents.find( + (candidate) => + candidate.id === command.taskId && + candidate.origin === "app_owned" && + candidate.remoteChild !== undefined, + ); + if (task === undefined) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Delegated task ${command.taskId} is not a remote task of thread ${command.parentThreadId}.`, + }); + } + const answers = new Map( + (thread.remoteTaskRequests ?? []) + .filter((request) => request.taskId === task.id) + .map((request) => [request.requestId, request.response]), + ); + const mirrored = + task.result !== null + ? [] + : command.requests.map((request) => { + const response = answers.get(request.requestId); + return { ...request, taskId: task.id, ...(response === undefined ? {} : { response }) }; + }); + yield* emitRemoteTaskRequests(events, command, thread, task.id, mirrored); + }); + + /** + * Records the user's answer to a request a remote task's thread waits on; its + * follower relays it there. False when `command` answers no such request. + */ + const dispatchRemoteTaskRequestRespond = Effect.fn( + "orchestrationV2.dispatch.remoteTaskRequestRespond", + )(function* ( + command: Extract, + events: Ref.Ref>, + ) { + const thread = yield* projectionStore + .getThread(command.threadId) + .pipe(mapDispatchError(command)); + const request = thread.remoteTaskRequests?.find( + (candidate) => candidate.requestId === command.requestId, + ); + if (request === undefined) return false; + const fail = (cause: string) => + new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause, + }); + if (request.response !== undefined) { + return yield* fail(`Runtime request ${command.requestId} is resolved.`); + } + if (Object.keys(command.attachmentsByQuestionId ?? {}).length > 0) { + return yield* fail("Attachments cannot reach a task in another environment."); + } + const dismissed = command.decision === "cancel" || command.decision === "decline"; + const response = + command.feedback !== undefined + ? undefined + : request.kind === "user_input" + ? command.answers !== undefined && command.decision === undefined + ? { answers: command.answers } + : command.answers === undefined && dismissed + ? { decision: command.decision! } + : undefined + : command.decision !== undefined && command.answers === undefined + ? { decision: command.decision } + : undefined; + if (response === undefined) { + return yield* fail("Answer an approval with a decision or a question with answers."); + } + yield* emitRemoteTaskRequests( + events, + command, + thread, + request.taskId, + (thread.remoteTaskRequests ?? []) + .filter((candidate) => candidate.taskId === request.taskId) + .map((candidate) => + candidate.requestId === request.requestId ? { ...candidate, response } : candidate, + ), + ); + return true; + }); + + const dispatchRemoteDelegatedTaskRespondFail = Effect.fn( + "orchestrationV2.dispatch.remoteDelegatedTaskRespondFail", + )(function* ( + command: Extract< + OrchestrationV2InternalCommand, + { readonly type: "delegated_task.remote.respond.fail" } + >, + events: Ref.Ref>, + ) { + const { thread, nodes } = yield* projectionStore + .getThreadRecords(command.parentThreadId, ["nodes"]) + .pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: command.parentThreadId, cause }), + ), + ); + const requests = (thread.remoteTaskRequests ?? []).filter( + (request) => request.taskId === command.taskId, + ); + if ( + !requests.some( + (request) => request.requestId === command.requestId && request.response !== undefined, + ) + ) + return; + yield* emitRemoteTaskRequests( + events, + command, + thread, + command.taskId, + requests.map((request) => { + if (request.requestId !== command.requestId) return request; + const { response: _answerThatFailed, ...unanswered } = request; + return unanswered; + }), + ); + const node = nodes.find((candidate) => candidate.id === command.taskId); + if (node === undefined) return; + const now = yield* DateTime.now; + yield* emit( + events, + command, + )({ + type: "turn-item.updated", + threadId: command.parentThreadId, + ...(node.runId === null ? {} : { runId: node.runId }), + nodeId: node.id, + providerInstanceId: thread.providerInstanceId, + occurredAt: now, + payload: { + id: TurnItemId.make(`turn-item:remote-request-answer-failure:${command.requestId}`), + threadId: command.parentThreadId, + runId: node.runId, + nodeId: node.id, + providerThreadId: node.providerThreadId, + providerTurnId: node.providerTurnId, + nativeItemRef: null, + parentItemId: null, + ordinal: yield* nextTurnItemOrdinal({ thread }), + status: "failed", + title: "Answer not delivered", + startedAt: now, + completedAt: now, + updatedAt: now, + type: "error", + failure: { + class: "transport_error", + message: command.message, + code: null, + retryable: null, + }, + }, + }); + }); + const emitRemoteStopOwed = Effect.fn("orchestrationV2.dispatch.remoteStopOwed")(function* ( events: Ref.Ref>, command: Extract< @@ -8419,6 +8651,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); const runtimeRequest = context.request; if (runtimeRequest === undefined) { + if (yield* dispatchRemoteTaskRequestRespond(command, events)) return; return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, @@ -12694,6 +12927,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "delegated_task.remote.stop": yield* dispatchRemoteDelegatedTaskStop(command, events); break; + case "delegated_task.remote.requests": + yield* dispatchRemoteDelegatedTaskRequests(command, events); + break; + case "delegated_task.remote.respond.fail": + yield* dispatchRemoteDelegatedTaskRespondFail(command, events); + break; case "delegated_task.wake-policy": yield* dispatchDelegatedTaskWakePolicy(command, events); break; @@ -12809,7 +13048,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command.type === "thread.stop" || command.type === "subagent.stop" || // Closing a share again, or after it ended, is not an error. - command.type === "thread.guest-share.close", + command.type === "thread.guest-share.close" || + // A follower's report repeats what the parent already shows. + command.type === "delegated_task.remote.requests" || + command.type === "delegated_task.remote.respond.fail", (_planned) => new OrchestratorDispatchError({ commandId: command.commandId, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 1e9176d1bed6..186f78a39a34 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1,4 +1,4 @@ -import { pendingGuestMessageCount } from "@t3tools/contracts"; +import { pendingGuestMessageCount, unansweredRemoteTaskRequests } from "@t3tools/contracts"; import { applyThreadPullRequestLinkUpdate } from "../sourceControl/threadPullRequests.ts"; import { projectTurnItemForWire } from "./WireProjection.ts"; import * as Stream from "effect/Stream"; @@ -1361,6 +1361,23 @@ function secretRequestAsPendingInput( }; } +/** The newest request a remote task's thread waits on that the user has not answered here. */ +function remoteTaskRequestAsPending( + thread: OrchestrationV2AppThread, +): OrchestrationV2PendingRuntimeRequestSummary | null { + const request = unansweredRemoteTaskRequests(thread).toSorted( + (left, right) => + DateTime.toEpochMillis(right.createdAt) - DateTime.toEpochMillis(left.createdAt), + )[0]; + if (request === undefined) return null; + return { + id: request.requestId, + kind: request.kind, + createdAt: request.createdAt, + blocking: true, + }; +} + /** * A secret card is answered while its agent waits in the run. A link card's * agent asked without waiting, so the card waits on the user after its run. @@ -1415,6 +1432,9 @@ export function threadShellFromProjection( )[0] ?? null, ) ?? pendingRuntimeRequest; } + if (pendingRuntimeRequest === null || pendingRuntimeRequest.blocking === false) { + pendingRuntimeRequest = remoteTaskRequestAsPending(projection.thread) ?? pendingRuntimeRequest; + } const userMessages = projection.messages .filter((message) => message.role === "user" && idleCompactionOf(message) === undefined) .toSorted( @@ -6198,6 +6218,9 @@ export const layer: Layer.Layer = )[0] ?? null, ) ?? pendingRuntimeRequest; } + if (pendingRuntimeRequest === null || pendingRuntimeRequest.blocking === false) { + pendingRuntimeRequest = remoteTaskRequestAsPending(thread) ?? pendingRuntimeRequest; + } let terminalFailureItem = row.terminal_failure_payload_json === null ? null diff --git a/apps/server/src/orchestration-v2/RemoteDelegatedTask.test.ts b/apps/server/src/orchestration-v2/RemoteDelegatedTask.test.ts index 4bb8c103f6df..d12f77d94087 100644 --- a/apps/server/src/orchestration-v2/RemoteDelegatedTask.test.ts +++ b/apps/server/src/orchestration-v2/RemoteDelegatedTask.test.ts @@ -10,6 +10,7 @@ import { ProviderInstanceId, ProviderThreadId, RunId, + RuntimeRequestId, ThreadId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; @@ -193,6 +194,26 @@ const remoteChild = { label: "Box", }; +/** A live parent with one task delegated to a linked environment, still running there. */ +const delegateRemotely = (name: string) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const parent = yield* seedLiveParent(name); + yield* orchestrator.dispatch({ + type: "delegated_task.remote.request", + commandId: CommandId.make(`command:remote-delegate:${name}`), + parentThreadId: parent.threadId, + parentRunId: parent.runId, + parentNodeId: parent.rootNodeId, + task: "Run a check.", + driver: ProviderDriverKind.make("remote"), + modelSelection, + remoteChild, + }); + const taskId = (yield* orchestrator.getThreadProjection(parent.threadId)).subagents[0]!.id; + return { parent, taskId }; + }); + it.layer(layerTest)("a delegated task in a linked environment", (it) => { it.effect("is recorded on its parent without a child thread here, once", () => Effect.gen(function* () { @@ -408,4 +429,202 @@ it.layer(layerTest)("a delegated task in a linked environment", (it) => { assert.equal(offered[0]?.threadId, parent.threadId); }), ); + it.effect("shows its thread's questions and approvals there on the parent, answered here", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const { parent, taskId } = yield* delegateRemotely("mirrors"); + const now = yield* DateTime.now; + const approval = { + requestId: RuntimeRequestId.make("request:there:approval"), + kind: "command" as const, + createdAt: now, + detail: "rm -rf build", + }; + const question = { + requestId: RuntimeRequestId.make("request:there:question"), + kind: "user_input" as const, + createdAt: DateTime.add(now, { seconds: 1 }), + questions: [{ id: "size", header: "Size", question: "Which size?", options: [] }], + }; + const mirror = (name: string, requests: ReadonlyArray) => + orchestrator.dispatch({ + type: "delegated_task.remote.requests", + commandId: CommandId.make(`command:mirror:${name}`), + parentThreadId: parent.threadId, + taskId, + requests, + }); + const pending = orchestrator + .getThreadShell(parent.threadId) + .pipe(Effect.map((shell) => shell?.pendingRuntimeRequest?.id ?? null)); + + yield* mirror("both", [approval, question]); + assert.equal(yield* pending, question.requestId); + + const wrongAnswer = yield* orchestrator + .dispatch({ + type: "runtime-request.respond", + commandId: CommandId.make("command:answer-approval-with-text"), + threadId: parent.threadId, + requestId: approval.requestId, + answers: { size: "Large" }, + }) + .pipe(Effect.flip); + assert.equal(wrongAnswer._tag, "OrchestratorDispatchError"); + yield* orchestrator.dispatch({ + type: "runtime-request.respond", + commandId: CommandId.make("command:answer-question"), + threadId: parent.threadId, + requestId: question.requestId, + answers: { size: "Large" }, + }); + // Answered here, the question waits on the relay, and the approval is what is left. + assert.equal(yield* pending, approval.requestId); + const answeredTwice = yield* orchestrator + .dispatch({ + type: "runtime-request.respond", + commandId: CommandId.make("command:answer-question-again"), + threadId: parent.threadId, + requestId: question.requestId, + answers: { size: "Small" }, + }) + .pipe(Effect.flip); + assert.equal(answeredTwice._tag, "OrchestratorDispatchError"); + + // The follower reads both still waiting there; the answer given here stands. + yield* mirror("both-again", [approval, question]); + const requests = (yield* orchestrator.getThreadProjection(parent.threadId)).thread + .remoteTaskRequests; + assert.deepEqual( + requests?.map((request) => [request.requestId, request.response]), + [ + [approval.requestId, undefined], + [question.requestId, { answers: { size: "Large" } }], + ], + ); + + // The question left the thread there once it got the answer. + yield* mirror("approval", [approval]); + assert.deepEqual( + (yield* orchestrator.getThreadProjection(parent.threadId)).thread.remoteTaskRequests?.map( + (request) => request.requestId, + ), + [approval.requestId], + ); + const settle = yield* orchestrator + .dispatch({ + type: "thread.settle", + commandId: CommandId.make("command:settle:mirrors"), + threadId: parent.threadId, + }) + .pipe(Effect.flip); + assert.include(String(settle.cause), "is waiting on an approval or question"); + const snooze = yield* orchestrator + .dispatch({ + type: "thread.snooze", + commandId: CommandId.make("command:snooze:mirrors"), + threadId: parent.threadId, + snoozedUntil: DateTime.formatIso(DateTime.add(now, { hours: 1 })), + }) + .pipe(Effect.flip); + assert.include(String(snooze.cause), "cannot be snoozed"); + + yield* orchestrator.dispatch({ + type: "delegated_task.remote.complete", + commandId: CommandId.make("command:complete:mirrors"), + parentThreadId: parent.threadId, + taskId, + status: "completed", + result: "Done there.", + stoppedThere: "ran-out", + }); + assert.equal(yield* pending, null); + // A late report from the follower mirrors nothing for a task that has its result. + yield* mirror("late", [approval]); + assert.equal(yield* pending, null); + }), + ); + + it.effect("records nothing when a task there still waits on what the parent shows", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const { parent, taskId } = yield* delegateRemotely("unchanged"); + yield* orchestrator.dispatch({ + type: "delegated_task.remote.request", + commandId: CommandId.make("command:remote-delegate:unchanged-second"), + parentThreadId: parent.threadId, + parentRunId: parent.runId, + parentNodeId: parent.rootNodeId, + task: "Run another check.", + driver: ProviderDriverKind.make("remote"), + modelSelection, + remoteChild, + }); + const secondTaskId = (yield* orchestrator.getThreadProjection( + parent.threadId, + )).subagents.find((task) => task.id !== taskId)!.id; + const now = yield* DateTime.now; + const mirror = (name: string, task: NodeId) => + orchestrator.dispatch({ + type: "delegated_task.remote.requests", + commandId: CommandId.make(`command:mirror:unchanged:${name}`), + parentThreadId: parent.threadId, + taskId: task, + requests: [ + { + requestId: RuntimeRequestId.make(`request:there:${task}`), + kind: "command", + createdAt: now, + }, + ], + }); + yield* mirror("first", taskId); + yield* mirror("second", secondTaskId); + // A restarted follower mirrors the first task's requests again. + const again = yield* mirror("first-again", taskId); + assert.deepEqual(again.storedEvents, []); + }), + ); + + it.effect("asks again here for an answer that could not reach its thread there", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const { parent, taskId } = yield* delegateRemotely("respond-fails"); + const requestId = RuntimeRequestId.make("request:there:refused"); + yield* orchestrator.dispatch({ + type: "delegated_task.remote.requests", + commandId: CommandId.make("command:mirror:respond-fails"), + parentThreadId: parent.threadId, + taskId, + requests: [{ requestId, kind: "command", createdAt: yield* DateTime.now }], + }); + yield* orchestrator.dispatch({ + type: "runtime-request.respond", + commandId: CommandId.make("command:approve:respond-fails"), + threadId: parent.threadId, + requestId, + decision: "accept", + }); + yield* orchestrator.dispatch({ + type: "delegated_task.remote.respond.fail", + commandId: CommandId.make("command:refused:respond-fails"), + parentThreadId: parent.threadId, + taskId, + requestId, + message: "This link is read-only.", + }); + + const projection = yield* orchestrator.getThreadProjection(parent.threadId); + assert.equal(projection.thread.remoteTaskRequests?.[0]?.response, undefined); + assert.equal( + (yield* orchestrator.getThreadShell(parent.threadId))?.pendingRuntimeRequest?.id, + requestId, + ); + const failure = projection.turnItems.find((item) => item.type === "error"); + assert.equal( + failure?.type === "error" ? failure.failure.message : undefined, + "This link is read-only.", + ); + }), + ); }); diff --git a/apps/server/src/orchestration-v2/SubagentProjection.test.ts b/apps/server/src/orchestration-v2/SubagentProjection.test.ts index d3506649f6a6..dd3930619acd 100644 --- a/apps/server/src/orchestration-v2/SubagentProjection.test.ts +++ b/apps/server/src/orchestration-v2/SubagentProjection.test.ts @@ -12,6 +12,7 @@ import { ProjectId, ProviderInstanceId, ProviderThreadId, + RuntimeRequestId, ThreadId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; @@ -79,7 +80,17 @@ function makeParentThread(): OrchestrationV2AppThread { } it("keeps a subagent child awake when its parent thread is snoozed", () => { - const parentThread = makeParentThread(); + const parentThread: OrchestrationV2AppThread = { + ...makeParentThread(), + remoteTaskRequests: [ + { + requestId: RuntimeRequestId.make("runtime-request:remote-question"), + kind: "user_input", + createdAt: snoozedAt, + taskId: NodeId.make("node:remote-task"), + }, + ], + }; const childProviderThreadId = ProviderThreadId.make("provider-thread:subagent-awake-child"); const parentNodeId = NodeId.make("node:subagent-parent"); const childThread = makeSubagentChildThread({ @@ -106,6 +117,7 @@ it("keeps a subagent child awake when its parent thread is snoozed", () => { assert.deepEqual(childThread.modelSelection, childModelSelection); assert.equal(childThread.activeProviderThreadId, childProviderThreadId); assert.isUndefined(childThread.historyOrigin); + assert.isUndefined(childThread.remoteTaskRequests); assert.deepEqual(childThread.lineage, { parentThreadId, relationshipToParent: "subagent", diff --git a/apps/server/src/orchestration-v2/ThreadForkService.test.ts b/apps/server/src/orchestration-v2/ThreadForkService.test.ts index b6b89758f7b5..e9ef5ec7f076 100644 --- a/apps/server/src/orchestration-v2/ThreadForkService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadForkService.test.ts @@ -3,6 +3,7 @@ import { ContextTransferId, MessageId, type ModelSelection, + NodeId, type OrchestrationV2AppThread, type OrchestrationV2Run, type OrchestrationV2ThreadProjection, @@ -10,6 +11,7 @@ import { ProviderInstanceId, ProviderThreadId, RunId, + RuntimeRequestId, ThreadId, type ThreadPullRequestLink, } from "@t3tools/contracts"; @@ -159,6 +161,29 @@ it.effect("keeps a fork awake when its source thread is snoozed", () => }), ); +it.effect("leaves the questions of the source's tasks in linked environments to the source", () => + Effect.gen(function* () { + const sourceRun = makeSourceRun("completed"); + const sourceProjection = makeSourceProjection(sourceRun); + const result = yield* planFork(sourceRun, { + ...sourceProjection, + thread: { + ...sourceProjection.thread, + remoteTaskRequests: [ + { + requestId: RuntimeRequestId.make("request:remote-question"), + taskId: NodeId.make("node:remote-task"), + kind: "user_input", + createdAt: forkCreatedAt, + }, + ], + }, + }); + + assert.isUndefined(result.targetThread.remoteTaskRequests); + }), +); + it.effect("preserves pull request context without inheriting any watches", () => Effect.gen(function* () { const sourceRun = makeSourceRun("completed"); diff --git a/apps/server/src/orchestration-v2/ThreadForkService.ts b/apps/server/src/orchestration-v2/ThreadForkService.ts index a660de9a4e6a..02fb93edaae9 100644 --- a/apps/server/src/orchestration-v2/ThreadForkService.ts +++ b/apps/server/src/orchestration-v2/ThreadForkService.ts @@ -80,8 +80,11 @@ export const layer: Layer.Layer = Layer.succeed( cause: forkableSourceRunStatusError(input.sourceRun), }); } + // A remote task's requests are answered on the thread that owns the task. + const { remoteTaskRequests: _remoteTaskRequests, ...sourceThread } = + input.sourceProjection.thread; const targetThread: OrchestrationV2AppThread = { - ...input.sourceProjection.thread, + ...sourceThread, // PR context follows the conversation; watching remains the source thread's responsibility. pullRequests: input.sourceProjection.thread.pullRequests?.map( ({ watch: _watch, ...link }) => link, diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 917fdc807c2f..6867735ac86a 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -112,6 +112,8 @@ function existingThreadIdsForCommand( case "delegated_task.remote.request": case "delegated_task.remote.complete": case "delegated_task.remote.stop": + case "delegated_task.remote.requests": + case "delegated_task.remote.respond.fail": return [command.parentThreadId]; case "thread.created.record": return command.parentThreadId === command.targetThreadId diff --git a/apps/server/src/peer/PeerForwarding.test.ts b/apps/server/src/peer/PeerForwarding.test.ts index 50be40682066..e6195c60bc35 100644 --- a/apps/server/src/peer/PeerForwarding.test.ts +++ b/apps/server/src/peer/PeerForwarding.test.ts @@ -35,6 +35,7 @@ import * as ServerSettings from "../serverSettings.ts"; import * as RemoteDelegation from "./RemoteDelegation.ts"; import * as McpHttpServer from "../mcp/McpHttpServer.ts"; import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; +import { FORWARDED_WAIT_CHUNK_MS } from "../mcp/toolkits/thread/tools.ts"; import { liveThreadShell } from "../mcp/McpToolAccess.testkit.ts"; import * as PeerForwarding from "./PeerForwarding.ts"; import * as PeerLinks from "./PeerLinks.ts"; @@ -451,9 +452,9 @@ it.effect("keeps a forwarded wait short and its retries scoped to the caller", ( expect( (yield* Ref.get(seen)).waits.map(({ timeoutMs, runId }) => [timeoutMs, runId]), ).toEqual([ - [PeerForwarding.FORWARDED_WAIT_CHUNK_MS, undefined], - [PeerForwarding.FORWARDED_WAIT_CHUNK_MS, boxRun], - [PeerForwarding.FORWARDED_WAIT_CHUNK_MS, boxRun], + [FORWARDED_WAIT_CHUNK_MS, undefined], + [FORWARDED_WAIT_CHUNK_MS, boxRun], + [FORWARDED_WAIT_CHUNK_MS, boxRun], ]); const send = (caller: string, clientRequestId: string) => diff --git a/apps/server/src/peer/PeerForwarding.ts b/apps/server/src/peer/PeerForwarding.ts index b8391e285c18..4159bbdb1739 100644 --- a/apps/server/src/peer/PeerForwarding.ts +++ b/apps/server/src/peer/PeerForwarding.ts @@ -21,16 +21,11 @@ import type * as Tool from "effect/ai/Tool"; import { assertNotLinked } from "../mcp/linkOrigin.ts"; import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; import { readCaller, unavailable } from "../mcp/threadAccess.ts"; +import { FORWARDED_WAIT_CHUNK_MS, LinkRequestsTool } from "../mcp/toolkits/thread/tools.ts"; import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; import * as PeerLinks from "./PeerLinks.ts"; import * as PeerMcpClient from "./PeerMcpClient.ts"; -/** - * A peer's `/mcp` holds a wait open for its whole timeout without sending - * anything, and T3 Connect's edge drops a request idle for about 100 s, so a - * forwarded wait goes out in pieces no longer than this. - */ -export const FORWARDED_WAIT_CHUNK_MS = 50_000; const DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1_000; const MAX_WAIT_TIMEOUT_MS = 60 * 60 * 1_000; @@ -73,6 +68,12 @@ export class PeerForwarding extends Context.Service< environmentId: EnvironmentId, input: OrchestratorMcpThreadWaitInput, ) => Effect.Effect; + /** `t3_link_requests` there, as the environment that delegated the thread. */ + readonly linkRequests: ( + scope: Scope, + environmentId: EnvironmentId, + input: Tool.Parameters, + ) => Effect.Effect, OrchestratorMcpFailure>; /** The environments this one links to, as an agent sees them. */ readonly links: ( scope: Scope, @@ -179,6 +180,16 @@ const make = Effect.gen(function* () { input, ); + const linkRequests: PeerForwarding["Service"]["linkRequests"] = (scope, environmentId, input) => + forward>( + scope, + LinkRequestsTool.name, + LinkRequestsTool.parametersSchema, + LinkRequestsTool.successSchema, + environmentId, + input, + ); + const waitForThread: PeerForwarding["Service"]["waitForThread"] = Effect.fn( "PeerForwarding.waitForThread", )(function* (scope, environmentId, input) { @@ -212,7 +223,8 @@ const make = Effect.gen(function* () { }); // While the beta is off, agents start nothing over links. Tasks already running there keep - // following through readThread, waitForThread and interruptThread, so they still finish. + // following through readThread, waitForThread, interruptThread and linkRequests, so they + // still finish. const requireEnabled = links.enabled.pipe( Effect.flatMap((enabled) => (enabled ? Effect.void : Effect.fail(linkedEnvironmentsOff))), ); @@ -268,6 +280,7 @@ const make = Effect.gen(function* () { readThread, interruptThread, waitForThread, + linkRequests, links: listLinks, }); }); diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index a344f63c523a..a7adc0305f25 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -7,14 +7,19 @@ import { EventId, MessageId, NodeId, + type OrchestrationV2AppThread, type OrchestrationV2Run, + type OrchestrationV2RuntimeRequest, + type OrchestrationV2TurnItem, ProjectId, + ProviderSessionId, ProviderDriverKind, ProviderInstanceId, ProviderThreadId, type Project, type RepositoryIdentity, RunId, + RuntimeRequestId, ThreadId, TurnItemId, } from "@t3tools/contracts"; @@ -46,6 +51,7 @@ import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterReg import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; import * as ThreadLaunch from "../orchestration-v2/ThreadLaunchService.ts"; import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; +import * as ThreadSearch from "../orchestration-v2/ThreadSearch.ts"; import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts"; @@ -127,6 +133,12 @@ interface BoxThread { readonly sent: Ref.Ref>; /** Runs as the launch there is answered, before the laptop hears back. */ readonly afterLaunch: Ref.Ref>; + /** The questions and approvals the thread there has asked, answered or not. */ + readonly requests: SubscriptionRef.SubscriptionRef>; + /** The answers the thread there was given, in order. */ + readonly answers: Ref.Ref>; + /** While set, the thread there refuses an answer with this. */ + readonly refuseAnswers: Ref.Ref; } const makeBoxThread = Effect.gen(function* () { @@ -161,6 +173,9 @@ const makeBoxThread = Effect.gen(function* () { releaseStop: yield* Deferred.make(), sent: yield* Ref.make>([]), afterLaunch: yield* Ref.make>(Effect.void), + requests: yield* SubscriptionRef.make>([]), + answers: yield* Ref.make>([]), + refuseAnswers: yield* Ref.make(null), } satisfies BoxThread; }); @@ -204,6 +219,39 @@ const serveBox = (thread: BoxThread) => { const boxRunId = RunId.make("run:box"); const boxLaterRunId = RunId.make("run:box-later"); +const boxApprovalId = RuntimeRequestId.make("request:box-approval"); +const boxApproval: OrchestrationV2RuntimeRequest = { + id: boxApprovalId, + nodeId: NodeId.make("node:box"), + providerTurnId: null, + nativeRequestRef: null, + kind: "command", + status: "pending", + responseCapability: { type: "live", providerSessionId: ProviderSessionId.make("session:box") }, + createdAt: DateTime.makeUnsafe(0), + resolvedAt: null, +}; +const boxApprovalItem: OrchestrationV2TurnItem = { + id: TurnItemId.make("item:box-approval"), + threadId: ThreadId.make("thread:box"), + runId: boxRunId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 0, + status: "waiting", + title: null, + startedAt: DateTime.makeUnsafe(0), + completedAt: null, + updatedAt: DateTime.makeUnsafe(0), + type: "approval_request", + requestId: boxApprovalId, + requestKind: "command", + prompt: "pnpm test --update-snapshots", +}; + const boxToolkitLayer = (thread: BoxThread) => { const runId = boxRunId; const interrupts = new Set(); @@ -235,7 +283,12 @@ const boxToolkitLayer = (thread: BoxThread) => { const runless = yield* Ref.get(thread.runless); return shell === null ? (null as never) - : { ...idleThreadProjection(shell), runs: runless ? [] : [runOf(status)] }; + : { + ...idleThreadProjection(shell), + runs: runless ? [] : [runOf(status)], + runtimeRequests: yield* SubscriptionRef.get(thread.requests), + turnItems: [boxApprovalItem], + }; }); // One record per command id, as its receipt replays a repeat. const recordStop = (commandId: string, kind: "interrupt" | "thread.stop" | "delegated-tasks") => @@ -250,6 +303,11 @@ const boxToolkitLayer = (thread: BoxThread) => { }); const layerBoxThreads = Layer.mock(ThreadManagement.ThreadManagementService)({ getThreadShell: shellOf, + getThreadEventSequence: () => Effect.succeed(0), + streamStoredEventsFrom: (input) => + input?.eventType === "runtime-request.updated" + ? (SubscriptionRef.changes(thread.requests).pipe(Stream.drop(1)) as never) + : Stream.never, getProjectThreadRecords: (input) => projectionOf(input.threadId), getThreadRecords: (threadId) => projectionOf(threadId), waitForThread: (input) => @@ -396,18 +454,36 @@ const boxToolkitLayer = (thread: BoxThread) => { Effect.as({ type: "interrupt_requested", run: runOf("running") } as never), ), dispatch: (command) => - command.type === "thread.stop" - ? recordStop(command.commandId, "thread.stop").pipe( - Effect.tap(() => - Effect.gen(function* () { - if (!(yield* Ref.getAndSet(thread.holdFirstStop, false))) return; - yield* Deferred.succeed(thread.stopStarted, undefined); - yield* Deferred.await(thread.releaseStop); - }), - ), - Effect.as({ sequence: 0, storedEvents: [] } as never), - ) - : Effect.die(`The box does not expect ${command.type}`), + command.type === "runtime-request.respond" + ? Effect.gen(function* () { + const refusal = yield* Ref.get(thread.refuseAnswers); + if (refusal !== null) { + return yield* new Orchestrator.OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: refusal, + }); + } + yield* Ref.update(thread.answers, (seen) => [...seen, command.decision]); + yield* SubscriptionRef.update(thread.requests, (requests) => + requests.map((request) => + request.id === command.requestId ? { ...request, status: "resolved" } : request, + ), + ); + return { sequence: 0, storedEvents: [] } as never; + }) + : command.type === "thread.stop" + ? recordStop(command.commandId, "thread.stop").pipe( + Effect.tap(() => + Effect.gen(function* () { + if (!(yield* Ref.getAndSet(thread.holdFirstStop, false))) return; + yield* Deferred.succeed(thread.stopStarted, undefined); + yield* Deferred.await(thread.releaseStop); + }), + ), + Effect.as({ sequence: 0, storedEvents: [] } as never), + ) + : Effect.die(`The box does not expect ${command.type}`), stopDelegatedTasks: (input) => recordStop(input.commandId, "delegated-tasks"), sendToThread: (input) => Ref.update(thread.sent, (seen) => [...seen, input.text]).pipe( @@ -459,10 +535,12 @@ const boxToolkitLayer = (thread: BoxThread) => { ), ), }); - return Layer.merge( + return Layer.mergeAll( McpHttpServer.layerOrchestratorToolkit, McpHttpServer.layerProjectRegistration, + McpHttpServer.layerThreadToolkit, ).pipe( + Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})), Layer.provide(DocumentText.layer.pipe(Layer.provideMerge(NodeServices.layer))), Layer.provide(NodeCrypto.layer), Layer.provide(layerBoxThreads), @@ -1751,3 +1829,127 @@ it.effect("a local task runs in this thread's project", () => }), ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + +it.effect("the laptop's parent shows an approval the task waits on there, and answers it", () => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + yield* SubscriptionRef.set(thread.requests, [boxApproval]); + yield* Ref.set(thread.refuseAnswers, "The approval's provider session ended."); + 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 delegated = yield* delegateToBox(a); + /** The parent's record once `ready` holds for it. */ + const parentWhen = (ready: (thread: OrchestrationV2AppThread) => boolean) => + a.orchestrator + .streamStoredEventsFrom({ + threadId: parentThreadId, + afterSequence: 0, + eventType: "thread.metadata-updated", + }) + .pipe( + Stream.map((stored) => stored.event.payload as OrchestrationV2AppThread), + Stream.filter(ready), + Stream.runHead, + ); + const approve = (attempt: string) => + a.orchestrator.dispatch({ + type: "runtime-request.respond", + commandId: CommandId.make(`command:approve-on-laptop:${attempt}`), + threadId: parentThreadId, + requestId: boxApprovalId, + decision: "accept", + }); + + yield* parentWhen((parent) => parent.remoteTaskRequests?.length === 1); + expect( + (yield* a.orchestrator.getThreadShell(parentThreadId))?.pendingRuntimeRequest, + ).toMatchObject({ id: boxApprovalId, kind: "command", blocking: true }); + expect( + (yield* a.call("task_status", { taskId: delegated.taskId })).structuredContent, + ).toMatchObject({ + status: "running", + pendingApproval: { requestId: boxApprovalId, runId: boxRunId }, + }); + + // The box refuses the first answer while it still waits, so the laptop asks again. + yield* approve("refused"); + const failure = yield* a.orchestrator + .streamStoredEventsFrom({ + threadId: parentThreadId, + afterSequence: 0, + eventType: "turn-item.updated", + }) + .pipe( + Stream.map((stored) => stored.event.payload as OrchestrationV2TurnItem), + Stream.filter((item) => item.type === "error"), + Stream.runHead, + ); + expect( + Option.getOrUndefined(failure)?.type === "error" ? failure.pipe(Option.getOrThrow) : null, + ).toMatchObject({ failure: { message: "The approval's provider session ended." } }); + expect( + (yield* a.orchestrator.getThreadProjection(parentThreadId)).thread.remoteTaskRequests?.[0] + ?.response, + ).toBeUndefined(); + + yield* Ref.set(thread.refuseAnswers, null); + yield* approve("accepted"); + yield* parentWhen((parent) => (parent.remoteTaskRequests ?? []).length === 0); + expect(yield* Ref.get(thread.answers)).toEqual(["accept"]); + expect( + (yield* a.orchestrator.getThreadShell(parentThreadId))?.pendingRuntimeRequest, + ).toBeNull(); + + yield* Ref.set(thread.status, "completed"); + yield* Deferred.succeed(thread.finished, undefined); + yield* taskResult(a, delegated.taskId); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect( + "a wait on a task there ends on its approval, and its result still wakes the parent", + () => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + yield* SubscriptionRef.set(thread.requests, [boxApproval]); + 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 waited = yield* a.call("delegate_task", { + task: "Run the full suite on the box.", + target: { + environmentId: box.environmentId, + providerInstanceId: instanceId, + model: "gpt-5.4", + }, + mode: "wait", + clientRequestId: "suite-on-box-waited", + }); + expect(waited.structuredContent).toMatchObject({ + status: "running", + waitTimedOut: false, + pendingApproval: { requestId: boxApprovalId }, + }); + const taskId = (waited.structuredContent as { taskId: NodeId }).taskId; + const task = (yield* a.orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (candidate) => candidate.id === taskId, + ); + expect(task?.completionWake).toBe("always"); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index 739edc0a6039..ea07088634a8 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -6,12 +6,15 @@ import { type OrchestratorMcpDelegateTaskInput, type OrchestratorMcpTaskCancelResult, type OrchestratorMcpThreadTimelineItem, + type OrchestrationV2LinkedRequest, type OrchestrationV2RemoteTaskChild, + type OrchestrationV2RemoteTaskRequest, type OrchestrationV2Subagent, type ProjectId, type ProviderInstanceId, ProviderDriverKind, type RunId, + type RuntimeRequestId, type ThreadId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; @@ -27,6 +30,7 @@ import type * as Tool from "effect/ai/Tool"; import type * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; import { ProjectToolkit } from "../mcp/toolkits/project/tools.ts"; +import { FORWARDED_WAIT_CHUNK_MS } from "../mcp/toolkits/thread/tools.ts"; import { resolveInteractionMode, resolveRuntimeMode, @@ -45,21 +49,25 @@ import * as PeerLinks from "./PeerLinks.ts"; const FOLLOW_WAIT_MS = 10 * 60 * 1_000; /** A linked environment that stops answering is retried, backing off to this. */ const FOLLOW_MAX_BACKOFF = Duration.minutes(5); +const followBackoff = Schedule.exponential(Duration.seconds(2)).pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, FOLLOW_MAX_BACKOFF)), + ), +); /** How a follower retries; tests step it by hand. */ export const FollowRetrySchedule = Context.Reference>( "t3/peer/RemoteDelegation/FollowRetrySchedule", - { - defaultValue: () => - Schedule.exponential(Duration.seconds(2)).pipe( - Schedule.modifyDelay(({ duration }) => - Effect.succeed(Duration.min(duration, FOLLOW_MAX_BACKOFF)), - ), - ), - }, + { defaultValue: () => followBackoff }, ); type ThreadScope = McpInvocationContext.McpThreadInvocationScope; +/** What a task's relay has sent there, and the version of its requests it last mirrored. */ +interface RelayState { + readonly relayed: Set; + version: string | undefined; +} + /** * `delegate_task` to a linked environment. The task's child is an ordinary * thread there, launched through the link; the parent here records the task @@ -505,6 +513,153 @@ const make = Effect.gen(function* () { /** A task's thread there that ended without a reply. */ const noReply = "(The thread in the linked environment ended without a reply.)"; + const mirrorRequests = Effect.fn("RemoteDelegation.mirrorRequests")(function* ( + parentThreadId: ThreadId, + taskId: NodeId, + requests: ReadonlyArray, + ) { + const attempt = yield* crypto.randomUUIDv4().pipe(Effect.orDie); + yield* threads.dispatch({ + type: "delegated_task.remote.requests", + commandId: CommandId.make(`command:remote-task-requests:${taskId}:${attempt}`), + parentThreadId, + taskId, + requests, + }); + }); + + /** + * Sends an answer the user gave here to the task's thread there; true once it is there. An + * answer refused there while the thread still waits on its request is asked for again here; + * one refused because the request was answered there is left for the mirror to drop. + */ + const relayAnswer = Effect.fn("RemoteDelegation.relayAnswer")(function* ( + parentThreadId: ThreadId, + scope: ThreadScope, + remote: OrchestrationV2RemoteTaskChild, + request: OrchestrationV2RemoteTaskRequest, + ) { + // A request is answered at most once there, so every attempt can be its own call. + const attempt = yield* crypto.randomUUIDv4().pipe(Effect.orDie); + const refusal = yield* forwarding + .linkRequests(scope, remote.environmentId, { + action: "respond", + threadId: remote.threadId, + requestId: request.requestId, + ...request.response, + clientRequestId: `answer:${attempt}`, + }) + .pipe( + Effect.as(undefined), + Effect.catch((error) => Effect.succeed(error)), + ); + if (refusal === undefined) return true; + const there = yield* forwarding.linkRequests(scope, remote.environmentId, { + action: "watch", + threadId: remote.threadId, + timeoutMs: 1, + }); + const stillWaits = + "requests" in there && there.requests.some((open) => open.requestId === request.requestId); + if (stillWaits) { + yield* threads.dispatch({ + type: "delegated_task.remote.respond.fail", + commandId: CommandId.make( + `command:remote-task-answer-refused:${request.requestId}:${attempt}`, + ), + parentThreadId, + taskId: request.taskId, + requestId: request.requestId, + message: refusal.message, + }); + } + return false; + }); + + /** + * One pass of relaying requests for a task still running there: sends the answers given here + * that have not reached it, then waits until what its thread there waits on changes or the + * user answers here. + */ + const relayRequestsOnce = Effect.fn("RemoteDelegation.relayRequestsOnce")(function* ( + parentThreadId: ThreadId, + taskId: NodeId, + relay: RelayState, + ) { + const sequence = yield* threads.getThreadEventSequence(parentThreadId); + const { thread, task } = yield* readTask(parentThreadId, taskId); + if (task?.remoteChild === undefined || task.result !== null) return yield* Effect.never; + const remote = task.remoteChild; + const scope = followScope( + { threadId: parentThreadId, providerInstanceId: thread.providerInstanceId }, + remote, + ); + const mirrored = (thread.remoteTaskRequests ?? []).filter( + (request) => request.taskId === taskId, + ); + for (const request of mirrored) { + if (request.response === undefined || relay.relayed.has(request.requestId)) continue; + if (yield* relayAnswer(parentThreadId, scope, remote, request)) { + relay.relayed.add(request.requestId); + } + } + const answeredHere = threads + .streamStoredEventsFrom({ + threadId: parentThreadId, + afterSequence: sequence, + eventType: "thread.metadata-updated", + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "thread.metadata-updated" && + (stored.event.payload.remoteTaskRequests ?? []).some( + (request) => + request.taskId === taskId && + request.response !== undefined && + !relay.relayed.has(request.requestId), + ), + ), + Stream.runHead, + Effect.flatMap((answered) => (Option.isSome(answered) ? Effect.void : Effect.never)), + ); + const watched = yield* forwarding + .linkRequests(scope, remote.environmentId, { + action: "watch", + threadId: remote.threadId, + knownVersion: relay.version, + timeoutMs: FORWARDED_WAIT_CHUNK_MS, + }) + .pipe(Effect.raceFirst(answeredHere)); + if (watched === undefined || !("requests" in watched) || watched.version === relay.version) { + return; + } + yield* mirrorRequests(parentThreadId, taskId, watched.requests); + relay.version = watched.version; + }); + + /** + * Shows on the parent the questions and approvals the task's thread there waits on, and + * relays the user's answers given here, for as long as its follower runs. + */ + const relayRequests = (parentThreadId: ThreadId, taskId: NodeId): Effect.Effect => { + const relay: RelayState = { relayed: new Set(), version: undefined }; + // Its own back-off, so stepping the follower's retries never steps this one. A peer from + // before `t3_link_requests` refuses every pass, and is asked again as it may update. + return relayRequestsOnce(parentThreadId, taskId, relay).pipe( + Effect.tapError((error) => + Effect.logDebug("remote-delegation.relay-requests-retrying", { taskId, error }), + ), + Effect.retry(followBackoff), + Effect.forever, + Effect.catchCause((cause) => + Effect.logWarning("remote-delegation.relay-requests-failed", { taskId, cause }).pipe( + Effect.andThen(Effect.never), + ), + ), + ); + }; + /** * Waits for the task's run there to end, completes the task here, and stops it there if it * ended here first. A revoked, expired or forgotten link, or a thread or run gone there, @@ -527,6 +682,7 @@ const make = Effect.gen(function* () { }), Effect.retry(followRetry), Effect.ignoreCause({ log: true }), + Effect.raceFirst(relayRequests(parentThreadId, taskId)), ); /** diff --git a/apps/server/src/provider/AgentContextBudget.test.ts b/apps/server/src/provider/AgentContextBudget.test.ts index ad19948c0688..2ebb6b42d045 100644 --- a/apps/server/src/provider/AgentContextBudget.test.ts +++ b/apps/server/src/provider/AgentContextBudget.test.ts @@ -1,10 +1,11 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { expect, it } from "@effect/vitest"; -import { ProviderInstanceId } from "@t3tools/contracts"; +import { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; import { McpProviderSessions } from "@t3tools/provider-core/server/McpProviderSessions"; +import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; -import { McpServer } from "effect/ai"; +import { McpSchema, McpServer } from "effect/ai"; import { ServerSecretStore } from "../auth/ServerSecretStore.ts"; import { ServerConfig } from "../config.ts"; @@ -12,6 +13,7 @@ import { DeviceService } from "../device/DeviceService.ts"; import { ServerEnvironment } from "../environment/ServerEnvironment.ts"; import { GitWorkflowService } from "../git/GitWorkflowService.ts"; import * as McpHttpServer from "../mcp/McpHttpServer.ts"; +import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; import { PreviewAutomationBroker } from "../mcp/PreviewAutomationBroker.ts"; import { CLAUDE_PROVIDER, @@ -103,18 +105,42 @@ const codexAdditionalContext = () => .map((entry) => entry.value) .join(""); +/** An agent's own session: tools hidden from it by `EnabledWhen` cost it nothing. */ +const agentScope: McpInvocationContext.McpInvocationScope = { + environmentId: EnvironmentId.make("environment:context-budget"), + requestNamespace: "provider-session:context-budget", + thread: { + threadId: ThreadId.make("thread:context-budget"), + providerSessionId: "provider-session:context-budget", + providerInstanceId: ProviderInstanceId.make("codex"), + }, + client: undefined, + capabilities: new Set(["orchestration"]), + issuedAt: 0, +}; + +const toolsListedToAgents = McpServer.McpServer.pipe( + Effect.map((mcp) => + mcp.tools.filter(({ annotations }) => { + const enabledWhen = Context.getOrUndefined(annotations, McpSchema.EnabledWhen); + return enabledWhen === undefined || enabledWhen({ protocolVersion: "", capabilities: {} }); + }), + ), + Effect.provideService(McpInvocationContext.McpInvocationContext, agentScope), +); + // CI rejects a PR that raises any value above main's agentContextBudget.json, so growth // needs the agent-context-increase label. Lower the values when the text shrinks. it.effect("agent context matches the committed budget", () => Effect.gen(function* () { - const mcp = yield* McpServer.McpServer; + const tools = yield* toolsListedToAgents; const measured = { claudeSystemAppendBytes: utf8Bytes(claudeSystemAppend()), codexAdditionalContextBytes: utf8Bytes(codexAdditionalContext()), // The model sees these fields. outputSchema and annotations are protocol metadata. mcpToolListJsonBytes: utf8Bytes( JSON.stringify( - mcp.tools.map(({ tool }) => ({ + tools.map(({ tool }) => ({ name: tool.name, description: tool.description, inputSchema: tool.inputSchema, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 4796a8465e24..63d10d8ea259 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -1288,6 +1288,7 @@ export default function ChatView(props: ChatViewProps) { }); const { pendingApprovals, + composerApprovals, pendingUserInputs, activePendingApproval, activePendingUserInput, @@ -2805,7 +2806,7 @@ export default function ChatView(props: ChatViewProps) { needsLoadBalancing, loadBalancing, readPendingEnvironmentChange, - pendingApprovals, + pendingApprovals: composerApprovals, respondingRequestIds, activePendingProgress, onAdvanceActivePendingUserInput, @@ -3444,7 +3445,10 @@ export default function ChatView(props: ChatViewProps) { environmentId={environmentId} attachmentUploadsCapabilityKnown={attachmentUploadsCapabilityKnown} supportsAttachmentUploads={supportsAttachmentUploads} - supportsQuestionAttachments={supportsQuestionAttachments} + supportsQuestionAttachments={ + supportsQuestionAttachments && + activePendingUserInput?.remoteTaskId === undefined + } maxFileAttachmentBytes={maxFileAttachmentBytes} routeKind={routeKind} routeThreadRef={routeThreadRef} @@ -3499,8 +3503,8 @@ export default function ChatView(props: ChatViewProps) { : undefined } environmentUnavailable={activeEnvironmentUnavailableState} - activePendingApproval={activePendingApproval} - pendingApprovals={pendingApprovals} + activePendingApproval={composerApprovals[0] ?? null} + pendingApprovals={composerApprovals} onHeldRequestIdsChange={onHeldRequestIdsChange} pendingUserInputs={pendingUserInputs} activePendingProgress={activePendingProgress} diff --git a/apps/web/src/components/ComposerPromptEditorTiptap.tsx b/apps/web/src/components/ComposerPromptEditorTiptap.tsx index abc9549ee3d0..4cd2c2bf8910 100644 --- a/apps/web/src/components/ComposerPromptEditorTiptap.tsx +++ b/apps/web/src/components/ComposerPromptEditorTiptap.tsx @@ -121,7 +121,7 @@ import { Tooltip, TooltipPopup, TooltipTrigger } from "./ui/tooltip"; import { ComposerCodeBlockNodeView } from "./chat/ComposerCodeBlockNodeView"; import { resolveAssistantCitationCommentDismissal } from "./chat/assistantCitationCommentDismissal"; import { composerCodeBlockHighlight } from "./composerCodeBlockHighlight"; -import { importPastedComposerText } from "./composerInlineTokenPaste"; +import { type ImportedContextIds, importPastedComposerText } from "./composerInlineTokenPaste"; import { didComposerSelectionChangeVisibly } from "./composerSelection"; import type { ComposerDraftContextRecords } from "./composerContextPresentation"; @@ -151,7 +151,7 @@ export interface ComposerPromptEditorProps { | ((contextIds: ReadonlyArray) => string | null) | undefined; importContextFragment?: - | ((fragment: ComposerContextClipboardFragment) => ReadonlyMap) + | ((fragment: ComposerContextClipboardFragment) => ImportedContextIds) | undefined; skills: ReadonlyArray; disabled: boolean; diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index 57a3a8aa4190..7cfe276dc692 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -398,7 +398,7 @@ export interface ChatComposerProps { } | null; activePendingApproval: PendingApproval | null; - pendingApprovals: PendingApproval[]; + pendingApprovals: ReadonlyArray; onHeldRequestIdsChange: (requestIds: ReadonlySet) => void; pendingUserInputs: PendingUserInput[]; activePendingProgress: { @@ -1388,6 +1388,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) attachmentTargetKey, attachmentTargetKeyRef, questionAttachmentTarget, + supportsQuestionAttachments, pendingUserInputs, composerContextRecords, composerImages, diff --git a/apps/web/src/components/chat/ComposerPendingUserInputPanel.tsx b/apps/web/src/components/chat/ComposerPendingUserInputPanel.tsx index 697a0e0bfd6f..f9fda1eacb69 100644 --- a/apps/web/src/components/chat/ComposerPendingUserInputPanel.tsx +++ b/apps/web/src/components/chat/ComposerPendingUserInputPanel.tsx @@ -196,6 +196,15 @@ const ComposerPendingUserInputCard = memo(function ComposerPendingUserInputCard( > {activeQuestion.header} + {prompt.remoteEnvironmentLabel ? ( + <> + + + Delegated task on + {prompt.remoteEnvironmentLabel} + + + ) : null} {isCollapsed ? ( <> diff --git a/apps/web/src/components/chat/TimelinePendingApprovalRow.tsx b/apps/web/src/components/chat/TimelinePendingApprovalRow.tsx index c425d560bf13..304ebadf7b53 100644 --- a/apps/web/src/components/chat/TimelinePendingApprovalRow.tsx +++ b/apps/web/src/components/chat/TimelinePendingApprovalRow.tsx @@ -65,6 +65,11 @@ export const TimelinePendingApprovalRow = memo(function TimelinePendingApprovalR } >
+ {approval.remoteEnvironmentLabel ? ( + + Delegated task on {approval.remoteEnvironmentLabel} + + ) : null} {approval.appName ? ( {approval.appName} ) : null} diff --git a/apps/web/src/components/chat/useComposerContextImport.ts b/apps/web/src/components/chat/useComposerContextImport.ts index e859e816074e..ed09032881c4 100644 --- a/apps/web/src/components/chat/useComposerContextImport.ts +++ b/apps/web/src/components/chat/useComposerContextImport.ts @@ -35,6 +35,7 @@ import { terminalContextDraftFromRecord, terminalContextRecord, terminalContextReference, + pastedContextRecordFits, } from "~/lib/composerContextRecords"; import { assetEnvironment } from "~/state/assets"; import { copiedComposerAttachmentFile } from "~/lib/copiedComposerAttachments"; @@ -42,6 +43,7 @@ import { countQuestionAttachments, questionAttachmentDraftId } from "../../quest import { elementContextToPreviewAnnotation } from "../../lib/elementContext"; import { orchestrationEnvironment } from "~/state/orchestration"; import { pendingDraftWork } from "./pendingDraftWork"; +import type { ImportedContextIds } from "../composerInlineTokenPaste"; import { randomUUID } from "~/lib/utils"; import { readPreparedConnection } from "~/state/session"; import { resolveAssetUrl } from "~/assets/assetUrls"; @@ -83,6 +85,7 @@ export function useComposerContextImport(input: { readonly attachmentTargetKey: string; readonly attachmentTargetKeyRef: RefObject; readonly questionAttachmentTarget: DraftId | null; + readonly supportsQuestionAttachments: boolean; readonly pendingUserInputs: ReadonlyArray; readonly composerContextRecords: ComposerDraftContextRecords; readonly composerImages: ReadonlyArray; @@ -100,6 +103,7 @@ export function useComposerContextImport(input: { attachmentTargetKey, attachmentTargetKeyRef, questionAttachmentTarget, + supportsQuestionAttachments, pendingUserInputs, composerContextRecords, composerImages, @@ -271,10 +275,14 @@ export function useComposerContextImport(input: { records: ReadonlyArray, sourceEnvironmentId: EnvironmentId | null, threadDraftTarget = attachmentDraftTarget, - ): ReadonlyMap => { - const rewritten = new Map(); + ): ImportedContextIds => { + const rewritten = new Map(); const dependentAttachmentLocalIds = new Map(); const skippedDependentAttachmentIds = new Set(); + const target = { + answersQuestion: questionAttachmentTarget !== null, + takesAttachments: supportsQuestionAttachments, + }; // Resolve annotations before their dependent screenshot records even if a foreign // clipboard producer emitted the records in a different order. const orderedRecords = records.toSorted((left, right) => @@ -336,6 +344,10 @@ export function useComposerContextImport(input: { continue; } const conflicts = existing !== undefined; + if (!pastedContextRecordFits(record.kind, target)) { + rewritten.set(record.contextId, null); + continue; + } switch (record.kind) { case "terminal": { const threadId = activeThread?.id ?? activeThreadId; @@ -388,9 +400,7 @@ export function useComposerContextImport(input: { break; } case "environment": { - // A machine means the same machine from any thread, so a pasted one is kept, - // but not in an answer to an agent's question, which carries no machines. - if (questionAttachmentTarget !== null) break; + // A machine means the same machine from any thread, so a pasted one is kept. addComposerDraftEnvironmentContexts(composerDraftTarget, [record], { appendReference: false, }); @@ -416,9 +426,7 @@ export function useComposerContextImport(input: { } return rewritten; }; - const importContextFragment = ( - fragment: ComposerContextClipboardFragment, - ): ReadonlyMap => + const importContextFragment = (fragment: ComposerContextClipboardFragment): ImportedContextIds => importContextRecords(fragment.records, fragment.source.environmentId); const importContextFragmentStable = useStableCallback(importContextFragment); diff --git a/apps/web/src/components/chat/useComposerSend.ts b/apps/web/src/components/chat/useComposerSend.ts index a5fa30ffe15a..f0794998bfab 100644 --- a/apps/web/src/components/chat/useComposerSend.ts +++ b/apps/web/src/components/chat/useComposerSend.ts @@ -211,7 +211,7 @@ interface ComposerSendInput { readonly needsLoadBalancing: boolean; readonly loadBalancing: DraftEnvironmentRouting["loadBalancing"]; readonly readPendingEnvironmentChange: DraftEnvironmentRouting["readPendingEnvironmentChange"]; - readonly pendingApprovals: PendingThreadRequests["pendingApprovals"]; + readonly pendingApprovals: PendingThreadRequests["composerApprovals"]; readonly respondingRequestIds: PendingThreadRequests["respondingRequestIds"]; readonly activePendingProgress: PendingThreadRequests["activePendingProgress"]; readonly onAdvanceActivePendingUserInput: PendingThreadRequests["onAdvanceActivePendingUserInput"]; diff --git a/apps/web/src/components/chat/useComposerStash.ts b/apps/web/src/components/chat/useComposerStash.ts index 6f092cd2f008..4a62ac3349d1 100644 --- a/apps/web/src/components/chat/useComposerStash.ts +++ b/apps/web/src/components/chat/useComposerStash.ts @@ -19,6 +19,10 @@ import { useComposerDraftStore, } from "../../composerDraftStore"; import { type ComposerPromptEditorHandle } from "../ComposerPromptEditorTiptap"; +import { + type ImportedContextIds, + rewriteImportedContextReferences, +} from "../composerInlineTokenPaste"; import { MAX_STASH_ENTRIES, type PromptStashEntry, @@ -29,7 +33,6 @@ import { type PendingUserInput } from "../../session-logic"; import { type ReviewCommentContext } from "../../reviewCommentContext"; import { collapseExpandedComposerCursor } from "../../composer-logic"; import { compressImageForStash } from "../../lib/imageCompression"; -import { formatInlineContextReference } from "~/lib/composerContextReferences"; import { isCommandPaletteOpen } from "../../commandPaletteBus"; import { pendingDraftWork } from "./pendingDraftWork"; import { @@ -45,7 +48,6 @@ import { verifyStashedAttachmentUpload, } from "../../lib/attachmentUploadQueue"; import { remainingComposerAttachmentSlots } from "./composerThreadImport"; -import { replaceComposerContextReferences } from "@t3tools/shared/composerContextReferences"; import { resizeSnapShotSource } from "../../lib/snapShotSource"; import { registerCommands } from "../../shortcuts/dispatcher"; import { selectThreadTerminalUiState, useTerminalUiStateStore } from "../../terminalUiStateStore"; @@ -94,7 +96,7 @@ export function useComposerStash(input: { records: ReadonlyArray, sourceEnvironmentId: EnvironmentId | null, threadDraftTarget?: ChatComposerProps["composerDraftTarget"], - ) => ReadonlyMap; + ) => ImportedContextIds; readonly expandMobileComposer: () => void; }) { const { @@ -203,19 +205,12 @@ export function useComposerStash(input: { }); } - const rewrittenContextIds = entry.records - ? importContextRecords(entry.records, null, composerDraftTarget) - : new Map(); - const restoredPrompt = replaceComposerContextReferences(entry.prompt, (reference) => { - const contextId = rewrittenContextIds.get(reference.contextId); - return contextId - ? formatInlineContextReference({ - ...reference, - contextId, - kind: reference.kind === "element" ? "preview-annotation" : reference.kind, - }) - : reference.source; - }); + const restoredPrompt = entry.records + ? rewriteImportedContextReferences( + entry.prompt, + importContextRecords(entry.records, null, composerDraftTarget), + ) + : entry.prompt; const currentPrompt = promptRef.current; const nextPrompt = restoredPrompt.length === 0 diff --git a/apps/web/src/components/chat/usePendingThreadRequests.ts b/apps/web/src/components/chat/usePendingThreadRequests.ts index 35e4cc5f0855..dfa1cd20c2f9 100644 --- a/apps/web/src/components/chat/usePendingThreadRequests.ts +++ b/apps/web/src/components/chat/usePendingThreadRequests.ts @@ -12,6 +12,7 @@ import { derivePendingThreadRequests, onThreadRequestsResolved, pruneResolvedUserInputDrafts, + redirectableApprovals, seedUserInputDraftAnswers, userInputDraftKey, } from "@t3tools/client-runtime/state/thread-requests"; @@ -254,6 +255,7 @@ export function usePendingThreadRequests(input: { respondingUserInputRequestIds.includes(activePendingUserInput.requestId) : false; const activePendingApproval = pendingApprovals[0] ?? null; + const composerApprovals = redirectableApprovals(pendingApprovals); const onRespondToApproval = async ( requestId: RuntimeRequestId, @@ -487,6 +489,7 @@ export function usePendingThreadRequests(input: { return { pendingApprovals, + composerApprovals, pendingUserInputs, activePendingApproval, activePendingUserInput, diff --git a/apps/web/src/components/composerInlineTokenPaste.test.ts b/apps/web/src/components/composerInlineTokenPaste.test.ts new file mode 100644 index 000000000000..13a71fde5e45 --- /dev/null +++ b/apps/web/src/components/composerInlineTokenPaste.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { rewriteImportedContextReferences } from "./composerInlineTokenPaste"; + +describe("rewriteImportedContextReferences", () => { + const pasted = + "See ![shot.png](t3-context://v1/image/img_a) and [notes.md](t3-context://v1/file/file_b) in [build](t3-context://v1/terminal/term_c)"; + + it.each([ + [ + "points an imported record at its new id", + new Map([["img_a", "img_z"]]), + "See ![shot.png](t3-context://v1/image/img_z) and [notes.md](t3-context://v1/file/file_b) in [build](t3-context://v1/terminal/term_c)", + ], + [ + "pastes a declined record as its label", + new Map([ + ["img_a", null], + ["file_b", null], + ]), + "See shot.png and notes.md in [build](t3-context://v1/terminal/term_c)", + ], + ])("%s", (_, importedIds, expected) => { + expect(rewriteImportedContextReferences(pasted, importedIds)).toBe(expected); + }); +}); diff --git a/apps/web/src/components/composerInlineTokenPaste.ts b/apps/web/src/components/composerInlineTokenPaste.ts index 32722ad6a957..f78cf11e0dff 100644 --- a/apps/web/src/components/composerInlineTokenPaste.ts +++ b/apps/web/src/components/composerInlineTokenPaste.ts @@ -37,28 +37,34 @@ export function readPastedComposerContext( }; } +// null marks a record the target declined; its reference pastes as plain text. +export type ImportedContextIds = ReadonlyMap; + +export function rewriteImportedContextReferences( + text: string, + importedIds: ImportedContextIds, +): string { + if (importedIds.size === 0) return text; + return replaceComposerContextReferences(text, (occurrence) => { + const nextId = importedIds.get(occurrence.contextId); + if (nextId === null) return occurrence.label; + return nextId + ? formatComposerContextReference({ + ...occurrence, + contextId: ComposerContextId.make(nextId), + kind: occurrence.kind === "element" ? "preview-annotation" : occurrence.kind, + }) + : occurrence.source; + }); +} + export function importPastedComposerText( clipboardData: Pick, - importContextFragment?: ( - fragment: ComposerContextClipboardFragment, - ) => ReadonlyMap, + importContextFragment?: (fragment: ComposerContextClipboardFragment) => ImportedContextIds, ): string { const pastedText = clipboardData.getData("text/plain"); const fragment = importContextFragment ? readPastedComposerContext(clipboardData) : null; - const rewrittenIds = - fragment && fragment.records.length > 0 ? importContextFragment!(fragment) : null; - const text = - rewrittenIds && rewrittenIds.size > 0 - ? replaceComposerContextReferences(pastedText, (occurrence) => { - const nextId = rewrittenIds.get(occurrence.contextId); - return nextId - ? formatComposerContextReference({ - ...occurrence, - contextId: ComposerContextId.make(nextId), - kind: occurrence.kind === "element" ? "preview-annotation" : occurrence.kind, - }) - : occurrence.source; - }) - : pastedText; - return text; + return fragment && fragment.records.length > 0 + ? rewriteImportedContextReferences(pastedText, importContextFragment!(fragment)) + : pastedText; } diff --git a/apps/web/src/lib/composerContextRecords.test.ts b/apps/web/src/lib/composerContextRecords.test.ts index f612a302ac8c..c64d1d8d0f84 100644 --- a/apps/web/src/lib/composerContextRecords.test.ts +++ b/apps/web/src/lib/composerContextRecords.test.ts @@ -20,6 +20,7 @@ import { buildMessageContext, composerContextImportLookupIds, isSameComposerContextPayload, + pastedContextRecordFits, previewAnnotationContextRecord, previewAnnotationFromRecord, pullRequestContextKindLabel, @@ -67,6 +68,38 @@ const annotation: PreviewAnnotationPayload = { }; describe("composerContextRecords", () => { + it.each([ + ["image", "a message", { answersQuestion: false, takesAttachments: false }, true], + ["environment", "a message", { answersQuestion: false, takesAttachments: false }, true], + [ + "file", + "a question that takes files", + { answersQuestion: true, takesAttachments: true }, + true, + ], + [ + "image", + "a question that takes none", + { answersQuestion: true, takesAttachments: false }, + false, + ], + [ + "file", + "a question that takes none", + { answersQuestion: true, takesAttachments: false }, + false, + ], + [ + "thread", + "a question that takes none", + { answersQuestion: true, takesAttachments: false }, + true, + ], + ["environment", "any question", { answersQuestion: true, takesAttachments: true }, false], + ] as const)("a pasted %s record fits %s: %s", (kind, _, target, fits) => { + expect(pastedContextRecordFits(kind, target)).toBe(fits); + }); + it("copies only ready or persisted server-side attachment IDs", () => { const environmentId = EnvironmentId.make("env"); const image = { diff --git a/apps/web/src/lib/composerContextRecords.ts b/apps/web/src/lib/composerContextRecords.ts index c4f123f73dcc..b35e9983bdbb 100644 --- a/apps/web/src/lib/composerContextRecords.ts +++ b/apps/web/src/lib/composerContextRecords.ts @@ -347,6 +347,16 @@ export function asKnownContextRecord( return record; } +/** An answer to an agent's question carries no machines, and files only when the question takes them. */ +export function pastedContextRecordFits( + kind: KnownComposerContextRecord["kind"], + target: { readonly answersQuestion: boolean; readonly takesAttachments: boolean }, +): boolean { + if (!target.answersQuestion) return true; + if (kind === "environment") return false; + return target.takesAttachments || (kind !== "image" && kind !== "file"); +} + export function composerContextImportLookupIds( record: KnownComposerContextRecord, ): ReadonlyArray { diff --git a/docs/internals/remote.md b/docs/internals/remote.md index d88d2973f4e4..b78443c148ca 100644 --- a/docs/internals/remote.md +++ b/docs/internals/remote.md @@ -133,6 +133,17 @@ internal command reuses the parent half of a local finalize, so the parent wakes the same way. Open remote tasks are followed again at startup, which is also what keeps restart recovery from treating them as abandoned provider work. +The user's phone or browser may reach only the parent's environment, so the +follower also mirrors what the task's thread there waits on onto the parent's +record (`remoteTaskRequests`, through `t3_link_requests`). The parent's shell +reports it as its pending request, and an answer there takes the normal +`runtime-request.respond` path; the follower relays it through the link, which +holds it to the link's access and fence like any other write. An answer refused +there while the request still waits is asked for again on the parent. A message +sent on the parent redirects only its own agent, so it never declines a mirrored +approval. A peer without `t3_link_requests` keeps its requests to itself, as +before. + The link is routing, not isolation: an agent the link starts runs as the receiving environment's user. **A full-access link is therefore not fenced**: its agents can already do anything that user can, so the fence would only stop diff --git a/docs/user/remote-access.md b/docs/user/remote-access.md index 835476864a1f..d9649c0c6882 100644 --- a/docs/user/remote-access.md +++ b/docs/user/remote-access.md @@ -297,7 +297,10 @@ there as an ordinary thread, in the project with the same repository (or the one `target.projectId` names), and this thread wakes with its result when it ends, as for any delegated task. It keeps following the task across restarts. If the other machine revokes the link, or the link is removed here, the task -fails with that reason. +fails with that reason. When the task asks a question or needs an approval, +this thread shows it, so you can answer from any device connected to this +machine. A machine on an older T3 Code version keeps those requests to itself; +answer them there. ## Manage or revoke access diff --git a/packages/client-runtime/src/state/threadDetail.ts b/packages/client-runtime/src/state/threadDetail.ts index a53494ebc592..069bbe21daef 100644 --- a/packages/client-runtime/src/state/threadDetail.ts +++ b/packages/client-runtime/src/state/threadDetail.ts @@ -369,8 +369,11 @@ export function createEnvironmentThreadDetailAtoms( ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-worktree:${key}`)), ); const pendingRequestsAtomFamily = Atom.family((key: string) => { - let previous: Pick | null = - null; + let previous: + | (Pick & { + readonly thread: Pick; + }) + | null = null; let value: PendingThreadRequests | null = null; return Atom.make((get) => { const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data); @@ -391,12 +394,24 @@ export function createEnvironmentThreadDetailAtoms( (item.type === "approval_request" || item.type === "user_input_request") && ids.has(item.requestId), ); + const remoteTaskRequests = projection.thread.remoteTaskRequests; + const subagents = + remoteTaskRequests === undefined + ? [] + : projection.subagents.filter((task) => task.remoteChild !== undefined); if ( previous === null || !arrayElementsEqual(previous.runtimeRequests, runtimeRequests) || - !arrayElementsEqual(previous.turnItems, turnItems) + !arrayElementsEqual(previous.turnItems, turnItems) || + !arrayElementsEqual(previous.subagents, subagents) || + previous.thread.remoteTaskRequests !== remoteTaskRequests ) { - previous = { runtimeRequests, turnItems }; + previous = { + runtimeRequests, + turnItems, + subagents, + thread: remoteTaskRequests === undefined ? {} : { remoteTaskRequests }, + }; value = derivePendingThreadRequests(previous); } return value; diff --git a/packages/client-runtime/src/state/threadRequests.test.ts b/packages/client-runtime/src/state/threadRequests.test.ts index d5d2c4ec16d8..0dd6f8dad279 100644 --- a/packages/client-runtime/src/state/threadRequests.test.ts +++ b/packages/client-runtime/src/state/threadRequests.test.ts @@ -3,6 +3,7 @@ import { NodeId, ProviderSessionId, RuntimeRequestId, + ThreadId, TurnItemId, type OrchestrationV2ThreadProjection, } from "@t3tools/contracts"; @@ -13,6 +14,7 @@ import { v2Now, v2Projection } from "./orchestrationV2TestFixtures.ts"; import { createQuestionHistoryProjector, derivePendingThreadRequests, + redirectableApprovals, declinePendingApprovals, sendApprovalFeedback, makeThreadRequestsResolvedAnnouncer, @@ -316,6 +318,73 @@ describe("pending v2 questions", () => { }; expect(derivePendingThreadRequests(live).userInputs[0]?.dismissible).toBe(false); }); + it("lists what a task in a linked environment waits on until the user answers it here", () => { + const remote = { + runtimeRequests: [], + turnItems: [], + subagents: [ + { + id: NodeId.make("remote-task"), + remoteChild: { + environmentId: EnvironmentId.make("environment-box"), + threadId: ThreadId.make("thread-on-the-box"), + label: "Box", + }, + }, + ], + thread: { + remoteTaskRequests: [ + { + requestId: RuntimeRequestId.make("remote-approval"), + taskId: NodeId.make("remote-task"), + kind: "command" as const, + createdAt: v2Now, + detail: "pnpm test", + }, + { + requestId: RuntimeRequestId.make("remote-question"), + taskId: NodeId.make("remote-task"), + kind: "user_input" as const, + createdAt: v2Now, + questions: [{ id: "size", header: "Size", question: "Which size?", options: [] }], + }, + { + requestId: RuntimeRequestId.make("remote-answered"), + taskId: NodeId.make("remote-task"), + kind: "command" as const, + createdAt: v2Now, + response: { decision: "accept" as const }, + }, + ], + }, + }; + const pending = derivePendingThreadRequests(remote); + expect(pending.approvals).toEqual([ + { + requestId: "remote-approval", + requestKind: "command", + createdAt: "2026-06-20T00:00:00.000Z", + detail: "pnpm test", + responseCapability: "live", + remoteTaskId: "remote-task", + remoteEnvironmentLabel: "Box", + }, + ]); + expect(redirectableApprovals(pending.approvals)).toEqual([]); + expect(pending.userInputs).toEqual([ + { + requestId: "remote-question", + createdAt: "2026-06-20T00:00:00.000Z", + questions: [ + { id: "size", header: "Size", question: "Which size?", options: [], multiSelect: false }, + ], + responseCapability: "live", + dismissible: false, + remoteTaskId: "remote-task", + remoteEnvironmentLabel: "Box", + }, + ]); + }); }); it("restores old text answers without mutating history or replacing unchanged rows", () => { diff --git a/packages/client-runtime/src/state/threadRequests.ts b/packages/client-runtime/src/state/threadRequests.ts index e089eb36d08c..ec3f42f3192e 100644 --- a/packages/client-runtime/src/state/threadRequests.ts +++ b/packages/client-runtime/src/state/threadRequests.ts @@ -1,14 +1,18 @@ -import type { - EnvironmentId, - ThreadId, - OrchestrationV2ThreadProjection, - OrchestrationV2RuntimeRequest, - OrchestrationV2UserInputQuestion, - ProviderApprovalFullAccessOverride, - ProviderApprovalOption, - ProviderApprovalDecision, - ProviderRequestKind, - RuntimeRequestId, +import { + unansweredRemoteTaskRequests, + type EnvironmentId, + type NodeId, + type ThreadId, + type OrchestrationV2AppThread, + type OrchestrationV2Subagent, + type OrchestrationV2ThreadProjection, + type OrchestrationV2RuntimeRequest, + type OrchestrationV2UserInputQuestion, + type ProviderApprovalFullAccessOverride, + type ProviderApprovalOption, + type ProviderApprovalDecision, + type ProviderRequestKind, + type RuntimeRequestId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import type { AtomCommandResult } from "./runtime.ts"; @@ -31,6 +35,14 @@ export async function sendApprovalFeedback( return (await respond(approval.requestId, "decline", feedback.trim()))?._tag === "Success"; } +/** A message sent here redirects this thread's agent, never a task waiting in another environment. */ +export const redirectableApprovals = ( + approvals: ReadonlyArray, +): ReadonlyArray => + approvals.some((approval) => approval.remoteTaskId !== undefined) + ? approvals.filter((approval) => approval.remoteTaskId === undefined) + : approvals; + /** Declines every approval a redirect replaces; any one still pending would block the agent. */ export async function declinePendingApprovals( approvals: ReadonlyArray, @@ -55,6 +67,8 @@ export interface ThreadPendingApproval { readonly options?: ReadonlyArray; readonly responseCapability: "live" | "not_resumable"; readonly supportsFeedback?: boolean; + readonly remoteTaskId?: NodeId; + readonly remoteEnvironmentLabel?: string; } export const FULL_ACCESS_OVERRIDE_QUESTION = "Why am I being asked this?"; @@ -81,6 +95,8 @@ export interface ThreadPendingUserInput { readonly responseCapability: OrchestrationV2RuntimeRequest["responseCapability"]["type"]; readonly responseMode?: "message"; readonly dismissible: boolean; + readonly remoteTaskId?: NodeId; + readonly remoteEnvironmentLabel?: string; } export interface PendingThreadRequests { @@ -213,16 +229,27 @@ export function makeThreadRequestsResolvedAnnouncer( }; } +type WithRemoteTaskRequests = { + readonly thread?: Pick; + readonly subagents?: ReadonlyArray>; +}; + export function pendingRuntimeRequestIds( - projection: Pick, + projection: Pick & WithRemoteTaskRequests, ): RuntimeRequestId[] { - return projection.runtimeRequests.flatMap((request) => - request.status === "pending" ? [request.id] : [], - ); + return [ + ...projection.runtimeRequests.flatMap((request) => + request.status === "pending" ? [request.id] : [], + ), + ...remoteTaskRequestsOf(projection).map((request) => request.requestId), + ]; } +const remoteTaskRequestsOf = (projection: WithRemoteTaskRequests) => + projection.thread === undefined ? [] : unansweredRemoteTaskRequests(projection.thread); + export function derivePendingThreadRequests( - projection: Pick, + projection: Pick & WithRemoteTaskRequests, ): PendingThreadRequests { const approvals: ThreadPendingApproval[] = []; const userInputs: ThreadPendingUserInput[] = []; @@ -276,6 +303,40 @@ export function derivePendingThreadRequests( }); } + for (const request of remoteTaskRequestsOf(projection)) { + const createdAt = DateTime.formatIso(request.createdAt); + const environmentLabel = projection.subagents?.find((task) => task.id === request.taskId) + ?.remoteChild?.label; + const remoteTask = { + remoteTaskId: request.taskId, + ...(environmentLabel === undefined ? {} : { remoteEnvironmentLabel: environmentLabel }), + }; + if (request.kind === "user_input") { + userInputs.push({ + requestId: request.requestId, + createdAt, + questions: (request.questions ?? []).map((question) => ({ + ...question, + multiSelect: question.multiSelect ?? false, + })), + responseCapability: "live", + dismissible: false, + ...remoteTask, + }); + continue; + } + approvals.push({ + requestId: request.requestId, + requestKind: request.kind, + createdAt, + ...(request.detail ? { detail: request.detail } : {}), + ...(request.reason ? { reason: request.reason } : {}), + ...(request.options === undefined ? {} : { options: request.options }), + responseCapability: "live", + ...remoteTask, + }); + } + return { approvals, userInputs }; } diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 993fb62453d5..6daefedc80ef 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -433,6 +433,71 @@ const LenientThreadOpens = Schema.NullOr(OrchestrationV2ThreadOpens).pipe( Schema.catchDecoding(() => Effect.succeedSome(null)), ); +export const OrchestrationV2UserInputQuestion = Schema.Struct({ + id: TrimmedNonEmptyString, + header: TrimmedNonEmptyString, + question: TrimmedNonEmptyString, + options: Schema.Array( + Schema.Struct({ + label: TrimmedNonEmptyString, + description: TrimmedNonEmptyString, + value: Schema.optional(Schema.String), + preview: Schema.optional(Schema.String), + }), + ), + multiSelect: Schema.optional(Schema.Boolean), + allowCustomAnswer: Schema.optional(Schema.Boolean), + initialAnswer: Schema.optional(Schema.String), + required: Schema.optional(Schema.Boolean), +}); +export type OrchestrationV2UserInputQuestion = typeof OrchestrationV2UserInputQuestion.Type; + +/** + * A question or approval a thread waits on, as the environment that delegated + * it a task reads it through their link. `runId` is the run there it blocks. + */ +export const OrchestrationV2LinkedRequest = Schema.Struct({ + requestId: RuntimeRequestId, + kind: Schema.Union([ProviderRequestKind, Schema.Literal("user_input")]), + createdAt: Schema.DateTimeUtc, + runId: Schema.optional(RunId), + questions: Schema.optional(Schema.Array(OrchestrationV2UserInputQuestion)), + detail: Schema.optional(Schema.String), + reason: Schema.optional(Schema.String), + options: Schema.optional(Schema.Array(ProviderApprovalOption)), +}); +export type OrchestrationV2LinkedRequest = typeof OrchestrationV2LinkedRequest.Type; + +export const OrchestrationV2LinkedRequestJson = OrchestrationV2LinkedRequest.mapFields( + (fields) => ({ ...fields, createdAt: Schema.DateTimeUtcFromString }), +); + +/** + * A request the thread of a task delegated to a linked environment waits on, + * mirrored on the parent here so the user answers it here. `response` is the + * answer given here, owed to the thread there until it resolves the request. + */ +export const OrchestrationV2RemoteTaskRequest = Schema.Struct({ + ...OrchestrationV2LinkedRequest.fields, + taskId: NodeId, + response: Schema.optional( + Schema.Struct({ + decision: Schema.optional(ProviderApprovalDecision), + answers: Schema.optional(ProviderUserInputAnswers), + }), + ), +}); +export type OrchestrationV2RemoteTaskRequest = typeof OrchestrationV2RemoteTaskRequest.Type; + +const OrchestrationV2RemoteTaskRequestJson = OrchestrationV2RemoteTaskRequest.mapFields( + (fields) => ({ ...fields, createdAt: Schema.DateTimeUtcFromString }), +); + +export const unansweredRemoteTaskRequests = ( + thread: Pick, +): ReadonlyArray => + thread.remoteTaskRequests?.filter((request) => request.response === undefined) ?? []; + export const OrchestrationV2AppThread = Schema.Struct({ ...OrchestrationV2CreationFields, id: ThreadId, @@ -451,6 +516,8 @@ export const OrchestrationV2AppThread = Schema.Struct({ historyOrigin: Schema.optional(OrchestrationV2ThreadHistoryOrigin), linkOrigin: Schema.optional(OrchestrationV2LinkOrigin), delegatedFrom: Schema.optional(OrchestrationV2DelegatedFrom), + /** Questions and approvals of this thread's tasks in linked environments. */ + remoteTaskRequests: Schema.optional(Schema.Array(OrchestrationV2RemoteTaskRequest)), lineage: OrchestrationV2AppThreadLineage, forkedFrom: Schema.NullOr( Schema.Union([ @@ -1296,25 +1363,6 @@ export const OrchestrationV2PlanStep = Schema.Struct({ }); export type OrchestrationV2PlanStep = typeof OrchestrationV2PlanStep.Type; -export const OrchestrationV2UserInputQuestion = Schema.Struct({ - id: TrimmedNonEmptyString, - header: TrimmedNonEmptyString, - question: TrimmedNonEmptyString, - options: Schema.Array( - Schema.Struct({ - label: TrimmedNonEmptyString, - description: TrimmedNonEmptyString, - value: Schema.optional(Schema.String), - preview: Schema.optional(Schema.String), - }), - ), - multiSelect: Schema.optional(Schema.Boolean), - allowCustomAnswer: Schema.optional(Schema.Boolean), - initialAnswer: Schema.optional(Schema.String), - required: Schema.optional(Schema.Boolean), -}); -export type OrchestrationV2UserInputQuestion = typeof OrchestrationV2UserInputQuestion.Type; - const OrchestrationV2PlanArtifactBaseFields = { id: PlanId, threadId: ThreadId, @@ -2209,6 +2257,7 @@ export type OrchestrationV2StoredEvent = typeof OrchestrationV2StoredEvent.Type; export const OrchestrationV2AppThreadJson = OrchestrationV2AppThread.mapFields((fields) => ({ ...fields, + remoteTaskRequests: Schema.optional(Schema.Array(OrchestrationV2RemoteTaskRequestJson)), createdAt: Schema.DateTimeUtcFromString, updatedAt: Schema.DateTimeUtcFromString, archivedAt: Schema.NullOr(Schema.DateTimeUtcFromString), @@ -3395,6 +3444,30 @@ const OrchestrationV2InternalCommand = Schema.Union([ parentThreadId: ThreadId, taskId: NodeId, }), + /** + * Mirrors what a remote task's thread there waits on now, as its follower read it there, + * so the user answers it on the parent here. An answer given here stays until the request + * leaves the thread there; a task that has its result mirrors nothing. + */ + Schema.Struct({ + type: Schema.Literal("delegated_task.remote.requests"), + commandId: CommandId, + parentThreadId: ThreadId, + taskId: NodeId, + requests: Schema.Array(OrchestrationV2LinkedRequest), + }), + /** + * The answer given here to a remote task's request could not reach its thread there. The + * request is unanswered here again, and the parent shows why. + */ + Schema.Struct({ + type: Schema.Literal("delegated_task.remote.respond.fail"), + commandId: CommandId, + parentThreadId: ThreadId, + taskId: NodeId, + requestId: RuntimeRequestId, + message: Schema.String, + }), Schema.Struct({ type: Schema.Literal("checkpoint.rollback.fail"), commandId: CommandId, diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index f83a0b376233..e8f0eb1d4abb 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -54,8 +54,10 @@ export function makeSubagentChildThread(input: { readonly createdBy: OrchestrationV2Actor; readonly creationSource: OrchestrationV2CreationSource; }): OrchestrationV2AppThread { + // The parent answers its remote tasks' requests; a child has none of its own yet. + const { remoteTaskRequests: _remoteTaskRequests, ...parentThread } = input.parentThread; return { - ...input.parentThread, + ...parentThread, createdBy: input.createdBy, creationSource: input.creationSource, id: input.childThreadId,