fix(connector): deliver peer messages as nextTurn, not steer

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);

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();