diff --git a/packages/compass-agent/src/compassv1.ts b/packages/compass-agent/src/compassv1.ts index ba3cbe98..16f38e32 100644 --- a/packages/compass-agent/src/compassv1.ts +++ b/packages/compass-agent/src/compassv1.ts @@ -282,6 +282,9 @@ export { // message schemas are the oneof payloads it constructs with `create`. type SessionAssistantText, SessionAssistantTextSchema, + type SessionError, + SessionErrorKind, + SessionErrorSchema, type SessionEvent, SessionEventSchema, type SessionFileDiff, diff --git a/packages/compass-agent/src/mapping.test.ts b/packages/compass-agent/src/mapping.test.ts index 889ad181..74279c79 100644 --- a/packages/compass-agent/src/mapping.test.ts +++ b/packages/compass-agent/src/mapping.test.ts @@ -15,6 +15,7 @@ import { AgentPlanEntryStatus, AgentSessionState, AgentToolCallStatus, + SessionErrorKind, type SessionEvent, } from "./compassv1"; import { EventMapper, type MapOutput } from "./mapping"; @@ -355,29 +356,97 @@ describe("EventMapper — non-content inner events emit nothing", () => { } }); -describe("EventMapper — inner error: crash → session ERRORED vs abort → counted unmapped", () => { - // The pi-ai `error` variant carries reason "error" | "aborted" — NOT the same - // failure class: - // - "error" = an inner/provider failure → a session frame to ERRORED. - // - "aborted" = a deliberate cancel, NOT a crash → surfaced as a counted - // UnmappedEvent (never conflated with ERRORED, never dropped). - test('reason "error" → one session frame in state ERRORED', () => { - expect( - soleSessionState( - mapper().map(upd({ type: "error", reason: "error", error: partial() })), - ), - ).toBe(AgentSessionState.ERRORED); +describe("EventMapper — inner error: error → SessionError + ERRORED vs abort → SessionError only", () => { + // The pi-ai `error` variant carries reason "error" | "aborted". Both surface + // their content as a SessionError trace event; the failure class differs: + // - "error" = an inner/provider failure → the SessionError(ERROR) content + // frame AND the ERRORED lifecycle transition (content first). + // - "aborted" = a deliberate cancel, NOT a crash → the SessionError(ABORTED) + // content frame only, no lifecycle transition, never a counted unmapped. + // `status` rides the frame only when the provider surfaced an HTTP status. + const errored = (over: Partial): AssistantMessage => ({ + ...partial(), + ...over, }); - test('reason "aborted" → one counted UnmappedEvent, NOT ERRORED', () => { + test('reason "error" → SessionError(ERROR) content frame then ERRORED lifecycle', () => { const out = mapper().map( - upd({ type: "error", reason: "aborted", error: partial() }), + upd({ + type: "error", + reason: "error", + error: errored({ errorMessage: "provider exploded", errorStatus: 500 }), + }), ); - expect(out).toHaveLength(1); - const frame = out[0]; - expect(frame.kind).toBe("unmapped"); - if (frame.kind === "unmapped") - expect(frame.eventType).toBe("message_update:error"); + expect(out).toHaveLength(2); + const [content, lifecycle] = out; + if (content.kind !== "session") + throw new Error(`expected session frame, got ${content.kind}`); + const ev = defined(content.value.typedEvent, "typed session event").event; + if (ev.case !== "sessionError") + throw new Error(`expected sessionError, got ${ev.case}`); + expect(ev.value.kind).toBe(SessionErrorKind.ERROR); + expect(ev.value.message).toBe("provider exploded"); + expect(ev.value.status).toBe(500); + if (lifecycle.kind !== "session") + throw new Error(`expected session frame, got ${lifecycle.kind}`); + expect(lifecycle.value.typedEvent).toBeUndefined(); + expect(lifecycle.value.state).toBe(AgentSessionState.ERRORED); + }); + + test('reason "aborted" → one SessionError(ABORTED), no lifecycle, no unmapped', () => { + const out = mapper().map( + upd({ + type: "error", + reason: "aborted", + error: errored({ errorMessage: "user cancelled" }), + }), + ); + const ev = soleTyped(out).event; + if (ev.case !== "sessionError") + throw new Error(`expected sessionError, got ${ev.case}`); + expect(ev.value.kind).toBe(SessionErrorKind.ABORTED); + expect(ev.value.message).toBe("user cancelled"); + }); + + test("a surfaced errorStatus rides the frame; an absent one leaves status unset", () => { + const withStatus = soleTyped( + mapper().map( + upd({ + type: "error", + reason: "aborted", + error: errored({ errorStatus: 429 }), + }), + ), + ).event; + if (withStatus.case !== "sessionError") + throw new Error(`expected sessionError, got ${withStatus.case}`); + expect(withStatus.value.status).toBe(429); + + // The literal-0 boundary is the whole reason #sessionError guards on + // `errorStatus !== undefined` rather than truthiness: HTTP status 0 is a + // real provider sentinel (network-level failure / no response), and a + // truthiness guard would silently drop it. Status 0 must ride the frame. + const zeroStatus = soleTyped( + mapper().map( + upd({ + type: "error", + reason: "aborted", + error: errored({ errorStatus: 0 }), + }), + ), + ).event; + if (zeroStatus.case !== "sessionError") + throw new Error(`expected sessionError, got ${zeroStatus.case}`); + expect(zeroStatus.value.status).toBe(0); + + const noStatus = soleTyped( + mapper().map(upd({ type: "error", reason: "aborted", error: partial() })), + ).event; + if (noStatus.case !== "sessionError") + throw new Error(`expected sessionError, got ${noStatus.case}`); + expect(noStatus.value.status).toBeUndefined(); + // An absent errorMessage collapses to the empty string, never undefined. + expect(noStatus.value.message).toBe(""); }); }); diff --git a/packages/compass-agent/src/mapping.ts b/packages/compass-agent/src/mapping.ts index 67a01f44..57a0aa1b 100644 --- a/packages/compass-agent/src/mapping.ts +++ b/packages/compass-agent/src/mapping.ts @@ -33,7 +33,7 @@ // a monotonic per-event counter. `atUnixMs` comes from an injectable clock so // tests are deterministic. -import type { AssistantMessageEvent } from "@oh-my-pi/pi-ai"; +import type { AssistantMessage, AssistantMessageEvent } from "@oh-my-pi/pi-ai"; import type { AgentSessionEvent } from "@oh-my-pi/pi-coding-agent"; import { type AgentPlanEntry, @@ -43,6 +43,8 @@ import { AgentToolCallStatus, create, SessionAssistantTextSchema, + SessionErrorKind, + SessionErrorSchema, type SessionEvent, SessionEventSchema, type SessionFileDiff, @@ -246,6 +248,29 @@ export class EventMapper { return { kind: "session", value }; } + // Build one SessionError trace frame from a pi-ai inner-error AssistantMessage + // (mapping.ts error arm). `kind` discriminates an unexpected failure (ERROR, + // paired with the ERRORED lifecycle transition) from a deliberate abort + // (ABORTED, no transition). `message` is the failure text (empty when the SDK + // surfaced none); `status` is set only when the provider surfaced an HTTP + // status, so a subscriber can tell "no status" from a literal 0. Routes + // through `#sessionEvent` for uniform id/clock stamping. + #sessionError( + kind: SessionErrorKind, + error: AssistantMessage, + ): OutboundFrame { + return this.#sessionEvent({ + case: "sessionError", + value: create(SessionErrorSchema, { + kind, + message: error.errorMessage ?? "", + ...(error.errorStatus !== undefined + ? { status: error.errorStatus } + : {}), + }), + }); + } + // Build one SessionInjection trace frame — the agent-side observation that a // channel message was injected into the live session as a steer or a deliver // (design "steer/deliver split observation seam", T1). Public because the @@ -296,27 +321,26 @@ export class EventMapper { } case "error": { // The stream surfaced an inner error. `reason` splits the failure - // class (SDK: "aborted" | "error"): - // - "error" = an unexpected inner/provider failure → an ERRORED - // lifecycle transition (compass.proto:132 scopes ERRORED to the + // class (SDK: "aborted" | "error"), and both surface their content as + // a SessionError trace frame (DL-322): `inner.error.errorMessage` is + // the user-facing failure text, `inner.error.errorStatus` the provider + // HTTP status when one was surfaced. + // - "error" = an unexpected inner/provider failure → emit the + // SessionError(ERROR) content frame AND preserve the ERRORED + // lifecycle transition (compass.proto scopes ERRORED to the // OOM/panic/engine-restart class; an inner stream error is that - // class). + // class, and board/presence/delivery key off it). Content first. // - "aborted" = a deliberate steer/user cancel, NOT a crash — // conflating it with ERRORED would misreport a normal abort as an - // engine failure. Surface it as a counted UnmappedEvent so it is - // logged + counted, never dropped, pending the abort-surfacing - // contract (a follow-up). + // engine failure. Emit only the SessionError(ABORTED) content + // frame, with NO lifecycle transition. if (inner.reason === "error") { - return [this.#sessionState(AgentSessionState.ERRORED)]; + return [ + this.#sessionError(SessionErrorKind.ERROR, inner.error), + this.#sessionState(AgentSessionState.ERRORED), + ]; } - return [ - { - kind: "unmapped", - eventType: "message_update:error", - reason: - "agent abort (reason=aborted) — not a crash; abort-surfacing staged", - }, - ]; + return [this.#sessionError(SessionErrorKind.ABORTED, inner.error)]; } default: // start/text_start/thinking_start/thinking_end/image_end/toolcall_*/ diff --git a/packages/compass-agent/src/transport/frame-sink.test.ts b/packages/compass-agent/src/transport/frame-sink.test.ts index 5174e1c7..befaac93 100644 --- a/packages/compass-agent/src/transport/frame-sink.test.ts +++ b/packages/compass-agent/src/transport/frame-sink.test.ts @@ -26,6 +26,8 @@ import { AgentSessionState, DeliveryAckSchema, ForgeNotificationAckSchema, + SessionErrorKind, + SessionErrorSchema, SessionEventSchema, SessionFrameSchema, SessionInjectionKind, @@ -735,3 +737,46 @@ test("a SessionInjection rides the Publish PRIORITY sub-lane, never the drop-old // It never touched the loss-tolerable trace lane. expect(traceFrames.length).toBe(0); }); + +test("a SessionError rides the Publish PRIORITY sub-lane, never the drop-oldest trace queue", () => { + // DL-323: a SessionError is a "session" trace frame (state UNSPECIFIED), so by + // the default classification it would ride the bounded, drop-oldest trace lane + // — where a busy trace stream could silently drop the surfaced failure content + // (for a reason=aborted failure, the SOLE signal). isSessionError() pins it + // onto the never-drop priority lane instead. The socket recorder cannot + // distinguish the two Publish sub-lanes, so this spy-spine test is what pins + // the choice. Non-vacuity: drop the `|| isSessionError(frame)` arm in emit() + // (frame-sink.ts) → the priority assertion reddens (0) and the trace assertion + // reddens (2). Both SessionErrorKinds ride the same lane. + for (const kind of [SessionErrorKind.ERROR, SessionErrorKind.ABORTED]) { + const { spine, priorityFrames, traceFrames } = spySpine(); + const sink = createSocketFrameSink(spineTransport(spine)); + sink.emit({ + kind: "session", + value: create(SessionFrameSchema, { + state: AgentSessionState.UNSPECIFIED, + typedEvent: create(SessionEventSchema, { + event: { + case: "sessionError", + value: create(SessionErrorSchema, { + kind, + message: "boom", + }), + }, + }), + }), + }); + // Exactly one priority frame, carrying the sessionError oneof case + kind. + expect(priorityFrames.length).toBe(1); + const inner = priorityFrames[0]?.frame?.frame; + expect(inner?.case).toBe("session"); + const event = + inner?.case === "session" ? inner.value.typedEvent?.event : undefined; + expect(event?.case).toBe("sessionError"); + expect(event?.case === "sessionError" ? event.value.kind : undefined).toBe( + kind, + ); + // It never touched the loss-tolerable trace lane. + expect(traceFrames.length).toBe(0); + } +}); diff --git a/packages/compass-agent/src/transport/frame-sink.ts b/packages/compass-agent/src/transport/frame-sink.ts index eba0ce02..edfc95e4 100644 --- a/packages/compass-agent/src/transport/frame-sink.ts +++ b/packages/compass-agent/src/transport/frame-sink.ts @@ -109,6 +109,20 @@ function isInjection(frame: OutboundFrame): boolean { ); } +// A "session" frame carrying a SessionError trace event is NOT loss-tolerable +// (DL-323): the surfaced failure content is observation-critical — for a +// reason=aborted failure it is the SOLE signal (no lifecycle frame accompanies +// it), and for reason=error the paired ERRORED transition survives on the +// priority lane but the content itself would still vanish off the bounded trace +// queue under backlog. So it rides the never-drop priority lane, matching the +// SessionInjection carve-out, even though its board state is UNSPECIFIED. +function isSessionError(frame: OutboundFrame): boolean { + return ( + frame.kind === "session" && + frame.value.typedEvent?.event.case === "sessionError" + ); +} + export function createSocketFrameSink(transport: RunnerTransport): FrameSink { const spine = transport.publishSpine(); // Borrow the single transport-owned ManagedRuntime through the module-private @@ -266,7 +280,7 @@ export function createSocketFrameSink(transport: RunnerTransport): FrameSink { const request = create(PublishFrameRequestSchema, { frame: toAgentFrame(frame), }); - if (isLifecycle(frame) || isInjection(frame)) { + if (isLifecycle(frame) || isInjection(frame) || isSessionError(frame)) { spine.enqueuePriority(request); } else { spine.enqueueTrace(request);