Skip to content

feat(eve): run subagents as background tasks behind experimental.tasks #1521

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

Closed
ruiconti wants to merge 2 commits into from
Closed

feat(eve): run subagents as background tasks behind experimental.tasks #1521

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e70cfc0
feat(eve): run subagents as background tasks behind experimental.tasks
ruiconti Aug 1, 2026
eae4028
refactor(eve): split remote subagent startup
ruiconti Aug 5, 2026
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
191 changes: 76 additions & 115 deletions packages/eve/src/execution/dispatch-runtime-actions-step.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,27 +47,31 @@ import {
} from "#protocol/message.js";
import type {
RuntimeActionRequest,
RuntimeActionResult,
RuntimeRemoteAgentCallActionRequest,
RuntimeSubagentCallActionRequest,
RuntimeSubagentDispatchFailure,
RuntimeSubagentResult,
RuntimeToolCallActionRequest,
} from "#runtime/actions/types.js";
import {
beginDelegatedTask,
executeTaskControlAction,
failDelegatedDispatch,
isTaskControlAction,
settleDelegatedDispatch,
} from "#execution/tasks/dispatch.js";
import {
createDurableSessionState,
type DurableSessionState,
readDurableSession,
} from "#execution/durable-session-store.js";
import {
resolveRemoteAgentForAction,
startRemoteAgentSession,
} from "#execution/remote-agent-dispatch.js";
import {
createRecursiveAgentRootOnlyResult,
createRemoteAgentStartFailureResult,
createUnavailableDynamicSubagentResult,
getSubagentName,
} from "#execution/dispatch-action-failures.js";
import { mintStartOperation } from "#execution/dispatch-start-operation.js";
import { startRemoteSubagent } from "#execution/subagent-start-remote.js";
import { hydrateDurableSession } from "#execution/session.js";
import { buildSubagentRunInput, type SubagentInputSource } from "#execution/subagent-tool.js";
import { createWorkflowRuntime, workflowEntryReference } from "#execution/workflow-runtime.js";
Expand Down Expand Up @@ -102,7 +106,8 @@ type DispatchPlanEntry =
readonly dynamicRemoteAgent?: DynamicRemoteAgentConfig;
}
| { readonly kind: "reject"; readonly result: RuntimeSubagentDispatchFailure }
| { readonly kind: "start"; readonly target: DispatchStartTarget };
| { readonly kind: "start"; readonly target: DispatchStartTarget }
| { readonly kind: "task-control"; readonly action: RuntimeToolCallActionRequest };

type DispatchStartTarget =
| {
Expand All @@ -125,7 +130,7 @@ export async function dispatchRuntimeActionsStep(input: {
readonly serializedContext: Record<string, unknown>;
readonly sessionState: DurableSessionState;
}): Promise<{
readonly results: readonly RuntimeSubagentResult[];
readonly results: readonly RuntimeActionResult[];
readonly sessionState: DurableSessionState;
}> {
"use step";
Expand Down Expand Up @@ -157,8 +162,12 @@ export async function dispatchRuntimeActionsStep(input: {
// Read here, not in the child: trace state is scoped to one session's
// context, so this is the last place the parent's window is visible.
const parentTraceContext = readSessionTraceContext(input.serializedContext, session.sessionId);
const tasksEnabled = bundle.resolvedAgent.config.experimental?.tasks === true;
// Background tasks require resumable children: the flag implies
// conversation-mode dispatch so `experimental.tasks` and
// `experimental.subagentPersistentSessions` never produce a third mode.
const persistentSessions =
bundle.resolvedAgent.config.experimental?.subagentPersistentSessions === true;
tasksEnabled || bundle.resolvedAgent.config.experimental?.subagentPersistentSessions === true;
// A corrupt handle store throws; surface that before anything dispatches.
// A mid-loop throw after a sibling started would durably replay the whole
// batch and re-dispatch that sibling.
Expand All @@ -177,7 +186,7 @@ export async function dispatchRuntimeActionsStep(input: {
).length;

let nextSession = session;
const results: RuntimeSubagentResult[] = [];
const results: RuntimeActionResult[] = [];

try {
for (const entry of plan) {
Expand All @@ -186,6 +195,31 @@ export async function dispatchRuntimeActionsStep(input: {
continue;
}

if (entry.kind === "task-control") {
const control = await executeTaskControlAction({
action: entry.action,
bundle,
session: nextSession,
});
if (control.result !== undefined) {
results.push(control.result);
}
continue;
}

// Delegated execution: the durable task record exists before the
// child dispatch side effect, and the child's reply address is the
// task run's private hook instead of the parent turn's inbox.
const delegated = tasksEnabled
? await beginDelegatedTask({
...describeDelegatedEntry(entry),
parentSessionId: session.sessionId,
parentTurnId: batch.event.turnId,
session: nextSession,
})
: undefined;
const delegatedParentToken = delegated?.commandToken;

let outcome: DispatchOutcome;
switch (entry.kind) {
case "resume":
Expand All @@ -198,7 +232,8 @@ export async function dispatchRuntimeActionsStep(input: {
dynamicRemoteAgent: entry.dynamicRemoteAgent,
}),
currentSession: nextSession,
parentToken: input.parentContinuationToken ?? session.continuationToken,
parentToken:
delegatedParentToken ?? input.parentContinuationToken ?? session.continuationToken,
parentTurnId: batch.event.turnId,
});
break;
Expand All @@ -213,7 +248,7 @@ export async function dispatchRuntimeActionsStep(input: {
currentSession: nextSession,
fanoutSize,
initiatorAuth,
parentContinuationToken: input.parentContinuationToken,
parentContinuationToken: delegatedParentToken ?? input.parentContinuationToken,
parentTraceContext,
persistentSessions,
session,
Expand All @@ -224,10 +259,25 @@ export async function dispatchRuntimeActionsStep(input: {

nextSession = outcome.session;
if (outcome.kind === "error") {
if (delegated !== undefined) {
await failDelegatedDispatch({ error: outcome.result.output, task: delegated });
}
results.push(outcome.result);
continue;
}

if (delegated !== undefined) {
const settled = await settleDelegatedDispatch({
callId: outcome.callId,
childSessionId: outcome.address.sessionId,
session: nextSession,
subagentName: outcome.toolName,
task: delegated,
});
nextSession = settled.session;
results.push(settled.receipt);
}

// Emission is observability, not control flow: a failure here must not
// escape the loop, because a durable-step retry would re-dispatch the
// children that already started.
Expand Down Expand Up @@ -295,6 +345,10 @@ function planDispatch(input: {
const handles = getAgentHandleStore(input.session.state)?.handles ?? [];

return input.actions.map((action): DispatchPlanEntry => {
if (isTaskControlAction(action)) {
return { action, kind: "task-control" };
}

const rawAgentId = action.input.agentId;
const agentId =
typeof rawAgentId === "string" && rawAgentId.trim() !== "" ? rawAgentId : undefined;
Expand Down Expand Up @@ -582,110 +636,17 @@ async function startLocalSubagent(input: {
};
}

async function startRemoteSubagent(input: {
readonly action: RuntimeRemoteAgentCallActionRequest;
readonly auth: Parameters<typeof startRemoteAgentSession>[0]["auth"];
readonly batchEvent: { readonly sequence: number; readonly turnId: string };
readonly bundle: CompiledBundle;
readonly callbackBaseUrl: string | undefined;
readonly currentSession: RuntimeSession;
readonly dynamicRemoteAgent?: DynamicRemoteAgentConfig;
readonly initiatorAuth: Parameters<typeof startRemoteAgentSession>[0]["initiatorAuth"];
readonly parentContinuationToken: string | undefined;
readonly persistentSessions: boolean;
readonly session: RuntimeSession;
}): Promise<DispatchOutcome> {
const { action } = input;

// Preflight resolution failures happen before ownership exists, so they
// reject without touching the handle store.
let callbackBaseUrl: string;
let resolvedRemote: ReturnType<typeof resolveRemoteAgentForAction>;
try {
if (input.callbackBaseUrl === undefined) {
throw new Error("Cannot dispatch remote agent without a callback base URL.");
}
callbackBaseUrl = input.callbackBaseUrl;
resolvedRemote = resolveRemoteAgentForAction({
dynamicRemoteAgent: input.dynamicRemoteAgent,
nodeId: action.nodeId,
remoteAgentName: action.remoteAgentName,
registry: input.bundle.subagentRegistry.subagentsByNodeId,
});
} catch (error) {
logError(log, "remote agent start failed", error, {
remoteAgentName: action.remoteAgentName,
nodeId: action.nodeId,
callId: action.callId,
});
return {
kind: "error",
result: createRemoteAgentStartFailureResult({ action, error }),
session: input.currentSession,
};
}

const { identity, operation } = mintStartOperation({
callId: action.callId,
name: action.remoteAgentName,
nodeId: action.nodeId,
parentSessionId: input.session.sessionId,
parentTurnId: input.batchEvent.turnId,
});
const preparedSession = prepareAgentStart(input.currentSession, {
identity,
operation,
target: { callbackBaseUrl, kind: "agent/remote", url: resolvedRemote.url },
});

try {
const child = await startRemoteAgentSession({
action,
auth: input.auth,
callbackBaseUrl,
callbackToken: input.parentContinuationToken,
initiatorAuth: input.initiatorAuth,
persistentSessions: input.persistentSessions,
remote: resolvedRemote,
session: input.session,
});
const address = {
callbackBaseUrl,
kind: "agent/remote",
sessionId: child.sessionId,
url: resolvedRemote.url,
...(child.continuationToken === undefined
? {}
: { continuationToken: child.continuationToken }),
} as const;
return {
address,
callId: action.callId,
kind: "called",
name: action.name,
session: confirmAgentStarted(preparedSession, {
address,
operationId: operation.id,
}),
toolName: action.remoteAgentName,
};
} catch (error) {
logError(log, "remote agent start failed", error, {
remoteAgentName: action.remoteAgentName,
nodeId: action.nodeId,
callId: action.callId,
});
return {
kind: "error",
result: createRemoteAgentStartFailureResult({ action, error }),
session: rejectAgentEffect(preparedSession, {
disposition: "dead",
operationId: operation.id,
}),
};
}
/** Names one delegated dispatch for its task record, before any child exists. */
function describeDelegatedEntry(entry: Extract<DispatchPlanEntry, { kind: "resume" | "start" }>): {
readonly callId: string;
readonly mode: "local" | "remote";
readonly name: string;
} {
const action = entry.kind === "resume" ? entry.action : entry.target.action;
return action.kind === "remote-agent-call"
? { callId: action.callId, mode: "remote", name: action.remoteAgentName }
: { callId: action.callId, mode: "local", name: action.subagentName };
}

function isRecursiveAgentAction(
action: RuntimeActionRequest,
subagentsByNodeId: ReadonlyMap<string, unknown>,
Expand Down
4 changes: 2 additions & 2 deletions packages/eve/src/execution/dispatch-workflow-runtime-actions-step.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ import { BundleKey } from "#runtime/sessions/runtime-context-keys.js";
import type {
RuntimeActionRequest,
RuntimeSubagentDispatchFailure,
RuntimeSubagentResult,
RuntimeActionResult,
} from "#runtime/actions/types.js";
import { resolveEffectiveAgentRuntime } from "#execution/effective-agent-config.js";

Expand All @@ -33,7 +33,7 @@ export async function dispatchWorkflowRuntimeActionsStep(input: {
readonly serializedContext: Record<string, unknown>;
readonly sessionState: DurableSessionState;
}): Promise<{
readonly results: readonly RuntimeSubagentResult[];
readonly results: readonly RuntimeActionResult[];
readonly sessionState: DurableSessionState;
}> {
"use step";
Expand Down
Loading