diff --git a/adapter/strings/omp-spt.mjs b/adapter/strings/omp-spt.mjs index c051d53..ba7abfa 100644 --- a/adapter/strings/omp-spt.mjs +++ b/adapter/strings/omp-spt.mjs @@ -1,6 +1,20 @@ import { spawn } from "node:child_process"; const ADAPTER = "omp-spt"; +const BUSY_TITLE_GLYPHS = [..."⣾⣽⣻⢿⡿⣟⣯⣷"]; +const IDLE_TITLE_GLYPH = "○"; +const TITLE_FRAME_MS = 80; + +// [impl->REQ-OMP-SESSION-TITLES] +export function endpointDisplayName(id, node, project) { + const endpoint = String(id ?? "").trim(); + const nodeName = String(node ?? "").trim(); + const projectName = String(project ?? "").trim(); + if (!nodeName) return endpoint; + return projectName + ? `${endpoint} @ ${nodeName} (${projectName}/)` + : `${endpoint} @ ${nodeName}`; +} export function decodeBody(body) { return body @@ -130,24 +144,6 @@ function messageText(message) { .join(""); } -export function extractReply(messages, afterUserMessage) { - const allMessages = messages ?? []; - let start = 0; - if (afterUserMessage !== undefined) { - const userIndex = allMessages.findLastIndex((message) => { - if (message?.role !== "user") return false; - const text = messageText(message); - return text === afterUserMessage || text.startsWith(`${afterUserMessage}\n\n message?.role === "assistant"); - return assistant ? messageText(assistant) : ""; -} function assistantMessageIdentity(message) { if (message?.role !== "assistant") return undefined; if (typeof message.responseId === "string" && message.responseId) { @@ -187,18 +183,6 @@ function successfulAssistant(message) { return SUCCESSFUL_ASSISTANT_STOP_REASONS.has(message?.stopReason); } -function unsuccessfulAssistantPayload(message) { - if (!message) return failureMessage("turn ended without an assistant response"); - if (message.stopReason === "aborted") { - return failureMessage("assistant turn was aborted before completion", message.errorMessage); - } - if (message.stopReason === "error") { - return failureMessage("assistant turn failed before completion", message.errorMessage); - } - return failureMessage( - `assistant turn ended without a successful completion (${message.stopReason ?? "missing stopReason"})`, - ); -} function firstLine(text) { @@ -210,10 +194,6 @@ function errorSummary(error) { return firstLine(detail).trim() || "unknown error"; } -function failureMessage(reason, error) { - const detail = error === undefined ? "" : `: ${errorSummary(error)}`; - return `[omp-spt] ${reason}${detail}`; -} function senderStub(sender) { const escaped = sender @@ -621,6 +601,8 @@ export function createOmpSpt(overrides = {}) { const spawnProcess = overrides.spawnProcess ?? spawn; const setTimer = overrides.setTimeout ?? globalThis.setTimeout; const clearTimer = overrides.clearTimeout ?? globalThis.clearTimeout; + const setRepeatingTimer = overrides.setInterval ?? globalThis.setInterval; + const clearRepeatingTimer = overrides.clearInterval ?? globalThis.clearInterval; const env = overrides.env ?? process.env; const platform = overrides.platform ?? process.platform; const killGraceMs = overrides.killGraceMs ?? DEFAULT_KILL_GRACE_MS; @@ -663,9 +645,6 @@ export function createOmpSpt(overrides = {}) { const acceptedBytesLimit = overrides.acceptedBytesLimit ?? DEFAULT_ACCEPTED_BYTES_LIMIT; const restartDelaysMs = [...(overrides.restartDelaysMs ?? [250, 1000, 4000])]; - const outcomeRetryDelaysMs = [ - ...(overrides.outcomeRetryDelaysMs ?? [250, 1000, 4000]), - ]; const sessionEndRetryDelaysMs = [ ...(overrides.sessionEndRetryDelaysMs ?? [250, 1000]), ]; @@ -800,6 +779,32 @@ export function createOmpSpt(overrides = {}) { let current; let stopping = false; let ui; + let titleTimer; + let titleFrame = 0; + const displayName = () => + endpointDisplayName(id, env.OMP_SPT_NODE, env.OMP_SPT_PROJECT); + const setWindowTitle = (glyph) => ui?.setTitle(`${glyph} ${displayName()}`); + const stopTitleAnimation = () => { + if (titleTimer !== undefined) { + clearRepeatingTimer(titleTimer); + titleTimer = undefined; + } + titleFrame = 0; + }; + const showIdleTitle = () => { + stopTitleAnimation(); + setWindowTitle(IDLE_TITLE_GLYPH); + }; + const showBusyTitle = () => { + stopTitleAnimation(); + setWindowTitle(BUSY_TITLE_GLYPHS[titleFrame]); + titleFrame = (titleFrame + 1) % BUSY_TITLE_GLYPHS.length; + titleTimer = setRepeatingTimer(() => { + setWindowTitle(BUSY_TITLE_GLYPHS[titleFrame]); + titleFrame = (titleFrame + 1) % BUSY_TITLE_GLYPHS.length; + }, TITLE_FRAME_MS); + titleTimer?.unref?.(); + }; let runtimeCtx; let endpointState; let stateOperation = Promise.resolve(); @@ -1168,6 +1173,9 @@ export function createOmpSpt(overrides = {}) { activated = true; startupBriefPending = true; ui.setStatus("omp-spt", `spt:${id}`); + // [impl->REQ-OMP-SESSION-TITLES] + pi.setSessionName(displayName()); + showIdleTitle(); startListener(); beginUpdateProbe(); if (options.announce) { @@ -1438,33 +1446,6 @@ export function createOmpSpt(overrides = {}) { } } - // [impl->REQ-OMP-EXTENSION-CUSTODY] - function settleItem(item, payload) { - if (!item) return Promise.resolve(); - if (item.outcomePromise) return item.outcomePromise; - item.settling = true; - item.outcomePromise = (async () => { - if (!item.from) throw new Error("missing EVENT from attribute"); - for (let attempt = 0; ; attempt += 1) { - try { - await runCommand(["send", item.from, "--from", id], payload); - item.settled = true; - return; - } catch (error) { - if (shutdownMode || attempt >= outcomeRetryDelaysMs.length) throw error; - const delay = outcomeRetryDelaysMs[attempt]; - pi.logger.error( - `omp-spt could not send the outcome to ${item.from}; retrying ${ - attempt + 1 - }/${outcomeRetryDelaysMs.length} in ${delay}ms`, - { error: errorSummary(error) }, - ); - await waitForRetry(delay); - } - } - })(); - return item.outcomePromise; - } function releaseItem(item) { if (!item?.accounted) return; @@ -1496,6 +1477,7 @@ export function createOmpSpt(overrides = {}) { } async function stopResources() { + stopTitleAnimation(); if (dispatchTimer !== undefined) { clearTimer(dispatchTimer); dispatchTimer = undefined; @@ -1518,51 +1500,13 @@ export function createOmpSpt(overrides = {}) { } } - async function settlePendingItem(item, reason) { - try { - if (item.outcomePromise && !item.settled) { - let existingError; - try { - await item.outcomePromise; - } catch (error) { - existingError = error; - } - if (item.settled) return; - if (existingError && !shutdownMode) { - logError( - `omp-spt could not return custody to ${item.from ?? "unknown"}`, - existingError, - ); - return; - } - item.outcomePromise = undefined; - item.settling = false; - } - try { - await settleItem(item, failureMessage(reason)); - } catch (error) { - logError(`omp-spt could not return custody to ${item.from ?? "unknown"}`, error); - } - } finally { - releaseItem(item); - } - } - - async function failPending(reason) { + function releasePending() { const pending = current ? [current, ...queue] : [...queue]; if (overflowItem) pending.push(overflowItem); current = undefined; queue.length = 0; overflowItem = undefined; - for (let index = 0; index < pending.length; index += 1) { - if (shutdownMode) { - await Promise.all( - pending.slice(index).map((item) => settlePendingItem(item, reason)), - ); - return; - } - await settlePendingItem(pending[index], reason); - } + for (const item of pending) releaseItem(item); } function teardownSession(pendingReason) { @@ -1570,7 +1514,7 @@ export function createOmpSpt(overrides = {}) { stopping = true; const operation = (async () => { await stopResources(); - await failPending(pendingReason); + releasePending(); await bindPromise?.catch(() => {}); await endSessionWithRetry(); })(); @@ -1655,13 +1599,6 @@ export function createOmpSpt(overrides = {}) { async function rejectItem(item, reason, error) { if (stopping) return; logError(`omp-spt ${reason}`, error); - try { - await settleItem(item, failureMessage(reason, error)); - } catch (outcomeError) { - if (stopping) return; - await failClosed(`omp-spt could not send the outcome to ${item.from ?? "unknown"}`, outcomeError); - return; - } if (current === item) { current = undefined; releaseItem(item); @@ -1892,7 +1829,7 @@ export function createOmpSpt(overrides = {}) { item.assistantBaseline = baseline; current = item; } - if (current?.submitted && !current.settling) { + if (current?.submitted) { if (current.activeInjected) { const content = `${current.stub}\n\n${current.envelope}`; if ( @@ -1976,32 +1913,15 @@ export function createOmpSpt(overrides = {}) { // [impl->REQ-HAZARD-ABNORMAL-TURN-RECEIVABILITY] async function completeTurn(event) { agentActive = false; + showIdleTitle(); desiredState = "idle"; if (stopping) return; const messages = event.messages ?? []; - const completed = current; - const completedBaseline = completed?.assistantBaseline ?? turnAssistantBaseline; const currentTurnAssistant = assistantAfterBaseline(messages, turnAssistantBaseline); - const completedAssistant = assistantAfterBaseline(messages, completedBaseline); + const completed = current; const currentTurnReply = successfulAssistant(currentTurnAssistant) ? messageText(currentTurnAssistant) : ""; - if (completed?.submitted && !completed.settled) { - const reply = messageText(completedAssistant); - const payload = successfulAssistant(completedAssistant) - ? reply || failureMessage("turn completed without a textual assistant response") - : unsuccessfulAssistantPayload(completedAssistant); - try { - await settleItem(completed, payload); - } catch (error) { - if (stopping) return; - await failClosed( - `omp-spt could not send the outcome to ${completed.from ?? "unknown"}`, - error, - ); - return; - } - } if (current === completed) { current = undefined; releaseItem(completed); @@ -2024,6 +1944,7 @@ export function createOmpSpt(overrides = {}) { turnAssistantBaseline = observedAssistantBaseline; turnContextObserved = false; agentActive = true; + showBusyTitle(); desiredState = "busy"; try { await syncDesiredState(); diff --git a/tests/omp-extension.mjs b/tests/omp-extension.mjs index a1d9ee7..5af4c92 100644 --- a/tests/omp-extension.mjs +++ b/tests/omp-extension.mjs @@ -2,9 +2,9 @@ import assert from "node:assert/strict"; import { EventEmitter } from "node:events"; import { createOmpSpt, + endpointDisplayName, decodeBody, drainEvents, - extractReply, parsePeerShortforms, runSpt, } from "../adapter/strings/omp-spt.mjs"; @@ -118,6 +118,9 @@ function createHarness(options = {}) { const selections = []; const confirmations = []; const inputs = []; + const titles = []; + const sessionNames = []; + const intervals = []; const compactions = []; const clock = new FakeClock(); let shutdowns = 0; @@ -146,6 +149,9 @@ function createHarness(options = {}) { setStatus(key, text) { statuses.push({ key, text }); }, + setTitle(title) { + titles.push(title); + }, async select(title, values) { selections.push({ title, values }); return options.onSelect?.(title, values) ?? options.selectResults?.shift(); @@ -174,6 +180,9 @@ function createHarness(options = {}) { }; const pi = { zod: { z }, + setSessionName(name) { + sessionNames.push(name); + }, logger: { error(message, details) { errors.push({ message, details }); @@ -227,6 +236,8 @@ function createHarness(options = {}) { SPT_ENDPOINT_ID: Object.hasOwn(options, "id") ? options.id : "omp-agent", OMP_SPT_SUBNET: options.subnet, OMP_SPT_SPT_BIN: "spt-test", + OMP_SPT_NODE: options.node, + OMP_SPT_PROJECT: options.project, }, checkUpdates: options.checkUpdates ?? false, fetchLatestAdapterVersion: options.fetchLatestAdapterVersion, @@ -234,7 +245,6 @@ function createHarness(options = {}) { acceptedBytesLimit: options.acceptedBytesLimit, acceptedQueueLimit: options.acceptedQueueLimit, restartDelaysMs: options.restartDelaysMs ?? [5, 10], - outcomeRetryDelaysMs: options.outcomeRetryDelaysMs ?? [], sessionEndRetryDelaysMs: options.sessionEndRetryDelaysMs ?? [], shutdownBudgetMs: options.shutdownBudgetMs, shutdownCommandTimeoutMs: options.shutdownCommandTimeoutMs, @@ -247,6 +257,19 @@ function createHarness(options = {}) { spawnProcess, setTimeout: clock.setTimeout.bind(clock), clearTimeout: clock.clearTimeout.bind(clock), + setInterval(fn, delay) { + const interval = { + active: true, + delay, + fn, + unref() {}, + }; + intervals.push(interval); + return interval; + }, + clearInterval(interval) { + interval.active = false; + }, }); extension(pi); @@ -276,6 +299,9 @@ function createHarness(options = {}) { selections, sentMessages, statuses, + intervals, + sessionNames, + titles, submitted, tools, get shutdowns() { @@ -300,8 +326,15 @@ function assertNoAgentManagedPoll(harness) { } -async function testParsingAndReplies() { +async function testParsing() { assert.equal(decodeBody('a<b>
"c"
legacy & &lt;'), 'a\n"c"\nlegacy & <'); + // [unit->REQ-OMP-SESSION-TITLES] + assert.equal( + endpointDisplayName("emphasys", "HFENDULEAM", "omp-spt"), + "emphasys @ HFENDULEAM (omp-spt/)", + ); + assert.equal(endpointDisplayName("emphasys", "HFENDULEAM"), "emphasys @ HFENDULEAM"); + assert.equal(endpointDisplayName("emphasys", undefined, "omp-spt"), "emphasys"); const partialEnvelope = 'hello
wo'; const partial = drainEvents(`noise${partialEnvelope}`); @@ -346,27 +379,31 @@ async function testParsingAndReplies() { /EVENT frame exceeded/, ); - assert.equal( - extractReply([ - assistantMessage([{ type: "text", text: "first" }]), - { role: "toolResult", content: [] }, - assistantMessage([ - { type: "text", text: "final " }, - { type: "text", text: "answer" }, - ]), - ]), - "final answer", - ); - assert.equal( - extractReply( - [ - assistantMessage("stale answer"), - { role: "user", content: '' }, - ], - '', - ), - "", - ); +} + +// [unit->REQ-OMP-SESSION-TITLES] +async function testEndpointSessionNameAndAnimatedWindowTitle() { + const harness = createHarness({ + id: "emphasys", + node: "HFENDULEAM", + project: "omp-spt", + }); + await harness.emit("session_start"); + assert.deepEqual(harness.sessionNames, ["emphasys @ HFENDULEAM (omp-spt/)"]); + assert.deepEqual(harness.titles, ["○ emphasys @ HFENDULEAM (omp-spt/)"]); + + await harness.emit("agent_start"); + assert.equal(harness.titles.at(-1), "⣾ emphasys @ HFENDULEAM (omp-spt/)"); + assert.equal(harness.intervals.length, 1); + assert.equal(harness.intervals[0].delay, 80); + harness.intervals[0].fn(); + assert.equal(harness.titles.at(-1), "⣽ emphasys @ HFENDULEAM (omp-spt/)"); + + await harness.emit("agent_end", { messages: [] }); + assert.equal(harness.intervals[0].active, false); + assert.equal(harness.titles.at(-1), "○ emphasys @ HFENDULEAM (omp-spt/)"); + await harness.emit("session_shutdown"); + assert.equal(harness.intervals[0].active, false); } // [unit->REQ-OMP-EXTENSION-CUSTODY] @@ -490,16 +527,11 @@ async function testLifecycleCustodyAndContext() { }); const outcomes = commandCalls(harness, "send"); - assert.equal(outcomes.length, 2); assert.deepEqual( - outcomes.map((call) => call.args), - [ - ["send", "alice", "--from", "omp-agent"], - ["send", "bob", "--from", "omp-agent"], - ], + outcomes, + [], + "ordinary assistant output must never be forwarded to a peer", ); - assert.equal(outcomes[0].input, "alice reply"); - assert.match(outcomes[1].input, /turn ended without an assistant response/); assert.deepEqual( stateCalls(harness).map((call) => call.args[4]), ["idle", "busy", "idle", "busy", "idle"], @@ -516,6 +548,42 @@ async function testLifecycleCustodyAndContext() { assert.deepEqual(harness.clock.delays(), []); } +// [unit->REQ-OMP-EXTENSION-CUSTODY] +// [unit->REQ-HAZARD-ABNORMAL-TURN-RECEIVABILITY] +async function testLocalAssistantOutputDoesNotReplyToPeer() { + const harness = createHarness(); + await harness.emit("session_start"); + harness.children[0].stdout.emit( + "data", + 'peer reply', + ); + await flush(); + await harness.emit("context", { + messages: [{ role: "user", content: '' }], + }); + await harness.emit("agent_start"); + await harness.emit("agent_end", { + messages: [ + { role: "user", content: '' }, + { role: "user", content: "local user interjection" }, + assistantMessage("answer intended for the local user"), + ], + }); + + assert.deepEqual( + commandCalls(harness, "send"), + [], + "a local interjection must not be correlated back to the peer", + ); + assert.deepEqual(harness.clock.delays(), []); + assert.deepEqual( + harness.submitted, + [''], + "the received peer delivery completes without an implicit response", + ); + await harness.emit("session_shutdown"); +} + async function testDeferredBindLifecycleSerialization() { const busyBind = deferred(); const busyHarness = createHarness({ @@ -592,92 +660,6 @@ async function testDeferredBindLifecycleSerialization() { assert.deepEqual(hungBindHarness.clock.delays(), []); } -// [unit->REQ-OMP-EXTENSION-CUSTODY] -// [unit->REQ-OMP-LISTENER-FAIL-CLOSED] -async function testOutcomeSendRetriesAndExhaustion() { - let retryAttempts = 0; - const retryHarness = createHarness({ - outcomeRetryDelaysMs: [5, 10], - onRun(call) { - if (call.args[0] === "send" && call.args[1] === "retry") { - retryAttempts += 1; - if (retryAttempts < 3) throw new Error(`outcome failure ${retryAttempts}`); - } - }, - }); - await retryHarness.emit("session_start"); - retryHarness.children[0].stdout.emit( - "data", - 'work', - ); - await flush(); - await retryHarness.emit("agent_start"); - const retryEnding = retryHarness.emit("agent_end", { - messages: [ - { role: "user", content: '' }, - assistantMessage("eventual outcome"), - ], - }); - await flush(); - assert.equal(commandCalls(retryHarness, "send").length, 1); - assert.deepEqual(retryHarness.clock.delays(), [5]); - await retryHarness.clock.runNext(5); - assert.equal(commandCalls(retryHarness, "send").length, 2); - assert.deepEqual(retryHarness.clock.delays(), [10]); - await retryHarness.clock.runNext(10); - await retryEnding; - assert.equal(commandCalls(retryHarness, "send").length, 3); - assert.equal(commandCalls(retryHarness, "send").at(-1).input, "eventual outcome"); - assert.equal(retryHarness.shutdowns, 0); - await retryHarness.emit("session_shutdown"); - assert.deepEqual(retryHarness.clock.delays(), []); - - const exhaustedHarness = createHarness({ - outcomeRetryDelaysMs: [7], - onRun(call) { - if (call.args[0] === "send") throw new Error("outcome channel unavailable"); - }, - }); - await exhaustedHarness.emit("session_start"); - const exhaustedListener = exhaustedHarness.children[0]; - exhaustedListener.stdout.emit( - "data", - 'work', - ); - await flush(); - await exhaustedHarness.emit("agent_start"); - const exhaustedEnding = exhaustedHarness.emit("agent_end", { - messages: [ - { role: "user", content: '' }, - assistantMessage("undeliverable outcome"), - ], - }); - await flush(); - assert.equal(commandCalls(exhaustedHarness, "send").length, 1); - assert.deepEqual(exhaustedHarness.clock.delays(), [7]); - await exhaustedHarness.clock.runNext(7); - await exhaustedEnding; - - assert.equal(commandCalls(exhaustedHarness, "send").length, 2); - assert.deepEqual( - stateCalls(exhaustedHarness).map((call) => call.args[4]), - ["idle", "busy"], - "exhausted custody must never be advertised idle", - ); - assert.equal(exhaustedHarness.shutdowns, 1); - assert.equal(exhaustedListener.kills, 1); - assert.equal( - exhaustedHarness.calls.filter((call) => call.args[3] === "session-end").length, - 1, - ); - assert.ok( - exhaustedHarness.errors.some(({ message }) => - message.includes("could not send the outcome to exhausted"), - ), - ); - await exhaustedHarness.emit("session_shutdown"); - assert.deepEqual(exhaustedHarness.clock.delays(), []); -} // [unit->REQ-OMP-EXTENSION-CUSTODY] async function testSubmissionFailureAdvancesQueue() { @@ -693,10 +675,11 @@ async function testSubmissionFailureAdvancesQueue() { ); await flush(); - const firstOutcome = commandCalls(harness, "send"); - assert.equal(firstOutcome.length, 1); - assert.deepEqual(firstOutcome[0].args, ["send", "broken", "--from", "omp-agent"]); - assert.match(firstOutcome[0].input, /could not submit your message to OMP/); + assert.deepEqual( + commandCalls(harness, "send"), + [], + "a rejected local submission must not message the peer implicitly", + ); assert.deepEqual(harness.clock.delays(), [0]); await harness.clock.runNext(0); @@ -708,13 +691,11 @@ async function testSubmissionFailureAdvancesQueue() { assistantMessage("next reply"), ], }); - const outcomes = commandCalls(harness, "send"); - assert.equal(outcomes.length, 2); assert.deepEqual( - outcomes.map((call) => call.args[1]), - ["broken", "next"], + commandCalls(harness, "send"), + [], + "assistant output for the next delivery must remain local", ); - assert.equal(outcomes[1].input, "next reply"); await harness.emit("session_shutdown"); assert.deepEqual(harness.clock.delays(), []); } @@ -736,8 +717,7 @@ async function testFailedIdleRecoveryFailsClosed() { harness.children[0].stdout.emit("data", 'one'); await flush(); - assert.equal(commandCalls(harness, "send").length, 1); - assert.match(commandCalls(harness, "send")[0].input, /could not submit your message to OMP/); + assert.deepEqual(commandCalls(harness, "send"), []); assert.equal(harness.shutdowns, 1); assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 1); assert.ok( @@ -836,64 +816,6 @@ async function testListenerStableIntervalResetsRetries() { assert.deepEqual(harness.clock.delays(), []); } -// [unit->REQ-OMP-EXTENSION-CUSTODY] -// [unit->REQ-OMP-LISTENER-FAIL-CLOSED] -async function testFatalTeardownAwaitsInFlightOutcome() { - let releaseOutcome; - const outcomeGate = new Promise((resolve) => { - releaseOutcome = resolve; - }); - const harness = createHarness({ - restartDelaysMs: [], - onRun(call) { - if (call.args[0] === "send" && call.args[1] === "slow") return outcomeGate; - }, - }); - await harness.emit("session_start"); - harness.children[0].stdout.emit("data", 'work'); - await flush(); - await harness.emit("agent_start"); - const ending = harness.emit("agent_end", { - messages: [ - { role: "user", content: '' }, - assistantMessage("finished"), - ], - }); - await flush(); - assert.equal(commandCalls(harness, "send").length, 1); - - harness.children[0].emit("close", 11); - await flush(); - assert.equal(harness.shutdowns, 0, "fatal teardown must join the sender outcome"); - assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 0); - - const shutdown = harness.emit("session_shutdown"); - await flush(); - assert.equal( - harness.calls.filter((call) => call.args[3] === "session-end").length, - 0, - "concurrent lifecycle shutdown must join fatal custody teardown", - ); - for (const reason of ["new", "resume", "fork", "handoff"]) { - assert.deepEqual( - await harness.emit("session_before_switch", { reason }), - { cancel: true }, - `teardown must keep blocking the ${reason} switch while custody is pending`, - ); - } - assert.deepEqual( - await harness.emit("session_before_branch"), - { cancel: true }, - "teardown must keep blocking branches while custody is pending", - ); - - releaseOutcome(); - await Promise.all([ending, shutdown]); - await flush(); - assert.equal(harness.shutdowns, 1); - assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 1); - assert.deepEqual(harness.clock.delays(), []); -} async function testSessionEndRetriesAfterTransientFailure() { const firstEnd = deferred(); @@ -964,7 +886,7 @@ async function testHumanBusyFailureFailsClosed() { // [unit->REQ-OMP-EXTENSION-CUSTODY] // [unit->REQ-OMP-LISTENER-FAIL-CLOSED] -async function testShutdownReapsAndFailsQueuedCustody() { +async function testShutdownReapsAndReleasesQueuedCustody() { const harness = createHarness(); await harness.emit("session_start"); const listener = harness.children[0]; @@ -986,10 +908,10 @@ async function testShutdownReapsAndFailsQueuedCustody() { assert.equal(listener.kills, 1); assert.deepEqual(harness.clock.delays(), []); assert.deepEqual( - commandCalls(harness, "send").map((call) => call.args[1]), - ["first", "queued"], + commandCalls(harness, "send"), + [], + "shutdown must not synthesize outbound peer messages", ); - assert.match(commandCalls(harness, "send")[1].input, /OMP session shut down/); assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 1); assert.deepEqual(harness.statuses.at(-1), { key: "omp-spt", text: undefined }); assert.equal(harness.shutdowns, 0, "normal lifecycle shutdown must not recursively shut down OMP"); @@ -1155,7 +1077,7 @@ async function testProtocolCorruptionFailsClosed() { // [unit->REQ-OMP-EXTENSION-CUSTODY] // [unit->REQ-OMP-LISTENER-FAIL-CLOSED] -async function testInboundQueueOverflowReturnsAcceptedCustody() { +async function testInboundQueueOverflowReleasesAcceptedCustody() { const frames = ["a", "b", "overflow"].map( (from) => `work`, ); @@ -1170,13 +1092,10 @@ async function testInboundQueueOverflowReturnsAcceptedCustody() { assert.equal(harness.shutdowns, 1); assert.deepEqual(harness.submitted, []); assert.deepEqual( - commandCalls(harness, "send").map((call) => call.args[1]), - ["a", "b", "overflow"], - "every accepted item and the capacity-refused item receive an explicit terminal failure", + commandCalls(harness, "send"), + [], + "capacity failure must not synthesize outbound peer messages", ); - for (const call of commandCalls(harness, "send")) { - assert.match(call.input, /endpoint stopped before your message could complete/); - } assert.ok( harness.errors.some(({ message }) => message.includes("inbound custody capacity exceeded"), @@ -1196,61 +1115,12 @@ async function testInboundQueueOverflowReturnsAcceptedCustody() { await byteHarness.emit("session_start"); byteHarness.children[0].stdout.emit("data", `${byteFirst}${byteOverflow}`); await flush(); - assert.deepEqual( - commandCalls(byteHarness, "send").map((call) => call.args[1]), - ["a", "b"], - ); + assert.deepEqual(commandCalls(byteHarness, "send"), []); assert.equal(byteHarness.shutdowns, 1); assert.deepEqual(byteHarness.clock.delays(), []); await byteHarness.emit("session_shutdown"); } -async function testSessionStopAwaitsOutcomeWithoutEndingEndpoint() { - const firstOutcome = deferred(); - let attempts = 0; - const harness = createHarness({ - outcomeRetryDelaysMs: [5], - onRun(call) { - if (call.args[0] === "send" && call.args[1] === "awaited") { - attempts += 1; - if (attempts === 1) return firstOutcome.promise; - } - }, - }); - await harness.emit("session_start"); - harness.children[0].stdout.emit( - "data", - 'work', - ); - await flush(); - await harness.emit("agent_start"); - const messages = [ - { role: "user", content: '' }, - assistantMessage("completed outcome"), - ]; - const agentEnd = harness.emit("agent_end", { messages }); - await flush(); - const sessionStop = harness.emit("session_stop", { messages }); - await flush(); - assert.equal(commandCalls(harness, "send").length, 1); - assert.equal(harness.children[0].kills, 0); - assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 0); - - firstOutcome.reject(new Error("transient delayed outcome failure")); - await flush(); - assert.deepEqual(harness.clock.delays(), [5]); - await harness.clock.runNext(5); - await Promise.all([agentEnd, sessionStop]); - assert.equal(commandCalls(harness, "send").length, 2); - assert.equal(commandCalls(harness, "send").at(-1).input, "completed outcome"); - assert.equal(harness.children[0].kills, 0, "ordinary session_stop must leave the endpoint live"); - assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 0); - - await harness.emit("session_shutdown"); - assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 1); - assert.equal(harness.children[0].kills, 1); - assert.deepEqual(harness.clock.delays(), []); -} // [unit->REQ-OMP-EXTENSION-CUSTODY] // [unit->REQ-OMP-LISTENER-FAIL-CLOSED] @@ -1260,193 +1130,26 @@ async function testShutdownFallbackStaysBelowHostCap() { shutdownBudgetMs: 1_800, shutdownCommandTimeoutMs: 300, onRun(call) { - if (call.args[0] === "send" || call.args[3] === "session-end") return never; + if (call.args[3] === "session-end") return never; }, }); await harness.emit("session_start"); - await harness.emit("agent_start"); - harness.children[0].stdout.emit( - "data", - 'work', - ); - await flush(); - const shutdown = harness.emit("session_shutdown"); await flush(); + assert.deepEqual( harness.clock.delays().sort((a, b) => a - b), [300, 1_800], - "queued custody has a short command timeout inside the 2s host cap", + "session-end has a short command timeout inside the 2s host cap", ); await harness.clock.runNext(300); - assert.equal(commandCalls(harness, "send").length, 1); - assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 1); - assert.deepEqual(harness.clock.delays().sort((a, b) => a - b), [300, 1_800]); - await harness.clock.runNext(300); await shutdown; assert.equal(harness.children[0].kills, 1); - assert.ok( - harness.errors.some(({ message }) => - message.includes("could not return custody to queued"), - ), - ); + assert.deepEqual(commandCalls(harness, "send"), []); assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 1); assert.deepEqual(harness.clock.delays(), []); assert.ok(1_800 < 2_000); - - const queuedGates = [deferred(), deferred()]; - let queuedSendIndex = 0; - const concurrentHarness = createHarness({ - onRun(call) { - if (call.args[0] === "send") { - const gate = queuedGates[queuedSendIndex]; - queuedSendIndex += 1; - return gate.promise; - } - }, - }); - await concurrentHarness.emit("session_start"); - await concurrentHarness.emit("agent_start"); - concurrentHarness.children[0].stdout.emit( - "data", - 'onetwo', - ); - await flush(); - const concurrentShutdown = concurrentHarness.emit("session_shutdown"); - await flush(); - assert.deepEqual( - commandCalls(concurrentHarness, "send").map((call) => call.args[1]), - ["queued-a", "queued-b"], - "all pending custody failures must start concurrently", - ); - assert.deepEqual( - concurrentHarness.clock.delays().sort((a, b) => a - b), - [300, 300, 1_800], - ); - for (const gate of queuedGates) gate.resolve(); - await concurrentShutdown; - assert.equal( - concurrentHarness.calls.filter((call) => call.args[3] === "session-end").length, - 1, - ); - assert.deepEqual(concurrentHarness.clock.delays(), []); - - const busyGate = deferred(); - const dispatchHarness = createHarness({ - onRun(call) { - if (call.args[0] === "api" && call.args[3] === "state" && call.args[4] === "busy") { - return busyGate.promise; - } - }, - }); - await dispatchHarness.emit("session_start"); - dispatchHarness.children[0].stdout.emit( - "data", - 'work', - ); - await flush(); - assert.deepEqual(dispatchHarness.submitted, []); - assert.equal( - stateCalls(dispatchHarness).filter((call) => call.args[4] === "busy").length, - 1, - ); - const dispatchShutdown = dispatchHarness.emit("session_shutdown"); - await dispatchShutdown; - assert.deepEqual(dispatchHarness.submitted, []); - assert.deepEqual( - commandCalls(dispatchHarness, "send").map((call) => call.args[1]), - ["dispatching"], - ); - assert.match(commandCalls(dispatchHarness, "send")[0].input, /OMP session shut down/); - assert.equal( - dispatchHarness.calls.filter((call) => call.args[3] === "session-end").length, - 1, - ); - assert.equal(dispatchHarness.shutdowns, 0); - assert.deepEqual(dispatchHarness.clock.delays(), []); - busyGate.resolve(); - await flush(); - assert.deepEqual(dispatchHarness.submitted, []); - - const inFlightOutcome = deferred(); - let inFlightAttempts = 0; - const inFlightHarness = createHarness({ - onRun(call) { - if (call.args[0] === "send" && call.args[1] === "in-flight") { - inFlightAttempts += 1; - if (inFlightAttempts === 1) return inFlightOutcome.promise; - } - }, - }); - await inFlightHarness.emit("session_start"); - inFlightHarness.children[0].stdout.emit( - "data", - 'work', - ); - await flush(); - await inFlightHarness.emit("agent_start"); - const ending = inFlightHarness.emit("agent_end", { - messages: [ - { role: "user", content: '' }, - assistantMessage("answer racing shutdown"), - ], - }); - await flush(); - assert.equal(commandCalls(inFlightHarness, "send").length, 1); - const inFlightShutdown = inFlightHarness.emit("session_shutdown"); - await Promise.all([ending, inFlightShutdown]); - assert.deepEqual( - commandCalls(inFlightHarness, "send").map((call) => call.args[1]), - ["in-flight", "in-flight"], - ); - assert.match(commandCalls(inFlightHarness, "send")[1].input, /OMP session shut down/); - assert.equal( - inFlightHarness.calls.filter((call) => call.args[3] === "session-end").length, - 1, - ); - assert.equal(inFlightHarness.shutdowns, 0); - assert.deepEqual(inFlightHarness.clock.delays(), []); - inFlightOutcome.resolve(); - await flush(); - assert.equal(commandCalls(inFlightHarness, "send").length, 2); - - const hardCapHarness = createHarness({ - shutdownBudgetMs: 1_800, - shutdownCommandTimeoutMs: 5_000, - killForceMs: 100, - killGraceMs: 100, - onRun(call) { - if (call.args[0] === "send") return never; - }, - }); - await hardCapHarness.emit("session_start"); - await hardCapHarness.emit("agent_start"); - hardCapHarness.children[0].stdout.emit( - "data", - 'work', - ); - await flush(); - const hardCappedShutdown = hardCapHarness.emit("session_shutdown"); - await flush(); - assert.deepEqual( - hardCapHarness.clock.delays().sort((a, b) => a - b), - [600, 1_800], - ); - await hardCapHarness.clock.runNext(600); - await hardCappedShutdown; - assert.equal( - hardCapHarness.calls.filter((call) => call.args[3] === "session-end").length, - 1, - "the phase clamp must reserve time for one session-end attempt", - ); - assert.ok( - !hardCapHarness.errors.some(({ message }) => - message.includes("bounded shutdown expired"), - ), - ); - assert.equal(hardCapHarness.children[0].kills, 1); - assert.deepEqual(hardCapHarness.clock.delays(), []); } // [unit->REQ-PARITY-READY-ACTIVATION] @@ -1916,8 +1619,11 @@ async function testActiveTurnBoundaryDeliveryAndFallback() { assistantMessage("first outcome"), ], }); - assert.equal(commandCalls(harness, "send")[0].args[1], "first"); - assert.equal(commandCalls(harness, "send")[0].input, "first outcome"); + assert.deepEqual( + commandCalls(harness, "send"), + [], + "assistant output at an active boundary must remain local", + ); assert.deepEqual(harness.clock.delays(), [0]); await harness.clock.runNext(0); assert.deepEqual( @@ -1948,9 +1654,7 @@ async function testAbnormalTurnsRestoreReceivability() { messages: [{ role: "user", content: "active work" }], }); await harness.emit(completionEvent, { messages: boundary.messages }); - const outcome = commandCalls(harness, "send")[0]; - assert.equal(outcome.args[1], label); - assert.match(outcome.input, /turn ended without an assistant response/); + assert.deepEqual(commandCalls(harness, "send"), []); assert.equal(stateCalls(harness).at(-1).args[4], "idle"); harness.children[0].stdout.emit( @@ -1965,10 +1669,7 @@ async function testAbnormalTurnsRestoreReceivability() { async function testCompletionStopReasonGatesSideEffects() { - for (const [stopReason, expected] of [ - ["aborted", /aborted before completion/], - ["error", /failed before completion: provider unavailable/], - ]) { + for (const stopReason of ["aborted", "error"]) { const harness = createHarness(); await harness.emit("session_start"); harness.children[0].stdout.emit( @@ -1988,19 +1689,17 @@ async function testCompletionStopReasonGatesSideEffects() { }); const sends = commandCalls(harness, "send"); assert.deepEqual( - sends.map((call) => call.args[1]), - [`${stopReason}-sender`], - `${stopReason} partial output must settle custody without peer-message side effects`, + sends, + [], + `${stopReason} partial output must not produce peer-message side effects`, ); - assert.match(sends[0].input, expected); - assert.ok(!sends[0].input.includes("partial output")); assert.deepEqual(harness.sentMessages, []); assert.equal(stateCalls(harness).at(-1).args[4], "idle"); await harness.emit("session_shutdown"); } } -async function testCompactionSafeReplyCorrelation() { +async function testCompactionPreservesLocalAssistantOutput() { const highHistory = Array.from({ length: 64 }, (_unused, index) => assistantMessage(`historical-${index}`, { timestamp: 10_000 + index }), ); @@ -2026,7 +1725,7 @@ async function testCompactionSafeReplyCorrelation() { assistantMessage("valid reply after compaction", { timestamp: 20_000 }), ], }); - assert.equal(commandCalls(harness, "send")[0].input, "valid reply after compaction"); + assert.deepEqual(commandCalls(harness, "send"), []); await harness.emit("session_shutdown"); const stale = createHarness(); @@ -2042,9 +1741,7 @@ async function testCompactionSafeReplyCorrelation() { }); await stale.emit("context", { messages: [highHistory[63]] }); await stale.emit("agent_end", { messages: [highHistory[63]] }); - const staleOutcome = commandCalls(stale, "send")[0].input; - assert.match(staleOutcome, /turn ended without an assistant response/); - assert.ok(!staleOutcome.includes("historical-63")); + assert.deepEqual(commandCalls(stale, "send"), []); await stale.emit("session_shutdown"); } @@ -2278,23 +1975,22 @@ async function testNativeCheckpointTool() { await failed.emit("session_shutdown"); } -await testParsingAndReplies(); +await testParsing(); +await testEndpointSessionNameAndAnimatedWindowTitle(); await testRunSptRejectsStdinErrorsAndHungCommands(); await testLifecycleCustodyAndContext(); +await testLocalAssistantOutputDoesNotReplyToPeer(); await testDeferredBindLifecycleSerialization(); -await testOutcomeSendRetriesAndExhaustion(); await testSubmissionFailureAdvancesQueue(); await testFailedIdleRecoveryFailsClosed(); await testListenerRestartExhaustion(); await testListenerStableIntervalResetsRetries(); -await testFatalTeardownAwaitsInFlightOutcome(); await testSessionEndRetriesAfterTransientFailure(); await testHumanBusyFailureFailsClosed(); -await testShutdownReapsAndFailsQueuedCustody(); +await testShutdownReapsAndReleasesQueuedCustody(); await testListenerTerminationEscalatesAndReaps(); await testProtocolCorruptionFailsClosed(); -await testInboundQueueOverflowReturnsAcceptedCustody(); -await testSessionStopAwaitsOutcomeWithoutEndingEndpoint(); +await testInboundQueueOverflowReleasesAcceptedCustody(); await testShutdownFallbackStaysBelowHostCap(); await testNativeActivationCommandsAndErrors(); await testNativeActivationSelectionAndCompletion(); @@ -2304,7 +2000,7 @@ await testStartupBriefHintsAndUpdateNotices(); await testActiveTurnBoundaryDeliveryAndFallback(); await testAbnormalTurnsRestoreReceivability(); await testCompletionStopReasonGatesSideEffects(); -await testCompactionSafeReplyCorrelation(); +await testCompactionPreservesLocalAssistantOutput(); await testPeerShortformParsingAndDispatch(); await testNativeCheckpointTool();