/** * @fileoverview Unit tests for the blocking-wait registry that backs the agent * wait primitives (`GET /api/sessions/:id/wait`, `.../wait-output`, and the * `wait` field on `POST .../input`). Plan: `docs/agent-control-plan.md`. * * The registry holds no IO and no Session reference, so everything here runs * against real (short) timers with no mocks beyond one clearTimeout spy. */ import { describe, it, expect, vi } from 'vitest'; import { SessionWaitRegistry, WaitCapacityError, hooksAvailableForMode, parseWaitSignals, resolveWaitSignals, signalForStatus, DEFAULT_WAIT_SIGNALS, WAIT_SIGNALS, } from '../src/web/session-wait-registry.js'; import type { SessionMode } from '../src/types.js'; const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); describe('parseWaitSignals', () => { it('parses a comma-separated string', () => { expect(parseWaitSignals('stop,idle')).toEqual({ signals: ['stop', 'idle'], invalid: [] }); }); it('parses an array, including comma-joined entries', () => { expect(parseWaitSignals(['stop', 'idle,exit'])).toEqual({ signals: ['stop', 'idle', 'exit'], invalid: [], }); }); it('trims and lowercases', () => { expect(parseWaitSignals(' STOP , Idle ')).toEqual({ signals: ['stop', 'idle'], invalid: [] }); }); it('dedups, first occurrence wins', () => { expect(parseWaitSignals('idle,stop,idle')).toEqual({ signals: ['idle', 'stop'], invalid: [] }); }); it('reports invalid tokens instead of silently dropping them', () => { // The whole point: a typo must surface as a 400, never as "waiting for the default". const parsed = parseWaitSignals('stop,stpo,done'); expect(parsed.signals).toEqual(['stop']); expect(parsed.invalid).toEqual(['stpo', 'done']); }); it('dedups invalid tokens too', () => { expect(parseWaitSignals('nope,nope').invalid).toEqual(['nope']); }); it('returns empty for absent / non-string input', () => { expect(parseWaitSignals(undefined)).toEqual({ signals: [], invalid: [] }); expect(parseWaitSignals(null)).toEqual({ signals: [], invalid: [] }); expect(parseWaitSignals(42)).toEqual({ signals: [], invalid: [] }); expect(parseWaitSignals('')).toEqual({ signals: [], invalid: [] }); expect(parseWaitSignals(',, ,')).toEqual({ signals: [], invalid: [] }); }); it('accepts every documented signal', () => { expect(parseWaitSignals(WAIT_SIGNALS.join(','))).toEqual({ signals: [...WAIT_SIGNALS], invalid: [], }); }); it('the default set is itself valid', () => { expect(parseWaitSignals(DEFAULT_WAIT_SIGNALS.join(','))).toEqual({ signals: [...DEFAULT_WAIT_SIGNALS], invalid: [], }); }); it('the default set includes exit, so a crashed worker never burns the full timeout', () => { expect(DEFAULT_WAIT_SIGNALS).toContain('exit'); }); }); describe('signalForStatus', () => { it('maps live statuses', () => { expect(signalForStatus('idle')).toBe('idle'); expect(signalForStatus('busy')).toBe('working'); }); it('maps both dead statuses to exit, so a caller that sees one does not hang', () => { // Narrower than it looks, and the JSDoc says so: a PTY that merely exits parks the // session at 'idle', so 'stopped' is reached only on a spawn failure or an explicit // stop(killMux:true). This mapping is right; it is not a liveness check. expect(signalForStatus('stopped')).toBe('exit'); expect(signalForStatus('error')).toBe('exit'); }); it('cannot distinguish a finished session from a dead one', () => { // Pinning the documented limitation: both PTY onExit handlers set 'idle', so this // is what a caller sees for a crashed worker. Liveness must come from elsewhere. expect(signalForStatus('idle')).toBe('idle'); }); }); describe('hooksAvailableForMode', () => { it('is true only for claude', () => { expect(hooksAvailableForMode('claude')).toBe(true); }); it('is false for shell, which installs no hooks despite not being an external CLI', () => { // The bug this replaced keyed off isExternalCliMode(), which excludes shell, so // until=stop on a bash PTY was accepted and then blocked for the full timeout. expect(hooksAvailableForMode('shell')).toBe(false); }); it('is false for every external CLI mode', () => { for (const mode of ['opencode', 'codex', 'gemini', 'antigravity'] as const) { expect(hooksAvailableForMode(mode)).toBe(false); } }); }); describe('resolveWaitSignals', () => { const claude = { mode: 'claude' as SessionMode }; const codex = { mode: 'codex' as SessionMode }; const shell = { mode: 'shell' as SessionMode }; it('returns the explicit set for a normal request', () => { expect(resolveWaitSignals('stop,exit', claude)).toEqual({ until: ['stop', 'exit'], error: null }); }); it('falls back to the default set when nothing is asked for', () => { expect(resolveWaitSignals(undefined, claude)).toEqual({ until: ['stop', 'idle', 'exit'], error: null }); // `wait: true` on the input route arrives here as undefined, same path. expect(resolveWaitSignals('', claude).until).toEqual(['stop', 'idle', 'exit']); }); it('errors on an unknown token rather than silently defaulting', () => { const result = resolveWaitSignals('stop,stpo', claude); expect(result.until).toEqual([]); expect(result.error).toContain('stpo'); expect(result.error).toContain('idle, working, stop, blocked, exit'); }); it('errors when a hook-only signal is asked for EXPLICITLY on an external CLI', () => { const result = resolveWaitSignals('stop', codex); expect(result.until).toEqual([]); expect(result.error).toContain('codex'); expect(resolveWaitSignals('blocked', codex).error).toBeTruthy(); expect(resolveWaitSignals('idle,stop', codex).error).toBeTruthy(); }); it('drops hook-only signals from the DEFAULT set instead of erroring', () => { // Omitting the parameter must never 400, whatever the mode. expect(resolveWaitSignals(undefined, codex)).toEqual({ until: ['idle', 'exit'], error: null }); }); it('still allows the supported signals explicitly on an external CLI', () => { expect(resolveWaitSignals('idle,exit', codex)).toEqual({ until: ['idle', 'exit'], error: null }); }); it('reports an unknown token even when the mode would also reject', () => { // Validity is checked before mode support, so the message names the typo. expect(resolveWaitSignals('stpo,stop', codex).error).toContain('stpo'); }); it('rejects hook-only signals on a SHELL session too', () => { // A shell session is a plain bash PTY with no Claude Code hooks, so `stop` can // never arrive; accepting it was a guaranteed ten-minute hold by construction. const result = resolveWaitSignals('stop', shell); expect(result.until).toEqual([]); expect(result.error).toContain('shell'); expect(resolveWaitSignals('blocked', shell).error).toBeTruthy(); }); it('drops hook-only signals from the DEFAULT set for shell, without erroring', () => { expect(resolveWaitSignals(undefined, shell)).toEqual({ until: ['idle', 'exit'], error: null }); }); it('names the mode in the rejection so the caller can see why', () => { expect(resolveWaitSignals('stop', codex).error).toContain('codex'); expect(resolveWaitSignals('stop', { mode: 'gemini' as SessionMode }).error).toContain('gemini'); }); }); describe('SessionWaitRegistry: signal waits', () => { it('resolves immediately when the session is already in a requested state', async () => { const reg = new SessionWaitRegistry(); const result = await reg.waitForSignal('s1', { until: ['idle'], timeoutMs: 5000, currentSignal: 'idle', }); expect(result).toEqual({ signal: 'idle', timedOut: false, immediate: true, ended: false, aborted: false, waitedMs: 0, timeoutMs: 5000, }); expect(reg.totalWaiterCount()).toBe(0); }); it('does not resolve immediately when the current signal was not requested', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000, currentSignal: 'idle', }); expect(reg.signalWaiterCount('s1')).toBe(1); reg.notifySignal('s1', 'stop'); expect((await promise).signal).toBe('stop'); }); it('requireTransition ignores the current state and waits for the next one', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForSignal('s1', { until: ['idle'], timeoutMs: 5000, currentSignal: 'idle', requireTransition: true, }); expect(reg.signalWaiterCount('s1')).toBe(1); const result = await Promise.resolve().then(() => { reg.notifySignal('s1', 'idle'); return promise; }); expect(result.signal).toBe('idle'); expect(result.immediate).toBe(false); }); it('resolves on the first of several requested signals', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForSignal('s1', { until: ['stop', 'blocked'], timeoutMs: 5000 }); reg.notifySignal('s1', 'blocked'); const result = await promise; expect(result.signal).toBe('blocked'); expect(result.timedOut).toBe(false); expect(result.ended).toBe(false); expect(result.waitedMs).toBeGreaterThanOrEqual(0); }); it('ignores signals that were not requested', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); expect(reg.notifySignal('s1', 'working')).toBe(0); expect(reg.notifySignal('s1', 'idle')).toBe(0); expect(reg.signalWaiterCount('s1')).toBe(1); reg.notifySignal('s1', 'stop'); expect((await promise).signal).toBe('stop'); }); it('ignores signals for other sessions', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); expect(reg.notifySignal('s2', 'stop')).toBe(0); expect(reg.signalWaiterCount('s1')).toBe(1); reg.notifySignal('s1', 'stop'); await promise; }); it('wakes every waiter that asked for the signal, and only those', async () => { const reg = new SessionWaitRegistry(); const a = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); const b = reg.waitForSignal('s1', { until: ['stop', 'idle'], timeoutMs: 5000 }); const c = reg.waitForSignal('s1', { until: ['exit'], timeoutMs: 5000 }); expect(reg.signalWaiterCount('s1')).toBe(3); expect(reg.notifySignal('s1', 'stop')).toBe(2); expect((await a).signal).toBe('stop'); expect((await b).signal).toBe('stop'); expect(reg.signalWaiterCount('s1')).toBe(1); reg.notifySignal('s1', 'exit'); expect((await c).signal).toBe('exit'); expect(reg.totalWaiterCount()).toBe(0); }); it('times out without erroring, and reports it as a normal outcome', async () => { const reg = new SessionWaitRegistry(); const result = await reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 20 }); expect(result.timedOut).toBe(true); expect(result.signal).toBeNull(); expect(result.ended).toBe(false); expect(result.waitedMs).toBeGreaterThanOrEqual(0); expect(reg.totalWaiterCount()).toBe(0); }); it('an empty until set can only time out', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForSignal('s1', { until: [], timeoutMs: 20 }); expect(reg.notifySignal('s1', 'stop')).toBe(0); expect((await promise).timedOut).toBe(true); }); it('cancelAll resolves pending waiters with ended, never leaving them hanging', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); expect(reg.cancelAll('s1')).toBe(1); const result = await promise; expect(result.ended).toBe(true); expect(result.timedOut).toBe(false); expect(result.signal).toBeNull(); expect(reg.totalWaiterCount()).toBe(0); }); it('honors the exit-before-cancel ordering contract', async () => { // The wiring must notify 'exit' BEFORE cancelAll, so an until=exit caller sees // the signal rather than a bare ended. const reg = new SessionWaitRegistry(); const promise = reg.waitForSignal('s1', { until: ['exit'], timeoutMs: 5000 }); reg.notifySignal('s1', 'exit'); expect(reg.cancelAll('s1')).toBe(0); const result = await promise; expect(result.signal).toBe('exit'); expect(result.ended).toBe(false); }); it('clears the timer when a signal resolves the wait', async () => { const reg = new SessionWaitRegistry(); const spy = vi.spyOn(globalThis, 'clearTimeout'); const before = spy.mock.calls.length; const promise = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); reg.notifySignal('s1', 'stop'); await promise; expect(spy.mock.calls.length).toBeGreaterThan(before); spy.mockRestore(); }); it('echoes the effective timeout on every outcome', async () => { // A caller that asked for 30 minutes and got 600s must be able to SEE that, or it // reads the timeout as a stalled worker and kills a session that was fine. const reg = new SessionWaitRegistry(); expect((await reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 20 })).timeoutMs).toBe(20); const signalled = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 4321 }); reg.notifySignal('s1', 'stop'); expect((await signalled).timeoutMs).toBe(4321); const cancelled = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 777 }); reg.cancelAll('s1'); expect((await cancelled).timeoutMs).toBe(777); }); }); describe('SessionWaitRegistry: cancellation', () => { it('frees the slot as soon as the caller hangs up', async () => { // The whole point: the response can no longer be sent, so holding the waiter for // the rest of its timeout only denies the pool to somebody else. const reg = new SessionWaitRegistry(); const controller = new AbortController(); const promise = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 60_000, abortSignal: controller.signal, }); expect(reg.signalWaiterCount('s1')).toBe(1); controller.abort(); const result = await promise; expect(result.aborted).toBe(true); expect(result.ended).toBe(true); expect(result.timedOut).toBe(false); expect(reg.totalWaiterCount()).toBe(0); }); it('frees an output waiter the same way', async () => { const reg = new SessionWaitRegistry(); const controller = new AbortController(); const promise = reg.waitForOutput('s1', { match: 'never', timeoutMs: 60_000, abortSignal: controller.signal, }); expect(reg.outputWaiterCount('s1')).toBe(1); controller.abort(); const result = await promise; expect(result.aborted).toBe(true); expect(result.ended).toBe(true); expect(result.matched).toBe(false); expect(reg.totalWaiterCount()).toBe(0); }); it('registers nothing at all when the signal is already aborted', () => { const reg = new SessionWaitRegistry(); const controller = new AbortController(); controller.abort(); void reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 60_000, abortSignal: controller.signal }); void reg.waitForOutput('s1', { match: 'x', timeoutMs: 60_000, abortSignal: controller.signal }); expect(reg.totalWaiterCount()).toBe(0); }); it('an already-aborted caller is never rejected for capacity', async () => { // It takes no slot, so refusing it would be a lie AND would cost the caller a // retry it cannot act on. const reg = new SessionWaitRegistry({ maxWaitersPerSession: 1, maxWaitersTotal: 1 }); void reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 60_000 }); const controller = new AbortController(); controller.abort(); const result = await reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 60_000, abortSignal: controller.signal, }); expect(result.aborted).toBe(true); reg.cancelEverything(); }); it('aborting an already-resolved waiter is a no-op', async () => { const reg = new SessionWaitRegistry(); const controller = new AbortController(); const promise = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000, abortSignal: controller.signal, }); reg.notifySignal('s1', 'stop'); const result = await promise; expect(result.signal).toBe('stop'); controller.abort(); await sleep(5); // The settled result is unchanged and no bookkeeping went negative. expect(result.aborted).toBe(false); expect(reg.totalWaiterCount()).toBe(0); }); it('detaches its abort listener when the wait resolves another way', async () => { // A caller reusing one controller across a loop of waits must not accumulate // listeners on it for the life of the request. const reg = new SessionWaitRegistry(); const controller = new AbortController(); for (let i = 0; i < 5; i++) { const promise = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000, abortSignal: controller.signal, }); reg.notifySignal('s1', 'stop'); await promise; } // Node exposes the count only through the internal getter, so assert the effect: // nothing is registered and a late abort still changes nothing. controller.abort(); expect(reg.totalWaiterCount()).toBe(0); }); it('aborted is false on every non-abort outcome', async () => { const reg = new SessionWaitRegistry(); expect((await reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 20 })).aborted).toBe(false); expect((await reg.waitForOutput('s1', { match: 'x', timeoutMs: 20 })).aborted).toBe(false); const immediate = await reg.waitForSignal('s1', { until: ['idle'], timeoutMs: 20, currentSignal: 'idle' }); expect(immediate.aborted).toBe(false); }); }); describe('SessionWaitRegistry: shutdown latch', () => { it('refuses to register a new waiter once stopped, resolving ended instead', async () => { // server.stop() cancels waiters well before app.close(), with several awaits in // between while the listener still accepts requests. A wait landing in that window // used to register a timer nothing would ever cancel and stall shutdown for the // full MAX_WAIT_MS. const reg = new SessionWaitRegistry(); reg.stop(); expect(reg.isStopped).toBe(true); const signal = await reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 600_000 }); expect(signal.ended).toBe(true); expect(signal.timedOut).toBe(false); expect(signal.aborted).toBe(false); const output = await reg.waitForOutput('s1', { match: 'never', timeoutMs: 600_000 }); expect(output.ended).toBe(true); expect(output.matched).toBe(false); expect(reg.totalWaiterCount()).toBe(0); }); it('stop() resolves the waiters that were already pending', async () => { const reg = new SessionWaitRegistry(); const pending = [ reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 600_000 }), reg.waitForOutput('s2', { match: 'never', timeoutMs: 600_000 }), ]; expect(reg.stop()).toBe(2); for (const result of await Promise.all(pending)) expect(result.ended).toBe(true); }); it('cancelEverything latches too, so the existing shutdown call site is covered', async () => { const reg = new SessionWaitRegistry(); reg.cancelEverything(); expect(reg.isStopped).toBe(true); expect((await reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 600_000 })).ended).toBe(true); }); it('a fresh registry is not stopped', () => { expect(new SessionWaitRegistry().isStopped).toBe(false); }); }); describe('SessionWaitRegistry: capacity', () => { it('rejects past the per-session cap with scope "session"', () => { const reg = new SessionWaitRegistry({ maxWaitersPerSession: 2, maxWaitersTotal: 100 }); void reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); void reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); try { reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); expect.unreachable('third wait should have been rejected'); } catch (err) { expect(err).toBeInstanceOf(WaitCapacityError); expect((err as WaitCapacityError).scope).toBe('session'); } // A different session is unaffected by another session's cap. expect(() => reg.waitForSignal('s2', { until: ['stop'], timeoutMs: 5000 })).not.toThrow(); reg.cancelEverything(); }); it('rejects past the global cap with scope "total"', () => { const reg = new SessionWaitRegistry({ maxWaitersPerSession: 10, maxWaitersTotal: 2 }); void reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); void reg.waitForSignal('s2', { until: ['stop'], timeoutMs: 5000 }); try { reg.waitForSignal('s3', { until: ['stop'], timeoutMs: 5000 }); expect.unreachable('third wait should have been rejected'); } catch (err) { expect(err).toBeInstanceOf(WaitCapacityError); expect((err as WaitCapacityError).scope).toBe('total'); } reg.cancelEverything(); }); it('counts both waiter kinds against the caps', () => { const reg = new SessionWaitRegistry({ maxWaitersPerSession: 2, maxWaitersTotal: 100 }); void reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); void reg.waitForOutput('s1', { match: 'done', timeoutMs: 5000 }); expect(reg.waiterCount('s1')).toBe(2); expect(() => reg.waitForOutput('s1', { match: 'x', timeoutMs: 5000 })).toThrow(WaitCapacityError); reg.cancelEverything(); }); it('frees capacity as waiters resolve', async () => { const reg = new SessionWaitRegistry({ maxWaitersPerSession: 1, maxWaitersTotal: 100 }); const first = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); expect(() => reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 })).toThrow(WaitCapacityError); reg.notifySignal('s1', 'stop'); await first; expect(() => reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 })).not.toThrow(); reg.cancelEverything(); }); it('an immediate resolve is never rejected by a full pool', async () => { // An already-satisfied wait holds no resource, so it must not be capacity-checked. const reg = new SessionWaitRegistry({ maxWaitersPerSession: 1, maxWaitersTotal: 1 }); void reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); const result = await reg.waitForSignal('s1', { until: ['idle'], timeoutMs: 5000, currentSignal: 'idle', }); expect(result.immediate).toBe(true); reg.cancelEverything(); }); it('rejects past the per-owner cap with scope "owner", across sessions', () => { // One user must not be able to occupy the whole process-wide pool and deny the // primitive to everyone else, admin included. const reg = new SessionWaitRegistry({ maxWaitersPerSession: 10, maxWaitersPerOwner: 2, maxWaitersTotal: 100 }); void reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000, owner: 'alice' }); void reg.waitForOutput('s2', { match: 'x', timeoutMs: 5000, owner: 'alice' }); expect(reg.ownerWaiterCount('alice')).toBe(2); try { reg.waitForSignal('s3', { until: ['stop'], timeoutMs: 5000, owner: 'alice' }); expect.unreachable('third wait should have been rejected'); } catch (err) { expect(err).toBeInstanceOf(WaitCapacityError); expect((err as WaitCapacityError).scope).toBe('owner'); expect((err as Error).message).toContain('owner'); } // Another user is unaffected, which is the entire point. expect(() => reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000, owner: 'bob' })).not.toThrow(); reg.cancelEverything(); }); it('ignores the owner cap when no owner is given (single-user mode)', () => { const reg = new SessionWaitRegistry({ maxWaitersPerSession: 10, maxWaitersPerOwner: 1, maxWaitersTotal: 100 }); for (let i = 0; i < 5; i++) { expect(() => reg.waitForSignal(`s${i}`, { until: ['stop'], timeoutMs: 5000 })).not.toThrow(); } reg.cancelEverything(); }); it('frees owner capacity as waiters resolve, time out and abort', async () => { const reg = new SessionWaitRegistry({ maxWaitersPerSession: 10, maxWaitersPerOwner: 3, maxWaitersTotal: 100 }); const resolved = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000, owner: 'alice' }); const timedOut = reg.waitForSignal('s2', { until: ['stop'], timeoutMs: 20, owner: 'alice' }); const controller = new AbortController(); const aborted = reg.waitForOutput('s3', { match: 'x', timeoutMs: 5000, owner: 'alice', abortSignal: controller.signal, }); expect(reg.ownerWaiterCount('alice')).toBe(3); reg.notifySignal('s1', 'stop'); controller.abort(); await Promise.all([resolved, timedOut, aborted]); expect(reg.ownerWaiterCount('alice')).toBe(0); expect(() => reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000, owner: 'alice' })).not.toThrow(); reg.cancelEverything(); }); it('does not double-release an owner slot when an abort races a resolve', async () => { const reg = new SessionWaitRegistry({ maxWaitersPerSession: 10, maxWaitersPerOwner: 5, maxWaitersTotal: 100 }); const held = reg.waitForSignal('s1', { until: ['exit'], timeoutMs: 5000, owner: 'alice' }); const controller = new AbortController(); const racing = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000, owner: 'alice', abortSignal: controller.signal, }); reg.notifySignal('s1', 'stop'); controller.abort(); await racing; // A second release would have dropped the count to 0 and handed out a slot the // still-pending waiter is using. expect(reg.ownerWaiterCount('alice')).toBe(1); reg.cancelAll('s1'); await held; expect(reg.ownerWaiterCount('alice')).toBe(0); }); it('assertCapacity is public, so a route can refuse BEFORE doing expensive work', () => { const reg = new SessionWaitRegistry({ maxWaitersPerSession: 1, maxWaitersTotal: 100 }); expect(() => reg.assertCapacity('s1')).not.toThrow(); void reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); expect(() => reg.assertCapacity('s1')).toThrow(WaitCapacityError); reg.cancelEverything(); }); it('checks capacity BEFORE scanning initialText', async () => { // from=buffer hands us up to MAX_BUFFER_SCAN_BYTES that the route produced by // joining the whole 32MB accumulator. Paying that for a request about to be // refused turns the cap into an amplifier instead of a protection. const reg = new SessionWaitRegistry({ maxWaitersPerSession: 1, maxWaitersTotal: 100 }); void reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); expect(() => reg.waitForOutput('s1', { match: 'PRESENT', timeoutMs: 5000, initialText: 'already PRESENT here' }) ).toThrow(WaitCapacityError); reg.cancelEverything(); }); it('reports the scope in the message for every cap', () => { const reg = new SessionWaitRegistry({ maxWaitersPerSession: 1, maxWaitersPerOwner: 1, maxWaitersTotal: 2 }); void reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000, owner: 'alice' }); expect(() => reg.assertCapacity('s1', 'alice')).toThrow(/scope: owner/); expect(() => reg.assertCapacity('s1')).toThrow(/scope: session/); void reg.waitForSignal('s2', { until: ['stop'], timeoutMs: 5000 }); expect(() => reg.assertCapacity('s3')).toThrow(/scope: total/); reg.cancelEverything(); }); }); describe('SessionWaitRegistry: output waits', () => { it('matches a literal string in a chunk', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'BUILD OK', timeoutMs: 5000 }); expect(reg.notifyOutput('s1', 'running tests...\nBUILD OK\n')).toBe(1); const result = await promise; expect(result.matched).toBe(true); expect(result.immediate).toBe(false); expect(result.snippet).toContain('BUILD OK'); }); it('strips ANSI before matching', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'BUILD OK', timeoutMs: 5000 }); reg.notifyOutput('s1', '\x1b[32mBUILD\x1b[0m OK\n'); expect((await promise).matched).toBe(true); }); it('matches across a chunk boundary', async () => { // The carry buffer exists for exactly this: PTY chunking is arbitrary. const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'BUILD OK', timeoutMs: 5000 }); expect(reg.notifyOutput('s1', 'tail of a line ... BUIL')).toBe(0); expect(reg.notifyOutput('s1', 'D OK and more')).toBe(1); const result = await promise; expect(result.matched).toBe(true); expect(result.snippet).toContain('BUILD OK'); }); it('matches when an ANSI escape is SPLIT across two chunks', async () => { // A PTY read boundary lands inside an escape all the time under tmux. stripAnsi // needs a complete sequence, so the fragment used to survive the strip, land in // the carry and split the needle: the same output matched or not depending on // where the kernel cut the read. const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'OK', timeoutMs: 5000 }); expect(reg.notifyOutput('s1', 'O\x1b[')).toBe(0); expect(reg.notifyOutput('s1', '0mK\n')).toBe(1); const result = await promise; expect(result.matched).toBe(true); expect(result.snippet).toContain('OK'); }); it('matches when the escape is split at every offset inside the sequence', async () => { const sequence = '\x1b[1;32m'; for (let cut = 1; cut < sequence.length; cut++) { const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'DONE', timeoutMs: 5000 }); reg.notifyOutput('s1', `DO${sequence.slice(0, cut)}`); reg.notifyOutput('s1', `${sequence.slice(cut)}NE`); expect((await promise).matched, `split after ${cut} chars`).toBe(true); } }); it('matches when an OSC sequence is split across chunks', async () => { // tmux emits ESC ] 0 ; ESC \ on every pane-title change, and the ST // terminator is itself an ESC, which a naive "last ESC is pending" rule mishandles. const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'READY', timeoutMs: 5000 }); expect(reg.notifyOutput('s1', 'REA\x1b]0;some pane title')).toBe(0); expect(reg.notifyOutput('s1', '\x1b\\DY')).toBe(1); expect((await promise).matched).toBe(true); }); it('matches when a charset-select escape is split across chunks', async () => { // ESC ( B is only removed once its FINAL byte arrives, so a cut between the '(' and // the 'B' smuggles an escape into the haystack: neither half is a complete sequence // on its own, and holding back only a lone trailing ESC does not cover it. This cut // point is the one the widened pending rule exists for. const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'MARKER', timeoutMs: 5000 }); expect(reg.notifyOutput('s1', 'MAR\x1b(')).toBe(0); expect(reg.notifyOutput('s1', 'BKER')).toBe(1); expect((await promise).matched).toBe(true); }); it('matches when a DCS string is split across chunks', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'ABCD', timeoutMs: 5000 }); expect(reg.notifyOutput('s1', 'AB\x1bPsome')).toBe(0); expect(reg.notifyOutput('s1', '-dcs\x1b\\CD')).toBe(1); expect((await promise).matched).toBe(true); }); it('releases a held-back fragment that turns out not to be an escape', async () => { // A lone trailing ESC is withheld; if the next chunk shows it was never the start of // a sequence we remove, the text after it must still reach the haystack rather than // being withheld behind it. const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'HELLO', timeoutMs: 5000 }); expect(reg.notifyOutput('s1', 'X\x1b')).toBe(0); expect(reg.notifyOutput('s1', '\nHELLO')).toBe(1); expect((await promise).matched).toBe(true); }); it('does not withhold an unterminated escape forever', async () => { // An OSC that never terminates would otherwise stall every match on the session. const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'MARKER', timeoutMs: 5000 }); reg.notifyOutput('s1', `\x1b]0;${'x'.repeat(600)}`); reg.notifyOutput('s1', 'MARKER'); expect((await promise).matched).toBe(true); }); it('drops the held-back fragment when the last waiter goes away', async () => { const reg = new SessionWaitRegistry(); const first = reg.waitForOutput('s1', { match: 'never', timeoutMs: 20 }); reg.notifyOutput('s1', 'text\x1b['); await first; // Nothing per-session may outlive the waiter set. A leaked '\x1b[' would prefix // the next chunk and be stripped away together with the '0m' the needle wants. const next = reg.waitForOutput('s1', { match: '0mFOUND', timeoutMs: 5000 }); expect(reg.notifyOutput('s1', '0mFOUND')).toBe(1); expect((await next).matched).toBe(true); }); it('does not re-match text already scanned in a previous chunk', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'zzz', timeoutMs: 30 }); reg.notifyOutput('s1', 'aaa bbb ccc'); reg.notifyOutput('s1', 'ddd eee fff'); expect((await promise).timedOut).toBe(true); }); it('honors nocase but reports the original-case text', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'build ok', nocase: true, timeoutMs: 5000 }); reg.notifyOutput('s1', '>>> BUILD OK <<<'); const result = await promise; expect(result.matched).toBe(true); expect(result.snippet).toContain('BUILD OK'); }); it('is case-sensitive by default', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'build ok', timeoutMs: 30 }); reg.notifyOutput('s1', 'BUILD OK'); expect((await promise).timedOut).toBe(true); }); it('keeps the nocase snippet on the match when lowercasing changes length', async () => { // 'İ' (U+0130) lowercases to TWO code units, so the index found in the lowercased // haystack is not an index into the original. With enough of them ahead of the // match the window slid clean off it and the snippet came back empty. const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'done', nocase: true, timeoutMs: 5000 }); reg.notifyOutput('s1', `${'\u0130'.repeat(200)}${'x'.repeat(300)} DONE marker`); const result = await promise; expect(result.matched).toBe(true); expect(result.snippet).toContain('DONE marker'); }); it('reports original-case context around a length-shifted match', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'result', nocase: true, timeoutMs: 5000 }); reg.notifyOutput('s1', `${'\u0130'.repeat(40)}before RESULT after`); const snippet = (await promise).snippet ?? ''; expect(snippet).toContain('before RESULT after'); }); it('carries enough text for a nocase needle that lowercases longer than it is', async () => { // needleCmp, not needle, sizes the carry: a needle of 'İ' compares as two code // units, so two code units of pane output satisfy one needle character and a carry // measured from the original needle drops the straddle. const reg = new SessionWaitRegistry(); const needle = '\u0130'.repeat(120); const printed = 'i\u0307'.repeat(120); const promise = reg.waitForOutput('s1', { match: needle, nocase: true, timeoutMs: 5000 }); expect(reg.notifyOutput('s1', printed.slice(0, printed.length - 1))).toBe(0); expect(reg.notifyOutput('s1', printed.slice(printed.length - 1))).toBe(1); expect((await promise).matched).toBe(true); }); it('scans initialText first and resolves immediately (from=buffer)', async () => { const reg = new SessionWaitRegistry(); const result = await reg.waitForOutput('s1', { match: 'BUILD OK', timeoutMs: 5000, initialText: 'earlier output\n\x1b[32mBUILD OK\x1b[0m\n', }); expect(result.matched).toBe(true); expect(result.immediate).toBe(true); expect(result.waitedMs).toBe(0); expect(reg.totalWaiterCount()).toBe(0); }); it('carries the tail of initialText so a match can straddle buffer and stream', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'BUILD OK', timeoutMs: 5000, initialText: 'stuff that ends with BUIL', }); expect(reg.notifyOutput('s1', 'D OK')).toBe(1); expect((await promise).matched).toBe(true); }); it('collapses blank runs in the snippet so pane padding does not eat the context', async () => { // A real pane pads with dozens of \r\n between the prompt and the output; without // collapsing, the whole 80-char context window is newlines. const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'MARKER', timeoutMs: 5000 }); reg.notifyOutput('s1', `prompt$${'\n\r'.repeat(30)}echo MARKER`); const snippet = (await promise).snippet ?? ''; expect(snippet).toContain('echo MARKER'); expect(snippet).not.toMatch(/[\r\n]{2}/); // Context on the far side of the padding survives the collapse. expect(snippet).toContain('prompt$'); }); it('matches across the charset-select escape a real bash prompt emits', async () => { // The E2E transcript: a stock prompt renders `arkon@tnode:~/dir$` and emits // `arkon@tnode\x1b(B\x1b[m:`. stripAnsi removes the CSI and leaves ESC ( B, so // `match=tnode:` failed on a prompt that plainly reads `tnode:` while `match=(B` // succeeded. Silent: a full-length timeout, no error anywhere. const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'tnode:~/codeman-cases', timeoutMs: 5000 }); reg.notifyOutput('s1', 'clear\narkon@tnode\x1b(B\x1b[m:~/codeman-cases/e2edocs-shell$ '); expect((await promise).matched).toBe(true); }); it('leaves no charset residue in the snippet either', async () => { // The same defect seen from the other side: the ESC was removed but the literal // "(B" stayed as visible text in the snippet handed to the agent. const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'MARKER', timeoutMs: 5000 }); reg.notifyOutput('s1', 'arkon@tnode\x1b(B:/tmp/e2e\x1b(B$ MARKER done'); const snippet = (await promise).snippet ?? ''; expect(snippet).toContain('arkon@tnode:/tmp/e2e$ MARKER done'); expect(snippet).not.toContain('(B'); }); it('matches across the other escape families stripAnsi does not know', async () => { const cases: Array<[string, string]> = [ ['ESC c full reset', 'AB\x1bcCD'], ['ESC 7 / ESC 8 cursor save', 'AB\x1b7CD\x1b8'], ['ESC ( 0 line drawing', 'AB\x1b(0CD'], ['ESC # 8 DEC alignment', 'AB\x1b#8CD'], ['DCS string', 'AB\x1bPsome-dcs\x1b\\CD'], ['APC string', 'AB\x1b_Gfile=1\x1b\\CD'], ['PM string', 'AB\x1b^private\x1b\\CD'], ]; for (const [label, printed] of cases) { const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'ABCD', timeoutMs: 5000 }); reg.notifyOutput('s1', printed); expect((await promise).matched, label).toBe(true); } }); it('does not delete real output after an UNTERMINATED string sequence', async () => { // The terminator is required on purpose: a greedy match would swallow every byte to // the end of the window, which is the output an agent is waiting for. const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'IMPORTANT', timeoutMs: 5000 }); reg.notifyOutput('s1', `\x1bP${'x'.repeat(600)} IMPORTANT\n`); expect((await promise).matched).toBe(true); }); it('leaves ordinary text alone', async () => { // The residue pass is anchored on ESC, so nothing without one can be eaten. const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'a(B)c [0m ]0; #8 P^_', timeoutMs: 5000 }); reg.notifyOutput('s1', 'literal a(B)c [0m ]0; #8 P^_ text'); expect((await promise).matched).toBe(true); }); it('never ships raw control bytes in the snippet', async () => { // stripAnsi handles three escape families and nothing else, so ESC ( B (real, seen // in a live bash pane) and ESC c (a full terminal reset) used to reach the calling // agent's own terminal through `jq -r .data.snippet`. const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'MARKER', timeoutMs: 5000 }); reg.notifyOutput('s1', 'user@host\x1b(B:/tmp\x1bc \x07\x00MARKER\x1b(0qq done\n'); const snippet = (await promise).snippet ?? ''; expect(snippet).toContain('MARKER'); expect(snippet).toContain('user@host'); // eslint-disable-next-line no-control-regex expect(snippet).not.toMatch(/[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]/); }); it('keeps ordinary whitespace in the snippet', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'MARKER', timeoutMs: 5000 }); reg.notifyOutput('s1', 'col1\tcol2\nMARKER here'); const snippet = (await promise).snippet ?? ''; expect(snippet).toContain('col1\tcol2'); expect(snippet).toContain('\n'); }); it('bounds the snippet around the match', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'NEEDLE', timeoutMs: 5000 }); reg.notifyOutput('s1', `${'x'.repeat(500)}NEEDLE${'y'.repeat(500)}`); const snippet = (await promise).snippet ?? ''; expect(snippet).toContain('NEEDLE'); expect(snippet.length).toBeLessThan(300); }); it('times out without erroring', async () => { const reg = new SessionWaitRegistry(); const result = await reg.waitForOutput('s1', { match: 'never', timeoutMs: 20 }); expect(result.timedOut).toBe(true); expect(result.matched).toBe(false); expect(result.snippet).toBeNull(); expect(reg.totalWaiterCount()).toBe(0); }); it('cancelAll resolves output waiters with ended', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForOutput('s1', { match: 'never', timeoutMs: 5000 }); expect(reg.cancelAll('s1')).toBe(1); const result = await promise; expect(result.ended).toBe(true); expect(result.matched).toBe(false); }); it('notifyOutput is a no-op with no waiters', () => { const reg = new SessionWaitRegistry(); expect(reg.notifyOutput('s1', 'anything at all')).toBe(0); }); it('resolves only the waiters whose needle matched', async () => { const reg = new SessionWaitRegistry(); const a = reg.waitForOutput('s1', { match: 'one', timeoutMs: 5000 }); const b = reg.waitForOutput('s1', { match: 'two', timeoutMs: 5000 }); expect(reg.outputWaiterCount('s1')).toBe(2); expect(reg.notifyOutput('s1', 'one')).toBe(1); expect((await a).matched).toBe(true); expect(reg.outputWaiterCount('s1')).toBe(1); expect(reg.notifyOutput('s1', 'two')).toBe(1); expect((await b).matched).toBe(true); expect(reg.totalWaiterCount()).toBe(0); }); }); describe('SessionWaitRegistry: teardown and leak prevention', () => { it('drops the per-session bookkeeping once the last waiter resolves', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 5000 }); expect(reg.totalWaiterCount()).toBe(1); reg.notifySignal('s1', 'stop'); await promise; expect(reg.signalWaiterCount('s1')).toBe(0); expect(reg.waiterCount('s1')).toBe(0); expect(reg.totalWaiterCount()).toBe(0); }); it('a timed-out waiter is removed, not left in the map', async () => { const reg = new SessionWaitRegistry(); await reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 20 }); await reg.waitForOutput('s1', { match: 'x', timeoutMs: 20 }); expect(reg.totalWaiterCount()).toBe(0); }); it('cancelEverything resolves every waiter across every session', async () => { const reg = new SessionWaitRegistry(); const promises = [ reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 60_000 }), reg.waitForSignal('s2', { until: ['idle'], timeoutMs: 60_000 }), reg.waitForOutput('s2', { match: 'nope', timeoutMs: 60_000 }), ]; expect(reg.totalWaiterCount()).toBe(3); expect(reg.cancelEverything()).toBe(3); const results = await Promise.all(promises); for (const result of results) expect(result.ended).toBe(true); expect(reg.totalWaiterCount()).toBe(0); }); it('a resolved waiter is not re-settled when its timeout would have fired', async () => { const reg = new SessionWaitRegistry(); const promise = reg.waitForSignal('s1', { until: ['stop'], timeoutMs: 25 }); reg.notifySignal('s1', 'stop'); const result = await promise; await sleep(60); expect(result.signal).toBe('stop'); expect(result.timedOut).toBe(false); expect(reg.totalWaiterCount()).toBe(0); }); });