feat(gate): messageQuietPeriod — one wake per message burst

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.

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) => {

src/gate/event-gate.ts

@@ -520,6 +520,14 @@ export class EventGate {
 
   // Debounce state
   private debounceTimers = new Map<string, DebounceState>();
+  // 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<typeof setTimeout> | null = null;
+  private quietEvents: PendingEvent[] = [];
+  private quietFirstAt = 0;
 
   // Rate-limit state — policy name → key → bucket.
   private rateLimitBuckets = new Map<string, Map<string, RateBucket>>();
@@ -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;
 

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.

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