feat(scheduler): messageQuietPeriod — one turn per message burst (carries #179)

changelog.d/feat-gate-message-quiet-period.added.md

@@ -0,0 +1,12 @@
+- Opt-in `FrameworkConfig.messageQuietPeriod: { quietMs, maxWaitMs? }`: an
+  idle agent's turn waits while message wakes for it keep arriving, so a
+  burst (a caption and its attachments, a reply split into several messages)
+  starts one turn instead of one per part. The scheduler holds the start
+  until `quietMs` have passed since the agent's newest message wake, or
+  `maxWaitMs` (default 30 s) since its oldest held one, then starts one turn
+  with everything queued, as it merges queued wakes today. Each message's
+  own decision stays what the gate gives it: who wakes a sleeping agent,
+  where a reply goes, what tune-out and conversation routing see. A batch
+  holding any other wake (a timer, an operator, a tool result, a
+  continuation or a retry of the agent's turn) starts now; the heartbeat is
+  no message wake. (#179, carried by #856.)

src/framework.ts

@@ -881,6 +881,53 @@ const turnRearmDelays = (value: unknown): readonly number[] => {
   return [...value];
 };
 
+/** FrameworkConfig.messageQuietPeriod, read once: both numbers finite and
+ *  0 or more, maxWaitMs defaulting to 30 s; anything else refuses the config. */
+const quietPeriodOf = (value: unknown): { quietMs: number; maxWaitMs: number } | undefined => {
+  if (value === undefined) return undefined;
+  const bad = (): never => {
+    throw new Error('FrameworkConfig.messageQuietPeriod must be { quietMs, maxWaitMs? }, each a finite number ' +
+      `of ms, 0 or more, not ${shownValue(value)}`);
+  };
+  if (!value || typeof value !== 'object' || Array.isArray(value)) return bad();
+  const { quietMs, maxWaitMs } = value as { quietMs?: unknown; maxWaitMs?: unknown };
+  if (typeof quietMs !== 'number' || !Number.isFinite(quietMs) || quietMs < 0) return bad();
+  if (maxWaitMs !== undefined && (typeof maxWaitMs !== 'number' || !Number.isFinite(maxWaitMs) || maxWaitMs < 0)) return bad();
+  return { quietMs, maxWaitMs: maxWaitMs ?? 30_000 };
+};
+
+/**
+ * How much longer an idle agent's turn waits for its message burst to end
+ * (FrameworkConfig.messageQuietPeriod), in ms; 0 starts it now. The batch
+ * waits only while every wake in it is a message wake: a channel message's
+ * or a server event's (other than the heartbeat's), and none a continuation
+ * or a retry of the agent's turn. It ends `quietMs` after the newest message
+ * wake, or `maxWaitMs` after the oldest, whichever is first.
+ */
+export function quietHoldRemaining(
+  requests: readonly InferenceRequest[],
+  now: number,
+  quiet: { quietMs: number; maxWaitMs: number } | undefined,
+): number {
+  if (!quiet || quiet.quietMs <= 0 || requests.length === 0) return 0;
+  let newest = -Infinity;
+  let oldest = Infinity;
+  for (const r of requests) {
+    if (!isMessageWake(r)) return 0;
+    if (r.timestamp > newest) newest = r.timestamp;
+    if (r.timestamp < oldest) oldest = r.timestamp;
+  }
+  return Math.max(0, Math.min(quiet.quietMs - (now - newest), quiet.maxWaitMs - (now - oldest)));
+}
+
+/** A wake a message made: a channel message's or a server event's, other
+ *  than the heartbeat's, that neither continues nor retries a turn. */
+function isMessageWake(r: InferenceRequest): boolean {
+  return (r.reason === 'mcpl:channel-incoming' || r.reason === 'mcpl:push-event')
+    && r.source !== 'heartbeat'
+    && !continuesTurn(r) && !carriesContinuation(r) && mergedBatch(r) === undefined;
+}
+
 const settingBoolean = (name: string, value: unknown): boolean => {
   if (typeof value === 'boolean') return value;
   if (value === 'true' || value === 'false') return value === 'true';
@@ -3882,6 +3929,10 @@ export class AgentFramework {
    *  Drives hard-down escalation — see noteInferenceExhausted. */
   private consecutiveInferenceFailures: Map<string, number> = 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<string>();
   private turnRearms: Map<string, number> = new Map();
   private exhaustingRearms: Set<string> = new Set();
   private rearmDeadlineCuts: Set<string> = new Set();
@@ -4689,6 +4740,7 @@ export class AgentFramework {
     );
     framework.providerHoldHook = config.providerHold;
     framework.turnRearmMs = turnRearmDelays(config.exhaustedTurnRearmMs);
+    framework.quietPeriod = quietPeriodOf(config.messageQuietPeriod);
     framework.inferenceBounds = resolveInferenceBounds(config.inferenceBounds);
 
     // Before any delivery.
@@ -19725,6 +19777,27 @@ export class AgentFramework {
         continue;
       }
 
+      // A message quiet period (FrameworkConfig.messageQuietPeriod): an idle
+      // agent's turn waits while message wakes for it keep arriving, and
+      // starts once with everything queued (the merge below). Each wake
+      // keeps what the gate and the ingress gave it; only the start waits.
+      // The loop's next pass reads the queue again (runLoop's backoff), so
+      // the requests go back as a busy agent's do, with no timer of their
+      // own. A batch holding any other wake starts now (quietHoldRemaining).
+      // The set is read as surgeryHold is: a prototype-built harness leaves
+      // both slots undefined, which reads as "not held".
+      const quietLeft = quietHoldRemaining(requests, now, this.quietPeriod);
+      if (quietLeft > 0) {
+        const held = (this.quietHeldAgents ??= new Set());
+        if (!held.has(agentName)) {
+          held.add(agentName);
+          console.error(`[quiet-period] agent=${agentName}: holding the turn for a message burst (${requests.length} wake(s) so far)`);
+        }
+        this.pendingRequests.push(...requests);
+        continue;
+      }
+      this.quietHeldAgents?.delete(agentName);
+
       // A wake whose sole cause was a write deferred for this agent
       // (readsDeferredWrite), or a message stored for it as it arrived
       // (readsStoredMessage), has nothing left to do once a request that

src/index.ts

@@ -2,7 +2,7 @@
 export * from './types/index.js';
 
 // Core classes
-export { AgentFramework, AgentStoppedError, BudgetPreflightError, DefaultErrorPolicy, ProviderWaitNotSleptError, QuiesceBlockedError, ResumeBlockedError } from './framework.js';
+export { AgentFramework, AgentStoppedError, BudgetPreflightError, DefaultErrorPolicy, ProviderWaitNotSleptError, QuiesceBlockedError, ResumeBlockedError, quietHoldRemaining } from './framework.js';
 export type { RuntimeSettingsPreview, HostModeStatus, DeriveAgentOptions } from './framework.js';
 export { Agent } from './agent.js';
 export { FLAT_MINIMUM_MS, type InferenceBoundsOptions } from './inference-bounds.js';

src/types/framework.ts

@@ -400,6 +400,22 @@ export interface FrameworkConfig {
    * `abortInference()` doesn't end it.
    */
   exhaustedTurnRearmMs?: number[] | false;
+  /**
+   * A message quiet period: an idle agent's turn waits while message wakes
+   * for it keep arriving, so a burst (a caption and its images, a reply split
+   * into several messages) starts one turn, not one per part. The scheduler
+   * holds the start until `quietMs` have passed since the agent's newest
+   * message wake, or `maxWaitMs` (default 30000) since its oldest held one,
+   * then starts one turn with everything queued, as it merges queued wakes
+   * today. A message wake is a channel message's or a server event's
+   * (`mcpl:channel-incoming`, `mcpl:push-event`) other than the heartbeat's.
+   * Each message's own decision stays what the gate gives it: who wakes a
+   * sleeping agent, where a reply goes, what tune-out and conversation
+   * routing see. A batch holding any other wake (a timer, an operator, a
+   * tool result, a continuation or a retry of the agent's turn) starts now.
+   * Off when omitted.
+   */
+  messageQuietPeriod?: { quietMs: number; maxWaitMs?: number };
 
   /** Interval for periodic store sync in milliseconds (default: 1000ms, 0 to disable) */
   syncIntervalMs?: number;

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