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/oh-my-pi-extension.smoke.ts b/extensions/connector-oh-my-pi/oh-my-pi-extension.smoke.ts index b9148b22..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) { @@ -28,10 +29,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 +44,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 +111,100 @@ 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 } }); + 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", + `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" } }); + 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)"); + } + // (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 } }); + 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 6d3fce5a..b3471972 100644 --- a/extensions/connector-oh-my-pi/src/extension.ts +++ b/extensions/connector-oh-my-pi/src/extension.ts @@ -68,14 +68,32 @@ 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; } + // 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 (reject OR a getSessionName throw) must never break the join, so the whole path is + // detached and fully guarded. + void (async () => { + try { + 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"); + } + })(); }); const loop = runPeerLoop({ mesh: agent, host: pi }); 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();