From 17034b1544c94aec6aa158d5b3fcbb88ff18a40c Mon Sep 17 00:00:00 2001 From: Iris-308 Date: Fri, 2 Oct 2026 01:18:09 -0400 Subject: [PATCH 1/7] fix(routing): deliver speech failure notices to the speaking agent Use the route failure callback's conversationId to target registered agents through addMessage. This keeps a fork's failure notice in its isolated window and evaluates mid-turn deferral against the speaker instead of the primary. Preserve the primary fallback for unknown conversation IDs and the context-only, no-wake behavior. Add five framework integration tests using the real ChannelRegistry callback and Chronicle contexts. Cover primary and unknown-ID behavior, secondary-agent manager selection, secondary live-turn deferral, and an actual incoming-DM fork whose failed reply is followed by its notice without another inference. Ordinary residents share a message slot, so the secondary test checks the receiving manager while the fork test verifies window isolation. Validation: compiled new tests on original source gave 2 pass / 3 fail (secondary targeting, deferral, and fork isolation). After the fix, the compiled targeting and existing registry routing tests gave 27 pass / 0 fail. Build and git diff --check passed. Full compiled suite is running separately before PR submission. Fixes #188. Co-Authored-By: GPT-6 Astra --- CHANGELOG.md | 4 + src/framework.ts | 5 +- test/route-failure-targeting.test.ts | 141 +++++++++++++++++++++++++++ 3 files changed, 149 insertions(+), 1 deletion(-) create mode 100644 test/route-failure-targeting.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b9a30b4..3bf98d34 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 +### Fixed + +- Speech-route failure notices now target the speaking agent, including conversation forks, and respect that agent's mid-turn deferral. Unregistered conversation IDs retain the primary-agent fallback; notices remain context-only and do not trigger inference (#188). + ## 0.19.0 — 2026-09-28 ### Added diff --git a/src/framework.ts b/src/framework.ts index 510b9183..b18147ac 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -12541,7 +12541,7 @@ export class AgentFramework { // addMessage() alone does not request inference, so this never wakes // her (matching the `discord-send-failed-skip` gate intent: context // yes, wake no). - onRouteFailure: ({ channelId, reason, textLen }) => { + onRouteFailure: ({ conversationId, channelId, reason, textLen }) => { try { // Render a human-readable channel name when we can — a bare // snowflake in the marker is unresolvable for the agent (the @@ -12563,6 +12563,9 @@ export class AgentFramework { text: `[discord-send-failed] Your previous reply (${textLen} chars) could not be delivered to ${where} (${reason}). It was saved to your archive but the human did not receive it.`, }], { system: true, kind: 'discord-send-failed', channelId: channelId ?? '', reason }, + // Keep the notice with its speaker, including mid-turn deferral. + // Unknown conversation IDs retain the primary-agent fallback. + this.agents.has(conversationId) ? { forAgent: conversationId } : undefined, ); } catch (err) { console.error('onRouteFailure: failed to record send-failure marker:', err); diff --git a/test/route-failure-targeting.test.ts b/test/route-failure-targeting.test.ts new file mode 100644 index 00000000..e17e55a2 --- /dev/null +++ b/test/route-failure-targeting.test.ts @@ -0,0 +1,141 @@ +import { describe, it, beforeEach, afterEach } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { AgentFramework } from '../src/index.js'; +import type { ProcessEvent } from '../src/index.js'; +import type { ChannelRegistry } from '../src/mcpl/channel-registry.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; + +function internals(framework: AgentFramework) { + return framework as unknown as { + initializeMcpl(servers: []): Promise; + channelRegistry: ChannelRegistry; + activeTurnTokens: Map; + deferredMessages: Array<{ forAgent?: string; metadata?: Record }>; + pendingRequests: unknown[]; + }; +} + +describe('speech route failure notice targeting', () => { + let tempDir: string; + let framework: AgentFramework; + let membrane: MockMembrane; + + beforeEach(async () => { + tempDir = mkdtempSync(join(tmpdir(), 'route-failure-targeting-')); + membrane = new MockMembrane(); + framework = await AgentFramework.create({ + storePath: join(tempDir, 'test.chronicle'), + membrane: membrane.asMembrane(), + agents: [ + { name: 'primary', model: 'test-model', systemPrompt: 'Primary.', proseRouting: 'locus' }, + { name: 'secondary', model: 'test-model', systemPrompt: 'Secondary.' }, + ], + modules: [], + conversations: { templateAgent: 'primary' }, + }); + // Install the real framework callback and registry without a live connector. + await internals(framework).initializeMcpl([]); + }); + + afterEach(async () => { + internals(framework).activeTurnTokens.clear(); + await framework.stop(); + rmSync(tempDir, { recursive: true, force: true }); + }); + + it('records a secondary agent failure through that agent\'s context manager', async (t) => { + // Ordinary residents share a Chronicle message slot. Check the receiving + // manager here; the real conversation fork below checks window isolation. + const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); + const secondary = t.mock.method(framework.getAgent('secondary')!.getContextManager(), 'addMessage'); + const result = await internals(framework).channelRegistry.routeSpeech('secondary', 'hello', 'missing-channel'); + + assert.equal(result, null); + assert.equal(primary.mock.callCount(), 0); + assert.equal(secondary.mock.callCount(), 1); + const [participant, content, metadata] = secondary.mock.calls[0]!.arguments; + assert.equal(participant, 'user'); + assert.match(JSON.stringify(content), /\[discord-send-failed\].*5 chars.*missing-channel/); + assert.equal(metadata?.system, true); + assert.equal(metadata?.kind, 'discord-send-failed'); + assert.equal(metadata?.channelId, 'missing-channel'); + assert.match(String(metadata?.reason), /no registered channel/); + assert.equal(internals(framework).pendingRequests.length, 0, 'a failure notice must not request inference'); + assert.equal(membrane.calls.length, 0); + }); + + it('keeps primary failures in the primary context manager', async (t) => { + const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); + const secondary = t.mock.method(framework.getAgent('secondary')!.getContextManager(), 'addMessage'); + + await internals(framework).channelRegistry.routeSpeech('primary', 'hello', null); + + assert.equal(primary.mock.callCount(), 1); + assert.equal(secondary.mock.callCount(), 0); + assert.equal(primary.mock.calls[0]!.arguments[2]?.kind, 'discord-send-failed'); + assert.equal(internals(framework).pendingRequests.length, 0); + }); + + it('falls back to the primary for an unregistered conversation', async (t) => { + const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); + const secondary = t.mock.method(framework.getAgent('secondary')!.getContextManager(), 'addMessage'); + + await internals(framework).channelRegistry.routeSpeech('unknown-conversation', 'hello', null); + + assert.equal(primary.mock.callCount(), 1); + assert.equal(secondary.mock.callCount(), 0); + assert.equal(primary.mock.calls[0]!.arguments[2]?.kind, 'discord-send-failed'); + assert.equal(internals(framework).pendingRequests.length, 0); + }); + + it('defers a failure notice against the speaking agent\'s live turn', async (t) => { + const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); + const secondary = t.mock.method(framework.getAgent('secondary')!.getContextManager(), 'addMessage'); + const state = internals(framework); + state.activeTurnTokens.set('secondary', 1); + + await state.channelRegistry.routeSpeech('secondary', 'hello', null); + + assert.equal(primary.mock.callCount(), 0); + assert.equal(secondary.mock.callCount(), 0, 'the notice waits for the owner\'s safe boundary'); + assert.equal(state.deferredMessages.length, 1); + assert.equal(state.deferredMessages[0]!.forAgent, 'secondary'); + assert.equal(state.deferredMessages[0]!.metadata?.kind, 'discord-send-failed'); + assert.equal(state.pendingRequests.length, 0); + }); + + it('delivers a failed fork reply notice to the fork after its turn, without waking it again', async () => { + const reply = 'Reply from the fork.'; + membrane.pushResponse(createMockResponse([{ type: 'text', text: reply }])); + framework.pushEvent({ + type: 'mcpl:channel-incoming', + serverId: 'srv', + channelId: 'missing-channel', + messageId: 'incoming-1', + author: { id: 'user-1', name: 'User' }, + content: [{ type: 'text', text: 'Hello fork.' }], + timestamp: new Date().toISOString(), + metadata: { channel_type: 'im' }, + triggerInference: true, + } as unknown as ProcessEvent); + await framework.runUntilIdle(); + + const binding = framework.getConversationRouter()!.getBinding('missing-channel'); + assert.ok(binding, 'the incoming DM must spawn a real conversation fork'); + const fork = framework.getAgent(binding.agentName)!; + const { messages: forkMessages } = await fork.getContextManager().compile(); + const { messages: primaryMessages } = await framework.getAgent('primary')!.getContextManager().compile(); + const notices = forkMessages.filter((m) => JSON.stringify(m.content).includes('[discord-send-failed]')); + assert.equal(notices.length, 1, 'the fork must see its own failure notice exactly once'); + assert.ok(!primaryMessages.some((m) => JSON.stringify(m.content).includes('[discord-send-failed]'))); + const replyIndex = forkMessages.findIndex((m) => m.participant === binding.agentName && JSON.stringify(m.content).includes(reply)); + assert.ok(replyIndex >= 0, 'the attempted reply is still archived'); + assert.ok(forkMessages.indexOf(notices[0]!) > replyIndex, 'failure notice follows the attempted reply'); + assert.equal(internals(framework).deferredMessages.length, 0, 'the completed turn flushes its notice'); + assert.equal(internals(framework).pendingRequests.length, 0); + assert.equal(membrane.calls.length, 1, 'failure notice is context only, not another wake'); + }); +}); From 0b417696aa31e92fe315404899fa2e77a57ca0b1 Mon Sep 17 00:00:00 2001 From: Iris-308 Date: Fri, 2 Oct 2026 02:04:30 -0400 Subject: [PATCH 2/7] fix(routing): retain closure failure notices before disposing forks Greptile's review of PR #204 identified a real teardown ordering bug: a failed TTL-closure publish queues its owner-targeted notice, but driveStream disposed the fork before the deferred flush. The subsequent addMessage call dropped the notice because its target was already unregistered. Release the turn token and flush owner-scoped deferred writes before disposing a terminal closure fork. Scope the pending-tool check to that agent so another agent's tool cycle cannot prevent its final notice from landing. Disposal now retains an agent with an active turn/stream/tool cycle or pending/unacknowledged context writes. This also protects the fallback reaper: a quiesced closure stays registered until its deferred notice is flushed, then the reaper disposes it safely. Add three real TTL-closure regressions with a server-reported delivered:false response: ordinary closure, another agent's tool cycle, and a quiesced closure plus fallback reaping after flush. Each opens a fresh ContextManager on the archived fork namespace and checks final-reply/notice order, exactly one notice, primary isolation, eventual disposal, and no additional inference. All three fail on PR head 17034b1 (5 pass / 3 fail in the extended file). Corrected targeting, per-agent delivery, deferred-wake, and host-quiesce tests pass 47/47 without force-exit; build and diff check pass. Full bounded compiled suite is running before publication. Addresses review https://github.com/anima-research/agent-framework/pull/204#discussion_r4163104814. Co-Authored-By: GPT-6 Astra --- CHANGELOG.md | 2 +- src/framework.ts | 27 ++++++--- test/route-failure-targeting.test.ts | 86 ++++++++++++++++++++++++++++ 3 files changed, 107 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3bf98d34..15576265 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,7 @@ Releases up to and including 0.7.3 predate this file; for their contents see ### Fixed -- Speech-route failure notices now target the speaking agent, including conversation forks, and respect that agent's mid-turn deferral. Unregistered conversation IDs retain the primary-agent fallback; notices remain context-only and do not trigger inference (#188). +- Speech-route failure notices now target the speaking agent, including conversation forks, and respect that agent's mid-turn deferral. Unregistered conversation IDs retain the primary-agent fallback; notices remain context-only and do not trigger inference. A fork's final closure notice is flushed before disposal, and deferred writes retain their target through quiesce until they can be stored (#188). ## 0.19.0 — 2026-09-28 diff --git a/src/framework.ts b/src/framework.ts index b18147ac..c761b2cb 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -7169,6 +7169,18 @@ export class AgentFramework { * forever. The fork's context stays in Chronicle for investigation. */ private disposeConversationAgent(agentName: string): void { + // A closure can finish while writes are still deferred (e.g. quiesce), + // or while a successor owns the turn token. Keep the target registered + // until those writes land; the fallback reaper can dispose it afterward. + const belongsToAgent = (msg: { forAgent?: string }) => + (msg.forAgent ?? this.primaryAgentName) === agentName; + if ( + this.activeTurnTokens.has(agentName) || + this.activeStreams.has(agentName) || + this.pendingAssistantBlocks.has(agentName) || + this.deferredMessages.some(belongsToAgent) || + this.unackedDeferredWrites.some(belongsToAgent) + ) return; this.closingConversationAgents.delete(agentName); const channelId = this.conversationAgentHomes.get(agentName); const agent = this.agents.get(agentName); @@ -10366,12 +10378,6 @@ export class AgentFramework { this.pendingAssistantBlocks.delete(agent.name); } - // A conversation fork whose TTL closure turn just finished is done for - // good — dispose it so the agent map doesn't grow monotonically. - if (ownsPhysicalStream && this.closingConversationAgents.has(agent.name)) { - this.disposeConversationAgent(agent.name); - } - // The turn is torn down — release the turn-alive marker BEFORE the // deferred flush below, so the flush appends at the settled tail. // Token-matched: if a successor turn already owns the agent (endTurn @@ -10388,7 +10394,7 @@ export class AgentFramework { // pending). Only THIS agent's messages: other targets' entries wait // for their own boundaries (re-adding via addMessage re-defers if the // target has meanwhile started a turn). - if (frameReachedTerminal && this.deferredMessages.length > 0 && this.pendingAssistantBlocks.size === 0) { + if (frameReachedTerminal && this.deferredMessages.length > 0 && !this.pendingAssistantBlocks.has(agent.name)) { const deferred = this.drainDeferredFor(agent.name); for (const msg of deferred) { this.addMessage(msg.participant, msg.content, msg.metadata, { @@ -10398,6 +10404,13 @@ export class AgentFramework { } this.ackDeferredWrites(); } + + // Flush the closure turn's notices while its owner is still registered. + // If quiesce or a successor re-deferred them, disposal waits for the + // fallback reaper after their eventual safe write boundary. + if (frameReachedTerminal && ownsPhysicalStream && this.closingConversationAgents.has(agent.name)) { + this.disposeConversationAgent(agent.name); + } } } diff --git a/test/route-failure-targeting.test.ts b/test/route-failure-targeting.test.ts index e17e55a2..2b8890de 100644 --- a/test/route-failure-targeting.test.ts +++ b/test/route-failure-targeting.test.ts @@ -4,6 +4,9 @@ import { mkdtempSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { AgentFramework } from '../src/index.js'; +import { ContextManager, PassthroughStrategy } from '@animalabs/context-manager'; +import { CapabilityGrant } from '../src/mcpl/capability-grant.js'; +import type { McplServerRegistry } from '../src/mcpl/server-registry.js'; import type { ProcessEvent } from '../src/index.js'; import type { ChannelRegistry } from '../src/mcpl/channel-registry.js'; import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; @@ -15,6 +18,12 @@ function internals(framework: AgentFramework) { activeTurnTokens: Map; deferredMessages: Array<{ forAgent?: string; metadata?: Record }>; pendingRequests: unknown[]; + pendingAssistantBlocks: Map; + mcplServerRegistry: McplServerRegistry; + lastConversationSweep: number; + sweepExpiredConversations(): void; + quiesced: boolean; + flushDeferredWrites(label: string): Promise; }; } @@ -42,6 +51,8 @@ describe('speech route failure notice targeting', () => { afterEach(async () => { internals(framework).activeTurnTokens.clear(); + internals(framework).pendingAssistantBlocks.clear(); + internals(framework).quiesced = false; await framework.stop(); rmSync(tempDir, { recursive: true, force: true }); }); @@ -107,6 +118,81 @@ describe('speech route failure notice targeting', () => { assert.equal(state.pendingRequests.length, 0); }); + for (const condition of ['ordinary', 'other-tool-cycle', 'quiesced'] as const) { + it('retains a failed TTL-closure reply notice before disposal: ' + condition, async (t) => { + const state = internals(framework); + const channelId = 'closure-channel'; + let failPublishing = false; + const publisher = { + grant: new CapabilityGrant(new Set(['channels.publish']), []), + sendChannelsTyping() {}, + async sendChannelsPublish() { + if (failPublishing && condition === 'quiesced') state.quiesced = true; + return { delivered: !failPublishing }; + }, + }; + t.mock.method(state.mcplServerRegistry, 'getServer', () => publisher as never); + (state.channelRegistry as unknown as { channels: Map }).channels.set('srv:' + channelId, { + serverId: 'srv', descriptor: { id: channelId, type: 'srv', label: 'closure test' }, open: true, + }); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Initial reply.' }])); + framework.pushEvent({ + type: 'mcpl:channel-incoming', serverId: 'srv', channelId, messageId: 'closure-incoming', + author: { id: 'user', name: 'User' }, + content: [{ type: 'text', text: 'Please help.' }], + timestamp: new Date().toISOString(), + metadata: { mentioned: true }, + triggerInference: true, + } as unknown as ProcessEvent); + await framework.runUntilIdle(); + const binding = framework.getConversationRouter()!.getBinding(channelId)!; + assert.ok(binding); + const fork = framework.getAgent(binding.agentName)!; + + failPublishing = true; + const finalReply = 'Final reply that the server cannot deliver.'; + membrane.pushResponse(createMockResponse([{ type: 'text', text: finalReply }])); + if (condition === 'other-tool-cycle') state.pendingAssistantBlocks.set('secondary', []); + binding.lastActivity = 0; + state.lastConversationSweep = 0; + state.sweepExpiredConversations(); + await framework.runUntilIdle(); + state.pendingAssistantBlocks.clear(); + + if (condition === 'quiesced') { + assert.ok(framework.getAgent(fork.name) === fork, 'a deferred write keeps its owner registered while quiesced'); + assert.equal(state.deferredMessages.length, 1); + state.lastConversationSweep = 0; + state.sweepExpiredConversations(); + assert.ok(framework.getAgent(fork.name) === fork, 'the fallback reaper must also retain the pending write target'); + state.quiesced = false; + await state.flushDeferredWrites('closure test resume'); + state.lastConversationSweep = 0; + state.sweepExpiredConversations(); + } + + assert.equal(framework.getAgent(fork.name), null, 'the settled fork is eventually disposed'); + assert.equal(state.deferredMessages.length, 0); + // Open a fresh manager over the retained fork namespace, rather than + // relying only on the disposed Agent's in-memory context manager. + const archive = await ContextManager.open({ + store: framework.getStore(), namespace: 'conversations/' + fork.name, + isolate: true, strategy: new PassthroughStrategy(), + }); + const { messages } = await archive.compile(); + const noticeIndex = messages.findIndex((m) => JSON.stringify(m.content).includes('[discord-send-failed]')); + const replyIndex = messages.findIndex((m) => m.participant === fork.name && JSON.stringify(m.content).includes(finalReply)); + assert.ok(replyIndex >= 0, 'the failed closure reply remains archived'); + assert.ok(noticeIndex > replyIndex, 'the archived failure notice follows the closure reply'); + assert.equal(messages.filter((m) => JSON.stringify(m.content).includes('[discord-send-failed]')).length, 1); + assert.match(JSON.stringify(messages[noticeIndex]!.content), /delivered:false/); + const primary = await framework.getAgent('primary')!.getContextManager().compile(); + assert.ok(!primary.messages.some((m) => JSON.stringify(m.content).includes('[discord-send-failed]'))); + assert.equal(membrane.calls.length, 2, 'only the initial and closure turns ran'); + assert.equal(state.pendingRequests.length, 0, 'a failed closure reply does not wake another turn'); + }); + } + it('delivers a failed fork reply notice to the fork after its turn, without waking it again', async () => { const reply = 'Reply from the fork.'; membrane.pushResponse(createMockResponse([{ type: 'text', text: reply }])); From 28dbcbac72e64396aa33b5203a1e000854f60b0b Mon Sep 17 00:00:00 2001 From: Ada-1017 Date: Fri, 9 Oct 2026 21:13:43 -0400 Subject: [PATCH 3/7] chore(changelog): move #188's entry into a changelog.d fragment The branch edited CHANGELOG.md's Unreleased section directly, which conflicts with main (02adf27) now that main's own entries live in changelog.d/ (CONTRIBUTING.md#changelog). Same text, moved verbatim into changelog.d/fix-188-route-failure-owner.fixed.md; CHANGELOG.md is back to the merge base's version, so the branch no longer touches it. --- CHANGELOG.md | 4 ---- changelog.d/fix-188-route-failure-owner.fixed.md | 1 + 2 files changed, 1 insertion(+), 4 deletions(-) create mode 100644 changelog.d/fix-188-route-failure-owner.fixed.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 15576265..6b9a30b4 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 -### Fixed - -- Speech-route failure notices now target the speaking agent, including conversation forks, and respect that agent's mid-turn deferral. Unregistered conversation IDs retain the primary-agent fallback; notices remain context-only and do not trigger inference. A fork's final closure notice is flushed before disposal, and deferred writes retain their target through quiesce until they can be stored (#188). - ## 0.19.0 — 2026-09-28 ### Added diff --git a/changelog.d/fix-188-route-failure-owner.fixed.md b/changelog.d/fix-188-route-failure-owner.fixed.md new file mode 100644 index 00000000..4cd786d7 --- /dev/null +++ b/changelog.d/fix-188-route-failure-owner.fixed.md @@ -0,0 +1 @@ +- Speech-route failure notices now target the speaking agent, including conversation forks, and respect that agent's mid-turn deferral. Unregistered conversation IDs retain the primary-agent fallback; notices remain context-only and do not trigger inference. A fork's final closure notice is flushed before disposal, and deferred writes retain their target through quiesce until they can be stored (#188). From 5fed304403bb8b2c835f2168b5c4da5009b62848 Mon Sep 17 00:00:00 2001 From: Ada-1017 Date: Fri, 9 Oct 2026 21:28:39 -0400 Subject: [PATCH 4/7] fix(routing): address an agent's notices by the slot it reads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Skill pass on #204 (preparing-upstream-pull-requests), reading main 02adf27. Three findings changed the design; probes in ~/.local/state/ada-1017/tmp/chk/pr204/route-failure-probes.test.ts ran on the trial merge and on main. 1. Residents share ONE message slot (sharedSlotAgents; forks and ephemeral runs are isolated). Addressing a resident (forAgent) only changed which context manager wrote the same slot, and swapped addMessage's guard from "primary's turn + any agent's tool cycle" to "the resident's own turn + cycle". Probes: with the primary mid-turn, or mid tool cycle, a secondary's notice was written at once (main defers it); with a secondary mid cycle, the primary's own notice was written at once (main defers it). The speaker already reads the slot, so residents stay on the default path. 2. Ephemeral runs (connectome-host subagents inherit the parent's proseRouting, so their prose routes) are isolated speakers too. On the old head their last notice was deferred against their live turn, then dropped by the teardown flush after runEphemeralToCompletion's finally had deregistered them ("not a registered agent — message dropped"); main gave it to the primary. 3. The same "a notice about X lands in the primary's window" defect sits in bounceProse (notice to the primary, wake and clipboard to the speaker) and injectScriptWake (envelope to the primary, wake to the script's agent). recordChannelAutoOpenNotice is left: a channel's open state is host-wide, and later traffic can reach the primary. Changes: - noticeTargetFor(name): forAgent unless the agent is a shared-slot resident; used by onRouteFailure, bounceProse, injectScriptWake. An unregistered name is addressed too, so addMessage drops it, logged, instead of giving the primary a notice about something it didn't do (was: primary fallback). - landQueuedWritesBeforeRelease(agent): when a fork is disposed or an ephemeral run deregistered, write what is still queued for it into its own context manager (Chronicle keeps it), then ack. Skipped while quiesced (a write could land mid-surgery); the next flush drops them, logged. Replaces 0b41769's mechanism (disposal guard with five conditions, disposal moved after the flush, per-agent flush gate), which covered forks only and whose per-agent teardown gate let a resident's forAgent writes (EventGate) land while another resident sharing the slot was mid tool cycle. Fork disposal is back at main's position. - Context: main c49f394 deprecated per-channel conversation routing (forks); ephemeral runs are the isolated speakers that stay. Tests (15, rewritten): residents x5 (default path and its guard), fork reply, fork own-turn deferral, unregistered drop, ephemeral last notice, ephemeral bounce, fork script wake, closure x4 (ordinary, other agent's tool cycle, quiesced: dropped and logged, refused write: logged and still disposed). Local Node 24.19: 15/15. --- .../fix-188-route-failure-owner.fixed.md | 6 +- src/framework.ts | 78 +++- test/route-failure-targeting.test.ts | 421 +++++++++++------- 3 files changed, 328 insertions(+), 177 deletions(-) diff --git a/changelog.d/fix-188-route-failure-owner.fixed.md b/changelog.d/fix-188-route-failure-owner.fixed.md index 4cd786d7..c2376aed 100644 --- a/changelog.d/fix-188-route-failure-owner.fixed.md +++ b/changelog.d/fix-188-route-failure-owner.fixed.md @@ -1 +1,5 @@ -- Speech-route failure notices now target the speaking agent, including conversation forks, and respect that agent's mid-turn deferral. Unregistered conversation IDs retain the primary-agent fallback; notices remain context-only and do not trigger inference. A fork's final closure notice is flushed before disposal, and deferred writes retain their target through quiesce until they can be stored (#188). +- Notices about an agent's own action now reach the window that agent reads (#188): + - A conversation fork's or an ephemeral run's failed-reply notice (`[discord-send-failed]`), bounced-prose notice and background-script wake land in its own window, deferred against its own turn, instead of the primary's. In connectome-host this moves a subagent's notices from its parent to the subagent. + - Residents share one message slot, so a resident's notices stay on the default path, unchanged. + - A notice for a speaker that is no longer registered is dropped and logged, rather than given to the primary. + - Writes still queued for a fork or an ephemeral run when it is released land in its own window first, which Chronicle keeps. While the host is quiesced they are dropped instead, logged. diff --git a/src/framework.ts b/src/framework.ts index c761b2cb..27e6b385 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -4137,6 +4137,7 @@ export class AgentFramework { // The late stream finalizer intentionally refuses name-keyed cleanup once // deregistered, so disposal must release EventGate liveness itself. this.eventGate?.onInferenceEnded(agent.name); + this.landQueuedWritesBeforeRelease(agent); this.agents.delete(agent.name); } this.toolImageLedgers.delete(agent.name); @@ -7169,21 +7170,10 @@ export class AgentFramework { * forever. The fork's context stays in Chronicle for investigation. */ private disposeConversationAgent(agentName: string): void { - // A closure can finish while writes are still deferred (e.g. quiesce), - // or while a successor owns the turn token. Keep the target registered - // until those writes land; the fallback reaper can dispose it afterward. - const belongsToAgent = (msg: { forAgent?: string }) => - (msg.forAgent ?? this.primaryAgentName) === agentName; - if ( - this.activeTurnTokens.has(agentName) || - this.activeStreams.has(agentName) || - this.pendingAssistantBlocks.has(agentName) || - this.deferredMessages.some(belongsToAgent) || - this.unackedDeferredWrites.some(belongsToAgent) - ) return; this.closingConversationAgents.delete(agentName); const channelId = this.conversationAgentHomes.get(agentName); const agent = this.agents.get(agentName); + if (agent) this.landQueuedWritesBeforeRelease(agent); this.agents.delete(agentName); this.toolImageLedgers.delete(agentName); this.agentConfigs.delete(agentName); @@ -8568,6 +8558,7 @@ export class AgentFramework { 'user', [{ type: 'text', text: notice }], { system: true, kind: 'prose-bounce' }, + this.noticeTargetFor(name), ); if (id) this.emitTrace({ type: 'message:added', messageId: id, source: 'prose-bounce' }); else this.emitTrace({ type: 'message:added', messageId: 'deferred', source: 'prose-bounce' }); @@ -10378,6 +10369,12 @@ export class AgentFramework { this.pendingAssistantBlocks.delete(agent.name); } + // A conversation fork whose TTL closure turn just finished is done for + // good — dispose it so the agent map doesn't grow monotonically. + if (ownsPhysicalStream && this.closingConversationAgents.has(agent.name)) { + this.disposeConversationAgent(agent.name); + } + // The turn is torn down — release the turn-alive marker BEFORE the // deferred flush below, so the flush appends at the settled tail. // Token-matched: if a successor turn already owns the agent (endTurn @@ -10394,7 +10391,7 @@ export class AgentFramework { // pending). Only THIS agent's messages: other targets' entries wait // for their own boundaries (re-adding via addMessage re-defers if the // target has meanwhile started a turn). - if (frameReachedTerminal && this.deferredMessages.length > 0 && !this.pendingAssistantBlocks.has(agent.name)) { + if (frameReachedTerminal && this.deferredMessages.length > 0 && this.pendingAssistantBlocks.size === 0) { const deferred = this.drainDeferredFor(agent.name); for (const msg of deferred) { this.addMessage(msg.participant, msg.content, msg.metadata, { @@ -10404,13 +10401,6 @@ export class AgentFramework { } this.ackDeferredWrites(); } - - // Flush the closure turn's notices while its owner is still registered. - // If quiesce or a successor re-deferred them, disposal waits for the - // fallback reaper after their eventual safe write boundary. - if (frameReachedTerminal && ownsPhysicalStream && this.closingConversationAgents.has(agent.name)) { - this.disposeConversationAgent(agent.name); - } } } @@ -11046,7 +11036,7 @@ export class AgentFramework { source: 'background-script', scriptId: record.id, tags: ['script:wake'], - }); + }, this.noticeTargetFor(record.agentName)); this.pendingRequests.push({ agentName: record.agentName, reason: 'script:wake', @@ -11964,6 +11954,48 @@ export class AgentFramework { return mine; } + /** + * Where a notice about an agent's own action belongs: its failed reply, its + * bounced prose, its background script's wake. A conversation fork or an + * ephemeral run reads only its own isolated slot, so the notice is addressed + * to it. Residents share one message slot, which the default path already + * writes, under the guard that waits for any agent's tool cycle and the + * primary's turn; addressing a resident would narrow that guard to its own + * turn and cycle. A name no longer registered (a fork or run released before + * its failure was known) is addressed too, so addMessage drops the notice, + * logged, rather than give it to the primary, which did nothing. + */ + private noticeTargetFor(agentName: string): { forAgent: string } | undefined { + const agent = this.agents.get(agentName); + return agent && this.sharedSlotAgents.has(agent) ? undefined : { forAgent: agentName }; + } + + /** + * A conversation fork or an ephemeral run is about to be deregistered. + * Writes still queued for its window would wait for a boundary it never + * reaches, and the flush that finds it gone would drop them; land them in + * its own window now, which stays in Chronicle for investigation. Its + * stream is over or cancelled, so nothing it compiles can diverge from + * them. While quiesced a write could land mid-surgery, so they stay + * queued, and the next flush to find their target gone drops them, logged. + */ + private landQueuedWritesBeforeRelease(agent: Agent): void { + if (this.quiesced) return; + const cm = agent.getContextManager(); + for (const msg of this.drainDeferredFor(agent.name)) { + try { + cm.addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id)); + } catch (err) { + console.error( + `[deferred-flush] ${agent.name}: failed to store a queued write before release ` + + `(participant=${msg.participant}):`, + err, + ); + } + } + this.ackDeferredWrites(); + } + private editMessage(id: MessageId, content: ContentBlock[]): void { const agent = this.primaryAgentName ? this.agents.get(this.primaryAgentName) @@ -12576,9 +12608,7 @@ export class AgentFramework { text: `[discord-send-failed] Your previous reply (${textLen} chars) could not be delivered to ${where} (${reason}). It was saved to your archive but the human did not receive it.`, }], { system: true, kind: 'discord-send-failed', channelId: channelId ?? '', reason }, - // Keep the notice with its speaker, including mid-turn deferral. - // Unknown conversation IDs retain the primary-agent fallback. - this.agents.has(conversationId) ? { forAgent: conversationId } : undefined, + this.noticeTargetFor(conversationId), ); } catch (err) { console.error('onRouteFailure: failed to record send-failure marker:', err); diff --git a/test/route-failure-targeting.test.ts b/test/route-failure-targeting.test.ts index 2b8890de..351dd411 100644 --- a/test/route-failure-targeting.test.ts +++ b/test/route-failure-targeting.test.ts @@ -23,11 +23,19 @@ function internals(framework: AgentFramework) { lastConversationSweep: number; sweepExpiredConversations(): void; quiesced: boolean; + deferredWritesPersisted: boolean; + unackedDeferredWrites: unknown[]; flushDeferredWrites(label: string): Promise; + injectScriptWake(record: { id: string; agentName: string }, envelope: string): void; }; } -describe('speech route failure notice targeting', () => { +const SEND_FAILED = '[discord-send-failed]'; +const BOUNCED = 'was not delivered —'; +const count = (messages: Array<{ content: unknown }>, needle: string) => + messages.filter((m) => JSON.stringify(m.content).includes(needle)).length; + +describe('notices about an agent\'s own action', () => { let tempDir: string; let framework: AgentFramework; let membrane: MockMembrane; @@ -57,144 +65,8 @@ describe('speech route failure notice targeting', () => { rmSync(tempDir, { recursive: true, force: true }); }); - it('records a secondary agent failure through that agent\'s context manager', async (t) => { - // Ordinary residents share a Chronicle message slot. Check the receiving - // manager here; the real conversation fork below checks window isolation. - const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); - const secondary = t.mock.method(framework.getAgent('secondary')!.getContextManager(), 'addMessage'); - const result = await internals(framework).channelRegistry.routeSpeech('secondary', 'hello', 'missing-channel'); - - assert.equal(result, null); - assert.equal(primary.mock.callCount(), 0); - assert.equal(secondary.mock.callCount(), 1); - const [participant, content, metadata] = secondary.mock.calls[0]!.arguments; - assert.equal(participant, 'user'); - assert.match(JSON.stringify(content), /\[discord-send-failed\].*5 chars.*missing-channel/); - assert.equal(metadata?.system, true); - assert.equal(metadata?.kind, 'discord-send-failed'); - assert.equal(metadata?.channelId, 'missing-channel'); - assert.match(String(metadata?.reason), /no registered channel/); - assert.equal(internals(framework).pendingRequests.length, 0, 'a failure notice must not request inference'); - assert.equal(membrane.calls.length, 0); - }); - - it('keeps primary failures in the primary context manager', async (t) => { - const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); - const secondary = t.mock.method(framework.getAgent('secondary')!.getContextManager(), 'addMessage'); - - await internals(framework).channelRegistry.routeSpeech('primary', 'hello', null); - - assert.equal(primary.mock.callCount(), 1); - assert.equal(secondary.mock.callCount(), 0); - assert.equal(primary.mock.calls[0]!.arguments[2]?.kind, 'discord-send-failed'); - assert.equal(internals(framework).pendingRequests.length, 0); - }); - - it('falls back to the primary for an unregistered conversation', async (t) => { - const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); - const secondary = t.mock.method(framework.getAgent('secondary')!.getContextManager(), 'addMessage'); - - await internals(framework).channelRegistry.routeSpeech('unknown-conversation', 'hello', null); - - assert.equal(primary.mock.callCount(), 1); - assert.equal(secondary.mock.callCount(), 0); - assert.equal(primary.mock.calls[0]!.arguments[2]?.kind, 'discord-send-failed'); - assert.equal(internals(framework).pendingRequests.length, 0); - }); - - it('defers a failure notice against the speaking agent\'s live turn', async (t) => { - const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); - const secondary = t.mock.method(framework.getAgent('secondary')!.getContextManager(), 'addMessage'); - const state = internals(framework); - state.activeTurnTokens.set('secondary', 1); - - await state.channelRegistry.routeSpeech('secondary', 'hello', null); - - assert.equal(primary.mock.callCount(), 0); - assert.equal(secondary.mock.callCount(), 0, 'the notice waits for the owner\'s safe boundary'); - assert.equal(state.deferredMessages.length, 1); - assert.equal(state.deferredMessages[0]!.forAgent, 'secondary'); - assert.equal(state.deferredMessages[0]!.metadata?.kind, 'discord-send-failed'); - assert.equal(state.pendingRequests.length, 0); - }); - - for (const condition of ['ordinary', 'other-tool-cycle', 'quiesced'] as const) { - it('retains a failed TTL-closure reply notice before disposal: ' + condition, async (t) => { - const state = internals(framework); - const channelId = 'closure-channel'; - let failPublishing = false; - const publisher = { - grant: new CapabilityGrant(new Set(['channels.publish']), []), - sendChannelsTyping() {}, - async sendChannelsPublish() { - if (failPublishing && condition === 'quiesced') state.quiesced = true; - return { delivered: !failPublishing }; - }, - }; - t.mock.method(state.mcplServerRegistry, 'getServer', () => publisher as never); - (state.channelRegistry as unknown as { channels: Map }).channels.set('srv:' + channelId, { - serverId: 'srv', descriptor: { id: channelId, type: 'srv', label: 'closure test' }, open: true, - }); - membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Initial reply.' }])); - framework.pushEvent({ - type: 'mcpl:channel-incoming', serverId: 'srv', channelId, messageId: 'closure-incoming', - author: { id: 'user', name: 'User' }, - content: [{ type: 'text', text: 'Please help.' }], - timestamp: new Date().toISOString(), - metadata: { mentioned: true }, - triggerInference: true, - } as unknown as ProcessEvent); - await framework.runUntilIdle(); - const binding = framework.getConversationRouter()!.getBinding(channelId)!; - assert.ok(binding); - const fork = framework.getAgent(binding.agentName)!; - - failPublishing = true; - const finalReply = 'Final reply that the server cannot deliver.'; - membrane.pushResponse(createMockResponse([{ type: 'text', text: finalReply }])); - if (condition === 'other-tool-cycle') state.pendingAssistantBlocks.set('secondary', []); - binding.lastActivity = 0; - state.lastConversationSweep = 0; - state.sweepExpiredConversations(); - await framework.runUntilIdle(); - state.pendingAssistantBlocks.clear(); - - if (condition === 'quiesced') { - assert.ok(framework.getAgent(fork.name) === fork, 'a deferred write keeps its owner registered while quiesced'); - assert.equal(state.deferredMessages.length, 1); - state.lastConversationSweep = 0; - state.sweepExpiredConversations(); - assert.ok(framework.getAgent(fork.name) === fork, 'the fallback reaper must also retain the pending write target'); - state.quiesced = false; - await state.flushDeferredWrites('closure test resume'); - state.lastConversationSweep = 0; - state.sweepExpiredConversations(); - } - - assert.equal(framework.getAgent(fork.name), null, 'the settled fork is eventually disposed'); - assert.equal(state.deferredMessages.length, 0); - // Open a fresh manager over the retained fork namespace, rather than - // relying only on the disposed Agent's in-memory context manager. - const archive = await ContextManager.open({ - store: framework.getStore(), namespace: 'conversations/' + fork.name, - isolate: true, strategy: new PassthroughStrategy(), - }); - const { messages } = await archive.compile(); - const noticeIndex = messages.findIndex((m) => JSON.stringify(m.content).includes('[discord-send-failed]')); - const replyIndex = messages.findIndex((m) => m.participant === fork.name && JSON.stringify(m.content).includes(finalReply)); - assert.ok(replyIndex >= 0, 'the failed closure reply remains archived'); - assert.ok(noticeIndex > replyIndex, 'the archived failure notice follows the closure reply'); - assert.equal(messages.filter((m) => JSON.stringify(m.content).includes('[discord-send-failed]')).length, 1); - assert.match(JSON.stringify(messages[noticeIndex]!.content), /delivered:false/); - const primary = await framework.getAgent('primary')!.getContextManager().compile(); - assert.ok(!primary.messages.some((m) => JSON.stringify(m.content).includes('[discord-send-failed]'))); - assert.equal(membrane.calls.length, 2, 'only the initial and closure turns ran'); - assert.equal(state.pendingRequests.length, 0, 'a failed closure reply does not wake another turn'); - }); - } - - it('delivers a failed fork reply notice to the fork after its turn, without waking it again', async () => { - const reply = 'Reply from the fork.'; + /** A real conversation fork: an incoming DM binds one, and its reply fails (no such channel). */ + async function spawnFork(reply = 'Reply from the fork.') { membrane.pushResponse(createMockResponse([{ type: 'text', text: reply }])); framework.pushEvent({ type: 'mcpl:channel-incoming', @@ -208,20 +80,265 @@ describe('speech route failure notice targeting', () => { triggerInference: true, } as unknown as ProcessEvent); await framework.runUntilIdle(); - const binding = framework.getConversationRouter()!.getBinding('missing-channel'); assert.ok(binding, 'the incoming DM must spawn a real conversation fork'); - const fork = framework.getAgent(binding.agentName)!; - const { messages: forkMessages } = await fork.getContextManager().compile(); - const { messages: primaryMessages } = await framework.getAgent('primary')!.getContextManager().compile(); - const notices = forkMessages.filter((m) => JSON.stringify(m.content).includes('[discord-send-failed]')); - assert.equal(notices.length, 1, 'the fork must see its own failure notice exactly once'); - assert.ok(!primaryMessages.some((m) => JSON.stringify(m.content).includes('[discord-send-failed]'))); - const replyIndex = forkMessages.findIndex((m) => m.participant === binding.agentName && JSON.stringify(m.content).includes(reply)); - assert.ok(replyIndex >= 0, 'the attempted reply is still archived'); - assert.ok(forkMessages.indexOf(notices[0]!) > replyIndex, 'failure notice follows the attempted reply'); - assert.equal(internals(framework).deferredMessages.length, 0, 'the completed turn flushes its notice'); - assert.equal(internals(framework).pendingRequests.length, 0); - assert.equal(membrane.calls.length, 1, 'failure notice is context only, not another wake'); + return framework.getAgent(binding.agentName)!; + } + + async function runEphemeral(proseRouting: 'locus' | 'explicit', speech: string) { + const created = await framework.createEphemeralAgent({ + name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', proseRouting, + }); + created.contextManager.addMessage('user', [{ type: 'text', text: 'Run once.' }]); + membrane.pushResponse(createMockResponse([{ type: 'text', text: speech }])); + const run = framework.runEphemeralToCompletion(created.agent, created.contextManager); + framework.start(); + await run; + await framework.runUntilIdle(); + // The run is released; read what it left in Chronicle. + const archive = await ContextManager.open({ + store: framework.getStore(), namespace: 'subagent/worker', isolate: true, strategy: new PassthroughStrategy(), + }); + return (await archive.compile()).messages; + } + + describe('residents share one message slot', () => { + it('keeps a secondary resident\'s notice on the default path, where it reads it', async (t) => { + const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); + const secondary = t.mock.method(framework.getAgent('secondary')!.getContextManager(), 'addMessage'); + const result = await internals(framework).channelRegistry.routeSpeech('secondary', 'hello', 'missing-channel'); + + assert.equal(result, null); + assert.equal(primary.mock.callCount(), 1, 'the default path writes the shared slot'); + assert.equal(secondary.mock.callCount(), 0); + const [participant, content, metadata] = primary.mock.calls[0]!.arguments; + assert.equal(participant, 'user'); + assert.match(JSON.stringify(content), /\[discord-send-failed\].*5 chars.*missing-channel/); + assert.equal(metadata?.system, true); + assert.equal(metadata?.kind, 'discord-send-failed'); + assert.equal(metadata?.channelId, 'missing-channel'); + assert.match(String(metadata?.reason), /no registered channel/); + t.mock.restoreAll(); + const { messages } = await framework.getAgent('secondary')!.getContextManager().compile(); + assert.equal(count(messages, SEND_FAILED), 1, 'the speaking resident reads the shared slot'); + assert.equal(internals(framework).pendingRequests.length, 0, 'a failure notice must not request inference'); + assert.equal(membrane.calls.length, 0); + }); + + it('keeps primary failures in the primary context manager', async (t) => { + const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); + const secondary = t.mock.method(framework.getAgent('secondary')!.getContextManager(), 'addMessage'); + + await internals(framework).channelRegistry.routeSpeech('primary', 'hello', null); + + assert.equal(primary.mock.callCount(), 1); + assert.equal(secondary.mock.callCount(), 0); + assert.equal(primary.mock.calls[0]!.arguments[2]?.kind, 'discord-send-failed'); + assert.equal(internals(framework).pendingRequests.length, 0); + }); + + it('waits for the primary\'s live turn before writing a secondary resident\'s notice', async () => { + const state = internals(framework); + state.activeTurnTokens.set('primary', 1); + + await state.channelRegistry.routeSpeech('secondary', 'hello', null); + + assert.equal(state.deferredMessages.length, 1, 'the shared slot is mid-turn for the primary'); + assert.equal(state.deferredMessages[0]!.forAgent, undefined); + const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); + assert.equal(count(messages, SEND_FAILED), 0); + }); + + for (const [speaker, cycling] of [['secondary', 'primary'], ['primary', 'secondary']] as const) { + it(`waits for ${cycling}'s tool cycle before writing ${speaker}'s notice`, async () => { + const state = internals(framework); + state.pendingAssistantBlocks.set(cycling, []); + + await state.channelRegistry.routeSpeech(speaker, 'hello', null); + + assert.equal(state.deferredMessages.length, 1, 'a write now would split a tool_use from its result'); + const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); + assert.equal(count(messages, SEND_FAILED), 0); + }); + } + }); + + describe('a fork or an ephemeral run reads only its own window', () => { + it('delivers a failed fork reply notice to the fork after its turn, without waking it again', async () => { + const reply = 'Reply from the fork.'; + const fork = await spawnFork(reply); + const { messages: forkMessages } = await fork.getContextManager().compile(); + const { messages: primaryMessages } = await framework.getAgent('primary')!.getContextManager().compile(); + const notices = forkMessages.filter((m) => JSON.stringify(m.content).includes(SEND_FAILED)); + assert.equal(notices.length, 1, 'the fork must see its own failure notice exactly once'); + assert.equal(count(primaryMessages, SEND_FAILED), 0); + const replyIndex = forkMessages.findIndex((m) => m.participant === fork.name && JSON.stringify(m.content).includes(reply)); + assert.ok(replyIndex >= 0, 'the attempted reply is still archived'); + assert.ok(forkMessages.indexOf(notices[0]!) > replyIndex, 'failure notice follows the attempted reply'); + assert.equal(internals(framework).deferredMessages.length, 0, 'the completed turn flushes its notice'); + assert.equal(internals(framework).pendingRequests.length, 0); + assert.equal(membrane.calls.length, 1, 'failure notice is context only, not another wake'); + }); + + it('defers a fork\'s notice against the fork\'s own live turn', async (t) => { + const fork = await spawnFork(); + const state = internals(framework); + const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); + const forkCm = t.mock.method(fork.getContextManager(), 'addMessage'); + state.activeTurnTokens.set(fork.name, 99); + + await state.channelRegistry.routeSpeech(fork.name, 'hello', null); + + assert.equal(primary.mock.callCount(), 0); + assert.equal(forkCm.mock.callCount(), 0, 'the notice waits for the fork\'s safe boundary'); + assert.equal(state.deferredMessages.length, 1); + assert.equal(state.deferredMessages[0]!.forAgent, fork.name); + assert.equal(state.deferredMessages[0]!.metadata?.kind, 'discord-send-failed'); + assert.equal(state.pendingRequests.length, 0); + }); + + it('drops a notice for a speaker that is no longer registered, logged, rather than give it to the primary', async (t) => { + const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); + const secondary = t.mock.method(framework.getAgent('secondary')!.getContextManager(), 'addMessage'); + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + + await internals(framework).channelRegistry.routeSpeech('released-conversation', 'hello', null); + + assert.equal(primary.mock.callCount(), 0); + assert.equal(secondary.mock.callCount(), 0); + assert.ok(errors.some((e) => /"released-conversation" is not a registered agent — message dropped/.test(e))); + assert.equal(internals(framework).deferredMessages.length, 0); + assert.equal(internals(framework).pendingRequests.length, 0); + }); + + it('lands an ephemeral run\'s last failure notice in its own window before releasing it', async (t) => { + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + const speech = 'Worker speech with no channel to go to.'; + + const archived = await runEphemeral('locus', speech); + + const replyIndex = archived.findIndex((m) => m.participant === 'worker' && JSON.stringify(m.content).includes(speech)); + const noticeIndex = archived.findIndex((m) => JSON.stringify(m.content).includes(SEND_FAILED)); + assert.ok(replyIndex >= 0, 'the run\'s reply is archived'); + assert.ok(noticeIndex > replyIndex, 'its failure notice follows the reply in its own window'); + assert.equal(count(archived, SEND_FAILED), 1); + const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); + assert.equal(count(messages, SEND_FAILED), 0, 'the primary did not write that reply'); + assert.ok(!errors.some((e) => /not a registered agent/.test(e))); + assert.equal(internals(framework).deferredMessages.length, 0); + }); + + it('sends an ephemeral run\'s bounced prose notice to its own window, beside its clipboard', async () => { + const archived = await runEphemeral('explicit', 'Worker prose with no destination.'); + + assert.equal(count(archived, BOUNCED), 1, 'the speaker reads why its text did not go out'); + const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); + assert.equal(count(messages, BOUNCED), 0, 'the primary wrote no such text'); + assert.equal(internals(framework).deferredMessages.length, 0); + }); + + it('wakes a fork\'s background script into the fork\'s own window', async () => { + const fork = await spawnFork(); + const envelope = '[background script s-1] Your background script DIED 0m after start.'; + + internals(framework).injectScriptWake({ id: 's-1', agentName: fork.name }, envelope); + + const { messages: forkMessages } = await fork.getContextManager().compile(); + const { messages: primaryMessages } = await framework.getAgent('primary')!.getContextManager().compile(); + assert.equal(count(forkMessages, envelope), 1); + assert.equal(count(primaryMessages, envelope), 0); + }); + + for (const condition of ['ordinary', 'other-tool-cycle', 'quiesced', 'write-fails'] as const) { + it('handles a failed TTL-closure reply notice at disposal: ' + condition, async (t) => { + const state = internals(framework); + const channelId = 'closure-channel'; + let failPublishing = false; + const publisher = { + grant: new CapabilityGrant(new Set(['channels.publish']), []), + sendChannelsTyping() {}, + async sendChannelsPublish() { + if (failPublishing && condition === 'quiesced') state.quiesced = true; + return { delivered: !failPublishing }; + }, + }; + t.mock.method(state.mcplServerRegistry, 'getServer', () => publisher as never); + (state.channelRegistry as unknown as { channels: Map }).channels.set('srv:' + channelId, { + serverId: 'srv', descriptor: { id: channelId, type: 'srv', label: 'closure test' }, open: true, + }); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Initial reply.' }])); + framework.pushEvent({ + type: 'mcpl:channel-incoming', serverId: 'srv', channelId, messageId: 'closure-incoming', + author: { id: 'user', name: 'User' }, + content: [{ type: 'text', text: 'Please help.' }], + timestamp: new Date().toISOString(), + metadata: { mentioned: true }, + triggerInference: true, + } as unknown as ProcessEvent); + await framework.runUntilIdle(); + const binding = framework.getConversationRouter()!.getBinding(channelId)!; + assert.ok(binding); + const fork = framework.getAgent(binding.agentName)!; + + failPublishing = true; + const finalReply = 'Final reply that the server cannot deliver.'; + membrane.pushResponse(createMockResponse([{ type: 'text', text: finalReply }])); + if (condition === 'other-tool-cycle') state.pendingAssistantBlocks.set('secondary', []); + // A host with a recovery queue: what lands at release is acked out of it. + if (condition === 'ordinary') state.deferredWritesPersisted = true; + if (condition === 'write-fails') { + const cm = fork.getContextManager(); + const store = cm.addMessage.bind(cm); + t.mock.method(cm, 'addMessage', (...args: Parameters) => { + if (JSON.stringify(args[1]).includes(SEND_FAILED)) throw new Error('store refused the write'); + return store(...args); + }); + } + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + binding.lastActivity = 0; + state.lastConversationSweep = 0; + state.sweepExpiredConversations(); + await framework.runUntilIdle(); + state.pendingAssistantBlocks.clear(); + + assert.equal(framework.getAgent(fork.name), null, 'the fork is disposed when its closure turn ends'); + assert.equal(state.deferredMessages.length, 0, 'nothing is left queued for a disposed fork'); + assert.equal(state.unackedDeferredWrites.length, 0, 'nor waiting in the recovery queue'); + // Open a fresh manager over the retained fork namespace, rather than + // relying only on the disposed Agent's in-memory context manager. + const archive = await ContextManager.open({ + store: framework.getStore(), namespace: 'conversations/' + fork.name, + isolate: true, strategy: new PassthroughStrategy(), + }); + const { messages } = await archive.compile(); + const replyIndex = messages.findIndex((m) => m.participant === fork.name && JSON.stringify(m.content).includes(finalReply)); + assert.ok(replyIndex >= 0, 'the failed closure reply remains archived'); + if (condition === 'quiesced') { + // A write while quiesced could land mid-surgery: the notice is not + // written, and the flush that finds its target gone says so. + assert.equal(count(messages, SEND_FAILED), 0); + assert.ok(errors.some((e) => new RegExp(`"${fork.name}" is not a registered agent — message dropped`).test(e))); + } else if (condition === 'write-fails') { + // A write the store refuses is logged; disposal still completes. + assert.equal(count(messages, SEND_FAILED), 0); + assert.ok(errors.some((e) => e.includes(`[deferred-flush] ${fork.name}: failed to store a queued write before release`))); + } else { + const noticeIndex = messages.findIndex((m) => JSON.stringify(m.content).includes(SEND_FAILED)); + assert.ok(noticeIndex > replyIndex, 'the archived failure notice follows the closure reply'); + assert.equal(count(messages, SEND_FAILED), 1); + assert.match(JSON.stringify(messages[noticeIndex]!.content), /delivered:false/); + // Every write of a drained entry carries its durable id (drainDeferredFor). + const stored = archive.getAllMessages().find((m) => JSON.stringify(m.content).includes(SEND_FAILED)); + assert.equal(typeof (stored?.metadata as { deferredWriteId?: unknown } | undefined)?.deferredWriteId, 'string'); + } + const primary = await framework.getAgent('primary')!.getContextManager().compile(); + assert.equal(count(primary.messages, SEND_FAILED), 0); + assert.equal(membrane.calls.length, 2, 'only the initial and closure turns ran'); + assert.equal(state.pendingRequests.length, 0, 'a failed closure reply does not wake another turn'); + }); + } }); }); From ce0c3b29d0ca4f652bab9f8def1c8f4dcb24203b Mon Sep 17 00:00:00 2001 From: Ada-1017 Date: Fri, 9 Oct 2026 21:42:28 -0400 Subject: [PATCH 5/7] fix(routing): pass an ended speaker's failed reply to the primary, by name Basil-1007's review of 5fed304 (room-203 #65318): a notice that a reply never reached the human is only useful to an agent that can still act on it. Landing an ended speaker's notice in its archive, or dropping it when the speaker was already unregistered, told no agent. - passFailedReplyToPrimary(speaker, metadata): for a discord-send-failed notice whose delivery named a channel, add to the primary (default path, so the shared-slot guard and quiesce defer it as usual) "[discord-send- failed] A reply by (N chars) could not be delivered to (). has ended, and the human did not receive it.", metadata { system, kind, channelId, reason, textLen, speaker }. With no channel (no locus) no human was waiting: it logs "[route-failure] has ended, and its reply had no channel to reach: no one else is told" and writes nothing. Not applied to bounces or script wakes: an ended run's typical bounce is a subagent's final answer, which its parent already has as the tool result, and a released agent's script wake has no one to wake. - onRouteFailure: an unregistered speaker (released before its delivery failed, e.g. an async delivered:false) goes to passFailedReplyToPrimary. The notice metadata gains textLen, so a queued notice can be re-told. - landQueuedWritesBeforeRelease: drain first and return when nothing is queued, before touching the context manager (tool-wrapper-prose-guard- integration's "conversation disposal reclaims logical-turn state" disposes a fake fork with no context manager; dropping this early return in 5fed304 failed it on the trial merge's Node 24 run). While quiesced the archive write is dropped, logged, instead of leaving the queue untouched, so the failed reply can still be passed on; ack always, as the teardown flush right after does anyway. Each queued failed reply is also passed to the primary. - channelForNotice(channelId): the label rendering moved out of onRouteFailure, shared with the by-name notice. Two tests added for pieces no test reached: a script wake for an agent no longer registered is dropped, logged, not given to the primary (5fed304's noticeTargetFor); a bounce at release is not passed on (no "[route-failure]" log). Tests 18 (local 18/18 on the PR base): unregistered speaker with a channel -> primary told by name (not woken); without a channel -> nobody, logged; ephemeral with a failing default channel -> its archive plus the primary by name; ephemeral with no locus -> archive only, logged; closure x4 -> the primary is told by name (after the other agent's tool cycle / after resume where deferred). --- .../fix-188-route-failure-owner.fixed.md | 3 +- src/framework.ts | 107 +++++++++++++----- test/route-failure-targeting.test.ts | 81 +++++++++++-- 3 files changed, 147 insertions(+), 44 deletions(-) diff --git a/changelog.d/fix-188-route-failure-owner.fixed.md b/changelog.d/fix-188-route-failure-owner.fixed.md index c2376aed..e1d289d6 100644 --- a/changelog.d/fix-188-route-failure-owner.fixed.md +++ b/changelog.d/fix-188-route-failure-owner.fixed.md @@ -1,5 +1,6 @@ - Notices about an agent's own action now reach the window that agent reads (#188): - A conversation fork's or an ephemeral run's failed-reply notice (`[discord-send-failed]`), bounced-prose notice and background-script wake land in its own window, deferred against its own turn, instead of the primary's. In connectome-host this moves a subagent's notices from its parent to the subagent. - Residents share one message slot, so a resident's notices stay on the default path, unchanged. - - A notice for a speaker that is no longer registered is dropped and logged, rather than given to the primary. + - When a fork or an ephemeral run has ended without reading that its reply to a channel failed, the primary is told by name instead: "A reply by `` (N chars) could not be delivered to … (…). `` has ended, and the human did not receive it." That notice's metadata carries `speaker`, and every failed-reply notice's metadata now carries `textLen`. - Writes still queued for a fork or an ephemeral run when it is released land in its own window first, which Chronicle keeps. While the host is quiesced they are dropped instead, logged. + - A background-script wake for an agent that is no longer registered is dropped, logged, rather than given to the primary. diff --git a/src/framework.ts b/src/framework.ts index 27e6b385..a5349e4e 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -11961,9 +11961,10 @@ export class AgentFramework { * to it. Residents share one message slot, which the default path already * writes, under the guard that waits for any agent's tool cycle and the * primary's turn; addressing a resident would narrow that guard to its own - * turn and cycle. A name no longer registered (a fork or run released before - * its failure was known) is addressed too, so addMessage drops the notice, - * logged, rather than give it to the primary, which did nothing. + * turn and cycle. A name no longer registered (a fork or run already + * released) is addressed too, so addMessage drops the notice, logged, rather + * than give it to the primary as if it were the primary's; a failed reply is + * passed on by name instead (passFailedReplyToPrimary). */ private noticeTargetFor(agentName: string): { forAgent: string } | undefined { const agent = this.agents.get(agentName); @@ -11973,29 +11974,78 @@ export class AgentFramework { /** * A conversation fork or an ephemeral run is about to be deregistered. * Writes still queued for its window would wait for a boundary it never - * reaches, and the flush that finds it gone would drop them; land them in - * its own window now, which stays in Chronicle for investigation. Its - * stream is over or cancelled, so nothing it compiles can diverge from - * them. While quiesced a write could land mid-surgery, so they stay - * queued, and the next flush to find their target gone drops them, logged. + * reaches, and the flush that finds it gone would drop them. Land them in + * its own window now, which Chronicle keeps for investigation; its stream + * is over or cancelled, so nothing it compiles can diverge from them. A + * failed reply among them is also passed to the primary, because the + * speaker will never read it. While quiesced, a write could land + * mid-surgery, so the writes are dropped instead, logged. */ private landQueuedWritesBeforeRelease(agent: Agent): void { - if (this.quiesced) return; + const queued = this.drainDeferredFor(agent.name); + if (queued.length === 0) return; const cm = agent.getContextManager(); - for (const msg of this.drainDeferredFor(agent.name)) { - try { - cm.addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id)); - } catch (err) { + for (const msg of queued) { + if (this.quiesced) { console.error( - `[deferred-flush] ${agent.name}: failed to store a queued write before release ` + - `(participant=${msg.participant}):`, - err, + `[deferred-flush] ${agent.name}: released while quiesced, so a queued write for its window ` + + `is dropped (participant=${msg.participant})`, ); + } else { + try { + cm.addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id)); + } catch (err) { + console.error( + `[deferred-flush] ${agent.name}: failed to store a queued write before release ` + + `(participant=${msg.participant}):`, + err, + ); + } } + this.passFailedReplyToPrimary(agent.name, msg.metadata); } this.ackDeferredWrites(); } + /** + * A fork or an ephemeral run ended before reading that its reply never + * reached the human: it was released with the notice still queued, or its + * delivery failed after release. Tell the primary instead, naming the + * speaker, so an agent who can still act learns of it: to resend, or to + * tell the human. Only for a delivery to a channel; a reply that had no + * channel to go to (no locus) had no human waiting for it. + */ + private passFailedReplyToPrimary(speaker: string, metadata: MessageMetadata | undefined): void { + const failure = metadata as { kind?: unknown; channelId?: unknown; reason?: unknown; textLen?: unknown } | undefined; + if (failure?.kind !== 'discord-send-failed') return; + if (typeof failure.channelId !== 'string' || failure.channelId === '') { + console.error(`[route-failure] ${speaker} has ended, and its reply had no channel to reach: no one else is told`); + return; + } + this.addMessage( + 'user', + [{ + type: 'text', + text: `[discord-send-failed] A reply by ${speaker} (${String(failure.textLen)} chars) could not be delivered to ` + + `${this.channelForNotice(failure.channelId)} (${String(failure.reason)}). ${speaker} has ended, and the human did not receive it.`, + }], + { system: true, kind: 'discord-send-failed', channelId: failure.channelId, reason: failure.reason, textLen: failure.textLen, speaker } as MessageMetadata, + ); + } + + /** + * A channel as a notice names it: the label when the registry has one, since + * a bare snowflake is unresolvable for the agent (the 2026-07-21 incident + * read as "a stale artifact", not a live failure). + */ + private channelForNotice(channelId: string | null): string { + if (!channelId) return 'the channel'; + const label = this.channelRegistry?.getDescriptor(channelId)?.label; + return label && label !== channelId + ? `${label.startsWith('#') ? label : `#${label}`} (${channelId})` + : channelId; + } + private editMessage(id: MessageId, content: ContentBlock[]): void { const agent = this.primaryAgentName ? this.agents.get(this.primaryAgentName) @@ -12588,26 +12638,21 @@ export class AgentFramework { // yes, wake no). onRouteFailure: ({ conversationId, channelId, reason, textLen }) => { try { - // Render a human-readable channel name when we can — a bare - // snowflake in the marker is unresolvable for the agent (the - // 2026-07-21 incident read as "a stale artifact", not a live - // failure). The marker is `system: true`, so it is never - // conversational and never influences routing. - const label = channelId - ? this.channelRegistry?.getDescriptor(channelId)?.label - : undefined; - const where = channelId - ? label && label !== channelId - ? `${label.startsWith('#') ? label : `#${label}`} (${channelId})` - : channelId - : 'the channel'; + // The marker is `system: true`, so it is never conversational and + // never influences routing. + const metadata: MessageMetadata = { system: true, kind: 'discord-send-failed', channelId: channelId ?? '', reason, textLen }; + if (!this.agents.has(conversationId)) { + // Released before its delivery failed: no window left to read it. + this.passFailedReplyToPrimary(conversationId, metadata); + return; + } this.addMessage( 'user', [{ type: 'text', - text: `[discord-send-failed] Your previous reply (${textLen} chars) could not be delivered to ${where} (${reason}). It was saved to your archive but the human did not receive it.`, + text: `[discord-send-failed] Your previous reply (${textLen} chars) could not be delivered to ${this.channelForNotice(channelId)} (${reason}). It was saved to your archive but the human did not receive it.`, }], - { system: true, kind: 'discord-send-failed', channelId: channelId ?? '', reason }, + metadata, this.noticeTargetFor(conversationId), ); } catch (err) { diff --git a/test/route-failure-targeting.test.ts b/test/route-failure-targeting.test.ts index 351dd411..c31daadb 100644 --- a/test/route-failure-targeting.test.ts +++ b/test/route-failure-targeting.test.ts @@ -197,19 +197,32 @@ describe('notices about an agent\'s own action', () => { assert.equal(state.pendingRequests.length, 0); }); - it('drops a notice for a speaker that is no longer registered, logged, rather than give it to the primary', async (t) => { + it('tells the primary, by name, when the speaker ended before its delivery failed', async (t) => { const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); const secondary = t.mock.method(framework.getAgent('secondary')!.getContextManager(), 'addMessage'); + + await internals(framework).channelRegistry.routeSpeech('released-conversation', 'hello', 'missing-channel'); + + assert.equal(secondary.mock.callCount(), 0); + assert.equal(primary.mock.callCount(), 1); + const [, content, metadata] = primary.mock.calls[0]!.arguments; + assert.match(JSON.stringify(content), + /\[discord-send-failed\] A reply by released-conversation \(5 chars\) could not be delivered to missing-channel \(no registered channel[^)]*\)\. released-conversation has ended, and the human did not receive it\./); + assert.equal(metadata?.speaker, 'released-conversation'); + assert.equal(metadata?.textLen, 5); + assert.equal(internals(framework).pendingRequests.length, 0, 'context only: the primary is not woken'); + }); + + it('tells no one, logged, when an ended speaker\'s reply had no channel to reach', async (t) => { + const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); const errors: string[] = []; t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); await internals(framework).channelRegistry.routeSpeech('released-conversation', 'hello', null); assert.equal(primary.mock.callCount(), 0); - assert.equal(secondary.mock.callCount(), 0); - assert.ok(errors.some((e) => /"released-conversation" is not a registered agent — message dropped/.test(e))); assert.equal(internals(framework).deferredMessages.length, 0); - assert.equal(internals(framework).pendingRequests.length, 0); + assert.ok(errors.some((e) => e.includes('[route-failure] released-conversation has ended, and its reply had no channel to reach'))); }); it('lands an ephemeral run\'s last failure notice in its own window before releasing it', async (t) => { @@ -227,18 +240,48 @@ describe('notices about an agent\'s own action', () => { const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); assert.equal(count(messages, SEND_FAILED), 0, 'the primary did not write that reply'); assert.ok(!errors.some((e) => /not a registered agent/.test(e))); + assert.ok(errors.some((e) => e.includes('[route-failure] worker has ended, and its reply had no channel to reach'))); assert.equal(internals(framework).deferredMessages.length, 0); }); - it('sends an ephemeral run\'s bounced prose notice to its own window, beside its clipboard', async () => { + it('passes an ephemeral run\'s failed delivery to a channel on to the primary, by name', async () => { + (internals(framework).channelRegistry as unknown as { defaultPublishChannel: string | null }).defaultPublishChannel = 'missing-channel'; + const speech = 'Worker speech for a channel that is not registered.'; + + const archived = await runEphemeral('locus', speech); + + assert.equal(count(archived, '[discord-send-failed] Your previous reply'), 1, 'the run keeps its own record'); + const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); + const named = messages.filter((m) => JSON.stringify(m.content).includes(SEND_FAILED)); + assert.equal(named.length, 1); + assert.match(JSON.stringify(named[0]!.content), /A reply by worker \(\d+ chars\) could not be delivered to missing-channel .*worker has ended/); + assert.equal(internals(framework).deferredMessages.length, 0); + }); + + it('sends an ephemeral run\'s bounced prose notice to its own window, beside its clipboard', async (t) => { + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + const archived = await runEphemeral('explicit', 'Worker prose with no destination.'); assert.equal(count(archived, BOUNCED), 1, 'the speaker reads why its text did not go out'); const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); - assert.equal(count(messages, BOUNCED), 0, 'the primary wrote no such text'); + assert.equal(count(messages, BOUNCED), 0, 'the primary wrote no such text, and is not told of it'); + assert.ok(!errors.some((e) => e.includes('[route-failure]')), 'a bounce is not a failed reply'); assert.equal(internals(framework).deferredMessages.length, 0); }); + it('drops, logged, a background-script wake for an agent no longer registered', async (t) => { + const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + + internals(framework).injectScriptWake({ id: 's-2', agentName: 'released-run' }, '[background script s-2] Your background script DIED.'); + + assert.equal(primary.mock.callCount(), 0, 'the primary started no such script'); + assert.ok(errors.some((e) => /"released-run" is not a registered agent — message dropped/.test(e))); + }); + it('wakes a fork\'s background script into the fork\'s own window', async () => { const fork = await spawnFork(); const envelope = '[background script s-1] Your background script DIED 0m after start.'; @@ -305,7 +348,20 @@ describe('notices about an agent\'s own action', () => { state.pendingAssistantBlocks.clear(); assert.equal(framework.getAgent(fork.name), null, 'the fork is disposed when its closure turn ends'); - assert.equal(state.deferredMessages.length, 0, 'nothing is left queued for a disposed fork'); + if (condition === 'quiesced') { + // A write while quiesced could land mid-surgery: the archive write + // is dropped, logged, and the primary's notice waits for resume. + assert.ok(errors.some((e) => e.includes(`[deferred-flush] ${fork.name}: released while quiesced, so a queued write for its window is dropped`))); + assert.equal(state.deferredMessages.length, 1); + state.quiesced = false; + await state.flushDeferredWrites('closure test resume'); + } + if (condition === 'other-tool-cycle') { + // The primary's notice waits for the shared slot's tool cycle. + assert.equal(state.deferredMessages.length, 1); + await state.flushDeferredWrites('the primary\'s next boundary'); + } + assert.equal(state.deferredMessages.length, 0, 'nothing is left queued'); assert.equal(state.unackedDeferredWrites.length, 0, 'nor waiting in the recovery queue'); // Open a fresh manager over the retained fork namespace, rather than // relying only on the disposed Agent's in-memory context manager. @@ -317,10 +373,7 @@ describe('notices about an agent\'s own action', () => { const replyIndex = messages.findIndex((m) => m.participant === fork.name && JSON.stringify(m.content).includes(finalReply)); assert.ok(replyIndex >= 0, 'the failed closure reply remains archived'); if (condition === 'quiesced') { - // A write while quiesced could land mid-surgery: the notice is not - // written, and the flush that finds its target gone says so. assert.equal(count(messages, SEND_FAILED), 0); - assert.ok(errors.some((e) => new RegExp(`"${fork.name}" is not a registered agent — message dropped`).test(e))); } else if (condition === 'write-fails') { // A write the store refuses is logged; disposal still completes. assert.equal(count(messages, SEND_FAILED), 0); @@ -329,13 +382,17 @@ describe('notices about an agent\'s own action', () => { const noticeIndex = messages.findIndex((m) => JSON.stringify(m.content).includes(SEND_FAILED)); assert.ok(noticeIndex > replyIndex, 'the archived failure notice follows the closure reply'); assert.equal(count(messages, SEND_FAILED), 1); - assert.match(JSON.stringify(messages[noticeIndex]!.content), /delivered:false/); + assert.match(JSON.stringify(messages[noticeIndex]!.content), /Your previous reply.*delivered:false/); // Every write of a drained entry carries its durable id (drainDeferredFor). const stored = archive.getAllMessages().find((m) => JSON.stringify(m.content).includes(SEND_FAILED)); assert.equal(typeof (stored?.metadata as { deferredWriteId?: unknown } | undefined)?.deferredWriteId, 'string'); } + // The fork never reads its notice again, so the primary is told, by name. const primary = await framework.getAgent('primary')!.getContextManager().compile(); - assert.equal(count(primary.messages, SEND_FAILED), 0); + const told = primary.messages.filter((m) => JSON.stringify(m.content).includes(SEND_FAILED)); + assert.equal(told.length, 1); + assert.match(JSON.stringify(told[0]!.content), + new RegExp(`A reply by ${fork.name} \\(${finalReply.length} chars\\) could not be delivered to #closure test \\(closure-channel\\) \\(.*delivered:false.*\\)\\. ${fork.name} has ended, and the human did not receive it\\.`)); assert.equal(membrane.calls.length, 2, 'only the initial and closure turns ran'); assert.equal(state.pendingRequests.length, 0, 'a failed closure reply does not wake another turn'); }); From 43dd1c7e4cb657a9f6fc33649bbb6fc00f6fcfe0 Mon Sep 17 00:00:00 2001 From: Ada-1017 Date: Fri, 9 Oct 2026 21:47:46 -0400 Subject: [PATCH 6/7] fix(routing): a release never fails on the primary's by-name notice Basil-1007's approval of ce0c3b2 (room-203 #65323) asked for one guard: passFailedReplyToPrimary runs inside a release, before agents.delete, and called addMessage bare. A throw ("No agents configured", or a store error from the CM write) would leave a fork registered, or escape runEphemeralToCompletion's finally before deregistration. - try/catch around the write, logged: "[route-failure] failed to tell the primary that 's reply to failed:". - Found while writing its test: with no primary (agents: []), the default path falls back to the first registered agent, which can be the very run being released, so it would have been told about itself. Check for a registered primary first; without one, log "[route-failure] has ended, and there is no primary to tell that its reply to failed" and write nothing. Tests (20, local 20/20): a store error on the primary's notice -> the run is still released, its archive keeps its own notice, logged; no primary -> the run's archive has no by-name notice about itself, logged. Removing the primary check fails the second (checked locally). --- src/framework.ts | 31 +++++++++++------ test/route-failure-targeting.test.ts | 51 ++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 10 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index a5349e4e..664578c3 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -12013,7 +12013,8 @@ export class AgentFramework { * delivery failed after release. Tell the primary instead, naming the * speaker, so an agent who can still act learns of it: to resend, or to * tell the human. Only for a delivery to a channel; a reply that had no - * channel to go to (no locus) had no human waiting for it. + * channel to go to (no locus) had no human waiting for it. Never throws: + * it runs inside a release, which must complete whatever this write meets. */ private passFailedReplyToPrimary(speaker: string, metadata: MessageMetadata | undefined): void { const failure = metadata as { kind?: unknown; channelId?: unknown; reason?: unknown; textLen?: unknown } | undefined; @@ -12022,15 +12023,25 @@ export class AgentFramework { console.error(`[route-failure] ${speaker} has ended, and its reply had no channel to reach: no one else is told`); return; } - this.addMessage( - 'user', - [{ - type: 'text', - text: `[discord-send-failed] A reply by ${speaker} (${String(failure.textLen)} chars) could not be delivered to ` + - `${this.channelForNotice(failure.channelId)} (${String(failure.reason)}). ${speaker} has ended, and the human did not receive it.`, - }], - { system: true, kind: 'discord-send-failed', channelId: failure.channelId, reason: failure.reason, textLen: failure.textLen, speaker } as MessageMetadata, - ); + // The default path falls back to the first registered agent when there + // is no primary, which could be the speaker being released. + if (!this.primaryAgentName || !this.agents.has(this.primaryAgentName)) { + console.error(`[route-failure] ${speaker} has ended, and there is no primary to tell that its reply to ${failure.channelId} failed`); + return; + } + try { + this.addMessage( + 'user', + [{ + type: 'text', + text: `[discord-send-failed] A reply by ${speaker} (${String(failure.textLen)} chars) could not be delivered to ` + + `${this.channelForNotice(failure.channelId)} (${String(failure.reason)}). ${speaker} has ended, and the human did not receive it.`, + }], + { system: true, kind: 'discord-send-failed', channelId: failure.channelId, reason: failure.reason, textLen: failure.textLen, speaker } as MessageMetadata, + ); + } catch (err) { + console.error(`[route-failure] failed to tell the primary that ${speaker}'s reply to ${failure.channelId} failed:`, err); + } } /** diff --git a/test/route-failure-targeting.test.ts b/test/route-failure-targeting.test.ts index c31daadb..a28c3db1 100644 --- a/test/route-failure-targeting.test.ts +++ b/test/route-failure-targeting.test.ts @@ -258,6 +258,57 @@ describe('notices about an agent\'s own action', () => { assert.equal(internals(framework).deferredMessages.length, 0); }); + it('still releases an ephemeral run when the primary cannot be told, logged', async (t) => { + (internals(framework).channelRegistry as unknown as { defaultPublishChannel: string | null }).defaultPublishChannel = 'missing-channel'; + const primaryCm = framework.getAgent('primary')!.getContextManager(); + const store = primaryCm.addMessage.bind(primaryCm); + t.mock.method(primaryCm, 'addMessage', (...args: Parameters) => { + if (JSON.stringify(args[1]).includes('A reply by worker')) throw new Error('store refused the write'); + return store(...args); + }); + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + + const archived = await runEphemeral('locus', 'Worker speech for a channel that is not registered.'); + + assert.equal(framework.getAgent('worker'), null, 'the run is released'); + assert.equal(count(archived, '[discord-send-failed] Your previous reply'), 1); + assert.ok(errors.some((e) => e.includes('[route-failure] failed to tell the primary that worker\'s reply to missing-channel failed'))); + }); + + it('tells no one, logged, when there is no primary to tell', async (t) => { + const dir = mkdtempSync(join(tmpdir(), 'route-failure-no-primary-')); + const bare = await AgentFramework.create({ + storePath: join(dir, 'test.chronicle'), membrane: membrane.asMembrane(), agents: [], modules: [], + }); + try { + await internals(bare).initializeMcpl([]); + (internals(bare).channelRegistry as unknown as { defaultPublishChannel: string | null }).defaultPublishChannel = 'missing-channel'; + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + const created = await bare.createEphemeralAgent({ + name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', proseRouting: 'locus', + }); + created.contextManager.addMessage('user', [{ type: 'text', text: 'Run once.' }]); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Worker speech with nobody to tell.' }])); + const run = bare.runEphemeralToCompletion(created.agent, created.contextManager); + bare.start(); + await run; + await bare.runUntilIdle(); + + const archive = await ContextManager.open({ + store: bare.getStore(), namespace: 'subagent/worker', isolate: true, strategy: new PassthroughStrategy(), + }); + const { messages } = await archive.compile(); + assert.equal(count(messages, 'A reply by worker'), 0, 'the run is not told about itself'); + assert.equal(count(messages, '[discord-send-failed] Your previous reply'), 1); + assert.ok(errors.some((e) => e.includes('[route-failure] worker has ended, and there is no primary to tell that its reply to missing-channel failed'))); + } finally { + await bare.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + it('sends an ephemeral run\'s bounced prose notice to its own window, beside its clipboard', async (t) => { const errors: string[] = []; t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); From d65cb181f3448c29203ba165e9e542ae314d786e Mon Sep 17 00:00:00 2001 From: Oscar-1808 Date: Sun, 11 Oct 2026 16:58:39 -0400 Subject: [PATCH 7/7] fix(routing): every notice goes to the agent it's about, and a released agent's unread failures reach the primary Review of this branch found that a failed reply already written into a fork's or a run's window was never passed on, that a release while quiesced lost its writes, and that some notices still went to the wrong place or left a misleading trace: - At release, every failed-reply notice the agent never read reaches the primary by name, once: each one still queued for it, and each one stored in its window above its consumed watermark (flushed there at its turn's end, or stored while it was idle). A notice injected into its live stream at a tool boundary was read, though no compile ran after it, so it is recorded at injection and not passed on. - A release asked for while the host is quiesced waits for resume(), which runs it before its own flush; the agent stays registered meanwhile and can't start a turn. Before, its queued writes were dropped and acked, lost from the window and the recovery queue. - The fallback reaper treats a held turn token as busy, so a closing fork mid-compile, before its stream registers, isn't disposed. - noticeTargetFor addresses every registered agent's notice to that agent, residents included, so a resident's notice waits on its own turn, stream and tool cycle rather than the primary's. Residents share one slot, so one may land there during another resident's turn; agent-framework #291 covers that. - An agent no longer registered gets nothing: noticeTargetFor logs the drop once and returns null, and its caller writes nothing, traces nothing and requests no wake (a bounce's wake and a background script's wake included). - A channel opened by an agent's own send is announced to that agent; a policy open, with no agent behind it, still goes to the primary. - A restored deferred write whose target is gone (boot recovery, or a resume's flush) passes a failed-reply notice to the primary by name; other writes are dropped, logged, as before. - The by-name notice says only what it has: an entry queued by an earlier build, without textLen or reason, no longer reads "undefined chars". - A background script's deferred wake is traced as deferred, not with an empty id. Tests (test/route-failure-targeting.test.ts, 33, up from 20): - a resident's notice addressed to the speaker and waiting on its own turn and tool cycle, and the primary's own wording pinned; - a fork's stored, unread notice passed on at its release, and one its later turn compiled kept its own; - the reaper and a held turn token; - a queued ordinary write landing in a released fork's window with its durable id, and acked; - a fork's notice waiting on its own tool cycle and its own stream, and landing while another agent's cycle is pending; - the auto-open notice to its opener, to the primary for a policy open, and dropped for an agent that is gone; - a background-script wake for an agent that is gone: no write, trace or wake; - restored notices for gone agents passed on by name, an earlier build's without "undefined"; - the TTL closure while quiesced, through resume() itself; - with a module tool in an ephemeral run: a notice heard at a tool boundary isn't passed on; a bounced prose notice is injected at the next tool boundary and follows its prose in the window; a run that ends while quiesced is released at resume, its queued write landed. Red checks: against the previous commit's framework.ts, 13 of the tests fail, the four residents' tests and nine aimed at the fixes. Each of these single changes to the new framework.ts fails the test aimed at it: no stored notice counted unread; a live-injected notice counted unread; the reaper ignoring turn tokens; releases run at once while quiesced; a fork's dispose not waiting; resume running no postponed release; a restored notice for a gone agent not passed on; a missing textLen printed; an unregistered agent's notice still addressed; an auto-open notice ignoring who caused it; a release not acking what it landed; a fork's notice waiting on every agent's tool cycle; a notice not waiting on its agent's stream. Changelog: the header names its exceptions; a subagent's notices moved from the primary, not from a parent; residents' notices are addressed to them, with #291 named; every failed-reply notice carries textLen; with no primary, no one is told, and it's logged. Co-Authored-By: Claude Opus 5.5 --- .../fix-188-route-failure-owner.fixed.md | 14 +- src/framework.ts | 278 +++++++++---- test/route-failure-targeting.test.ts | 376 ++++++++++++++++-- 3 files changed, 561 insertions(+), 107 deletions(-) diff --git a/changelog.d/fix-188-route-failure-owner.fixed.md b/changelog.d/fix-188-route-failure-owner.fixed.md index e1d289d6..86c49735 100644 --- a/changelog.d/fix-188-route-failure-owner.fixed.md +++ b/changelog.d/fix-188-route-failure-owner.fixed.md @@ -1,6 +1,8 @@ -- Notices about an agent's own action now reach the window that agent reads (#188): - - A conversation fork's or an ephemeral run's failed-reply notice (`[discord-send-failed]`), bounced-prose notice and background-script wake land in its own window, deferred against its own turn, instead of the primary's. In connectome-host this moves a subagent's notices from its parent to the subagent. - - Residents share one message slot, so a resident's notices stay on the default path, unchanged. - - When a fork or an ephemeral run has ended without reading that its reply to a channel failed, the primary is told by name instead: "A reply by `` (N chars) could not be delivered to … (…). `` has ended, and the human did not receive it." That notice's metadata carries `speaker`, and every failed-reply notice's metadata now carries `textLen`. - - Writes still queued for a fork or an ephemeral run when it is released land in its own window first, which Chronicle keeps. While the host is quiesced they are dropped instead, logged. - - A background-script wake for an agent that is no longer registered is dropped, logged, rather than given to the primary. +- Notices about an agent's own action now reach the window that agent reads, deferred against that agent's own turn, stream and tool cycle (#188). Two exceptions: a failed reply its speaker will never read goes to the primary by name, and a notice for an agent no longer registered goes nowhere, logged. + - The notices: a failed reply (`[discord-send-failed]`), bounced prose (beside its `{{unsent}}` clipboard), a background script's wake, and a channel opened by the agent's own send. A channel opened by subscription policy, with no agent behind it, is still announced to the primary. + - A conversation fork's or an ephemeral run's notices land in its own window instead of the primary's. In connectome-host this moves a subagent's notices from the primary, where the default path put them, to the subagent. + - A resident's notices are addressed to it too, in the slot the residents share, and wait on its own turn rather than the primary's. Whether it may land there while another resident's turn is alive is the question of #291. + - When a fork or an ephemeral run is released, each failed-reply notice it never read is passed to the primary, once, by name: "A reply by `` (N chars) could not be delivered to … (…). `` has ended, and the human did not receive it." Never read means still queued for it, or stored in its window since its last compile and not heard mid-turn at a tool boundary. The same goes for a delivery that fails after its speaker is released, and for such a notice restored at boot whose agent is gone. That notice's metadata carries `speaker`, and every failed-reply notice's metadata now carries `textLen`. With no primary registered, no one is told, and it's logged. + - Writes still queued for a fork or an ephemeral run when it is released land in its own window first, which Chronicle keeps. A release asked for while the host is quiesced waits for resume, and lands them then. + - The fallback reaper no longer disposes a closing fork whose turn is still alive before its stream has registered. + - A notice or a background-script wake for an agent that is no longer registered is dropped, logged once, with no trace and no wake. diff --git a/src/framework.ts b/src/framework.ts index 664578c3..84a5b383 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -1277,6 +1277,15 @@ export class AgentFramework { * ephemeral agents have isolated slots and are their own only readers. */ private readonly sharedSlotAgents = new Set(); + /** + * A fork's or an ephemeral run's failed-reply notices that reached its live + * stream at a tool boundary, by deferred-write id. Read, though stored above + * its consumed watermark, since no compile ran after them; a release + * (unreadFailedReplies) doesn't pass them on. Forgotten at release. + */ + private readonly injectedFailureNotices = new Map>(); + /** Releases asked for while the host was quiesced, by agent name; resume() runs them. */ + private readonly releasesAfterResume = new Map void>(); /** Group commit: one fsync per event-loop turn, shared by every waiter. */ private coalescingCommit: Promise | null = null; private inferenceRouter: InferenceRouter | null = null; @@ -3060,10 +3069,14 @@ export class AgentFramework { this.eventGate?.setQuiesced(false); try { - // Flush context writes deferred by the quiesce window (module events, - // api message.send) — with no turn alive there is no other flush - // point. Per-message try/catch and per-message durable ack (see - // flushDeferredWrites); nothing here may reach the plane reopen below. + // Releases asked for during the window run first, each landing its + // agent's queued writes and passing on its unread failed replies, as at + // any release. Then flush context writes deferred by the quiesce window + // (module events, api message.send) — with no turn alive there is no + // other flush point. Per-message try/catch and per-message durable ack + // (see flushDeferredWrites); nothing here may reach the plane reopen + // below. + this.runReleasesAfterResume(); await this.flushDeferredWrites('resume'); this.persistHostMode(null); } finally { @@ -3132,7 +3145,20 @@ export class AgentFramework { this.deferredMessages = keep; this.handOffDeferredWrites(flush); let stored = 0; + let passed = 0; for (const msg of flush) { + // A target no longer registered (a fork or a run gone before a crash + // that left this write to boot recovery) has no window to land in. A + // failed reply it never read is passed to the primary, by name; any + // other write is dropped, logged, by addMessage. + if ( + msg.forAgent && !this.agents.has(msg.forAgent) && + (msg.metadata as { kind?: unknown } | undefined)?.kind === 'discord-send-failed' + ) { + this.passFailedReplyToPrimary(msg.forAgent, msg.metadata); + passed++; + continue; + } try { this.addMessage(msg.participant, msg.content, msg.metadata, { deferredWriteId: msg.id, @@ -3152,6 +3178,7 @@ export class AgentFramework { this.ackDeferredWrites(); console.error( `[host-mode] ${label}: flushed ${stored}/${flush.length} deferred context write(s)` + + (passed > 0 ? `, ${passed} failed repl${passed === 1 ? 'y' : 'ies'} of a released agent passed to the primary` : '') + (keep.length > 0 ? `, ${keep.length} kept for a still-alive turn` : ''), ); } @@ -4137,8 +4164,11 @@ export class AgentFramework { // The late stream finalizer intentionally refuses name-keyed cleanup once // deregistered, so disposal must release EventGate liveness itself. this.eventGate?.onInferenceEnded(agent.name); - this.landQueuedWritesBeforeRelease(agent); - this.agents.delete(agent.name); + this.releaseNowOrAfterResume(agent.name, () => { + if (this.agents.get(agent.name) !== agent) return; + this.releaseWindow(agent); + this.agents.delete(agent.name); + }); } this.toolImageLedgers.delete(agent.name); // Spawn-and-dispose bookkeeping (main, d453165/fee96a7): without this, @@ -6486,6 +6516,7 @@ export class AgentFramework { content: msg.content, ...(msg.metadata ? { metadata: msg.metadata } : {}), }); + this.noteInjectedFailureNotice(agent, msg.id, msg.metadata); } } } @@ -7170,10 +7201,16 @@ export class AgentFramework { * forever. The fork's context stays in Chronicle for investigation. */ private disposeConversationAgent(agentName: string): void { + if (this.quiesced) { + // Released by resume() (releaseNowOrAfterResume). It stays closing, so + // a sweep meanwhile reaches here again and waits too. + this.releasesAfterResume.set(agentName, () => this.disposeConversationAgent(agentName)); + return; + } this.closingConversationAgents.delete(agentName); const channelId = this.conversationAgentHomes.get(agentName); const agent = this.agents.get(agentName); - if (agent) this.landQueuedWritesBeforeRelease(agent); + if (agent) this.releaseWindow(agent); this.agents.delete(agentName); this.toolImageLedgers.delete(agentName); this.agentConfigs.delete(agentName); @@ -7259,9 +7296,12 @@ export class AgentFramework { // Fallback reaper: a closing fork whose closure inference never ran // (request dropped as stale, inference policy veto) would otherwise stay - // registered forever. No active stream + no pending request = done. + // registered forever. No turn, no active stream and no pending request = + // done. A turn holds its token from dequeue, through its hooks and + // compile, before its stream registers, so a fork mid-compile is busy. for (const agentName of [...this.closingConversationAgents]) { if ( + !this.activeTurnTokens.has(agentName) && !this.activeStreams.has(agentName) && !this.pendingRequests.some((r) => r.agentName === agentName) ) { @@ -8335,13 +8375,21 @@ export class AgentFramework { (plural ? 'these channels' : 'this channel') + ` will reach you. If you don't want ${plural ? 'one of them' : 'it'}, use channel_close — ` + 'your choice sticks; neither policy nor a later send will silently reopen it.'; + // An open caused by an agent's own send is that agent's to hear about, in + // the window it reads; a policy admission with no agent behind it goes to + // the primary. + let target: { forAgent: string } | undefined; + if (agentName) { + const addressed = this.noticeTargetFor(agentName, 'channel auto-open'); + if (!addressed) return; + target = addressed; + } try { // Through framework.addMessage for mid-turn safety (deferral + // hear-while-acting injection) — a delivery-forced open happens inside // a live turn, and a direct context-manager append there would store - // the notice out of order. Targets the primary agent; connectome-host - // runs trunk-only, so this is the delivering agent in practice. - this.addMessage('user', [{ type: 'text', text }], { system: true, kind: 'channel-notice' }); + // the notice out of order. + this.addMessage('user', [{ type: 'text', text }], { system: true, kind: 'channel-notice' }, target); console.error(`[channel] auto-open notice (${cause}) -> ${agentName ?? 'primary'}: ${shown}`); } catch (err) { console.error('recordChannelAutoOpenNotice failed:', err); @@ -8546,31 +8594,34 @@ export class AgentFramework { `destination plus the token {{unsent}}, e.g. "${prefix}#channel {{unsent}}" or "${prefix}@person {{unsent}}". ` + `"${prefix}skip_reply {{unsent}}" keeps it in context only.` + (prefix === '>>' ? ' The prose_help tool shows the full syntax.' : ''); - try { - // Through framework.addMessage, NOT the context manager directly: while - // the turn is still streaming this defers the notice to the next tool - // boundary, where it is BOTH stored in order and injected into the live - // stream (hear-while-acting) — the model sees the error and can correct - // the prefix within the same turn. A direct append here would land the - // notice before the turn's assistant blocks (cache bust, invisible - // until the next compile). - const id = this.addMessage( - 'user', - [{ type: 'text', text: notice }], - { system: true, kind: 'prose-bounce' }, - this.noticeTargetFor(name), - ); - if (id) this.emitTrace({ type: 'message:added', messageId: id, source: 'prose-bounce' }); - else this.emitTrace({ type: 'message:added', messageId: 'deferred', source: 'prose-bounce' }); - } catch (err) { - console.error('bounceProse: failed to record bounce notice:', err); + const target = this.noticeTargetFor(name, 'prose-bounce'); + if (target) { + try { + // Through framework.addMessage, NOT the context manager directly: while + // the turn is still streaming this defers the notice to the next tool + // boundary, where it is BOTH stored in order and injected into the live + // stream (hear-while-acting) — the model sees the error and can correct + // the prefix within the same turn. A direct append here would land the + // notice before the turn's assistant blocks (cache bust, invisible + // until the next compile). + const id = this.addMessage( + 'user', + [{ type: 'text', text: notice }], + { system: true, kind: 'prose-bounce' }, + target, + ); + if (id) this.emitTrace({ type: 'message:added', messageId: id, source: 'prose-bounce' }); + else this.emitTrace({ type: 'message:added', messageId: 'deferred', source: 'prose-bounce' }); + } catch (err) { + console.error('bounceProse: failed to record bounce notice:', err); + } } const capped = streak > AgentFramework.PROSE_BOUNCE_WAKE_CAP; console.error( `[prose] ${name}: BOUNCED ${text.length} chars — ${reason}` + (capped ? ' (bounce-wake cap reached; notice appended without wake)' : ''), ); - if (!capped) { + if (!capped && target) { // Wake so the resend can happen immediately — an unsent DM reply must // not sit invisible until the next unrelated event. this.pendingRequests.push({ @@ -11031,12 +11082,14 @@ export class AgentFramework { * later if that ever needs revisiting. */ private injectScriptWake(record: BackgroundScriptRecord, envelope: string): void { + const target = this.noticeTargetFor(record.agentName, `background-script ${record.id} wake`); + if (!target) return; try { const messageId = this.addMessage('user', [{ type: 'text', text: envelope }], { source: 'background-script', scriptId: record.id, tags: ['script:wake'], - }, this.noticeTargetFor(record.agentName)); + }, target); this.pendingRequests.push({ agentName: record.agentName, reason: 'script:wake', @@ -11045,7 +11098,7 @@ export class AgentFramework { }); this.emitTrace({ type: 'message:added', - messageId, + messageId: messageId || 'deferred', source: `background-script:${record.id}`, }); } catch (err) { @@ -11956,60 +12009,131 @@ export class AgentFramework { /** * Where a notice about an agent's own action belongs: its failed reply, its - * bounced prose, its background script's wake. A conversation fork or an - * ephemeral run reads only its own isolated slot, so the notice is addressed - * to it. Residents share one message slot, which the default path already - * writes, under the guard that waits for any agent's tool cycle and the - * primary's turn; addressing a resident would narrow that guard to its own - * turn and cycle. A name no longer registered (a fork or run already - * released) is addressed too, so addMessage drops the notice, logged, rather - * than give it to the primary as if it were the primary's; a failed reply is - * passed on by name instead (passFailedReplyToPrimary). - */ - private noticeTargetFor(agentName: string): { forAgent: string } | undefined { - const agent = this.agents.get(agentName); - return agent && this.sharedSlotAgents.has(agent) ? undefined : { forAgent: agentName }; + * bounced prose (beside its `{{unsent}}` clipboard), its background script's + * wake, a channel its own send opened. Every registered agent's notice is + * addressed to that agent, so it lands in the window that agent reads and + * waits on that agent's own turn, stream and tool cycle: a conversation + * fork or an ephemeral run reads only its isolated slot, and a resident's + * notice would otherwise wait on the primary's turn rather than its own. + * Residents share one message slot, so a resident's notice can land there + * while another resident's turn is alive; agent-framework #291 covers that. + * + * null when the agent is no longer registered (a fork or a run already + * released): there is no window to write, so the caller writes nothing, + * traces nothing and requests no wake. The drop is logged here, once. A + * failed reply is passed on by name instead (passFailedReplyToPrimary). + */ + private noticeTargetFor(agentName: string, notice: string): { forAgent: string } | null { + if (this.agents.has(agentName)) return { forAgent: agentName }; + console.error(`[notice] ${agentName} is no longer registered, so its ${notice} notice is dropped`); + return null; } /** - * A conversation fork or an ephemeral run is about to be deregistered. - * Writes still queued for its window would wait for a boundary it never - * reaches, and the flush that finds it gone would drop them. Land them in - * its own window now, which Chronicle keeps for investigation; its stream - * is over or cancelled, so nothing it compiles can diverge from them. A - * failed reply among them is also passed to the primary, because the - * speaker will never read it. While quiesced, a write could land - * mid-surgery, so the writes are dropped instead, logged. - */ - private landQueuedWritesBeforeRelease(agent: Agent): void { + * A conversation fork or an ephemeral run is about to be deregistered. A + * release never runs while the host is quiesced: it waits for resume() + * (releaseNowOrAfterResume), since a write could land mid-surgery. + * + * The agent never reads its window again, so every failed-reply notice it + * never read is passed to the primary, by name, once: each one stored in + * its window above its consumed watermark (flushed there at its turn's end, + * or stored while it was idle) and not heard at a tool boundary, and each + * one still queued for it. Writes still + * queued would wait for a boundary it never reaches, and the flush that + * finds it gone would drop them, so they land in its own window now, which + * Chronicle keeps for investigation; its stream is over or cancelled, so + * nothing it compiles can diverge from them. + */ + private releaseWindow(agent: Agent): void { + for (const metadata of this.unreadFailedReplies(agent)) this.passFailedReplyToPrimary(agent.name, metadata); const queued = this.drainDeferredFor(agent.name); if (queued.length === 0) return; const cm = agent.getContextManager(); for (const msg of queued) { - if (this.quiesced) { + try { + cm.addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id)); + } catch (err) { console.error( - `[deferred-flush] ${agent.name}: released while quiesced, so a queued write for its window ` + - `is dropped (participant=${msg.participant})`, + `[deferred-flush] ${agent.name}: failed to store a queued write before release ` + + `(participant=${msg.participant}):`, + err, ); - } else { - try { - cm.addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id)); - } catch (err) { - console.error( - `[deferred-flush] ${agent.name}: failed to store a queued write before release ` + - `(participant=${msg.participant}):`, - err, - ); - } } this.passFailedReplyToPrimary(agent.name, msg.metadata); } this.ackDeferredWrites(); } + /** + * The failed-reply notices stored in `agent`'s window that it never read: + * above its consumed watermark (or any, when it never compiled on this + * branch), and not injected into its live stream at a tool boundary. + */ + private unreadFailedReplies(agent: Agent): MessageMetadata[] { + const injected = this.injectedFailureNotices.get(agent.name); + this.injectedFailureNotices.delete(agent.name); + try { + const cm = agent.getContextManager(); + const watermark = agent.getConsumedWatermark(); + const floor = watermark && watermark.branch === cm.currentBranch().name + ? watermark.sequence + : Number.NEGATIVE_INFINITY; + return cm.getAllMessages() + .filter((m) => { + const metadata = m.metadata as { kind?: unknown; deferredWriteId?: unknown } | undefined; + return m.sequence > floor && metadata?.kind === 'discord-send-failed' && + !(typeof metadata.deferredWriteId === 'string' && injected?.has(metadata.deferredWriteId)); + }) + .map((m) => m.metadata as MessageMetadata); + } catch (err) { + // A release must complete whatever its window's read meets. + console.error(`[route-failure] ${agent.name}: could not read its window for unread failed replies:`, err); + return []; + } + } + + /** + * Record a failed-reply notice injected into `agent`'s live stream at a + * tool boundary (injectedFailureNotices). Residents are never released, so + * only an isolated agent's are kept. + */ + private noteInjectedFailureNotice(agent: Agent, id: string, metadata: MessageMetadata | undefined): void { + if (this.sharedSlotAgents.has(agent)) return; + if ((metadata as { kind?: unknown } | undefined)?.kind !== 'discord-send-failed') return; + let ids = this.injectedFailureNotices.get(agent.name); + if (!ids) this.injectedFailureNotices.set(agent.name, (ids = new Set())); + ids.add(id); + } + + /** + * Run `release` now, or, while the host is quiesced, when it resumes. A + * release lands the agent's queued writes in its window, and while quiesced + * a write could land mid-surgery; dropping them instead would lose them + * from both the window and the recovery queue. The agent stays registered + * meanwhile, and can't start a turn while quiesced. + */ + private releaseNowOrAfterResume(agentName: string, release: () => void): void { + if (this.quiesced) this.releasesAfterResume.set(agentName, release); + else release(); + } + + /** resume(): the releases asked for during the quiesce window, each on its own. */ + private runReleasesAfterResume(): void { + const releases = [...this.releasesAfterResume.values()]; + this.releasesAfterResume.clear(); + for (const release of releases) { + try { + release(); + } catch (err) { + console.error('[host-mode] resume: a release postponed by the quiesce window failed:', err); + } + } + } + /** * A fork or an ephemeral run ended before reading that its reply never - * reached the human: it was released with the notice still queued, or its + * reached the human: it was released with the notice unread or still + * queued, its queued notice was restored at boot after it was gone, or its * delivery failed after release. Tell the primary instead, naming the * speaker, so an agent who can still act learns of it: to resend, or to * tell the human. Only for a delivery to a channel; a reply that had no @@ -12029,15 +12153,22 @@ export class AgentFramework { console.error(`[route-failure] ${speaker} has ended, and there is no primary to tell that its reply to ${failure.channelId} failed`); return; } + // A notice queued by an earlier build may carry neither; say only what it has. + const textLen = typeof failure.textLen === 'number' ? failure.textLen : undefined; + const reason = typeof failure.reason === 'string' && failure.reason !== '' ? failure.reason : undefined; try { this.addMessage( 'user', [{ type: 'text', - text: `[discord-send-failed] A reply by ${speaker} (${String(failure.textLen)} chars) could not be delivered to ` + - `${this.channelForNotice(failure.channelId)} (${String(failure.reason)}). ${speaker} has ended, and the human did not receive it.`, + text: `[discord-send-failed] A reply by ${speaker}${textLen !== undefined ? ` (${textLen} chars)` : ''} could not be delivered to ` + + `${this.channelForNotice(failure.channelId)}${reason !== undefined ? ` (${reason})` : ''}. ${speaker} has ended, and the human did not receive it.`, }], - { system: true, kind: 'discord-send-failed', channelId: failure.channelId, reason: failure.reason, textLen: failure.textLen, speaker } as MessageMetadata, + { + system: true, kind: 'discord-send-failed', channelId: failure.channelId, speaker, + ...(reason !== undefined ? { reason } : {}), + ...(textLen !== undefined ? { textLen } : {}), + } as MessageMetadata, ); } catch (err) { console.error(`[route-failure] failed to tell the primary that ${speaker}'s reply to ${failure.channelId} failed:`, err); @@ -12657,6 +12788,7 @@ export class AgentFramework { this.passFailedReplyToPrimary(conversationId, metadata); return; } + // A registered speaker's notice is addressed to it (noticeTargetFor). this.addMessage( 'user', [{ @@ -12664,7 +12796,7 @@ export class AgentFramework { text: `[discord-send-failed] Your previous reply (${textLen} chars) could not be delivered to ${this.channelForNotice(channelId)} (${reason}). It was saved to your archive but the human did not receive it.`, }], metadata, - this.noticeTargetFor(conversationId), + { forAgent: conversationId }, ); } catch (err) { console.error('onRouteFailure: failed to record send-failure marker:', err); diff --git a/test/route-failure-targeting.test.ts b/test/route-failure-targeting.test.ts index a28c3db1..41ebe724 100644 --- a/test/route-failure-targeting.test.ts +++ b/test/route-failure-targeting.test.ts @@ -7,7 +7,8 @@ import { AgentFramework } from '../src/index.js'; import { ContextManager, PassthroughStrategy } from '@animalabs/context-manager'; import { CapabilityGrant } from '../src/mcpl/capability-grant.js'; import type { McplServerRegistry } from '../src/mcpl/server-registry.js'; -import type { ProcessEvent } from '../src/index.js'; +import type { ProcessEvent, Module, ModuleContext, ToolDefinition, ToolCall, ToolResult, EventResponse } from '../src/index.js'; +import type { ContentBlock } from '@animalabs/membrane'; import type { ChannelRegistry } from '../src/mcpl/channel-registry.js'; import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; @@ -27,9 +28,39 @@ function internals(framework: AgentFramework) { unackedDeferredWrites: unknown[]; flushDeferredWrites(label: string): Promise; injectScriptWake(record: { id: string; agentName: string }, envelope: string): void; + closingConversationAgents: Set; + activeStreams: Map>; + recordChannelAutoOpenNotice(agentName: string | undefined, channels: Array<{ channelId: string; label?: string }>, cause: 'subscription-policy' | 'opened-by-delivery' | 'opened-by-reply'): void; + addMessage(participant: string, content: ContentBlock[], metadata?: Record, opts?: { forAgent?: string }): string; }; } +/** A tool whose call makes `speaker`'s reply fail to reach a channel while its + * turn is alive, and a tool that does nothing; either can quiesce the host + * from inside the turn. */ +class SenderModule implements Module { + readonly name = 'sender'; + framework: AgentFramework | null = null; + speaker = 'worker'; + quiesceOnCall = false; + async start(_ctx: ModuleContext): Promise {} + async stop(): Promise {} + getTools(): ToolDefinition[] { + return [ + { name: 'post', description: 'Post a reply', inputSchema: { type: 'object', properties: {} } }, + { name: 'noop', description: 'Do nothing', inputSchema: { type: 'object', properties: {} } }, + ]; + } + async handleToolCall(call: ToolCall): Promise { + if (call.name.endsWith('post')) { + await internals(this.framework!).channelRegistry.routeSpeech(this.speaker, 'mid-turn reply', 'missing-channel'); + } + if (this.quiesceOnCall) internals(this.framework!).quiesced = true; + return { success: true, data: { ok: true } }; + } + async onProcess(): Promise { return {}; } +} + const SEND_FAILED = '[discord-send-failed]'; const BOUNCED = 'was not delivered —'; const count = (messages: Array<{ content: unknown }>, needle: string) => @@ -103,24 +134,25 @@ describe('notices about an agent\'s own action', () => { } describe('residents share one message slot', () => { - it('keeps a secondary resident\'s notice on the default path, where it reads it', async (t) => { + it('addresses a secondary resident\'s notice to it, in the slot every resident reads', async (t) => { const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); const secondary = t.mock.method(framework.getAgent('secondary')!.getContextManager(), 'addMessage'); const result = await internals(framework).channelRegistry.routeSpeech('secondary', 'hello', 'missing-channel'); assert.equal(result, null); - assert.equal(primary.mock.callCount(), 1, 'the default path writes the shared slot'); - assert.equal(secondary.mock.callCount(), 0); - const [participant, content, metadata] = primary.mock.calls[0]!.arguments; + assert.equal(primary.mock.callCount(), 0); + assert.equal(secondary.mock.callCount(), 1, 'written through the speaker\'s own window'); + const [participant, content, metadata] = secondary.mock.calls[0]!.arguments; assert.equal(participant, 'user'); - assert.match(JSON.stringify(content), /\[discord-send-failed\].*5 chars.*missing-channel/); + assert.match(JSON.stringify(content), /\[discord-send-failed\] Your previous reply \(5 chars\) could not be delivered to missing-channel/); assert.equal(metadata?.system, true); assert.equal(metadata?.kind, 'discord-send-failed'); assert.equal(metadata?.channelId, 'missing-channel'); assert.match(String(metadata?.reason), /no registered channel/); + assert.equal(metadata?.textLen, 5); t.mock.restoreAll(); const { messages } = await framework.getAgent('secondary')!.getContextManager().compile(); - assert.equal(count(messages, SEND_FAILED), 1, 'the speaking resident reads the shared slot'); + assert.equal(count(messages, SEND_FAILED), 1, 'the speaking resident reads it'); assert.equal(internals(framework).pendingRequests.length, 0, 'a failure notice must not request inference'); assert.equal(membrane.calls.length, 0); }); @@ -134,30 +166,32 @@ describe('notices about an agent\'s own action', () => { assert.equal(primary.mock.callCount(), 1); assert.equal(secondary.mock.callCount(), 0); assert.equal(primary.mock.calls[0]!.arguments[2]?.kind, 'discord-send-failed'); + assert.match(JSON.stringify(primary.mock.calls[0]!.arguments[1]), /Your previous reply \(5 chars\) could not be delivered to the channel/); assert.equal(internals(framework).pendingRequests.length, 0); }); - it('waits for the primary\'s live turn before writing a secondary resident\'s notice', async () => { + it('waits for the speaking resident\'s own live turn', async () => { const state = internals(framework); - state.activeTurnTokens.set('primary', 1); + state.activeTurnTokens.set('secondary', 1); await state.channelRegistry.routeSpeech('secondary', 'hello', null); - assert.equal(state.deferredMessages.length, 1, 'the shared slot is mid-turn for the primary'); - assert.equal(state.deferredMessages[0]!.forAgent, undefined); - const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); + assert.equal(state.deferredMessages.length, 1, 'the speaker is mid-turn'); + assert.equal(state.deferredMessages[0]!.forAgent, 'secondary'); + const { messages } = await framework.getAgent('secondary')!.getContextManager().compile(); assert.equal(count(messages, SEND_FAILED), 0); }); - for (const [speaker, cycling] of [['secondary', 'primary'], ['primary', 'secondary']] as const) { - it(`waits for ${cycling}'s tool cycle before writing ${speaker}'s notice`, async () => { + for (const speaker of ['secondary', 'primary'] as const) { + it(`waits for ${speaker}'s own tool cycle before writing its notice`, async () => { const state = internals(framework); - state.pendingAssistantBlocks.set(cycling, []); + state.pendingAssistantBlocks.set(speaker, []); await state.channelRegistry.routeSpeech(speaker, 'hello', null); - assert.equal(state.deferredMessages.length, 1, 'a write now would split a tool_use from its result'); - const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); + assert.equal(state.deferredMessages.length, 1, 'a write now would split its tool_use from its result'); + assert.equal(state.deferredMessages[0]!.forAgent, speaker); + const { messages } = await framework.getAgent(speaker)!.getContextManager().compile(); assert.equal(count(messages, SEND_FAILED), 0); }); } @@ -322,15 +356,20 @@ describe('notices about an agent\'s own action', () => { assert.equal(internals(framework).deferredMessages.length, 0); }); - it('drops, logged, a background-script wake for an agent no longer registered', async (t) => { + it('drops, logged once, a background-script wake for an agent no longer registered: no write, trace or wake', async (t) => { const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); const errors: string[] = []; t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + const traces: string[] = []; + framework.onTrace((event) => { if (event.type === 'message:added') traces.push(JSON.stringify(event)); }); internals(framework).injectScriptWake({ id: 's-2', agentName: 'released-run' }, '[background script s-2] Your background script DIED.'); assert.equal(primary.mock.callCount(), 0, 'the primary started no such script'); - assert.ok(errors.some((e) => /"released-run" is not a registered agent — message dropped/.test(e))); + assert.deepEqual(errors, ['[notice] released-run is no longer registered, so its background-script s-2 wake notice is dropped']); + assert.deepEqual(traces, [], 'nothing was queued or stored'); + assert.equal(internals(framework).deferredMessages.length, 0); + assert.equal(internals(framework).pendingRequests.length, 0, 'no wake for an agent that is gone'); }); it('wakes a fork\'s background script into the fork\'s own window', async () => { @@ -345,6 +384,187 @@ describe('notices about an agent\'s own action', () => { assert.equal(count(primaryMessages, envelope), 0); }); + /** Run the fallback reaper over a closing fork, as a minute's sweep would. */ + function reap(forkName: string) { + const state = internals(framework); + state.closingConversationAgents.add(forkName); + state.lastConversationSweep = 0; + state.sweepExpiredConversations(); + } + + it('tells the primary, once, of a failure notice a fork stored but never read, when the fork is released', async () => { + const reply = 'Reply from the fork.'; + const fork = await spawnFork(reply); + // The notice landed in the fork's window after its turn: no compile has read it. + + reap(fork.name); + + assert.equal(framework.getAgent(fork.name), null, 'released with no closure turn'); + const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); + const told = messages.filter((m) => JSON.stringify(m.content).includes(SEND_FAILED)); + assert.equal(told.length, 1); + assert.match(JSON.stringify(told[0]!.content), + new RegExp(`A reply by ${fork.name} \\(${reply.length} chars\\) could not be delivered to missing-channel \\(no registered channel[^)]*\\)\\. ${fork.name} has ended, and the human did not receive it\\.`)); + }); + + it('passes on only what a fork never read: a notice its later turn compiled stays its own', async () => { + const fork = await spawnFork('Reply from the fork.'); + const second = 'A second reply, which fails as well.'; + membrane.pushResponse(createMockResponse([{ type: 'text', text: second }])); + framework.pushEvent({ + type: 'mcpl:channel-incoming', + serverId: 'srv', + channelId: 'missing-channel', + messageId: 'incoming-2', + author: { id: 'user-1', name: 'User' }, + content: [{ type: 'text', text: 'Hello again.' }], + timestamp: new Date().toISOString(), + metadata: { channel_type: 'im' }, + triggerInference: true, + } as unknown as ProcessEvent); + await framework.runUntilIdle(); + assert.equal(membrane.calls.length, 2, 'the fork\'s second turn compiled its first notice'); + + reap(fork.name); + + const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); + const told = messages.filter((m) => JSON.stringify(m.content).includes(SEND_FAILED)); + assert.equal(told.length, 1); + assert.match(JSON.stringify(told[0]!.content), new RegExp(`A reply by ${fork.name} \\(${second.length} chars\\)`)); + }); + + it('does not reap a closing fork whose turn holds its token before its stream registers', async () => { + const fork = await spawnFork(); + const state = internals(framework); + state.activeTurnTokens.set(fork.name, 7); + + reap(fork.name); + assert.equal(framework.getAgent(fork.name), fork, 'a turn mid-compile is busy'); + + state.activeTurnTokens.delete(fork.name); + reap(fork.name); + assert.equal(framework.getAgent(fork.name), null); + }); + + it('lands a queued ordinary write in a released fork\'s window, with its durable id, and acks it', async () => { + const fork = await spawnFork(); + const state = internals(framework); + state.deferredWritesPersisted = true; + state.activeTurnTokens.set(fork.name, 3); + state.addMessage('user', [{ type: 'text', text: 'queued for the fork' }], { kind: 'test-note' }, { forAgent: fork.name }); + assert.equal(state.deferredMessages.length, 1); + state.activeTurnTokens.delete(fork.name); + + reap(fork.name); + + assert.equal(framework.getAgent(fork.name), null); + assert.equal(state.deferredMessages.length, 0); + assert.equal(state.unackedDeferredWrites.length, 0, 'acked out of the recovery queue'); + const archive = await ContextManager.open({ + store: framework.getStore(), namespace: 'conversations/' + fork.name, isolate: true, strategy: new PassthroughStrategy(), + }); + const stored = archive.getAllMessages().find((m) => JSON.stringify(m.content).includes('queued for the fork')); + assert.ok(stored, 'it landed in the fork\'s own window'); + assert.equal(typeof (stored.metadata as { deferredWriteId?: unknown } | undefined)?.deferredWriteId, 'string'); + }); + + for (const [what, hold] of [ + ['its own tool cycle', (state: ReturnType, fork: string) => { state.pendingAssistantBlocks.set(fork, []); }], + ['its own stream', (state: ReturnType, fork: string) => { state.activeStreams.set(fork, new Promise(() => {})); }], + ] as const) { + it(`defers a fork's notice while only ${what} is alive`, async (t) => { + const fork = await spawnFork(); + const state = internals(framework); + const forkCm = t.mock.method(fork.getContextManager(), 'addMessage'); + hold(state, fork.name); + try { + await state.channelRegistry.routeSpeech(fork.name, 'hello', null); + + assert.equal(forkCm.mock.callCount(), 0); + assert.equal(state.deferredMessages.length, 1); + assert.equal(state.deferredMessages[0]!.forAgent, fork.name); + } finally { + // A stream that never settles would hold stop() in afterEach. + state.activeStreams.delete(fork.name); + } + }); + } + + it('lands a fork\'s notice while only another agent\'s tool cycle is pending', async (t) => { + const fork = await spawnFork(); + const state = internals(framework); + const forkCm = t.mock.method(fork.getContextManager(), 'addMessage'); + state.pendingAssistantBlocks.set('primary', []); + + await state.channelRegistry.routeSpeech(fork.name, 'hello', null); + + assert.equal(forkCm.mock.callCount(), 1, 'the fork\'s window has no cycle in flight'); + assert.equal(state.deferredMessages.length, 0); + }); + + it('tells the agent whose send opened a channel, in its own window; a policy open tells the primary', async (t) => { + const fork = await spawnFork(); + const state = internals(framework); + const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); + const forkCm = t.mock.method(fork.getContextManager(), 'addMessage'); + + state.recordChannelAutoOpenNotice(fork.name, [{ channelId: 'c-1', label: '#one' }], 'opened-by-reply'); + assert.equal(forkCm.mock.callCount(), 1); + assert.equal(primary.mock.callCount(), 0); + assert.match(JSON.stringify(forkCm.mock.calls[0]!.arguments[1]), /Now open: #one \(c-1\) — your own message into the closed channel engaged it/); + + state.recordChannelAutoOpenNotice(undefined, [{ channelId: 'c-2' }], 'subscription-policy'); + assert.equal(primary.mock.callCount(), 1, 'no agent caused a policy open'); + assert.equal(forkCm.mock.callCount(), 1); + }); + + it('drops, logged once, an auto-open notice for an agent no longer registered', async (t) => { + const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + + internals(framework).recordChannelAutoOpenNotice('released-conversation', [{ channelId: 'c-1' }], 'opened-by-delivery'); + + assert.equal(primary.mock.callCount(), 0); + assert.equal(internals(framework).deferredMessages.length, 0); + assert.deepEqual(errors, ['[notice] released-conversation is no longer registered, so its channel auto-open notice is dropped']); + }); + + it('passes a restored failure notice whose agent is gone to the primary, by name, and drops its other writes, logged', async (t) => { + const state = internals(framework); + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + (state.deferredMessages as unknown as Array>).push( + { + id: 'w-1', seq: 1, participant: 'user', forAgent: 'conversation-gone', + content: [{ type: 'text', text: '[discord-send-failed] Your previous reply (12 chars) could not be delivered.' }], + metadata: { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', reason: 'no registered channel', textLen: 12 }, + }, + { + // Queued by an earlier build: no textLen, and no reason either. + id: 'w-2', seq: 2, participant: 'user', forAgent: 'conversation-older', + content: [{ type: 'text', text: '[discord-send-failed] Your previous reply could not be delivered.' }], + metadata: { system: true, kind: 'discord-send-failed', channelId: 'missing-channel' }, + }, + { + id: 'w-3', seq: 3, participant: 'user', forAgent: 'conversation-gone', + content: [{ type: 'text', text: 'an ordinary write' }], + }, + ); + + await state.flushDeferredWrites('boot-recovery'); + + const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); + const told = messages.filter((m) => JSON.stringify(m.content).includes('has ended')).map((m) => JSON.stringify(m.content)); + assert.equal(told.length, 2); + assert.match(told[0]!, /A reply by conversation-gone \(12 chars\) could not be delivered to missing-channel \(no registered channel\)\. conversation-gone has ended/); + assert.match(told[1]!, /A reply by conversation-older could not be delivered to missing-channel\. conversation-older has ended/); + assert.ok(!told.some((t) => t.includes('undefined'))); + assert.equal(count(messages, 'an ordinary write'), 0); + assert.ok(errors.some((e) => /"conversation-gone" is not a registered agent — message dropped \(participant=user\)/.test(e))); + assert.equal(state.deferredMessages.length, 0); + }); + for (const condition of ['ordinary', 'other-tool-cycle', 'quiesced', 'write-fails'] as const) { it('handles a failed TTL-closure reply notice at disposal: ' + condition, async (t) => { const state = internals(framework); @@ -398,15 +618,18 @@ describe('notices about an agent\'s own action', () => { await framework.runUntilIdle(); state.pendingAssistantBlocks.clear(); - assert.equal(framework.getAgent(fork.name), null, 'the fork is disposed when its closure turn ends'); if (condition === 'quiesced') { - // A write while quiesced could land mid-surgery: the archive write - // is dropped, logged, and the primary's notice waits for resume. - assert.ok(errors.some((e) => e.includes(`[deferred-flush] ${fork.name}: released while quiesced, so a queued write for its window is dropped`))); + // A write while quiesced could land mid-surgery, so the release + // waits: the fork stays registered with its notice queued, and the + // primary hasn't been told. + assert.equal(framework.getAgent(fork.name), fork, 'released only once the host resumes'); assert.equal(state.deferredMessages.length, 1); - state.quiesced = false; - await state.flushDeferredWrites('closure test resume'); + assert.equal(state.deferredMessages[0]!.forAgent, fork.name); + const waiting = await framework.getAgent('primary')!.getContextManager().compile(); + assert.equal(count(waiting.messages, SEND_FAILED), 0); + await framework.resume({ force: true }); } + assert.equal(framework.getAgent(fork.name), null, 'the fork is disposed when its closure turn ends'); if (condition === 'other-tool-cycle') { // The primary's notice waits for the shared slot's tool cycle. assert.equal(state.deferredMessages.length, 1); @@ -423,9 +646,7 @@ describe('notices about an agent\'s own action', () => { const { messages } = await archive.compile(); const replyIndex = messages.findIndex((m) => m.participant === fork.name && JSON.stringify(m.content).includes(finalReply)); assert.ok(replyIndex >= 0, 'the failed closure reply remains archived'); - if (condition === 'quiesced') { - assert.equal(count(messages, SEND_FAILED), 0); - } else if (condition === 'write-fails') { + if (condition === 'write-fails') { // A write the store refuses is logged; disposal still completes. assert.equal(count(messages, SEND_FAILED), 0); assert.ok(errors.some((e) => e.includes(`[deferred-flush] ${fork.name}: failed to store a queued write before release`))); @@ -449,4 +670,103 @@ describe('notices about an agent\'s own action', () => { }); } }); + + describe('a run\'s notices at its tool boundaries, and its release', () => { + let runDir: string; + let runFramework: AgentFramework; + let runMembrane: MockMembrane; + let sender: SenderModule; + + beforeEach(async () => { + runDir = mkdtempSync(join(tmpdir(), 'route-failure-runs-')); + runMembrane = new MockMembrane(); + sender = new SenderModule(); + runFramework = await AgentFramework.create({ + storePath: join(runDir, 'test.chronicle'), + membrane: runMembrane.asMembrane(), + agents: [{ name: 'primary', model: 'test-model', systemPrompt: 'Primary.', proseRouting: 'locus' }], + modules: [sender], + }); + sender.framework = runFramework; + await internals(runFramework).initializeMcpl([]); + }); + + afterEach(async () => { + internals(runFramework).quiesced = false; + await runFramework.stop(); + rmSync(runDir, { recursive: true, force: true }); + }); + + /** Run 'worker' once through these rounds; every round but the last ends on its tool call. */ + async function run(rounds: ContentBlock[][]) { + const created = await runFramework.createEphemeralAgent({ + name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', proseRouting: 'explicit', + }); + created.contextManager.addMessage('user', [{ type: 'text', text: 'Run once.' }]); + rounds.forEach((blocks, i) => runMembrane.pushResponse(createMockResponse(blocks, i < rounds.length - 1 ? 'tool_use' : undefined))); + const done = runFramework.runEphemeralToCompletion(created.agent, created.contextManager); + runFramework.start(); + await done; + await runFramework.runUntilIdle(); + } + + async function archived() { + const archive = await ContextManager.open({ + store: runFramework.getStore(), namespace: 'subagent/worker', isolate: true, strategy: new PassthroughStrategy(), + }); + return (await archive.compile()).messages; + } + + const injected = () => runMembrane.lastStream!.receivedToolResultOptions + .flatMap((o) => o?.injectedMessages ?? []) + .map((m) => JSON.stringify(m.content)); + + it('passes on nothing a run read live at a tool boundary', async () => { + await run([ + [{ type: 'tool_use', id: 'p1', name: 'sender--post', input: {} }], + [{ type: 'text', text: '>>skip_reply done' }], + ]); + + assert.ok(injected().some((c) => c.includes(SEND_FAILED)), 'the notice reached the run\'s live stream'); + assert.equal(count(await archived(), SEND_FAILED), 1, 'and its window'); + const { messages } = await runFramework.getAgent('primary')!.getContextManager().compile(); + assert.equal(count(messages, 'A reply by worker'), 0, 'it read it, so no one else is told'); + }); + + it('injects a bounced prose notice at the next tool boundary, after the prose it bounced', async () => { + const prose = 'Worker prose with no destination.'; + await run([ + [{ type: 'text', text: prose }, { type: 'tool_use', id: 'n1', name: 'sender--noop', input: {} }], + [{ type: 'text', text: '>>skip_reply done' }], + ]); + + assert.ok(injected().some((c) => c.includes(BOUNCED)), 'the speaker hears it within the turn'); + const messages = await archived(); + const proseIndex = messages.findIndex((m) => m.participant === 'worker' && JSON.stringify(m.content).includes(prose)); + const noticeIndex = messages.findIndex((m) => JSON.stringify(m.content).includes(BOUNCED)); + assert.ok(proseIndex >= 0, 'the bounced prose is in its window'); + assert.ok(noticeIndex > proseIndex, 'the notice follows it there'); + assert.match(JSON.stringify(messages[noticeIndex]!.content), /\{\{unsent\}\}/, 'naming the clipboard that holds it'); + const { messages: primary } = await runFramework.getAgent('primary')!.getContextManager().compile(); + assert.equal(count(primary, BOUNCED), 0); + }); + + it('releases a run that ends while the host is quiesced only at resume, landing what it left queued', async () => { + sender.quiesceOnCall = true; + await run([ + [{ type: 'tool_use', id: 'n1', name: 'sender--noop', input: {} }], + [{ type: 'text', text: 'Worker prose that bounces at the end.' }], + ]); + const state = internals(runFramework); + assert.equal(state.quiesced, true); + assert.ok(runFramework.getAgent('worker'), 'still registered while quiesced'); + assert.ok(state.deferredMessages.some((m) => m.forAgent === 'worker'), 'its last notice is still queued'); + + await runFramework.resume({ force: true }); + + assert.equal(runFramework.getAgent('worker'), null, 'released at resume'); + assert.equal(state.deferredMessages.length, 0); + assert.equal(count(await archived(), BOUNCED), 1, 'what it left queued landed in its window'); + }); + }); });