diff --git a/packages/server/src/server/agent/providers/omp/agent.test.ts b/packages/server/src/server/agent/providers/omp/agent.test.ts index e591b366c0..8539e38336 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/packages/server/src/server/agent/providers/omp/agent.test.ts @@ -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([ + "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({ diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index 5f85b7c7d0..13f36b5b7a 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -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 { 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; diff --git a/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts b/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts index 0b17fe8c5b..68eed2fc3b 100644 --- a/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts +++ b/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts @@ -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 { @@ -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((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); }