fix: 支持长工具调用 interrupt 与硬超时回收
This commit is contained in:
Codex
+91
-25
@@ -423,16 +423,75 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess
|
||||
let client: CodexStdioClient | null = null;
|
||||
const requestTimeoutMs = Math.min(positiveTimeout(options.timeoutMs), requestTimeoutCapMs);
|
||||
let stopActiveTurn: (() => void) | undefined;
|
||||
let activeTurnKey: string | null = null;
|
||||
let interruptInFlight: Promise<void> | null = null;
|
||||
let stopAfterInterrupt = false;
|
||||
const controlRequestTimeoutMs = Math.min(requestTimeoutMs, 5_000);
|
||||
const activeTurnControl = (activeThreadId: string, activeTurnId: string): CodexActiveTurnControl => ({
|
||||
threadId: activeThreadId,
|
||||
turnId: activeTurnId,
|
||||
steer: async (prompt: string) => {
|
||||
await client!.request("turn/steer", { threadId: activeThreadId, expectedTurnId: activeTurnId, input: textInput(prompt) }, requestTimeoutMs);
|
||||
},
|
||||
interrupt: async () => {
|
||||
await client!.request("turn/interrupt", { threadId: activeThreadId, turnId: activeTurnId }, controlRequestTimeoutMs);
|
||||
},
|
||||
});
|
||||
const exposeActiveTurn = (source: string): void => {
|
||||
if (!client || !threadId || !turnId || !options.onActiveTurn) return;
|
||||
const key = `${threadId}:${turnId}`;
|
||||
if (activeTurnKey === key) return;
|
||||
stopActiveTurn?.();
|
||||
activeTurnKey = key;
|
||||
emitEvent({ type: "backend_status", payload: { phase: "active-turn-control-ready", source, threadId, turnId } });
|
||||
const maybeStop = options.onActiveTurn(activeTurnControl(threadId, turnId));
|
||||
stopActiveTurn = typeof maybeStop === "function" ? maybeStop : undefined;
|
||||
};
|
||||
const requestInterrupt = (reason: string, triggerPhase: string): Promise<void> => {
|
||||
const activeClient = client;
|
||||
const activeThreadId = threadId;
|
||||
const activeTurnId = turnId;
|
||||
emitEvent({ type: "backend_status", payload: { phase: "turn-interrupt-requested", reason, triggerPhase, threadId: activeThreadId ?? null, turnId: activeTurnId ?? null } });
|
||||
if (!activeClient || !activeThreadId || !activeTurnId) {
|
||||
emitEvent({ type: "backend_status", payload: { phase: "turn-interrupt-unavailable", reason, triggerPhase, hasClient: Boolean(activeClient), hasThreadId: Boolean(activeThreadId), hasTurnId: Boolean(activeTurnId) } });
|
||||
activeClient?.stop();
|
||||
return Promise.resolve();
|
||||
}
|
||||
return activeClient.request("turn/interrupt", { threadId: activeThreadId, turnId: activeTurnId }, controlRequestTimeoutMs)
|
||||
.then(() => {
|
||||
emitEvent({ type: "backend_status", payload: { phase: "turn/interrupt:completed", reason, triggerPhase, threadId: activeThreadId, turnId: activeTurnId } });
|
||||
})
|
||||
.catch((error) => {
|
||||
const failure = normalizeFailure(error);
|
||||
emitEvent({ type: "error", payload: { failureKind: failure.failureKind, message: failure.message, phase: "turn/interrupt:failed", triggerPhase, details: failure.details } });
|
||||
})
|
||||
.finally(() => {
|
||||
if (stopAfterInterrupt && !activeClient.isClosed) activeClient.stop();
|
||||
});
|
||||
};
|
||||
const beginInterruptAndStop = (reason: string, triggerPhase: string): void => {
|
||||
stopAfterInterrupt = true;
|
||||
if (!interruptInFlight) interruptInFlight = requestInterrupt(reason, triggerPhase);
|
||||
else interruptInFlight = interruptInFlight.then(() => requestInterrupt(reason, triggerPhase));
|
||||
};
|
||||
const abortTurn = (): void => {
|
||||
if (terminal) return;
|
||||
terminal = { status: "cancelled", failureKind: "cancelled", message: "cancel requested" };
|
||||
emitEvent({ type: "backend_status", payload: { phase: "turn-cancelled", failureKind: "cancelled" } });
|
||||
client?.stop();
|
||||
beginInterruptAndStop("cancel requested", "abort-signal");
|
||||
terminalResolve();
|
||||
};
|
||||
options.abortSignal?.addEventListener("abort", abortTurn, { once: true });
|
||||
const turnIdleTimeoutMs = positiveTimeout(options.timeoutMs);
|
||||
let hardTimeout: NodeJS.Timeout | null = null;
|
||||
let idleTimeout: NodeJS.Timeout | null = null;
|
||||
hardTimeout = setTimeout(() => {
|
||||
if (terminal) return;
|
||||
terminal = { status: "failed", failureKind: "backend-timeout", message: `codex stdio turn hard timed out after ${turnIdleTimeoutMs}ms` };
|
||||
emitEvent({ type: "error", payload: { failureKind: terminal.failureKind, message: terminal.message, phase: "turn:hard-timeout" } });
|
||||
beginInterruptAndStop("hard timeout", "turn:hard-timeout");
|
||||
terminalResolve();
|
||||
}, turnIdleTimeoutMs);
|
||||
const refreshTurnActivity = (): void => {
|
||||
if (terminal) return;
|
||||
if (idleTimeout) clearTimeout(idleTimeout);
|
||||
@@ -440,7 +499,7 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess
|
||||
if (terminal) return;
|
||||
terminal = { status: "failed", failureKind: "backend-timeout", message: `codex stdio turn idle timed out after ${turnIdleTimeoutMs}ms without activity` };
|
||||
emitEvent({ type: "error", payload: { failureKind: terminal.failureKind, message: terminal.message, phase: "turn:idle-timeout" } });
|
||||
client?.stop();
|
||||
beginInterruptAndStop("idle timeout", "turn:idle-timeout");
|
||||
terminalResolve();
|
||||
}, turnIdleTimeoutMs);
|
||||
};
|
||||
@@ -449,12 +508,18 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess
|
||||
clearTimeout(idleTimeout);
|
||||
idleTimeout = null;
|
||||
};
|
||||
const stopTurnHardTimeout = (): void => {
|
||||
if (!hardTimeout) return;
|
||||
clearTimeout(hardTimeout);
|
||||
hardTimeout = null;
|
||||
};
|
||||
refreshTurnActivity();
|
||||
const stopNotifications = session.addNotificationHandler((message) => {
|
||||
refreshTurnActivity();
|
||||
const normalized = normalizeCodexNotification(message, suppressedNotifications);
|
||||
if (normalized.threadId) threadId = normalized.threadId;
|
||||
if (normalized.turnId) turnId = normalized.turnId;
|
||||
exposeActiveTurn(normalized.turnId ? "turn-notification" : "notification");
|
||||
emitEvents(normalized.events);
|
||||
if (normalized.assistantDelta) {
|
||||
assistantText += normalized.assistantDelta.text;
|
||||
@@ -511,30 +576,28 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess
|
||||
|
||||
const promptInjection = initialPromptInjection(options.initialPrompt, willResumeThread);
|
||||
emitEvent({ type: "backend_status", payload: { phase: "initial-prompt-assembly", initialPromptInjected: promptInjection.injected, reason: promptInjection.reason, initialPrompt: options.initialPrompt?.summary ?? { available: false, valuesPrinted: false }, valuesPrinted: false } });
|
||||
const turnResponse = requireResponseRecord(await client.request("turn/start", withOptionalModel({ threadId, input: textInputForUserMessage(options.prompt, promptInjection), cwd: options.cwd, approvalPolicy: options.approvalPolicy }, options.model), requestTimeoutMs), "turn/start");
|
||||
turnId = requireNestedId(turnResponse, "turn/start", "turn");
|
||||
emitEvent({ type: "backend_status", payload: { phase: "turn/start:completed", turnId } });
|
||||
if (threadId && turnId && options.onActiveTurn) {
|
||||
const maybeStop = options.onActiveTurn({
|
||||
threadId,
|
||||
turnId,
|
||||
steer: async (prompt: string) => {
|
||||
await client!.request("turn/steer", { threadId: threadId!, expectedTurnId: turnId!, input: textInput(prompt) }, requestTimeoutMs);
|
||||
},
|
||||
interrupt: async () => {
|
||||
await client!.request("turn/interrupt", { threadId: threadId!, turnId: turnId! }, requestTimeoutMs);
|
||||
},
|
||||
});
|
||||
if (typeof maybeStop === "function") stopActiveTurn = maybeStop;
|
||||
const turnStart = await Promise.race([
|
||||
client.request("turn/start", withOptionalModel({ threadId, input: textInputForUserMessage(options.prompt, promptInjection), cwd: options.cwd, approvalPolicy: options.approvalPolicy }, options.model), requestTimeoutMs).then((response) => ({ kind: "response" as const, response })),
|
||||
terminalPromise.then(() => ({ kind: "terminal" as const })),
|
||||
]);
|
||||
if (turnStart.kind === "response") {
|
||||
const turnResponse = requireResponseRecord(turnStart.response, "turn/start");
|
||||
turnId = requireNestedId(turnResponse, "turn/start", "turn");
|
||||
emitEvent({ type: "backend_status", payload: { phase: "turn/start:completed", turnId } });
|
||||
exposeActiveTurn("turn-start-response");
|
||||
} else {
|
||||
emitEvent({ type: "backend_status", payload: { phase: "turn/start:interrupted-before-response", threadId: threadId ?? null, turnId: turnId ?? null } });
|
||||
}
|
||||
|
||||
const race = await Promise.race([
|
||||
terminalPromise.then(() => ({ kind: "terminal" as const })),
|
||||
client.closedPromise.then((closeInfo) => ({ kind: "closed" as const, closeInfo })),
|
||||
]);
|
||||
if (race.kind === "closed" && !terminal) {
|
||||
terminal = terminalFromClose(race.closeInfo);
|
||||
emitEvent({ type: "error", payload: { failureKind: terminal.failureKind, message: terminal.message, phase: "transport:closed-before-terminal", appServerExit: closeEvent(race.closeInfo) } });
|
||||
if (!terminal) {
|
||||
const race = await Promise.race([
|
||||
terminalPromise.then(() => ({ kind: "terminal" as const })),
|
||||
client.closedPromise.then((closeInfo) => ({ kind: "closed" as const, closeInfo })),
|
||||
]);
|
||||
if (race.kind === "closed" && !terminal) {
|
||||
terminal = terminalFromClose(race.closeInfo);
|
||||
emitEvent({ type: "error", payload: { failureKind: terminal.failureKind, message: terminal.message, phase: "transport:closed-before-terminal", appServerExit: closeEvent(race.closeInfo) } });
|
||||
}
|
||||
}
|
||||
if (!terminal) terminal = { status: "failed", failureKind: "backend-response-invalid", message: "codex app-server did not emit turn/completed" };
|
||||
} catch (error) {
|
||||
@@ -548,8 +611,11 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess
|
||||
stopNotifications();
|
||||
options.abortSignal?.removeEventListener("abort", abortTurn);
|
||||
stopTurnIdleTimeout();
|
||||
stopTurnHardTimeout();
|
||||
}
|
||||
if (!terminal) terminal = { status: "failed", failureKind: "backend-response-invalid", message: "codex app-server finished without terminal status" };
|
||||
const pendingInterrupt: Promise<void> | null = interruptInFlight as Promise<void> | null;
|
||||
if (pendingInterrupt) await pendingInterrupt.catch(() => undefined);
|
||||
if (terminal.status !== "completed") emitEvents(await session.close());
|
||||
emitEvents(flushAssistantDeltaProgress(assistantDeltaProgress));
|
||||
if (completedAssistantMessages.length === 0) emitEvents(assistantMessageEventsForTurn(assistantText, terminal.status === "completed"));
|
||||
@@ -754,7 +820,7 @@ function normalizeCodexNotification(message: JsonRecord, suppressed: SuppressedN
|
||||
const status = terminalStatusFromValue(turn.status);
|
||||
const error = asRecordAt(turn, "error");
|
||||
const messageText = typeof error.message === "string" ? redactText(error.message) : null;
|
||||
const failureKind = status === "completed" ? null : classifyCodexErrorRecord(Object.keys(error).length > 0 ? error : { message: turn.status }, "backend-failed");
|
||||
const failureKind = status === "completed" ? null : status === "cancelled" ? "cancelled" : classifyCodexErrorRecord(Object.keys(error).length > 0 ? error : { message: turn.status }, "backend-failed");
|
||||
const events: BackendEvent[] = [{ type: "backend_status", payload: { phase: method, terminalStatus: status } }];
|
||||
if (failureKind) events.push({ type: "error", payload: { failureKind, error: redactJson(error), phase: method } });
|
||||
return { events, terminal: { status, failureKind, message: messageText } };
|
||||
|
||||
Reference in New Issue
Block a user