From 2de01db95fcd92b090c1adf0799daaead29cd43a Mon Sep 17 00:00:00 2001 From: antra-tess Date: Thu, 1 Oct 2026 00:09:43 -0700 Subject: [PATCH 1/3] fix(gate): batched wakes route to their addressed message's channel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A debounced gate wake requested inference with telemetry provenance only — no channelId — so the turn froze on the process-global most-recent- inbound channel. channels/incoming traffic anywhere retargets that fallback; a DM push event never does. Both orders misrouted a DM reply: ambient chatter elsewhere one second AFTER the DM pulled the reply into that channel, and chatter BEFORE the DM kept it there. - EventGate resolves each debounced event's registered channel at enqueue (PendingEvent.routeChannelId) through a new optional resolveRouteChannel(info): default = a channel-incoming event's own id, nothing for push events. The framework supplies its push-origin derivation (derivePushEventChannel — the one ingestion uses to register the DM channel). A throwing resolver never breaks gating. - wakeProvenance reports routeChannelId only when the chosen event — the newest addressed one — is addressed and resolved; it never falls back to an older addressed event. - The framework's gate requestInference sets channelId + addressed from routeChannelId; the existing turn-start selection (addressed wins) does the rest. Ambient-only batches are unchanged. - Batched-wake lines name the registered id instead of the raw snowflake. Tests: an end-to-end suite over a new MCPL fixture that sends a Discord-shaped DM push event and channel traffic in one debounce window (DM→ambient, ambient→DM, mention→ambient route to the addressed channel — all three fail on main with the reply in the ambient channel; ambient-only keeps the fallback), plus gate unit tests. Co-Authored-By: Claude Opus 5.5 --- .../fix-batched-wake-addressed-locus.fixed.md | 10 ++ src/framework.ts | 25 ++- src/gate/event-gate.ts | 57 ++++++- test/batched-wake-locus.test.ts | 131 +++++++++++++++ test/event-gate.test.ts | 96 ++++++++++- test/fixtures/wake-locus-mcpl-server.mjs | 153 ++++++++++++++++++ 6 files changed, 461 insertions(+), 11 deletions(-) create mode 100644 changelog.d/fix-batched-wake-addressed-locus.fixed.md create mode 100644 test/batched-wake-locus.test.ts create mode 100644 test/fixtures/wake-locus-mcpl-server.mjs diff --git a/changelog.d/fix-batched-wake-addressed-locus.fixed.md b/changelog.d/fix-batched-wake-addressed-locus.fixed.md new file mode 100644 index 000000000..27153f4ac --- /dev/null +++ b/changelog.d/fix-batched-wake-addressed-locus.fixed.md @@ -0,0 +1,10 @@ +- A debounced (batched) gate wake that contains an addressed message — + mention, reply, or DM — now routes the turn's plain speech to that + message's channel, the same rule the direct wake path already applies. + Previously a batched wake carried no channel and fell back to the + process-global most-recent-inbound channel, so ambient traffic elsewhere + in the debounce window (or before a DM, which never updates that + fallback) received the reply. Push events are mapped to their registered + composite channel id via a new `EventGate` option `resolveRouteChannel` + (the framework supplies it); batched-wake lines now show that id instead + of the adapter's raw one. Ambient-only batches keep the legacy fallback. diff --git a/src/framework.ts b/src/framework.ts index 6724e0219..878a7b075 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -1569,16 +1569,35 @@ export class AgentFramework { framework.pendingRequests.push({ agentName, reason, source, timestamp: Date.now(), // gate-requested wakes carry where/who (EventGate wakeProvenance) - // as TELEMETRY fields — the host's stamp reads them; the turn's - // speech locus (channelId / addressed) is deliberately NOT set - // here, so a batched wake routes exactly as it did before. + // as TELEMETRY fields — the host's stamp reads them. ...(provenance?.channelId ? { wakeChannelId: provenance.channelId } : {}), ...(provenance?.counterparty ? { counterparty: provenance.counterparty } : {}), ...(provenance?.at ? { wakeAt: provenance.at } : {}), + // A batch that contains an ADDRESSED message (mention / reply / + // DM) routes like the direct path does: the reply goes to that + // message's channel. Without it the turn froze on the + // process-global most-recent-inbound channel, which an ambient + // message elsewhere could retarget between the DM's arrival and + // the debounce firing (2026-09-30). Ambient-only batches set no + // locus and keep the legacy fallback. + ...(provenance?.routeChannelId + ? { channelId: provenance.routeChannelId, addressed: true } + : {}), }); }, getAgentNames: () => [...framework.agents.keys()].filter( (n) => n !== framework.subconsciousAgentName), + // Push events reach the gate with the adapter's raw channel id; map + // them to the registered composite id the same way ingestion does + // (handleMcplPushEvent registers that channel on arrival, before any + // debounce can fire). + resolveRouteChannel: (info) => { + if (info.eventType === 'mcpl:channel-incoming') return info.channelId || undefined; + if (info.eventType === 'mcpl:push-event') { + return framework.derivePushEventChannel(info.metadata)?.channelId; + } + return undefined; + }, }); } diff --git a/src/gate/event-gate.ts b/src/gate/event-gate.ts index 6db0a737c..01e6cf505 100644 --- a/src/gate/event-gate.ts +++ b/src/gate/event-gate.ts @@ -88,6 +88,11 @@ interface PendingEvent { * newer ambient chatter when a batched wake picks its provenance — the same * rule the framework applies to the turn's speech locus. */ addressed: boolean; + /** The REGISTERED (composite) channel id the event arrived on, resolved by + * the host at enqueue time. Differs from `channelId` for push events, + * whose `channelId` is the adapter's raw id (a bare Discord snowflake for a + * DM) — unroutable. Absent when the host could not resolve one. */ + routeChannelId?: string; } /** @@ -106,6 +111,15 @@ export interface WakeProvenance { counterparty?: string; /** True when the chosen event was `chat:addressed`. */ addressed?: boolean; + /** Speech locus for the wake: the routable channel of the chosen event, + * set ONLY when that event addressed the agent (mention / reply / DM) and + * the host resolved its registered channel. A batched wake used to carry + * no locus at all, so its reply fell back to the process-global + * most-recent-inbound channel — which any ambient message elsewhere, + * arriving a second after the DM that caused the wake, could retarget + * (2026-09-30: a DM reply published into a guild channel). Ambient-only + * batches still set nothing and keep the legacy fallback. */ + routeChannelId?: string; /** Timestamp (ms) of the chosen event, so a consumer coalescing several * requests can order by event recency, not by flush order. */ at?: number; @@ -204,10 +218,20 @@ export function wakeProvenance(events: PendingEvent[]): WakeProvenance | undefin ...(channelId ? { channelId } : {}), ...(pick.authorId ? { counterparty: `${ns}:user:${pick.authorId}` } : {}), ...(pick.addressed ? { addressed: true } : {}), + ...(pick.addressed && pick.routeChannelId ? { routeChannelId: pick.routeChannelId } : {}), at: pick.timestamp, }; } +/** + * Default route-channel resolution when the host supplies none: a + * channel-incoming event's channel id is already the registered composite id; + * a push event's is the adapter's raw id, which only the host can map. + */ +function defaultRouteChannel(info: GateEventInfo): string | undefined { + return info.eventType === 'mcpl:channel-incoming' && info.channelId ? info.channelId : undefined; +} + /** Compact, log-friendly serialization of a GateBehavior. */ function formatBehavior(b: import('./types.js').GateBehavior): string { if (typeof b === 'string') return b; @@ -587,6 +611,7 @@ export class EventGate { ) => unknown; private requestInferenceFn: (agentName: string, reason: string, source: string, provenance?: WakeProvenance) => void; private getAgentNamesFn: () => string[]; + private resolveRouteChannelFn: (info: GateEventInfo) => string | undefined; /** Clock injection — keeps the new rate_limit / passive_sample paths * testable without monkey-patching Date.now globally. */ private now: () => number; @@ -608,6 +633,10 @@ export class EventGate { now?: () => number; /** Per-event timeout (ms) for the optional gate.js script. Default 50. */ scriptTimeoutMs?: number; + /** Map an event to the registered channel it arrived on (the host knows + * how push-event origins compose into channel ids). Default: a + * channel-incoming event's own channel id; nothing for push events. */ + resolveRouteChannel?: (info: GateEventInfo) => string | undefined; }) { this.configPath = opts.configPath; this.privilegedUsersPath = opts.privilegedUsersPath; @@ -615,6 +644,7 @@ export class EventGate { this.addMessageFn = opts.addMessage; this.requestInferenceFn = opts.requestInference; this.getAgentNamesFn = opts.getAgentNames; + this.resolveRouteChannelFn = opts.resolveRouteChannel ?? defaultRouteChannel; this.now = opts.now ?? (() => Date.now()); this.loadPrivileged(); @@ -1432,6 +1462,15 @@ export class EventGate { // Debounce // ========================================================================= + /** Registered channel for an event; a resolver fault never breaks gating. */ + private routeChannelFor(info: GateEventInfo): string | undefined { + try { + return this.resolveRouteChannelFn(info) || undefined; + } catch { + return undefined; + } + } + private handleDebounce(policy: GatePolicy, info: GateEventInfo): void { const debounceMs = (policy.behavior as { debounce: number }).debounce; @@ -1451,6 +1490,7 @@ export class EventGate { authorId: this.extractAuthorId(info.metadata) ?? undefined, serverId: info.serverId || undefined, addressed: Array.isArray(info.tags) && info.tags.includes('chat:addressed'), + routeChannelId: this.routeChannelFor(info), }; const existing = this.debounceTimers.get(policy.name); @@ -1514,7 +1554,10 @@ export class EventGate { if (referenced.length > 0) { const byChannel = new Map(); for (const e of referenced) { - const key = e.channelId ?? '(unknown channel)'; + // The registered id when known, so this line names the channel the + // same way routing notices and send tools do (a push event's own + // channelId is the adapter's raw id). + const key = e.routeChannelId ?? e.channelId ?? '(unknown channel)'; const entry = byChannel.get(key) ?? { count: 0, oldest: e.timestamp, label: e.channelLabel }; entry.count += 1; entry.oldest = Math.min(entry.oldest, e.timestamp); @@ -1541,11 +1584,13 @@ export class EventGate { policies: policyNames, }); - // Provenance of the batched wake (telemetry: who/where woke the agent): - // the newest ADDRESSED event first, else the newest event naming a - // channel or an author; channel + author + addressed from that ONE - // event. It does not set the turn's speech locus — the framework keeps - // its own routing rule for that. + // Provenance of the batched wake (who/where woke the agent): the newest + // ADDRESSED event first, else the newest event naming a channel or an + // author; channel + author + addressed from that ONE event. When that + // event addressed the agent and its registered channel is known, + // `routeChannelId` names the turn's speech locus too — the reply goes + // back to whoever addressed the agent, not to whichever channel last saw + // any traffic. Ambient-only batches carry telemetry only. const provenance = wakeProvenance(events); for (const agentName of this.getAgentNamesFn()) { this.requestInferenceFn(agentName, 'gate:debounce', 'gate', provenance); diff --git a/test/batched-wake-locus.test.ts b/test/batched-wake-locus.test.ts new file mode 100644 index 000000000..9ef4c4037 --- /dev/null +++ b/test/batched-wake-locus.test.ts @@ -0,0 +1,131 @@ +/** + * Batched (debounced) wakes and outbound routing, end to end over a real MCPL + * child: a DM push event (raw snowflake channel id, as discord-mcpl sends it) + * and ambient channel traffic land in the same debounce window. + * + * Before: a batched wake carried no channel, so the turn froze on the + * process-global most-recent-inbound channel. channels/incoming retargets + * that; a DM push event never does. Either order misrouted the DM reply: + * - ambient AFTER the DM: the ambient channel became the fallback; + * - ambient BEFORE the DM: the DM never displaced it. + * Now a batch containing an addressed message routes to that message's + * registered channel; an ambient-only batch keeps the legacy fallback. + */ +import { describe, it, beforeEach, afterEach } from 'node:test'; +import assert from 'node:assert/strict'; +import { tmpdir } from 'node:os'; +import { mkdtempSync, rmSync, readFileSync, existsSync, writeFileSync, appendFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { AgentFramework } from '../src/index.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; + +const FIXTURE = join(import.meta.dirname, 'fixtures/wake-locus-mcpl-server.mjs'); +const ROOM = 'discord:g1:room'; +const GENERAL = 'discord:g1:general'; +const RAW_DM = '1548000000000000001'; +const DM = `discord:dm:${RAW_DM}`; + +async function waitFor(cond: () => boolean, what: string, timeoutMs = 15_000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (cond()) return; + await new Promise((r) => setTimeout(r, 25)); + } + throw new Error(`timed out waiting for: ${what}`); +} + +describe('batched wake routing', () => { + let tempDir: string; + let statusPath: string; + let commandPath: string; + let membrane: MockMembrane; + let framework: AgentFramework; + let n = 0; + + beforeEach(async () => { + tempDir = mkdtempSync(join(tmpdir(), 'wake-locus-')); + statusPath = join(tempDir, 'status.jsonl'); + commandPath = join(tempDir, 'commands.txt'); + writeFileSync(commandPath, ''); + membrane = new MockMembrane(); + framework = await AgentFramework.create({ + storePath: join(tempDir, 'test.chronicle'), + membrane: membrane.asMembrane(), + agents: [{ name: 'scout', model: 'test-model', systemPrompt: 'You are scout.' }], + gate: { + config: { + policies: [{ + name: 'batch', + match: { scope: ['mcpl:channel-incoming', 'mcpl:push-event'] }, + behavior: { debounce: 400 }, + }], + default: 'skip', + }, + }, + mcplServers: [{ + id: 'discord', + command: process.execPath, + args: [FIXTURE], + env: { STATUS_PATH: statusPath, COMMAND_PATH: commandPath }, + }], + modules: [], + }); + await framework.start(); + const registry = (framework as unknown as { channelRegistry: { listChannelsRaw(): unknown[] } }).channelRegistry; + await waitFor(() => registry.listChannelsRaw().length >= 2, 'channel registration'); + }); + + afterEach(async () => { + await framework.stop(); + rmSync(tempDir, { recursive: true, force: true }); + }); + + const command = (line: string): void => appendFileSync(commandPath, line + '\n'); + const dm = (text: string): void => command(`dm ev-${++n} 134 ${RAW_DM} ${text}`); + const incoming = (channelId: string, mode: 'ambient' | 'addressed', text: string): void => + command(`incoming ${channelId} m-${++n} ${mode} ${text}`); + const sent = (event: string): number => !existsSync(statusPath) ? 0 + : readFileSync(statusPath, 'utf8').split('\n').filter(Boolean) + .filter((l) => (JSON.parse(l) as { event: string }).event === event).length; + const publishes = (): string[] => !existsSync(statusPath) ? [] + : readFileSync(statusPath, 'utf8').split('\n').filter(Boolean) + .map((l) => JSON.parse(l) as { event: string; channelId?: string }) + .filter((e) => e.event === 'publish') + .map((e) => e.channelId!); + + it('a DM followed one beat later by ambient chatter elsewhere: the reply goes to the DM', async () => { + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'answering the DM' }])); + dm('are you there?'); + await waitFor(() => sent('dm-sent') === 1, 'dm sent'); + incoming(GENERAL, 'ambient', 'unrelated chatter'); + await waitFor(() => publishes().length >= 1, 'reply published'); + assert.deepEqual(publishes(), [DM]); + }); + + it('ambient chatter followed by a DM in the same window: the reply goes to the DM', async () => { + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'answering the DM' }])); + incoming(GENERAL, 'ambient', 'unrelated chatter'); + await waitFor(() => sent('incoming-sent') === 1, 'ambient sent'); + dm('quick question'); + await waitFor(() => publishes().length >= 1, 'reply published'); + assert.deepEqual(publishes(), [DM]); + }); + + it('a mention in one channel then ambient chatter in another: the reply goes to the mention', async () => { + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'answering the mention' }])); + incoming(ROOM, 'addressed', '@scout hello'); + await waitFor(() => sent('incoming-sent') === 1, 'mention sent'); + incoming(GENERAL, 'ambient', 'unrelated chatter'); + await waitFor(() => publishes().length >= 1, 'reply published'); + assert.deepEqual(publishes(), [ROOM]); + }); + + it('an ambient-only batch keeps the legacy fallback (most recent inbound)', async () => { + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'joining in' }])); + incoming(ROOM, 'ambient', 'first'); + await waitFor(() => sent('incoming-sent') === 1, 'first sent'); + incoming(GENERAL, 'ambient', 'second'); + await waitFor(() => publishes().length >= 1, 'reply published'); + assert.deepEqual(publishes(), [GENERAL]); + }); +}); diff --git a/test/event-gate.test.ts b/test/event-gate.test.ts index 49c245fac..e469e084e 100644 --- a/test/event-gate.test.ts +++ b/test/event-gate.test.ts @@ -27,14 +27,18 @@ interface TraceEntry { [key: string]: unknown; } -function makeGate(configPath: string, opts?: { initialConfig?: GateConfig }) { +function makeGate(configPath: string, opts?: { + initialConfig?: GateConfig; + resolveRouteChannel?: (info: GateEventInfo) => string | undefined; +}) { const traces: TraceEntry[] = []; const messages: Array<{ participant: string; content: unknown; metadata?: unknown }> = []; - const inferenceRequests: Array<{ agentName: string; reason: string; source: string; channelId?: string; counterparty?: string; addressed?: boolean; at?: number }> = []; + const inferenceRequests: Array<{ agentName: string; reason: string; source: string; channelId?: string; counterparty?: string; addressed?: boolean; at?: number; routeChannelId?: string }> = []; const gate = new EventGate({ configPath, initialConfig: opts?.initialConfig, + ...(opts?.resolveRouteChannel ? { resolveRouteChannel: opts.resolveRouteChannel } : {}), emitTrace: (e) => traces.push(e as TraceEntry), addMessage: (p, c, m) => { messages.push({ participant: p, content: c, metadata: m }); return ''; }, requestInference: (a, r, s, p) => inferenceRequests.push({ agentName: a, reason: r, source: s, ...(p ?? {}) }), @@ -450,6 +454,94 @@ describe('debounce', () => { assert.strictEqual(inferenceRequests[0].addressed, true); }); + it('an addressed batch names its registered route channel; ambient-only batches name none', async () => { + const path = writeConfig('debounce-route.json', { + policies: [ + { name: 'chat', match: { scope: ['mcpl:channel-incoming', 'mcpl:push-event'] }, behavior: { debounce: 100 } }, + ], + default: 'skip', + }); + // The host maps a push origin to its registered composite id. + const resolveRouteChannel = (info: GateEventInfo) => + info.eventType === 'mcpl:push-event' ? `discord:dm:${String(info.metadata?.channelId)}` : info.channelId; + const { gate, inferenceRequests, messages } = makeGate(path, { resolveRouteChannel }); + + // DM (addressed, raw id) then newer ambient chatter elsewhere. + gate.evaluate(event({ eventType: 'mcpl:push-event', serverId: 'discord', channelId: '1548', content: 'dm', + tags: ['chat:dm', 'chat:addressed'], metadata: { authorId: '42', channelId: '1548' } })); + gate.evaluate(event({ eventType: 'mcpl:channel-incoming', serverId: 'discord', channelId: 'discord:g:general', content: 'chatter', + metadata: { authorId: '7' } })); + await new Promise(r => setTimeout(r, 150)); + assert.strictEqual(inferenceRequests.length, 1); + assert.strictEqual(inferenceRequests[0].routeChannelId, 'discord:dm:1548', 'the addressed event names the route'); + assert.strictEqual(inferenceRequests[0].counterparty, 'discord:user:42', 'route and author come from the same event'); + assert.strictEqual(inferenceRequests[0].channelId, undefined, 'telemetry channel unchanged for push events'); + // The batched-wake line names the registered id, not the raw snowflake. + const text = JSON.stringify(messages.at(-1)?.content); + assert.ok(text.includes('discord:dm:1548'), 'gate line shows the registered channel id'); + + // Ambient-only batch: telemetry provenance, no route. + gate.evaluate(event({ eventType: 'mcpl:channel-incoming', serverId: 'discord', channelId: 'discord:g:room', content: 'x', + metadata: { authorId: '8' } })); + await new Promise(r => setTimeout(r, 150)); + assert.strictEqual(inferenceRequests.length, 2); + assert.strictEqual(inferenceRequests[1].channelId, 'discord:g:room'); + assert.strictEqual(inferenceRequests[1].routeChannelId, undefined, 'ambient chatter never sets a route'); + }); + + it('the default resolver routes an addressed channel-incoming event; a push event needs the host resolver', async () => { + const path = writeConfig('debounce-route-default.json', { + policies: [ + { name: 'chat', match: { scope: ['mcpl:channel-incoming', 'mcpl:push-event'] }, behavior: { debounce: 100 } }, + ], + default: 'skip', + }); + const { gate, inferenceRequests } = makeGate(path); + gate.evaluate(event({ eventType: 'mcpl:channel-incoming', serverId: 'discord', channelId: 'discord:g:room', content: '@agent hi', + tags: ['chat:addressed'], metadata: { authorId: '1' } })); + await new Promise(r => setTimeout(r, 150)); + assert.strictEqual(inferenceRequests[0].routeChannelId, 'discord:g:room'); + + gate.evaluate(event({ eventType: 'mcpl:push-event', serverId: 'discord', channelId: '1548', content: 'dm', + tags: ['chat:dm', 'chat:addressed'], metadata: { authorId: '42', channelId: '1548' } })); + await new Promise(r => setTimeout(r, 150)); + assert.strictEqual(inferenceRequests[1].routeChannelId, undefined, 'a raw id is never used as a route'); + }); + + it('the newest addressed event decides the route even when it has none (no guessing an older one)', async () => { + const path = writeConfig('debounce-route-newest.json', { + policies: [ + { name: 'chat', match: { scope: ['mcpl:channel-incoming', 'mcpl:push-event'] }, behavior: { debounce: 100 } }, + ], + default: 'skip', + }); + const { gate, inferenceRequests } = makeGate(path); // default resolver: push → no route + gate.evaluate(event({ eventType: 'mcpl:channel-incoming', serverId: 'discord', channelId: 'discord:g:room', content: '@agent first', + tags: ['chat:addressed'], metadata: { authorId: '1' } })); + gate.evaluate(event({ eventType: 'mcpl:push-event', serverId: 'discord', channelId: '1548', content: 'dm later', + tags: ['chat:dm', 'chat:addressed'], metadata: { authorId: '42', channelId: '1548' } })); + await new Promise(r => setTimeout(r, 150)); + assert.strictEqual(inferenceRequests[0].counterparty, 'discord:user:42'); + assert.strictEqual(inferenceRequests[0].routeChannelId, undefined, + 'the DM is the newest addressed event; replying to the older mention would answer the wrong person'); + }); + + it('a throwing route resolver never breaks gating', async () => { + const path = writeConfig('debounce-route-throw.json', { + policies: [ + { name: 'chat', match: { scope: ['mcpl:channel-incoming'] }, behavior: { debounce: 100 } }, + ], + default: 'skip', + }); + const { gate, inferenceRequests } = makeGate(path, { resolveRouteChannel: () => { throw new Error('boom'); } }); + const decision = gate.evaluate(event({ eventType: 'mcpl:channel-incoming', serverId: 'discord', channelId: 'discord:g:room', + content: '@agent hi', tags: ['chat:addressed'] })); + assert.strictEqual(decision.policyName, 'chat'); + await new Promise(r => setTimeout(r, 150)); + assert.strictEqual(inferenceRequests.length, 1); + assert.strictEqual(inferenceRequests[0].routeChannelId, undefined); + }); + it('a batched wake with no channel-bearing event carries no provenance', async () => { const path = writeConfig('debounce-noprov.json', { policies: [ diff --git a/test/fixtures/wake-locus-mcpl-server.mjs b/test/fixtures/wake-locus-mcpl-server.mjs new file mode 100644 index 000000000..203b4e0e5 --- /dev/null +++ b/test/fixtures/wake-locus-mcpl-server.mjs @@ -0,0 +1,153 @@ +// MCPL fixture for batched-wake routing tests. +// +// Registers two open guild channels and, on command, sends either +// channels/incoming traffic or a Discord-shaped DM push/event (raw snowflake +// channel id in the origin, exactly what discord-mcpl sends for a DM). The +// TEST appends lines to COMMAND_PATH: +// incoming +// dm +// Host-side channels/publish and channels/open calls are recorded to +// STATUS_PATH as JSONL. +import { appendFileSync, existsSync, readFileSync } from 'node:fs'; +import { createInterface } from 'node:readline'; + +const statusPath = process.env.STATUS_PATH; +const commandPath = process.env.COMMAND_PATH; + +const log = (event, extra = {}) => { + if (!statusPath) return; + appendFileSync(statusPath, JSON.stringify({ event, ...extra }) + '\n'); +}; +const send = (message) => process.stdout.write(JSON.stringify(message) + '\n'); +const reply = (id, result) => send({ jsonrpc: '2.0', id, result }); + +const CHANNELS = [ + { id: 'discord:g1:room', label: 'room' }, + { id: 'discord:g1:general', label: 'general' }, +]; +let nextId = 500; +let processedCommands = 0; + +function pollCommands() { + if (!commandPath || !existsSync(commandPath)) return; + const lines = readFileSync(commandPath, 'utf8').split('\n').filter(Boolean); + for (const line of lines.slice(processedCommands)) { + processedCommands++; + const [kind, ...rest] = line.split(' '); + if (kind === 'incoming') { + const [channelId, messageId, mode, ...words] = rest; + log('incoming-sent', { channelId, messageId }); + send({ + jsonrpc: '2.0', + id: nextId++, + method: 'channels/incoming', + params: { + messages: [{ + channelId, + messageId, + author: { id: 'U-other', name: 'someone' }, + timestamp: new Date().toISOString(), + content: [{ type: 'text', text: words.join(' ') }], + tags: mode === 'addressed' + ? ['chat:mention', 'chat:addressed', 'chat:from-human'] + : ['chat:ambient', 'chat:from-human'], + }], + }, + }); + } else if (kind === 'dm') { + const [eventId, authorId, rawChannelId, ...words] = rest; + log('dm-sent', { eventId }); + send({ + jsonrpc: '2.0', + id: nextId++, + method: 'push/event', + params: { + featureSet: 'chat', + eventId, + timestamp: new Date().toISOString(), + origin: { + source: 'discord', + channelId: rawChannelId, + guildId: null, + isDM: true, + authorId, + authorName: 'antra', + messageId: eventId, + }, + tags: ['chat:dm', 'chat:addressed', 'chat:private'], + payload: { content: [{ type: 'text', text: words.join(' ') }] }, + }, + }); + } + } +} +const pollTimer = setInterval(pollCommands, 50); + +const rl = createInterface({ input: process.stdin }); +rl.on('close', () => { + clearInterval(pollTimer); + process.exit(0); +}); +rl.on('line', (line) => { + let msg; + try { + msg = JSON.parse(line); + } catch { + return; + } + if (msg.method === 'initialize') { + reply(msg.id, { + protocolVersion: '2024-11-05', + capabilities: { + experimental: { + mcpl: { + version: '0.5', + pushEvents: true, + channels: { register: true, lifecycle: true, incoming: true, publish: true }, + featureSets: { + chat: { description: 'chat connector', uses: ['pushEvents', 'channels.incoming', 'channels.publish'] }, + }, + }, + }, + }, + serverInfo: { name: 'wake-locus-fixture', version: '0.0.0' }, + }); + return; + } + if (msg.method === 'featureSets/update') { + if (msg.id !== undefined && msg.id !== null) reply(msg.id, { accepted: true }); + return; + } + if (msg.method === 'notifications/initialized') { + send({ + jsonrpc: '2.0', + id: 400, + method: 'channels/register', + params: { + channels: CHANNELS.map((c) => ({ + id: c.id, type: 'discord', label: c.label, direction: 'bidirectional', initiallyOpen: true, + })), + }, + }); + return; + } + if (msg.method === 'channels/open') { + log('channel-opened', { channelId: msg.params?.channelId }); + reply(msg.id, { opened: true }); + return; + } + if (msg.method === 'channels/close') { + reply(msg.id, { closed: true }); + return; + } + if (msg.method === 'channels/publish') { + log('publish', { channelId: msg.params?.channelId }); + if (msg.id !== undefined && msg.id !== null) reply(msg.id, { delivered: true }); + return; + } + if (msg.method === 'tools/list') { + reply(msg.id, { tools: [] }); + return; + } + // Responses to our own requests (register, incoming, push) need no handling. +}); From bc21aaa8008ffc79726fb38dc84352da9819fb7b Mon Sep 17 00:00:00 2001 From: antra-tess Date: Tue, 6 Oct 2026 16:31:46 -0700 Subject: [PATCH 2/3] fix(gate): batched-wake routing review fixes - The turn's channel pick orders by event time (wakeAt, else timestamp), not queue order. A gate wake queued at its debounce flush after a busy turn describes an older event; it no longer takes the turn from a newer addressed message's direct wake. Ties keep the later-queued request. - wakeProvenance counts a resolved route as naming a channel, so an addressed push that carries only origin.mcplChannelId (no raw id, no author) still routes. - The gate route applies to a conversation fork only for its home channel, and to no other agent when a fork owns that channel (mayTakeGateRoute). - An addressed routing winner is its own telemetry provenance; it never borrows another request's counterparty. - InferenceRequest.wakeChannelId docs no longer say gate wakes never set channelId. Co-Authored-By: Claude Opus 5.5 --- .../fix-batched-wake-addressed-locus.fixed.md | 7 +++ src/framework.ts | 49 ++++++++++++++--- src/gate/event-gate.ts | 6 ++- src/types/agent.ts | 6 ++- test/batched-wake-locus.test.ts | 52 +++++++++++++++++++ test/event-gate.test.ts | 21 ++++++++ test/turn-trigger-provenance.test.ts | 42 +++++++++++++++ 7 files changed, 172 insertions(+), 11 deletions(-) diff --git a/changelog.d/fix-batched-wake-addressed-locus.fixed.md b/changelog.d/fix-batched-wake-addressed-locus.fixed.md index 27153f4ac..2540a004e 100644 --- a/changelog.d/fix-batched-wake-addressed-locus.fixed.md +++ b/changelog.d/fix-batched-wake-addressed-locus.fixed.md @@ -8,3 +8,10 @@ composite channel id via a new `EventGate` option `resolveRouteChannel` (the framework supplies it); batched-wake lines now show that id instead of the adapter's raw one. Ambient-only batches keep the legacy fallback. +- When a batch holds several addressed wakes, the newest addressed EVENT now + decides the turn's channel, not the most recently queued request. A gate + wake flushed after a busy turn no longer outranks a newer mention's direct + wake. A conversation fork takes a batched wake's route only for its home + channel, so a fork bound elsewhere shows no typing in the addressed + channel. An addressed wake with no author no longer borrows another + request's author for its turn telemetry. diff --git a/src/framework.ts b/src/framework.ts index 878a7b075..777c89bf8 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -1580,7 +1580,13 @@ export class AgentFramework { // message elsewhere could retarget between the DM's arrival and // the debounce firing (2026-09-30). Ambient-only batches set no // locus and keep the legacy fallback. - ...(provenance?.routeChannelId + // + // The wake is broadcast to every agent, conversation forks + // included, and a fork speaks only in its home channel: the + // route applies to a fork only when it IS that home, and never to + // another agent when a fork owns that channel. Otherwise a fork + // bound elsewhere would show typing where it isn't replying. + ...(provenance?.routeChannelId && framework.mayTakeGateRoute(agentName, provenance.routeChannelId) ? { channelId: provenance.routeChannelId, addressed: true } : {}), }); @@ -7334,6 +7340,21 @@ export class AgentFramework { } } + /** + * Whether a batched gate wake's addressed route may become `agentName`'s + * speech locus. The gate broadcasts to every agent, conversation forks + * included. A fork speaks only in its home channel, and a channel bound to + * a fork is that fork's conversation, so no other agent takes it. + */ + private mayTakeGateRoute(agentName: string, routeChannelId: string): boolean { + const home = this.conversationAgentHomes.get(agentName); + if (home !== undefined) return home === routeChannelId; + for (const boundChannel of this.conversationAgentHomes.values()) { + if (boundChannel === routeChannelId) return false; + } + return true; + } + /** * Derive the MCPL composite channel id (the outbound routing locus) for a * push event from its server-defined `origin`, if it carries one. @@ -7599,19 +7620,31 @@ export class AgentFramework { // and counterparty must come from the same message, or a batch of // "ambient from A, then addressed from B" would report B's channel // with A's author (review finding on the provenance change). + // + // "Most recent" is by the EVENT's time (wakeAt, else the request's own + // timestamp), not queue order. A gate wake is queued when its debounce + // flushes, which can be long after its event: an addressed message + // held in the gate while the agent was busy lands BEHIND a newer + // addressed message's direct wake, and queue order would hand the turn + // to the older one. Ties keep the later-queued request, as before. + const eventAt = (r: InferenceRequest): number => r.wakeAt ?? r.timestamp; let ambientReq: InferenceRequest | undefined; let addressedReq: InferenceRequest | undefined; for (const r of requests) { if (!r.channelId) continue; - ambientReq = r; - if (r.addressed) addressedReq = r; + if (!ambientReq || eventAt(r) >= eventAt(ambientReq)) ambientReq = r; + if (r.addressed && (!addressedReq || eventAt(r) >= eventAt(addressedReq))) addressedReq = r; } const channelReq = addressedReq ?? ambientReq; - // Telemetry provenance (who/where woke the agent): from the routing - // winner when it carries one; else from the most recent request that - // does (gate wakes name a counterparty without setting a locus), - // ordered by the event's own time (wakeAt) rather than flush order. - let provReq: InferenceRequest | undefined = channelReq?.counterparty || channelReq?.wakeChannelId ? channelReq : undefined; + // Telemetry provenance (who/where woke the agent). An ADDRESSED winner + // is its own provenance, even when it names no author (a push event + // can route without an author id): borrowing another request's author + // would credit the turn to someone who did not address the agent. + // Otherwise from the routing winner when it carries one; else from the + // most recent request that does (ambient gate wakes name a counterparty + // without setting a locus), ordered by the event's own time. + let provReq: InferenceRequest | undefined = addressedReq + ?? (channelReq?.counterparty || channelReq?.wakeChannelId ? channelReq : undefined); if (!provReq) { for (const r of requests) { if (!r.counterparty && !r.wakeChannelId) continue; diff --git a/src/gate/event-gate.ts b/src/gate/event-gate.ts index 01e6cf505..c324dce0c 100644 --- a/src/gate/event-gate.ts +++ b/src/gate/event-gate.ts @@ -204,7 +204,11 @@ export function wakeProvenance(events: PendingEvent[]): WakeProvenance | undefin let newest: PendingEvent | undefined; let newestAddressed: PendingEvent | undefined; for (const e of events) { - if (!e.channelId && !e.authorId) continue; + // A resolved route counts as naming a channel: a push whose origin + // declares only `mcplChannelId` (no raw channel id, no author) is still a + // routable addressed message, and skipping it here would leave its wake + // on the legacy most-recent-inbound fallback. + if (!e.channelId && !e.authorId && !e.routeChannelId) continue; if (!newest || e.timestamp >= newest.timestamp) newest = e; if (e.addressed && (!newestAddressed || e.timestamp >= newestAddressed.timestamp)) newestAddressed = e; } diff --git a/src/types/agent.ts b/src/types/agent.ts index b5cd7471f..b18b6a637 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -330,8 +330,10 @@ export interface InferenceRequest { * Where the wake came from, for telemetry ONLY — never a speech locus. * Set by the EventGate for batched wakes (composite channel id of the * chosen event when the event carried one; push-event raw ids are not - * used). Routing keeps reading `channelId`, which only the direct channel - * paths set from normalized ids — a gate wake leaves it unset, as before. + * used). Routing keeps reading `channelId`, set from registered ids: by + * the direct channel paths, and by a gate wake only when its chosen event + * addressed the agent and the host resolved that event's registered + * channel (ambient-only batches leave it unset). */ wakeChannelId?: string; /** Timestamp of the event the wake provenance was taken from (ms). */ diff --git a/test/batched-wake-locus.test.ts b/test/batched-wake-locus.test.ts index 9ef4c4037..ab2df9035 100644 --- a/test/batched-wake-locus.test.ts +++ b/test/batched-wake-locus.test.ts @@ -129,3 +129,55 @@ describe('batched wake routing', () => { assert.deepEqual(publishes(), [GENERAL]); }); }); + +describe('batched wake routing with conversation forks', () => { + let tempDir: string; + let framework: AgentFramework; + + beforeEach(async () => { + tempDir = mkdtempSync(join(tmpdir(), 'wake-locus-fork-')); + framework = await AgentFramework.create({ + storePath: join(tempDir, 'test.chronicle'), + membrane: new MockMembrane().asMembrane(), + agents: [ + { name: 'trunk', model: 'test-model', systemPrompt: 'You are trunk.' }, + { name: 'fork-room', model: 'test-model', systemPrompt: 'You are a fork.' }, + { name: 'fork-other', model: 'test-model', systemPrompt: 'You are a fork.' }, + ], + gate: { + config: { + policies: [{ name: 'batch', match: { scope: ['mcpl:channel-incoming'] }, behavior: { debounce: 50 } }], + default: 'skip', + }, + }, + modules: [], + }); + }); + + afterEach(async () => { + await framework.stop(); + rmSync(tempDir, { recursive: true, force: true }); + }); + + it('the addressed route applies only to the fork homed there; other forks and the trunk get none', async () => { + const internals = framework as unknown as { + conversationAgentHomes: Map; + pendingRequests: Array<{ agentName: string; channelId?: string; addressed?: boolean; counterparty?: string }>; + eventGate: { evaluate(info: unknown): unknown }; + }; + internals.conversationAgentHomes.set('fork-room', ROOM); + internals.conversationAgentHomes.set('fork-other', GENERAL); + internals.eventGate.evaluate({ + content: '@agent hi', eventType: 'mcpl:channel-incoming', serverId: 'discord', channelId: ROOM, + tags: ['chat:addressed'], metadata: { authorId: '5' }, + }); + await waitFor(() => internals.pendingRequests.some((r) => r.agentName === 'fork-other'), 'gate wake broadcast'); + const byAgent = new Map(internals.pendingRequests.map((r) => [r.agentName, r])); + assert.equal(byAgent.get('fork-room')?.channelId, ROOM, 'the fork that owns the channel takes the route'); + assert.equal(byAgent.get('fork-room')?.addressed, true); + assert.equal(byAgent.get('fork-other')?.channelId, undefined, 'a fork bound elsewhere gets no locus'); + assert.equal(byAgent.get('trunk')?.channelId, undefined, 'the trunk never takes a fork-bound channel'); + // Telemetry provenance still reaches every agent. + assert.equal(byAgent.get('fork-other')?.counterparty, 'discord:user:5'); + }); +}); diff --git a/test/event-gate.test.ts b/test/event-gate.test.ts index e469e084e..7b55cf889 100644 --- a/test/event-gate.test.ts +++ b/test/event-gate.test.ts @@ -526,6 +526,27 @@ describe('debounce', () => { 'the DM is the newest addressed event; replying to the older mention would answer the wrong person'); }); + it('an addressed push that names only its registered channel (no raw id, no author) still routes', async () => { + const path = writeConfig('debounce-route-only.json', { + policies: [ + { name: 'chat', match: { scope: ['mcpl:channel-incoming', 'mcpl:push-event'] }, behavior: { debounce: 100 } }, + ], + default: 'skip', + }); + const resolveRouteChannel = (info: GateEventInfo) => + typeof info.metadata?.mcplChannelId === 'string' ? info.metadata.mcplChannelId : undefined; + const { gate, inferenceRequests } = makeGate(path, { resolveRouteChannel }); + gate.evaluate(event({ eventType: 'mcpl:channel-incoming', serverId: 'surface', channelId: 'surface:room:general', + content: 'chatter', metadata: { authorId: '7' } })); + gate.evaluate(event({ eventType: 'mcpl:push-event', serverId: 'surface', content: 'dm', + tags: ['chat:dm', 'chat:addressed'], metadata: { mcplChannelId: 'surface:dm:42' } })); + await new Promise(r => setTimeout(r, 150)); + assert.strictEqual(inferenceRequests.length, 1); + assert.strictEqual(inferenceRequests[0].routeChannelId, 'surface:dm:42'); + assert.strictEqual(inferenceRequests[0].addressed, true); + assert.strictEqual(inferenceRequests[0].counterparty, undefined, 'no author on the chosen event, none borrowed from the chatter'); + }); + it('a throwing route resolver never breaks gating', async () => { const path = writeConfig('debounce-route-throw.json', { policies: [ diff --git a/test/turn-trigger-provenance.test.ts b/test/turn-trigger-provenance.test.ts index 793dc4b15..df75affc1 100644 --- a/test/turn-trigger-provenance.test.ts +++ b/test/turn-trigger-provenance.test.ts @@ -102,6 +102,48 @@ describe('Turn trigger provenance', () => { await framework.stop(); }); + it('a gate wake queued after a NEWER addressed direct wake does not take the turn', async () => { + // The agent was busy: Alice's DM sat in the gate's buffer while Bob's + // newer mention queued a direct wake. The gate flushed at turn end, so its + // request is queued AFTER Bob's but describes an OLDER event. + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hi bob' }])); + const framework = await makeFramework(); + const i = internals(framework); + const captured = spy(framework); + const t = Date.now(); + i.pendingRequests.push( + { agentName: 'scout', reason: 'mcpl:channel-incoming', source: 'discord', timestamp: t, + channelId: 'discord:g:bob-room', counterparty: 'discord:user:bob', addressed: true }, + { agentName: 'scout', reason: 'gate:debounce', source: 'gate', timestamp: t + 5_000, + channelId: 'discord:dm:alice', addressed: true, counterparty: 'discord:user:alice', wakeAt: t - 2_000 }, + ); + await i.processInferenceRequests(); + await framework.runUntilIdle(); + assert.equal(captured.handed?.channelId, 'discord:g:bob-room', 'the newer addressed EVENT wins, not the later-queued request'); + assert.equal(captured.handed?.counterparty, 'discord:user:bob'); + await framework.stop(); + }); + + it('an addressed winner with no author never borrows another request\'s author', async () => { + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hi' }])); + const framework = await makeFramework(); + const i = internals(framework); + const captured = spy(framework); + const t = Date.now(); + i.pendingRequests.push( + { agentName: 'scout', reason: 'mcpl:channel-incoming', source: 'discord', timestamp: t, + channelId: 'discord:g:general', counterparty: 'discord:user:carol', addressed: false }, + // An addressed push routed by origin.mcplChannelId alone: no author id. + { agentName: 'scout', reason: 'gate:debounce', source: 'gate', timestamp: t + 1, + channelId: 'surface:dm:42', addressed: true, wakeAt: t + 1 }, + ); + await i.processInferenceRequests(); + await framework.runUntilIdle(); + assert.equal(captured.handed?.channelId, 'surface:dm:42'); + assert.equal(captured.handed?.counterparty, undefined, 'carol did not address the agent'); + await framework.stop(); + }); + it('a context-budget restart keeps the channel for routing but names no author', async () => { membrane.pushResponse(createMockResponse([{ type: 'text', text: 'continuing' }])); const framework = await makeFramework(); From 3a7a22a4d572071a671798c9dce488381f4ce7c3 Mon Sep 17 00:00:00 2001 From: antra-tess Date: Tue, 6 Oct 2026 16:40:55 -0700 Subject: [PATCH 3/3] fix(gate): an ambient route-only event never becomes the wake's provenance bc21aaa let any event with a resolved route qualify as a provenance candidate. The route is reported only for addressed events, so an AMBIENT push naming only its registered channel, if newest, became the pick and carried no channel or author telemetry at all, hiding an earlier event that named both. Only an addressed event's route now makes it a candidate. Co-Authored-By: Claude Opus 5.5 --- src/gate/event-gate.ts | 13 ++++++++----- test/event-gate.test.ts | 21 +++++++++++++++++++++ 2 files changed, 29 insertions(+), 5 deletions(-) diff --git a/src/gate/event-gate.ts b/src/gate/event-gate.ts index c324dce0c..bb8bd3d1a 100644 --- a/src/gate/event-gate.ts +++ b/src/gate/event-gate.ts @@ -204,11 +204,14 @@ export function wakeProvenance(events: PendingEvent[]): WakeProvenance | undefin let newest: PendingEvent | undefined; let newestAddressed: PendingEvent | undefined; for (const e of events) { - // A resolved route counts as naming a channel: a push whose origin - // declares only `mcplChannelId` (no raw channel id, no author) is still a - // routable addressed message, and skipping it here would leave its wake - // on the legacy most-recent-inbound fallback. - if (!e.channelId && !e.authorId && !e.routeChannelId) continue; + // A resolved route counts as naming a channel only for an ADDRESSED + // event: a push whose origin declares only `mcplChannelId` (no raw + // channel id, no author) is still a routable addressed message, and + // skipping it would leave its wake on the most-recent-inbound fallback. + // An ambient one stays out: the route is never reported for it, so as + // the pick it would carry no telemetry at all and hide an earlier event + // that names its channel and author. + if (!e.channelId && !e.authorId && !(e.addressed && e.routeChannelId)) continue; if (!newest || e.timestamp >= newest.timestamp) newest = e; if (e.addressed && (!newestAddressed || e.timestamp >= newestAddressed.timestamp)) newestAddressed = e; } diff --git a/test/event-gate.test.ts b/test/event-gate.test.ts index 7b55cf889..637cf1029 100644 --- a/test/event-gate.test.ts +++ b/test/event-gate.test.ts @@ -547,6 +547,27 @@ describe('debounce', () => { assert.strictEqual(inferenceRequests[0].counterparty, undefined, 'no author on the chosen event, none borrowed from the chatter'); }); + it('a newer AMBIENT route-only push does not hide an earlier event\'s channel and author', async () => { + const path = writeConfig('debounce-route-only-ambient.json', { + policies: [ + { name: 'chat', match: { scope: ['mcpl:channel-incoming', 'mcpl:push-event'] }, behavior: { debounce: 100 } }, + ], + default: 'skip', + }); + const resolveRouteChannel = (info: GateEventInfo) => + typeof info.metadata?.mcplChannelId === 'string' ? info.metadata.mcplChannelId : info.channelId || undefined; + const { gate, inferenceRequests } = makeGate(path, { resolveRouteChannel }); + gate.evaluate(event({ eventType: 'mcpl:channel-incoming', serverId: 'surface', channelId: 'surface:room:general', + content: 'chatter', metadata: { authorId: '7' } })); + gate.evaluate(event({ eventType: 'mcpl:push-event', serverId: 'surface', content: 'status ping', + metadata: { mcplChannelId: 'surface:room:feed' } })); + await new Promise(r => setTimeout(r, 150)); + assert.strictEqual(inferenceRequests.length, 1); + assert.strictEqual(inferenceRequests[0].channelId, 'surface:room:general', 'telemetry keeps the attributable event'); + assert.strictEqual(inferenceRequests[0].counterparty, 'surface:user:7'); + assert.strictEqual(inferenceRequests[0].routeChannelId, undefined, 'ambient batches never route'); + }); + it('a throwing route resolver never breaks gating', async () => { const path = writeConfig('debounce-route-throw.json', { policies: [