From 26e472f9cd411ca972e0b1b5d27569eb3dc7bc51 Mon Sep 17 00:00:00 2001 From: shaggy Date: Wed, 12 Aug 2026 14:44:00 -0500 Subject: [PATCH] refactor(chat): route browser chat through one runtime Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01ESFAnh2t9HmLwng8oW95St --- .../agent/__tests__/session-ownership.test.ts | 67 +++++- .../src/chat/__tests__/chat-security.test.ts | 217 +++++++++++++++++- apps/gateway/src/chat/chat.gateway.ts | 97 +++++++- .../gateway/src/chat/embedded-chat.runtime.ts | 49 +++- .../plugin/discord-ingress.security.spec.ts | 196 +++++++++++++++- 5 files changed, 607 insertions(+), 19 deletions(-) diff --git a/apps/gateway/src/agent/__tests__/session-ownership.test.ts b/apps/gateway/src/agent/__tests__/session-ownership.test.ts index 2ebb1235..b7892791 100644 --- a/apps/gateway/src/agent/__tests__/session-ownership.test.ts +++ b/apps/gateway/src/agent/__tests__/session-ownership.test.ts @@ -280,11 +280,15 @@ function buildGatewayModule( ): Promise { const brain = { conversations: { - findById: vi.fn().mockResolvedValue(undefined), + // The sender OWNS this durable conversation, so the browser-send admission gate lets the turn + // reach the router seam. Foreignness is asserted downstream at the in-memory agent session + // (getSession({USER_B}) -> undefined), not at durable admission — the admission-rejection + // property has its own dedicated coverage. + findById: vi.fn().mockResolvedValue({ id: CONVERSATION_ID, userId: USER_B.id }), create: vi.fn().mockResolvedValue(undefined), update: vi.fn().mockResolvedValue(undefined), findMessages: vi.fn().mockResolvedValue([]), - addMessage: vi.fn().mockResolvedValue(undefined), + addMessage: vi.fn().mockResolvedValue({ id: 'persisted-turn' }), }, }; return Test.createTestingModule({ @@ -706,6 +710,26 @@ function makeLeaseAgentService() { return { svc, unsubscribe, session }; } +/** + * getSession → a live owned session (REST resolveOrCreate succeeds), onEvent returns a `detach` + * spy, and `prompt` REJECTS with a non-timeout error. Drives the REST-turn catch path so the single + * idempotent teardown must clear the 120s timeout and detach the listener exactly once. + */ +function makeRejectingPromptAgentService() { + const session = makeAgentSession(USER_A); + const detach = vi.fn(); + const svc = { + getSession: vi.fn(() => session), + createSession: vi.fn(), + onEvent: vi.fn(() => detach), + addChannel: vi.fn(), + removeChannel: vi.fn(), + prompt: vi.fn().mockRejectedValue(new Error('agent backend exploded')), + recordTokenUsage: vi.fn(), + }; + return { svc, detach }; +} + describe('TESS Task-5 embedded ownership collapse (missing and foreign are indistinguishable, never throw)', () => { const ctx = ownConversation(CONVERSATION_ID, EMBEDDED_SCOPE); @@ -801,3 +825,42 @@ describe('TESS Task-5 embedded socket lease lifecycle (one-shot dispatch, idempo expect(unsubscribe).toHaveBeenCalledTimes(1); }); }); + +describe('TESS Task-5 embedded REST turn teardown (a prompt rejection frees the timer + listener exactly once)', () => { + const ctx = ownConversation(CONVERSATION_ID, EMBEDDED_SCOPE); + + it('clears the 120s timeout and detaches the listener exactly once when prompt() rejects, leaving no timer to reject the abandoned done-promise later (Task 5 finding 6)', async () => { + const { svc, detach } = makeRejectingPromptAgentService(); + const runtime = new EmbeddedChatRuntime(svc as never); + + // A rejected `done` promise firing after completeLegacyRestTurn has already returned would + // surface as an unhandledRejection — the leak this test fences. Capture any that escape. + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown): void => { + unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandled); + vi.useFakeTimers(); + try { + const result = await runtime.completeLegacyRestTurn(ctx, { + content: 'trigger a backend failure', + }); + + // The rejection collapses to a total safe failure (not a timeout) — never throws out of the port. + expect(result).toEqual({ ok: false, code: 'operation_failed', retryable: false }); + // The single idempotent dispose ran in the catch: listener detached exactly once. + expect(detach).toHaveBeenCalledTimes(1); + + // dispose() cleared the REST timeout, so advancing far past it (120s) fires nothing: no second + // detach, and — the actual leak — no live timer left to reject the now-abandoned `done` promise. + vi.advanceTimersByTime(600_000); + expect(detach).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + // Let any scheduled rejection surface on a real macrotask, then confirm none did. + await new Promise((resolve) => setTimeout(resolve, 0)); + process.off('unhandledRejection', onUnhandled); + expect(unhandled).toHaveLength(0); + }); +}); diff --git a/apps/gateway/src/chat/__tests__/chat-security.test.ts b/apps/gateway/src/chat/__tests__/chat-security.test.ts index 7343576f..eba0f671 100644 --- a/apps/gateway/src/chat/__tests__/chat-security.test.ts +++ b/apps/gateway/src/chat/__tests__/chat-security.test.ts @@ -683,8 +683,9 @@ describe('Non-Discord ("Telegram-equivalent") socket ingress rejection (Task Fiv findMessages: vi.fn().mockResolvedValue([]), create: vi.fn().mockResolvedValue(undefined), update: vi.fn().mockResolvedValue(undefined), - addMessage: vi.fn().mockImplementation(async (): Promise => { + addMessage: vi.fn().mockImplementation(async (): Promise<{ id: string }> => { order.push('persist'); + return { id: 'message-order-1' }; }), }, }; @@ -794,6 +795,220 @@ describe('Non-Discord ("Telegram-equivalent") socket ingress rejection (Task Fiv ); expect(harnessConversations.append).not.toHaveBeenCalled(); }); + + it('drops a legacy browser turn when persistence RESOLVES nullish (not only on rejection): zero info/ack/dispatch/prompt, one clean disposal (Task 5 finding 2)', async () => { + // Companion to the rejected-promise case above. A brain adapter that resolves `undefined`/`null` + // instead of throwing must be treated as a persistence FAILURE, never as a saved message — the + // pre-fix code accepted a nullish resolve and dispatched a turn whose user message was never + // durably stored. RED before finding-2: the turn acks + dispatches + prompts on a phantom persist. + process.env['CHAT_HARNESS_RUNTIME'] = 'legacy'; + const order: string[] = []; + const counters = { dispatch: 0, dispose: 0 }; + const unsub = vi.fn(); + const session = { + provider: 'configured-provider', + modelId: 'configured-model', + piSession: { + thinkingLevel: 'medium', + getAvailableThinkingLevels: (): string[] => ['medium'], + }, + }; + const agentService = { + getSession: vi.fn().mockReturnValue(undefined), + createSession: vi.fn().mockResolvedValue(session), + recordMessage: vi.fn(), + onEvent: vi.fn().mockReturnValue(unsub), + addChannel: vi.fn(), + removeChannel: vi.fn(), + prompt: vi.fn().mockResolvedValue(undefined), + }; + const harnessConversations = { append: vi.fn() }; + const brain = { + conversations: { + findById: vi.fn().mockResolvedValue({ id: 'conversation-nullish-1' }), + findMessages: vi.fn().mockResolvedValue([]), + create: vi.fn().mockResolvedValue(undefined), + update: vi.fn().mockResolvedValue(undefined), + // Resolves nullish rather than throwing: the fix must still fail the turn closed. + addMessage: vi.fn().mockResolvedValue(undefined), + }, + }; + const router = readyRouter('legacy', agentService, harnessConversations); + instrumentLease(router, order, counters); + const gateway = new ChatGateway( + router as never, + { api: { getSession: vi.fn() } } as never, + brain as never, + {} as never, + {} as never, + {} as never, + ); + const client = { + id: 'browser-nullish-1', + data: { user: { id: 'user-a' } }, + emit: vi.fn(), + }; + + await gateway.handleMessage( + client as never, + { + conversationId: 'conversation-nullish-1', + content: 'persist resolves undefined', + } as never, + ); + + expect(client.emit).not.toHaveBeenCalledWith('session:info', expect.anything()); + expect(client.emit).not.toHaveBeenCalledWith('message:ack', expect.anything()); + expect(counters.dispatch).toBe(0); + expect(agentService.prompt).not.toHaveBeenCalled(); + expect(counters.dispose).toBe(1); + expect(unsub).toHaveBeenCalledTimes(1); + expect(client.emit).toHaveBeenCalledWith( + 'error', + expect.objectContaining({ code: 'persist_failed' }), + ); + expect(harnessConversations.append).not.toHaveBeenCalled(); + }); + + it("fails a legacy browser send closed when the supplied conversationId is not the sender's: conversation_unavailable, zero persist/mint/dispatch/prompt (Task 5 finding 1)", async () => { + // A browser socket that supplies a conversationId it does not own must be refused at admission, + // BEFORE any runtime effect. Pre-fix, an unresolved/foreign id fell through to session mint + + // dispatch, letting a caller attach to (or resurrect) a conversation outside their scope. + process.env['CHAT_HARNESS_RUNTIME'] = 'legacy'; + const order: string[] = []; + const counters = { dispatch: 0, dispose: 0 }; + const agentService = { + getSession: vi.fn().mockReturnValue(undefined), + createSession: vi.fn().mockResolvedValue(undefined), + recordMessage: vi.fn(), + onEvent: vi.fn().mockReturnValue((): void => undefined), + addChannel: vi.fn(), + removeChannel: vi.fn(), + prompt: vi.fn().mockResolvedValue(undefined), + }; + const harnessConversations = { append: vi.fn() }; + const addMessage = vi.fn().mockResolvedValue({ id: 'must-not-persist' }); + const brain = { + conversations: { + // Scoped lookup: the sender does not own this id, so admission resolves undefined. + findById: vi.fn().mockResolvedValue(undefined), + findMessages: vi.fn().mockResolvedValue([]), + create: vi.fn().mockResolvedValue(undefined), + update: vi.fn().mockResolvedValue(undefined), + addMessage, + }, + }; + const router = readyRouter('legacy', agentService, harnessConversations); + instrumentLease(router, order, counters); + const gateway = new ChatGateway( + router as never, + { api: { getSession: vi.fn() } } as never, + brain as never, + {} as never, + {} as never, + {} as never, + ); + const client = { + id: 'browser-foreign-1', + data: { user: { id: 'user-a' } }, + emit: vi.fn(), + }; + + await gateway.handleMessage( + client as never, + { + conversationId: 'conversation-foreign-1', + content: 'let me into a conversation I do not own', + } as never, + ); + + // Admission consults the scoped durable record with the SENDER's id, then fails closed: no + // persist, no session mint, no lease dispatch/dispose, no prompt, no ack — only the typed refusal. + expect(brain.conversations.findById).toHaveBeenCalledWith('conversation-foreign-1', 'user-a'); + expect(addMessage).not.toHaveBeenCalled(); + expect(brain.conversations.create).not.toHaveBeenCalled(); + expect(agentService.createSession).not.toHaveBeenCalled(); + expect(agentService.onEvent).not.toHaveBeenCalled(); + expect(agentService.addChannel).not.toHaveBeenCalled(); + expect(agentService.prompt).not.toHaveBeenCalled(); + expect(counters.dispatch).toBe(0); + expect(counters.dispose).toBe(0); + expect(client.emit).not.toHaveBeenCalledWith('message:ack', expect.anything()); + expect(client.emit).toHaveBeenCalledWith( + 'error', + expect.objectContaining({ code: 'conversation_unavailable' }), + ); + expect(harnessConversations.append).not.toHaveBeenCalled(); + }); + + it('admits a legacy browser send with NO conversationId by minting a scoped durable record first, then dispatches once (Task 5 finding 1)', async () => { + // The distinct server-minted-new path: a client that omits conversationId is a brand-new + // conversation. Admission must CREATE the durable record (scoped to the sender) before the turn + // persists/dispatches, and must not consult the ownership lookup (there is nothing to authorize + // yet). This guards the fix from over-reaching and breaking new-conversation creation. + process.env['CHAT_HARNESS_RUNTIME'] = 'legacy'; + const order: string[] = []; + const counters = { dispatch: 0, dispose: 0 }; + const session = { + provider: 'configured-provider', + modelId: 'configured-model', + piSession: { + thinkingLevel: 'medium', + getAvailableThinkingLevels: (): string[] => ['medium'], + }, + }; + const agentService = { + getSession: vi.fn().mockReturnValue(undefined), + createSession: vi.fn().mockResolvedValue(session), + recordMessage: vi.fn(), + onEvent: vi.fn().mockReturnValue((): void => undefined), + addChannel: vi.fn(), + removeChannel: vi.fn(), + prompt: vi.fn().mockResolvedValue(undefined), + }; + const harnessConversations = { append: vi.fn() }; + const create = vi.fn().mockResolvedValue(undefined); + const brain = { + conversations: { + findById: vi.fn().mockResolvedValue(undefined), + findMessages: vi.fn().mockResolvedValue([]), + create, + update: vi.fn().mockResolvedValue(undefined), + addMessage: vi.fn().mockResolvedValue({ id: 'persisted-new' }), + }, + }; + const router = readyRouter('legacy', agentService, harnessConversations); + instrumentLease(router, order, counters); + const gateway = new ChatGateway( + router as never, + { api: { getSession: vi.fn() } } as never, + brain as never, + {} as never, + {} as never, + {} as never, + ); + const client = { + id: 'browser-new-1', + data: { user: { id: 'user-a' } }, + emit: vi.fn(), + }; + + await gateway.handleMessage( + client as never, + { + content: 'first message with no conversationId', + } as never, + ); + + // The durable record is minted for the sender (create with the same server id the turn persists + // under), and the admitted new-conversation turn dispatches exactly once. + expect(create).toHaveBeenCalledWith( + expect.objectContaining({ id: expect.any(String), userId: 'user-a' }), + ); + expect(counters.dispatch).toBe(1); + expect(agentService.prompt).toHaveBeenCalledTimes(1); + expect(harnessConversations.append).not.toHaveBeenCalled(); + }); }); describe('Chat DTO validation', () => { diff --git a/apps/gateway/src/chat/chat.gateway.ts b/apps/gateway/src/chat/chat.gateway.ts index a64950a3..c9fc1378 100644 --- a/apps/gateway/src/chat/chat.gateway.ts +++ b/apps/gateway/src/chat/chat.gateway.ts @@ -364,13 +364,32 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa return; } const data = rawData; - const conversationId = data.conversationId ?? uuid(); + const suppliedConversationId = data.conversationId; + const conversationId = suppliedConversationId ?? uuid(); const scope = this.getClientScope(client); if (!scope) { client.emit('error', { conversationId, error: 'Authenticated user scope is required.' }); return; } + // Durable ownership admission BEFORE any runtime/listener/channel effect (security fix, + // finding 1). A browser-supplied conversation id must resolve to THIS socket's own durable + // owner; a missing row and a row owned by another user both fail closed here, so runtime state + // is never allocated under an unowned conversation. A send that omits the id is the distinct + // server-minted-new path: the durable record is created first and a creation failure fails + // closed. Both rejections collapse to conversation_unavailable with zero downstream effects. + if ( + !(await this.admitBrowserConversation(suppliedConversationId, conversationId, scope.userId)) + ) { + client.emit('error', { + conversationId, + code: 'conversation_unavailable', + retryable: false, + error: 'That conversation is not available.', + }); + return; + } + this.logger.log(`Message from ${client.id} in conversation ${conversationId}`); // Dispose any prior turn on this exact channel BEFORE preparing the next: prepare re-adds the @@ -472,6 +491,31 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa }), ); + // Reconcile the verified binding against the durable agent record BEFORE claiming the message + // id (security fix, finding 3). The configured agent must exist and its record name must match + // the binding's instance id exactly; a missing record, a name mismatch, or a lookup error + // rejects the turn WITHOUT consuming the replay claim, so a corrected retry is still admitted. + // The branded identity is derived from the record (id + name), never from the raw binding + // strings — an unreconciled binding must not execute a default or different embedded agent + // under a verified label. + let configuredAgent: { readonly agentConfigId: string; readonly instanceId: string }; + try { + const record = await this.brain.agents.findById(binding.agentConfigId); + if (!record || record.name !== binding.instanceId) { + this.logger.warn( + `Rejected Discord ingress: configured agent not reconciled binding=${binding.agentConfigId} instance=${binding.instanceId}`, + ); + return; + } + configuredAgent = { agentConfigId: record.id, instanceId: record.name }; + } catch (err) { + this.logger.error( + `Discord configured-agent reconciliation failed binding=${binding.agentConfigId}`, + err instanceof Error ? err.stack : String(err), + ); + return; + } + // Atomic replay claim LAST — after the configured-identity, binding, forced-scope, and // attachment checks above have all passed, and immediately before the first effect (existing // session teardown + dispatch). An envelope rejected by any earlier gate consumes no claim, so @@ -496,12 +540,13 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa onEvent: (event: LegacyRuntimeEvent): void => this.relayEvent(client, conversationId, event), }; - // The configured agent comes from the verified binding, never from a brain lookup: the - // embedded runtime rechecks scope and mints/reuses the session under this exact identity. + // The configured agent is the record reconciled above (finding 3): the embedded runtime rechecks + // scope, refuses to reuse a same-scope session under a different identity, and mints under this + // exact reconciled identity. const context = verifyDiscordIngress({ conversationId, scope, - configuredAgent: { agentConfigId: binding.agentConfigId, instanceId: binding.instanceId }, + configuredAgent, content: ingress.content, ...(attachments && attachments.length > 0 ? { attachments } : {}), correlationId: ingress.correlationId, @@ -602,7 +647,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa if (!userId) return true; await this.ensureConversation(conversationId, userId); try { - await this.brain.conversations.addMessage( + const saved = await this.brain.conversations.addMessage( { conversationId, role: 'user', @@ -632,6 +677,16 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa }, userId, ); + // A nullish result is durable persistence failure, not success: `addMessage` returns + // undefined when the parent conversation is missing or owned by another user, inserting no + // row. Treat it exactly like a thrown error so the caller tears down the lease and surfaces + // `persist_failed` — never ack/dispatch/prompt on a turn whose user message was not stored. + if (saved === undefined || saved === null) { + this.logger.error( + `User message not persisted for conversation=${conversationId}: no durable record (missing or foreign conversation owner)`, + ); + return false; + } return true; } catch (err) { this.logger.error( @@ -1088,6 +1143,38 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa .filter((id: string): boolean => id.length > 0); } + /** + * Durable ownership admission for a browser send, run BEFORE any runtime/listener/channel effect + * (security fix, finding 1). + * + * A supplied conversation id must durably resolve to this socket's own user: `findById` scopes by + * owner, so a missing row and a row owned by another user both return undefined and admission + * fails (returns false) with zero runtime effect. A send that omits the id + * (`suppliedConversationId === undefined`) is the server-minted-new path — the durable record is + * created first and any creation failure fails closed. Never allocate runtime under a conversation + * this socket does not own or could not create. + */ + private async admitBrowserConversation( + suppliedConversationId: string | undefined, + conversationId: string, + userId: string, + ): Promise { + try { + if (suppliedConversationId !== undefined) { + const owned = await this.brain.conversations.findById(suppliedConversationId, userId); + return owned !== undefined; + } + await this.brain.conversations.create({ id: conversationId, userId }); + return true; + } catch (err) { + this.logger.error( + `Conversation admission failed for conversation=${conversationId}`, + err instanceof Error ? err.stack : String(err), + ); + return false; + } + } + private async ensureConversation(conversationId: string, userId: string): Promise { try { const existing = await this.brain.conversations.findById(conversationId, userId); diff --git a/apps/gateway/src/chat/embedded-chat.runtime.ts b/apps/gateway/src/chat/embedded-chat.runtime.ts index d8f69063..39018458 100644 --- a/apps/gateway/src/chat/embedded-chat.runtime.ts +++ b/apps/gateway/src/chat/embedded-chat.runtime.ts @@ -56,13 +56,27 @@ export class EmbeddedChatRuntime implements ChatRuntime, LegacyEmbeddedChatPort if (!resolved.ok) return resolved; let responseText = ''; + let timer: ReturnType | undefined; + let detach: (() => void) | undefined; + let disposed = false; + // One idempotent teardown owned OUTSIDE the completion promise: it clears the timeout and + // detaches the event listener exactly once, whichever of agent_end, timeout, or a prompt + // rejection fires first. Without this, a prompt() rejection surfaced through the catch below + // would return while leaving the listener attached (free to consume a later turn's events) and + // the 120s timer live (its rejection later going unobserved). + const dispose = (): void => { + if (disposed) return; + disposed = true; + if (timer !== undefined) clearTimeout(timer); + detach?.(); + }; const done = new Promise((resolve, reject) => { - const timer = setTimeout(() => { - cleanup(); + timer = setTimeout(() => { + dispose(); reject(new Error('Agent response timed out')); }, REST_TURN_TIMEOUT_MS); - const cleanup = this.agentService.onEvent( + detach = this.agentService.onEvent( conversationId, (event: AgentSessionEvent) => { if ( @@ -72,8 +86,7 @@ export class EmbeddedChatRuntime implements ChatRuntime, LegacyEmbeddedChatPort responseText += event.assistantMessageEvent.delta; } if (event.type === 'agent_end') { - clearTimeout(timer); - cleanup(); + dispose(); resolve(); } }, @@ -85,6 +98,7 @@ export class EmbeddedChatRuntime implements ChatRuntime, LegacyEmbeddedChatPort await this.agentService.prompt(conversationId, input.content, scope); await done; } catch (err) { + dispose(); const message = err instanceof Error ? err.message : String(err); if (message.includes('timed out')) { return { ok: false, code: 'timeout', retryable: true }; @@ -230,9 +244,15 @@ export class EmbeddedChatRuntime implements ChatRuntime, LegacyEmbeddedChatPort const scope = toScope(context.scope); const { conversationId } = context; - const resolved = await this.resolveOrCreate(conversationId, scope, { - agentConfigId: context.configuredAgent.agentConfigId, - }); + const resolved = await this.resolveOrCreate( + conversationId, + scope, + { agentConfigId: context.configuredAgent.agentConfigId }, + { + agentConfigId: context.configuredAgent.agentConfigId, + instanceId: context.configuredAgent.instanceId, + }, + ); if (!resolved.ok) return resolved; let detach: () => void; @@ -275,11 +295,24 @@ export class EmbeddedChatRuntime implements ChatRuntime, LegacyEmbeddedChatPort conversationId: string, scope: ActorTenantScope, extraOptions: Readonly<{ provider?: string; modelId?: string; agentConfigId?: string }>, + expectedAgent?: Readonly<{ agentConfigId: string; instanceId: string }>, ): Promise< | { readonly ok: true; readonly presentation: LegacySessionPresentation } | Exclude, { ok: true }> > { let session = this.agentService.getSession(conversationId, scope); + if (session && expectedAgent) { + // A verified-Discord turn reuses an existing in-process session only when its configured + // identity still matches the reconciled agent record. A pre-existing same-scope session + // carrying a different configured agent (config id or resolved name) is rejected rather than + // executed under the verified label — never silently run a different prompt/model/tool policy. + if ( + session.agentConfigId !== expectedAgent.agentConfigId || + session.agentName !== expectedAgent.instanceId + ) { + return CONVERSATION_UNAVAILABLE; + } + } if (!session) { try { session = await this.agentService.createSession(conversationId, { diff --git a/apps/gateway/src/plugin/discord-ingress.security.spec.ts b/apps/gateway/src/plugin/discord-ingress.security.spec.ts index 066e811d..9d6628ce 100644 --- a/apps/gateway/src/plugin/discord-ingress.security.spec.ts +++ b/apps/gateway/src/plugin/discord-ingress.security.spec.ts @@ -628,7 +628,7 @@ describe('Discord ingress security', () => { removeChannel: vi.fn(), prompt, }; - const addMessage = vi.fn().mockResolvedValue(undefined); + const addMessage = vi.fn().mockResolvedValue({ id: 'discord-persisted-message' }); const brain = { agents: { findById: vi.fn((id: string) => Promise.resolve({ id, name: 'Nova' })) }, conversations: { @@ -713,7 +713,7 @@ describe('Discord ingress security', () => { removeChannel: vi.fn(), prompt, }; - const addMessage = vi.fn().mockResolvedValue(undefined); + const addMessage = vi.fn().mockResolvedValue({ id: 'discord-persisted-message' }); const brain = { agents: { findById: vi.fn((id: string) => Promise.resolve({ id, name: 'Nova' })) }, conversations: { @@ -778,6 +778,191 @@ describe('Discord ingress security', () => { expect(harnessConversations.append).not.toHaveBeenCalled(); }); + it('a verified SEND whose configured agent record fails reconciliation consumes no replay claim, so a corrected byte-identical retry dispatches/persists/acks exactly once (Task 5 finding 3)', async () => { + configureDiscordEnv(); + process.env['CHAT_HARNESS_RUNTIME'] = 'pi-rpc'; + process.env['DISCORD_ALLOWED_CHANNEL_IDS'] = 'channel-001'; + process.env['DISCORD_INTERACTION_BINDINGS'] = JSON.stringify([ + { + instanceId: 'Nova', + agentConfigId: 'agent-config-nova', + guildId: 'guild-001', + channelId: 'channel-001', + pairedUsers: { + 'user-001': { role: 'operator', mosaicUserId: 'mosaic-operator-001' }, + }, + }, + ]); + const session = { + provider: 'configured-provider', + modelId: 'configured-model', + piSession: { + thinkingLevel: 'medium', + getAvailableThinkingLevels: (): string[] => ['medium'], + }, + }; + const createSession = vi.fn().mockResolvedValue(session); + const prompt = vi.fn().mockResolvedValue(undefined); + const agentService = { + getSession: vi.fn().mockReturnValue(undefined), + createSession, + recordMessage: vi.fn(), + onEvent: vi.fn().mockReturnValue((): void => undefined), + addChannel: vi.fn(), + removeChannel: vi.fn(), + prompt, + }; + const addMessage = vi.fn().mockResolvedValue({ id: 'discord-persisted-message' }); + // The durable agent record does not reconcile on the first delivery (its name no longer matches + // the verified binding's instance id), then reconciles cleanly on the corrected retry. + const findAgent = vi + .fn() + .mockResolvedValueOnce({ id: 'agent-config-nova', name: 'Renamed-Away' }) + .mockResolvedValue({ id: 'agent-config-nova', name: 'Nova' }); + const brain = { + agents: { findById: findAgent }, + conversations: { + findById: vi.fn().mockResolvedValue({ id: 'Nova:discord:channel-001' }), + findMessages: vi.fn().mockResolvedValue([]), + create: vi.fn().mockResolvedValue(undefined), + update: vi.fn().mockResolvedValue(undefined), + addMessage, + }, + }; + const harnessConversations = { append: vi.fn() }; + const gateway = new ChatGateway( + piRpcRouterFronting(agentService, harnessConversations) as never, + {} as never, + brain as never, + {} as never, + {} as never, + { resolve: vi.fn() } as never, + ); + const client = { + id: 'discord-client-reconcile', + data: { discordService: true }, + emit: vi.fn(), + }; + const ackCount = (): number => + client.emit.mock.calls.filter((call) => call[0] === 'message:ack').length; + + const envelope = ingressEnvelope( + 'verified once with stale agent record', + 'discord-reconcile-001', + { + conversationId: 'Nova:discord:channel-001', + }, + ); + + // (1) The configured-agent reconcile runs BEFORE the replay claim. A mismatch refuses the turn + // and, crucially, consumes no claim for discord-reconcile-001 — nothing dispatches. + await gateway.handleMessage(client as never, envelope); + expect(createSession).toHaveBeenCalledTimes(0); + expect(prompt).toHaveBeenCalledTimes(0); + expect(addMessage).toHaveBeenCalledTimes(0); + expect(ackCount()).toBe(0); + + // (2) The record now reconciles; because step (1) took no claim, this byte-identical retry claims + // once and runs the full embedded dispatch exactly once. (Pre-fix, the claim was spent ahead + // of the reconcile in step (1), so this retry was dropped as a replay — the RED this drives.) + await gateway.handleMessage(client as never, envelope); + expect(createSession).toHaveBeenCalledTimes(1); + expect(prompt).toHaveBeenCalledTimes(1); + expect(addMessage).toHaveBeenCalledTimes(1); + expect(ackCount()).toBe(1); + + // (3) A genuine duplicate after the committed turn stays fail-closed. + await gateway.handleMessage(client as never, envelope); + expect(createSession).toHaveBeenCalledTimes(1); + expect(prompt).toHaveBeenCalledTimes(1); + expect(addMessage).toHaveBeenCalledTimes(1); + expect(ackCount()).toBe(1); + expect(harnessConversations.append).not.toHaveBeenCalled(); + }); + + it('a verified SEND refuses to reuse a same-scope embedded session minted under a different configured identity, with zero prompt/persist/ack (Task 5 finding 3)', async () => { + configureDiscordEnv(); + process.env['CHAT_HARNESS_RUNTIME'] = 'pi-rpc'; + process.env['DISCORD_ALLOWED_CHANNEL_IDS'] = 'channel-001'; + process.env['DISCORD_INTERACTION_BINDINGS'] = JSON.stringify([ + { + instanceId: 'Nova', + agentConfigId: 'agent-config-nova', + guildId: 'guild-001', + channelId: 'channel-001', + pairedUsers: { + 'user-001': { role: 'operator', mosaicUserId: 'mosaic-operator-001' }, + }, + }, + ]); + // A live session already exists for this conversation/scope, but it was minted under a DIFFERENT + // configured agent (Orion). The verified binding reconciles to Nova, so reusing this session would + // execute one agent's turn under another agent's verified label — the reuse guard must refuse it. + const foreignIdentitySession = { + provider: 'configured-provider', + modelId: 'configured-model', + agentConfigId: 'agent-config-orion', + agentName: 'Orion', + piSession: { + thinkingLevel: 'medium', + getAvailableThinkingLevels: (): string[] => ['medium'], + }, + }; + const prompt = vi.fn().mockResolvedValue(undefined); + const createSession = vi.fn().mockResolvedValue(foreignIdentitySession); + const agentService = { + getSession: vi.fn().mockReturnValue(foreignIdentitySession), + createSession, + recordMessage: vi.fn(), + onEvent: vi.fn().mockReturnValue((): void => undefined), + addChannel: vi.fn(), + removeChannel: vi.fn(), + prompt, + }; + const addMessage = vi.fn().mockResolvedValue({ id: 'discord-persisted-message' }); + const brain = { + agents: { findById: vi.fn((id: string) => Promise.resolve({ id, name: 'Nova' })) }, + conversations: { + findById: vi.fn().mockResolvedValue({ id: 'Nova:discord:channel-001' }), + findMessages: vi.fn().mockResolvedValue([]), + create: vi.fn().mockResolvedValue(undefined), + update: vi.fn().mockResolvedValue(undefined), + addMessage, + }, + }; + const harnessConversations = { append: vi.fn() }; + const gateway = new ChatGateway( + piRpcRouterFronting(agentService, harnessConversations) as never, + {} as never, + brain as never, + {} as never, + {} as never, + { resolve: vi.fn() } as never, + ); + const client = { + id: 'discord-client-identity-swap', + data: { discordService: true }, + emit: vi.fn(), + }; + + await gateway.handleMessage( + client as never, + ingressEnvelope('reuse under a different identity', 'discord-identity-swap-001', { + conversationId: 'Nova:discord:channel-001', + }), + ); + + // Refused at the embedded reuse guard: no prompt, no persist, no ack — only a typed refusal. + expect(prompt).not.toHaveBeenCalled(); + expect(addMessage).not.toHaveBeenCalled(); + expect(client.emit).not.toHaveBeenCalledWith('message:ack', expect.anything()); + expect(client.emit).toHaveBeenCalledWith( + 'error', + expect.objectContaining({ conversationId: 'Nova:discord:channel-001' }), + ); + expect(harnessConversations.append).not.toHaveBeenCalled(); + }); + it('retains validated persisted attachments in resumed conversation history', async () => { const attachment = { id: 'attachment-history', @@ -840,10 +1025,14 @@ describe('Discord ingress security', () => { configureDiscordEnv(); process.env['CHAT_HARNESS_RUNTIME'] = 'pi-rpc'; const prompt = vi.fn().mockResolvedValue(undefined); - const addMessage = vi.fn().mockResolvedValue(undefined); + const addMessage = vi.fn().mockResolvedValue({ id: 'discord-persisted-message' }); const session = { provider: 'test-provider', modelId: 'test-model', + // The reused embedded session carries the SAME reconciled identity as the verified binding, + // so the finding-3 session-reuse guard admits it rather than refusing an identity swap. + agentConfigId: 'agent-config-nova', + agentName: 'Nova', piSession: { thinkingLevel: 'medium', getAvailableThinkingLevels: (): string[] => ['medium'], @@ -857,6 +1046,7 @@ describe('Discord ingress security', () => { prompt, }; const brain = { + agents: { findById: vi.fn((id: string) => Promise.resolve({ id, name: 'Nova' })) }, conversations: { findById: vi.fn().mockResolvedValue({ id: 'Nova:discord:channel-001' }), create: vi.fn().mockResolvedValue(undefined),