From ffa91d575243ff946f7d0b71ecf0707288a56c90 Mon Sep 17 00:00:00 2001 From: Ian de Marcellus <50731233+ian-de-marcellus@users.noreply.github.com> Date: Thu, 24 Sep 2026 03:57:24 +0200 Subject: [PATCH 1/4] =?UTF-8?q?feat(gate):=20messageQuietPeriod=20?= =?UTF-8?q?=E2=80=94=20one=20wake=20per=20message=20burst?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GateOptions.messageQuietPeriod { quietMs, maxWaitMs? }: a message-driven wake (MCPL channel-incoming / push-event the policies would trigger) is held until no further message has arrived for quietMs, capped at maxWaitMs (default 30 s) from the first held message, then fires once. Messages still enter context at arrival and no batch summary is added; an agent already mid-turn when the hold ends is not re-woken. Scheduled wakes, tool results and heartbeats are unaffected. Default off. Why: a caption and its forwarded images, or a reply Discord split into several messages, woke residents once per message — and in rooms with several AIs each part triggered replies from all the others. Measured gaps (4,758 messages): split parts arrive 0-2 s apart; caption+forward within ~10 s. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../feat-gate-message-quiet-period.added.md | 5 ++ src/framework.ts | 1 + src/gate/event-gate.ts | 58 +++++++++++++ src/gate/types.ts | 11 +++ test/event-gate-quiet-period.test.ts | 86 +++++++++++++++++++ 5 files changed, 161 insertions(+) create mode 100644 changelog.d/feat-gate-message-quiet-period.added.md create mode 100644 test/event-gate-quiet-period.test.ts 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..7b03d120f --- /dev/null +++ b/changelog.d/feat-gate-message-quiet-period.added.md @@ -0,0 +1,5 @@ +- Opt-in gate `messageQuietPeriod: { quietMs, maxWaitMs? }`: MCPL message + wakes are held until the sender goes quiet (default hard cap 30 s), so a + caption + attachments or a multi-part message wakes the agent once instead + of once per part. Agents already inferring are not held. Non-message wakes + (timers, tool results) are unaffected. diff --git a/src/framework.ts b/src/framework.ts index c9eb1c627..5c8c12592 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -4295,6 +4295,7 @@ export class AgentFramework { } : {}), ...(judge && judgeOptions?.timeoutMs !== undefined ? { judgeTimeoutMs: judgeOptions.timeoutMs } : {}), ...(judge && judgeOptions?.onError !== undefined ? { judgeOnError: judgeOptions.onError } : {}), + messageQuietPeriod: config.gate.messageQuietPeriod, emitTrace: (e) => framework.emitTrace(e as { type: TraceEvent['type']; [key: string]: unknown }), addMessage: (p, c, m, forAgent, placement) => framework.addMessage(p, c, m as MessageMetadata, forAgent || placement ? { ...(forAgent ? { forAgent } : {}), ...(placement ? { placement } : {}) } : undefined), diff --git a/src/gate/event-gate.ts b/src/gate/event-gate.ts index 71cea874a..11422a26c 100644 --- a/src/gate/event-gate.ts +++ b/src/gate/event-gate.ts @@ -1993,6 +1993,14 @@ export class EventGate { // Debounce state private debounceTimers = new Map(); + // Message quiet period (GateOptions.messageQuietPeriod): a message-driven + // wake is held until no further message has arrived for quietMs (capped at + // maxWaitMs from the first held one), then fires once. + private readonly quietMs: number; + private readonly quietMaxWaitMs: number; + private quietTimer: ReturnType | null = null; + private quietEvents: PendingEvent[] = []; + private quietFirstAt = 0; // Reply judge state (`judge` rules). The judge itself, when the host wires // one; without it a judge rule's batch is delivered as a debounce's. @@ -2153,7 +2161,11 @@ export class EventGate { * (WakeProvenance.line), for a fresh start of the wake's turn to decide * (WakeLine). Absent: every batched wake writes its line, as before. */ isNewestEntry?: (entry: GateEntryRef) => boolean; + /** See GateOptions.messageQuietPeriod. */ + messageQuietPeriod?: { quietMs: number; maxWaitMs?: number }; }) { + this.quietMs = Math.max(0, opts.messageQuietPeriod?.quietMs ?? 0); + this.quietMaxWaitMs = Math.max(this.quietMs, opts.messageQuietPeriod?.maxWaitMs ?? 30_000); this.configPath = opts.configPath; this.privilegedUsersPath = opts.privilegedUsersPath; this.emitTrace = opts.emitTrace; @@ -3107,6 +3119,14 @@ export class EventGate { this.defaultByEventType.set(info.eventType, bucket); } if (observed.length > 0) decision.observed = observed; + if ( + decision.trigger && + this.quietMs > 0 && + (info.eventType === 'mcpl:channel-incoming' || info.eventType === 'mcpl:push-event') + ) { + this.holdForQuietPeriod(info); + decision = { ...decision, trigger: false, quietHeld: true }; + } this.emitTrace({ type: 'gate:decision', @@ -3508,6 +3528,44 @@ export class EventGate { } } + /** Hold a message-driven wake until the senders go quiet. The message + * itself is already in context (stored at arrival); only the WAKE waits, + * so a caption and its images, or a reply split into several Discord + * messages, reach the agent as one turn instead of several. */ + private holdForQuietPeriod(info: GateEventInfo): void { + const now = this.now(); + if (this.quietEvents.length === 0) this.quietFirstAt = now; + this.quietEvents.push({ + policyName: 'quiet-period', + content: '', + eventType: info.eventType, + timestamp: now, + inContext: true, + channelId: info.channelId || undefined, + authorId: this.extractAuthorId(info.metadata) ?? undefined, + serverId: info.serverId || undefined, + addressed: Array.isArray(info.tags) && info.tags.includes('chat:addressed'), + }); + if (this.quietTimer) clearTimeout(this.quietTimer); + const wait = Math.max(0, Math.min(this.quietMs, this.quietFirstAt + this.quietMaxWaitMs - now)); + this.quietTimer = setTimeout(() => this.fireQuietPeriod(), wait); + (this.quietTimer as { unref?: () => void }).unref?.(); + } + + private fireQuietPeriod(): void { + this.quietTimer = null; + const events = this.quietEvents; + this.quietEvents = []; + if (events.length === 0) return; + const provenance = wakeProvenance(events); + for (const agentName of this.getAgentNamesFn()) { + // An agent already mid-turn received these messages as mid-turn input; + // waking it again afterwards would spend a redundant turn. + if (this.inferring.has(agentName)) continue; + this.requestInferenceFn(agentName, 'gate:quiet-period', 'gate', provenance); + } + } + private handleDebounce(policy: GatePolicy, info: GateEventInfo): void { const debounceMs = batchQuietMs(policy.behavior) ?? MIN_DEBOUNCE_MS; const judge = typeof policy.behavior === 'object' && 'judge' in policy.behavior ? policy.behavior.judge : undefined; diff --git a/src/gate/types.ts b/src/gate/types.ts index edc7eb241..5ec4f3b32 100644 --- a/src/gate/types.ts +++ b/src/gate/types.ts @@ -223,6 +223,15 @@ export interface GateOptions { * null), the judge is wired with its defaults; it runs only for `judge` * rules, so a gate with none makes no judge calls. */ judge?: ReplyJudgeOptions | null; + /** + * Hold message-driven wakes (MCPL channel-incoming / push-event that the + * policies would trigger) until no further message has arrived for + * `quietMs`, capped at `maxWaitMs` (default 30000) from the first held + * message; then wake once. Messages still enter context at arrival. Other + * wake sources (scheduled, tool results, heartbeat) are unaffected. + * Default off. + */ + messageQuietPeriod?: { quietMs: number; maxWaitMs?: number }; } /** @@ -342,6 +351,8 @@ export interface GateDecision { policyName: string | null; /** The behavior that was applied. */ behavior: GateBehavior; + /** The trigger was converted into a held quiet-period wake. */ + quietHeld?: boolean; /** * Names of `passthrough` policies that matched and counted this event but * did not fire, so evaluation continued past them. Absent when none did. diff --git a/test/event-gate-quiet-period.test.ts b/test/event-gate-quiet-period.test.ts new file mode 100644 index 000000000..104ca4906 --- /dev/null +++ b/test/event-gate-quiet-period.test.ts @@ -0,0 +1,86 @@ +/** + * GateOptions.messageQuietPeriod: a burst of messages (a caption and its + * images, a reply split across several Discord messages) wakes the agent once, + * after the senders go quiet — not once per message. + */ +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 { EventGate } from '../src/gate/event-gate.js'; +import type { GateEventInfo } from '../src/gate/types.js'; + +const TMP = join(import.meta.dirname, '../.test-tmp-gate-quiet'); + +function makeGate(quiet?: { quietMs: number; maxWaitMs?: number }) { + mkdirSync(TMP, { recursive: true }); + const wakes: Array<{ reason: string; channelId?: string }> = []; + const messages: unknown[] = []; + const gate = new EventGate({ + configPath: join(TMP, `gate-${Math.random().toString(36).slice(2)}.json`), + initialConfig: { default: 'always', policies: [] }, + emitTrace: () => {}, + addMessage: (...args) => { messages.push(args); return ''; }, + requestInference: (_agent, reason, _source, provenance) => wakes.push({ reason, channelId: provenance?.channelId }), + getAgentNames: () => ['agent'], + ...(quiet ? { messageQuietPeriod: quiet } : {}), + }); + return { gate, wakes, messages }; +} + +const msg = (text: string, extra: Partial = {}): GateEventInfo => ({ + content: text, + eventType: 'mcpl:channel-incoming', + serverId: 'discord', + channelId: 'discord:g:salon', + metadata: { authorId: 'u1' }, + ...extra, +}); +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +describe('message quiet period', () => { + afterEach(() => rmSync(TMP, { recursive: true, force: true })); + + it('holds a burst and wakes once after the senders go quiet', async () => { + const { gate, wakes, messages } = makeGate({ quietMs: 60 }); + assert.equal(gate.evaluate(msg('caption')).trigger, false); + await sleep(20); + assert.equal(gate.evaluate(msg('image 1')).trigger, false); + await sleep(20); + assert.equal(gate.evaluate(msg('image 2')).trigger, false); + assert.equal(wakes.length, 0, 'no wake while messages keep arriving'); + await sleep(90); + assert.equal(wakes.length, 1, 'one wake for the whole burst'); + assert.equal(wakes[0].reason, 'gate:quiet-period'); + assert.equal(wakes[0].channelId, 'discord:g:salon'); + assert.equal(messages.length, 0, 'no batch summary is added to context'); + }); + + it('caps the hold at maxWaitMs under a continuous stream', async () => { + const { gate, wakes } = makeGate({ quietMs: 50, maxWaitMs: 120 }); + const start = Date.now(); + while (Date.now() - start < 200) { + gate.evaluate(msg('chatter')); + await sleep(25); + if (wakes.length > 0) break; + } + assert.equal(wakes.length, 1, 'woke despite continuous messages'); + assert.ok(Date.now() - start < 190, 'within roughly the cap'); + }); + + it('does not re-wake an agent that is already mid-turn when the hold ends', async () => { + const { gate, wakes } = makeGate({ quietMs: 40 }); + gate.evaluate(msg('arrives during a turn')); + gate.onInferenceStarted('agent'); + await sleep(70); + assert.equal(wakes.length, 0); + gate.onInferenceEnded('agent'); + }); + + it('does not delay non-message wakes, and is off by default', () => { + const { gate } = makeGate({ quietMs: 60 }); + assert.equal(gate.evaluate(msg('tick', { eventType: 'external-message' })).trigger, true); + const plain = makeGate(); + assert.equal(plain.gate.evaluate(msg('hello')).trigger, true); + }); +}); From bbd1a80a900f4fad7029660ec6907408a069c26b Mon Sep 17 00:00:00 2001 From: Daryna Date: Tue, 27 Oct 2026 20:34:12 +0000 Subject: [PATCH 2/4] fix(gate): a quiet-period hold goes the way a debounce batch does (#179 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit From the 10/20 review of #179 and Greptile's 9/28 findings, on top of Ian's commit rebased onto main: - An agent mid-turn when a hold ends gets its wake after the turn, not never: the hold fires through the debounce path (fireBatch), so it is buffered for an agent in a turn or a quiesced host and delivered at onInferenceEnded / resume. (Review 1, blocking.) - The held event is shaped as a debounce policy's, routeChannelId included, so a held DM's wake routes its reply back to the DM. (Review 2, blocking.) - Holds are keyed by channel (the registered id, else the event's own), so chatter in one channel holds no other channel's wake. - dispose() clears pending holds. - A message the policies didn't trigger (a rate-limited part of the same burst) extends an open hold for its channel but opens none. - The heartbeat (serverId 'heartbeat') is never held. - maxWaitMs is the operator's cap even below quietMs; it is no longer raised to quietMs. - A hold that ends while the agent sleeps is dropped and counted as suppressed, as the messages' own wakes would have been. - deliverEvents writes no [Gate: …] line for a quiet-period batch: its messages are the newest entries right above the wake. The wake's reason is 'gate:quiet-period'; mixed with a debounce policy's events the line and reason are the debounce's. - Tests pin each of the above, the wake after onInferenceEnded (the old test 3 pinned the drop), and the framework's pass-through of the option; the changelog fragment and the option's doc say what holds. Left as `main` has it for a debounce batch, and said in the description: the wake fans out to every agent (the framework's mayTakeGateRoute keeps a fork's route to its home), and tune-out reads the held message's verdict as the gate's no. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01SkepYxcPem6pmgRxKd2RBY --- .../feat-gate-message-quiet-period.added.md | 16 +- src/gate/event-gate.ts | 161 ++++++++++++---- src/gate/types.ts | 15 +- test/event-gate-quiet-period.test.ts | 181 ++++++++++++++++-- 4 files changed, 311 insertions(+), 62 deletions(-) diff --git a/changelog.d/feat-gate-message-quiet-period.added.md b/changelog.d/feat-gate-message-quiet-period.added.md index 7b03d120f..f96a0c38a 100644 --- a/changelog.d/feat-gate-message-quiet-period.added.md +++ b/changelog.d/feat-gate-message-quiet-period.added.md @@ -1,5 +1,11 @@ -- Opt-in gate `messageQuietPeriod: { quietMs, maxWaitMs? }`: MCPL message - wakes are held until the sender goes quiet (default hard cap 30 s), so a - caption + attachments or a multi-part message wakes the agent once instead - of once per part. Agents already inferring are not held. Non-message wakes - (timers, tool results) are unaffected. +- Opt-in gate `messageQuietPeriod: { quietMs, maxWaitMs? }`: a message's + wake is held until its channel has been quiet for `quietMs` (at most + `maxWaitMs`, default 30 s, from the channel's first held message), so a + caption and its attachments, or a reply split into several messages, wake + the agent once instead of once per part. Messages still enter context at + arrival; only the wake waits, and no line is written for it. The held wake + goes as a `debounce` policy's batch does: it carries the burst's route, so + a held DM's reply goes back to the DM; an agent mid-turn gets it after the + turn; a quiesced host at resume; an agent asleep when the hold ends gets + none. Holds are per channel, so chatter in one channel delays no other's. + The heartbeat, scheduled wakes and tool results are unaffected. diff --git a/src/gate/event-gate.ts b/src/gate/event-gate.ts index 11422a26c..4f900a06d 100644 --- a/src/gate/event-gate.ts +++ b/src/gate/event-gate.ts @@ -239,6 +239,8 @@ interface RateBucket { /** Synthetic key used when a policy has no `keyBy` or the field is missing. */ const SHARED_BUCKET_KEY = '__shared__'; +/** The policy name a quiet-period hold's events carry (GateOptions.messageQuietPeriod). */ +const QUIET_PERIOD_POLICY = 'quiet-period'; interface CompiledPolicy { policy: GatePolicy; @@ -1994,13 +1996,13 @@ export class EventGate { // Debounce state private debounceTimers = new Map(); // Message quiet period (GateOptions.messageQuietPeriod): a message-driven - // wake is held until no further message has arrived for quietMs (capped at - // maxWaitMs from the first held one), then fires once. + // wake is held until no further message has arrived in its channel for + // quietMs (capped at maxWaitMs from the channel's first held one), then + // fires once, through the debounce path (fireDebounce): an agent mid-turn + // gets it after the turn, a quiesced host at resume. private readonly quietMs: number; private readonly quietMaxWaitMs: number; - private quietTimer: ReturnType | null = null; - private quietEvents: PendingEvent[] = []; - private quietFirstAt = 0; + private readonly quietHolds = new Map; events: PendingEvent[]; firstAt: number }>(); // Reply judge state (`judge` rules). The judge itself, when the host wires // one; without it a judge rule's batch is delivered as a debounce's. @@ -2165,7 +2167,9 @@ export class EventGate { messageQuietPeriod?: { quietMs: number; maxWaitMs?: number }; }) { this.quietMs = Math.max(0, opts.messageQuietPeriod?.quietMs ?? 0); - this.quietMaxWaitMs = Math.max(this.quietMs, opts.messageQuietPeriod?.maxWaitMs ?? 30_000); + // The cap is the operator's: a maxWaitMs below quietMs ends the hold at + // maxWaitMs, not at quietMs. + this.quietMaxWaitMs = Math.max(0, opts.messageQuietPeriod?.maxWaitMs ?? 30_000); this.configPath = opts.configPath; this.privilegedUsersPath = opts.privilegedUsersPath; this.emitTrace = opts.emitTrace; @@ -3119,13 +3123,24 @@ export class EventGate { this.defaultByEventType.set(info.eventType, bucket); } if (observed.length > 0) decision.observed = observed; + // Message quiet period: a message's wake is held, by channel, until the + // channel goes quiet. The heartbeat is no message (its wakes stay + // immediate, as under sleep). A message the policies didn't trigger + // (a rate-limited part of the same burst, say) extends a hold already + // open for its channel, so the wake waits for the whole burst, but opens + // none. if ( - decision.trigger && this.quietMs > 0 && + info.serverId !== 'heartbeat' && (info.eventType === 'mcpl:channel-incoming' || info.eventType === 'mcpl:push-event') ) { - this.holdForQuietPeriod(info); - decision = { ...decision, trigger: false, quietHeld: true }; + const key = this.quietKeyFor(info); + if (decision.trigger) { + this.holdForQuietPeriod(key, info); + decision = { ...decision, trigger: false, quietHeld: true }; + } else if (this.quietHolds.has(key)) { + this.armQuietHold(key); + } } this.emitTrace({ @@ -3528,42 +3543,85 @@ export class EventGate { } } - /** Hold a message-driven wake until the senders go quiet. The message + /** The hold a message belongs to: its registered channel, else its own + * channel id, else one shared hold, so chatter in one channel doesn't hold + * a DM in another. */ + private quietKeyFor(info: GateEventInfo): string { + return this.routeChannelFor(info) ?? (info.channelId || SHARED_BUCKET_KEY); + } + + /** Hold a message-driven wake until its channel goes quiet. The message * itself is already in context (stored at arrival); only the WAKE waits, * so a caption and its images, or a reply split into several Discord - * messages, reach the agent as one turn instead of several. */ - private holdForQuietPeriod(info: GateEventInfo): void { + * messages, reach the agent as one turn instead of several. The held event + * is shaped as a debounce policy's, and fires through the same path. */ + private holdForQuietPeriod(key: string, info: GateEventInfo): void { const now = this.now(); - if (this.quietEvents.length === 0) this.quietFirstAt = now; - this.quietEvents.push({ - policyName: 'quiet-period', - content: '', + const event: PendingEvent = { + policyName: QUIET_PERIOD_POLICY, + content: info.content.length > MAX_CONTENT_SNIPPET + ? safeSlice(info.content, 0, MAX_CONTENT_SNIPPET) + '...' + : info.content, eventType: info.eventType, timestamp: now, inContext: true, channelId: info.channelId || undefined, + channelLabel: + typeof info.metadata?.channelName === 'string' && info.metadata.channelName + ? (info.metadata.channelName as string) + : undefined, authorId: this.extractAuthorId(info.metadata) ?? undefined, serverId: info.serverId || undefined, addressed: Array.isArray(info.tags) && info.tags.includes('chat:addressed'), - }); - if (this.quietTimer) clearTimeout(this.quietTimer); - const wait = Math.max(0, Math.min(this.quietMs, this.quietFirstAt + this.quietMaxWaitMs - now)); - this.quietTimer = setTimeout(() => this.fireQuietPeriod(), wait); - (this.quietTimer as { unref?: () => void }).unref?.(); - } - - private fireQuietPeriod(): void { - this.quietTimer = null; - const events = this.quietEvents; - this.quietEvents = []; - if (events.length === 0) return; - const provenance = wakeProvenance(events); - for (const agentName of this.getAgentNamesFn()) { - // An agent already mid-turn received these messages as mid-turn input; - // waking it again afterwards would spend a redundant turn. - if (this.inferring.has(agentName)) continue; - this.requestInferenceFn(agentName, 'gate:quiet-period', 'gate', provenance); + // The held wake's route: a held DM's reply goes back to the DM. + routeChannelId: this.routeChannelFor(info), + ...(typeof info.metadata?.messageId === 'string' ? { messageId: info.metadata.messageId } : {}), + ...(typeof info.metadata?.eventId === 'string' ? { eventId: info.metadata.eventId } : {}), + }; + const hold = this.quietHolds.get(key); + if (hold) { + hold.events.push(event); + } else { + this.quietHolds.set(key, { timer: setTimeout(() => {}, 0), events: [event], firstAt: now }); } + this.armQuietHold(key); + } + + /** (Re)arm a hold's timer: quietMs from now, but no later than maxWaitMs + * from the hold's first message. */ + private armQuietHold(key: string): void { + const hold = this.quietHolds.get(key); + if (!hold) return; + clearTimeout(hold.timer); + const wait = Math.max(0, Math.min(this.quietMs, hold.firstAt + this.quietMaxWaitMs - this.now())); + hold.timer = setTimeout(() => this.fireQuietHold(key), wait); + } + + /** A hold's end. Asleep, the wake is dropped as the messages' own would + * have been (sleep suppresses external wakes; the messages are in context + * for the agent's wake). Otherwise the batch goes as a debounce batch + * does: buffered for an agent mid-turn or a quiesced host, delivered + * now otherwise. */ + private fireQuietHold(key: string): void { + const hold = this.quietHolds.get(key); + this.quietHolds.delete(key); + if (!hold || hold.events.length === 0) return; + const events = hold.events; + if (this.sleepUntil > this.now()) { + this.sleepSuppressed += events.length; + this.emitTrace({ + type: 'gate:decision', + eventType: events[events.length - 1].eventType, + serverId: events[events.length - 1].serverId, + channelId: events[events.length - 1].channelId, + matchedPolicy: 'asleep', + trigger: false, + behavior: 'skip', + timestamp: this.now(), + }); + return; + } + this.fireBatch(QUIET_PERIOD_POLICY, events); } private handleDebounce(policy: GatePolicy, info: GateEventInfo): void { @@ -3624,12 +3682,16 @@ export class EventGate { // A judge rule's batch goes to the reply judge, which delivers it, or // doesn't, once it has decided (judgeBatch). Without a judge wired it's - // delivered below, as a debounce's batch is. + // delivered as a debounce's batch is (fireBatch). if (state.judge && this.judgeFn) { this.judgeBatch(policyName, events, state.judge); return; } + this.fireBatch(policyName, events); + } + /** A batch's end, for a debounce policy's and a quiet-period hold's alike. */ + private fireBatch(policyName: string, events: PendingEvent[]): void { // If any agent is currently inferring — or the host is quiesced — buffer // for later (with cap). While quiesced, delivering would write a // "[Gate: N events]" message into the very context being operated on @@ -3669,15 +3731,20 @@ export class EventGate { private deliverEvents(events: PendingEvent[]): void { if (events.length === 0) return; - const policyNames = [...new Set(events.map(e => e.policyName))]; + // A quiet-period hold's messages get no line: they are the newest + // entries of their channel's window, right above the wake, and the hold + // exists to spend one turn on them, not to add to it. A debounce + // policy's events keep their line as before. + const lined = events.filter(e => e.policyName !== QUIET_PERIOD_POLICY); + const policyNames = [...new Set(lined.map(e => e.policyName))]; // Events whose content the framework already stored at arrival (MCPL // channel-incoming / push-event) get a compact REFERENCE — re-quoting // them here would inject a duplicate copy of the message minutes after // the fact, reading as fresh input the agent may have already handled. // Only events with no context entry of their own carry their content. - const quoted = events.filter(e => !e.inContext); - const referenced = events.filter(e => e.inContext); + const quoted = lined.filter(e => !e.inContext); + const referenced = lined.filter(e => e.inContext); const lines: string[] = quoted.map(e => `- [${e.policyName}] (${e.eventType}): ${e.content}`); // How each referenced channel is named in the line, in order. @@ -3724,10 +3791,12 @@ export class EventGate { // gate hands the line to the wake (WakeProvenance.line) and a fresh start // of the wake's turn decides it (WakeLine). Several messages, one with // anything after it, or content the line quotes, keep the line now. - const text = `[Gate: ${events.length} event${events.length > 1 ? 's' : ''} matched]\n\n${lines.join('\n')}`; - const only = events.length === 1 && events[0].inContext ? events[0] : undefined; + const text = `[Gate: ${lined.length} event${lined.length > 1 ? 's' : ''} matched]\n\n${lines.join('\n')}`; + const only = lined.length === 1 && lined[0].inContext ? lined[0] : undefined; let line: WakeLine | undefined; - if (only && this.newestEntry(only)) { + if (lined.length === 0) { + // Nothing to write: a quiet-period batch alone. + } else if (only && this.newestEntry(only)) { const shownChannel = only.routeChannelId ?? only.channelId; line = { entry: entryRef(only), @@ -3754,8 +3823,9 @@ export class EventGate { // shared by every agent the batch wakes, so it is written once. const chosen = wakeProvenance(events); const provenance = line ? { ...chosen, line } : chosen; + const reason = lined.length === 0 ? 'gate:quiet-period' : 'gate:debounce'; for (const agentName of this.getAgentNamesFn()) { - this.requestInferenceFn(agentName, 'gate:debounce', 'gate', provenance); + this.requestInferenceFn(agentName, reason, 'gate', provenance); } } @@ -4604,6 +4674,13 @@ export class EventGate { } this.judging.clear(); this.judgeHeld.clear(); + // A quiet-period hold goes with the gate: fired after a stop(), its wake + // would reach a stopped framework. Its messages are in context; the next + // lifecycle's first wake reads them. + for (const hold of this.quietHolds.values()) { + clearTimeout(hold.timer); + } + this.quietHolds.clear(); for (const timer of this.selfWakeTimers.values()) { clearTimeout(timer); } diff --git a/src/gate/types.ts b/src/gate/types.ts index 5ec4f3b32..1fc768808 100644 --- a/src/gate/types.ts +++ b/src/gate/types.ts @@ -225,10 +225,17 @@ export interface GateOptions { judge?: ReplyJudgeOptions | null; /** * Hold message-driven wakes (MCPL channel-incoming / push-event that the - * policies would trigger) until no further message has arrived for - * `quietMs`, capped at `maxWaitMs` (default 30000) from the first held - * message; then wake once. Messages still enter context at arrival. Other - * wake sources (scheduled, tool results, heartbeat) are unaffected. + * policies would trigger) until no further message has arrived in the same + * channel for `quietMs`, capped at `maxWaitMs` (default 30000) from the + * channel's first held message; then wake once for the burst. Messages + * still enter context at arrival; only the wake waits. A message the + * policies didn't trigger extends a hold already open for its channel (a + * rate-limited part of the same burst) but opens none. The held wake goes + * the way a `debounce` policy's batch does: it carries the burst's route + * (`routeChannelId`, so a held DM's reply goes back to the DM), an agent + * mid-turn gets it after the turn, a quiesced host at resume, and an agent + * asleep when the hold ends gets none (the messages are in context for its + * wake). The heartbeat, scheduled wakes and tool results are unaffected. * Default off. */ messageQuietPeriod?: { quietMs: number; maxWaitMs?: number }; diff --git a/test/event-gate-quiet-period.test.ts b/test/event-gate-quiet-period.test.ts index 104ca4906..1bcdf1f2b 100644 --- a/test/event-gate-quiet-period.test.ts +++ b/test/event-gate-quiet-period.test.ts @@ -1,29 +1,47 @@ /** * GateOptions.messageQuietPeriod: a burst of messages (a caption and its * images, a reply split across several Discord messages) wakes the agent once, - * after the senders go quiet — not once per message. + * after its channel goes quiet — not once per message. + * + * The held wake goes the way a `debounce` policy's batch does (#179's review): + * - an agent mid-turn when the hold ends gets the wake after the turn, not + * never (bufferForInference, flushed at onInferenceEnded); + * - the wake carries the burst's route (routeChannelId), so a held DM's + * reply goes back to the DM; + * - holds are per channel; dispose() ends them; sleep drops them; the + * heartbeat is never held; maxWaitMs is the operator's cap even below + * quietMs; a part the policies didn't trigger extends an open hold. */ 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 { EventGate } from '../src/gate/event-gate.js'; -import type { GateEventInfo } from '../src/gate/types.js'; +import type { GateConfig, GateEventInfo } from '../src/gate/types.js'; +import type { WakeProvenance } from '../src/gate/event-gate.js'; +import { AgentFramework } from '../src/index.js'; +import { MockMembrane } from './helpers/mock-membrane.js'; const TMP = join(import.meta.dirname, '../.test-tmp-gate-quiet'); -function makeGate(quiet?: { quietMs: number; maxWaitMs?: number }) { +type Wake = { agent: string; reason: string; provenance?: WakeProvenance }; + +function makeGate( + quiet?: { quietMs: number; maxWaitMs?: number }, + extra: { config?: GateConfig; resolveRouteChannel?: (info: GateEventInfo) => string | undefined } = {}, +) { mkdirSync(TMP, { recursive: true }); - const wakes: Array<{ reason: string; channelId?: string }> = []; + const wakes: Wake[] = []; const messages: unknown[] = []; const gate = new EventGate({ configPath: join(TMP, `gate-${Math.random().toString(36).slice(2)}.json`), - initialConfig: { default: 'always', policies: [] }, + initialConfig: extra.config ?? { default: 'always', policies: [] }, emitTrace: () => {}, addMessage: (...args) => { messages.push(args); return ''; }, - requestInference: (_agent, reason, _source, provenance) => wakes.push({ reason, channelId: provenance?.channelId }), + requestInference: (agent, reason, _source, provenance) => wakes.push({ agent, reason, provenance }), getAgentNames: () => ['agent'], ...(quiet ? { messageQuietPeriod: quiet } : {}), + ...(extra.resolveRouteChannel ? { resolveRouteChannel: extra.resolveRouteChannel } : {}), }); return { gate, wakes, messages }; } @@ -41,9 +59,11 @@ const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); describe('message quiet period', () => { afterEach(() => rmSync(TMP, { recursive: true, force: true })); - it('holds a burst and wakes once after the senders go quiet', async () => { + it('holds a burst and wakes once after the channel goes quiet, with no line in context', async () => { const { gate, wakes, messages } = makeGate({ quietMs: 60 }); - assert.equal(gate.evaluate(msg('caption')).trigger, false); + const first = gate.evaluate(msg('caption')); + assert.equal(first.trigger, false); + assert.equal(first.quietHeld, true); await sleep(20); assert.equal(gate.evaluate(msg('image 1')).trigger, false); await sleep(20); @@ -52,8 +72,9 @@ describe('message quiet period', () => { await sleep(90); assert.equal(wakes.length, 1, 'one wake for the whole burst'); assert.equal(wakes[0].reason, 'gate:quiet-period'); - assert.equal(wakes[0].channelId, 'discord:g:salon'); + assert.equal(wakes[0].provenance?.channelId, 'discord:g:salon'); assert.equal(messages.length, 0, 'no batch summary is added to context'); + gate.dispose(); }); it('caps the hold at maxWaitMs under a continuous stream', async () => { @@ -66,21 +87,159 @@ describe('message quiet period', () => { } assert.equal(wakes.length, 1, 'woke despite continuous messages'); assert.ok(Date.now() - start < 190, 'within roughly the cap'); + gate.dispose(); + }); + + it('a maxWaitMs below quietMs is the cap, not raised to quietMs', async () => { + const { gate, wakes } = makeGate({ quietMs: 300, maxWaitMs: 40 }); + gate.evaluate(msg('one')); + await sleep(100); + assert.equal(wakes.length, 1, 'the hold ended at maxWaitMs'); + gate.dispose(); }); - it('does not re-wake an agent that is already mid-turn when the hold ends', async () => { + it('an agent mid-turn when the hold ends gets the wake after its turn', async () => { const { gate, wakes } = makeGate({ quietMs: 40 }); gate.evaluate(msg('arrives during a turn')); gate.onInferenceStarted('agent'); await sleep(70); - assert.equal(wakes.length, 0); + assert.equal(wakes.length, 0, 'nothing wakes a turn in flight'); gate.onInferenceEnded('agent'); + await sleep(10); // the flush runs in a microtask + assert.equal(wakes.length, 1, 'the held wake follows the turn'); + assert.equal(wakes[0].reason, 'gate:quiet-period'); + assert.equal(wakes[0].provenance?.channelId, 'discord:g:salon'); + gate.dispose(); + }); + + it('a quiesced host gets the held wake at resume', async () => { + const { gate, wakes } = makeGate({ quietMs: 30 }); + gate.setQuiesced(true); + gate.evaluate(msg('while quiesced')); + await sleep(60); + assert.equal(wakes.length, 0, 'held through the quiesce'); + gate.setQuiesced(false); + await sleep(10); + assert.equal(wakes.length, 1, 'delivered at resume'); + gate.dispose(); + }); + + it("a held addressed message's wake carries its route, so a DM's reply goes back to the DM", async () => { + const { gate, wakes } = makeGate({ quietMs: 30 }, { + resolveRouteChannel: (info) => (info.channelId === 'raw-dm' ? 'discord:dm:u1' : undefined), + }); + gate.evaluate(msg('hello?', { channelId: 'raw-dm', tags: ['chat:addressed'] })); + await sleep(60); + assert.equal(wakes.length, 1); + assert.equal(wakes[0].provenance?.routeChannelId, 'discord:dm:u1'); + assert.equal(wakes[0].provenance?.routeServerId, 'discord'); + assert.equal(wakes[0].provenance?.addressed, true); + gate.dispose(); + }); + + it("holds are per channel: chatter in one channel doesn't delay another's wake", async () => { + const { gate, wakes } = makeGate({ quietMs: 60, maxWaitMs: 1000 }); + gate.evaluate(msg('dm', { channelId: 'discord:dm:u2', tags: ['chat:addressed'] })); + const start = Date.now(); + // The salon keeps talking past the DM's quiet period. + while (Date.now() - start < 150) { + gate.evaluate(msg('salon chatter')); + await sleep(20); + } + assert.ok(wakes.length >= 1, "the DM's wake came while the salon was still talking"); + assert.equal(wakes[0].provenance?.channelId, 'discord:dm:u2'); + await sleep(120); + assert.equal(wakes.length, 2, "then the salon's, once it went quiet"); + assert.equal(wakes[1].provenance?.channelId, 'discord:g:salon'); + gate.dispose(); + }); + + it('a part the policies did not trigger extends an open hold but opens none', async () => { + // One token, slow refill: the caption triggers, the image is denied. + const config: GateConfig = { + default: 'always', + policies: [{ name: 'one-per-window', match: {}, behavior: { rate_limit: { tokens: 1, refillIntervalMs: 60_000 } } }], + }; + const { gate, wakes } = makeGate({ quietMs: 60 }, { config }); + const caption = gate.evaluate(msg('caption')); + assert.equal(caption.quietHeld, true); + await sleep(40); + const image = gate.evaluate(msg('image')); + assert.equal(image.trigger, false); + assert.notEqual(image.quietHeld, true, 'a denied part is not a held wake of its own'); + await sleep(35); // 75 ms after the caption: past its own quietMs, within the image's + assert.equal(wakes.length, 0, 'the image extended the hold'); + await sleep(40); + assert.equal(wakes.length, 1, 'one wake once the burst is quiet'); + gate.dispose(); + + const alone = makeGate({ quietMs: 30 }, { config }); + alone.gate.evaluate(msg('a')); + await sleep(60); + assert.equal(alone.wakes.length, 1); + alone.gate.evaluate(msg('denied, no hold open')); + await sleep(60); + assert.equal(alone.wakes.length, 1, 'a denied message opens no hold'); + alone.gate.dispose(); + }); + + it('the heartbeat is not held', () => { + const { gate } = makeGate({ quietMs: 60 }); + const tick = gate.evaluate(msg('tick', { eventType: 'mcpl:push-event', serverId: 'heartbeat', channelId: '' })); + assert.equal(tick.trigger, true); + assert.notEqual(tick.quietHeld, true); + gate.dispose(); + }); + + it('an agent asleep when the hold ends gets no wake', async () => { + const { gate, wakes } = makeGate({ quietMs: 30 }); + gate.evaluate(msg('before sleep')); + gate.setSleep(10, 'resting', 'agent'); + await sleep(60); + assert.equal(wakes.length, 0, 'sleep suppresses the held wake as it would the message\'s own'); + assert.equal(gate.getSleepState()?.suppressed, 1); + gate.clearSleep(); + gate.dispose(); + }); + + it('dispose() ends a pending hold', async () => { + const { gate, wakes } = makeGate({ quietMs: 30 }); + gate.evaluate(msg('held')); + gate.dispose(); + await sleep(60); + assert.equal(wakes.length, 0, 'no wake reaches a disposed gate'); }); it('does not delay non-message wakes, and is off by default', () => { const { gate } = makeGate({ quietMs: 60 }); assert.equal(gate.evaluate(msg('tick', { eventType: 'external-message' })).trigger, true); + gate.dispose(); const plain = makeGate(); assert.equal(plain.gate.evaluate(msg('hello')).trigger, true); + plain.gate.dispose(); + }); + + it('the framework passes gate.messageQuietPeriod through to its gate', async () => { + mkdirSync(TMP, { recursive: true }); + const membrane = new MockMembrane(); + const framework = await AgentFramework.create({ + storePath: join(TMP, 'store-passthrough'), + membrane: membrane.asMembrane(), + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test' }], + gate: { + config: { policies: [], default: 'always' }, + configPath: join(TMP, 'gate-passthrough.json'), + messageQuietPeriod: { quietMs: 40 }, + }, + modules: [], + }); + const gate = (framework as unknown as { eventGate: EventGate }).eventGate; + try { + const held = gate.evaluate(msg('through the framework')); + assert.equal(held.quietHeld, true, 'the option reached the gate'); + } finally { + gate.dispose(); + await framework.stop(); + } }); }); From 2195f772d50e8c4e73654b309381398f86ab8558 Mon Sep 17 00:00:00 2001 From: Daryna Date: Wed, 28 Oct 2026 15:33:20 +0000 Subject: [PATCH 3/4] fix(scheduler): hold a message burst where the turn starts, not in the gate (#856 R2) From #856's review (nissa-seru's four runs, slimepriestess's re-read): a hold kept in the gate turned a message's own decision into "no" and woke later through the batch path, so every part of the framework that reads that decision saw a "no": a privileged author's ping no longer woke a sleeping agent (a), an ambient burst's reply could go to another channel (b), any skipped message kept a hold going (c), and tune-out and conversation routing saw the held message as skipped (d). The hold now lives where the scheduler starts an idle agent's turn (runInferencePass), as both reviewers suggested: - `FrameworkConfig.messageQuietPeriod: { quietMs, maxWaitMs? }`, validated once (finite, 0 or more; maxWaitMs defaults to 30 s). - `quietHoldRemaining`: an idle agent's batch waits while every wake in it is a message wake (mcpl:channel-incoming or mcpl:push-event, other than the heartbeat's, and neither a continuation nor a retry of the turn), until quietMs after the newest or maxWaitMs after the oldest. The requests go back to the queue as a busy agent's do; the loop's next pass reads them again. A batch holding any other wake starts now. - Each message keeps what the gate and the ingress gave it, so (a) to (d) are as on main: the gate is main's again, `quietHeld` and the gate's option are gone, and the scheduler's merge of queued wakes makes the one turn. - One log line per burst ([quiet-period]); no new trace type. Tests: quietHoldRemaining on fixed clocks (burst, quiet, the cap, a cap below quietMs, server events, the heartbeat, other wakes, continuations and retries), and the scheduler against a real framework with wide margins (one turn per burst, the option off, the cap under a stream, an operator's wake ending a hold, refused configs). The gate-level tests go with the gate's hold. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01SkepYxcPem6pmgRxKd2RBY --- .../feat-gate-message-quiet-period.added.md | 23 +- src/framework.ts | 71 ++++- src/gate/event-gate.ts | 151 +---------- src/gate/types.ts | 18 -- src/index.ts | 2 +- src/types/framework.ts | 16 ++ test/event-gate-quiet-period.test.ts | 245 ------------------ test/message-quiet-period.test.ts | 189 ++++++++++++++ 8 files changed, 296 insertions(+), 419 deletions(-) delete mode 100644 test/event-gate-quiet-period.test.ts create mode 100644 test/message-quiet-period.test.ts diff --git a/changelog.d/feat-gate-message-quiet-period.added.md b/changelog.d/feat-gate-message-quiet-period.added.md index f96a0c38a..27f0d1000 100644 --- a/changelog.d/feat-gate-message-quiet-period.added.md +++ b/changelog.d/feat-gate-message-quiet-period.added.md @@ -1,11 +1,12 @@ -- Opt-in gate `messageQuietPeriod: { quietMs, maxWaitMs? }`: a message's - wake is held until its channel has been quiet for `quietMs` (at most - `maxWaitMs`, default 30 s, from the channel's first held message), so a - caption and its attachments, or a reply split into several messages, wake - the agent once instead of once per part. Messages still enter context at - arrival; only the wake waits, and no line is written for it. The held wake - goes as a `debounce` policy's batch does: it carries the burst's route, so - a held DM's reply goes back to the DM; an agent mid-turn gets it after the - turn; a quiesced host at resume; an agent asleep when the hold ends gets - none. Holds are per channel, so chatter in one channel delays no other's. - The heartbeat, scheduled wakes and tool results are unaffected. +- 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 565e7df7d..e53b8008d 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -806,6 +806,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'; @@ -3508,6 +3555,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 statedWaitRuns: Map = new Map(); @@ -4251,6 +4302,7 @@ export class AgentFramework { ); framework.providerHoldHook = config.providerHold; framework.turnRearmMs = turnRearmDelays(config.exhaustedTurnRearmMs); + framework.quietPeriod = quietPeriodOf(config.messageQuietPeriod); // Before any delivery. await framework.reconcileDiscordAwarenessAtStartup(); @@ -4369,7 +4421,6 @@ export class AgentFramework { } : {}), ...(judge && judgeOptions?.timeoutMs !== undefined ? { judgeTimeoutMs: judgeOptions.timeoutMs } : {}), ...(judge && judgeOptions?.onError !== undefined ? { judgeOnError: judgeOptions.onError } : {}), - messageQuietPeriod: config.gate.messageQuietPeriod, emitTrace: (e) => framework.emitTrace(e as { type: TraceEvent['type']; [key: string]: unknown }), addMessage: (p, c, m, forAgent, placement) => framework.addMessage(p, c, m as MessageMetadata, forAgent || placement ? { ...(forAgent ? { forAgent } : {}), ...(placement ? { placement } : {}) } : undefined), @@ -17031,6 +17082,24 @@ 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). + const quietLeft = quietHoldRemaining(requests, now, this.quietPeriod); + if (quietLeft > 0) { + if (!this.quietHeldAgents.has(agentName)) { + this.quietHeldAgents.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/gate/event-gate.ts b/src/gate/event-gate.ts index 23e17a3ba..6bc724f51 100644 --- a/src/gate/event-gate.ts +++ b/src/gate/event-gate.ts @@ -241,8 +241,6 @@ interface RateBucket { /** Synthetic key used when a policy has no `keyBy` or the field is missing. */ const SHARED_BUCKET_KEY = '__shared__'; -/** The policy name a quiet-period hold's events carry (GateOptions.messageQuietPeriod). */ -const QUIET_PERIOD_POLICY = 'quiet-period'; interface CompiledPolicy { policy: GatePolicy; @@ -2056,14 +2054,6 @@ export class EventGate { // Debounce state private debounceTimers = new Map(); - // Message quiet period (GateOptions.messageQuietPeriod): a message-driven - // wake is held until no further message has arrived in its channel for - // quietMs (capped at maxWaitMs from the channel's first held one), then - // fires once, through the debounce path (fireDebounce): an agent mid-turn - // gets it after the turn, a quiesced host at resume. - private readonly quietMs: number; - private readonly quietMaxWaitMs: number; - private readonly quietHolds = new Map; events: PendingEvent[]; firstAt: number }>(); // Reply judge state (`judge` rules). The judge itself, when the host wires // one; without it a judge rule's batch is delivered as a debounce's. @@ -2226,18 +2216,12 @@ export class EventGate { * (WakeProvenance.line), for a fresh start of the wake's turn to decide * (WakeLine). Absent: every batched wake writes its line, as before. */ isNewestEntry?: (entry: GateEntryRef) => boolean; - /** See GateOptions.messageQuietPeriod. */ - messageQuietPeriod?: { quietMs: number; maxWaitMs?: number }; /** The event types the gate can see from a source, when the host knows * them (an MCPL server's id: MCPL_GATE_EVENT_TYPES), else undefined. * The rule-order notes (findShadowedPolicies) leave out types a rule's * source never sends. Absent: any source can send any type. */ eventTypesFrom?: EventTypesFrom; }) { - this.quietMs = Math.max(0, opts.messageQuietPeriod?.quietMs ?? 0); - // The cap is the operator's: a maxWaitMs below quietMs ends the hold at - // maxWaitMs, not at quietMs. - this.quietMaxWaitMs = Math.max(0, opts.messageQuietPeriod?.maxWaitMs ?? 30_000); this.configPath = opts.configPath; this.privilegedUsersPath = opts.privilegedUsersPath; this.emitTrace = opts.emitTrace; @@ -3192,25 +3176,6 @@ export class EventGate { this.defaultByEventType.set(info.eventType, bucket); } if (observed.length > 0) decision.observed = observed; - // Message quiet period: a message's wake is held, by channel, until the - // channel goes quiet. The heartbeat is no message (its wakes stay - // immediate, as under sleep). A message the policies didn't trigger - // (a rate-limited part of the same burst, say) extends a hold already - // open for its channel, so the wake waits for the whole burst, but opens - // none. - if ( - this.quietMs > 0 && - info.serverId !== 'heartbeat' && - (info.eventType === 'mcpl:channel-incoming' || info.eventType === 'mcpl:push-event') - ) { - const key = this.quietKeyFor(info); - if (decision.trigger) { - this.holdForQuietPeriod(key, info); - decision = { ...decision, trigger: false, quietHeld: true }; - } else if (this.quietHolds.has(key)) { - this.armQuietHold(key); - } - } this.emitTrace({ type: 'gate:decision', @@ -3622,87 +3587,6 @@ export class EventGate { } } - /** The hold a message belongs to: its registered channel, else its own - * channel id, else one shared hold, so chatter in one channel doesn't hold - * a DM in another. */ - private quietKeyFor(info: GateEventInfo): string { - return this.routeChannelFor(info) ?? (info.channelId || SHARED_BUCKET_KEY); - } - - /** Hold a message-driven wake until its channel goes quiet. The message - * itself is already in context (stored at arrival); only the WAKE waits, - * so a caption and its images, or a reply split into several Discord - * messages, reach the agent as one turn instead of several. The held event - * is shaped as a debounce policy's, and fires through the same path. */ - private holdForQuietPeriod(key: string, info: GateEventInfo): void { - const now = this.now(); - const event: PendingEvent = { - policyName: QUIET_PERIOD_POLICY, - content: info.content.length > MAX_CONTENT_SNIPPET - ? safeSlice(info.content, 0, MAX_CONTENT_SNIPPET) + '...' - : info.content, - eventType: info.eventType, - timestamp: now, - inContext: true, - channelId: info.channelId || undefined, - channelLabel: - typeof info.metadata?.channelName === 'string' && info.metadata.channelName - ? (info.metadata.channelName as string) - : undefined, - authorId: this.extractAuthorId(info.metadata) ?? undefined, - serverId: info.serverId || undefined, - addressed: Array.isArray(info.tags) && info.tags.includes('chat:addressed'), - // The held wake's route: a held DM's reply goes back to the DM. - routeChannelId: this.routeChannelFor(info), - ...(typeof info.metadata?.messageId === 'string' ? { messageId: info.metadata.messageId } : {}), - ...(typeof info.metadata?.eventId === 'string' ? { eventId: info.metadata.eventId } : {}), - }; - const hold = this.quietHolds.get(key); - if (hold) { - hold.events.push(event); - } else { - this.quietHolds.set(key, { timer: setTimeout(() => {}, 0), events: [event], firstAt: now }); - } - this.armQuietHold(key); - } - - /** (Re)arm a hold's timer: quietMs from now, but no later than maxWaitMs - * from the hold's first message. */ - private armQuietHold(key: string): void { - const hold = this.quietHolds.get(key); - if (!hold) return; - clearTimeout(hold.timer); - const wait = Math.max(0, Math.min(this.quietMs, hold.firstAt + this.quietMaxWaitMs - this.now())); - hold.timer = setTimeout(() => this.fireQuietHold(key), wait); - } - - /** A hold's end. Asleep, the wake is dropped as the messages' own would - * have been (sleep suppresses external wakes; the messages are in context - * for the agent's wake). Otherwise the batch goes as a debounce batch - * does: buffered for an agent mid-turn or a quiesced host, delivered - * now otherwise. */ - private fireQuietHold(key: string): void { - const hold = this.quietHolds.get(key); - this.quietHolds.delete(key); - if (!hold || hold.events.length === 0) return; - const events = hold.events; - if (this.sleepUntil > this.now()) { - this.sleepSuppressed += events.length; - this.emitTrace({ - type: 'gate:decision', - eventType: events[events.length - 1].eventType, - serverId: events[events.length - 1].serverId, - channelId: events[events.length - 1].channelId, - matchedPolicy: 'asleep', - trigger: false, - behavior: 'skip', - timestamp: this.now(), - }); - return; - } - this.fireBatch(QUIET_PERIOD_POLICY, events); - } - private handleDebounce(policy: GatePolicy, info: GateEventInfo): void { const debounceMs = batchQuietMs(policy.behavior) ?? MIN_DEBOUNCE_MS; const judge = typeof policy.behavior === 'object' && 'judge' in policy.behavior ? policy.behavior.judge : undefined; @@ -3761,16 +3645,12 @@ export class EventGate { // A judge rule's batch goes to the reply judge, which delivers it, or // doesn't, once it has decided (judgeBatch). Without a judge wired it's - // delivered as a debounce's batch is (fireBatch). + // delivered below, as a debounce's batch is. if (state.judge && this.judgeFn) { this.judgeBatch(policyName, events, state.judge); return; } - this.fireBatch(policyName, events); - } - /** A batch's end, for a debounce policy's and a quiet-period hold's alike. */ - private fireBatch(policyName: string, events: PendingEvent[]): void { // If any agent is currently inferring — or the host is quiesced — buffer // for later (with cap). While quiesced, delivering would write a // "[Gate: N events]" message into the very context being operated on @@ -3810,20 +3690,15 @@ export class EventGate { private deliverEvents(events: PendingEvent[]): void { if (events.length === 0) return; - // A quiet-period hold's messages get no line: they are the newest - // entries of their channel's window, right above the wake, and the hold - // exists to spend one turn on them, not to add to it. A debounce - // policy's events keep their line as before. - const lined = events.filter(e => e.policyName !== QUIET_PERIOD_POLICY); - const policyNames = [...new Set(lined.map(e => e.policyName))]; + const policyNames = [...new Set(events.map(e => e.policyName))]; // Events whose content the framework already stored at arrival (MCPL // channel-incoming / push-event) get a compact REFERENCE — re-quoting // them here would inject a duplicate copy of the message minutes after // the fact, reading as fresh input the agent may have already handled. // Only events with no context entry of their own carry their content. - const quoted = lined.filter(e => !e.inContext); - const referenced = lined.filter(e => e.inContext); + const quoted = events.filter(e => !e.inContext); + const referenced = events.filter(e => e.inContext); const lines: string[] = quoted.map(e => `- [${e.policyName}] (${e.eventType}): ${suppliedValue(e.content)}`); // How each referenced channel is named in the line, in order. @@ -3870,12 +3745,10 @@ export class EventGate { // gate hands the line to the wake (WakeProvenance.line) and a fresh start // of the wake's turn decides it (WakeLine). Several messages, one with // anything after it, or content the line quotes, keep the line now. - const text = `[${NOTICE_TAG.gate}: ${lined.length} event${lined.length > 1 ? 's' : ''} matched]\n\n${lines.join('\n')}`; - const only = lined.length === 1 && lined[0].inContext ? lined[0] : undefined; + const text = `[${NOTICE_TAG.gate}: ${events.length} event${events.length > 1 ? 's' : ''} matched]\n\n${lines.join('\n')}`; + const only = events.length === 1 && events[0].inContext ? events[0] : undefined; let line: WakeLine | undefined; - if (lined.length === 0) { - // Nothing to write: a quiet-period batch alone. - } else if (only && this.newestEntry(only)) { + if (only && this.newestEntry(only)) { const shownChannel = only.routeChannelId ?? only.channelId; line = { entry: entryRef(only), @@ -3902,9 +3775,8 @@ export class EventGate { // shared by every agent the batch wakes, so it is written once. const chosen = wakeProvenance(events); const provenance = line ? { ...chosen, line } : chosen; - const reason = lined.length === 0 ? 'gate:quiet-period' : 'gate:debounce'; for (const agentName of this.getAgentNamesFn()) { - this.requestInferenceFn(agentName, reason, 'gate', provenance); + this.requestInferenceFn(agentName, 'gate:debounce', 'gate', provenance); } } @@ -4755,13 +4627,6 @@ export class EventGate { } this.judging.clear(); this.judgeHeld.clear(); - // A quiet-period hold goes with the gate: fired after a stop(), its wake - // would reach a stopped framework. Its messages are in context; the next - // lifecycle's first wake reads them. - for (const hold of this.quietHolds.values()) { - clearTimeout(hold.timer); - } - this.quietHolds.clear(); for (const timer of this.selfWakeTimers.values()) { clearTimeout(timer); } diff --git a/src/gate/types.ts b/src/gate/types.ts index c8238563b..50398b77c 100644 --- a/src/gate/types.ts +++ b/src/gate/types.ts @@ -223,22 +223,6 @@ export interface GateOptions { * null), the judge is wired with its defaults; it runs only for `judge` * rules, so a gate with none makes no judge calls. */ judge?: ReplyJudgeOptions | null; - /** - * Hold message-driven wakes (MCPL channel-incoming / push-event that the - * policies would trigger) until no further message has arrived in the same - * channel for `quietMs`, capped at `maxWaitMs` (default 30000) from the - * channel's first held message; then wake once for the burst. Messages - * still enter context at arrival; only the wake waits. A message the - * policies didn't trigger extends a hold already open for its channel (a - * rate-limited part of the same burst) but opens none. The held wake goes - * the way a `debounce` policy's batch does: it carries the burst's route - * (`routeChannelId`, so a held DM's reply goes back to the DM), an agent - * mid-turn gets it after the turn, a quiesced host at resume, and an agent - * asleep when the hold ends gets none (the messages are in context for its - * wake). The heartbeat, scheduled wakes and tool results are unaffected. - * Default off. - */ - messageQuietPeriod?: { quietMs: number; maxWaitMs?: number }; } /** @@ -358,8 +342,6 @@ export interface GateDecision { policyName: string | null; /** The behavior that was applied. */ behavior: GateBehavior; - /** The trigger was converted into a held quiet-period wake. */ - quietHeld?: boolean; /** * Names of `passthrough` policies that matched and counted this event but * did not fire, so evaluation continued past them. Absent when none did. diff --git a/src/index.ts b/src/index.ts index 8bad14e76..781b7843b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,7 +2,7 @@ export * from './types/index.js'; // Core classes -export { AgentFramework, BudgetPreflightError, DefaultErrorPolicy, ProviderWaitNotSleptError, QuiesceBlockedError, ResumeBlockedError } from './framework.js'; +export { AgentFramework, BudgetPreflightError, DefaultErrorPolicy, ProviderWaitNotSleptError, QuiesceBlockedError, ResumeBlockedError, quietHoldRemaining } from './framework.js'; export type { RuntimeSettingsPreview, HostModeStatus } from './framework.js'; export { Agent } from './agent.js'; export { storedInjectionMarkOf, type StoredInjectionMark } from './stored-injections.js'; diff --git a/src/types/framework.ts b/src/types/framework.ts index be5ed5455..d55812177 100644 --- a/src/types/framework.ts +++ b/src/types/framework.ts @@ -261,6 +261,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/event-gate-quiet-period.test.ts b/test/event-gate-quiet-period.test.ts deleted file mode 100644 index 1bcdf1f2b..000000000 --- a/test/event-gate-quiet-period.test.ts +++ /dev/null @@ -1,245 +0,0 @@ -/** - * GateOptions.messageQuietPeriod: a burst of messages (a caption and its - * images, a reply split across several Discord messages) wakes the agent once, - * after its channel goes quiet — not once per message. - * - * The held wake goes the way a `debounce` policy's batch does (#179's review): - * - an agent mid-turn when the hold ends gets the wake after the turn, not - * never (bufferForInference, flushed at onInferenceEnded); - * - the wake carries the burst's route (routeChannelId), so a held DM's - * reply goes back to the DM; - * - holds are per channel; dispose() ends them; sleep drops them; the - * heartbeat is never held; maxWaitMs is the operator's cap even below - * quietMs; a part the policies didn't trigger extends an open hold. - */ -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 { EventGate } from '../src/gate/event-gate.js'; -import type { GateConfig, GateEventInfo } from '../src/gate/types.js'; -import type { WakeProvenance } from '../src/gate/event-gate.js'; -import { AgentFramework } from '../src/index.js'; -import { MockMembrane } from './helpers/mock-membrane.js'; - -const TMP = join(import.meta.dirname, '../.test-tmp-gate-quiet'); - -type Wake = { agent: string; reason: string; provenance?: WakeProvenance }; - -function makeGate( - quiet?: { quietMs: number; maxWaitMs?: number }, - extra: { config?: GateConfig; resolveRouteChannel?: (info: GateEventInfo) => string | undefined } = {}, -) { - mkdirSync(TMP, { recursive: true }); - const wakes: Wake[] = []; - const messages: unknown[] = []; - const gate = new EventGate({ - configPath: join(TMP, `gate-${Math.random().toString(36).slice(2)}.json`), - initialConfig: extra.config ?? { default: 'always', policies: [] }, - emitTrace: () => {}, - addMessage: (...args) => { messages.push(args); return ''; }, - requestInference: (agent, reason, _source, provenance) => wakes.push({ agent, reason, provenance }), - getAgentNames: () => ['agent'], - ...(quiet ? { messageQuietPeriod: quiet } : {}), - ...(extra.resolveRouteChannel ? { resolveRouteChannel: extra.resolveRouteChannel } : {}), - }); - return { gate, wakes, messages }; -} - -const msg = (text: string, extra: Partial = {}): GateEventInfo => ({ - content: text, - eventType: 'mcpl:channel-incoming', - serverId: 'discord', - channelId: 'discord:g:salon', - metadata: { authorId: 'u1' }, - ...extra, -}); -const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); - -describe('message quiet period', () => { - afterEach(() => rmSync(TMP, { recursive: true, force: true })); - - it('holds a burst and wakes once after the channel goes quiet, with no line in context', async () => { - const { gate, wakes, messages } = makeGate({ quietMs: 60 }); - const first = gate.evaluate(msg('caption')); - assert.equal(first.trigger, false); - assert.equal(first.quietHeld, true); - await sleep(20); - assert.equal(gate.evaluate(msg('image 1')).trigger, false); - await sleep(20); - assert.equal(gate.evaluate(msg('image 2')).trigger, false); - assert.equal(wakes.length, 0, 'no wake while messages keep arriving'); - await sleep(90); - assert.equal(wakes.length, 1, 'one wake for the whole burst'); - assert.equal(wakes[0].reason, 'gate:quiet-period'); - assert.equal(wakes[0].provenance?.channelId, 'discord:g:salon'); - assert.equal(messages.length, 0, 'no batch summary is added to context'); - gate.dispose(); - }); - - it('caps the hold at maxWaitMs under a continuous stream', async () => { - const { gate, wakes } = makeGate({ quietMs: 50, maxWaitMs: 120 }); - const start = Date.now(); - while (Date.now() - start < 200) { - gate.evaluate(msg('chatter')); - await sleep(25); - if (wakes.length > 0) break; - } - assert.equal(wakes.length, 1, 'woke despite continuous messages'); - assert.ok(Date.now() - start < 190, 'within roughly the cap'); - gate.dispose(); - }); - - it('a maxWaitMs below quietMs is the cap, not raised to quietMs', async () => { - const { gate, wakes } = makeGate({ quietMs: 300, maxWaitMs: 40 }); - gate.evaluate(msg('one')); - await sleep(100); - assert.equal(wakes.length, 1, 'the hold ended at maxWaitMs'); - gate.dispose(); - }); - - it('an agent mid-turn when the hold ends gets the wake after its turn', async () => { - const { gate, wakes } = makeGate({ quietMs: 40 }); - gate.evaluate(msg('arrives during a turn')); - gate.onInferenceStarted('agent'); - await sleep(70); - assert.equal(wakes.length, 0, 'nothing wakes a turn in flight'); - gate.onInferenceEnded('agent'); - await sleep(10); // the flush runs in a microtask - assert.equal(wakes.length, 1, 'the held wake follows the turn'); - assert.equal(wakes[0].reason, 'gate:quiet-period'); - assert.equal(wakes[0].provenance?.channelId, 'discord:g:salon'); - gate.dispose(); - }); - - it('a quiesced host gets the held wake at resume', async () => { - const { gate, wakes } = makeGate({ quietMs: 30 }); - gate.setQuiesced(true); - gate.evaluate(msg('while quiesced')); - await sleep(60); - assert.equal(wakes.length, 0, 'held through the quiesce'); - gate.setQuiesced(false); - await sleep(10); - assert.equal(wakes.length, 1, 'delivered at resume'); - gate.dispose(); - }); - - it("a held addressed message's wake carries its route, so a DM's reply goes back to the DM", async () => { - const { gate, wakes } = makeGate({ quietMs: 30 }, { - resolveRouteChannel: (info) => (info.channelId === 'raw-dm' ? 'discord:dm:u1' : undefined), - }); - gate.evaluate(msg('hello?', { channelId: 'raw-dm', tags: ['chat:addressed'] })); - await sleep(60); - assert.equal(wakes.length, 1); - assert.equal(wakes[0].provenance?.routeChannelId, 'discord:dm:u1'); - assert.equal(wakes[0].provenance?.routeServerId, 'discord'); - assert.equal(wakes[0].provenance?.addressed, true); - gate.dispose(); - }); - - it("holds are per channel: chatter in one channel doesn't delay another's wake", async () => { - const { gate, wakes } = makeGate({ quietMs: 60, maxWaitMs: 1000 }); - gate.evaluate(msg('dm', { channelId: 'discord:dm:u2', tags: ['chat:addressed'] })); - const start = Date.now(); - // The salon keeps talking past the DM's quiet period. - while (Date.now() - start < 150) { - gate.evaluate(msg('salon chatter')); - await sleep(20); - } - assert.ok(wakes.length >= 1, "the DM's wake came while the salon was still talking"); - assert.equal(wakes[0].provenance?.channelId, 'discord:dm:u2'); - await sleep(120); - assert.equal(wakes.length, 2, "then the salon's, once it went quiet"); - assert.equal(wakes[1].provenance?.channelId, 'discord:g:salon'); - gate.dispose(); - }); - - it('a part the policies did not trigger extends an open hold but opens none', async () => { - // One token, slow refill: the caption triggers, the image is denied. - const config: GateConfig = { - default: 'always', - policies: [{ name: 'one-per-window', match: {}, behavior: { rate_limit: { tokens: 1, refillIntervalMs: 60_000 } } }], - }; - const { gate, wakes } = makeGate({ quietMs: 60 }, { config }); - const caption = gate.evaluate(msg('caption')); - assert.equal(caption.quietHeld, true); - await sleep(40); - const image = gate.evaluate(msg('image')); - assert.equal(image.trigger, false); - assert.notEqual(image.quietHeld, true, 'a denied part is not a held wake of its own'); - await sleep(35); // 75 ms after the caption: past its own quietMs, within the image's - assert.equal(wakes.length, 0, 'the image extended the hold'); - await sleep(40); - assert.equal(wakes.length, 1, 'one wake once the burst is quiet'); - gate.dispose(); - - const alone = makeGate({ quietMs: 30 }, { config }); - alone.gate.evaluate(msg('a')); - await sleep(60); - assert.equal(alone.wakes.length, 1); - alone.gate.evaluate(msg('denied, no hold open')); - await sleep(60); - assert.equal(alone.wakes.length, 1, 'a denied message opens no hold'); - alone.gate.dispose(); - }); - - it('the heartbeat is not held', () => { - const { gate } = makeGate({ quietMs: 60 }); - const tick = gate.evaluate(msg('tick', { eventType: 'mcpl:push-event', serverId: 'heartbeat', channelId: '' })); - assert.equal(tick.trigger, true); - assert.notEqual(tick.quietHeld, true); - gate.dispose(); - }); - - it('an agent asleep when the hold ends gets no wake', async () => { - const { gate, wakes } = makeGate({ quietMs: 30 }); - gate.evaluate(msg('before sleep')); - gate.setSleep(10, 'resting', 'agent'); - await sleep(60); - assert.equal(wakes.length, 0, 'sleep suppresses the held wake as it would the message\'s own'); - assert.equal(gate.getSleepState()?.suppressed, 1); - gate.clearSleep(); - gate.dispose(); - }); - - it('dispose() ends a pending hold', async () => { - const { gate, wakes } = makeGate({ quietMs: 30 }); - gate.evaluate(msg('held')); - gate.dispose(); - await sleep(60); - assert.equal(wakes.length, 0, 'no wake reaches a disposed gate'); - }); - - it('does not delay non-message wakes, and is off by default', () => { - const { gate } = makeGate({ quietMs: 60 }); - assert.equal(gate.evaluate(msg('tick', { eventType: 'external-message' })).trigger, true); - gate.dispose(); - const plain = makeGate(); - assert.equal(plain.gate.evaluate(msg('hello')).trigger, true); - plain.gate.dispose(); - }); - - it('the framework passes gate.messageQuietPeriod through to its gate', async () => { - mkdirSync(TMP, { recursive: true }); - const membrane = new MockMembrane(); - const framework = await AgentFramework.create({ - storePath: join(TMP, 'store-passthrough'), - membrane: membrane.asMembrane(), - agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test' }], - gate: { - config: { policies: [], default: 'always' }, - configPath: join(TMP, 'gate-passthrough.json'), - messageQuietPeriod: { quietMs: 40 }, - }, - modules: [], - }); - const gate = (framework as unknown as { eventGate: EventGate }).eventGate; - try { - const held = gate.evaluate(msg('through the framework')); - assert.equal(held.quietHeld, true, 'the option reached the gate'); - } finally { - gate.dispose(); - await framework.stop(); - } - }); -}); 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), + ); + } + }); +}); From 72ee758c5c1b41eaaff9a94f7c4cb5ed1a6eca03 Mon Sep 17 00:00:00 2001 From: Daryna Date: Wed, 28 Oct 2026 16:35:17 +0000 Subject: [PATCH 4/4] fix(scheduler): read the quiet-period set as surgeryHold is read, for a prototype-built harness busy-agent-scheduler.test.ts builds the framework from the prototype, so the class's field initializers don't run and quietHeldAgents is undefined: deleting from it threw, with the option off. The set is now made on first use and read with ?., as the surgery hold's slot is. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01SkepYxcPem6pmgRxKd2RBY --- src/framework.ts | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index e53b8008d..556fbd5b7 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -17089,16 +17089,19 @@ export class AgentFramework { // 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) { - if (!this.quietHeldAgents.has(agentName)) { - this.quietHeldAgents.add(agentName); + 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); + 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