Merge PR #8889: fix(cursor): answer interactionQuery and resume idle-stall MCP turns (@bnivanov)
This commit is contained in:
can1357
@@ -54,7 +54,7 @@ Current retryable categories include:
|
||||
|
||||
The normalized classifier recognizes the transient categories above from structured flags/status and provider-aware text patterns. Classifier refusals remain a separate typed `stopDetails` decision.
|
||||
|
||||
Beyond `isRetryableError(...)`, empty generic aborts may enter the same retry engine when no user, dispose, or streaming-edit-guard abort is in progress. An interrupted turn whose tool calls already have matching results can also be continued safely: the failed assistant/tool-result sequence is preserved so completed side effects are not replayed. Resolved stream stalls and HTTP/2 stream resets (`NGHTTP2_INTERNAL_ERROR`, `NGHTTP2_REFUSED_STREAM`, `HTTP2StreamReset`) use the same preserve-and-continue path. Cursor idle-stall recovery still requires the exec-resolved marker (the Connect stream may still be open); an HTTP/2 RST does not, because the stream is already dead.
|
||||
Beyond `isRetryableError(...)`, empty generic aborts may enter the same retry engine when no user, dispose, or streaming-edit-guard abort is in progress. An interrupted turn whose tool calls already have matching results can also be continued safely: the failed assistant/tool-result sequence is preserved so completed side effects are not replayed. Resolved stream stalls and HTTP/2 stream resets (`NGHTTP2_INTERNAL_ERROR`, `NGHTTP2_REFUSED_STREAM`, `HTTP2StreamReset`) use the same preserve-and-continue path. Cursor idle-stall recovery continues after every emitted tool call has a result; the Connect stream is already closed by the idle abort. An HTTP/2 RST is the same: the stream is already dead.
|
||||
|
||||
Retry state is owned by `TurnRecovery`:
|
||||
|
||||
|
||||
@@ -450,8 +450,12 @@ Cursor's integration in `packages/ai` operates over an HTTP/2 Connect RPC transp
|
||||
- **Trailer & Transport Error Handling**:
|
||||
- Monitors HTTP/2 trailers (`grpc-status`, `grpc-message`) and maps socket or TLS disconnects using `mapH2TransportError`.
|
||||
- **Bi-Directional RPC Dispatch**:
|
||||
- Server streams `AgentServerMessage` (`interactionUpdate`, `execServerMessage`, `kvServerMessage`).
|
||||
- Client writes `AgentClientMessage` (`runRequest`, periodic `clientHeartbeat` every 5 seconds) and `ExecClientMessage` tool responses (`readResult`, `writeResult`, `execClientThrow`, `requestContextResult`).
|
||||
- Server streams `AgentServerMessage` (`interactionUpdate`, `execServerMessage`, `kvServerMessage`, `interactionQuery`).
|
||||
- Client writes `AgentClientMessage` (`runRequest`, periodic `clientHeartbeat` every 5 seconds, `interactionResponse`) and `ExecClientMessage` tool responses (`readResult`, `writeResult`, `execClientThrow`, `requestContextResult`).
|
||||
- **Interaction Query Handshake**:
|
||||
- Hosted web search / Exa / unnamed field-9 WebFetch send `interactionQuery` and block the turn until the client writes `interactionResponse`.
|
||||
- Heartbeats keep HTTP/2 alive but are not semantic progress; an unanswered query sits silent until the 300s idle watchdog (`Provider stream stalled while waiting for the next event`).
|
||||
- `handleInteractionQuery` approves network permission gates and rejects interactive ask / switch-mode / create-plan. VM setup is left unanswered because its result oneof is success-only.
|
||||
- **Async Execution Drain & Turn Completion**:
|
||||
- `handleServerMessage` processes frames asynchronously so the socket continues draining. Dispatches are tracked in `inFlightDispatches` and bounded by `options.signal` abort handling before finalizing stream completion.
|
||||
- Stream completion verifies `turnEnded` (`sawTurnEnded`) or throws `incomplete-stream`.
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
### Fixed
|
||||
|
||||
- Cloud Code Assist Gemini 3.6/3.7 Flash requests at `minimal` now send `thinkingLevel: LOW` on the aliased `-low` SKU instead of `MINIMAL`, which the API rejects with HTTP 400.
|
||||
- Answer Cursor `interaction_query` permission gates (hosted web search, Exa, unnamed field-9 WebFetch) so the Run RPC continues instead of sitting silent until the 300s idle watchdog.
|
||||
|
||||
## [17.3.7] - 2026-08-17
|
||||
|
||||
|
||||
@@ -220,6 +220,7 @@ import {
|
||||
piReadPathHasRange,
|
||||
piTimeout,
|
||||
} from "./cursor/exec-modern";
|
||||
import { handleInteractionQuery } from "./cursor/interaction-query";
|
||||
|
||||
export const CURSOR_API_URL = "https://api2.cursor.sh";
|
||||
export const CURSOR_CLIENT_VERSION = "cli-2026.07.23-e383d2b";
|
||||
@@ -1024,6 +1025,8 @@ export async function handleServerMessage(
|
||||
state,
|
||||
),
|
||||
);
|
||||
} else if (msgCase === "interactionQuery") {
|
||||
handleInteractionQuery(msg.message.value, h2Request);
|
||||
} else if (msgCase === "conversationCheckpointUpdate") {
|
||||
handleConversationCheckpointUpdate(msg.message.value, output, usageState, onConversationCheckpoint);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,200 @@
|
||||
import type http2 from "node:http2";
|
||||
import { create, toBinary } from "@bufbuild/protobuf";
|
||||
import {
|
||||
AgentClientMessageSchema,
|
||||
AskQuestionInteractionResponseSchema,
|
||||
AskQuestionRejectedSchema,
|
||||
AskQuestionResultSchema,
|
||||
CreatePlanErrorSchema,
|
||||
CreatePlanRequestResponseSchema,
|
||||
CreatePlanResultSchema,
|
||||
ExaFetchRequestResponse_ApprovedSchema,
|
||||
ExaFetchRequestResponseSchema,
|
||||
ExaSearchRequestResponse_ApprovedSchema,
|
||||
ExaSearchRequestResponseSchema,
|
||||
type InteractionQuery,
|
||||
type InteractionResponse,
|
||||
InteractionResponseSchema,
|
||||
SwitchModeRequestResponse_RejectedSchema,
|
||||
SwitchModeRequestResponseSchema,
|
||||
WebSearchRequestResponse_ApprovedSchema,
|
||||
WebSearchRequestResponseSchema,
|
||||
} from "@oh-my-pi/pi-catalog/discovery/cursor-gen/agent_pb";
|
||||
import { $env } from "@oh-my-pi/pi-utils";
|
||||
|
||||
const NOT_IMPLEMENTED_SUFFIX = "not implemented by this client";
|
||||
|
||||
type ProtoUnknownField = { no: number; wireType: number; data: Uint8Array };
|
||||
type ProtoUnknownBag = { $unknown?: ProtoUnknownField[] };
|
||||
|
||||
type InteractionQueryCase = NonNullable<InteractionQuery["query"]["case"]>;
|
||||
type InteractionResult = Exclude<InteractionResponse["result"], { case: undefined; value?: undefined }>;
|
||||
|
||||
function frameConnectMessage(data: Uint8Array, flags = 0): Buffer {
|
||||
const frame = Buffer.alloc(5 + data.length);
|
||||
frame[0] = flags;
|
||||
frame.writeUInt32BE(data.length, 1);
|
||||
frame.set(data, 5);
|
||||
return frame;
|
||||
}
|
||||
|
||||
function isProtoUnknownField(value: unknown): value is ProtoUnknownField {
|
||||
if (!value || typeof value !== "object") return false;
|
||||
if (!("no" in value) || !("wireType" in value) || !("data" in value)) return false;
|
||||
return typeof value.no === "number" && typeof value.wireType === "number" && value.data instanceof Uint8Array;
|
||||
}
|
||||
|
||||
function protoUnknownFields(message: object): ProtoUnknownField[] {
|
||||
if (!("$unknown" in message) || !Array.isArray(message.$unknown)) return [];
|
||||
return message.$unknown.filter(isProtoUnknownField);
|
||||
}
|
||||
|
||||
function attachUnknownApprovedField(response: InteractionResponse, fieldNo: number): void {
|
||||
const bag: ProtoUnknownBag = response; // protobuf-es unnamed oneof members live on $unknown
|
||||
// protobuf-es writes tag + raw(data); LEN fields need the length prefix in data.
|
||||
const field: ProtoUnknownField = { no: fieldNo, wireType: 2, data: new Uint8Array([0x02, 0x0a, 0x00]) };
|
||||
const existing = bag.$unknown;
|
||||
if (Array.isArray(existing)) {
|
||||
existing.push(field);
|
||||
return;
|
||||
}
|
||||
bag.$unknown = [field];
|
||||
}
|
||||
|
||||
function log(type: string, subtype?: string, data?: unknown): void {
|
||||
if (!$env.DEBUG_CURSOR) return;
|
||||
const verbose = $env.DEBUG_CURSOR === "2" || $env.DEBUG_CURSOR === "verbose";
|
||||
const dataStr = verbose && data ? ` ${JSON.stringify(data)?.slice(0, 500)}` : "";
|
||||
console.error(`[CURSOR] ${type}${subtype ? `: ${subtype}` : ""}${dataStr}`);
|
||||
}
|
||||
|
||||
function sendInteractionResponse(h2Request: http2.ClientHttp2Stream, queryId: number, result: InteractionResult): void {
|
||||
const response = create(InteractionResponseSchema, { id: queryId, result });
|
||||
const clientMessage = create(AgentClientMessageSchema, {
|
||||
message: { case: "interactionResponse", value: response },
|
||||
});
|
||||
h2Request.write(frameConnectMessage(toBinary(AgentClientMessageSchema, clientMessage)));
|
||||
log("interactionResponse", result.case, { id: queryId });
|
||||
}
|
||||
|
||||
function sendUnknownApprovedInteractionResponse(
|
||||
h2Request: http2.ClientHttp2Stream,
|
||||
queryId: number,
|
||||
fieldNo: number,
|
||||
): void {
|
||||
// `approved {}` on the matching response oneof: field N, empty message whose
|
||||
// first member is field 1 (`approved`) with an empty length-delimited payload.
|
||||
const response = create(InteractionResponseSchema, { id: queryId });
|
||||
attachUnknownApprovedField(response, fieldNo);
|
||||
const clientMessage = create(AgentClientMessageSchema, {
|
||||
message: { case: "interactionResponse", value: response },
|
||||
});
|
||||
h2Request.write(frameConnectMessage(toBinary(AgentClientMessageSchema, clientMessage)));
|
||||
log("interactionResponse", "unknownApproved", { id: queryId, field: fieldNo });
|
||||
}
|
||||
|
||||
/**
|
||||
* Answer a Cursor `interaction_query` so the Run RPC can continue.
|
||||
*
|
||||
* Hosted web search / Exa / unnamed permission gates (field 9 = WebFetch on
|
||||
* current Cursor builds) block the turn until the client writes an
|
||||
* `interaction_response`. Dropping the frame leaves the HTTP/2 stream alive
|
||||
* on heartbeats that are not semantic progress; the lazy idle watchdog then
|
||||
* aborts with "Provider stream stalled while waiting for the next event".
|
||||
*
|
||||
* Unsupported interactive queries are rejected so the server is not stranded.
|
||||
* VM setup is left unanswered rather than reporting a fake success.
|
||||
*/
|
||||
export function handleInteractionQuery(query: InteractionQuery, h2Request: http2.ClientHttp2Stream): void {
|
||||
const queryCase = query.query.case;
|
||||
log("interactionQuery", queryCase, query.query.value);
|
||||
if (!queryCase) {
|
||||
const unknown = protoUnknownFields(query).find(field => field.wireType === 2 && field.no >= 2);
|
||||
if (unknown) {
|
||||
log("warn", "unknownInteractionQueryApproved", { id: query.id, field: unknown.no });
|
||||
sendUnknownApprovedInteractionResponse(h2Request, query.id, unknown.no);
|
||||
return;
|
||||
}
|
||||
log("warn", "unknownInteractionQuery", { id: query.id });
|
||||
return;
|
||||
}
|
||||
|
||||
switch (queryCase) {
|
||||
case "webSearchRequestQuery":
|
||||
sendInteractionResponse(h2Request, query.id, {
|
||||
case: "webSearchRequestResponse",
|
||||
value: create(WebSearchRequestResponseSchema, {
|
||||
result: { case: "approved", value: create(WebSearchRequestResponse_ApprovedSchema, {}) },
|
||||
}),
|
||||
});
|
||||
return;
|
||||
case "exaSearchRequestQuery":
|
||||
sendInteractionResponse(h2Request, query.id, {
|
||||
case: "exaSearchRequestResponse",
|
||||
value: create(ExaSearchRequestResponseSchema, {
|
||||
result: { case: "approved", value: create(ExaSearchRequestResponse_ApprovedSchema, {}) },
|
||||
}),
|
||||
});
|
||||
return;
|
||||
case "exaFetchRequestQuery":
|
||||
sendInteractionResponse(h2Request, query.id, {
|
||||
case: "exaFetchRequestResponse",
|
||||
value: create(ExaFetchRequestResponseSchema, {
|
||||
result: { case: "approved", value: create(ExaFetchRequestResponse_ApprovedSchema, {}) },
|
||||
}),
|
||||
});
|
||||
return;
|
||||
case "askQuestionInteractionQuery":
|
||||
sendInteractionResponse(h2Request, query.id, {
|
||||
case: "askQuestionInteractionResponse",
|
||||
value: create(AskQuestionInteractionResponseSchema, {
|
||||
result: create(AskQuestionResultSchema, {
|
||||
result: {
|
||||
case: "rejected",
|
||||
value: create(AskQuestionRejectedSchema, {
|
||||
reason: `Interactive questions are ${NOT_IMPLEMENTED_SUFFIX}`,
|
||||
}),
|
||||
},
|
||||
}),
|
||||
}),
|
||||
});
|
||||
return;
|
||||
case "switchModeRequestQuery":
|
||||
sendInteractionResponse(h2Request, query.id, {
|
||||
case: "switchModeRequestResponse",
|
||||
value: create(SwitchModeRequestResponseSchema, {
|
||||
result: {
|
||||
case: "rejected",
|
||||
value: create(SwitchModeRequestResponse_RejectedSchema, {
|
||||
reason: `Mode switches are ${NOT_IMPLEMENTED_SUFFIX}`,
|
||||
}),
|
||||
},
|
||||
}),
|
||||
});
|
||||
return;
|
||||
case "createPlanRequestQuery":
|
||||
sendInteractionResponse(h2Request, query.id, {
|
||||
case: "createPlanRequestResponse",
|
||||
value: create(CreatePlanRequestResponseSchema, {
|
||||
result: create(CreatePlanResultSchema, {
|
||||
result: {
|
||||
case: "error",
|
||||
value: create(CreatePlanErrorSchema, {
|
||||
error: `Plan files are ${NOT_IMPLEMENTED_SUFFIX}`,
|
||||
}),
|
||||
},
|
||||
}),
|
||||
}),
|
||||
});
|
||||
return;
|
||||
case "setupVmEnvironmentArgs":
|
||||
// Result oneof is success-only. Do not invent a VM; silence is better
|
||||
// than a false SetupVmEnvironmentSuccess (review of #8047).
|
||||
log("warn", "unhandledInteractionQuery", { queryCase, id: query.id });
|
||||
return;
|
||||
default: {
|
||||
const _exhaustive: InteractionQueryCase = queryCase;
|
||||
log("warn", "unhandledInteractionQuery", { queryCase: _exhaustive, id: query.id });
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,219 @@
|
||||
import { describe, expect, it } from "bun:test";
|
||||
import { create, fromBinary } from "@bufbuild/protobuf";
|
||||
import { type BlockState, handleServerMessage, type ToolCallState } from "@oh-my-pi/pi-ai/providers/cursor";
|
||||
import type { AssistantMessage } from "@oh-my-pi/pi-ai/types";
|
||||
import { AssistantMessageEventStream } from "@oh-my-pi/pi-ai/utils/event-stream";
|
||||
import {
|
||||
type AgentClientMessage,
|
||||
AgentClientMessageSchema,
|
||||
AgentServerMessageSchema,
|
||||
AskQuestionInteractionQuerySchema,
|
||||
CreatePlanRequestQuerySchema,
|
||||
ExaFetchRequestQuerySchema,
|
||||
ExaSearchRequestQuerySchema,
|
||||
type InteractionQuery,
|
||||
InteractionQuerySchema,
|
||||
SetupVmEnvironmentArgsSchema,
|
||||
SwitchModeRequestQuerySchema,
|
||||
WebSearchRequestQuerySchema,
|
||||
} from "@oh-my-pi/pi-catalog/discovery/cursor-gen/agent_pb";
|
||||
|
||||
type ProtoUnknownField = { no: number; wireType: number; data: Uint8Array };
|
||||
type ProtoUnknownBag = { $unknown?: ProtoUnknownField[] };
|
||||
|
||||
function cursorAssistantMessage(): AssistantMessage {
|
||||
return {
|
||||
role: "assistant",
|
||||
api: "cursor-agent",
|
||||
provider: "cursor",
|
||||
model: "cursor-composer-2.5",
|
||||
content: [],
|
||||
usage: {
|
||||
input: 0,
|
||||
output: 0,
|
||||
cacheRead: 0,
|
||||
cacheWrite: 0,
|
||||
totalTokens: 0,
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
|
||||
},
|
||||
stopReason: "stop",
|
||||
timestamp: 1,
|
||||
};
|
||||
}
|
||||
|
||||
function emptyBlockState(): BlockState {
|
||||
let textBlock: BlockState["currentTextBlock"] = null;
|
||||
let thinkingBlock: BlockState["currentThinkingBlock"] = null;
|
||||
let toolCall: ToolCallState | null = null;
|
||||
return {
|
||||
get currentTextBlock() {
|
||||
return textBlock;
|
||||
},
|
||||
get currentThinkingBlock() {
|
||||
return thinkingBlock;
|
||||
},
|
||||
get currentToolCall() {
|
||||
return toolCall;
|
||||
},
|
||||
openToolCalls: new Map(),
|
||||
resolvedMcpToolCallIds: new Set(),
|
||||
firstTokenTime: undefined,
|
||||
setTextBlock: b => {
|
||||
textBlock = b;
|
||||
},
|
||||
setThinkingBlock: b => {
|
||||
thinkingBlock = b;
|
||||
},
|
||||
setToolCall: t => {
|
||||
toolCall = t;
|
||||
},
|
||||
setFirstTokenTime: () => {},
|
||||
};
|
||||
}
|
||||
|
||||
function decodeConnectFrame(frame: Buffer): AgentClientMessage {
|
||||
return fromBinary(AgentClientMessageSchema, frame.subarray(5));
|
||||
}
|
||||
|
||||
async function dispatchQuery(query: InteractionQuery): Promise<Buffer[]> {
|
||||
const frames: Buffer[] = [];
|
||||
const h2Request = {
|
||||
write(chunk: Buffer) {
|
||||
frames.push(chunk);
|
||||
return true;
|
||||
},
|
||||
} as unknown as Parameters<typeof handleServerMessage>[5];
|
||||
const serverMsg = create(AgentServerMessageSchema, {
|
||||
message: { case: "interactionQuery", value: query },
|
||||
});
|
||||
await handleServerMessage(
|
||||
serverMsg,
|
||||
cursorAssistantMessage(),
|
||||
new AssistantMessageEventStream(),
|
||||
emptyBlockState(),
|
||||
new Map(),
|
||||
h2Request,
|
||||
undefined,
|
||||
undefined,
|
||||
{ sawTokenDelta: false },
|
||||
[],
|
||||
);
|
||||
return frames;
|
||||
}
|
||||
|
||||
describe("cursor interaction query handshake", () => {
|
||||
it("approves hosted web search so the Run RPC can continue", async () => {
|
||||
const frames = await dispatchQuery(
|
||||
create(InteractionQuerySchema, {
|
||||
id: 11,
|
||||
query: { case: "webSearchRequestQuery", value: create(WebSearchRequestQuerySchema, {}) },
|
||||
}),
|
||||
);
|
||||
expect(frames).toHaveLength(1);
|
||||
const client = decodeConnectFrame(frames[0]!);
|
||||
expect(client.message.case).toBe("interactionResponse");
|
||||
expect(client.message.value).toMatchObject({
|
||||
id: 11,
|
||||
result: { case: "webSearchRequestResponse", value: { result: { case: "approved" } } },
|
||||
});
|
||||
});
|
||||
|
||||
it("approves hosted Exa search and fetch permission queries", async () => {
|
||||
const search = await dispatchQuery(
|
||||
create(InteractionQuerySchema, {
|
||||
id: 12,
|
||||
query: { case: "exaSearchRequestQuery", value: create(ExaSearchRequestQuerySchema, {}) },
|
||||
}),
|
||||
);
|
||||
const fetch = await dispatchQuery(
|
||||
create(InteractionQuerySchema, {
|
||||
id: 13,
|
||||
query: { case: "exaFetchRequestQuery", value: create(ExaFetchRequestQuerySchema, {}) },
|
||||
}),
|
||||
);
|
||||
expect(decodeConnectFrame(search[0]!).message.value).toMatchObject({
|
||||
id: 12,
|
||||
result: { case: "exaSearchRequestResponse", value: { result: { case: "approved" } } },
|
||||
});
|
||||
expect(decodeConnectFrame(fetch[0]!).message.value).toMatchObject({
|
||||
id: 13,
|
||||
result: { case: "exaFetchRequestResponse", value: { result: { case: "approved" } } },
|
||||
});
|
||||
});
|
||||
|
||||
it("approves unnamed field-9 permission queries used by hosted WebFetch", async () => {
|
||||
const query = create(InteractionQuerySchema, { id: 18 });
|
||||
const bag: ProtoUnknownBag = query; // protobuf-es unnamed query oneof (field 9)
|
||||
bag.$unknown = [{ no: 9, wireType: 2, data: new Uint8Array([0x02, 0x0a, 0x00]) }];
|
||||
const frames = await dispatchQuery(query);
|
||||
expect(frames).toHaveLength(1);
|
||||
const client = decodeConnectFrame(frames[0]!);
|
||||
expect(client.message.case).toBe("interactionResponse");
|
||||
if (client.message.case !== "interactionResponse") {
|
||||
throw new Error("expected interactionResponse");
|
||||
}
|
||||
expect(client.message.value.id).toBe(18);
|
||||
const responseBag: ProtoUnknownBag = client.message.value;
|
||||
expect(responseBag.$unknown?.some(field => field.no === 9 && field.wireType === 2)).toBe(true);
|
||||
});
|
||||
|
||||
it("rejects interactive ask / switch-mode / create-plan queries", async () => {
|
||||
const ask = decodeConnectFrame(
|
||||
(
|
||||
await dispatchQuery(
|
||||
create(InteractionQuerySchema, {
|
||||
id: 14,
|
||||
query: { case: "askQuestionInteractionQuery", value: create(AskQuestionInteractionQuerySchema, {}) },
|
||||
}),
|
||||
)
|
||||
)[0]!,
|
||||
);
|
||||
const mode = decodeConnectFrame(
|
||||
(
|
||||
await dispatchQuery(
|
||||
create(InteractionQuerySchema, {
|
||||
id: 15,
|
||||
query: { case: "switchModeRequestQuery", value: create(SwitchModeRequestQuerySchema, {}) },
|
||||
}),
|
||||
)
|
||||
)[0]!,
|
||||
);
|
||||
const plan = decodeConnectFrame(
|
||||
(
|
||||
await dispatchQuery(
|
||||
create(InteractionQuerySchema, {
|
||||
id: 16,
|
||||
query: { case: "createPlanRequestQuery", value: create(CreatePlanRequestQuerySchema, {}) },
|
||||
}),
|
||||
)
|
||||
)[0]!,
|
||||
);
|
||||
expect(ask.message.value).toMatchObject({
|
||||
id: 14,
|
||||
result: {
|
||||
case: "askQuestionInteractionResponse",
|
||||
value: { result: { result: { case: "rejected" } } },
|
||||
},
|
||||
});
|
||||
expect(mode.message.value).toMatchObject({
|
||||
id: 15,
|
||||
result: { case: "switchModeRequestResponse", value: { result: { case: "rejected" } } },
|
||||
});
|
||||
expect(plan.message.value).toMatchObject({
|
||||
id: 16,
|
||||
result: { case: "createPlanRequestResponse", value: { result: { result: { case: "error" } } } },
|
||||
});
|
||||
});
|
||||
|
||||
it("does not invent a VM success or a reply for an empty query", async () => {
|
||||
const vm = await dispatchQuery(
|
||||
create(InteractionQuerySchema, {
|
||||
id: 19,
|
||||
query: { case: "setupVmEnvironmentArgs", value: create(SetupVmEnvironmentArgsSchema, {}) },
|
||||
}),
|
||||
);
|
||||
const empty = await dispatchQuery(create(InteractionQuerySchema, { id: 17 }));
|
||||
expect(vm).toEqual([]);
|
||||
expect(empty).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -14,6 +14,9 @@
|
||||
### Fixed
|
||||
|
||||
- Fixed images returned by tools mounted under `xd://` rendering only as file links instead of inline terminal graphics.
|
||||
### Fixed
|
||||
|
||||
- Resume Cursor idle-stall turns after completed MCP/todo tool results. The watchdog already closes the Connect stream, so unmarked blocks no longer need the `exec-resolved` marker to continue.
|
||||
|
||||
## [17.3.7] - 2026-08-17
|
||||
|
||||
|
||||
@@ -20,7 +20,6 @@ import type {
|
||||
} from "@oh-my-pi/pi-ai";
|
||||
import { calculateRateLimitBackoffMs, parseRateLimitReason } from "@oh-my-pi/pi-ai";
|
||||
import * as AIError from "@oh-my-pi/pi-ai/error";
|
||||
import { kCursorExecResolved } from "@oh-my-pi/pi-ai/utils/block-symbols";
|
||||
import { isFireworksFastModelId, toFireworksBaseModelId } from "@oh-my-pi/pi-catalog/fireworks-model-id";
|
||||
import { modelsAreEqual } from "@oh-my-pi/pi-catalog/models";
|
||||
import { extractRetryHint, logger, prompt } from "@oh-my-pi/pi-utils";
|
||||
@@ -1177,25 +1176,15 @@ export class TurnRecovery {
|
||||
if (!reasonlessAbort && !streamStall && !transportReset) return undefined;
|
||||
if (reasonlessAbort && genericAbort) message.errorId = AIError.create(AIError.Flag.Abort);
|
||||
|
||||
// The Cursor server-execution marker gate applies only to the idle stream-stall
|
||||
// path: an unmarked/unresolved Cursor block there means the server has not
|
||||
// finished executing, so resuming would race it. A reasonless abort instead
|
||||
// ends the turn and the agent loop pairs every un-run call (Cursor's unmarked
|
||||
// `todo`/MCP blocks included) with a synthetic `executed: false` result, so
|
||||
// the tool-result reconciliation below is the safety gate and the marker is
|
||||
// irrelevant. An HTTP/2 RST_STREAM / NGHTTP2_* close also ends the Connect
|
||||
// stream, so there is no in-flight server exec to race — unmarked MCP/todo
|
||||
// blocks are safe to continue once every emitted call has a result.
|
||||
// Idle stall and HTTP/2 RST both close the Cursor Connect stream:
|
||||
// the lazy watchdog aborts the request signal, and cursor.ts then
|
||||
// calls `h2Request.close()`. There is no in-flight server exec to
|
||||
// race, so unmarked MCP/todo blocks can continue once every emitted
|
||||
// call has a matching result. A reasonless abort ends the turn and
|
||||
// the agent loop pairs leftover calls with `executed: false`.
|
||||
const resolvedToolCallIds: string[] = [];
|
||||
for (const block of message.content) {
|
||||
if (block.type !== "toolCall") continue;
|
||||
if (
|
||||
streamStall &&
|
||||
message.provider === "cursor" &&
|
||||
(!(kCursorExecResolved in block) || block[kCursorExecResolved] !== true)
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
resolvedToolCallIds.push(block.id);
|
||||
}
|
||||
if (resolvedToolCallIds.length === 0) return undefined;
|
||||
|
||||
@@ -1386,6 +1386,128 @@ describe("AgentSession retry delay cap", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("resumes a Cursor idle stall after an unmarked MCP tool result", async () => {
|
||||
const stallMessage = "Provider stream stalled while waiting for the next event";
|
||||
const model = createMockModel({
|
||||
id: "composer-2.5",
|
||||
provider: "cursor",
|
||||
});
|
||||
authStorage.setRuntimeApiKey("cursor", "cursor-test-key");
|
||||
const toolCall: ToolCall = {
|
||||
type: "toolCall",
|
||||
id: "cursor-mcp-idle-1",
|
||||
name: "mcp__databricks_production_execute_sql",
|
||||
arguments: { query: "SELECT 1" },
|
||||
};
|
||||
const toolResult: ToolResultMessage = {
|
||||
role: "toolResult",
|
||||
toolCallId: toolCall.id,
|
||||
toolName: toolCall.name,
|
||||
content: [{ type: "text", text: "1" }],
|
||||
isError: false,
|
||||
timestamp: Date.now(),
|
||||
};
|
||||
let streamCalls = 0;
|
||||
let resumedWithToolResult = false;
|
||||
const agent = new Agent({
|
||||
getApiKey: requestedModel => `${requestedModel.provider}-test-key`,
|
||||
initialState: {
|
||||
model,
|
||||
systemPrompt: ["Test"],
|
||||
tools: [],
|
||||
messages: [],
|
||||
},
|
||||
cursorOnToolResult: message => message,
|
||||
streamFn: (_requestedModel, context, options) => {
|
||||
streamCalls += 1;
|
||||
if (streamCalls > 1) {
|
||||
resumedWithToolResult = context.messages.some(
|
||||
message => message.role === "toolResult" && message.toolCallId === toolCall.id,
|
||||
);
|
||||
model.push({ content: ["Recovered after Cursor idle stall"] });
|
||||
return model.stream(model, context, options);
|
||||
}
|
||||
|
||||
const stream = new AssistantMessageEventStream();
|
||||
queueMicrotask(async () => {
|
||||
await options?.cursorOnToolResult?.(toolResult);
|
||||
const partial: AssistantMessage = {
|
||||
role: "assistant",
|
||||
content: [toolCall],
|
||||
api: model.api,
|
||||
provider: model.provider,
|
||||
model: model.id,
|
||||
usage: {
|
||||
input: 0,
|
||||
output: 0,
|
||||
cacheRead: 0,
|
||||
cacheWrite: 0,
|
||||
totalTokens: 0,
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
|
||||
},
|
||||
stopReason: "stop",
|
||||
timestamp: Date.now(),
|
||||
};
|
||||
stream.push({ type: "start", partial });
|
||||
stream.push({ type: "toolcall_start", contentIndex: 0, partial });
|
||||
stream.push({
|
||||
type: "toolcall_delta",
|
||||
contentIndex: 0,
|
||||
delta: JSON.stringify(toolCall.arguments),
|
||||
partial,
|
||||
});
|
||||
stream.push({ type: "toolcall_end", contentIndex: 0, toolCall, partial });
|
||||
stream.push({
|
||||
type: "error",
|
||||
reason: "error",
|
||||
error: {
|
||||
...partial,
|
||||
stopReason: "error",
|
||||
errorMessage: stallMessage,
|
||||
},
|
||||
});
|
||||
});
|
||||
return stream;
|
||||
},
|
||||
});
|
||||
|
||||
const settings = Settings.isolated({
|
||||
"compaction.enabled": false,
|
||||
"retry.baseDelayMs": 5,
|
||||
"retry.maxRetries": 1,
|
||||
});
|
||||
settings.setModelRole("default", `${model.provider}/${model.id}`);
|
||||
session = new AgentSession({
|
||||
agent,
|
||||
sessionManager: SessionManager.inMemory(),
|
||||
settings,
|
||||
modelRegistry,
|
||||
});
|
||||
const retryStartEvents: AutoRetryStartEvent[] = [];
|
||||
const retryEndEvents: AutoRetryEndEvent[] = [];
|
||||
session.subscribe(event => {
|
||||
if (event.type === "auto_retry_start") retryStartEvents.push(event);
|
||||
if (event.type === "auto_retry_end") retryEndEvents.push(event);
|
||||
});
|
||||
|
||||
await session.prompt("Run the query");
|
||||
await session.waitForIdle();
|
||||
|
||||
expect(streamCalls).toBe(2);
|
||||
expect(resumedWithToolResult).toBe(true);
|
||||
expect(
|
||||
session.agent.state.messages.some(
|
||||
message => message.role === "toolResult" && message.toolCallId === toolCall.id,
|
||||
),
|
||||
).toBe(true);
|
||||
expect(retryStartEvents).toHaveLength(1);
|
||||
expect(retryEndEvents).toContainEqual(expect.objectContaining({ success: true, attempt: 1 }));
|
||||
expect(lastAssistant(session).content).toContainEqual({
|
||||
type: "text",
|
||||
text: "Recovered after Cursor idle stall",
|
||||
});
|
||||
});
|
||||
|
||||
it("resumes a Cursor reasonless abort after an unmarked client-side tool call", async () => {
|
||||
const model = createMockModel({
|
||||
id: "composer-2.5",
|
||||
|
||||
@@ -603,10 +603,10 @@ describe("TurnRecovery replay-unsafe output classification", () => {
|
||||
expect(recovery.classifyResolvedInterruptedToolTurn(message)).toBe("stream-stall");
|
||||
});
|
||||
|
||||
it("does not continue a Cursor idle stall after an unmarked MCP call", () => {
|
||||
it("continues a Cursor idle stall after an unmarked MCP call", () => {
|
||||
const message = cursorMessage([mcpToolCall("mcp-1")], stallMessage);
|
||||
const recovery = recoveryForReset(message, [realResult("mcp-1", "mcp__databricks_production_execute_sql")]);
|
||||
expect(recovery.classifyResolvedInterruptedToolTurn(message)).toBeUndefined();
|
||||
expect(recovery.classifyResolvedInterruptedToolTurn(message)).toBe("stream-stall");
|
||||
});
|
||||
|
||||
it("does not continue an HTTP/2 reset whose tool call has no result", () => {
|
||||
|
||||
Reference in New Issue
Block a user