diff --git a/extensions/connector-oh-my-pi/interactive-loop.smoke.ts b/extensions/connector-oh-my-pi/interactive-loop.smoke.ts index 33248812..93255044 100644 --- a/extensions/connector-oh-my-pi/interactive-loop.smoke.ts +++ b/extensions/connector-oh-my-pi/interactive-loop.smoke.ts @@ -88,27 +88,59 @@ class FakeMesh implements PeerMesh { interface SentCall { message: { customType: string; content: string; display: boolean; details: unknown; attribution: "user" | "agent" }; - options: { deliverAs: "steer"; triggerTurn: true }; + options: { deliverAs: "steer" | "nextTurn"; triggerTurn: true }; } -/** A fake host that records every sendMessage(message, options). */ +/** A fake host that records every sendMessage AND models OMP's delivery routing, so a test can + * assert the user-visible landing zone (not just the envelope). Mirrors the real routing in + * agent-session.ts `sendCustomMessage`: the deliverAs branch at 7458-7479 sends `nextTurn` to the + * hidden `#queueHiddenNextTurnMessage` queue (never `this.agent.steer()`), while a `steer` reaching + * a session that is still tearing down a user-interrupted (ESC) turn (`interruptUnwinding` here + * models isStreaming still true + #advisorAutoResumeSuppressed latched at 7766-7768) folds via + * this.agent.steer() into the editable pending-message UI = the composer. When idle, either mode + * wakes a fresh turn via #promptAgentInitiatedMessage (7472-7478), so the happy path is identical. + * Keep this fake in sync with agent-session.ts:7458-7479 if OMP's routing changes. */ class FakeHost implements PeerHost { readonly sent: SentCall[] = []; + /** Set true to model OMP still unwinding a user-interrupted (ESC) turn — the bug window. */ + interruptUnwinding = false; + readonly composer: string[] = []; // editable pending UI — a bleed lands here (the defect) + readonly held: string[] = []; // hidden next-turn queue — parked, awaiting the deferred continuation + readonly turns: string[] = []; // content a real turn actually consumed (idle wake, or a held flush) sendMessage(message: SentCall["message"], options: SentCall["options"]): void { this.sent.push({ message, options }); + if (this.interruptUnwinding) { + // Mid-unwind: nextTurn parks in the hidden queue (#queueHiddenNextTurnMessage, 7459); a + // steer would instead fold via this.agent.steer() into the editable composer (7466). + if (options.deliverAs === "nextTurn") this.held.push(message.content); + else this.composer.push(message.content); + } else { + // Idle: either mode wakes a fresh turn via #promptAgentInitiatedMessage (7472-7478) that + // consumes the message immediately. + this.turns.push(message.content); + } + } + /** Model OMP draining the hidden next-turn queue into a real turn — the deferred + * #promptQueuedHiddenNextTurnMessages continuation (7323-7346) that the post-prompt task + * (7298-7321) runs once the interrupted prompt has settled. Parked content becomes consumed + * turn content, exactly as a clean redelivery would. Returns what it flushed. */ + flushHeld(): string[] { + const flushed = this.held.splice(0, this.held.length); + this.turns.push(...flushed); + return flushed; } get last(): SentCall { return this.sent[this.sent.length - 1]; } } -/** Assert a sendMessage call carries the fixed steer/turn envelope the loop always uses. */ +/** Assert a sendMessage call carries the fixed nextTurn/turn envelope the loop always uses. */ function assertEnvelope(call: SentCall, customType: string, ctx: string): void { assert(call.message.customType === customType, `${ctx}: customType === ${customType}`); assert(call.message.display === true, `${ctx}: display true`); assert(call.message.attribution === "user", `${ctx}: attribution "user"`); assert(JSON.stringify(call.message.details) === "{}", `${ctx}: details {}`); - assert(call.options.deliverAs === "steer", `${ctx}: deliverAs "steer"`); + assert(call.options.deliverAs === "nextTurn", `${ctx}: deliverAs "nextTurn"`); assert(call.options.triggerTurn === true, `${ctx}: triggerTurn true`); } @@ -325,5 +357,53 @@ function assertEnvelope(call: SentCall, customType: string, ctx: string): void { console.log("8) shutdown stops mesh OK ✅"); } +// ---- 9. ESC-interrupt: a queued message is HELD for redelivery, never bled into the composer ---- +// Repro for the reported defect: hitting ESC to interrupt a running turn while a cotal message is +// waiting must not land the message text in the editable composer. The connector cannot observe the +// interrupt (agent_end/ExtensionContext carry no abort reason), so it must deliver in a mode that is +// hidden-from-composer under OMP's own contract. `nextTurn` is that mode: idle → a fresh turn wakes +// as before; still-unwinding after an ESC → parked in the hidden next-turn queue, redelivered clean. +// A `steer` (the pre-fix envelope) bleeds into the composer in that window — this asserts it doesn't. +{ + const mesh = new FakeMesh(); + const host = new FakeHost(); + const loop = runPeerLoop({ mesh, host }); + + // The turn the user is about to ESC out of. + loop.onAgentStart(); + // A directed DM arrives while that turn is live — buffered by the no-interrupt gate, not delivered. + const dm = item({ id: "esc1", kind: "dm", text: "peer ping during a turn" }); + mesh.inbox = [dm]; + mesh.emit("incoming", dm); + assert(host.sent.length === 0, "9) message arriving mid-turn is buffered, not delivered"); + + // User hits ESC: OMP aborts with USER_INTERRUPT and is still tearing the turn down when the + // connector's turn-end hook fires and flushes the buffered message (pendingWake mirrors the real + // MeshAgent reporting the mid-turn arrival as a pending wake, exactly as test 5 drives it). + mesh.setPendingWake(1); + host.interruptUnwinding = true; + loop.onAgentEnd(); + + assert(host.sent.length === 1, "9) the buffered message is delivered at turn end"); + assert(host.composer.length === 0, `9) message must NOT bleed into the composer (got ${JSON.stringify(host.composer)})`); + assert(host.held.length === 1 && host.held[0].includes("peer ping during a turn"), "9) message is HELD for clean redelivery"); + // It stays unacked (still leads the inbox) so it redelivers as a normal peer message next turn. + assert(mesh.peekInbox().some((i) => i.id === "esc1"), "9) held message stays on the inbox for redelivery"); + // The other half of the acceptance ("held + redelivered next turn"): prove held-AND-DELIVERED, + // not just held-AND-acked. OMP's deferred continuation (#promptQueuedHiddenNextTurnMessages, + // 7323-7346) drains the hidden queue into a real turn once the interrupted prompt settles — model + // that flush and assert the parked content actually reached a turn before we let the ack stand. + host.interruptUnwinding = false; + const flushed = host.flushHeld(); + assert(flushed.length === 1 && flushed[0].includes("peer ping during a turn"), "9) held message is flushed into a real turn (delivered, not dropped)"); + assert(host.turns.some((t) => t.includes("peer ping during a turn")), "9) the redelivered message was consumed by a turn"); + // Only now, when that clean turn ends, does the connector ack — draining the peer inbox. This is + // the ack-on-turn-end path (interactive-loop.ts ackSurfaced) that nextTurn preserves. + mesh.setPendingWake(0); + loop.onAgentEnd(); + assert(mesh.peekInbox().length === 0, "9) the message is acked (drained) only after it was delivered into a turn"); + console.log("9) ESC-interrupt holds the message, no composer bleed OK ✅"); +} + console.log("\nCOTAL-MESH LOOP SMOKE OK ✅"); process.exit(0); diff --git a/extensions/connector-oh-my-pi/src/interactive-loop.ts b/extensions/connector-oh-my-pi/src/interactive-loop.ts index 4ce86f69..3fffdd19 100644 --- a/extensions/connector-oh-my-pi/src/interactive-loop.ts +++ b/extensions/connector-oh-my-pi/src/interactive-loop.ts @@ -26,11 +26,15 @@ export interface PeerMesh { on(event: "wake", handler: () => void): void; } -/** The host-session surface the loop drives. The extension's `ExtensionAPI` satisfies this. */ +/** The host-session surface the loop drives. The extension's `ExtensionAPI` satisfies this. + * Delivery uses `deliverAs: "nextTurn"`, the one mode OMP's contract keeps hidden from the + * editable pending-message UI: when idle it wakes a fresh turn (same #promptAgentInitiatedMessage + * path as a steer), and when the session is still tearing down a user-interrupted (ESC) turn it is + * parked in the hidden next-turn queue and redelivered — never bled into the composer. */ export interface PeerHost { sendMessage( message: { customType: string; content: string; display: boolean; details: unknown; attribution: "user" | "agent" }, - options: { deliverAs: "steer"; triggerTurn: true }, + options: { deliverAs: "nextTurn"; triggerTurn: true }, ): void; } @@ -82,11 +86,16 @@ export function runPeerLoop({ mesh, host }: { mesh: PeerMesh; host: PeerHost }): } busy = true; surfaced = ids; - // The content participates in LLM context (a CustomMessage); triggerTurn wakes an idle session, - // steer folds into a live one. Attribution "user" — a peer message is external input here. + // The content participates in LLM context (a CustomMessage); triggerTurn wakes an idle session + // into a fresh turn. `nextTurn` (not `steer`) is deliberate: the loop only ever delivers when it + // believes the session idle (drive() early-returns while busy), so it never needs steer's mid- + // turn fold — and steer's one distinguishing behavior is that, arriving while OMP is still + // unwinding a user-interrupted (ESC) turn, it surfaces into the editable composer. `nextTurn` is + // hidden-from-composer by contract: idle → the same fresh-turn path, mid-unwind → parked + + // redelivered. Attribution "user" — a peer message is external input here. host.sendMessage( { customType: override ? NUDGE : INCOMING, content: text, display: true, details: {}, attribution: "user" }, - { deliverAs: "steer", triggerTurn: true }, + { deliverAs: "nextTurn", triggerTurn: true }, ); } @@ -127,6 +136,19 @@ export function runPeerLoop({ mesh, host }: { mesh: PeerMesh; host: PeerHost }): }, onAgentEnd(): void { // turn-end: release the no-interrupt gate, ack the surfaced batch, flush the next. + // + // Why ackSurfaced() can't ack a still-unconsumed message during an ESC interrupt: we + // deliver with deliverAs "nextTurn", so on an interrupt the batch is parked in OMP's hidden + // next-turn queue and consumed by a deferred continuation — NOT the interrupted turn. That + // looks like it could race (surfaced is armed in drive() before the continuation runs), but + // OMP coalesces the wire-level agent_end this handler fires on: #emitSessionEvent + // (agent-session.ts:2787-2799) HOLDS agent_end while #promptInFlightCount > 0 and lets a + // later agent_end supersede the pending one, so a wire-level subscriber sees ONE agent_end + // at the true settle. The interrupted turn + the nextTurn continuation therefore collapse + // into a single agent_end that fires AFTER the continuation consumed the batch — so this + // ack runs post-consume, never on a stray interrupted-turn event. Backstop even if that + // invariant ever broke: ackSurfaced drains only ids still at the inbox front, so an + // unconsumed survivor is left unacked and redelivers (fails safe — redelivery, not loss). busy = false; ackSurfaced(); if (mesh.pendingWake() > 0) drive();