From a4835d9c0e7d520c4acec79c17a5bc2129b47360 Mon Sep 17 00:00:00 2001 From: Sage-307 Date: Fri, 2 Oct 2026 01:31:52 -0400 Subject: [PATCH 1/2] fix(routing): refresh idle activity after delivery instead of queries ConversationRouter.route currently refreshes lastActivity while deciding where a message would go. A diagnostic lookup or failed context write therefore keeps a fork alive, while fixed-audience coalesced delivery bypasses that refresh entirely. Make route a pure query and expose touch(channelId, now) as the explicit activity command. Remove the unused query clock from IncomingMessageFacts and document the direct-caller migration. The framework touches after a successful context append, including fixed-audience delivery only when the recipient is still the active binding. Delivery to an older engagement cannot prolong its replacement. Self-echo identity policy (#47) stays outside this change. Add query immutability and delivery-boundary regressions, preserve ambient activity, cover failed writes and old-generation corrections, and move deterministic expiry coverage to touch. With the added tests and original source, the focused compiled suite gives 24 pass / 3 fail; changed source gives 27 pass / 0 fail. Build and diff checks pass. Full suite and independent review follow. Co-Authored-By: GPT-6 Astra --- CHANGELOG.md | 4 ++ src/framework.ts | 6 ++ src/mcpl/conversation-router.ts | 20 ++++--- test/conversation-router.test.ts | 54 +++++++++++------- test/conversation-routing.test.ts | 93 +++++++++++++++++++++++++++++++ 5 files changed, 150 insertions(+), 27 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b9a30b4c..1cf0cd586 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,10 @@ Releases up to and including 0.7.3 predate this file; for their contents see ## Unreleased +### Breaking + +- **Direct ConversationRouter callers:** `route()` is now a pure query and no longer refreshes idle activity (#46). After successfully delivering a message to the bound fork, call `touch(channelId)`; deterministic clocks move from `route({ now })` to `touch(channelId, now)`. Framework-managed delivery performs this step automatically, including ambient messages and coalesced fixed-audience delivery to the current binding. Queries, failed writes, and delivery to an older engagement leave the current binding's clock unchanged. Bind rules, trigger rules, generation counters, and fork names are unchanged. No store migration or sibling dependency upgrade is required. + ## 0.19.0 — 2026-09-28 ### Added diff --git a/src/framework.ts b/src/framework.ts index 510b9183a..f65ce1480 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -6890,6 +6890,11 @@ export class AgentFramework { if (this.conversationRouter && this.conversationAgentHomes.has(target.name)) { metadata.triggered = event.triggerInference ?? false; const id = target.getContextManager().addMessage('user', event.content, metadata); + // A correction may target an older engagement. Refresh only the + // binding whose agent actually received the message. + if (this.conversationRouter.getBinding(event.channelId)?.agentName === target.name) { + this.conversationRouter.touch(event.channelId); + } this.emitTrace({ type: 'message:added', messageId: id, source: 'mcpl:channel-incoming' }); if (event.triggerInference) { this.pendingRequests.push({ @@ -7069,6 +7074,7 @@ export class AgentFramework { messageMetadata.triggered = trigger; const id = agent.getContextManager().addMessage('user', event.content, messageMetadata); + router.touch(event.channelId); this.emitTrace({ type: 'message:added', messageId: id, source: 'mcpl:channel-incoming' }); if (trigger) { diff --git a/src/mcpl/conversation-router.ts b/src/mcpl/conversation-router.ts index e96cb5703..1dba1de96 100644 --- a/src/mcpl/conversation-router.ts +++ b/src/mcpl/conversation-router.ts @@ -95,8 +95,6 @@ export interface IncomingMessageFacts { mentioned: boolean; /** Channel classification (see classifyChannel). */ kind: ChannelKind; - /** Clock injection for tests; defaults to Date.now(). */ - now?: number; } export type RoutingDecision = @@ -149,17 +147,15 @@ export class ConversationRouter { } /** - * Decide where an incoming message goes. Pure read except for refreshing - * lastActivity on an existing binding — a 'spawn' decision does NOT create - * the binding; the framework calls bind() after the fork agent actually - * exists, so a failed spawn doesn't leave a dangling route. + * Decide where an incoming message goes without changing router state. + * The framework calls bind() after a fork exists and touch() after a + * message is delivered. Queries and failed deliveries leave idle clocks + * unchanged; a failed spawn doesn't leave a dangling route. */ route(facts: IncomingMessageFacts): RoutingDecision { - const now = facts.now ?? Date.now(); const existing = this.bindings.get(facts.channelId); if (existing) { - existing.lastActivity = now; return { kind: 'existing', agentName: existing.agentName, @@ -202,6 +198,14 @@ export class ConversationRouter { return binding; } + /** Record delivered activity on an existing binding. Does not bind a + * channel or advance its generation. Call after successful delivery, not + * when previewing a route. The optional clock supports deterministic use. */ + touch(channelId: string, now = Date.now()): void { + const binding = this.bindings.get(channelId); + if (binding) binding.lastActivity = now; + } + unbind(channelId: string): void { this.bindings.delete(channelId); } diff --git a/test/conversation-router.test.ts b/test/conversation-router.test.ts index 9a1697784..3e921a6fd 100644 --- a/test/conversation-router.test.ts +++ b/test/conversation-router.test.ts @@ -15,7 +15,7 @@ function makeRouter(overrides: Partial { const router = makeRouter(); - const decision = router.route({ channelId: 'slack:D1', mentioned: false, kind: 'dm', now: T0 }); + const decision = router.route({ channelId: 'slack:D1', mentioned: false, kind: 'dm' }); assert.equal(decision.kind, 'spawn'); if (decision.kind === 'spawn') { assert.equal(decision.generation, 1); @@ -26,19 +26,19 @@ test('DM message binds without a mention (default dm bind: always)', () => { test('channel message without mention stays unbound (default channel bind: mention)', () => { const router = makeRouter(); - const decision = router.route({ channelId: 'slack:C1', mentioned: false, kind: 'channel', now: T0 }); + const decision = router.route({ channelId: 'slack:C1', mentioned: false, kind: 'channel' }); assert.equal(decision.kind, 'unbound'); }); test('channel mention spawns a fork', () => { const router = makeRouter(); - const decision = router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel', now: T0 }); + const decision = router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel' }); assert.equal(decision.kind, 'spawn'); }); test('bind rule never keeps DMs unbound', () => { const router = makeRouter({ bind: { dm: 'never' } }); - const decision = router.route({ channelId: 'slack:D1', mentioned: true, kind: 'dm', now: T0 }); + const decision = router.route({ channelId: 'slack:D1', mentioned: true, kind: 'dm' }); assert.equal(decision.kind, 'unbound'); }); @@ -48,17 +48,17 @@ test('bind rule never keeps DMs unbound', () => { test('spawn decision does not create the binding until bind() is called', () => { const router = makeRouter(); - const d1 = router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel', now: T0 }); + const d1 = router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel' }); assert.equal(d1.kind, 'spawn'); assert.equal(router.getBinding('slack:C1'), undefined); // Framework failed to spawn → next mention proposes the same generation. - const d2 = router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel', now: T0 + 1 }); + const d2 = router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel' }); assert.equal(d2.kind, 'spawn'); if (d2.kind === 'spawn') assert.equal(d2.generation, 1); if (d2.kind === 'spawn') router.bind('slack:C1', d2.agentName, d2.generation, T0 + 1); - const d3 = router.route({ channelId: 'slack:C1', mentioned: false, kind: 'channel', now: T0 + 2 }); + const d3 = router.route({ channelId: 'slack:C1', mentioned: false, kind: 'channel' }); assert.equal(d3.kind, 'existing'); }); @@ -70,11 +70,11 @@ test('bound channel: non-mentions land without triggering, mentions trigger', () const router = makeRouter(); router.bind('slack:C1', 'conversation-slack-C1-g1', 1, T0); - const ambient = router.route({ channelId: 'slack:C1', mentioned: false, kind: 'channel', now: T0 + 1 }); + const ambient = router.route({ channelId: 'slack:C1', mentioned: false, kind: 'channel' }); assert.equal(ambient.kind, 'existing'); if (ambient.kind === 'existing') assert.equal(ambient.trigger, false); - const mention = router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel', now: T0 + 2 }); + const mention = router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel' }); assert.equal(mention.kind, 'existing'); if (mention.kind === 'existing') assert.equal(mention.trigger, true); }); @@ -82,7 +82,7 @@ test('bound channel: non-mentions land without triggering, mentions trigger', () test('bound DM: every message triggers', () => { const router = makeRouter(); router.bind('slack:D1', 'conversation-slack-D1-g1', 1, T0); - const decision = router.route({ channelId: 'slack:D1', mentioned: false, kind: 'dm', now: T0 + 1 }); + const decision = router.route({ channelId: 'slack:D1', mentioned: false, kind: 'dm' }); assert.equal(decision.kind, 'existing'); if (decision.kind === 'existing') assert.equal(decision.trigger, true); }); @@ -91,25 +91,41 @@ test('bound DM: every message triggers', () => { // TTL + generations // --------------------------------------------------------------------------- -test('expired() respects lastActivity refresh from route()', () => { +test('routing queries leave the binding and its expiry clock unchanged', () => { + const router = makeRouter({ idleTtlMs: 1000 }); + router.bind('slack:C1', 'fork-a', 3, T0); + const before = { ...router.getBinding('slack:C1')! }; + const generations = router.exportGenerations(); + + assert.equal(router.route({ channelId: 'slack:C1', mentioned: false, kind: 'channel' }).kind, 'existing'); + assert.equal(router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel' }).kind, 'existing'); + assert.deepEqual(router.getBinding('slack:C1'), before, 'previewing a route must not prolong an engagement'); + assert.deepEqual(router.exportGenerations(), generations); + assert.equal(router.expired(T0 + 1001).length, 1); +}); + +test('touch() refreshes expiry explicitly and leaves unbound channels alone', () => { const router = makeRouter({ idleTtlMs: 1000 }); router.bind('slack:C1', 'conversation-slack-C1-g1', 1, T0); - // Ambient traffic keeps the binding alive even without triggering. - router.route({ channelId: 'slack:C1', mentioned: false, kind: 'channel', now: T0 + 900 }); + // Delivery records activity independently of the routing query. + router.touch('slack:C1', T0 + 900); + router.touch('slack:unknown', T0 + 900); + assert.equal(router.getBinding('slack:unknown'), undefined); + assert.deepEqual(router.exportGenerations(), { 'slack:C1': 1 }); assert.deepEqual(router.expired(T0 + 1500), []); assert.equal(router.expired(T0 + 1901).length, 1); }); test('rebind after expiry gets a fresh generation and agent name', () => { const router = makeRouter({ idleTtlMs: 1000 }); - const d1 = router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel', now: T0 }); + const d1 = router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel' }); if (d1.kind !== 'spawn') assert.fail('expected spawn'); router.bind('slack:C1', d1.agentName, d1.generation, T0); router.unbind('slack:C1'); // framework does this after the closure turn - const d2 = router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel', now: T0 + 5000 }); + const d2 = router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel' }); assert.equal(d2.kind, 'spawn'); if (d2.kind === 'spawn') { assert.equal(d2.generation, 2); @@ -153,11 +169,11 @@ test('classifyChannel reads descriptor metadata and message channel_type', () => test('group DMs bind on any message but trigger only on mention by default', () => { const router = new ConversationRouter({ templateAgent: 'main' }); - const ambient = router.route({ channelId: 'slack:G1', mentioned: false, kind: 'groupDm', now: T0 }); + const ambient = router.route({ channelId: 'slack:G1', mentioned: false, kind: 'groupDm' }); assert.equal(ambient.kind, 'spawn', 'group DM binds without a mention'); assert.equal((ambient as { trigger: boolean }).trigger, false, 'no firehose replies in group DMs'); - const mention = router.route({ channelId: 'slack:G2', mentioned: true, kind: 'groupDm', now: T0 }); + const mention = router.route({ channelId: 'slack:G2', mentioned: true, kind: 'groupDm' }); assert.equal(mention.kind, 'spawn'); assert.equal((mention as { trigger: boolean }).trigger, true, 'mention triggers'); }); @@ -171,14 +187,14 @@ test('generation counters export/hydrate round-trips and never regresses', () => const restored = new ConversationRouter({ templateAgent: 'main' }); restored.hydrateGenerations(exported); - const decision = restored.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel', now: T0 }); + const decision = restored.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel' }); assert.equal(decision.kind, 'spawn'); assert.equal((decision as { generation: number }).generation, 4, 'restart must not reuse generation names'); // Hydration never regresses a counter that advanced in the meantime. restored.bind('slack:D1', 'conversation-slack-D1-g5', 5, T0); restored.hydrateGenerations(exported); - const d1 = restored.route({ channelId: 'slack:D2', mentioned: false, kind: 'dm', now: T0 }); + const d1 = restored.route({ channelId: 'slack:D2', mentioned: false, kind: 'dm' }); assert.equal(d1.kind, 'spawn', 'sanity'); assert.equal(restored.exportGenerations()['slack:D1'], 5); }); diff --git a/test/conversation-routing.test.ts b/test/conversation-routing.test.ts index cdd3ed954..db3239b06 100644 --- a/test/conversation-routing.test.ts +++ b/test/conversation-routing.test.ts @@ -181,6 +181,99 @@ describe('Conversation routing', () => { await framework.stop(); }); + // Exercise the delivery boundary without scheduling inference. The same + // handler receives ordinary channel traffic and coalesced fixed-audience notices. + async function deliver(framework: AgentFramework, channelId: string, text: string, deliverTo?: string) { + const internals = framework as unknown as { + handleMcplChannelIncoming(event: unknown): Promise; + }; + return internals.handleMcplChannelIncoming({ + ...incomingEvent({ channelId, text, mentioned: true }), + triggerInference: false, + ...(deliverTo ? { deliverTo } : {}), + }); + } + + it('successful ambient delivery refreshes the bound fork idle clock', async () => { + const framework = await makeFramework(); + try { + await deliver(framework, 'slack:C1', 'first'); + const router = framework.getConversationRouter()!; + const binding = router.getBinding('slack:C1')!; + binding.lastActivity = 1; + const before = Date.now(); + framework.pushEvent({ + ...incomingEvent({ channelId: 'slack:C1', text: 'ambient detail' }), + triggerInference: false, + } as ProcessEvent); + await framework.runUntilIdle(); + + assert.ok(binding.lastActivity >= before, 'successful delivery records activity'); + assert.equal(membrane.calls.length, 0, 'ambient delivery needs no inference'); + assert.deepEqual(router.expired(before), []); + } finally { + await framework.stop(); + } + }); + + it('failed delivery leaves the bound fork idle clock unchanged', async () => { + const framework = await makeFramework(); + try { + await deliver(framework, 'slack:C1', 'first'); + const router = framework.getConversationRouter()!; + const binding = router.getBinding('slack:C1')!; + binding.lastActivity = 1; + const context = framework.getAgent(binding.agentName)!.getContextManager(); + const addMessage = context.addMessage; + context.addMessage = () => { throw new Error('injected context write failure'); }; + try { + await assert.rejects(deliver(framework, 'slack:C1', 'lost'), /injected context write failure/); + assert.equal(binding.lastActivity, 1, 'a routing decision is not delivered activity'); + } finally { + context.addMessage = addMessage; + } + } finally { + await framework.stop(); + } + }); + + it('fixed-audience coalesced delivery refreshes the current fork idle clock', async () => { + const framework = await makeFramework(); + try { + await deliver(framework, 'slack:C1', 'first'); + const router = framework.getConversationRouter()!; + const binding = router.getBinding('slack:C1')!; + binding.lastActivity = 1; + const before = Date.now(); + await deliver(framework, 'slack:C1', 'updated detail', binding.agentName); + assert.ok(binding.lastActivity >= before, 'delivery bypassing fresh routing still records activity'); + const { messages } = await framework.getAgent(binding.agentName)!.getContextManager().compile(); + assert.ok(JSON.stringify(messages).includes('updated detail')); + } finally { + await framework.stop(); + } + }); + + it('delivery to an older fork does not refresh a newer binding on the same channel', async () => { + const framework = await makeFramework(); + try { + await deliver(framework, 'slack:C1', 'first'); + const router = framework.getConversationRouter()!; + const oldName = router.getBinding('slack:C1')!.agentName; + router.unbind('slack:C1'); + await deliver(framework, 'slack:C1', 'new engagement'); + const binding = router.getBinding('slack:C1')!; + binding.lastActivity = 1; + await deliver(framework, 'slack:C1', 'old engagement correction', oldName); + + assert.equal(binding.lastActivity, 1, 'the correction belongs to the old engagement'); + const { messages } = await framework.getAgent(oldName)!.getContextManager().compile(); + assert.ok(JSON.stringify(messages).includes('old engagement correction')); + } finally { + await framework.stop(); + } + }); + it('idle TTL runs a closure turn, unbinds, disposes the fork, and the next message spawns g2', async () => { const IDLE_TTL_MS = 60_000; membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hello!' }])); From c7da9b4e090faa4b92c878800312d22414e08295 Mon Sep 17 00:00:00 2001 From: Sage-307 Date: Fri, 2 Oct 2026 14:11:32 -0400 Subject: [PATCH 2/2] docs: move PR release notes to a unique changelog fragment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The upstream reviewer identified conflicts among concurrent PRs that edit Unreleased directly. Felix explicitly authorized this unique fragment in Commons message 19989. Move only this PR’s own bullets verbatim, preserving existing release history and the reviewed runtime/test tree. The fragment satisfies the added-file changelog check; integration and release-fold validation follow before push. Co-Authored-By: GPT-6 Astra --- CHANGELOG.md | 4 ---- changelog.d/209-pure-routing-query.breaking.md | 1 + 2 files changed, 1 insertion(+), 4 deletions(-) create mode 100644 changelog.d/209-pure-routing-query.breaking.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 1cf0cd586..6b9a30b4c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,10 +12,6 @@ Releases up to and including 0.7.3 predate this file; for their contents see ## Unreleased -### Breaking - -- **Direct ConversationRouter callers:** `route()` is now a pure query and no longer refreshes idle activity (#46). After successfully delivering a message to the bound fork, call `touch(channelId)`; deterministic clocks move from `route({ now })` to `touch(channelId, now)`. Framework-managed delivery performs this step automatically, including ambient messages and coalesced fixed-audience delivery to the current binding. Queries, failed writes, and delivery to an older engagement leave the current binding's clock unchanged. Bind rules, trigger rules, generation counters, and fork names are unchanged. No store migration or sibling dependency upgrade is required. - ## 0.19.0 — 2026-09-28 ### Added diff --git a/changelog.d/209-pure-routing-query.breaking.md b/changelog.d/209-pure-routing-query.breaking.md new file mode 100644 index 000000000..93027b4db --- /dev/null +++ b/changelog.d/209-pure-routing-query.breaking.md @@ -0,0 +1 @@ +- **Direct ConversationRouter callers:** `route()` is now a pure query and no longer refreshes idle activity (#46). After successfully delivering a message to the bound fork, call `touch(channelId)`; deterministic clocks move from `route({ now })` to `touch(channelId, now)`. Framework-managed delivery performs this step automatically, including ambient messages and coalesced fixed-audience delivery to the current binding. Queries, failed writes, and delivery to an older engagement leave the current binding's clock unchanged. Bind rules, trigger rules, generation counters, and fork names are unchanged. No store migration or sibling dependency upgrade is required.