mirror of
https://github.com/vercel/eve.git
synced 2026-09-20 05:35:39 +08:00
refactor(eve): remove approval fix from workflow sandbox change
Signed-off-by: Rui Conti <ruiconti@gmail.com>
This commit is contained in:
Rui Conti
@@ -1,5 +0,0 @@
|
||||
---
|
||||
"eve": patch
|
||||
---
|
||||
|
||||
Prevent blocking workflow tools from starting before their approval request is resolved.
|
||||
@@ -333,9 +333,7 @@ if (answer === undefined) return { deployed: false, reason: "timed out" };
|
||||
normal `turnPolicy`.
|
||||
|
||||
Compare the [`approval`](/docs/human-in-the-loop) policy, which gates the call before `execute` runs
|
||||
and can only show the model's input. For both blocking and background workflow tools, a pending
|
||||
approval prevents the workflow from starting; denying it prevents execution. Both compose:
|
||||
`approval` before the run, `ctx.ask` inside it.
|
||||
and can only show the model's input. Both compose: `approval` before the run, `ctx.ask` inside it.
|
||||
|
||||
## Delegate work: `ctx.agent`
|
||||
|
||||
|
||||
@@ -70,7 +70,6 @@ function respond(request: MockModelRequest): MockModelResponse | string {
|
||||
["WORKFLOW-DEPLOY-START", "deploy_service"],
|
||||
["WORKFLOW-SANDBOX-BLOCKING-START", "sandbox_blocking"],
|
||||
["WORKFLOW-SANDBOX-BACKGROUND-START", "sandbox_background"],
|
||||
["WORKFLOW-GUARDED-START", "guarded_workflow"],
|
||||
["WORKFLOW-CONFIRM-START", "confirm_deploy"],
|
||||
["WORKFLOW-REPORT-START", "report_deploy"],
|
||||
["WORKFLOW-ESCALATE-START", "escalate_deploy"],
|
||||
|
||||
@@ -1,15 +0,0 @@
|
||||
import { defineWorkflowTool } from "eve/tools";
|
||||
import { always } from "eve/tools/approval";
|
||||
import { z } from "zod";
|
||||
|
||||
export default defineWorkflowTool({
|
||||
approval: always(),
|
||||
description: "Report workflow execution only after approval.",
|
||||
inputSchema: z.strictObject({ service: z.string() }),
|
||||
async *execute({ service }) {
|
||||
"use workflow";
|
||||
|
||||
yield "WORKFLOW-GUARDED-EXECUTING";
|
||||
return { service, result: "WORKFLOW-GUARDED-COMPLETE" };
|
||||
},
|
||||
});
|
||||
@@ -1,27 +0,0 @@
|
||||
import { defineEval } from "eve/evals";
|
||||
|
||||
export default defineEval({
|
||||
description: "A blocking workflow starts only after approval.",
|
||||
async test(t) {
|
||||
const parked = await t.send("WORKFLOW-GUARDED-START");
|
||||
parked.expectOk();
|
||||
parked.session.requireInputRequest({
|
||||
display: "confirmation",
|
||||
toolName: "guarded_workflow",
|
||||
});
|
||||
parked.notEvent("action.partial", {
|
||||
data: { result: { toolName: "guarded_workflow" } },
|
||||
});
|
||||
parked.notEvent("action.result", {
|
||||
data: { result: { toolName: "guarded_workflow" } },
|
||||
});
|
||||
|
||||
const resolved = await parked.session.respondAll("approve");
|
||||
resolved.expectOk();
|
||||
resolved.event("action.partial", {
|
||||
data: { result: { toolName: "guarded_workflow", output: "WORKFLOW-GUARDED-EXECUTING" } },
|
||||
});
|
||||
resolved.messageIncludes("WORKFLOW-GUARDED-COMPLETE");
|
||||
t.noFailedActions();
|
||||
},
|
||||
});
|
||||
@@ -1,30 +0,0 @@
|
||||
import { defineEval } from "eve/evals";
|
||||
|
||||
export default defineEval({
|
||||
description: "A blocking workflow never starts when approval is denied.",
|
||||
async test(t) {
|
||||
const parked = await t.send("WORKFLOW-GUARDED-START");
|
||||
parked.expectOk();
|
||||
parked.session.requireInputRequest({
|
||||
display: "confirmation",
|
||||
toolName: "guarded_workflow",
|
||||
});
|
||||
parked.notEvent("action.partial", {
|
||||
data: { result: { toolName: "guarded_workflow" } },
|
||||
});
|
||||
parked.notEvent("action.result", {
|
||||
data: { result: { toolName: "guarded_workflow" } },
|
||||
});
|
||||
|
||||
const resolved = await parked.session.respondAll("cancel");
|
||||
resolved.expectOk();
|
||||
resolved.notEvent("action.partial", {
|
||||
data: { result: { toolName: "guarded_workflow" } },
|
||||
});
|
||||
resolved.event("action.result", {
|
||||
count: 1,
|
||||
data: { result: { toolName: "guarded_workflow" }, status: "rejected" },
|
||||
});
|
||||
t.succeeded();
|
||||
},
|
||||
});
|
||||
@@ -31,13 +31,8 @@ export function isSessionStateIdleForHandoff(sessionState: DurableSessionState):
|
||||
"eve.harness.pendingWorkflowInterrupt",
|
||||
];
|
||||
if (pendingKeys.some((key) => state?.[key] !== undefined)) return false;
|
||||
for (const key of [
|
||||
"eve.runtime.pendingInputBatches",
|
||||
"eve.runtime.pendingApprovalCoordinationBatches",
|
||||
]) {
|
||||
const value = state?.[key];
|
||||
if (value !== undefined && (!Array.isArray(value) || value.length > 0)) return false;
|
||||
}
|
||||
const batches = state?.["eve.runtime.pendingInputBatches"];
|
||||
if (batches !== undefined && (!Array.isArray(batches) || batches.length > 0)) return false;
|
||||
const proxyRequests = state?.["eve.runtime.proxyInputRequests"];
|
||||
if (
|
||||
proxyRequests !== undefined &&
|
||||
|
||||
@@ -255,7 +255,6 @@ describe("handoff state inspection", () => {
|
||||
["eve.runtime.pendingInputBatch", {}],
|
||||
["eve.runtime.pendingInputBatches", [null]],
|
||||
["eve.runtime.pendingCoordinationBatch", {}],
|
||||
["eve.runtime.pendingApprovalCoordinationBatches", [{}]],
|
||||
["eve.runtime.deferredStepInput", {}],
|
||||
["eve.harness.pendingWorkflowInterrupt", {}],
|
||||
["eve.runtime.proxyInputRequests", { malformed: null }],
|
||||
|
||||
@@ -36,7 +36,6 @@ import {
|
||||
defineWorkflowTool,
|
||||
type BlockingWorkflowToolDefinition,
|
||||
} from "#tools/workflow-definition.js";
|
||||
import { always } from "#tools/approval/policies.js";
|
||||
import { serializeInputSchema, toInputSchema } from "#tools/schema.js";
|
||||
|
||||
const DEPLOY_INPUT_SCHEMA = toInputSchema({
|
||||
@@ -79,7 +78,6 @@ function buildSerializedContext(input: {
|
||||
*/
|
||||
async function createWorkflowToolRuntime(input: {
|
||||
readonly agentName: string;
|
||||
readonly approval?: BlockingWorkflowToolDefinition["approval"];
|
||||
readonly background?: boolean;
|
||||
readonly execute: (...args: never[]) => unknown;
|
||||
readonly inputSchema?: ResolvedToolDefinition["inputSchema"];
|
||||
@@ -92,7 +90,6 @@ async function createWorkflowToolRuntime(input: {
|
||||
logicalPath: `tools/${input.toolName}.ts`,
|
||||
loadNamespace: async () => ({
|
||||
default: defineWorkflowTool({
|
||||
approval: input.approval,
|
||||
execution: input.background === true ? "background" : undefined,
|
||||
description: `Deploys a service (${input.toolName}).`,
|
||||
execute: input.execute as BlockingWorkflowToolDefinition["execute"],
|
||||
@@ -656,109 +653,6 @@ describe("workflow tools", () => {
|
||||
expect(output).toContain("ctx.getSandbox() is not available inside a workflow tool");
|
||||
});
|
||||
|
||||
it.each(
|
||||
[
|
||||
{ background: false, decision: "approve" },
|
||||
{ background: false, decision: "cancel" },
|
||||
{ background: true, decision: "approve" },
|
||||
{ background: true, decision: "cancel" },
|
||||
].flatMap((testCase) => [
|
||||
{ ...testCase, sandbox: false },
|
||||
{ ...testCase, sandbox: true },
|
||||
]),
|
||||
)(
|
||||
"gates workflow execution on approval (background=$background, decision=$decision, sandbox=$sandbox)",
|
||||
async ({ background, decision, sandbox }) => {
|
||||
vi.stubEnv("VERCEL_DEPLOYMENT_ID", "dpl_inline");
|
||||
const runtime = await createWorkflowToolRuntime({
|
||||
agentName: "workflow-tool-approval",
|
||||
approval: always(),
|
||||
background,
|
||||
execute: sandbox ? sandboxAcrossStepsWorkflow : deployServiceWorkflow,
|
||||
toolName: "deploy_service",
|
||||
});
|
||||
|
||||
await runtime.run(async () => {
|
||||
const bundle = await getCompiledRuntimeAgentBundle({
|
||||
compiledArtifactsSource: createBundledRuntimeCompiledArtifactsSource(),
|
||||
});
|
||||
const backend = bundle.graph.root.sandboxRegistry.sandbox!.definition.backend;
|
||||
const provision = vi.spyOn(backend, "create");
|
||||
const before = await listWorkflowToolRunIds();
|
||||
const run = await start(workflowEntry, [
|
||||
{
|
||||
kind: "initial",
|
||||
ownerDeploymentId: "dpl_inline",
|
||||
input: { message: 'Run deploy_service with service "api"' },
|
||||
serializedContext: buildSerializedContext({
|
||||
continuationToken: "http:workflow-tool-approval",
|
||||
mode: "conversation",
|
||||
requestInput: true,
|
||||
}),
|
||||
},
|
||||
]);
|
||||
const stream = captureTurnEvents(run);
|
||||
|
||||
try {
|
||||
const asked = await stream.nextTurn();
|
||||
expect(asked.at(-1)?.type).toBe("session.waiting");
|
||||
const requested = filterEventsByType(asked, "input.requested");
|
||||
expect(requested).toHaveLength(1);
|
||||
const request = (requested[0] as InputRequestedStreamEvent).data.requests[0]!;
|
||||
expect(request).toMatchObject({
|
||||
action: { kind: "tool-call", toolName: "deploy_service" },
|
||||
kind: "tool-approval",
|
||||
});
|
||||
expect(await listWorkflowToolRunIds()).toEqual(before);
|
||||
expect(provision).not.toHaveBeenCalled();
|
||||
|
||||
const commandToken = sessionCommandHookToken(run.runId);
|
||||
await resumeSessionInbox(commandToken, {
|
||||
kind: "send",
|
||||
payload: { inputResponses: [{ optionId: decision, requestId: request.requestId }] },
|
||||
});
|
||||
|
||||
if (decision === "cancel") {
|
||||
const denied = await stream.nextTurn();
|
||||
expect(filterEventsByType(denied, "turn.failed")).toEqual([]);
|
||||
expect(await listWorkflowToolRunIds()).toEqual(before);
|
||||
expect(provision).not.toHaveBeenCalled();
|
||||
expect(filterEventsByType(denied, "action.result")).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({ data: expect.objectContaining({ status: "rejected" }) }),
|
||||
]),
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const executorRunId = await waitForNewWorkflowToolRun(before, 15_000);
|
||||
expect(await waitForWorkflowToolRunTerminal(executorRunId)).toBe("completed");
|
||||
const completed = await stream.nextTurn();
|
||||
expect(filterEventsByType(completed, "turn.failed")).toEqual([]);
|
||||
expect(completed.at(-1)?.type).toBe("session.waiting");
|
||||
if (sandbox) expect(provision).toHaveBeenCalled();
|
||||
else expect(provision).not.toHaveBeenCalled();
|
||||
if (!background) {
|
||||
expect(JSON.stringify(completed)).toContain(
|
||||
sandbox ? "workflow-sandbox:api" : "plan:api",
|
||||
);
|
||||
}
|
||||
const continuedTurns = filterEventsByType(completed, "turn.started");
|
||||
expect(continuedTurns).toHaveLength(1);
|
||||
expect(continuedTurns[0]?.data.turnId).toMatch(/^turn_\d+$/u);
|
||||
expect(filterEventsByType(completed, "turn.completed")[0]?.data.turnId).toBe(
|
||||
continuedTurns[0]?.data.turnId,
|
||||
);
|
||||
} finally {
|
||||
provision.mockRestore();
|
||||
stream.dispose();
|
||||
await run.cancel();
|
||||
}
|
||||
});
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
it("settles the call with an error when the workflow body throws", async () => {
|
||||
const runtime = await createWorkflowToolRuntime({
|
||||
agentName: "workflow-tool-fail",
|
||||
|
||||
@@ -1,149 +0,0 @@
|
||||
import type { ModelMessage } from "ai";
|
||||
|
||||
import {
|
||||
assertUniqueCoordinationCallIds,
|
||||
getPendingCoordinationBatch,
|
||||
setPendingCoordinationBatch,
|
||||
type PendingCoordinationBatch,
|
||||
} from "#harness/coordination.js";
|
||||
import { getPendingInputBatches } from "#harness/pending-input-batches.js";
|
||||
import type { HarnessSession, SessionStateMap } from "#harness/types.js";
|
||||
import type {
|
||||
RuntimeToolCallActionRequest,
|
||||
RuntimeWorkflowTaskRequest,
|
||||
} from "#shared/action-types.js";
|
||||
|
||||
const PENDING_APPROVAL_COORDINATION_BATCHES_KEY = "eve.runtime.pendingApprovalCoordinationBatches";
|
||||
|
||||
/** Returns coordination batches withheld until their matching approvals settle. */
|
||||
export function getPendingApprovalCoordinationBatches(
|
||||
state: SessionStateMap | undefined,
|
||||
): readonly PendingCoordinationBatch[] {
|
||||
const value = state?.[PENDING_APPROVAL_COORDINATION_BATCHES_KEY];
|
||||
if (!Array.isArray(value)) return [];
|
||||
|
||||
return value.filter((entry): entry is PendingCoordinationBatch => {
|
||||
if (typeof entry !== "object" || entry === null) return false;
|
||||
const batch = entry as PendingCoordinationBatch;
|
||||
return (
|
||||
Array.isArray(batch.runtimeActions) &&
|
||||
Array.isArray(batch.tasks) &&
|
||||
Array.isArray(batch.responseMessages) &&
|
||||
typeof batch.event === "object" &&
|
||||
batch.event !== null
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
function setPendingApprovalCoordinationBatches(
|
||||
session: HarnessSession,
|
||||
batches: readonly PendingCoordinationBatch[],
|
||||
): HarnessSession {
|
||||
const state = { ...session.state };
|
||||
if (batches.length === 0) {
|
||||
delete state[PENDING_APPROVAL_COORDINATION_BATCHES_KEY];
|
||||
} else {
|
||||
state[PENDING_APPROVAL_COORDINATION_BATCHES_KEY] = batches.map((batch) => ({
|
||||
runtimeActions: [...batch.runtimeActions],
|
||||
tasks: [...batch.tasks],
|
||||
event: batch.event,
|
||||
localFanoutSize: batch.localFanoutSize,
|
||||
responseMessages: [...batch.responseMessages],
|
||||
}));
|
||||
}
|
||||
return { ...session, state: Object.keys(state).length > 0 ? state : undefined };
|
||||
}
|
||||
|
||||
/** Stores coordination that must not dispatch before approval settles. */
|
||||
export function appendPendingApprovalCoordinationBatch(input: {
|
||||
readonly runtimeActions: readonly RuntimeToolCallActionRequest[];
|
||||
readonly tasks: readonly RuntimeWorkflowTaskRequest[];
|
||||
readonly event: PendingCoordinationBatch["event"];
|
||||
readonly localFanoutSize?: number;
|
||||
readonly responseMessages: readonly ModelMessage[];
|
||||
readonly session: HarnessSession;
|
||||
}): HarnessSession {
|
||||
const batch = {
|
||||
runtimeActions: [...input.runtimeActions],
|
||||
tasks: [...input.tasks],
|
||||
event: input.event,
|
||||
localFanoutSize: input.localFanoutSize,
|
||||
responseMessages: [...input.responseMessages],
|
||||
} satisfies PendingCoordinationBatch;
|
||||
const existing = getPendingApprovalCoordinationBatches(input.session.state);
|
||||
assertUniqueCoordinationCallIds([...batch.runtimeActions, ...batch.tasks]);
|
||||
return setPendingApprovalCoordinationBatches(input.session, [...existing, batch]);
|
||||
}
|
||||
|
||||
/** Removes approval-blocked coordination requests rejected before dispatch. */
|
||||
export function removePendingApprovalCoordinationRequests(
|
||||
session: HarnessSession,
|
||||
rejections: readonly {
|
||||
readonly event: PendingCoordinationBatch["event"];
|
||||
readonly results: readonly { readonly callId: string }[];
|
||||
}[],
|
||||
): HarnessSession {
|
||||
if (rejections.length === 0) return session;
|
||||
const batches = getPendingApprovalCoordinationBatches(session.state);
|
||||
let changed = false;
|
||||
const filtered = batches.flatMap((batch) => {
|
||||
const callIds = new Set(
|
||||
rejections
|
||||
.filter(
|
||||
({ event }) =>
|
||||
event.sequence === batch.event.sequence &&
|
||||
event.stepIndex === batch.event.stepIndex &&
|
||||
event.turnId === batch.event.turnId,
|
||||
)
|
||||
.flatMap(({ results }) => results.map((result) => result.callId)),
|
||||
);
|
||||
const runtimeActions = batch.runtimeActions.filter((request) => !callIds.has(request.callId));
|
||||
const tasks = batch.tasks.filter((request) => !callIds.has(request.callId));
|
||||
changed ||= runtimeActions.length !== batch.runtimeActions.length;
|
||||
changed ||= tasks.length !== batch.tasks.length;
|
||||
return runtimeActions.length === 0 && tasks.length === 0
|
||||
? []
|
||||
: [{ ...batch, runtimeActions, tasks }];
|
||||
});
|
||||
return changed ? setPendingApprovalCoordinationBatches(session, filtered) : session;
|
||||
}
|
||||
|
||||
/** Promotes the first blocked batch whose matching approval is no longer pending. */
|
||||
export function promoteApprovedCoordinationBatch(session: HarnessSession): HarnessSession {
|
||||
if (getPendingCoordinationBatch(session.state) !== undefined) return session;
|
||||
|
||||
const batches = getPendingApprovalCoordinationBatches(session.state);
|
||||
const readyIndex = batches.findIndex((batch) => !hasMatchingPendingApproval(session, batch));
|
||||
if (readyIndex < 0) return session;
|
||||
|
||||
const batch = batches[readyIndex];
|
||||
if (batch === undefined) return session;
|
||||
const remaining = batches.filter((_, index) => index !== readyIndex);
|
||||
return setPendingCoordinationBatch({
|
||||
runtimeActions: batch.runtimeActions,
|
||||
tasks: batch.tasks,
|
||||
event: batch.event,
|
||||
localFanoutSize: batch.localFanoutSize,
|
||||
responseMessages: batch.responseMessages,
|
||||
session: setPendingApprovalCoordinationBatches(session, remaining),
|
||||
});
|
||||
}
|
||||
|
||||
function hasMatchingPendingApproval(
|
||||
session: HarnessSession,
|
||||
batch: PendingCoordinationBatch,
|
||||
): boolean {
|
||||
const callIds = new Set(
|
||||
[...batch.runtimeActions, ...batch.tasks].map((request) => request.callId),
|
||||
);
|
||||
return getPendingInputBatches(session.state).some(
|
||||
(inputBatch) =>
|
||||
inputBatch.event !== undefined &&
|
||||
inputBatch.event.sequence === batch.event.sequence &&
|
||||
inputBatch.event.stepIndex === batch.event.stepIndex &&
|
||||
inputBatch.event.turnId === batch.event.turnId &&
|
||||
inputBatch.requests.some(
|
||||
(request) => request.kind === "tool-approval" && callIds.has(request.action.callId),
|
||||
),
|
||||
);
|
||||
}
|
||||
@@ -1,4 +1,3 @@
|
||||
import { getPendingApprovalCoordinationBatches } from "#harness/approval-coordination.js";
|
||||
import { BoundaryHookError } from "#shared/boundary-hook-error.js";
|
||||
import { context as otelContext, trace } from "#compiled/@opentelemetry/api/index.js";
|
||||
import {
|
||||
@@ -80,7 +79,6 @@ import {
|
||||
appendPendingInputBatch,
|
||||
} from "#harness/input-requests.js";
|
||||
import { activeTurnId } from "#harness/active-turn-id.js";
|
||||
import { derivePendingState } from "#execution/session/pending-turn-state.js";
|
||||
import { registerWorkflowToolRun } from "#harness/workflow-tool-runs.js";
|
||||
import { getPendingCoordinationBatch } from "#harness/coordination.js";
|
||||
import { AGENT_HANDLES_STATE_KEY } from "#subagents/handles/store.js";
|
||||
@@ -1918,162 +1916,6 @@ describe("createToolLoopHarness", () => {
|
||||
expect(getPendingCoordinationBatch(result.session.state)).toBeUndefined();
|
||||
});
|
||||
|
||||
it("does not coordinate a blocking workflow tool before approval", async () => {
|
||||
const toolCall = {
|
||||
input: { action: "run" },
|
||||
toolCallId: "workflow-1",
|
||||
toolName: "guarded_workflow",
|
||||
type: "tool-call" as const,
|
||||
};
|
||||
setupMockAgent({
|
||||
content: [
|
||||
toolCall,
|
||||
{
|
||||
approvalId: "approval-workflow",
|
||||
toolCallId: toolCall.toolCallId,
|
||||
type: "tool-approval-request",
|
||||
},
|
||||
],
|
||||
finishReason: "tool-calls",
|
||||
response: {
|
||||
messages: [
|
||||
{
|
||||
content: [
|
||||
toolCall,
|
||||
{
|
||||
approvalId: "approval-workflow",
|
||||
toolCallId: toolCall.toolCallId,
|
||||
type: "tool-approval-request",
|
||||
},
|
||||
],
|
||||
role: "assistant",
|
||||
},
|
||||
],
|
||||
},
|
||||
responseMessages: [
|
||||
{
|
||||
content: [
|
||||
toolCall,
|
||||
{
|
||||
approvalId: "approval-workflow",
|
||||
toolCallId: toolCall.toolCallId,
|
||||
type: "tool-approval-request",
|
||||
},
|
||||
],
|
||||
role: "assistant",
|
||||
},
|
||||
],
|
||||
text: "",
|
||||
toolCalls: [toolCall],
|
||||
toolResults: [],
|
||||
});
|
||||
|
||||
const { emit, events } = createEventCollector();
|
||||
const runStep = createToolLoopHarness(
|
||||
createTestConfig("conversation", emit, {
|
||||
tools: new Map([
|
||||
[
|
||||
"guarded_workflow",
|
||||
{
|
||||
approval: () => "user-approval",
|
||||
description: "Run a guarded workflow.",
|
||||
inputSchema: jsonSchema({ type: "object" }),
|
||||
name: "guarded_workflow",
|
||||
workflowId: "workflow//./agent/tools/guarded-workflow//execute",
|
||||
},
|
||||
],
|
||||
[
|
||||
"unguarded_workflow",
|
||||
{
|
||||
description: "Run an unguarded workflow.",
|
||||
inputSchema: jsonSchema({ type: "object" }),
|
||||
name: "unguarded_workflow",
|
||||
workflowId: "workflow//./agent/tools/unguarded-workflow//execute",
|
||||
},
|
||||
],
|
||||
]),
|
||||
}),
|
||||
);
|
||||
|
||||
const pending = await runStep(createTestSession(), { message: "Run the workflow." });
|
||||
|
||||
expect(getPendingApprovalCoordinationBatches(pending.session.state)[0]?.tasks).toMatchObject([
|
||||
{ callId: "workflow-1" },
|
||||
]);
|
||||
expect(getPendingCoordinationBatch(pending.session.state)).toBeUndefined();
|
||||
expect(derivePendingState(pending.session).pendingCoordinationCallIds).toBeUndefined();
|
||||
expect(getPendingInputRequestIds(pending.session.state)).toEqual(
|
||||
new Set(["approval-workflow"]),
|
||||
);
|
||||
expect(events.slice(-3).map((event) => event.type)).toEqual([
|
||||
"input.requested",
|
||||
"turn.completed",
|
||||
"session.waiting",
|
||||
]);
|
||||
|
||||
const unrelatedToolCall = {
|
||||
input: {},
|
||||
toolCallId: "workflow-2",
|
||||
toolName: "unguarded_workflow",
|
||||
type: "tool-call" as const,
|
||||
};
|
||||
setupMockAgent({
|
||||
content: [unrelatedToolCall],
|
||||
finishReason: "tool-calls",
|
||||
response: {
|
||||
messages: [{ content: [unrelatedToolCall], role: "assistant" }],
|
||||
},
|
||||
responseMessages: [{ content: [unrelatedToolCall], role: "assistant" }],
|
||||
text: "",
|
||||
toolCalls: [unrelatedToolCall],
|
||||
toolResults: [],
|
||||
});
|
||||
const unrelated = await runStep(pending.session, {
|
||||
message: "Run something else while that waits.",
|
||||
});
|
||||
|
||||
expect(getPendingCoordinationBatch(unrelated.session.state)?.tasks).toMatchObject([
|
||||
{ callId: "workflow-2" },
|
||||
]);
|
||||
expect(derivePendingState(unrelated.session).pendingCoordinationCallIds).toEqual([
|
||||
"workflow-2",
|
||||
]);
|
||||
expect(getPendingApprovalCoordinationBatches(unrelated.session.state)[0]?.tasks).toMatchObject([
|
||||
{ callId: "workflow-1" },
|
||||
]);
|
||||
expect(unrelated.session.history).toContainEqual({
|
||||
content: "Run something else while that waits.",
|
||||
kind: "user",
|
||||
role: "user",
|
||||
});
|
||||
|
||||
const eventCountBeforeApproval = events.length;
|
||||
const approved = await runStep(pending.session, {
|
||||
inputResponses: [{ optionId: "approve", requestId: "approval-workflow" }],
|
||||
});
|
||||
|
||||
expect(hasPendingInputBatch(approved.session.state)).toBe(false);
|
||||
expect(derivePendingState(approved.session).pendingCoordinationCallIds).toEqual(["workflow-1"]);
|
||||
const approvalEvents = events.slice(eventCountBeforeApproval);
|
||||
const approvalTurnStarted = approvalEvents.find((event) => event.type === "turn.started");
|
||||
expect(approvalTurnStarted?.data.turnId).toMatch(/^turn_\d+$/u);
|
||||
|
||||
setupMockAgent({
|
||||
finishReason: "stop",
|
||||
response: { messages: [{ content: "Cancelled.", role: "assistant" }] },
|
||||
text: "Cancelled.",
|
||||
toolCalls: [],
|
||||
toolResults: [],
|
||||
});
|
||||
const denied = await runStep(pending.session, {
|
||||
inputResponses: [{ optionId: "cancel", requestId: "approval-workflow" }],
|
||||
});
|
||||
|
||||
expect(getPendingCoordinationBatch(denied.session.state)).toBeUndefined();
|
||||
expect(getPendingApprovalCoordinationBatches(denied.session.state)).toEqual([]);
|
||||
expect(hasPendingInputBatch(denied.session.state)).toBe(false);
|
||||
});
|
||||
|
||||
it("parks on both batches when one step carries a workflow task and an approval", async () => {
|
||||
const gateToolCall = {
|
||||
input: { action: "run" },
|
||||
|
||||
@@ -1,8 +1,3 @@
|
||||
import {
|
||||
appendPendingApprovalCoordinationBatch,
|
||||
promoteApprovedCoordinationBatch,
|
||||
removePendingApprovalCoordinationRequests,
|
||||
} from "#harness/approval-coordination.js";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import { BoundaryHookError } from "#shared/boundary-hook-error.js";
|
||||
import { GenerationSteering } from "#harness/generation-steering.js";
|
||||
@@ -210,7 +205,6 @@ import {
|
||||
import { resolveFrameworkToolFromUpstreamType } from "#harness/provider-tools.js";
|
||||
import {
|
||||
createCoordinationRequestFromToolCall,
|
||||
getPendingCoordinationBatch,
|
||||
resolvePendingCoordination,
|
||||
setPendingCoordinationBatch,
|
||||
} from "#harness/coordination.js";
|
||||
@@ -666,7 +660,6 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn {
|
||||
});
|
||||
session = stepInput.session;
|
||||
|
||||
session = promoteApprovedCoordinationBatch(session);
|
||||
const resolvedCoordination = await resolvePendingCoordination({
|
||||
emit,
|
||||
session,
|
||||
@@ -850,7 +843,7 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn {
|
||||
};
|
||||
}
|
||||
|
||||
let pending = resolvePendingInput({
|
||||
const pending = resolvePendingInput({
|
||||
deferMessagesWhileApprovalsPending: config.mode !== "conversation",
|
||||
history: resolvedCoordination.messages,
|
||||
resolveApprovalKey: resolveApprovalKeyFromTools(responseAuthorizationTools),
|
||||
@@ -994,26 +987,6 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn {
|
||||
}
|
||||
}
|
||||
|
||||
if ((pending.rejectedActions?.length ?? 0) > 0) {
|
||||
pending = {
|
||||
...pending,
|
||||
session: removePendingApprovalCoordinationRequests(
|
||||
pending.session,
|
||||
pending.rejectedActions ?? [],
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
pending = { ...pending, session: promoteApprovedCoordinationBatch(pending.session) };
|
||||
|
||||
const promotedCoordination = getPendingCoordinationBatch(pending.session.state) !== undefined;
|
||||
if (promotedCoordination && config.mode !== "conversation") {
|
||||
return {
|
||||
next: null,
|
||||
session: { ...pending.session, history: validateHarnessModelMessages(pending.messages) },
|
||||
};
|
||||
}
|
||||
|
||||
// --- Turn preamble ------------------------------------------------------
|
||||
|
||||
const turnId = activeTurnId(emissionState);
|
||||
@@ -1173,13 +1146,6 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn {
|
||||
|
||||
messages = [...messages, ...preparedTurnInput];
|
||||
|
||||
if (promotedCoordination) {
|
||||
return {
|
||||
next: null,
|
||||
session: { ...session, history: messages },
|
||||
};
|
||||
}
|
||||
|
||||
const createModelMessages = (
|
||||
durableMessages: readonly HarnessModelMessage[],
|
||||
): HarnessModelMessage[] => {
|
||||
@@ -2740,74 +2706,14 @@ async function handleStepResult(input: {
|
||||
};
|
||||
}
|
||||
|
||||
const event = {
|
||||
sequence: emissionState.sequence,
|
||||
stepIndex: emissionState.stepIndex,
|
||||
turnId: emissionState.turnId,
|
||||
};
|
||||
const approvalCallIds = new Set(
|
||||
inputRequests
|
||||
.filter((request) => request.kind === "tool-approval")
|
||||
.map((request) => request.action.callId),
|
||||
);
|
||||
const mustWaitForApproval = [...runtimeActions, ...tasks].some((request) =>
|
||||
approvalCallIds.has(request.callId),
|
||||
);
|
||||
|
||||
if (mustWaitForApproval) {
|
||||
let parkedSession = appendPendingApprovalCoordinationBatch({
|
||||
runtimeActions,
|
||||
tasks,
|
||||
event,
|
||||
responseMessages: [],
|
||||
session: { ...baseSession, history: parkedInputHistory },
|
||||
});
|
||||
const responseAuthorizationTools = buildResponseAuthorizationTools({
|
||||
authoredTools: config.tools,
|
||||
context: contextStorage.getStore(),
|
||||
});
|
||||
parkedSession = appendPendingInputBatch({
|
||||
event,
|
||||
requests: inputRequests,
|
||||
responseAuthRequiredRequestIds: approvalRequests
|
||||
.filter((request) => {
|
||||
const approval = responseAuthorizationTools.get(request.action.toolName)?.approval;
|
||||
return (
|
||||
approval !== undefined &&
|
||||
typeof approval !== "function" &&
|
||||
approval.response !== undefined
|
||||
);
|
||||
})
|
||||
.map((request) => request.requestId),
|
||||
responseMessages: pendingResponseMessages,
|
||||
session: parkedSession,
|
||||
});
|
||||
|
||||
if (emit) {
|
||||
await emit(
|
||||
createInputRequestedEvent({
|
||||
requests: inputRequests,
|
||||
sequence: event.sequence,
|
||||
stepIndex: event.stepIndex,
|
||||
turnId: event.turnId,
|
||||
}),
|
||||
);
|
||||
if (config.mode === "conversation") {
|
||||
emissionState = await emitTurnEpilogue(emit, emissionState, config.mode);
|
||||
parkedSession = setHarnessEmissionState(parkedSession, emissionState);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
next: hasDeferredStepInput(parkedSession) ? runStep : null,
|
||||
session: parkedSession,
|
||||
};
|
||||
}
|
||||
|
||||
let parkedSession = setPendingCoordinationBatch({
|
||||
runtimeActions,
|
||||
tasks,
|
||||
event,
|
||||
event: {
|
||||
sequence: emissionState.sequence,
|
||||
stepIndex: emissionState.stepIndex,
|
||||
turnId: emissionState.turnId,
|
||||
},
|
||||
responseMessages: pendingResponseMessages,
|
||||
session: { ...baseSession, history: parkedInputHistory },
|
||||
});
|
||||
|
||||
@@ -8,7 +8,7 @@ last_updated: "2026-09-19"
|
||||
|
||||
Authored workflow steps access the session sandbox lazily through `ctx.getSandbox()`; the session retains initialization and lifecycle ownership.
|
||||
|
||||
The approval ordering fix in [#3198](https://github.com/vercel/eve/issues/3198) must land with or before this change.
|
||||
Approval ordering remains tracked separately in [#3198](https://github.com/vercel/eve/issues/3198); this change does not fix it.
|
||||
|
||||
## Authoring contract
|
||||
|
||||
@@ -16,7 +16,7 @@ Use `defineWorkflowTool({ execute })` and pass `ctx` directly to a `"use step"`
|
||||
|
||||
## Boundaries
|
||||
|
||||
- Tool approval gates dispatch. A workflow that never calls `getSandbox` does not open a sandbox.
|
||||
- A workflow that never calls `getSandbox` does not open a sandbox.
|
||||
- On first access in each step, the step requests initialization through its owner inbox. The owning session checks the recorded run, opens or reconnects its sandbox, and persists the updated session checkpoint before returning a serializable reconnect record. Background tasks forward this request through their existing parent delivery path.
|
||||
- A durable response stream keyed by the requesting step lets retries reuse the response. Concurrent steps are initialized through the owning session so they share its initialization state. This adds an owner round trip on first access in each step.
|
||||
- The existing workflow step context wrapper reconnects the backend and binds operations to the step's abort signal. Repeated calls in one step reuse its handle.
|
||||
@@ -25,4 +25,4 @@ Use `defineWorkflowTool({ execute })` and pass `ctx` directly to a `"use step"`
|
||||
|
||||
## Validation
|
||||
|
||||
Runtime integration coverage checks file persistence across a durable sleep and successive steps for blocking and background tools, plus concurrent first accesses, step retries, unused getters, approval/denial, and a clear failure from the workflow body. Fixture evals exercise the same contract through an agent. CI is required for the fixture evals.
|
||||
Runtime integration coverage checks file persistence across a durable sleep and successive steps for blocking and background tools, plus concurrent first accesses, step retries, and a clear failure from the workflow body. Fixture evals exercise the same contract through an agent. CI is required for the fixture evals.
|
||||
|
||||
Reference in New Issue
Block a user