Skip to content

Commit 609cc0a

Browse files
tatoalotatoalo
authored
feat(cloud): negotiate idempotent steer delivery (#3538)
1 parent 27236a6 commit 609cc0a

7 files changed

Lines changed: 569 additions & 172 deletions

File tree

‎packages/agent/src/server/agent-server.test.ts‎

Lines changed: 258 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2267,6 +2267,106 @@ describe("AgentServer HTTP Mode", () => {
22672267
expect(prompt).toHaveBeenCalledTimes(4);
22682268
}, 20000);
22692269

2270+
it("steers an active turn without emitting a separate turn completion", async () => {
2271+
const s = createServer();
2272+
await s.start();
2273+
const prompt = vi.fn(async () => ({
2274+
stopReason: "end_turn",
2275+
_meta: { steer: true },
2276+
}));
2277+
const broadcastTurnComplete = vi.fn();
2278+
const resetTurnMessages = vi.fn();
2279+
const serverInternals = s as unknown as {
2280+
activeOwnedTurnCount: number;
2281+
broadcastTurnComplete: typeof broadcastTurnComplete;
2282+
session: {
2283+
clientConnection: { prompt: typeof prompt };
2284+
logWriter: { resetTurnMessages: typeof resetTurnMessages };
2285+
};
2286+
};
2287+
serverInternals.activeOwnedTurnCount = 1;
2288+
serverInternals.broadcastTurnComplete = broadcastTurnComplete;
2289+
serverInternals.session.clientConnection.prompt = prompt;
2290+
serverInternals.session.logWriter.resetTurnMessages = resetTurnMessages;
2291+
2292+
const response = await fetch(`http://localhost:${port}/command`, {
2293+
method: "POST",
2294+
headers: {
2295+
Authorization: `Bearer ${createToken()}`,
2296+
"Content-Type": "application/json",
2297+
},
2298+
body: JSON.stringify({
2299+
jsonrpc: "2.0",
2300+
id: "steer-1",
2301+
method: "user_message",
2302+
params: { content: "change direction", steer: true },
2303+
}),
2304+
});
2305+
2306+
expect(response.status).toBe(200);
2307+
await expect(response.json()).resolves.toMatchObject({
2308+
result: { stopReason: "steered", steered: true },
2309+
});
2310+
expect(prompt).toHaveBeenCalledWith(
2311+
expect.objectContaining({
2312+
_meta: expect.objectContaining({ steer: true }),
2313+
}),
2314+
);
2315+
expect(broadcastTurnComplete).not.toHaveBeenCalled();
2316+
expect(resetTurnMessages).not.toHaveBeenCalled();
2317+
}, 20000);
2318+
2319+
it("declines steering without blocking on a fallback normal turn", async () => {
2320+
const s = createServer();
2321+
await s.start();
2322+
const prompt = vi.fn();
2323+
const broadcastTurnComplete = vi.fn();
2324+
const resetTurnMessages = vi.fn();
2325+
const serverInternals = s as unknown as {
2326+
activeOwnedTurnCount: number;
2327+
broadcastTurnComplete: typeof broadcastTurnComplete;
2328+
session: {
2329+
clientConnection: { prompt: typeof prompt };
2330+
logWriter: { resetTurnMessages: typeof resetTurnMessages };
2331+
};
2332+
};
2333+
serverInternals.activeOwnedTurnCount = 1;
2334+
prompt.mockImplementationOnce(async () => {
2335+
serverInternals.activeOwnedTurnCount = 0;
2336+
return { stopReason: "end_turn", _meta: { steer: false } };
2337+
});
2338+
serverInternals.broadcastTurnComplete = broadcastTurnComplete;
2339+
serverInternals.session.clientConnection.prompt = prompt;
2340+
serverInternals.session.logWriter.resetTurnMessages = resetTurnMessages;
2341+
2342+
const response = await fetch(`http://localhost:${port}/command`, {
2343+
method: "POST",
2344+
headers: {
2345+
Authorization: `Bearer ${createToken()}`,
2346+
"Content-Type": "application/json",
2347+
},
2348+
body: JSON.stringify({
2349+
jsonrpc: "2.0",
2350+
id: "steer-race",
2351+
method: "user_message",
2352+
params: { content: "continue normally", steer: true },
2353+
}),
2354+
});
2355+
2356+
expect(response.status).toBe(200);
2357+
await expect(response.json()).resolves.toMatchObject({
2358+
result: { stopReason: "steer_declined", steered: false },
2359+
});
2360+
expect(prompt).toHaveBeenCalledTimes(1);
2361+
expect(prompt.mock.calls[0]?.[0]).toEqual(
2362+
expect.objectContaining({
2363+
_meta: expect.objectContaining({ steer: true }),
2364+
}),
2365+
);
2366+
expect(resetTurnMessages).not.toHaveBeenCalled();
2367+
expect(broadcastTurnComplete).not.toHaveBeenCalled();
2368+
}, 20000);
2369+
22702370
it("redelivers a messageId whose first delivery failed before producing a turn", async () => {
22712371
const s = createServer();
22722372
await s.start();
@@ -2307,6 +2407,163 @@ describe("AgentServer HTTP Mode", () => {
23072407
expect(prompt).toHaveBeenCalledTimes(2);
23082408
}, 20000);
23092409

2410+
it("keeps a recoverable delivery committed across an ambiguous retry", async () => {
2411+
const s = createServer();
2412+
await s.start();
2413+
const prompt = vi
2414+
.fn()
2415+
.mockRejectedValue(new Error("API Error: The operation timed out."));
2416+
const serverInternals = s as unknown as {
2417+
session: { clientConnection: { prompt: typeof prompt } };
2418+
};
2419+
serverInternals.session.clientConnection.prompt = prompt;
2420+
2421+
const token = createToken();
2422+
const send = async (requestId: string) =>
2423+
fetch(`http://localhost:${port}/command`, {
2424+
method: "POST",
2425+
headers: {
2426+
Authorization: `Bearer ${token}`,
2427+
"Content-Type": "application/json",
2428+
},
2429+
body: JSON.stringify({
2430+
jsonrpc: "2.0",
2431+
id: requestId,
2432+
method: "user_message",
2433+
params: {
2434+
content: "do the thing",
2435+
messageId: "m-recoverable",
2436+
},
2437+
}),
2438+
});
2439+
2440+
const first = await send("first-attempt");
2441+
await expect(first.json()).resolves.toMatchObject({
2442+
result: { stopReason: "error_recoverable" },
2443+
});
2444+
expect(prompt).toHaveBeenCalledTimes(1);
2445+
2446+
const retry = await send("ambiguous-retry");
2447+
await expect(retry.json()).resolves.toMatchObject({
2448+
result: { stopReason: "duplicate_delivery", duplicate: true },
2449+
});
2450+
expect(prompt).toHaveBeenCalledTimes(1);
2451+
}, 20000);
2452+
2453+
it("shares a failed in-flight messageId outcome with concurrent retries", async () => {
2454+
const s = createServer();
2455+
await s.start();
2456+
let rejectFirstDelivery!: (error: Error) => void;
2457+
const prompt = vi
2458+
.fn()
2459+
.mockImplementationOnce(
2460+
() =>
2461+
new Promise((_resolve, reject) => {
2462+
rejectFirstDelivery = reject;
2463+
}),
2464+
)
2465+
.mockResolvedValueOnce({ stopReason: "end_turn" });
2466+
const serverInternals = s as unknown as {
2467+
logger: { info: (...args: unknown[]) => void };
2468+
session: { clientConnection: { prompt: typeof prompt } };
2469+
};
2470+
serverInternals.session.clientConnection.prompt = prompt;
2471+
const infoLog = vi.spyOn(serverInternals.logger, "info");
2472+
2473+
const token = createToken();
2474+
const send = async (requestId: string) =>
2475+
fetch(`http://localhost:${port}/command`, {
2476+
method: "POST",
2477+
headers: {
2478+
Authorization: `Bearer ${token}`,
2479+
"Content-Type": "application/json",
2480+
},
2481+
body: JSON.stringify({
2482+
jsonrpc: "2.0",
2483+
id: requestId,
2484+
method: "user_message",
2485+
params: { content: "do the thing", messageId: "m-concurrent" },
2486+
}),
2487+
});
2488+
2489+
const firstResponse = send("first-attempt");
2490+
await vi.waitFor(() => expect(prompt).toHaveBeenCalledTimes(1));
2491+
2492+
let retrySettled = false;
2493+
const retryResponse = send("concurrent-retry").finally(() => {
2494+
retrySettled = true;
2495+
});
2496+
await vi.waitFor(() => {
2497+
expect(infoLog).toHaveBeenCalledWith(
2498+
"Awaiting in-flight user_message delivery",
2499+
{ messageId: "m-concurrent" },
2500+
);
2501+
expect(prompt).toHaveBeenCalledTimes(1);
2502+
expect(retrySettled).toBe(false);
2503+
});
2504+
2505+
rejectFirstDelivery(new Error("sdk connection lost"));
2506+
const [first, retry] = await Promise.all([firstResponse, retryResponse]);
2507+
await expect(first.json()).resolves.toMatchObject({
2508+
error: { message: "sdk connection lost" },
2509+
});
2510+
await expect(retry.json()).resolves.toMatchObject({
2511+
error: { message: "sdk connection lost" },
2512+
});
2513+
expect(prompt).toHaveBeenCalledTimes(1);
2514+
}, 20000);
2515+
2516+
it("keeps an accepted messageId committed when teardown clears the active session", async () => {
2517+
const s = createServer();
2518+
await s.start();
2519+
let finishPrompt!: (result: { stopReason: "end_turn" }) => void;
2520+
const prompt = vi.fn(
2521+
() =>
2522+
new Promise<{ stopReason: "end_turn" }>((resolve) => {
2523+
finishPrompt = resolve;
2524+
}),
2525+
);
2526+
const serverInternals = s as unknown as {
2527+
session: { clientConnection: { prompt: typeof prompt } } | null;
2528+
};
2529+
const acceptedSession = serverInternals.session;
2530+
if (!acceptedSession) throw new Error("expected active test session");
2531+
acceptedSession.clientConnection.prompt = prompt;
2532+
2533+
const token = createToken();
2534+
const send = async (requestId: string) =>
2535+
fetch(`http://localhost:${port}/command`, {
2536+
method: "POST",
2537+
headers: {
2538+
Authorization: `Bearer ${token}`,
2539+
"Content-Type": "application/json",
2540+
},
2541+
body: JSON.stringify({
2542+
jsonrpc: "2.0",
2543+
id: requestId,
2544+
method: "user_message",
2545+
params: { content: "do the thing", messageId: "m-teardown" },
2546+
}),
2547+
});
2548+
2549+
const firstResponse = send("first-attempt");
2550+
await vi.waitFor(() => expect(prompt).toHaveBeenCalledTimes(1));
2551+
2552+
serverInternals.session = null;
2553+
finishPrompt({ stopReason: "end_turn" });
2554+
const first = await firstResponse;
2555+
await expect(first.json()).resolves.toMatchObject({
2556+
result: { stopReason: "end_turn" },
2557+
});
2558+
2559+
serverInternals.session = acceptedSession;
2560+
const retry = await send("retry");
2561+
await expect(retry.json()).resolves.toMatchObject({
2562+
result: { stopReason: "duplicate_delivery", duplicate: true },
2563+
});
2564+
expect(prompt).toHaveBeenCalledTimes(1);
2565+
}, 20000);
2566+
23102567
// Shared plumbing for the relay-echo tests: install a controllable
23112568
// prompt, stub the log writer so relayAgentResponse has an answer to
23122569
// relay, and spy on the relay_message client call.
@@ -2447,6 +2704,7 @@ describe("AgentServer HTTP Mode", () => {
24472704
expect(runStarted?.notification?.params).toMatchObject({
24482705
runId: "test-run-id",
24492706
taskId: "test-task-id",
2707+
steering: "native",
24502708
});
24512709
// Agent reports its semver so clients can gate UI features
24522710
// against agent capabilities (e.g. `>=0.40.1`). The exact value

0 commit comments

Comments
 (0)