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); + }); +});