diff --git a/tests/omp-extension.mjs b/tests/omp-extension.mjs index a1d9ee7..4adfcf5 100644 --- a/tests/omp-extension.mjs +++ b/tests/omp-extension.mjs @@ -1,32 +1,31 @@ import assert from "node:assert/strict"; import { EventEmitter } from "node:events"; import { createOmpSpt, decodeBody, drainEvents, - extractReply, parsePeerShortforms, runSpt, } from "../adapter/strings/omp-spt.mjs"; const flush = () => new Promise((resolve) => setImmediate(resolve)); function deferred() { let resolve; let reject; const promise = new Promise((resolvePromise, rejectPromise) => { resolve = resolvePromise; reject = rejectPromise; }); return { promise, reject, resolve }; } let nextAssistantTimestamp = 1; function assistantMessage(content, options = {}) { return { role: "assistant", content, stopReason: options.stopReason ?? "stop", timestamp: options.timestamp ?? nextAssistantTimestamp++, ...options, }; } @@ -212,51 +211,50 @@ function createHarness(options = {}) { return options.bindOutput ?? "BOUND endpoint token=token-123"; } return ""; }; const spawnProcess = (binary, args, spawnOptions) => { const child = new FakeChild(); child.binary = binary; child.args = [...args]; child.spawnOptions = spawnOptions; options.onSpawn?.(child); children.push(child); return child; }; const extension = createOmpSpt({ env: { SPT_ENDPOINT_ID: Object.hasOwn(options, "id") ? options.id : "omp-agent", OMP_SPT_SUBNET: options.subnet, OMP_SPT_SPT_BIN: "spt-test", }, checkUpdates: options.checkUpdates ?? false, fetchLatestAdapterVersion: options.fetchLatestAdapterVersion, platform: options.platform, acceptedBytesLimit: options.acceptedBytesLimit, acceptedQueueLimit: options.acceptedQueueLimit, restartDelaysMs: options.restartDelaysMs ?? [5, 10], - outcomeRetryDelaysMs: options.outcomeRetryDelaysMs ?? [], sessionEndRetryDelaysMs: options.sessionEndRetryDelaysMs ?? [], shutdownBudgetMs: options.shutdownBudgetMs, shutdownCommandTimeoutMs: options.shutdownCommandTimeoutMs, shortformCommandTimeoutMs: options.shortformCommandTimeoutMs, listenerStableMs: options.listenerStableMs ?? false, killForceMs: options.killForceMs ?? 4, killGraceMs: options.killGraceMs ?? 3, listenerBufferLimit: options.listenerBufferLimit, runSptCommand, spawnProcess, setTimeout: clock.setTimeout.bind(clock), clearTimeout: clock.clearTimeout.bind(clock), }); extension(pi); async function emit(name, event = {}) { let result; for (const handler of handlers.get(name) ?? []) { const returned = await handler({ type: name, ...event }, ctx); if (returned !== undefined) result = returned; } return result; } return { @@ -278,117 +276,96 @@ function createHarness(options = {}) { statuses, submitted, tools, get shutdowns() { return shutdowns; }, }; } function commandCalls(harness, command) { return harness.calls.filter((call) => call.args[0] === command); } function stateCalls(harness) { return harness.calls.filter((call) => call.args[0] === "api" && call.args[3] === "state"); } function assertNoAgentManagedPoll(harness) { assert.ok( !harness.calls.some((call) => call.args.includes("poll")) && !harness.children.some((child) => child.args.includes("poll")), "extension activation and delivery must never launch an agent-managed background poll", ); } -async function testParsingAndReplies() { +async function testParsing() { assert.equal(decodeBody('a<b>
"c"
legacy & &lt;'), 'a\n"c"\nlegacy & <'); const partialEnvelope = 'hello
wo'; const partial = drainEvents(`noise${partialEnvelope}`); assert.deepEqual(partial.events, []); assert.equal(partial.rest, partialEnvelope); const envelope = `${partial.rest}rld`; const complete = drainEvents(`${envelope}skip`); assert.deepEqual(complete.events, [ { from: "doyle", body: "hello\nworld", envelope }, ]); assert.equal(complete.rest, ""); const literalEventBody = 'No inbound containing the message has surfaced.'; assert.deepEqual(drainEvents(literalEventBody).events, [ { from: "hertz", body: "No inbound containing the message has surfaced.", envelope: literalEventBody, }, ]); const truncatedA = 'truncatedvalid'; const nested = drainEvents(truncatedA); assert.deepEqual(nested.events, []); assert.equal(nested.rest, truncatedA); assert.match(nested.error.message, /nested EVENT/); assert.match( drainEvents('missing sender').error.message, /missing EVENT from/, ); assert.match( drainEvents('bad attrs').error.message, /malformed EVENT attributes/, ); assert.match( drainEvents('0123456789', { maxFrameChars: 32, }).error.message, /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-EXTENSION-CUSTODY] // [unit->REQ-OMP-SESSION-IMMUTABLE] // [unit->REQ-OMP-MESSAGE-CONTEXT] // [unit->REQ-OMP-NATIVE-TUI] async function testLifecycleCustodyAndContext() { const harness = createHarness({ subnet: "mesh-a" }); assert.deepEqual([...harness.handlers.keys()], [ "session_start", "session_before_switch", "session_before_branch", "context", "before_agent_start", "agent_start", "agent_end", "session_stop", "session_shutdown", ]); await harness.emit("session_start"); assert.deepEqual(harness.calls[0], { args: [ "api", "--adapter", @@ -468,76 +445,107 @@ async function testLifecycleCustodyAndContext() { stateCalls(harness).map((call) => call.args[4]), ["idle", "busy"], "agent_start must not duplicate the already-honest busy transition", ); const aliceReply = assistantMessage([{ type: "text", text: "alice reply" }]); await harness.emit("agent_end", { messages: [ { role: "user", content: '' }, aliceReply, ], }); assert.deepEqual(harness.submitted, ['']); assert.deepEqual(harness.clock.delays(), [0]); await harness.clock.runNext(0); assert.deepEqual(harness.submitted, ['', '']); await harness.emit("agent_start"); await harness.emit("agent_end", { messages: [ aliceReply, { role: "user", content: '' }, ], }); 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"], ); for (const call of [...stateCalls(harness), ...harness.calls.filter((candidate) => candidate.args[3] === "session-end")]) { assert.deepEqual(call.args.slice(-2), ["--token", "token-123"]); } await harness.emit("session_shutdown"); const ended = harness.calls.filter((call) => call.args[3] === "session-end"); assert.equal(ended.length, 1); assert.deepEqual(ended[0].args.slice(-2), ["--token", "token-123"]); assert.equal(harness.children[0].kills, 1); 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({ onRun(call) { if (call.args[0] === "api" && call.args[3] === "bind") return busyBind.promise; }, }); const startingBusy = busyHarness.emit("session_start"); await flush(); const becomingBusy = busyHarness.emit("agent_start"); await flush(); assert.deepEqual(stateCalls(busyHarness), []); assert.equal(busyHarness.children.length, 0); busyBind.resolve("BOUND endpoint token=token-busy"); await Promise.all([startingBusy, becomingBusy]); assert.deepEqual( stateCalls(busyHarness).map((call) => call.args[4]), ["busy"], "agent_start before bind completion must suppress the stale idle publication", ); assert.equal(busyHarness.children.length, 1); await busyHarness.emit("session_shutdown"); assert.deepEqual(busyHarness.clock.delays(), []); @@ -570,196 +578,108 @@ async function testDeferredBindLifecycleSerialization() { "bind completion after shutdown must not restore live status", ); assert.deepEqual(shutdownHarness.clock.delays(), []); const hungBindHarness = createHarness({ onRun(call) { if (call.args[0] === "api" && call.args[3] === "bind") return new Promise(() => {}); }, }); const hungStart = hungBindHarness.emit("session_start"); await flush(); const hungShutdown = hungBindHarness.emit("session_shutdown"); await flush(); assert.deepEqual(hungBindHarness.clock.delays().sort((a, b) => a - b), [300, 1_800]); await hungBindHarness.clock.runNext(300); await Promise.all([hungStart, hungShutdown]); assert.equal(hungBindHarness.children.length, 0); assert.equal( hungBindHarness.calls.filter((call) => call.args[3] === "session-end").length, 0, "session-end cannot run without a completed bind token", ); 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() { const harness = createHarness({ onSubmit(content) { if (content === '') throw new Error("OMP prompt flow rejected input"); }, }); await harness.emit("session_start"); harness.children[0].stdout.emit( "data", 'onetwo', ); 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); assert.deepEqual(harness.submitted, ['', '']); await harness.emit("agent_start"); await harness.emit("agent_end", { messages: [ { role: "user", content: '' }, 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(), []); } async function testFailedIdleRecoveryFailsClosed() { let idleCalls = 0; const harness = createHarness({ onSubmit() { throw new Error("OMP prompt flow rejected input"); }, onRun(call) { if (call.args[0] === "api" && call.args[3] === "state" && call.args[4] === "idle") { idleCalls += 1; if (idleCalls === 2) throw new Error("state channel unavailable"); } }, }); await harness.emit("session_start"); 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( harness.errors.some(({ message }) => message.includes("could not restore idle state after a failed submission"), ), ); assert.deepEqual(harness.clock.delays(), []); } // [unit->REQ-OMP-LISTENER-FAIL-CLOSED] async function testListenerRestartExhaustion() { const harness = createHarness({ restartDelaysMs: [5, 10] }); await harness.emit("session_start"); const first = harness.children[0]; first.emit("close", 7); assert.deepEqual(harness.clock.delays(), [5]); await harness.clock.runNext(5); const second = harness.children[1]; second.stdout.emit("data", 'half'); second.emit("error", new Error("listener crashed")); second.emit("close", 8); await flush(); assert.deepEqual(harness.clock.delays(), [10], "error plus close schedules one restart"); @@ -814,108 +734,50 @@ async function testListenerStableIntervalResetsRetries() { [10], "a parsed event followed by an immediate crash remains in the consecutive crash loop", ); await harness.clock.runNext(10); const stable = harness.children[2]; assert.deepEqual(harness.clock.delays(), [20]); await harness.clock.runNext(20); stable.emit("close", 3); assert.deepEqual( harness.clock.delays(), [5], "a listener surviving the stable interval resets the next retry to attempt one", ); assert.match( harness.notifications.filter(({ type }) => type === "warning").at(-1).message, /restarting 1\/2 in 5ms/, ); await harness.clock.runNext(5); assert.equal(harness.children.length, 4); await harness.emit("session_shutdown"); 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(); let endAttempts = 0; const harness = createHarness({ restartDelaysMs: [], sessionEndRetryDelaysMs: [5], onRun(call) { if (call.args[0] === "api" && call.args[3] === "session-end") { endAttempts += 1; if (endAttempts === 1) return firstEnd.promise; } }, }); await harness.emit("session_start"); harness.children[0].emit("close", 12); await flush(); assert.equal( harness.calls.filter((call) => call.args[3] === "session-end").length, 1, ); firstEnd.reject(new Error("transient teardown failure")); await flush(); assert.equal(harness.shutdowns, 0, "fatal close must wait for the bounded teardown retry"); @@ -942,76 +804,76 @@ async function testSessionEndRetriesAfterTransientFailure() { assert.deepEqual(harness.clock.delays(), []); } async function testHumanBusyFailureFailsClosed() { const harness = createHarness({ onRun(call) { if (call.args[0] === "api" && call.args[3] === "state" && call.args[4] === "busy") { throw new Error("state channel unavailable"); } }, }); await harness.emit("session_start"); await harness.emit("agent_start"); assert.equal(harness.shutdowns, 1); assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 1); assert.ok( harness.errors.some(({ message }) => message.includes("could not mark the endpoint busy")), ); assert.equal(harness.children[0].kills, 1); assert.deepEqual(harness.clock.delays(), []); } // [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]; listener.stdout.emit( "data", 'onetwo', ); await flush(); await harness.emit("agent_start"); await harness.emit("agent_end", { messages: [ { role: "user", content: '' }, assistantMessage("done"), ], }); assert.deepEqual(harness.clock.delays(), [0]); await harness.emit("session_shutdown"); 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"); } async function testRunSptRejectsStdinErrorsAndHungCommands() { const epipeClock = new FakeClock(); const epipeChild = new FakeChild(); const epipe = Object.assign(new Error("write EPIPE"), { code: "EPIPE" }); epipeChild.stdin.onEnd = () => { epipeChild.stdin.emit("error", epipe); return false; }; await assert.rejects( runSpt(["send", "peer", "--from", "omp-agent"], "reply", { clearTimeout: epipeClock.clearTimeout.bind(epipeClock), commandTimeoutMs: 20, killForceMs: 4, killGraceMs: 3, setTimeout: epipeClock.setTimeout.bind(epipeClock), spawnProcess: () => epipeChild, }), (error) => error === epipe && error.code === "EPIPE", ); assert.deepEqual(epipeChild.killSignals, ["SIGTERM"]); assert.deepEqual(epipeClock.delays(), []); @@ -1133,342 +995,123 @@ async function testProtocolCorruptionFailsClosed() { ); assert.equal(harness.children[0].kills, 1); assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 1); assert.deepEqual(harness.clock.delays(), []); await harness.emit("session_shutdown"); return harness; } const truncatedA = 'truncatedvalid'; const nested = await failProtocol(truncatedA, /nested EVENT/); assert.deepEqual( commandCalls(nested, "send").map((call) => call.args[1]), [], "the later valid b frame must not be merged into or consumed as a", ); await failProtocol('missing sender', /missing EVENT from/); await failProtocol('bad attrs', /malformed EVENT attributes/); await failProtocol('never closes'.padEnd(80, "x"), /buffer exceeded/, { listenerBufferLimit: 64, }); } // [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`, ); const harness = createHarness({ acceptedQueueLimit: 2, restartDelaysMs: [], }); await harness.emit("session_start"); harness.children[0].stdout.emit("data", frames.join("")); await flush(); 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"), ), ); assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 1); assert.deepEqual(harness.clock.delays(), []); await harness.emit("session_shutdown"); const byteFirst = 'x'; const byteOverflow = '😀'; const byteHarness = createHarness({ acceptedBytesLimit: Buffer.byteLength(byteFirst, "utf8") + byteOverflow.length, acceptedQueueLimit: 10, restartDelaysMs: [], }); 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] async function testShutdownFallbackStaysBelowHostCap() { const never = new Promise(() => {}); const harness = createHarness({ 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] // [unit->REQ-PARITY-LIVE-ACTIVATION] async function testNativeActivationCommandsAndErrors() { const inert = createHarness({ id: null }); assert.deepEqual([...inert.commands.keys()], ["ready", "live"]); assert.ok(inert.tools.has("spt_checkpoint")); await inert.emit("session_start"); assert.deepEqual(inert.calls, [], "an ordinary OMP session must remain lifecycle-inert"); assert.equal( await inert.emit("session_before_switch", { reason: "new" }), undefined, "an unbound extension must not block native session changes", ); await inert.commands.get("ready").handler("--auto", inert.ctx); await inert.commands.get("live").handler("two identities", inert.ctx); assert.deepEqual(inert.calls, [], "invalid activation syntax must not bind or guess"); assert.ok( inert.notifications.some(({ message }) => message.includes("supported only by `/live`")), ); assert.ok(inert.notifications.some(({ message }) => message.includes("Usage: /live"))); await inert.commands.get("ready").handler("ready-one", inert.ctx); @@ -1894,179 +1537,173 @@ async function testActiveTurnBoundaryDeliveryAndFallback() { assert.equal( boundary.messages.at(-1).content, `\n\n${firstEnvelope}`, "the first accepted message must enter at the next model boundary", ); const afterToolBoundary = await harness.emit("context", { messages: [ { role: "user", content: "operator prompt" }, assistantMessage("", { stopReason: "toolUse", toolCalls: [{ name: "read" }] }), { role: "toolResult", content: "tool output" }, ], }); assert.equal( afterToolBoundary.messages.at(-1).content, `\n\n${firstEnvelope}`, "ephemeral context must be re-injected before every later model continuation", ); await harness.emit("agent_end", { messages: [ { role: "user", content: "operator prompt" }, assistantMessage("", { stopReason: "toolUse", toolCalls: [{ name: "read" }] }), { role: "toolResult", content: "tool output" }, 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( harness.submitted, [''], "a later accepted message must preserve order and fall back to an ordinary next turn", ); assertNoAgentManagedPoll(harness); await harness.emit("session_shutdown"); } // [unit->REQ-HAZARD-ABNORMAL-TURN-RECEIVABILITY] async function testAbnormalTurnsRestoreReceivability() { for (const [label, completionEvent] of [ ["cancelled", "session_stop"], ["interrupted", "session_stop"], ["failed", "agent_end"], ]) { const harness = createHarness(); await harness.emit("session_start"); await harness.emit("agent_start"); harness.children[0].stdout.emit( "data", `work`, ); await flush(); const boundary = await harness.emit("context", { 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( "data", `next`, ); await flush(); assert.deepEqual(harness.submitted, [``]); await harness.emit("session_shutdown"); } } 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( "data", `work`, ); await flush(); await harness.emit("agent_start"); await harness.emit("agent_end", { messages: [ { role: "user", content: `` }, assistantMessage("@partial output", { stopReason, errorMessage: stopReason === "error" ? "provider unavailable" : undefined, }), ], }); 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 }), ); const harness = createHarness(); await harness.emit("session_start"); harness.children[0].stdout.emit( "data", 'work', ); await flush(); await harness.emit("agent_start"); await harness.emit("context", { messages: [...highHistory, { role: "user", content: '' }], }); const compactedHistory = [ { role: "user", content: "[auto-compaction summary]" }, highHistory[12], ]; await harness.emit("context", { messages: compactedHistory }); await harness.emit("agent_end", { messages: [ ...compactedHistory, 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(); await stale.emit("session_start"); stale.children[0].stdout.emit( "data", 'work', ); await flush(); await stale.emit("agent_start"); await stale.emit("context", { messages: [...highHistory, { role: "user", content: '' }], }); 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"); } // [unit->REQ-PARITY-PEER-SHORTFORM] async function testPeerShortformParsingAndDispatch() { assert.deepEqual( parsePeerShortforms("Before @ after @").map( ({ targets, body }) => ({ targets, body }), ), [ { targets: ["alpha", "beta"], body: "hello there" }, { targets: ["gamma"], body: "second\nline" }, ], ); assert.deepEqual( parsePeerShortforms( "ordinary @alice; `@`\n```\n@\n```\n@", ), [], ); const alphaGate = deferred(); const harness = createHarness({ onRun(call) { if (call.args[0] === "send" && call.args[1] === "alpha") return alphaGate.promise; @@ -2256,56 +1893,54 @@ async function testNativeCheckpointTool() { assert.equal(launchedRefusal.details.reason, "not-live"); assert.deepEqual(launchedReady.compactions, []); await launchedReady.emit("session_shutdown"); const failed = createHarness({ onCompact() { throw new Error("native compaction cancelled"); }, onRun(call) { if (call.args[0] === "--json" && call.args[2] === "endpoint-info") { return JSON.stringify({ endpoint_type: "live_agent" }); } }, }); await failed.emit("session_start"); const failure = await failed.tools .get("spt_checkpoint") .execute("tool-3", {}, undefined, undefined, failed.ctx); assert.equal(failure.isError, true); assert.match(failure.content[0].text, /native compaction cancelled/); assert.deepEqual(failed.sentMessages, [], "a failed reset must never queue a false wake"); await failed.emit("session_shutdown"); } -await testParsingAndReplies(); +await testParsing(); 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(); await testPlatformAwareActivationCandidatePaths(); await testExplicitLiveAutoResume(); await testStartupBriefHintsAndUpdateNotices(); await testActiveTurnBoundaryDeliveryAndFallback(); await testAbnormalTurnsRestoreReceivability(); await testCompletionStopReasonGatesSideEffects(); -await testCompactionSafeReplyCorrelation(); +await testCompactionPreservesLocalAssistantOutput(); await testPeerShortformParsingAndDispatch(); await testNativeCheckpointTool(); console.log("OMP-EXTENSION OK");