Skip to content

fix(orchestrator): Wake settled parents when delegated children finish #4499

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mwolson wants to merge 1 commit into
base: t3code/codex-turn-mapping
Choose a base branch
from
Open

fix(orchestrator): Wake settled parents when delegated children finish #4499

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 50 additions & 5 deletions apps/server/src/mcp/OrchestratorMcpService.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1053,6 +1053,10 @@ const make = Effect.gen(function* () {
modelSelection: target.modelSelection,
runtimeMode,
interactionMode,
// Async delegations wake the parent on every child terminal; wait
// delegations deliver through the blocking tool call, so a wake is
// only needed if the parent settled first (timeout, disconnect).
completionWake: input.mode === "wait" ? "settled_only" : "always",
})
.pipe(
Effect.mapError((error) =>
Expand All @@ -1072,18 +1076,59 @@ const make = Effect.gen(function* () {
"Delegated task command did not produce a task projection.",
);
}
const taskId = taskEvent.event.payload.id;

if (input.mode !== "wait") {
return yield* readTask(scope, taskEvent.event.payload.id);
return yield* readTask(scope, taskId);
}
const timeoutMs = Math.min(
MAX_WAIT_TIMEOUT_MS,
Math.max(1, input.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS),
);
const waited = yield* waitForTask(scope, taskEvent.event.payload.id, timeoutMs);
return Option.isSome(waited)
? waited.value
: yield* readTask(scope, taskEvent.event.payload.id, true);
const waited = yield* waitForTask(scope, taskId, timeoutMs);
if (Option.isSome(waited)) {
return waited.value;
}
// The blocking wait timed out, so it no longer owns delivery: upgrade
// the task so a later terminal wakes the parent even mid-turn. Best
// effort; on failure the settled_only policy still wakes a settled
// parent.
yield* threadManagement
.dispatch({
type: "delegated_task.wake-policy",
commandId: stableCommandId({
scope,
requestKey: key,
operation: "delegate-task-wake-policy",
}),
parentThreadId: scope.threadId,
taskId,
completionWake: "always",
})
.pipe(
// The tool result is the timed-out task either way, so failures
// stay warnings. Keep the two shapes apart: a rejected receipt
// means this exact command id already failed (a replay of a
// no-op upgrade), while anything else is a fresh dispatch fault.
Effect.catch((error) =>
Effect.logWarning("orchestrator-mcp.delegate-task.wake-policy-failed", {
taskId,
outcome:
error._tag === "OrchestratorCommandPreviouslyRejectedError"
? "previously_rejected"
: "dispatch_failed",
error,
}),
),
Effect.catchCause((cause) =>
Effect.logWarning("orchestrator-mcp.delegate-task.wake-policy-failed", {
taskId,
outcome: "defect",
cause,
}),
),
);
return yield* readTask(scope, taskId, true);
}),
taskStatus: (scope, taskId) => readTask(scope, taskId),
cancelTask: (scope, input) =>
Expand Down
227 changes: 226 additions & 1 deletion apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,10 @@ import {
type ProviderAdapterV2TurnInput,
} from "../orchestration-v2/ProviderAdapter.ts";
import { makeLayer as makeProviderAdapterRegistryLayer } from "../orchestration-v2/ProviderAdapterRegistry.ts";
import {
type ProviderContinuationRequest,
ProviderContinuationRequests,
} from "../orchestration-v2/ProviderContinuationRequests.ts";
import { checkpointWorkspace } from "../orchestration-v2/testkit/ReplayFixtureWorkspace.ts";
import { makeOrchestratorV2ReplayLayerWithRegistry } from "../orchestration-v2/testkit/ProviderReplayHarness.ts";
import { makeProviderRegistryLayer } from "../provider/testUtils/providerRegistryMock.ts";
Expand Down Expand Up @@ -411,6 +415,38 @@ describe("orchestrator MCP toolkit", () => {
: `Claude completed: ${turn.message.text}`,
}),
]);
// Captures parent-wake offers made when a delegated child
// terminalizes after the parent run settled.
const continuationOffers = yield* Ref.make<ReadonlyArray<ProviderContinuationRequest>>(
[],
);
const continuationProbeLayer = Layer.succeed(ProviderContinuationRequests, {
offer: (request) =>
Ref.update(continuationOffers, (existing) => [...existing, request]),
take: Effect.never,
});
// Offers land after the finalize projection writes, so poll briefly
// instead of asserting counts immediately.
const waitForContinuationOffers = (count: number) =>
Effect.gen(function* () {
for (let attempt = 0; attempt < 1_000; attempt += 1) {
const current = yield* Ref.get(continuationOffers);
if (current.length >= count) {
return current;
}
yield* Effect.sleep("5 millis");
}
return yield* Ref.get(continuationOffers);
});
// Absence has no event to await, so sample repeatedly instead of
// trusting a single sleep to outlast a late offer.
const expectOffersToStay = (count: number) =>
Effect.gen(function* () {
for (let sample = 0; sample < 4; sample += 1) {
yield* Effect.sleep("50 millis");
expect(yield* Ref.get(continuationOffers)).toHaveLength(count);
}
});
const orchestratorLayer = makeOrchestratorV2ReplayLayerWithRegistry(
{
name: "orchestrator-mcp-toolkit",
Expand All @@ -425,7 +461,7 @@ describe("orchestrator MCP toolkit", () => {
},
},
registryLayer,
);
).pipe(Layer.provide(continuationProbeLayer));
const orchestrationLayer = Layer.merge(
orchestratorLayer,
threadManagementServiceLayer.pipe(Layer.provide(orchestratorLayer)),
Expand Down Expand Up @@ -781,6 +817,11 @@ describe("orchestrator MCP toolkit", () => {
expect(delegatedStatus.status).toBe("completed");
expect(delegatedStatus.resultContextTransferId).not.toBeNull();

// A wait-mode child (completionWake settled_only) that completes
// while the parent run is live does not offer a wake: the
// blocking delegate_task call above already returned the result.
yield* expectOffersToStay(0);

const repeatedDelegatedCall = yield* invoke("delegate_task", {
task: delegatedPrompt,
target: {
Expand Down Expand Up @@ -886,6 +927,18 @@ describe("orchestrator MCP toolkit", () => {
).pipe(Effect.orDie);
expect(cancelledStatus.status).toBe("interrupted");

// The cancelled async child carries completionWake "always", so
// its terminal offers a wake even though the parent run is live;
// queue_after_active sequences the continuation behind it.
const offersAfterCancel = yield* waitForContinuationOffers(1);
expect(offersAfterCancel).toHaveLength(1);
expect(offersAfterCancel[0]).toMatchObject({
threadId: parentThreadId,
delivery: "message_text",
});
expect(offersAfterCancel[0]?.detail).toContain(cancellable.taskId);
expect(offersAfterCancel[0]?.detail).toContain("interrupted");

const createInput = {
clientRequestId: "create-thread-batch-1",
threads: [
Expand Down Expand Up @@ -1232,6 +1285,178 @@ describe("orchestrator MCP toolkit", () => {
expect(
listed.threads.some((thread) => thread.relationshipToParent === "subagent"),
).toBe(false);

// A wait-mode delegation whose blocking wait times out no longer
// owns delivery, so delegate_task upgrades the task to "always".
// Its terminal then wakes the parent even mid-turn.
const upgradedCall = yield* invoke("delegate_task", {
task: cancellationPrompt,
target: {
providerInstanceId: codexInstanceId,
model: codexModel,
},
mode: "wait",
timeoutMs: 1,
clientRequestId: "delegate-wait-upgrade-1",
});
const upgradedDelegated = yield* decodeDelegateTaskResult(
upgradedCall.structuredContent,
).pipe(Effect.orDie);
expect(upgradedDelegated.status).toBe("running");
yield* waitForProjection(orchestrator, upgradedDelegated.childThreadId, (projection) =>
projection.providerTurns.some((turn) => turn.status === "running"),
);
expect(
(yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find(
(task) => task.id === upgradedDelegated.taskId,
)?.completionWake,
).toBe("always");
const upgradeCancelCall = yield* invoke("task_cancel", {
taskId: upgradedDelegated.taskId,
reason: "Terminalize while the parent run is live.",
clientRequestId: "cancel-wait-upgrade-1",
});
expect(upgradeCancelCall.isError).toBe(false);
yield* waitForProjection(orchestrator, parentThreadId, (projection) =>
projection.subagents.some(
(task) => task.id === upgradedDelegated.taskId && task.status === "interrupted",
),
);
const offersAfterUpgrade = yield* waitForContinuationOffers(2);
expect(offersAfterUpgrade).toHaveLength(2);
expect(offersAfterUpgrade[1]).toMatchObject({
threadId: parentThreadId,
delivery: "message_text",
});
expect(offersAfterUpgrade[1]?.detail).toContain(upgradedDelegated.taskId);

// The MCP tool cannot force the reverse interleaving (child
// terminal before the upgrade lands), so dispatch the command
// directly. The first wait-mode delegation completed while the
// parent run was live, so finalize skipped its offer under
// settled_only; the upgrade must accept, persist the policy, and
// deliver the wake finalize declined.
const terminalUpgrade = yield* orchestrator.dispatch({
type: "delegated_task.wake-policy",
commandId: CommandId.make("command:mcp-parent:wake-policy-terminal"),
parentThreadId,
taskId: delegated.taskId,
completionWake: "always",
});
expect(
terminalUpgrade.storedEvents.some(
(stored) =>
stored.event.type === "subagent.updated" &&
stored.event.payload.id === delegated.taskId &&
stored.event.payload.completionWake === "always",
),
).toBe(true);
expect(
(yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find(
(task) => task.id === delegated.taskId,
)?.completionWake,
).toBe("always");
const offersAfterTerminalUpgrade = yield* waitForContinuationOffers(3);
expect(offersAfterTerminalUpgrade).toHaveLength(3);
expect(offersAfterTerminalUpgrade[2]).toMatchObject({
threadId: parentThreadId,
delivery: "message_text",
});
expect(offersAfterTerminalUpgrade[2]?.detail).toContain(delegated.taskId);

// Legacy records omit completionWake and stay settled_only. The
// MCP service always sets the field now, so dispatch the request
// directly to cover the legacy shape.
if (parentRun === undefined || parentRun.rootNodeId === null) {
return yield* Effect.die(new Error("Parent run missing."));
}
const legacyDispatch = yield* orchestrator.dispatch({
type: "delegated_task.request",
createdBy: "agent",
creationSource: "mcp",
commandId: CommandId.make("command:mcp-parent:delegate-legacy"),
parentThreadId,
parentRunId: parentRun.id,
parentNodeId: parentRun.rootNodeId,
task: cancellationPrompt,
modelSelection: codexSelection,
runtimeMode: "full-access",
interactionMode: "default",
});
const legacyTaskEvent = legacyDispatch.storedEvents.find(
(stored) =>
stored.event.type === "subagent.updated" &&
stored.event.payload.origin === "app_owned",
);
if (legacyTaskEvent?.event.type !== "subagent.updated") {
return yield* Effect.die(new Error("Legacy delegated task projection missing."));
}
const legacyTask = legacyTaskEvent.event.payload;
expect(legacyTask.completionWake).toBeUndefined();
if (legacyTask.childThreadId === null) {
return yield* Effect.die(new Error("Legacy delegated task child thread missing."));
}
const legacyChildThreadId = legacyTask.childThreadId;
yield* waitForProjection(orchestrator, legacyChildThreadId, (projection) =>
projection.providerTurns.some((turn) => turn.status === "running"),
);
// Nothing terminalized here, so the offer count must hold.
yield* expectOffersToStay(3);

yield* orchestrator.dispatch({
type: "run.interrupt",
commandId: CommandId.make("command:mcp-parent:interrupt-wake"),
threadId: parentThreadId,
runId: parentRun.id,
reason: "Settle the parent before the legacy child terminalizes.",
});
yield* waitForProjection(orchestrator, parentThreadId, (projection) =>
projection.runs.every(
(run) =>
run.status !== "preparing" &&
run.status !== "starting" &&
run.status !== "running",
),
);
const legacyCancelCall = yield* invoke("task_cancel", {
taskId: legacyTask.id,
reason: "Terminalize the legacy child after the parent settled.",
clientRequestId: "cancel-legacy-1",
});
expect(legacyCancelCall.isError).toBe(false);
yield* waitForProjection(orchestrator, legacyChildThreadId, (projection) =>
projection.runs.some((run) => run.status === "interrupted"),
);
const offers = yield* waitForContinuationOffers(4);
expect(offers).toHaveLength(4);
expect(offers[3]).toMatchObject({ threadId: parentThreadId, delivery: "message_text" });
expect(offers[3]?.detail).toContain(legacyTask.id);
expect(offers[3]?.detail).toContain("task_status");

// Same command against a terminal task whose finalize already
// offered (the parent was settled then, and still is): the policy
// must persist without a second offer, or the parent wakes twice.
const settledUpgrade = yield* orchestrator.dispatch({
type: "delegated_task.wake-policy",
commandId: CommandId.make("command:mcp-parent:wake-policy-settled"),
parentThreadId,
taskId: legacyTask.id,
completionWake: "always",
});
expect(
settledUpgrade.storedEvents.some(
(stored) =>
stored.event.type === "subagent.updated" &&
stored.event.payload.id === legacyTask.id &&
stored.event.payload.completionWake === "always",
),
).toBe(true);
expect(
(yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find(
(task) => task.id === legacyTask.id,
)?.completionWake,
).toBe("always");
yield* expectOffersToStay(4);
}).pipe(Effect.provide(testLayer));
}),
),
Expand Down
2 changes: 1 addition & 1 deletion apps/server/src/mcp/toolkits/orchestrator/tools.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ export const OrchestratorCapabilitiesTool = Tool.make("orchestrator_capabilities

export const DelegateTaskTool = Tool.make("delegate_task", {
description:
"Delegate one task to a T3-owned child agent/subagent of THIS thread and run it with only the supplied task prompt, without copying parent conversation history. Use this whenever the user asks for an agent, subagent, worker, delegated task, or parallel help—including cross-provider work. The childThreadId is backing storage, not an ordinary top-level thread. Provider, model, model options (see orchestrator_capabilities), runtime mode, and interaction mode inherit unless target overrides them. Prefer mode='async' and poll task_status for long work; mode='wait' blocks until completion or timeout.",
"Delegate one task to a T3-owned child agent/subagent of THIS thread and run it with only the supplied task prompt, without copying parent conversation history. Use this whenever the user asks for an agent, subagent, worker, delegated task, or parallel help—including cross-provider work. The childThreadId is backing storage, not an ordinary top-level thread. Provider, model, model options (see orchestrator_capabilities), runtime mode, and interaction mode inherit unless target overrides them. Prefer mode='async' for long work; mode='wait' blocks until completion or timeout. An async child's completion wakes this thread with a continuation message naming the task (queued behind any turn in progress), so end the turn instead of polling or spawning watchers; use task_status only when the result is needed mid-turn.",
parameters: OrchestratorMcpDelegateTaskInput,
success: OrchestratorMcpDelegateTaskResult,
failure: OrchestratorMcpFailure,
Expand Down
Loading
Loading