diff --git a/changelog.d/tune-out-timer-failures.fixed.md b/changelog.d/tune-out-timer-failures.fixed.md new file mode 100644 index 000000000..6660857b8 --- /dev/null +++ b/changelog.d/tune-out-timer-failures.fixed.md @@ -0,0 +1,12 @@ +- A tune-out's cadence, coalesced wake or end time whose work throws, such + as a store write that fails during the subconscious's turn, no longer + throws into the event loop, where an uncaught exception ends the host + process. The coordinator writes the error to stderr and reports a + `tune-out:timer-failed` trace naming the timer. Before, a cadence whose + turn threw also never armed again. Now it arms its next period before + the turn, and that next turn counts the failed turn's messages. An end + time whose cancel throws before the channel registry has changed + anything tries the cancel again a minute later, until it succeeds. +- A failed acknowledgement of a suppressed mention is written to stderr + as well as traced, and a rejection with `undefined` or `null` no + longer throws inside its handler. diff --git a/src/tune-out/coordinator.ts b/src/tune-out/coordinator.ts index 021610336..d3215d480 100644 --- a/src/tune-out/coordinator.ts +++ b/src/tune-out/coordinator.ts @@ -34,6 +34,20 @@ export const TUNE_OUT_DEFAULTS = { * subconscious turn, and max-wakes counts invocations, not messages. */ const WAKE_COALESCE_MS = 2_000; +/** How long after an end time whose cancel threw the cancel is tried again. */ +const EXPIRY_RETRY_MS = 60_000; + +/** What a caught value says: an Error's message, or the value as a string. */ +function describeThrown(err: unknown): string { + if (err instanceof Error) return err.message; + try { + return String(err); + } catch { + // A value with no string form, such as an object with no prototype. + return Object.prototype.toString.call(err); + } +} + export interface TuneOutFrameworkHooks { /** Deliver a message into an agent's window (per-agent seam). */ addMessage: ( @@ -65,9 +79,11 @@ export interface TuneOutFrameworkHooks { export class TuneOutCoordinator { private cadenceTimers = new Map>(); - /** Each duration's pending expiry step (params.expiresAtMs). stop clears - * it, cancel does while the registry still holds the tune-out, and the - * channel's next enter does in any case. */ + /** Each duration's pending timer: a step toward its end time + * (params.expiresAtMs), or the retry of an end-time cancel that threw. + * stop clears it, cancel does when it finds the tune-out in the registry, + * the end time does when its cancel finds the tune-out already ended, + * and the channel's next accepted enter does in any case. */ private expiryTimers = new Map>(); /** Channels with a wake invocation pending inside the coalescing window. */ private pendingWakes = new Map }>(); @@ -111,9 +127,35 @@ export class TuneOutCoordinator { const key = this.key(serverId, channelId); const existing = this.expiryTimers.get(key); if (existing) clearTimeout(existing); - armAt(params.expiresAtMs, (timer) => { this.expiryTimers.set(key, timer); }, () => { - this.expiryTimers.delete(key); - this.cancel(serverId, channelId, 'duration', 'duration elapsed'); + const hold = (timer: ReturnType): void => { this.expiryTimers.set(key, timer); }; + const end = (): void => { + try { + // The retry is armed before the cancel, as the cadence arms its next + // period before its turn, so a cancel that throws is tried again. A + // cancel that ends the tune-out clears it with the channel's other + // timers, and so does finding the tune-out already ended. + hold(setTimeout(end, EXPIRY_RETRY_MS)); + if (!this.cancel(serverId, channelId, 'duration', 'duration elapsed').ok) { + this.clearTimers(serverId, channelId); + } + } catch (err) { + this.timerFailed('expiry', serverId, channelId, err); + } + }; + armAt(params.expiresAtMs, hold, end); + } + + /** A timer callback's work threw. Report it on stderr and as a trace rather + * than throw it into the event loop, where an uncaught exception ends the + * host process. */ + private timerFailed(timer: 'cadence' | 'wake' | 'expiry', serverId: string, channelId: string, err: unknown): void { + console.error(`[tune-out] the ${timer} timer for ${channelId} on ${serverId} failed:`, err); + this.hooks.emitTrace({ + type: 'tune-out:timer-failed', + timer, + serverId, + channelId, + error: describeThrown(err), }); } @@ -359,11 +401,12 @@ export class TuneOutCoordinator { intent: 'suppressed-tuned-out', }); } catch (err) { + console.error(`[tune-out] acknowledging ${messageId} in ${channelId} on ${serverId} failed:`, err); this.hooks.emitTrace({ type: 'tune-out:ack-failed', serverId, channelId, - error: (err as Error).message, + error: describeThrown(err), }); } } @@ -377,14 +420,20 @@ export class TuneOutCoordinator { const key = this.key(serverId, channelId); const hold = (timer: ReturnType): void => { this.cadenceTimers.set(key, timer); }; const tick = () => { - const state = this.channelRegistry.getTuneOutState(serverId, channelId); - if (!state) return; // cancelled meanwhile - const diverted = this.divertedSinceInvocation.get(key) ?? 0; - // An empty period needs no turn at all — skip silently. - if (diverted > 0) { - this.invokeSubconscious(serverId, channelId, 'cadence', diverted); + try { + const state = this.channelRegistry.getTuneOutState(serverId, channelId); + if (!state) return; // cancelled meanwhile + // The next period is armed before this one's turn, so a turn that + // fails can't end the cadence. + armAfter(cadenceSeconds * 1000, hold, tick); + const diverted = this.divertedSinceInvocation.get(key) ?? 0; + // An empty period needs no turn at all — skip silently. + if (diverted > 0) { + this.invokeSubconscious(serverId, channelId, 'cadence', diverted); + } + } catch (err) { + this.timerFailed('cadence', serverId, channelId, err); } - armAfter(cadenceSeconds * 1000, hold, tick); }; armAfter(cadenceSeconds * 1000, hold, tick); } @@ -417,21 +466,25 @@ export class TuneOutCoordinator { return; } const timer = setTimeout(() => { - const entry = this.pendingWakes.get(key); - this.pendingWakes.delete(key); - // Count ONE durable wake per coalesced invocation. - const counted = this.channelRegistry.recordTuneOutWake(serverId, channelId); - if (!counted) return; // cancelled inside the window - if (counted.wakeCount > counted.params.maxWakes) { - this.cancel( - serverId, - channelId, - 'tune-out-max-wakes', - `wake budget exhausted (${counted.params.maxWakes})`, - ); - return; + try { + const entry = this.pendingWakes.get(key); + this.pendingWakes.delete(key); + // Count ONE durable wake per coalesced invocation. + const counted = this.channelRegistry.recordTuneOutWake(serverId, channelId); + if (!counted) return; // cancelled inside the window + if (counted.wakeCount > counted.params.maxWakes) { + this.cancel( + serverId, + channelId, + 'tune-out-max-wakes', + `wake budget exhausted (${counted.params.maxWakes})`, + ); + return; + } + this.invokeSubconscious(serverId, channelId, 'wake', entry?.count ?? 1); + } catch (err) { + this.timerFailed('wake', serverId, channelId, err); } - this.invokeSubconscious(serverId, channelId, 'wake', entry?.count ?? 1); }, WAKE_COALESCE_MS); this.pendingWakes.set(key, { count: 1, timer }); } @@ -450,7 +503,6 @@ export class TuneOutCoordinator { const subName = this.hooks.subconsciousName(); if (!subName) return; const key = this.key(serverId, channelId); - this.divertedSinceInvocation.set(key, 0); const notice = trigger === 'wake' ? `[Tune-out wake: ${channelId} — ${count} addressed message${count === 1 ? '' : 's'}]` @@ -462,6 +514,9 @@ export class TuneOutCoordinator { subName, ); this.hooks.requestInference(subName, `tune-out ${trigger}: ${channelId}`, 'tune-out'); + // Counted from here only once the turn has been asked for, so a turn that + // throws leaves its messages for the next period's turn. + this.divertedSinceInvocation.set(key, 0); } // ========================================================================== diff --git a/test/tune-out-timers.test.ts b/test/tune-out-timers.test.ts index 5ffc6e59c..f71d4d680 100644 --- a/test/tune-out-timers.test.ts +++ b/test/tune-out-timers.test.ts @@ -1,6 +1,7 @@ /** - * A tune-out's timers hold any length the resident asks for, and enter() - * refuses an option it can't use before anything changes (F76). + * A tune-out's timers hold any length the resident asks for, enter() + * refuses an option it can't use before anything changes, and a timer whose + * work throws reports it rather than throwing it into the event loop. * * Node fires a setTimeout longer than 2^31 - 1 ms (about 24.8 days) almost * at once, so a single timer ended a longer tune-out as soon as it began, @@ -18,6 +19,7 @@ import { TuneOutCoordinator, type TuneOutFrameworkHooks } from '../src/tune-out/ import { armAt } from '../src/deadline-timer.js'; import type { ChannelRegistry, TuneOutParams } from '../src/mcpl/channel-registry.js'; import type { McplServerRegistry } from '../src/mcpl/server-registry.js'; +import { CapabilityGrant } from '../src/mcpl/capability-grant.js'; /** The longest delay one setTimeout holds. */ const ONE_TIMER_MS = 2_147_483_647; @@ -29,9 +31,14 @@ const STORED: TuneOutParams = { epochId: 'e1', cadenceSeconds: 1800, backlogCap: * A coordinator over an in-memory registry for one channel, `disc` `#a`, with * setTimeout mocked and each delay asked of it recorded, and the wall clock * mocked unless `wallClock` is false. `stored` is a tune-out the registry - * already holds, as after a restart. + * already holds, as after a restart. `servers` is the MCPL server registry, + * by default one with no servers. */ -function harness(t: TestContext, stored?: TuneOutParams, { wallClock = true } = {}) { +function harness( + t: TestContext, + stored?: TuneOutParams, + { wallClock = true, servers = { getServer: () => undefined } as unknown as McplServerRegistry } = {}, +) { if (wallClock) t.mock.timers.enable({ apis: ['Date', 'setTimeout'], now: NOW }); else t.mock.timers.enable({ apis: ['setTimeout'] }); const mocked = globalThis.setTimeout; @@ -42,9 +49,24 @@ function harness(t: TestContext, stored?: TuneOutParams, { wallClock = true } = }) as unknown as typeof setTimeout; let state: { params: TuneOutParams; wakeCount: number } | null = stored ? { params: stored, wakeCount: 0 } : null; + /** What each step throws on its next call, as a failed store write would. */ + const throwsNext = new Map<'addMessage' | 'recordTuneOutWake' | 'cancelTuneOut', unknown>(); + const throwIfSet = (step: 'addMessage' | 'recordTuneOutWake' | 'cancelTuneOut') => { + if (!throwsNext.has(step)) return; + const thrown = throwsNext.get(step); + throwsNext.delete(step); + throw thrown; + }; + const traces: Array> = []; + /** The text of each message added: the subconscious's notices among them. */ + const notices: string[] = []; + /** What the coordinator writes to stderr. */ + const stderr = t.mock.method(console, 'error', () => {}); const entered: TuneOutParams[] = []; /** The source of each cancel that ended a tune-out: 'duration' for an end time. */ const cancels: string[] = []; + /** Every call to the registry's cancel, whatever came of it. */ + let cancelCalls = 0; /** The reason of each inference requested, the subconscious's turns among them. */ const inferences: string[] = []; const registry = { @@ -55,15 +77,24 @@ function harness(t: TestContext, stored?: TuneOutParams, { wallClock = true } = state = { params, wakeCount: 0 }; }, cancelTuneOut: (_s: string, _c: string, _next: string, source: string) => { + cancelCalls++; + throwIfSet('cancelTuneOut'); const ended = state; state = null; if (ended) cancels.push(source); return ended; }, - recordTuneOutWake: () => (state ? { params: state.params, wakeCount: ++state.wakeCount } : null), + recordTuneOutWake: () => { + throwIfSet('recordTuneOutWake'); + return state ? { params: state.params, wakeCount: ++state.wakeCount } : null; + }, } as unknown as ChannelRegistry; const hooks = { - addMessage: () => '', + addMessage: (_participant: string, content: Array<{ text: string }>) => { + throwIfSet('addMessage'); + notices.push(content[0]?.text ?? ''); + return ''; + }, requestInference: (_agent: string, reason: string) => { inferences.push(reason); }, subconsciousName: () => 'Subconscious', primaryName: () => 'scout', @@ -73,9 +104,9 @@ function harness(t: TestContext, stored?: TuneOutParams, { wallClock = true } = isForkBound: () => false, isPrivilegedAuthor: () => false, allowChannelSpeech: () => false, - emitTrace: () => {}, + emitTrace: (event: Record) => { traces.push(event); }, } as unknown as TuneOutFrameworkHooks; - const coordinator = new TuneOutCoordinator(registry, {} as McplServerRegistry, hooks); + const coordinator = new TuneOutCoordinator(registry, servers, hooks); t.after(() => { coordinator.stop(); globalThis.setTimeout = mocked; @@ -90,7 +121,19 @@ function harness(t: TestContext, stored?: TuneOutParams, { wallClock = true } = /** End the tune-out in the registry alone, as a send into the channel or a close does. */ const endInRegistry = () => { state = null; }; const active = () => state !== null; - return { coordinator, entered, cancels, cadenceTurns, wakeTurns, delays, noTimerPastOne, endInRegistry, active }; + /** Make the step's next call throw `thrown`. */ + const failNext = ( + step: 'addMessage' | 'recordTuneOutWake' | 'cancelTuneOut' = 'addMessage', + thrown: unknown = new Error('store write failed'), + ) => { throwsNext.set(step, thrown); }; + const timerFailures = () => traces.filter((e) => e.type === 'tune-out:timer-failed'); + const ackFailures = () => traces.filter((e) => e.type === 'tune-out:ack-failed'); + /** The arguments of each line written to stderr. */ + const stderrLines = () => stderr.mock.calls.map((c) => c.arguments as unknown[]); + return { + coordinator, entered, cancels, cadenceTurns, wakeTurns, delays, noTimerPastOne, endInRegistry, active, + failNext, timerFailures, ackFailures, notices, stderrLines, cancelCalls: () => cancelCalls, + }; } it('a duration past one timer ends at its end time and not before', (t) => { @@ -299,3 +342,182 @@ it('armAt arms a deadline that has passed for the next turn of the loop, with no process.off('warning', onWarning); } }); + +// A timer callback's work can throw, as a failed store write would. Thrown into +// the event loop, it's an uncaught exception, which ends the host process. +// Under node:test's mock timers a throw also comes out of tick(), and the mock +// then runs that callback once more at its next tick, so these tests assert +// that nothing leaves tick(), not only what comes after. + +const failure = (timer: string, error = 'store write failed') => + ({ type: 'tune-out:timer-failed', timer, serverId: 'disc', channelId: '#a', error }); + +it("a cadence turn that throws stays in its timer, is reported, and the next period's turn counts its messages", (t) => { + const h = harness(t); + assert.equal(h.coordinator.enter('disc', '#a', { cadenceSeconds: 60 }, 'agent-tool').ok, true); + h.coordinator.onIncoming('disc', '#a', 'm1', undefined); + h.failNext(); + assert.doesNotThrow(() => t.mock.timers.tick(60_000), "the turn's error stays in the timer"); + assert.deepEqual(h.timerFailures(), [failure('cadence')]); + assert.deepEqual(h.cadenceTurns(), [], 'the failed turn requested nothing'); + // No new traffic: the next period's turn is for m1. + t.mock.timers.tick(60_000); + assert.deepEqual(h.cadenceTurns(), ['tune-out cadence: #a'], "the next period's turn comes"); + assert.match(h.notices.at(-1) ?? '', /— 1 diverted message since your last look\]$/); +}); + +it('a timer failure is written to stderr with the error', (t) => { + const h = harness(t); + assert.equal(h.coordinator.enter('disc', '#a', { cadenceSeconds: 60 }, 'agent-tool').ok, true); + h.coordinator.onIncoming('disc', '#a', 'm1', undefined); + const thrown = new Error('store write failed'); + h.failNext('addMessage', thrown); + t.mock.timers.tick(60_000); + assert.deepEqual(h.stderrLines(), [['[tune-out] the cadence timer for #a on disc failed:', thrown]]); +}); + +it('after a turn, the count starts again: a quiet period gives no turn, and the next notice counts only new messages', (t) => { + const h = harness(t); + assert.equal(h.coordinator.enter('disc', '#a', { cadenceSeconds: 60 }, 'agent-tool').ok, true); + h.coordinator.onIncoming('disc', '#a', 'm1', undefined); + h.coordinator.onIncoming('disc', '#a', 'm2', undefined); + t.mock.timers.tick(60_000); + assert.match(h.notices.at(-1) ?? '', /— 2 diverted messages since your last look\]$/); + t.mock.timers.tick(60_000); + assert.equal(h.cadenceTurns().length, 1, 'no turn for a quiet period after a turn'); + h.coordinator.onIncoming('disc', '#a', 'm3', undefined); + t.mock.timers.tick(60_000); + assert.equal(h.cadenceTurns().length, 2); + assert.match(h.notices.at(-1) ?? '', /— 1 diverted message since your last look\]$/); +}); + +it("a wake turn that throws leaves its message for the next cadence turn", (t) => { + const h = harness(t); + assert.equal(h.coordinator.enter('disc', '#a', { cadenceSeconds: 60 }, 'agent-tool').ok, true); + h.coordinator.onIncoming('disc', '#a', 'm1', ['chat:addressed']); + h.failNext(); + t.mock.timers.tick(2_000); + assert.deepEqual(h.wakeTurns(), [], 'the wake turn failed'); + t.mock.timers.tick(58_000); + assert.deepEqual(h.cadenceTurns(), ['tune-out cadence: #a'], 'the cadence turn comes'); + assert.match(h.notices.at(-1) ?? '', /— 1 diverted message since your last look\]$/); +}); + +it("a cadence tick that finds the tune-out ended in the registry arms nothing and gives no turn", (t) => { + const h = harness(t); + assert.equal(h.coordinator.enter('disc', '#a', { cadenceSeconds: 60 }, 'agent-tool').ok, true); + h.coordinator.onIncoming('disc', '#a', 'm1', undefined); + h.endInRegistry(); + t.mock.timers.tick(60_000); + t.mock.timers.tick(60_000); + assert.deepEqual(h.cadenceTurns(), [], 'no turn for an ended tune-out'); + assert.equal(h.delays.filter((ms) => ms === 60_000).length, 1, 'only the first period was armed'); +}); + +it("a quiet cadence period arms the next, whose turn comes when a message is diverted", (t) => { + const h = harness(t); + assert.equal(h.coordinator.enter('disc', '#a', { cadenceSeconds: 60 }, 'agent-tool').ok, true); + t.mock.timers.tick(60_000); + assert.deepEqual(h.cadenceTurns(), [], 'a quiet period gives no turn'); + h.coordinator.onIncoming('disc', '#a', 'm1', undefined); + t.mock.timers.tick(60_000); + assert.deepEqual(h.cadenceTurns(), ['tune-out cadence: #a'], "the next period's turn comes"); +}); + +it('a coalesced wake whose turn throws stays in its timer and is reported', (t) => { + const h = harness(t); + assert.equal(h.coordinator.enter('disc', '#a', {}, 'agent-tool').ok, true); + h.coordinator.onIncoming('disc', '#a', 'm1', ['chat:addressed']); + h.failNext(); + assert.doesNotThrow(() => t.mock.timers.tick(2_000), "the turn's error stays in the timer"); + assert.deepEqual(h.timerFailures(), [failure('wake')]); + h.coordinator.onIncoming('disc', '#a', 'm2', ['chat:addressed']); + t.mock.timers.tick(2_000); + assert.deepEqual(h.wakeTurns(), ['tune-out wake: #a'], "the next wake's turn comes"); +}); + +it("a coalesced wake whose count can't be recorded stays in its timer and is reported", (t) => { + const h = harness(t); + assert.equal(h.coordinator.enter('disc', '#a', {}, 'agent-tool').ok, true); + h.coordinator.onIncoming('disc', '#a', 'm1', ['chat:addressed']); + h.failNext('recordTuneOutWake'); + assert.doesNotThrow(() => t.mock.timers.tick(2_000), "the error stays in the timer"); + assert.deepEqual(h.timerFailures(), [failure('wake')]); + assert.deepEqual(h.wakeTurns(), [], 'no turn for the wake it could not count'); +}); + +it('a coalesced wake for a tune-out the registry ended in its window gives no turn and reports nothing', (t) => { + const h = harness(t); + assert.equal(h.coordinator.enter('disc', '#a', {}, 'agent-tool').ok, true); + h.coordinator.onIncoming('disc', '#a', 'm1', ['chat:addressed']); + h.endInRegistry(); + t.mock.timers.tick(2_000); + assert.deepEqual(h.wakeTurns(), []); + assert.deepEqual(h.timerFailures(), []); +}); + +it('a failed acknowledgement is written to stderr and traced, whatever it rejects with', async (t) => { + const servers = { + getServer: () => ({ + grant: new CapabilityGrant(new Set(['channels.acknowledge']), []), + sendChannelsAcknowledge: async () => { throw undefined; }, + }), + } as unknown as McplServerRegistry; + const h = harness(t, undefined, { servers }); + assert.equal(h.coordinator.enter('disc', '#a', {}, 'agent-tool').ok, true); + h.coordinator.onIncoming('disc', '#a', 'm1', ['chat:addressed']); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(h.ackFailures(), [ + { type: 'tune-out:ack-failed', serverId: 'disc', channelId: '#a', error: 'undefined' }, + ]); + assert.deepEqual(h.stderrLines(), [['[tune-out] acknowledging m1 in #a on disc failed:', undefined]]); +}); + +it('an end time whose cancel throws after the registry ended the tune-out stays in its timer, is reported, and tries no more', (t) => { + const h = harness(t); + assert.equal(h.coordinator.enter('disc', '#a', { durationSeconds: 120 }, 'agent-tool').ok, true); + h.failNext(); + assert.doesNotThrow(() => t.mock.timers.tick(120_000), "the cancel's error stays in the timer"); + assert.deepEqual(h.timerFailures(), [failure('expiry')]); + t.mock.timers.tick(10 * 60_000); + assert.equal(h.cancelCalls(), 1, 'the registry ended it, so nothing tries again'); +}); + +it('an end time whose registry cancel throws is reported, and the cancel is tried again a minute later', (t) => { + // The stand-in's cancel throws before it changes anything, as a registry + // that writes its record before it changes memory does when the write fails. + const h = harness(t); + assert.equal(h.coordinator.enter('disc', '#a', { durationSeconds: 120 }, 'agent-tool').ok, true); + h.failNext('cancelTuneOut'); + assert.doesNotThrow(() => t.mock.timers.tick(120_000), "the cancel's error stays in the timer"); + assert.deepEqual(h.timerFailures(), [failure('expiry')]); + assert.equal(h.active(), true, 'still tuned out after the failed cancel'); + t.mock.timers.tick(60_000 - 1); + assert.deepEqual(h.cancels, [], 'not before the minute'); + t.mock.timers.tick(1); + assert.deepEqual(h.cancels, ['duration'], 'ended a minute later'); + t.mock.timers.tick(10 * 60_000); + assert.equal(h.cancelCalls(), 2, 'and nothing tries again after it ended'); +}); + +it('an end time that finds the tune-out already ended in the registry tries no more', (t) => { + const h = harness(t); + assert.equal(h.coordinator.enter('disc', '#a', { durationSeconds: 120 }, 'agent-tool').ok, true); + h.endInRegistry(); + t.mock.timers.tick(120_000); + t.mock.timers.tick(10 * 60_000); + assert.equal(h.cancelCalls(), 1); + assert.deepEqual(h.timerFailures(), []); +}); + +it('a thrown value that is not an Error is reported as a string, even one with no string form', (t) => { + const h = harness(t); + assert.equal(h.coordinator.enter('disc', '#a', { cadenceSeconds: 60 }, 'agent-tool').ok, true); + h.coordinator.onIncoming('disc', '#a', 'm1', undefined); + h.failNext('addMessage', 'disk full'); + assert.doesNotThrow(() => t.mock.timers.tick(60_000)); + // String() throws for an object with no prototype. + h.failNext('addMessage', Object.create(null)); + assert.doesNotThrow(() => t.mock.timers.tick(60_000)); + assert.deepEqual(h.timerFailures().map((e) => e.error), ['disk full', '[object Object]']); +});