From 06dfbc841e3664de21ef5b290648c7ffb4fa3d8d Mon Sep 17 00:00:00 2001 From: Matt Wilkinson Date: Wed, 8 Jul 2026 17:13:52 -0400 Subject: [PATCH 1/5] fix(connector): name the oh-my-pi session after COTAL_NAME at join The launcher forwards COTAL_NAME but OMP had no agent-reachable way to set the session (pane/terminal) title, so every agent's title was wrong. The extension now calls `pi.setSessionName(config.name)` at session_start on the interactive branch, guarded on an unset name so a resumed or manually-renamed session is never clobbered, and best-effort so a title failure never breaks the mesh-join. Co-Authored-By: seal --- .../oh-my-pi-extension.smoke.ts | 71 +++++++++++++++---- .../connector-oh-my-pi/src/extension.ts | 15 +++- 2 files changed, 73 insertions(+), 13 deletions(-) diff --git a/extensions/connector-oh-my-pi/oh-my-pi-extension.smoke.ts b/extensions/connector-oh-my-pi/oh-my-pi-extension.smoke.ts index b9148b22..5dcffb71 100644 --- a/extensions/connector-oh-my-pi/oh-my-pi-extension.smoke.ts +++ b/extensions/connector-oh-my-pi/oh-my-pi-extension.smoke.ts @@ -28,10 +28,11 @@ interface RegisteredTool { } /** A fake ExtensionAPI that records everything the factory does. */ -function fakePi() { +function fakePi(opts?: { rejectSessionName?: boolean }) { const tools = new Map(); const events = new Map unknown>(); const sent: { message: Record; options: Record }[] = []; + const sessionNameSets: string[] = []; const z = zodV4.z; const pi = { zod: zodV4, @@ -42,8 +43,17 @@ function fakePi() { sent.push({ message, options }), registerCommand: () => {}, setLabel: () => {}, + // The title fix calls `await pi.setSessionName(config.name)`. Record every name it sets so we + // can assert on WHAT was set (and how many times); `rejectSessionName` models a host that + // refuses the rename, exercising the best-effort catch in session_start. + setSessionName: (name: string) => { + sessionNameSets.push(name); + return opts?.rejectSessionName + ? Promise.reject(new Error("smoke: setSessionName rejected")) + : Promise.resolve(); + }, }; - return { pi, tools, events, sent, z }; + return { pi, tools, events, sent, sessionNameSets, z }; } // ---- 1. inert without identity ------------------------------------------------ @@ -100,32 +110,69 @@ process.env.COTAL_SERVERS = "nats://127.0.0.1:4222"; // never actually connected // Each branch loads a fresh factory: the `started` guard is per-instance, so one instance can't be // re-driven. Env from block 2 (COTAL_NAME/COTAL_SERVERS) is still set → identity is present. { + // The title fix makes session_start `async` and, on an interactive session, name the session + // after the mesh identity via `ctx.sessionManager.getSessionName()` / `pi.setSessionName()`, so + // the ctx now carries a sessionManager. One shape for every sub-case below. + type SessionStart = ( + event: unknown, + ctx: { hasUI: boolean; sessionManager: { getSessionName: () => string | undefined } }, + ) => unknown | Promise; const origStart = MeshAgent.prototype.start; let startCalls = 0; MeshAgent.prototype.start = function () { startCalls++; }; try { - // (a) non-interactive session (subagent/print/RPC): hasUI:false → stays off the mesh. + // (a) non-interactive session (subagent/print/RPC): hasUI:false → stays off the mesh AND does + // no title work (behavior #3: a subagent must never rename the parent's session). { - const { pi, events } = fakePi(); + const { pi, events, sessionNameSets } = fakePi(); cotalMesh(pi as never); - const sessionStart = events.get("session_start") as - | ((event: unknown, ctx: { hasUI: boolean }) => unknown) - | undefined; + const sessionStart = events.get("session_start") as SessionStart | undefined; assert(sessionStart, "identity → subscribes to session_start"); startCalls = 0; - await sessionStart(undefined, { hasUI: false }); + await sessionStart(undefined, { hasUI: false, sessionManager: { getSessionName: () => undefined } }); assert(startCalls === 0, "hasUI:false → agent.start NOT invoked (subagent stays off mesh)"); + assert(sessionNameSets.length === 0, "hasUI:false → setSessionName NOT called (no title work off the mesh)"); } - // (b) interactive top-level session: hasUI:true → joins the mesh. + // (b) interactive top-level session, unnamed: hasUI:true + getSessionName() undefined → joins + // the mesh AND names the session after the mesh identity (behavior #1: title IS set when + // unset; the arg must be config.name == COTAL_NAME == "smoke-peer" from block 2's env). { - const { pi, events } = fakePi(); + const { pi, events, sessionNameSets } = fakePi(); cotalMesh(pi as never); - const sessionStart = events.get("session_start") as (event: unknown, ctx: { hasUI: boolean }) => unknown; + const sessionStart = events.get("session_start") as SessionStart; startCalls = 0; - await sessionStart(undefined, { hasUI: true }); + await sessionStart(undefined, { hasUI: true, sessionManager: { getSessionName: () => undefined } }); assert(startCalls === 1, "hasUI:true → agent.start invoked (interactive session joins)"); + assert( + sessionNameSets.length === 1 && sessionNameSets[0] === "smoke-peer", + `hasUI:true + unnamed → setSessionName called once with "smoke-peer" (got ${JSON.stringify(sessionNameSets)})`, + ); + } + // (c) interactive session already named (resumed / manual `/rename`, source:"user"): hasUI:true + // + getSessionName() returns a non-empty name → the guard suppresses the rename (behavior + // #2: an existing title is NEVER clobbered), yet the mesh-join still proceeds. + { + const { pi, events, sessionNameSets } = fakePi(); + cotalMesh(pi as never); + const sessionStart = events.get("session_start") as SessionStart; + startCalls = 0; + await sessionStart(undefined, { hasUI: true, sessionManager: { getSessionName: () => "user-renamed" } }); + assert(sessionNameSets.length === 0, "hasUI:true + already named → setSessionName NEVER called (guard protects /rename + resume)"); + assert(startCalls === 1, "hasUI:true + already named → agent.start still invoked (join unaffected by the guard)"); + } + // (d) best-effort: setSessionName rejects (host refuses the rename). session_start must still + // resolve and agent.start must still fire (behavior #4: a title failure must never break + // the mesh-join). The rename is attempted exactly once before the failure is swallowed. + { + const { pi, events, sessionNameSets } = fakePi({ rejectSessionName: true }); + cotalMesh(pi as never); + const sessionStart = events.get("session_start") as SessionStart; + startCalls = 0; + await sessionStart(undefined, { hasUI: true, sessionManager: { getSessionName: () => undefined } }); + assert(sessionNameSets.length === 1 && sessionNameSets[0] === "smoke-peer", "setSessionName rejects → rename attempted exactly once"); + assert(startCalls === 1, "setSessionName rejects → agent.start still invoked (best-effort: title failure never breaks the join)"); } } finally { MeshAgent.prototype.start = origStart; diff --git a/extensions/connector-oh-my-pi/src/extension.ts b/extensions/connector-oh-my-pi/src/extension.ts index 6d3fce5a..1ff6fc3c 100644 --- a/extensions/connector-oh-my-pi/src/extension.ts +++ b/extensions/connector-oh-my-pi/src/extension.ts @@ -68,13 +68,26 @@ export default function cotalMesh(pi: ExtensionAPI): void { // NOTE: a future headless launcher (e.g. Compass spawning a real worker) is also hasUI:false and // WOULD need to join — revisit with an explicit signal (agentKind/env opt-in) when that lands. let started = false; - pi.on("session_start", (_event, ctx: ExtensionContext) => { + pi.on("session_start", async (_event, ctx: ExtensionContext) => { if (started) return; started = true; if (!ctx.hasUI) { log("non-interactive session (subagent/print/RPC) — staying off the mesh"); return; } + // Name the session after the mesh identity so the terminal/pane title reflects WHO this agent + // is (COTAL_NAME) instead of a generic auto-title — the launcher forwards the name but OMP has + // no other agent-reachable way to set it (`/rename` isn't agent-invokable, the auto-title never + // fired). Connector-side, not an OMP→Cotal dependency. Guarded on an unset name so a resumed + // session or a manual `/rename` (both source:"user") is never clobbered; best-effort — a title + // failure must never break the join. + if (!ctx.sessionManager.getSessionName()) { + try { + await pi.setSessionName(config.name); + } catch (e) { + log(`could not set session name to "${config.name}": ${e instanceof Error ? e.message : String(e)}`, "warn"); + } + } agent.start(); // background connect with retry — never blocks }); From 4222fdd312f92bd2b8bf17d4ee2a4b8b85db754d Mon Sep 17 00:00:00 2001 From: Matt Wilkinson Date: Wed, 8 Jul 2026 17:47:34 -0400 Subject: [PATCH 2/5] fix(connector): never gate the mesh-join on the cosmetic session rename MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit greptile + cubic both flagged that `await pi.setSessionName()` ran before agent.start(), so a getSessionName() throw (session manager not ready) or a setSessionName() promise that stalls instead of rejecting would leave the pane off the mesh. Start the join first (non-blocking background connect), then do the title work in a detached fire-and-forget IIFE with getSessionName() inside the try — a title failure of any kind can no longer block connect. Smoke: flush the detached title work before asserting it (cases b/c/d no longer race) and add case (e) — getSessionName() throws => agent.start still fires, no rename, handler resolves. Red-green demonstrated: (e) reddens on the pre-fix ordering, green after. Co-Authored-By: seal --- .../oh-my-pi-extension.smoke.ts | 32 +++++++++++++++++++ .../connector-oh-my-pi/src/extension.ts | 15 ++++++--- 2 files changed, 42 insertions(+), 5 deletions(-) diff --git a/extensions/connector-oh-my-pi/oh-my-pi-extension.smoke.ts b/extensions/connector-oh-my-pi/oh-my-pi-extension.smoke.ts index 5dcffb71..54bd86b7 100644 --- a/extensions/connector-oh-my-pi/oh-my-pi-extension.smoke.ts +++ b/extensions/connector-oh-my-pi/oh-my-pi-extension.smoke.ts @@ -10,6 +10,7 @@ import cotalMesh from "./src/extension.ts"; import * as zodV4 from "zod/v4"; import { MeshAgent } from "@cotal-ai/connector-core"; +import { setImmediate as settle } from "node:timers/promises"; function assert(cond: unknown, msg: string): asserts cond { if (!cond) { @@ -144,6 +145,7 @@ process.env.COTAL_SERVERS = "nats://127.0.0.1:4222"; // never actually connected const sessionStart = events.get("session_start") as SessionStart; startCalls = 0; await sessionStart(undefined, { hasUI: true, sessionManager: { getSessionName: () => undefined } }); + await settle(); // the title work is now a detached fire-and-forget IIFE — let it settle before asserting sessionNameSets assert(startCalls === 1, "hasUI:true → agent.start invoked (interactive session joins)"); assert( sessionNameSets.length === 1 && sessionNameSets[0] === "smoke-peer", @@ -159,6 +161,7 @@ process.env.COTAL_SERVERS = "nats://127.0.0.1:4222"; // never actually connected const sessionStart = events.get("session_start") as SessionStart; startCalls = 0; await sessionStart(undefined, { hasUI: true, sessionManager: { getSessionName: () => "user-renamed" } }); + await settle(); // detached title IIFE: flush the microtask queue before asserting the (suppressed) rename assert(sessionNameSets.length === 0, "hasUI:true + already named → setSessionName NEVER called (guard protects /rename + resume)"); assert(startCalls === 1, "hasUI:true + already named → agent.start still invoked (join unaffected by the guard)"); } @@ -171,9 +174,38 @@ process.env.COTAL_SERVERS = "nats://127.0.0.1:4222"; // never actually connected const sessionStart = events.get("session_start") as SessionStart; startCalls = 0; await sessionStart(undefined, { hasUI: true, sessionManager: { getSessionName: () => undefined } }); + await settle(); // detached title IIFE: flush before asserting the attempted-then-swallowed rename assert(sessionNameSets.length === 1 && sessionNameSets[0] === "smoke-peer", "setSessionName rejects → rename attempted exactly once"); assert(startCalls === 1, "setSessionName rejects → agent.start still invoked (best-effort: title failure never breaks the join)"); } + // (e) P1 regression: getSessionName() THROWS (session manager not ready). The cosmetic rename + // must NEVER gate the mesh-join. With the fix, agent.start() fires FIRST and the guard read + // lives INSIDE the detached IIFE's try, so the throw is swallowed there and can't reach the + // handler — the join proceeds regardless (this is the exact both-bots P1: a not-ready + // session manager once left the pane off the mesh). + { + const { pi, events, sessionNameSets } = fakePi(); + cotalMesh(pi as never); + const sessionStart = events.get("session_start") as SessionStart; + startCalls = 0; + let handlerThrew = false; + try { + await sessionStart(undefined, { + hasUI: true, + sessionManager: { + getSessionName: () => { + throw new Error("session manager not ready"); + }, + }, + }); + } catch { + handlerThrew = true; + } + await settle(); // let the detached IIFE run (its try/catch swallows the getSessionName throw) + assert(!handlerThrew, "getSessionName throws → session_start still resolves (throw confined to the detached IIFE)"); + assert(startCalls === 1, "getSessionName throws → agent.start STILL invoked (the crux: a not-ready session manager never gates the join)"); + assert(sessionNameSets.length === 0, "getSessionName throws → setSessionName NEVER called (the guard read threw before any rename)"); + } } finally { MeshAgent.prototype.start = origStart; } diff --git a/extensions/connector-oh-my-pi/src/extension.ts b/extensions/connector-oh-my-pi/src/extension.ts index 1ff6fc3c..b3471972 100644 --- a/extensions/connector-oh-my-pi/src/extension.ts +++ b/extensions/connector-oh-my-pi/src/extension.ts @@ -75,20 +75,25 @@ export default function cotalMesh(pi: ExtensionAPI): void { log("non-interactive session (subagent/print/RPC) — staying off the mesh"); return; } + // Start the mesh join FIRST — it's a non-blocking background connect with retry. The session + // naming below is cosmetic and must never gate the join: a `getSessionName()` throw (session + // manager not ready) or a `setSessionName()` promise that stalls instead of rejecting would + // otherwise leave the pane off the mesh. So connect, then name in fire-and-forget. + agent.start(); // background connect with retry — never blocks // Name the session after the mesh identity so the terminal/pane title reflects WHO this agent // is (COTAL_NAME) instead of a generic auto-title — the launcher forwards the name but OMP has // no other agent-reachable way to set it (`/rename` isn't agent-invokable, the auto-title never // fired). Connector-side, not an OMP→Cotal dependency. Guarded on an unset name so a resumed // session or a manual `/rename` (both source:"user") is never clobbered; best-effort — a title - // failure must never break the join. - if (!ctx.sessionManager.getSessionName()) { + // failure (reject OR a getSessionName throw) must never break the join, so the whole path is + // detached and fully guarded. + void (async () => { try { - await pi.setSessionName(config.name); + if (!ctx.sessionManager.getSessionName()) await pi.setSessionName(config.name); } catch (e) { log(`could not set session name to "${config.name}": ${e instanceof Error ? e.message : String(e)}`, "warn"); } - } - agent.start(); // background connect with retry — never blocks + })(); }); const loop = runPeerLoop({ mesh: agent, host: pi }); From 905a6a471fcb22e99319c2e7e292b152d7b759de Mon Sep 17 00:00:00 2001 From: Matt Wilkinson Date: Wed, 8 Jul 2026 21:19:29 -0400 Subject: [PATCH 3/5] 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 4/5] 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 5/5] 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();