From 6c05acea48ad79f78f4edea910ddd356dfe20898 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] =?UTF-8?q?feat(gate):=20messageQuietPeriod=20=E2=80=94=20?= =?UTF-8?q?one=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 87f9c2a41..4c6e65bb2 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -1548,6 +1548,7 @@ export class AgentFramework { configPath, initialConfig: config.gate.config, privilegedUsersPath: config.gate.privilegedUsersPath, + messageQuietPeriod: config.gate.messageQuietPeriod, emitTrace: (e) => framework.emitTrace(e as { type: TraceEvent['type']; [key: string]: unknown }), addMessage: (p, c, m, forAgent) => framework.addMessage(p, c, m as MessageMetadata, forAgent ? { forAgent } : undefined), requestInference: (agentName, reason, source, provenance) => { diff --git a/src/gate/event-gate.ts b/src/gate/event-gate.ts index 6db0a737c..75f9d3e9e 100644 --- a/src/gate/event-gate.ts +++ b/src/gate/event-gate.ts @@ -520,6 +520,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; // Rate-limit state — policy name → key → bucket. private rateLimitBuckets = new Map>(); @@ -608,7 +616,11 @@ export class EventGate { now?: () => number; /** Per-event timeout (ms) for the optional gate.js script. Default 50. */ scriptTimeoutMs?: number; + /** 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; @@ -1067,6 +1079,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', @@ -1432,6 +1452,44 @@ export class EventGate { // Debounce // ========================================================================= + /** 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 = (policy.behavior as { debounce: number }).debounce; diff --git a/src/gate/types.ts b/src/gate/types.ts index 29c67244a..5b3f38491 100644 --- a/src/gate/types.ts +++ b/src/gate/types.ts @@ -164,6 +164,15 @@ export interface GateOptions { * missing, no one bypasses sleep. */ privilegedUsersPath?: string; + /** + * 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 }; } // --------------------------------------------------------------------------- @@ -178,6 +187,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); + }); +});