diff --git a/adapter/omp-spt.toml b/adapter/omp-spt.toml index b0e72ab..fd4b0d0 100644 --- a/adapter/omp-spt.toml +++ b/adapter/omp-spt.toml @@ -37,6 +37,14 @@ loads the packaged extension fresh on each endpoint bringup. +# OMP's native extension is the non-disruptive injection seam in both activity +# states. Busy active-only traffic is pulled at context boundaries; idle and +# unrestricted listener traffic uses the same native user-message API. +# [impl->REQ-OMP-CORE-DELIVERY] +[inject] +activity = ["hook"] +idle = ["hook"] + [identity] # OMP reports the bound session id after spawn; the process-tree fallback is # the native host executable, never the adapter helper. diff --git a/adapter/strings/omp-spt.mjs b/adapter/strings/omp-spt.mjs index 3849ce0..0e7d9fd 100644 --- a/adapter/strings/omp-spt.mjs +++ b/adapter/strings/omp-spt.mjs @@ -16,9 +16,12 @@ export function endpointDisplayName(id, node, project) { : `${endpoint} @ ${nodeName}`; } -export function endpointInlineStatus(id, node, project, theme) { +export function endpointInlineStatus(id, node, project, theme, recovering = false) { const text = endpointDisplayName(id, node, project); - return theme?.fg ? theme.fg("statusLineModel", text) : text; + const identity = theme?.fg ? theme.fg("statusLineModel", text) : text; + if (!recovering) return identity; + const warning = " · comms recovering..."; + return identity + (theme?.fg ? theme.fg("warning", warning) : warning); } export function decodeBody(body) { @@ -218,7 +221,7 @@ export function formatInboundEnvelope(envelope) { } function injectEnvelope(messages, item) { - const index = messages.findLastIndex( + const index = messages.findIndex( (message) => message?.role === "user" && messageText(message) === item.stub, ); if (index < 0) return messages; @@ -776,11 +779,11 @@ export function createOmpSpt(overrides = {}) { let listenerRestartCount = 0; let listenerStableTimer; let restartTimer; - let dispatchTimer; + let stateRetryTimer; + let stateRetryAttempt = 0; let bindPromise; let agentActive = false; let desiredState = "idle"; - let dispatching = false; let turnCompletionPromise; let turnAssistantBaseline = captureAssistantBaseline([]); let turnContextObserved = false; @@ -790,7 +793,6 @@ export function createOmpSpt(overrides = {}) { let shutdownDeadlineExpired = false; const activeCommands = new Map(); const retryWaiters = new Set(); - let current; let stopping = false; let ui; let titleTimer; @@ -821,19 +823,47 @@ export function createOmpSpt(overrides = {}) { }; let runtimeCtx; let endpointState; - let stateOperation = Promise.resolve(); let endPromise; let fatalPromise; let teardownPromise; let acceptedBytes = 0; let overflowItem; - const queue = []; + const pendingListener = []; + const commsFailures = new Set(); const logError = (message, error) => { pi.logger.error(message, { error: errorSummary(error) }); ui?.notify(`${message}: ${errorSummary(error)}`, "error"); }; + function renderEndpointStatus() { + if (!id || !ui) return; + ui.setStatus( + "omp-spt", + endpointInlineStatus( + id, + env.OMP_SPT_NODE, + env.OMP_SPT_PROJECT, + ui.theme, + commsFailures.size > 0, + ), + ); + } + + // [impl->REQ-OMP-COMMS-RECOVERY] + function markCommsFailure(component, message, error) { + const first = !commsFailures.has(component); + commsFailures.add(component); + pi.logger.error(message, { error: errorSummary(error) }); + if (first) ui?.notify(`${message}: ${errorSummary(error)}`, "warning"); + renderEndpointStatus(); + } + + function clearCommsFailure(component) { + if (!commsFailures.delete(component)) return; + renderEndpointStatus(); + } + function runCommand(args, input, options = {}) { if (shutdownDeadlineExpired) { return Promise.reject(new Error("omp-spt shutdown deadline expired")); @@ -1158,7 +1188,6 @@ export function createOmpSpt(overrides = {}) { sid = ctx.sessionManager.getSessionId(); id = nextId; activationType = type; - let bound = false; const operation = (async () => { const bindArgs = [ "api", @@ -1175,11 +1204,14 @@ export function createOmpSpt(overrides = {}) { const response = await runCommand(bindArgs); token = response.match(/\btoken=([^\s]+)/)?.[1]; if (!token) throw new Error("spt bind response did not include token="); - bound = true; })(); try { await bindPromise; - await syncDesiredState(); + try { + await syncDesiredState(); + } catch { + // Bind established the endpoint; communications now recover in place. + } if (stopping) { await teardownSession("OMP session shut down before initialization completed"); return false; @@ -1204,8 +1236,8 @@ export function createOmpSpt(overrides = {}) { return true; } catch (error) { ui.setStatus("omp-spt", "spt bind failed"); - if (options.fatal || bound) { - await failClosed(`omp-spt could not bind ${id}`, error); + if (options.fatal) { + await failActivation(`omp-spt could not bind ${id}`, error); return false; } pi.logger.error(`omp-spt could not bind ${id}`, { @@ -1409,30 +1441,61 @@ export function createOmpSpt(overrides = {}) { return ["--token", token]; } - function setState(state) { - if (!sid || !token || stopping) return Promise.resolve(); - const operation = stateOperation.catch(() => {}).then(async () => { - if (endpointState === state || stopping) return; + function scheduleStateRetry() { + if (stopping || stateRetryTimer !== undefined) return; + const index = Math.min(stateRetryAttempt, Math.max(0, restartDelaysMs.length - 1)); + const delay = restartDelaysMs[index] ?? 0; + stateRetryAttempt += 1; + stateRetryTimer = setTimer(() => { + stateRetryTimer = undefined; + if (stopping) return; + void setState(desiredState).catch(() => {}); + }, delay); + stateRetryTimer?.unref?.(); + } + + // [impl->REQ-OMP-COMMS-RECOVERY] + async function setState(state) { + if (!sid || !token || stopping) return; + if (endpointState === state) { + if (state === desiredState) clearCommsFailure("state"); + return; + } + try { await runCommand(["api", "--adapter", ADAPTER, "state", state, id, ...authArgs()]); endpointState = state; - }); - stateOperation = operation; - return operation; + if (state === desiredState) { + stateRetryAttempt = 0; + if (stateRetryTimer !== undefined) { + clearTimer(stateRetryTimer); + stateRetryTimer = undefined; + } + clearCommsFailure("state"); + } else { + markCommsFailure( + "state", + "omp-spt activity changed during state publication; reconciling", + new Error(`published stale ${state} state while current OMP state is ${desiredState}`), + ); + scheduleStateRetry(); + } + } catch (error) { + markCommsFailure("state", `omp-spt could not mark the endpoint ${state}`, error); + scheduleStateRetry(); + throw error; + } } async function syncDesiredState() { if (!bindPromise) return; await bindPromise; - while (!stopping && endpointState !== desiredState) { - await setState(desiredState); - } + await setState(desiredState); } function endSession() { if (!sid || !token) return Promise.resolve(); if (!endPromise) { const operation = (async () => { - await stateOperation.catch(() => {}); await runCommand(["api", "--adapter", ADAPTER, "session-end", id, ...authArgs()]); endpointState = undefined; })(); @@ -1495,9 +1558,9 @@ export function createOmpSpt(overrides = {}) { async function stopResources() { stopTitleAnimation(); - if (dispatchTimer !== undefined) { - clearTimer(dispatchTimer); - dispatchTimer = undefined; + if (stateRetryTimer !== undefined) { + clearTimer(stateRetryTimer); + stateRetryTimer = undefined; } if (restartTimer !== undefined) { clearTimer(restartTimer); @@ -1518,10 +1581,9 @@ export function createOmpSpt(overrides = {}) { } function releasePending() { - const pending = current ? [current, ...queue] : [...queue]; + const pending = [...pendingListener]; if (overflowItem) pending.push(overflowItem); - current = undefined; - queue.length = 0; + pendingListener.length = 0; overflowItem = undefined; for (const item of pending) releaseItem(item); } @@ -1579,15 +1641,14 @@ export function createOmpSpt(overrides = {}) { } } - // [impl->REQ-OMP-LISTENER-FAIL-CLOSED] - async function failClosed(message, error) { + async function failActivation(message, error) { if (stopping && shutdownMode) return teardownPromise ?? Promise.resolve(); if (fatalPromise) return fatalPromise; fatalPromise = (async () => { - ui?.setStatus("omp-spt", "spt failed"); + ui?.setStatus("omp-spt", "spt activation failed"); logError(message, error); try { - await teardownSession("endpoint stopped before your message could complete"); + await teardownSession("endpoint activation failed"); } catch (teardownError) { logError("omp-spt session teardown failed", teardownError); } @@ -1596,76 +1657,35 @@ export function createOmpSpt(overrides = {}) { return fatalPromise; } - function scheduleDispatch() { - if ( - stopping || - dispatching || - current || - queue.length === 0 || - dispatchTimer !== undefined - ) { - return; + // [impl->REQ-OMP-CORE-DELIVERY] + function submitListenerItem(item) { + if (stopping) return false; + item.stub ??= senderStub(item.from ?? "unknown"); + try { + pi.sendUserMessage(item.stub); + return true; + } catch (error) { + const index = pendingListener.indexOf(item); + if (index >= 0) pendingListener.splice(index, 1); + releaseItem(item); + logError("omp-spt could not submit your message to OMP", error); + return false; } - dispatchTimer = setTimer(() => { - dispatchTimer = undefined; - void dispatchNext().catch((error) => { - if (!stopping) return failClosed("omp-spt dispatch failed", error); - }); - }, 0); - dispatchTimer?.unref?.(); } - async function rejectItem(item, reason, error) { - if (stopping) return; - logError(`omp-spt ${reason}`, error); - if (current === item) { - current = undefined; - releaseItem(item); - } - if (!stopping) { - desiredState = "idle"; - try { - await setState("idle"); - } catch (stateError) { - await failClosed( - "omp-spt could not restore idle state after a failed submission", - stateError, - ); - } - } + function resubmitUnobservedListenerItems() { + for (const item of pendingListener) submitListenerItem(item); } - // [impl->REQ-OMP-EXTENSION-CUSTODY] - async function dispatchNext() { - if (stopping || dispatching || current || queue.length === 0) return; - dispatching = true; - const item = queue.shift(); - current = item; - try { - try { - desiredState = "busy"; - await setState("busy"); - } catch (error) { - await rejectItem(item, "could not accept your message", error); - return; - } - if (stopping) return; - item.stub = senderStub(item.from ?? "unknown"); - item.submitted = true; - item.hiddenSubmitted = false; - try { - pi.sendUserMessage(item.stub); - } catch (error) { - item.submitted = false; - await rejectItem(item, "could not submit your message to OMP", error); - } - } finally { - dispatching = false; - scheduleDispatch(); - } + function admitOverflowItem() { + if (!overflowItem || pendingListener.length >= acceptedQueueLimit) return; + const item = overflowItem; + overflowItem = undefined; + pendingListener.push(item); + submitListenerItem(item); } - // [impl->REQ-OMP-LISTENER-FAIL-CLOSED] + // [impl->REQ-OMP-COMMS-RECOVERY] function handleListenerDeath(reason) { listenerBuffer = ""; if (listenerStableTimer !== undefined) { @@ -1673,17 +1693,15 @@ export function createOmpSpt(overrides = {}) { listenerStableTimer = undefined; } if (stopping || restartTimer !== undefined) return; - if (listenerRestartCount >= restartDelaysMs.length) { - void failClosed("omp-spt listener restart budget exhausted", reason); - return; - } const attempt = listenerRestartCount + 1; - const delay = restartDelaysMs[listenerRestartCount]; + const index = Math.min(listenerRestartCount, Math.max(0, restartDelaysMs.length - 1)); + const delay = restartDelaysMs[index] ?? 0; listenerRestartCount = attempt; - const message = `omp-spt listener stopped; restarting ${attempt}/${restartDelaysMs.length} in ${delay}ms`; - pi.logger.error(message, { error: errorSummary(reason) }); - ui?.setStatus("omp-spt", `spt reconnecting (${attempt}/${restartDelaysMs.length})`); - ui?.notify(message, "warning"); + markCommsFailure( + "listener", + `omp-spt listener stopped; retrying in ${delay}ms`, + reason, + ); restartTimer = setTimer(() => { restartTimer = undefined; startListener(); @@ -1722,7 +1740,10 @@ export function createOmpSpt(overrides = {}) { if (listenerStableMs !== undefined) { listenerStableTimer = setTimer(() => { listenerStableTimer = undefined; - if (listener === child && !stopping) listenerRestartCount = 0; + if (listener === child && !stopping) { + listenerRestartCount = 0; + clearCommsFailure("listener"); + } }, listenerStableMs); listenerStableTimer?.unref?.(); } @@ -1732,12 +1753,11 @@ export function createOmpSpt(overrides = {}) { if (listener !== child || stopping) return; listenerBuffer += String(chunk); if (listenerBuffer.length > listenerBufferLimit) { - void failClosed( - "omp-spt listener protocol corruption", - protocolError( - `EVENT buffer exceeded ${listenerBufferLimit} characters without a complete drain`, - ), + const error = protocolError( + `EVENT buffer exceeded ${listenerBufferLimit} characters without a complete drain`, ); + markCommsFailure("listener", "omp-spt listener protocol corruption", error); + died(error, false); return; } while (!stopping) { @@ -1747,32 +1767,40 @@ export function createOmpSpt(overrides = {}) { }); listenerBuffer = drained.rest; if (drained.error) { - void failClosed("omp-spt listener protocol corruption", drained.error); + markCommsFailure( + "listener", + "omp-spt listener protocol corruption", + drained.error, + ); + died(drained.error, false); return; } if (drained.events.length === 0) break; const event = drained.events[0]; - const acceptedCount = queue.length + (current ? 1 : 0); + const acceptedCount = pendingListener.length + (overflowItem ? 1 : 0); const eventBytes = Buffer.byteLength(event.envelope, "utf8"); + event.acceptedBytes = eventBytes; + event.accounted = true; + acceptedBytes += eventBytes; if ( acceptedCount >= acceptedQueueLimit || - acceptedBytes + eventBytes > acceptedBytesLimit + acceptedBytes > acceptedBytesLimit ) { overflowItem = event; - void failClosed( - "omp-spt inbound custody capacity exceeded", - new Error( - `accepted queue limit is ${acceptedQueueLimit} messages and ${acceptedBytesLimit} bytes`, - ), + const error = new Error( + `accepted listener limit is ${acceptedQueueLimit} messages and ${acceptedBytesLimit} bytes`, + ); + markCommsFailure( + "listener", + "omp-spt inbound listener capacity exceeded", + error, ); + died(error, false); return; } - event.acceptedBytes = eventBytes; - event.accounted = true; - acceptedBytes += eventBytes; - queue.push(event); + pendingListener.push(event); + submitListenerItem(event); } - if (!current && !dispatching) void dispatchNext(); }); child.stderr.on("data", (chunk) => pi.logger.debug("omp-spt listener", { output: String(chunk).trim() }), @@ -1782,10 +1810,7 @@ export function createOmpSpt(overrides = {}) { const status = signal ? `signal ${signal}` : code; died(new Error(`spt api listen exited ${status}`), true); }); - ui?.setStatus( - "omp-spt", - endpointInlineStatus(id, env.OMP_SPT_NODE, env.OMP_SPT_PROJECT, ui?.theme), - ); + renderEndpointStatus(); } // [impl->REQ-OMP-NATIVE-TUI] @@ -1823,7 +1848,8 @@ export function createOmpSpt(overrides = {}) { // [impl->REQ-OMP-MESSAGE-CONTEXT] // [impl->REQ-PARITY-SAFE-BOUNDARY-DELIVERY] - pi.on("context", (event) => { + // [impl->REQ-OMP-CORE-DELIVERY] + pi.on("context", async (event) => { const baseline = captureAssistantBaseline(event.messages); if (agentActive) { if (!turnContextObserved) { @@ -1834,17 +1860,63 @@ export function createOmpSpt(overrides = {}) { observedAssistantBaseline = baseline; } let messages = event.messages; - if (current?.submitted && !current.hiddenSubmitted) { - current.assistantBaseline ??= baseline; - messages = injectEnvelope(messages, current); + for (const item of [...pendingListener]) { + const injected = injectEnvelope(messages, item); + if (injected === messages) continue; + messages = injected; + const index = pendingListener.indexOf(item); + if (index >= 0) pendingListener.splice(index, 1); + releaseItem(item); + } + admitOverflowItem(); + + if (activated && !stopping && (agentActive || desiredState === "busy")) { + try { + await setState("busy"); + } catch { + return messages === event.messages ? undefined : { messages }; + } + try { + const polled = await runCommand([ + "api", + "--adapter", + ADAPTER, + "poll", + id, + "--include-deferred", + ...authArgs(), + ]); + clearCommsFailure("poll"); + if (String(polled).trim()) { + messages = [ + ...messages, + { + role: "custom", + customType: "spt-event", + content: String(polled), + display: false, + attribution: "user", + timestamp: Date.now(), + }, + ]; + } + } catch (error) { + markCommsFailure("poll", "omp-spt could not poll active messages", error); + } } if (messages !== event.messages) return { messages }; }); // [impl->REQ-PARITY-STARTUP-BRIEF] // [impl->REQ-PARITY-TARGETED-HINTS] // [impl->REQ-PARITY-UPDATE-NOTICE] - pi.on("before_agent_start", (event) => { + pi.on("before_agent_start", async (event) => { if (!activated || !id || stopping) return; + desiredState = "busy"; + try { + await setState("busy"); + } catch { + // Local model work proceeds while the state retry loop restores routing truth. + } const additions = []; if (startupBriefPending) { startupBriefPending = false; @@ -1899,24 +1971,17 @@ export function createOmpSpt(overrides = {}) { if (stopping) return; const messages = event.messages ?? []; const currentTurnAssistant = assistantAfterBaseline(messages, turnAssistantBaseline); - const completed = current; const currentTurnReply = successfulAssistant(currentTurnAssistant) ? messageText(currentTurnAssistant) : ""; - if (current === completed) { - current = undefined; - releaseItem(completed); - } observedAssistantBaseline = captureAssistantBaseline(messages); try { await setState("idle"); - } catch (error) { - if (stopping) return; - await failClosed("omp-spt could not mark the endpoint idle", error); - return; + } catch { + // Local completion wins; the retry loop converges on current idle truth. } await dispatchShortforms(currentTurnReply); - if (!stopping) await dispatchNext(); + if (!stopping) resubmitUnobservedListenerItems(); } pi.on("agent_start", async () => { @@ -1929,12 +1994,12 @@ export function createOmpSpt(overrides = {}) { desiredState = "busy"; try { await syncDesiredState(); - } catch (error) { - if (!stopping) await failClosed("omp-spt could not mark the endpoint busy", error); + } catch { + // before_agent_start already opened recovery; never terminate local work. } }); - // [impl->REQ-OMP-EXTENSION-CUSTODY] + // [impl->REQ-OMP-CORE-DELIVERY] pi.on("agent_end", (event) => { turnCompletionPromise ??= completeTurn(event); return turnCompletionPromise; diff --git a/tests/manifest-shortcut.sh b/tests/manifest-shortcut.sh index 011f670..dc4fde8 100644 --- a/tests/manifest-shortcut.sh +++ b/tests/manifest-shortcut.sh @@ -131,12 +131,18 @@ if grep -Eq '^[[:space:]]*transport[[:space:]]*=[[:space:]]*"gh"' "$MANIFEST"; t if grep -Eq '^[[:space:]]*message[[:space:]]*=' "$MANIFEST"; then echo "ok [update].message present"; else echo "FAIL [update] has no message field"; fail=1; fi if grep -Eq 'endpoint run --adapter omp-spt --id --create --start' "$MANIFEST"; then echo "FAIL fresh-endpoint update notice still uses --start (hides the harness PTY)"; fail=1; else echo "ok fresh-endpoint update notice omits --start (attach-default PTY)"; fi +# [unit->REQ-OMP-CORE-DELIVERY] +inject_activity=$(field_of inject '^[[:space:]]*activity[[:space:]]*=') +case "$inject_activity" in + 'activity = ["hook"]') echo "ok [inject] routes idle and active custody through OMP hooks" ;; + *) echo "FAIL [inject].activity must be exactly [\"hook\"]: [$inject_activity]"; fail=1 ;; +esac + # ── retired foreign-harness seams stay absent ──────────────────────────────────────────────────── # [unit->REQ-OMP-NATIVE-TUI] for stale in \ '^\[update\.post\]' \ '^\[hooks\.' \ - '^\[inject\]' \ '^\[message-idle-translation-binary\]' \ '^\[env\.SPT_INJECT_VERIFY_ECHO\]' \ '^\[env\.CLAUDE_CONFIG_DIR\]' \ @@ -147,7 +153,7 @@ for stale in \ fail=1 fi done -if [ "$fail" -eq 0 ]; then echo "ok retired update/hook/inject/translation/env surfaces absent"; fi +if [ "$fail" -eq 0 ]; then echo "ok retired update/hook/translation/env surfaces absent"; fi # OMP model routing is native; the adapter ships no foreign profile overlay. if grep -Eq '^\[profiles\.' "$MANIFEST"; then echo "FAIL a shipped [profiles.*] table lingers"; fail=1; else echo "ok no shipped profile overlays"; fi diff --git a/tests/omp-extension.mjs b/tests/omp-extension.mjs index d9c242f..ab2c225 100644 --- a/tests/omp-extension.mjs +++ b/tests/omp-extension.mjs @@ -438,7 +438,7 @@ async function testEndpointSessionNameAndAnimatedWindowTitle() { assert.equal(harness.intervals[0].active, false); } -// [unit->REQ-OMP-EXTENSION-CUSTODY] +// [unit->REQ-OMP-CORE-DELIVERY] // [unit->REQ-OMP-SESSION-IMMUTABLE] // [unit->REQ-OMP-MESSAGE-CONTEXT] // [unit->REQ-OMP-NATIVE-TUI] @@ -457,33 +457,17 @@ async function testLifecycleCustodyAndContext() { ]); await harness.emit("session_start"); - assert.deepEqual(harness.calls[0], { - args: [ - "api", - "--adapter", - "omp-spt", - "bind", - "omp-agent", - "--set-session-id", - "session-1", - "--subnet", - "mesh-a", - ], - input: undefined, - }); - assert.deepEqual(harness.calls[1].args, [ + assert.deepEqual(harness.calls[0].args.slice(0, 7), [ "api", "--adapter", "omp-spt", - "state", - "idle", + "bind", "omp-agent", - "--token", - "token-123", + "--set-session-id", + "session-1", ]); - assert.equal(harness.children.length, 1); - assert.equal(harness.children[0].binary, "spt-test"); - assert.deepEqual(harness.children[0].args, [ + assert.equal(stateCalls(harness).at(-1).args[4], "idle"); + assert.deepEqual(harness.children[0].args.slice(0, 7), [ "api", "--adapter", "omp-spt", @@ -491,93 +475,56 @@ async function testLifecycleCustodyAndContext() { "omp-agent", "--session-id", "session-1", - "--subnet", - "mesh-a", ]); for (const reason of ["new", "resume", "fork", "handoff"]) { assert.deepEqual(await harness.emit("session_before_switch", { reason }), { cancel: true }); - assert.ok( - harness.notifications.some(({ message }) => - message.includes(`${reason} session switch`), - ), - ); } assert.deepEqual(await harness.emit("session_before_branch"), { cancel: true }); - assert.ok( - harness.notifications.some(({ message }) => message.includes("session branch")), - ); const aliceEnvelope = 'hello<world
line'; const bobEnvelope = 'second'; harness.children[0].stdout.emit("data", `${aliceEnvelope}${bobEnvelope}`); await flush(); - assert.deepEqual(harness.submitted, ['']); - assert.deepEqual(harness.submittedDeliveries, [undefined]); - assert.deepEqual(harness.sentMessages, []); assert.deepEqual( - stateCalls(harness).map((call) => call.args[4]), - ["idle", "busy"], + harness.submitted, + ['', ''], + "listener events surface independently as they arrive", ); - const originalMessages = [{ role: "user", content: '' }]; - assert.deepEqual(await harness.emit("context", { messages: originalMessages }), { - messages: [ - { - role: "user", - content: `\n\n${formatInboundEnvelope(aliceEnvelope)}`, - }, - ], - }); - + await harness.emit("before_agent_start", { prompt: "listener wake", systemPrompt: [] }); await harness.emit("agent_start"); - assert.deepEqual( - 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", { + const boundary = await harness.emit("context", { messages: [ { role: "user", content: '' }, - aliceReply, - ], - }); - assert.deepEqual(harness.clock.delays(), []); - assert.deepEqual(harness.submitted, ['', '']); - assert.deepEqual(harness.submittedDeliveries, [undefined, undefined]); - await harness.emit("agent_start"); - await harness.emit("agent_end", { - messages: [ - aliceReply, { role: "user", content: '' }, ], }); - - const outcomes = commandCalls(harness, "send"); - assert.deepEqual( - outcomes, - [], - "ordinary assistant output must never be forwarded to a peer", + assert.equal( + boundary.messages[0].content, + `\n\n${formatInboundEnvelope(aliceEnvelope)}`, ); + assert.equal( + boundary.messages[1].content, + `\n\n${formatInboundEnvelope(bobEnvelope)}`, + ); + assert.equal(harness.calls.filter((call) => call.args[3] === "poll").length, 1); + + await harness.emit("agent_end", { + messages: [...boundary.messages, assistantMessage("local result")], + }); + assert.deepEqual(commandCalls(harness, "send"), []); assert.deepEqual( stateCalls(harness).map((call) => call.args[4]), - ["idle", "busy", "idle", "busy", "idle"], + ["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.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-CORE-DELIVERY] // [unit->REQ-HAZARD-ABNORMAL-TURN-RECEIVABILITY] async function testLocalAssistantOutputDoesNotReplyToPeer() { const harness = createHarness(); @@ -631,8 +578,8 @@ async function testDeferredBindLifecycleSerialization() { 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", + ["busy", "busy"], + "lifecycle callbacks publish current truth directly without replaying stale idle state", ); assert.equal(busyHarness.children.length, 1); await busyHarness.emit("session_shutdown"); @@ -690,7 +637,43 @@ async function testDeferredBindLifecycleSerialization() { } -// [unit->REQ-OMP-EXTENSION-CUSTODY] +// [unit->REQ-OMP-COMMS-RECOVERY] +async function testStateReconciliationUsesLatestActivity() { + const busyState = deferred(); + let heldBusy = false; + const harness = createHarness({ + onRun(call) { + if ( + !heldBusy && + call.args[0] === "api" && + call.args[3] === "state" && + call.args[4] === "busy" + ) { + heldBusy = true; + return busyState.promise; + } + }, + }); + await harness.emit("session_start"); + const starting = harness.emit("agent_start"); + await flush(); + await harness.emit("agent_end", { messages: [assistantMessage("done")] }); + busyState.resolve(""); + await starting; + + assert.deepEqual(harness.clock.delays(), [5]); + assert.match(harness.statuses.at(-1).text, /comms recovering/); + await harness.clock.runNext(5); + assert.deepEqual( + stateCalls(harness).map((call) => call.args[4]), + ["idle", "busy", "idle"], + "a late busy completion is corrected to the latest idle truth", + ); + assert.doesNotMatch(harness.statuses.at(-1).text, /comms recovering/); + await harness.emit("session_shutdown"); +} + +// [unit->REQ-OMP-CORE-DELIVERY] async function testSubmissionFailureAdvancesQueue() { const harness = createHarness({ onSubmit(content) { @@ -711,10 +694,11 @@ async function testSubmissionFailureAdvancesQueue() { [], "a rejected local submission must not message the peer implicitly", ); - assert.deepEqual(harness.clock.delays(), [0]); - - await harness.clock.runNext(0); + assert.deepEqual(harness.clock.delays(), []); assert.deepEqual(harness.submitted, ['', '']); + assert.ok( + harness.errors.some(({ message }) => message.includes("could not submit your message")), + ); await harness.emit("agent_start"); await harness.emit("agent_end", { messages: [ @@ -731,12 +715,10 @@ async function testSubmissionFailureAdvancesQueue() { assert.deepEqual(harness.clock.delays(), []); } +// [unit->REQ-OMP-COMMS-RECOVERY] 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; @@ -745,21 +727,21 @@ async function testFailedIdleRecoveryFailsClosed() { }, }); await harness.emit("session_start"); - harness.children[0].stdout.emit("data", 'one'); - await flush(); + await harness.emit("agent_start"); + await harness.emit("agent_end", { messages: [assistantMessage("local result")] }); - 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(), []); + assert.equal(harness.shutdowns, 0); + assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 0); + assert.deepEqual(harness.clock.delays(), [5]); + assert.match(harness.statuses.at(-1).text, /comms recovering/); + + await harness.clock.runNext(5); + assert.equal(idleCalls, 3); + assert.doesNotMatch(harness.statuses.at(-1).text, /comms recovering/); + await harness.emit("session_shutdown"); } -// [unit->REQ-OMP-LISTENER-FAIL-CLOSED] +// [unit->REQ-OMP-COMMS-RECOVERY] async function testListenerRestartExhaustion() { const harness = createHarness({ restartDelaysMs: [5, 10] }); await harness.emit("session_start"); @@ -781,23 +763,16 @@ async function testListenerRestartExhaustion() { await flush(); assert.equal(harness.children.length, 3); - assert.equal(harness.shutdowns, 1); - assert.ok( - harness.errors.some(({ message }) => message.includes("listener restart budget exhausted")), - ); - assert.ok( - harness.notifications.some(({ message, type }) => - type === "error" && message.includes("listener restart budget exhausted"), - ), - ); - assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 1); - assert.deepEqual(harness.clock.delays(), []); + assert.equal(harness.shutdowns, 0); + assert.deepEqual(harness.clock.delays(), [10]); + assert.match(harness.statuses.at(-1).text, /comms recovering/); + await harness.clock.runNext(10); + assert.equal(harness.children.length, 4, "listener retries indefinitely at capped backoff"); await harness.emit("session_shutdown"); assert.equal( harness.calls.filter((call) => call.args[3] === "session-end").length, 1, - "fatal shutdown and lifecycle shutdown share one teardown", ); } @@ -838,7 +813,7 @@ async function testListenerStableIntervalResetsRetries() { ); assert.match( harness.notifications.filter(({ type }) => type === "warning").at(-1).message, - /restarting 1\/2 in 5ms/, + /retrying in 5ms/, ); await harness.clock.runNext(5); @@ -848,75 +823,33 @@ async function testListenerStableIntervalResetsRetries() { } -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"); - assert.deepEqual(harness.clock.delays(), [5]); - assert.ok( - harness.errors.some( - ({ message, details }) => - message.includes("session teardown failed; retrying") && - details.error.includes("transient teardown failure"), - ), - ); - - await harness.clock.runNext(5); - await flush(); - const shutdown = harness.emit("session_shutdown"); - await shutdown; - await flush(); - assert.equal( - harness.calls.filter((call) => call.args[3] === "session-end").length, - 2, - "normal fatal teardown may retry before bounded lifecycle shutdown begins", - ); - assert.equal(harness.shutdowns, 1); - assert.deepEqual(harness.clock.delays(), []); -} +// [unit->REQ-OMP-COMMS-RECOVERY] async function testHumanBusyFailureFailsClosed() { + let busyCalls = 0; const harness = createHarness({ onRun(call) { if (call.args[0] === "api" && call.args[3] === "state" && call.args[4] === "busy") { - throw new Error("state channel unavailable"); + busyCalls += 1; + if (busyCalls === 1) 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(), []); + assert.equal(harness.shutdowns, 0); + assert.equal(harness.children[0].kills, 0); + assert.deepEqual(harness.clock.delays(), [5]); + assert.match(harness.statuses.at(-1).text, /comms recovering/); + await harness.clock.runNext(5); + assert.equal(busyCalls, 2); + assert.doesNotMatch(harness.statuses.at(-1).text, /comms recovering/); + await harness.emit("session_shutdown"); } -// [unit->REQ-OMP-EXTENSION-CUSTODY] -// [unit->REQ-OMP-LISTENER-FAIL-CLOSED] +// [unit->REQ-OMP-CORE-DELIVERY] +// [unit->REQ-OMP-COMMS-RECOVERY] async function testShutdownReapsAndReleasesQueuedCustody() { let listener; let listenerWasLiveAtSessionEnd = false; @@ -935,12 +868,15 @@ async function testShutdownReapsAndReleasesQueuedCustody() { ); await flush(); await harness.emit("agent_start"); - await harness.emit("agent_end", { + const boundary = await harness.emit("context", { messages: [ { role: "user", content: '' }, - assistantMessage("done"), + { role: "user", content: '' }, ], }); + await harness.emit("agent_end", { + messages: [...boundary.messages, assistantMessage("done")], + }); assert.deepEqual(harness.submitted, ['', '']); await harness.emit("session_shutdown"); @@ -1028,7 +964,7 @@ async function testRunSptRejectsStdinErrorsAndHungCommands() { } } -// [unit->REQ-OMP-LISTENER-FAIL-CLOSED] +// [unit->REQ-OMP-COMMS-RECOVERY] async function testListenerTerminationEscalatesAndReaps() { const harness = createHarness({ killForceMs: 4, @@ -1080,15 +1016,14 @@ async function testListenerTerminationEscalatesAndReaps() { assert.deepEqual(errorHarness.clock.delays(), []); } -// [unit->REQ-OMP-EXTENSION-CUSTODY] -// [unit->REQ-OMP-LISTENER-FAIL-CLOSED] +// [unit->REQ-OMP-COMMS-RECOVERY] async function testProtocolCorruptionFailsClosed() { async function failProtocol(payload, expected, options = {}) { - const harness = createHarness({ restartDelaysMs: [], ...options }); + const harness = createHarness({ restartDelaysMs: [5], ...options }); await harness.emit("session_start"); harness.children[0].stdout.emit("data", payload); await flush(); - assert.equal(harness.shutdowns, 1); + assert.equal(harness.shutdowns, 0); assert.deepEqual(harness.submitted, []); assert.deepEqual(commandCalls(harness, "send"), []); assert.ok( @@ -1098,8 +1033,10 @@ 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(), []); + assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 0); + assert.deepEqual(harness.clock.delays(), [5]); + await harness.clock.runNext(5); + assert.equal(harness.children.length, 2); await harness.emit("session_shutdown"); return harness; } @@ -1119,34 +1056,44 @@ async function testProtocolCorruptionFailsClosed() { }); } -// [unit->REQ-OMP-EXTENSION-CUSTODY] -// [unit->REQ-OMP-LISTENER-FAIL-CLOSED] +// [unit->REQ-OMP-CORE-DELIVERY] +// [unit->REQ-OMP-COMMS-RECOVERY] async function testInboundQueueOverflowReleasesAcceptedCustody() { const frames = ["a", "b", "overflow"].map( (from) => `work`, ); const harness = createHarness({ acceptedQueueLimit: 2, - restartDelaysMs: [], + restartDelaysMs: [5], }); 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"), - [], - "capacity failure must not synthesize outbound peer messages", - ); + assert.equal(harness.shutdowns, 0); + assert.deepEqual(harness.submitted, ['', '']); + assert.deepEqual(commandCalls(harness, "send"), []); assert.ok( harness.errors.some(({ message }) => - message.includes("inbound custody capacity exceeded"), + message.includes("inbound listener capacity exceeded"), ), ); - assert.equal(harness.calls.filter((call) => call.args[3] === "session-end").length, 1); - assert.deepEqual(harness.clock.delays(), []); + assert.equal(harness.children[0].kills, 1); + assert.deepEqual(harness.clock.delays(), [5]); + + await harness.emit("agent_start"); + const boundary = await harness.emit("context", { + messages: [ + { role: "user", content: '' }, + { role: "user", content: '' }, + ], + }); + assert.equal(boundary.messages.length, 2); + assert.deepEqual( + harness.submitted, + ['', '', ''], + "freeing observed custody admits the held overflow item without loss", + ); await harness.emit("session_shutdown"); const byteFirst = 'x'; @@ -1154,20 +1101,20 @@ async function testInboundQueueOverflowReleasesAcceptedCustody() { const byteHarness = createHarness({ acceptedBytesLimit: Buffer.byteLength(byteFirst, "utf8") + byteOverflow.length, acceptedQueueLimit: 10, - restartDelaysMs: [], + restartDelaysMs: [5], }); await byteHarness.emit("session_start"); byteHarness.children[0].stdout.emit("data", `${byteFirst}${byteOverflow}`); await flush(); - assert.deepEqual(commandCalls(byteHarness, "send"), []); - assert.equal(byteHarness.shutdowns, 1); - assert.deepEqual(byteHarness.clock.delays(), []); + assert.deepEqual(byteHarness.submitted, ['']); + assert.equal(byteHarness.shutdowns, 0); + assert.deepEqual(byteHarness.clock.delays(), [5]); await byteHarness.emit("session_shutdown"); } -// [unit->REQ-OMP-EXTENSION-CUSTODY] -// [unit->REQ-OMP-LISTENER-FAIL-CLOSED] +// [unit->REQ-OMP-CORE-DELIVERY] +// [unit->REQ-OMP-COMMS-RECOVERY] async function testShutdownFallbackStaysBelowHostCap() { const never = new Promise(() => {}); const harness = createHarness({ @@ -1623,59 +1570,63 @@ async function testStartupBriefHintsAndUpdateNotices() { await delayed.emit("session_shutdown"); } +// [unit->REQ-OMP-CORE-DELIVERY] // [unit->REQ-PARITY-SAFE-BOUNDARY-DELIVERY] // [unit->REQ-HAZARD-ABNORMAL-TURN-RECEIVABILITY] async function testActiveTurnBoundaryDeliveryAndFallback() { - const harness = createHarness(); + let pollCalls = 0; + const polledEnvelope = 'busy'; + const harness = createHarness({ + onRun(call) { + if (call.args[0] === "api" && call.args[3] === "poll") { + pollCalls += 1; + return pollCalls === 1 ? polledEnvelope : ""; + } + }, + }); await harness.emit("session_start"); + await harness.emit("before_agent_start", { prompt: "operator prompt", systemPrompt: [] }); await harness.emit("agent_start"); const firstEnvelope = 'one'; const secondEnvelope = 'two'; harness.children[0].stdout.emit("data", `${firstEnvelope}${secondEnvelope}`); await flush(); - assert.deepEqual(harness.submitted, ['']); + assert.deepEqual( + harness.submitted, + ['', ''], + "live-listener arrivals use the native steer path independently", + ); const boundary = await harness.emit("context", { messages: [ - { role: "user", content: "operator prompt" }, { role: "user", content: '' }, + { role: "user", content: '' }, ], }); assert.equal( - boundary.messages.at(-1).content, + boundary.messages[0].content, `\n\n${formatInboundEnvelope(firstEnvelope)}`, - "an active-turn arrival must steer the running model at its next boundary", ); - const afterToolBoundary = await harness.emit("context", { - messages: [ - { role: "user", content: "operator prompt" }, - { role: "user", content: '' }, - assistantMessage("", { stopReason: "toolUse", toolCalls: [{ name: "read" }] }), - { role: "toolResult", content: "tool output" }, - ], - }); assert.equal( - afterToolBoundary.messages[1].content, - `\n\n${formatInboundEnvelope(firstEnvelope)}`, - "the steered delivery envelope must persist across later model continuations", - ); + boundary.messages[1].content, + `\n\n${formatInboundEnvelope(secondEnvelope)}`, + ); + assert.deepEqual(boundary.messages.at(-1), { + role: "custom", + customType: "spt-event", + content: polledEnvelope, + display: false, + attribution: "user", + timestamp: boundary.messages.at(-1).timestamp, + }); + assert.equal(pollCalls, 1, "busy context boundaries drain active-only core custody"); + assert.deepEqual(harness.sentMessages, [], "busy poll output adds no user-visible panel"); + 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"), - ], + messages: [...boundary.messages, assistantMessage("local outcome")], }); - assert.deepEqual( - commandCalls(harness, "send"), - [], - "assistant output at an active boundary must remain local", - ); + assert.deepEqual(commandCalls(harness, "send"), []); assert.deepEqual(harness.clock.delays(), []); - assert.deepEqual(harness.submitted, ['', '']); - assert.deepEqual(harness.submittedDeliveries, [undefined, undefined]); - assertNoAgentManagedPoll(harness); await harness.emit("session_shutdown"); } @@ -2031,11 +1982,11 @@ await testRunSptRejectsStdinErrorsAndHungCommands(); await testLifecycleCustodyAndContext(); await testLocalAssistantOutputDoesNotReplyToPeer(); await testDeferredBindLifecycleSerialization(); +await testStateReconciliationUsesLatestActivity(); await testSubmissionFailureAdvancesQueue(); await testFailedIdleRecoveryFailsClosed(); await testListenerRestartExhaustion(); await testListenerStableIntervalResetsRetries(); -await testSessionEndRetriesAfterTransientFailure(); await testHumanBusyFailureFailsClosed(); await testShutdownReapsAndReleasesQueuedCustody(); await testListenerTerminationEscalatesAndReaps();