diff --git a/changelog.d/feat-gate-message-quiet-period.added.md b/changelog.d/feat-gate-message-quiet-period.added.md new file mode 100644 index 000000000..27f0d1000 --- /dev/null +++ b/changelog.d/feat-gate-message-quiet-period.added.md @@ -0,0 +1,12 @@ +- Opt-in `FrameworkConfig.messageQuietPeriod: { quietMs, maxWaitMs? }`: an + idle agent's turn waits while message wakes for it keep arriving, so a + burst (a caption and its attachments, a reply split into several messages) + starts one turn instead of one per part. The scheduler holds the start + until `quietMs` have passed since the agent's newest message wake, or + `maxWaitMs` (default 30 s) since its oldest held one, then starts one turn + with everything queued, as it merges queued wakes today. Each message's + own decision stays what the gate gives it: who wakes a sleeping agent, + where a reply goes, what tune-out and conversation routing see. A batch + holding any other wake (a timer, an operator, a tool result, a + continuation or a retry of the agent's turn) starts now; the heartbeat is + no message wake. (#179, carried by #856.) diff --git a/src/framework.ts b/src/framework.ts index 79dbddd41..4412a572c 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -881,6 +881,53 @@ const turnRearmDelays = (value: unknown): readonly number[] => { return [...value]; }; +/** FrameworkConfig.messageQuietPeriod, read once: both numbers finite and + * 0 or more, maxWaitMs defaulting to 30 s; anything else refuses the config. */ +const quietPeriodOf = (value: unknown): { quietMs: number; maxWaitMs: number } | undefined => { + if (value === undefined) return undefined; + const bad = (): never => { + throw new Error('FrameworkConfig.messageQuietPeriod must be { quietMs, maxWaitMs? }, each a finite number ' + + `of ms, 0 or more, not ${shownValue(value)}`); + }; + if (!value || typeof value !== 'object' || Array.isArray(value)) return bad(); + const { quietMs, maxWaitMs } = value as { quietMs?: unknown; maxWaitMs?: unknown }; + if (typeof quietMs !== 'number' || !Number.isFinite(quietMs) || quietMs < 0) return bad(); + if (maxWaitMs !== undefined && (typeof maxWaitMs !== 'number' || !Number.isFinite(maxWaitMs) || maxWaitMs < 0)) return bad(); + return { quietMs, maxWaitMs: maxWaitMs ?? 30_000 }; +}; + +/** + * How much longer an idle agent's turn waits for its message burst to end + * (FrameworkConfig.messageQuietPeriod), in ms; 0 starts it now. The batch + * waits only while every wake in it is a message wake: a channel message's + * or a server event's (other than the heartbeat's), and none a continuation + * or a retry of the agent's turn. It ends `quietMs` after the newest message + * wake, or `maxWaitMs` after the oldest, whichever is first. + */ +export function quietHoldRemaining( + requests: readonly InferenceRequest[], + now: number, + quiet: { quietMs: number; maxWaitMs: number } | undefined, +): number { + if (!quiet || quiet.quietMs <= 0 || requests.length === 0) return 0; + let newest = -Infinity; + let oldest = Infinity; + for (const r of requests) { + if (!isMessageWake(r)) return 0; + if (r.timestamp > newest) newest = r.timestamp; + if (r.timestamp < oldest) oldest = r.timestamp; + } + return Math.max(0, Math.min(quiet.quietMs - (now - newest), quiet.maxWaitMs - (now - oldest))); +} + +/** A wake a message made: a channel message's or a server event's, other + * than the heartbeat's, that neither continues nor retries a turn. */ +function isMessageWake(r: InferenceRequest): boolean { + return (r.reason === 'mcpl:channel-incoming' || r.reason === 'mcpl:push-event') + && r.source !== 'heartbeat' + && !continuesTurn(r) && !carriesContinuation(r) && mergedBatch(r) === undefined; +} + const settingBoolean = (name: string, value: unknown): boolean => { if (typeof value === 'boolean') return value; if (value === 'true' || value === 'false') return value === 'true'; @@ -3882,6 +3929,10 @@ export class AgentFramework { * Drives hard-down escalation — see noteInferenceExhausted. */ private consecutiveInferenceFailures: Map = new Map(); private turnRearmMs: readonly number[] = DEFAULT_TURN_REARM_MS; + /** FrameworkConfig.messageQuietPeriod, validated; undefined when off. */ + private quietPeriod: { quietMs: number; maxWaitMs: number } | undefined; + /** Agents whose turn start a quiet period holds now, for one log line per burst. */ + private quietHeldAgents = new Set(); private turnRearms: Map = new Map(); private exhaustingRearms: Set = new Set(); private rearmDeadlineCuts: Set = new Set(); @@ -4689,6 +4740,7 @@ export class AgentFramework { ); framework.providerHoldHook = config.providerHold; framework.turnRearmMs = turnRearmDelays(config.exhaustedTurnRearmMs); + framework.quietPeriod = quietPeriodOf(config.messageQuietPeriod); framework.inferenceBounds = resolveInferenceBounds(config.inferenceBounds); // Before any delivery. @@ -19725,6 +19777,27 @@ export class AgentFramework { continue; } + // A message quiet period (FrameworkConfig.messageQuietPeriod): an idle + // agent's turn waits while message wakes for it keep arriving, and + // starts once with everything queued (the merge below). Each wake + // keeps what the gate and the ingress gave it; only the start waits. + // The loop's next pass reads the queue again (runLoop's backoff), so + // the requests go back as a busy agent's do, with no timer of their + // own. A batch holding any other wake starts now (quietHoldRemaining). + // The set is read as surgeryHold is: a prototype-built harness leaves + // both slots undefined, which reads as "not held". + const quietLeft = quietHoldRemaining(requests, now, this.quietPeriod); + if (quietLeft > 0) { + const held = (this.quietHeldAgents ??= new Set()); + if (!held.has(agentName)) { + held.add(agentName); + console.error(`[quiet-period] agent=${agentName}: holding the turn for a message burst (${requests.length} wake(s) so far)`); + } + this.pendingRequests.push(...requests); + continue; + } + this.quietHeldAgents?.delete(agentName); + // A wake whose sole cause was a write deferred for this agent // (readsDeferredWrite), or a message stored for it as it arrived // (readsStoredMessage), has nothing left to do once a request that diff --git a/src/index.ts b/src/index.ts index 49c4a448c..1ebf95dc6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,7 +2,7 @@ export * from './types/index.js'; // Core classes -export { AgentFramework, AgentStoppedError, BudgetPreflightError, DefaultErrorPolicy, ProviderWaitNotSleptError, QuiesceBlockedError, ResumeBlockedError } from './framework.js'; +export { AgentFramework, AgentStoppedError, BudgetPreflightError, DefaultErrorPolicy, ProviderWaitNotSleptError, QuiesceBlockedError, ResumeBlockedError, quietHoldRemaining } from './framework.js'; export type { RuntimeSettingsPreview, HostModeStatus, DeriveAgentOptions } from './framework.js'; export { Agent } from './agent.js'; export { FLAT_MINIMUM_MS, type InferenceBoundsOptions } from './inference-bounds.js'; diff --git a/src/types/framework.ts b/src/types/framework.ts index 71e1bbd97..1626c4150 100644 --- a/src/types/framework.ts +++ b/src/types/framework.ts @@ -400,6 +400,22 @@ export interface FrameworkConfig { * `abortInference()` doesn't end it. */ exhaustedTurnRearmMs?: number[] | false; + /** + * A message quiet period: an idle agent's turn waits while message wakes + * for it keep arriving, so a burst (a caption and its images, a reply split + * into several messages) starts one turn, not one per part. The scheduler + * holds the start until `quietMs` have passed since the agent's newest + * message wake, or `maxWaitMs` (default 30000) since its oldest held one, + * then starts one turn with everything queued, as it merges queued wakes + * today. A message wake is a channel message's or a server event's + * (`mcpl:channel-incoming`, `mcpl:push-event`) other than the heartbeat's. + * Each message's own decision stays what the gate gives it: who wakes a + * sleeping agent, where a reply goes, what tune-out and conversation + * routing see. A batch holding any other wake (a timer, an operator, a + * tool result, a continuation or a retry of the agent's turn) starts now. + * Off when omitted. + */ + messageQuietPeriod?: { quietMs: number; maxWaitMs?: number }; /** Interval for periodic store sync in milliseconds (default: 1000ms, 0 to disable) */ syncIntervalMs?: number; diff --git a/test/message-quiet-period.test.ts b/test/message-quiet-period.test.ts new file mode 100644 index 000000000..c850a9c28 --- /dev/null +++ b/test/message-quiet-period.test.ts @@ -0,0 +1,189 @@ +/** + * FrameworkConfig.messageQuietPeriod: an idle agent's turn waits while + * message wakes for it keep arriving, and starts once with everything queued. + * + * The hold lives where the scheduler starts a turn (runInferencePass), not in + * the gate: each message keeps the decision the gate gave it, so who wakes a + * sleeping agent, where a reply goes, and what tune-out and conversation + * routing see are as on `main` (#856's review). What the hold decides is + * `quietHoldRemaining`, pinned here on fixed clocks; the scheduler's use of it + * runs against a real framework with wide margins. + */ +import { describe, it, afterEach } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdirSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; +import { AgentFramework, quietHoldRemaining } from '../src/index.js'; +import type { InferenceRequest } from '../src/types/agent.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; + +const TMP = join(import.meta.dirname, '../.test-tmp-quiet-period'); +const T0 = 1_000_000; +const QUIET = { quietMs: 2_000, maxWaitMs: 10_000 }; + +const message = (at: number, extra: Partial = {}): InferenceRequest => ({ + agentName: 'agent', reason: 'mcpl:channel-incoming', source: 'discord', timestamp: at, ...extra, +}); + +describe('quietHoldRemaining', () => { + it('is 0 with the option off, or with nothing queued', () => { + assert.equal(quietHoldRemaining([message(T0)], T0 + 100, undefined), 0); + assert.equal(quietHoldRemaining([], T0, QUIET), 0); + assert.equal(quietHoldRemaining([message(T0)], T0 + 100, { quietMs: 0, maxWaitMs: 10_000 }), 0); + }); + + it('holds a burst for quietMs after its newest message wake', () => { + const burst = [message(T0), message(T0 + 500), message(T0 + 900)]; + assert.equal(quietHoldRemaining(burst, T0 + 1_000, QUIET), 1_900, 'quietMs counted from the newest'); + assert.equal(quietHoldRemaining(burst, T0 + 2_899, QUIET), 1); + assert.equal(quietHoldRemaining(burst, T0 + 2_900, QUIET), 0, 'quiet: the turn starts'); + }); + + it('ends the hold maxWaitMs after the oldest message wake, however the burst goes on', () => { + const stream = [message(T0), message(T0 + 9_000), message(T0 + 9_900)]; + assert.equal(quietHoldRemaining(stream, T0 + 9_950, QUIET), 50, 'the cap, not quietMs, is what is left'); + assert.equal(quietHoldRemaining(stream, T0 + 10_000, QUIET), 0); + }); + + it('a maxWaitMs below quietMs is the cap', () => { + assert.equal(quietHoldRemaining([message(T0)], T0 + 100, { quietMs: 2_000, maxWaitMs: 300 }), 200); + }); + + it('a server event is a message wake; the heartbeat is not', () => { + const push = message(T0, { reason: 'mcpl:push-event', source: 'tsm' }); + assert.equal(quietHoldRemaining([push], T0 + 100, QUIET), 1_900); + const tick = message(T0, { reason: 'mcpl:push-event', source: 'heartbeat' }); + assert.equal(quietHoldRemaining([tick], T0 + 100, QUIET), 0); + assert.equal(quietHoldRemaining([message(T0), tick], T0 + 100, QUIET), 0, 'a tick in the batch starts it'); + }); + + it('any other wake in the batch starts the turn now', () => { + for (const other of [ + message(T0, { reason: 'self-wake', source: 'gate' }), + message(T0, { reason: 'external-message', source: 'api' }), + message(T0, { reason: 'operator', source: 'api', addressed: true }), + message(T0, { reason: 'gate:debounce', source: 'gate' }), + ]) { + assert.equal(quietHoldRemaining([message(T0), other], T0 + 100, QUIET), 0, other.reason); + } + }); + + it('a continuation or a retry of the turn is not held', () => { + const restart = message(T0, { reason: 'context_budget_restart', source: 'framework' }); + assert.equal(quietHoldRemaining([restart], T0 + 100, QUIET), 0); + const guard = message(T0, { reason: 'tool_result_guard_retry', source: 'framework' }); + assert.equal(quietHoldRemaining([guard], T0 + 100, QUIET), 0); + const auto = message(T0, { autoContinueRun: { stop: 'call skip_reply' } }); + assert.equal(quietHoldRemaining([auto], T0 + 100, QUIET), 0); + }); +}); + +describe('messageQuietPeriod in the scheduler', () => { + afterEach(() => rmSync(TMP, { recursive: true, force: true })); + + async function start(quiet?: { quietMs: number; maxWaitMs?: number }) { + mkdirSync(TMP, { recursive: true }); + const membrane = new MockMembrane(); + const framework = await AgentFramework.create({ + storePath: join(TMP, `store-${Math.random().toString(36).slice(2)}`), + membrane: membrane.asMembrane(), + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test' }], + modules: [], + ...(quiet ? { messageQuietPeriod: quiet } : {}), + }); + const internals = framework as unknown as { pendingRequests: InferenceRequest[] }; + framework.start(); + return { framework, membrane, internals }; + } + const wake = (internals: { pendingRequests: InferenceRequest[] }, extra: Partial = {}) => + internals.pendingRequests.push(message(Date.now(), extra)); + const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + const waitFor = async (cond: () => boolean, what: string, timeoutMs: number) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (cond()) return; + await sleep(20); + } + throw new Error(`timed out waiting for: ${what}`); + }; + + it('a burst of message wakes starts one turn, once the burst is quiet', async () => { + const { framework, membrane, internals } = await start({ quietMs: 600 }); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'one turn' }])); + try { + const t0 = Date.now(); + wake(internals); await sleep(100); + wake(internals); await sleep(100); + wake(internals); + await sleep(300); + assert.equal(membrane.calls.length, 0, 'no turn while the burst goes on'); + await waitFor(() => membrane.calls.length >= 1, 'the turn', 3_000); + assert.ok(Date.now() - t0 >= 700, 'the turn started after the quiet period'); + await sleep(400); + assert.equal(membrane.calls.length, 1, 'one turn for the burst'); + } finally { + await framework.stop(); + } + }); + + it('with the option off, the same wakes start a turn at once', async () => { + const { framework, membrane, internals } = await start(); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'now' }])); + try { + const t0 = Date.now(); + wake(internals); + await waitFor(() => membrane.calls.length >= 1, 'the turn', 3_000); + assert.ok(Date.now() - t0 < 500, 'started without a hold'); + } finally { + await framework.stop(); + } + }); + + it('a stream that never goes quiet starts a turn at maxWaitMs', async () => { + const { framework, membrane, internals } = await start({ quietMs: 600, maxWaitMs: 900 }); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'capped' }])); + try { + const t0 = Date.now(); + while (Date.now() - t0 < 1_500 && membrane.calls.length === 0) { + wake(internals); + await sleep(150); + } + assert.ok(membrane.calls.length >= 1, 'the cap started the turn'); + assert.ok(Date.now() - t0 < 1_500, 'before the stream ended'); + } finally { + await framework.stop(); + } + }); + + it('a wake that is no message starts the held turn now', async () => { + const { framework, membrane, internals } = await start({ quietMs: 2_000 }); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'now' }])); + try { + const t0 = Date.now(); + wake(internals); + await sleep(100); + wake(internals, { reason: 'operator', source: 'api', addressed: true }); + await waitFor(() => membrane.calls.length >= 1, 'the turn', 3_000); + assert.ok(Date.now() - t0 < 1_000, 'the operator wake ended the hold'); + } finally { + await framework.stop(); + } + }); + + it('refuses a quiet period that is not finite numbers of ms', async () => { + mkdirSync(TMP, { recursive: true }); + for (const bad of [{ quietMs: Number.NaN }, { quietMs: -1 }, { quietMs: 10, maxWaitMs: 'soon' }, 'later', { maxWaitMs: 5 }]) { + await assert.rejects( + AgentFramework.create({ + storePath: join(TMP, `store-bad-${Math.random().toString(36).slice(2)}`), + membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test' }], + modules: [], + messageQuietPeriod: bad as unknown as { quietMs: number }, + }), + /messageQuietPeriod must be/, + JSON.stringify(bad), + ); + } + }); +});