From 905a6a471fcb22e99319c2e7e292b152d7b759de Mon Sep 17 00:00:00 2001 From: Matt Wilkinson Date: Wed, 8 Jul 2026 21:19:29 -0400 Subject: [PATCH 1/3] fix(connector): deliver peer messages as nextTurn, not steer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hitting ESC to interrupt a running OMP turn while a cotal message was waiting bled the message text into the editable composer instead of holding it for redelivery. The connector delivered every inbound mesh message with deliverAs:"steer". When idle that wakes a fresh turn, but a steer arriving while OMP is still tearing down a user-interrupted (ESC) turn folds via agent.steer() into the editable pending- message UI = the composer (agent-session.ts:7466; the ESC path latches #advisorAutoResumeSuppressed at 7766-7768 and isStreaming stays true through the unwind). The connector cannot observe the interrupt to guard against it — AgentEndEvent carries no reason and ExtensionContext exposes no aborting signal — so it must deliver in a mode that is hidden-from-composer by contract. deliverAs:"nextTurn" is that mode (agent-session.ts:7458-7479): idle+triggerTurn routes to the same #promptAgentInitiatedMessage fresh turn as steer, while the still-unwinding case is parked in the hidden #queueHiddenNextTurnMessage queue and redelivered on the next clean turn — never the composer. The connector never intends a mid-turn fold (drive() early-returns while busy), so steer's only distinguishing behavior was exactly the misroute frame; nextTurn loses zero intended behavior and preserves ack-on-turn-end. Regression: interactive-loop.smoke.ts test 9 models OMP's routing in FakeHost (steer-during-interrupt -> composer; nextTurn -> hidden held queue) and asserts a DM queued mid-turn lands HELD not composer, then is delivered + acked on the next clean turn-end. Fails on the pre-fix steer envelope, passes after. Co-Authored-By: seal --- .../interactive-loop.smoke.ts | 72 +++++++++++++++++-- .../src/interactive-loop.ts | 19 +++-- 2 files changed, 82 insertions(+), 9 deletions(-) diff --git a/extensions/connector-oh-my-pi/interactive-loop.smoke.ts b/extensions/connector-oh-my-pi/interactive-loop.smoke.ts index 33248812..de6c188c 100644 --- a/extensions/connector-oh-my-pi/interactive-loop.smoke.ts +++ b/extensions/connector-oh-my-pi/interactive-loop.smoke.ts @@ -88,27 +88,46 @@ 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 — redelivered cleanly on the next turn + readonly turns: string[] = []; // a fresh turn woke on this message (the idle happy path) sendMessage(message: SentCall["message"], options: SentCall["options"]): void { this.sent.push({ message, options }); + if (this.interruptUnwinding) { + if (options.deliverAs === "nextTurn") this.held.push(message.content); + else this.composer.push(message.content); + } else { + this.turns.push(message.content); + } } 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 +344,50 @@ 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"): the nextTurn delivery already + // woke a fresh turn (drive() sent it and armed the surfaced batch). When THAT turn ends cleanly — + // no interrupt this time — the held message must ack normally, proving it was held-AND-delivered, + // not held-AND-dropped. This is the ack-on-turn-end path (interactive-loop.ts ackSurfaced), which + // nextTurn preserves (the message rode a real #promptAgentInitiatedMessage turn). + host.interruptUnwinding = false; + mesh.setPendingWake(0); + loop.onAgentEnd(); + assert(mesh.peekInbox().length === 0, "9) on the next clean turn-end the redelivered message is acked (drained)"); + 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..1aa69bf2 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 }, ); } From 621eddab361e3ea53e73fcc19a4bf9d4a864480d Mon Sep 17 00:00:00 2001 From: Matt Wilkinson Date: Wed, 8 Jul 2026 21:30:59 -0400 Subject: [PATCH 2/3] test(connector): model the held-queue flush before asserting the ack MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cubic P3 on #8: the ESC regression asserted the inbox drains on the next clean turn-end, but FakeHost never modeled the held nextTurn content actually being delivered into a turn — so it proved held-AND-acked, not held-AND-delivered (the drain came purely from the connector's surfaced bookkeeping). A future regression where the hidden queue never redelivers would still pass. Model OMP's deferred continuation (#promptQueuedHiddenNextTurnMessages, agent-session.ts:7323-7346) with FakeHost.flushHeld(): it drains the hidden next-turn queue into consumed turn content, exactly as a clean redelivery does. Test 9 now flushes the held message into a real turn and asserts it was consumed BEFORE the clean turn-end acks it — so the ack is proven to follow an actual delivery, closing cubic's gap and mirroring the real deliver-then-ack ordering. Refs #3. Co-Authored-By: seal --- .../interactive-loop.smoke.ts | 34 ++++++++++++++----- 1 file changed, 25 insertions(+), 9 deletions(-) diff --git a/extensions/connector-oh-my-pi/interactive-loop.smoke.ts b/extensions/connector-oh-my-pi/interactive-loop.smoke.ts index de6c188c..93255044 100644 --- a/extensions/connector-oh-my-pi/interactive-loop.smoke.ts +++ b/extensions/connector-oh-my-pi/interactive-loop.smoke.ts @@ -105,17 +105,30 @@ class FakeHost implements PeerHost { /** 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 — redelivered cleanly on the next turn - readonly turns: string[] = []; // a fresh turn woke on this message (the idle happy path) + 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]; } @@ -376,16 +389,19 @@ function assertEnvelope(call: SentCall, customType: string, ctx: string): void { 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"): the nextTurn delivery already - // woke a fresh turn (drive() sent it and armed the surfaced batch). When THAT turn ends cleanly — - // no interrupt this time — the held message must ack normally, proving it was held-AND-delivered, - // not held-AND-dropped. This is the ack-on-turn-end path (interactive-loop.ts ackSurfaced), which - // nextTurn preserves (the message rode a real #promptAgentInitiatedMessage turn). + // 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) on the next clean turn-end the redelivered message is acked (drained)"); + 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 ✅"); } From 8b98869a0ae6b064faee2893dc2369808738ded0 Mon Sep 17 00:00:00 2001 From: Matt Wilkinson Date: Wed, 8 Jul 2026 21:36:04 -0400 Subject: [PATCH 3/3] docs(connector): explain why ESC-interrupt can't early-ack a held message MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Capture the coalescing invariant at the ack site (greptile P1 on #8): a future reader touching ackSurfaced must see WHY a nextTurn message parked during an ESC unwind can't be acked before it's consumed. OMP holds the wire-level agent_end while #promptInFlightCount > 0 and lets a later one supersede it (agent-session.ts:2787-2799), so a wire-level subscriber sees one agent_end at the true settle — the interrupted turn and the deferred continuation collapse into a single post-consume event. Backstop: ackSurfaced drains only front-matching ids, so an unconsumed survivor redelivers (fails safe). Comment only, no behavior change. Refs #3. Co-Authored-By: seal --- .../connector-oh-my-pi/src/interactive-loop.ts | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/extensions/connector-oh-my-pi/src/interactive-loop.ts b/extensions/connector-oh-my-pi/src/interactive-loop.ts index 1aa69bf2..3fffdd19 100644 --- a/extensions/connector-oh-my-pi/src/interactive-loop.ts +++ b/extensions/connector-oh-my-pi/src/interactive-loop.ts @@ -136,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();