mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-01 14:12:51 +00:00
fix(omp): show a stopped OMP turn as canceled, not failed (#5243)
Pressing Stop on an OMP turn could land in the thread as a failed turn with OMP's interrupt text as the error, because OMP ends the turn with a terminal response that carries "Interrupted by user" in errorMessage and "aborted" in stopReason, and completeTurn classified any errorMessage as a failure. A terminal response whose stop reason is aborted now ends the turn as turn_canceled and terminalizes the turn's in-flight tool calls and subagents, matching what interrupt() already emits when it wins the race with the provider's teardown. A provider-idle state check that resolves after the turn already settled no longer completes that turn, so a stop can no longer emit a second lifecycle event behind the cancellation. Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>
This commit is contained in:
@@ -1,12 +1,37 @@
|
||||
import { describe, expect, test } from "vitest";
|
||||
import { setImmediate as waitForImmediate } from "node:timers/promises";
|
||||
|
||||
import type { AgentStreamEvent } from "../../agent-sdk-types.js";
|
||||
import type { PaseoToolCatalog } from "../../tools/types.js";
|
||||
import type { OmpAgentMessage } from "./rpc-types.js";
|
||||
import type { OmpNoTurnScheduler, OmpProviderIdleScheduler } from "./agent.js";
|
||||
import type { OmpUsagePollScheduler } from "./usage-poller.js";
|
||||
import { resolveOmpProviderParams } from "./provider-config.js";
|
||||
import { OmpHarness } from "./test-utils/omp-harness.js";
|
||||
|
||||
const TURN_LIFECYCLE_EVENTS = new Set<AgentStreamEvent["type"]>([
|
||||
"turn_started",
|
||||
"turn_completed",
|
||||
"turn_failed",
|
||||
"turn_canceled",
|
||||
]);
|
||||
|
||||
function isTurnLifecycle(type: AgentStreamEvent["type"]): boolean {
|
||||
return TURN_LIFECYCLE_EVENTS.has(type);
|
||||
}
|
||||
|
||||
// What OMP reports for a turn the user stopped: an error message on a terminal
|
||||
// response whose stop reason says the request was aborted.
|
||||
const ABORTED_TERMINAL_RESPONSE: OmpAgentMessage = {
|
||||
role: "assistant",
|
||||
content: [],
|
||||
provider: "ai-harness-omp",
|
||||
model: "glm-5.3-flash-high",
|
||||
responseId: "chatcmpl-aborted",
|
||||
stopReason: "aborted",
|
||||
errorMessage: "Interrupted by user",
|
||||
};
|
||||
|
||||
test("OMP ready timeout defaults to 20 seconds and RPC timeout overrides both", () => {
|
||||
expect(resolveOmpProviderParams({}).runtimeProviderParams).toMatchObject({
|
||||
readyTimeoutMs: 20_000,
|
||||
@@ -624,6 +649,74 @@ describe("OMP agent client and session", () => {
|
||||
expect(omp.runningToolCallIds()).toEqual([]);
|
||||
});
|
||||
|
||||
test("an interrupt that OMP reports as an aborted turn cancels instead of failing", async () => {
|
||||
const omp = new OmpHarness();
|
||||
await omp.start();
|
||||
|
||||
await omp.requireStartTurn("do something long");
|
||||
const runtime = omp.runtime();
|
||||
runtime.beginTurn();
|
||||
runtime.emit({
|
||||
type: "tool_execution_start",
|
||||
toolCallId: "tool-1",
|
||||
toolName: "bash",
|
||||
args: { command: "sleep 30" },
|
||||
});
|
||||
runtime.streamAssistantText("working on it");
|
||||
// OMP ends the turn with an aborted terminal response before answering the abort.
|
||||
runtime.onAbort = () => {
|
||||
runtime.emit({ type: "message_end", message: ABORTED_TERMINAL_RESPONSE });
|
||||
runtime.finishTurn(ABORTED_TERMINAL_RESPONSE);
|
||||
};
|
||||
|
||||
await omp.interrupt();
|
||||
await waitForImmediate();
|
||||
await waitForImmediate();
|
||||
|
||||
expect(omp.eventTypes().filter(isTurnLifecycle)).toEqual(["turn_started", "turn_canceled"]);
|
||||
expect(omp.runningToolCallIds()).toEqual([]);
|
||||
});
|
||||
|
||||
test("an aborted turn that settles after the interrupt cancels exactly once", async () => {
|
||||
const omp = new OmpHarness();
|
||||
await omp.start();
|
||||
|
||||
await omp.requireStartTurn("do something long");
|
||||
const runtime = omp.runtime();
|
||||
runtime.beginTurn();
|
||||
runtime.streamAssistantText("working on it");
|
||||
// The provider-idle check is still in flight when the abort is acknowledged.
|
||||
const releaseState = runtime.holdStateRequests();
|
||||
runtime.emit({ type: "message_end", message: ABORTED_TERMINAL_RESPONSE });
|
||||
runtime.finishTurn(ABORTED_TERMINAL_RESPONSE);
|
||||
|
||||
await omp.interrupt();
|
||||
releaseState();
|
||||
await waitForImmediate();
|
||||
await waitForImmediate();
|
||||
|
||||
expect(omp.eventTypes().filter(isTurnLifecycle)).toEqual(["turn_started", "turn_canceled"]);
|
||||
});
|
||||
|
||||
test("an autonomous OMP turn aborted with no client turn id cancels", async () => {
|
||||
const omp = new OmpHarness();
|
||||
await omp.start();
|
||||
|
||||
const runtime = omp.runtime();
|
||||
runtime.beginTurn();
|
||||
runtime.streamAssistantText("autonomous work");
|
||||
runtime.onAbort = () => {
|
||||
runtime.emit({ type: "message_end", message: ABORTED_TERMINAL_RESPONSE });
|
||||
runtime.finishTurn(ABORTED_TERMINAL_RESPONSE);
|
||||
};
|
||||
|
||||
await omp.interrupt();
|
||||
await waitForImmediate();
|
||||
await waitForImmediate();
|
||||
|
||||
expect(omp.eventTypes().filter(isTurnLifecycle)).toEqual(["turn_started", "turn_canceled"]);
|
||||
});
|
||||
|
||||
test("a resumed session does not re-emit replayed events as live timeline items", async () => {
|
||||
const omp = new OmpHarness();
|
||||
await omp.resume({
|
||||
|
||||
@@ -526,6 +526,11 @@ function latestOmpErrorMessage(messages: OmpAgentMessage[]): string | null {
|
||||
return formatOmpErrorMessage(latestAssistant);
|
||||
}
|
||||
|
||||
function isOmpAbortedTerminalResponse(messages: OmpAgentMessage[]): boolean {
|
||||
const latestAssistant = messages.findLast((message) => message.role === "assistant");
|
||||
return latestAssistant?.stopReason?.toLowerCase() === "aborted";
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
@@ -2133,6 +2138,19 @@ export class OmpAgentSession implements AgentSession {
|
||||
this.activeTurnStarted = false;
|
||||
this.activeTurnHasUserMessage = false;
|
||||
this.clearNoTurnBuffers();
|
||||
// OMP reports a stopped turn as a terminal response carrying its interrupt
|
||||
// text as an error. That is the user's own Stop, not a failed turn.
|
||||
if (isOmpAbortedTerminalResponse(messages)) {
|
||||
this.usagePoller.stopTurn();
|
||||
this.terminalizeActiveWork();
|
||||
this.emit({
|
||||
type: "turn_canceled",
|
||||
provider: this.provider,
|
||||
turnId,
|
||||
reason: "interrupted",
|
||||
});
|
||||
return;
|
||||
}
|
||||
const errorMessage = latestOmpErrorMessage(messages);
|
||||
if (typeof errorMessage === "string" && errorMessage.length > 0) {
|
||||
this.usagePoller.stopTurn();
|
||||
@@ -2161,6 +2179,10 @@ export class OmpAgentSession implements AgentSession {
|
||||
try {
|
||||
const state = await this.runtimeSession.getState();
|
||||
this.state = state;
|
||||
if (this.closed || !this.activeTurnStarted || this.currentTurnIdForEvent() !== turnId) {
|
||||
// An interrupt settled this turn while the state check was in flight.
|
||||
return;
|
||||
}
|
||||
if (!state.isStreaming && !state.isCompacting) {
|
||||
this.completeTurn(turnId, messages);
|
||||
return;
|
||||
|
||||
@@ -113,6 +113,8 @@ export class FakeOmpSession implements OmpRuntimeSession {
|
||||
getStateRequestCount = 0;
|
||||
abortRequested = false;
|
||||
abortError: Error | null = null;
|
||||
onAbort: (() => void) | null = null;
|
||||
private heldStateRequests: Array<() => void> | null = null;
|
||||
readonly canceledExtensionUiRequests: string[] = [];
|
||||
readonly extensionUiResponses: Array<{
|
||||
id: string;
|
||||
@@ -243,6 +245,8 @@ export class FakeOmpSession implements OmpRuntimeSession {
|
||||
throw this.abortError;
|
||||
}
|
||||
this.abortRequested = true;
|
||||
// OMP can stream a turn's teardown before it answers the abort request.
|
||||
this.onAbort?.();
|
||||
}
|
||||
|
||||
async getState(): Promise<OmpSessionState> {
|
||||
@@ -251,6 +255,10 @@ export class FakeOmpSession implements OmpRuntimeSession {
|
||||
if (this.getStateRequestCount >= waiter.count) waiter.resolve();
|
||||
else this.stateRequestWaiters.push(waiter);
|
||||
}
|
||||
if (this.heldStateRequests) {
|
||||
const held = this.heldStateRequests;
|
||||
await new Promise<void>((resolve) => held.push(resolve));
|
||||
}
|
||||
if (this.getStateError) {
|
||||
throw this.getStateError;
|
||||
}
|
||||
@@ -261,6 +269,16 @@ export class FakeOmpSession implements OmpRuntimeSession {
|
||||
return this.state;
|
||||
}
|
||||
|
||||
/** Holds every state request until the returned function releases them. */
|
||||
holdStateRequests(): () => void {
|
||||
const held: Array<() => void> = [];
|
||||
this.heldStateRequests = held;
|
||||
return () => {
|
||||
this.heldStateRequests = null;
|
||||
for (const resolve of held.splice(0)) resolve();
|
||||
};
|
||||
}
|
||||
|
||||
queueStateReports(states: OmpSessionState[]): void {
|
||||
this.stateReports.push(...states);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user