Skip to content

fix(gate): batched wakes route to their addressed message's channel #200

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
antra-tess merged 3 commits into from
Oct 7, 2026
Merged

fix(gate): batched wakes route to their addressed message's channel #200

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2de01db
fix(gate): batched wakes route to their addressed message's channel
antra-tess Oct 1, 2026
bc21aaa
fix(gate): batched-wake routing review fixes
antra-tess Oct 6, 2026
3a7a22a
fix(gate): an ambient route-only event never becomes the wake's prove…
antra-tess Oct 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions changelog.d/fix-batched-wake-addressed-locus.fixed.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
- 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.
- 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.
72 changes: 62 additions & 10 deletions src/framework.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1569,16 +1569,41 @@ 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.
//
// 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 }
: {}),
});
Comment on lines +1589 to 1592

@greptile-apps greptile-apps Bot Oct 6, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Fork wake misroutes trunk speech When an addressed batched wake belongs to a conversation fork, this guard withholds the channel from the trunk but still queues the trunk for inference. If ambient traffic has made another channel the global fallback, the trunk can publish its response there instead of in the addressed channel.

Knowledge Base Used: MCPL conversation and inference routing

Prompt To Fix With AI 
This is a comment left during a code review.
Path: src/framework.ts
Line: 1589-1592

Comment:
**Fork wake misroutes trunk speech** When an addressed batched wake belongs to a conversation fork, this guard withholds the channel from the trunk but still queues the trunk for inference. If ambient traffic has made another channel the global fallback, the trunk can publish its response there instead of in the addressed channel.

**Knowledge Base Used:** [MCPL conversation and inference routing](https://app.greptile.com/anima-labs/-/custom-context/knowledge-base/anima-research/agent-framework/-/docs/mcpl-conversation-and-inference.md)

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

@antra-tess antra-tess Oct 6, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not fixing this here. It already happens on main. There, getAgentNames broadcasts every gate wake to the trunk, and requestInference sets no locus, so a fork-owned addressed batch already wakes the trunk onto the most-recent-inbound fallback. The guard leaves the trunk exactly where main has it. Before the guard, the trunk would have answered inside the fork's own channel, a second reply next to the fork's.

The real fix is per-agent wake provenance: work out each agent's provenance only from events it owns, and skip agents that own none. That changes gate broadcast semantics. Conversation routing is also being deprecated (chore/deprecate-conversation-routing), and no deployed config enables it. That makes it a follow-up at most, not part of this PR.

@greptile-apps greptile-apps Bot Oct 6, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That’s fair. Given the current broadcast semantics, mayTakeGateRoute prevents the trunk from producing a second reply in the fork’s channel while preserving the existing main behavior for the trunk (waking it on the legacy fallback). Fixing that remaining misroute requires per-agent provenance and changing which agents receive or act on a gate wake, which is separate from this PR—especially with conversation routing being deprecated and disabled in deployed configurations. I’ll treat the earlier finding as non-blocking and withdraw it for this change.

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

Expand Down Expand Up @@ -7315,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.
Expand Down Expand Up @@ -7580,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;
Expand Down
66 changes: 59 additions & 7 deletions src/gate/event-gate.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}

/**
Expand All @@ -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;
Expand Down Expand Up @@ -190,7 +204,14 @@ 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 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;
}
Expand All @@ -204,10 +225,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 } : {}),
Comment thread
greptile-apps[bot] marked this conversation as resolved.
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;
Expand Down Expand Up @@ -587,6 +618,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;
Expand All @@ -608,13 +640,18 @@ 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;
this.emitTrace = opts.emitTrace;
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();

Expand Down Expand Up @@ -1432,6 +1469,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;

Expand All @@ -1451,6 +1497,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);
Expand Down Expand Up @@ -1514,7 +1561,10 @@ export class EventGate {
if (referenced.length > 0) {
const byChannel = new Map<string, { count: number; oldest: number; label?: string }>();
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);
Expand All @@ -1541,11 +1591,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);
Expand Down
6 changes: 4 additions & 2 deletions src/types/agent.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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). */
Expand Down
Loading
Loading