diff --git a/apps/gateway/src/__tests__/conversation-persistence.test.ts b/apps/gateway/src/__tests__/conversation-persistence.test.ts index 781d2957..0907b155 100644 --- a/apps/gateway/src/__tests__/conversation-persistence.test.ts +++ b/apps/gateway/src/__tests__/conversation-persistence.test.ts @@ -417,7 +417,7 @@ describe('ConversationsController — search endpoint', () => { }, ]; brain = createMockBrain({ searchResults }); - controller = new ConversationsController(brain as never); + controller = new ConversationsController(brain as never, { runtimeMode: 'legacy' }); }); it('returns matching messages for a valid search query', async () => { @@ -479,7 +479,7 @@ describe('ConversationsController — search endpoint', () => { describe('ConversationsController — message CRUD', () => { it('listMessages returns 404 when conversation is not owned by user', async () => { const brain = createMockBrain({ conversation: undefined }); - const controller = new ConversationsController(brain as never); + const controller = new ConversationsController(brain as never, { runtimeMode: 'legacy' }); await expect(controller.listMessages(CONV_ID, { id: USER_ID })).rejects.toBeInstanceOf( NotFoundException, @@ -489,7 +489,7 @@ describe('ConversationsController — message CRUD', () => { it('listMessages returns the messages for an owned conversation', async () => { const msgs = [makeMessage('user', 'Test message'), makeMessage('assistant', 'Test reply')]; const brain = createMockBrain({ conversation: makeConversation(), messages: msgs }); - const controller = new ConversationsController(brain as never); + const controller = new ConversationsController(brain as never, { runtimeMode: 'legacy' }); const result = await controller.listMessages(CONV_ID, { id: USER_ID }); @@ -500,7 +500,7 @@ describe('ConversationsController — message CRUD', () => { it('addMessage returns the persisted message', async () => { const brain = createMockBrain({ conversation: makeConversation() }); - const controller = new ConversationsController(brain as never); + const controller = new ConversationsController(brain as never, { runtimeMode: 'legacy' }); const result = await controller.addMessage( CONV_ID, diff --git a/apps/gateway/src/__tests__/resource-ownership.test.ts b/apps/gateway/src/__tests__/resource-ownership.test.ts index 821a7a62..cd2bbacf 100644 --- a/apps/gateway/src/__tests__/resource-ownership.test.ts +++ b/apps/gateway/src/__tests__/resource-ownership.test.ts @@ -60,7 +60,7 @@ describe('Resource ownership checks', () => { // The repo enforces ownership via the WHERE clause; it returns undefined when the // conversation does not belong to the requesting user. brain.conversations.findById.mockResolvedValue(undefined); - const controller = new ConversationsController(brain as never); + const controller = new ConversationsController(brain as never, { runtimeMode: 'legacy' }); await expect(controller.findOne('conv-1', { id: 'user-1' })).rejects.toBeInstanceOf( NotFoundException, diff --git a/apps/gateway/src/chat/chat-runtime-router.spec.ts b/apps/gateway/src/chat/chat-runtime-router.spec.ts index 9c25e937..b5fb9660 100644 --- a/apps/gateway/src/chat/chat-runtime-router.spec.ts +++ b/apps/gateway/src/chat/chat-runtime-router.spec.ts @@ -19,8 +19,14 @@ import { import { ChatRuntimeRouter } from './chat-runtime-router.js'; import { ChatRuntimeUnavailableError, + ownConversation, type ChatRuntime, type ChatRuntimeMode, + type LegacyEmbeddedChatPort, + type LegacyRuntimeStream, + type LegacySessionPresentation, + type LegacySocketTurnLease, + type OwnedConversationContext, } from './chat-runtime.js'; import { AppModule } from '../app.module.js'; import { ProviderService } from '../agent/provider.service.js'; @@ -176,6 +182,253 @@ describe('ChatRuntimeRouter', () => { }); }); +/** + * Task Five, Step Three — legacy port operations fail closed under pi-rpc (direct valid-input). + * + * The unit suite above constructs the router but never invokes a legacy port operation, so the + * six per-operation inner `if (this.mode === 'pi-rpc')` guards are unexercised — a mutation that + * deletes one of them SURVIVES for lack of a test that drives that operation. This group closes + * that gap the right way: it drives each of the six operations DIRECTLY, in pi-rpc mode, with a + * valid branded {@link OwnedConversationContext} and valid input, against a recording embedded + * stub whose method returns a distinguishable `ok:true` success and increments a per-op counter. + * + * For each operation: + * - pi-rpc test asserts the exact frozen `{ ok:false, code:'runtime_unsupported', retryable:false }` + * result AND that the embedded stub was touched zero times (no effects); + * - the paired legacy test proves that same stub method IS reached and returns its distinguishable + * success when the mode does not refuse — so the pi-rpc zero-invocation assertion is meaningful, + * not vacuously true because the stub could never be called. + * + * Deleting ONLY one operation's inner guard makes THAT operation's pi-rpc test behaviorally RED + * (the router returns the embedded `ok:true` value and records the call), with every outer guard + * and the other five inner guards intact. `next` is untouched; nothing here changes production. + */ +describe('ChatRuntimeRouter — legacy port ops fail closed under pi-rpc (Task Five, Step Three)', () => { + const RUNTIME_UNSUPPORTED = { + ok: false, + code: 'runtime_unsupported', + retryable: false, + } as const; + + const PRESENTATION: LegacySessionPresentation = { + provider: 'embedded-provider', + modelId: 'embedded-model', + thinkingLevel: 'low', + availableThinkingLevels: ['low', 'high'], + }; + + const stream: LegacyRuntimeStream = { + channelId: 'websocket:test-socket', + onEvent: () => {}, + }; + + const ctx = (): OwnedConversationContext => + ownConversation('conversation-1', { userId: 'user-1', tenantId: 'tenant-1' }); + + /** + * Per-operation invocation counters with declared keys (not an index signature) so each + * `calls.` is definitely `number` under `noUncheckedIndexedAccess`. + */ + type LegacyPortCallCounts = { + completeLegacyRestTurn: number; + prepareLegacySocketTurn: number; + setLegacyThinking: number; + abortLegacyTurn: number; + applyLegacyModelOverride: number; + readLegacySessionPresentation: number; + dispatchVerifiedDiscordIngress: number; + }; + + /** + * An embedded port that records every invocation and returns a distinguishable `ok:true` + * value per operation. If a router op reaches it (its guard removed), both the recorded call + * count and the returned `ok:true` value diverge from the frozen `runtime_unsupported` result. + */ + function recordingEmbeddedPort(): { + port: ChatRuntime & LegacyEmbeddedChatPort; + calls: LegacyPortCallCounts; + } { + const calls: LegacyPortCallCounts = { + completeLegacyRestTurn: 0, + prepareLegacySocketTurn: 0, + setLegacyThinking: 0, + abortLegacyTurn: 0, + applyLegacyModelOverride: 0, + readLegacySessionPresentation: 0, + dispatchVerifiedDiscordIngress: 0, + }; + const lease: LegacySocketTurnLease = { + presentation: PRESENTATION, + dispatch: () => Promise.resolve({ ok: true, value: undefined }), + dispose: () => Promise.resolve(), + }; + const port: ChatRuntime & LegacyEmbeddedChatPort = { + kind: 'embedded', + completeLegacyRestTurn: () => { + calls.completeLegacyRestTurn += 1; + return Promise.resolve({ + ok: true, + value: { text: 'EMBEDDED-REST', presentation: PRESENTATION }, + }); + }, + prepareLegacySocketTurn: () => { + calls.prepareLegacySocketTurn += 1; + return Promise.resolve({ ok: true, value: lease }); + }, + setLegacyThinking: () => { + calls.setLegacyThinking += 1; + return { ok: true, value: PRESENTATION }; + }, + abortLegacyTurn: () => { + calls.abortLegacyTurn += 1; + return Promise.resolve({ ok: true, value: undefined }); + }, + applyLegacyModelOverride: () => { + calls.applyLegacyModelOverride += 1; + return { ok: true, value: PRESENTATION }; + }, + readLegacySessionPresentation: () => { + calls.readLegacySessionPresentation += 1; + return { ok: true, value: PRESENTATION }; + }, + dispatchVerifiedDiscordIngress: () => { + calls.dispatchVerifiedDiscordIngress += 1; + return Promise.resolve({ + ok: true, + value: { + presentation: PRESENTATION, + dispatch: () => Promise.resolve({ ok: true, value: undefined }), + dispose: () => Promise.resolve(), + }, + }); + }, + }; + return { port, calls }; + } + + function piRouter(port: ChatRuntime & LegacyEmbeddedChatPort): ChatRuntimeRouter { + return new ChatRuntimeRouter( + registryWith(['pi']), + boundConversationService, + port, + harness, + 'pi-rpc', + ); + } + function legacyRouter(port: ChatRuntime & LegacyEmbeddedChatPort): ChatRuntimeRouter { + return new ChatRuntimeRouter( + registryWith([]), + boundConversationService, + port, + harness, + 'legacy', + ); + } + + // completeLegacyRestTurn --------------------------------------------------- + it('completeLegacyRestTurn refuses with runtime_unsupported and never touches embedded under pi-rpc', async () => { + const { port, calls } = recordingEmbeddedPort(); + const result = await piRouter(port).completeLegacyRestTurn(ctx(), { content: 'hello' }); + expect(result).toEqual(RUNTIME_UNSUPPORTED); + expect(calls.completeLegacyRestTurn).toBe(0); + }); + it('completeLegacyRestTurn delegates to embedded under legacy (guard is the sole gate)', async () => { + const { port, calls } = recordingEmbeddedPort(); + const result = await legacyRouter(port).completeLegacyRestTurn(ctx(), { content: 'hello' }); + expect(result.ok).toBe(true); + expect(calls.completeLegacyRestTurn).toBe(1); + }); + + // prepareLegacySocketTurn -------------------------------------------------- + it('prepareLegacySocketTurn refuses with runtime_unsupported and never touches embedded under pi-rpc', async () => { + const { port, calls } = recordingEmbeddedPort(); + const result = await piRouter(port).prepareLegacySocketTurn( + ctx(), + { content: 'hello' }, + stream, + ); + expect(result).toEqual(RUNTIME_UNSUPPORTED); + expect(calls.prepareLegacySocketTurn).toBe(0); + }); + it('prepareLegacySocketTurn delegates to embedded under legacy (guard is the sole gate)', async () => { + const { port, calls } = recordingEmbeddedPort(); + const result = await legacyRouter(port).prepareLegacySocketTurn( + ctx(), + { content: 'hello' }, + stream, + ); + expect(result.ok).toBe(true); + expect(calls.prepareLegacySocketTurn).toBe(1); + }); + + // setLegacyThinking (sync) ------------------------------------------------- + it('setLegacyThinking refuses with runtime_unsupported and never touches embedded under pi-rpc', () => { + const { port, calls } = recordingEmbeddedPort(); + const result = piRouter(port).setLegacyThinking(ctx(), 'high'); + expect(result).toEqual(RUNTIME_UNSUPPORTED); + expect(calls.setLegacyThinking).toBe(0); + }); + it('setLegacyThinking delegates to embedded under legacy (guard is the sole gate)', () => { + const { port, calls } = recordingEmbeddedPort(); + const result = legacyRouter(port).setLegacyThinking(ctx(), 'high'); + expect(result.ok).toBe(true); + expect(calls.setLegacyThinking).toBe(1); + }); + + // abortLegacyTurn ---------------------------------------------------------- + it('abortLegacyTurn refuses with runtime_unsupported and never touches embedded under pi-rpc', async () => { + const { port, calls } = recordingEmbeddedPort(); + const result = await piRouter(port).abortLegacyTurn(ctx()); + expect(result).toEqual(RUNTIME_UNSUPPORTED); + expect(calls.abortLegacyTurn).toBe(0); + }); + it('abortLegacyTurn delegates to embedded under legacy (guard is the sole gate)', async () => { + const { port, calls } = recordingEmbeddedPort(); + const result = await legacyRouter(port).abortLegacyTurn(ctx()); + expect(result.ok).toBe(true); + expect(calls.abortLegacyTurn).toBe(1); + }); + + // applyLegacyModelOverride (sync) ------------------------------------------ + it('applyLegacyModelOverride refuses with runtime_unsupported and never touches embedded under pi-rpc', () => { + const { port, calls } = recordingEmbeddedPort(); + const result = piRouter(port).applyLegacyModelOverride(ctx(), 'model-x'); + expect(result).toEqual(RUNTIME_UNSUPPORTED); + expect(calls.applyLegacyModelOverride).toBe(0); + }); + it('applyLegacyModelOverride delegates to embedded under legacy (guard is the sole gate)', () => { + const { port, calls } = recordingEmbeddedPort(); + const result = legacyRouter(port).applyLegacyModelOverride(ctx(), 'model-x'); + expect(result.ok).toBe(true); + expect(calls.applyLegacyModelOverride).toBe(1); + }); + + // readLegacySessionPresentation (sync) ------------------------------------- + it('readLegacySessionPresentation refuses with runtime_unsupported and never touches embedded under pi-rpc', () => { + const { port, calls } = recordingEmbeddedPort(); + const result = piRouter(port).readLegacySessionPresentation(ctx()); + expect(result).toEqual(RUNTIME_UNSUPPORTED); + expect(calls.readLegacySessionPresentation).toBe(0); + }); + it('readLegacySessionPresentation delegates to embedded under legacy (guard is the sole gate)', () => { + const { port, calls } = recordingEmbeddedPort(); + const result = legacyRouter(port).readLegacySessionPresentation(ctx()); + expect(result.ok).toBe(true); + expect(calls.readLegacySessionPresentation).toBe(1); + }); + + // dispatchVerifiedDiscordIngress delegates in BOTH modes (embedded-only, no guard) --------- + it('dispatchVerifiedDiscordIngress delegates to embedded under pi-rpc (embedded-only, no mode guard)', async () => { + const { port, calls } = recordingEmbeddedPort(); + const discordCtx = ctx() as unknown as Parameters< + ChatRuntimeRouter['dispatchVerifiedDiscordIngress'] + >[0]; + const result = await piRouter(port).dispatchVerifiedDiscordIngress(discordCtx, stream); + expect(result.ok).toBe(true); + expect(calls.dispatchVerifiedDiscordIngress).toBe(1); + }); +}); + /** * Task Five, Step Two — group 1 (real Nest module-graph readiness). * diff --git a/apps/gateway/src/chat/chat.gateway.ts b/apps/gateway/src/chat/chat.gateway.ts index 4f63c5a9..a64950a3 100644 --- a/apps/gateway/src/chat/chat.gateway.ts +++ b/apps/gateway/src/chat/chat.gateway.ts @@ -441,7 +441,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa this.logger.warn(`Rejected malformed Discord ingress from ${client.id}`); return; } - const ingress = this.resolveDiscordIngress(client, rawData); + const ingress = this.resolveDiscordIngress(client, rawData, 'send', false); if (!ingress) return; const discordServiceUserId = process.env['DISCORD_SERVICE_USER_ID']; @@ -472,6 +472,18 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa }), ); + // 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 + // a corrected byte-identical retry dispatches once; once a turn commits here, a true duplicate + // finds the claim taken and fails closed with no additional dispatch/persist/ack. + if (!this.discordReplayProtector.claim(ingress.messageId)) { + this.logger.warn( + `Rejected replayed Discord message=${ingress.messageId} correlation=${ingress.correlationId}`, + ); + return; + } + this.logger.log( `Message from ${client.id} in conversation ${conversationId} correlation=${ingress.correlationId}`, ); @@ -1008,6 +1020,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa client: Socket, envelope: DiscordIngressEnvelope, operation: 'send' | 'approve' | 'stop' = 'send', + claimReplay = true, ): DiscordIngressPayload | null { const payload = verifyDiscordIngressEnvelope( envelope, @@ -1041,7 +1054,12 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa ); return null; } - if (!this.discordReplayProtector.claim(payload.messageId)) { + // The SEND path passes `claimReplay: false` and claims the message itself only after the + // configured-identity, binding, forced-scope, and attachment checks succeed — immediately + // before its first effect — so a SEND rejected by one of those later gates burns no claim and + // a corrected retry is not mistaken for a replay. The approve/stop paths have no such + // post-resolve gates, so they claim here, at the moment the envelope is fully verified. + if (claimReplay && !this.discordReplayProtector.claim(payload.messageId)) { this.logger.warn( `Rejected replayed Discord message=${payload.messageId} correlation=${payload.correlationId}`, ); diff --git a/apps/gateway/src/chat/chat.module.ts b/apps/gateway/src/chat/chat.module.ts index dacb35ad..17a0146c 100644 --- a/apps/gateway/src/chat/chat.module.ts +++ b/apps/gateway/src/chat/chat.module.ts @@ -54,6 +54,6 @@ import { HarnessChatRuntime } from './harness-chat.runtime.js'; ], }, ], - exports: [ChatGateway], + exports: [ChatGateway, ChatRuntimeRouter], }) export class ChatModule {} diff --git a/apps/gateway/src/conversations/conversations-harness-fence.spec.ts b/apps/gateway/src/conversations/conversations-harness-fence.spec.ts index 90cc58a2..d95355f7 100644 --- a/apps/gateway/src/conversations/conversations-harness-fence.spec.ts +++ b/apps/gateway/src/conversations/conversations-harness-fence.spec.ts @@ -1,4 +1,5 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { ChatRuntimeMode } from '../chat/chat-runtime.js'; import { ConversationsController } from './conversations.controller.js'; /** @@ -9,12 +10,11 @@ import { ConversationsController } from './conversations.controller.js'; * fixed typed `runtime_unsupported` BEFORE the repository is touched — never a duplicate write. * Under `legacy` the endpoint keeps its current behaviour and writes through `brain.conversations`. * - * The refusal is env-driven (`resolveChatRuntimeMode(process.env)` via `CHAT_HARNESS_RUNTIME`), so - * the controller's constructor signature does not change; the mode is read at request time. - * - * RED today: the controller writes unconditionally, so the pi-rpc case both writes (repo spy > 0) - * and returns a message instead of the typed refusal — a behavioural failure, not a DI/import one. - * GREEN (HELD until the RED checkpoint independently passes) adds the pre-write mode guard. + * Item 3 (single runtime-mode source of truth): the mode is the router's ONE init-time resolution, + * injected into the controller and read as `router.runtimeMode`. It is NOT re-derived from + * `process.env` at request time. The two "env is flipped after construction" tests below are the + * load-bearing guard: they pass only because the controller reads the fixed injected mode, and turn + * RED the instant the fence is reverted to `resolveChatRuntimeMode(process.env)`. */ const CONVERSATION_ID = '22222222-2222-4222-8222-222222222222'; const USER = { id: 'user-1' }; @@ -40,6 +40,11 @@ function brainWithMessageSpy() { }; } +/** The controller only needs the router's immutable `runtimeMode`; supply exactly that. */ +function routerFixedTo(mode: ChatRuntimeMode) { + return { runtimeMode: mode }; +} + let priorMode: string | undefined; describe('conversations REST write path — Task 5 harness fence', () => { @@ -52,10 +57,9 @@ describe('conversations REST write path — Task 5 harness fence', () => { else process.env['CHAT_HARNESS_RUNTIME'] = priorMode; }); - it('refuses the legacy repository write in pi-rpc mode with a fixed typed unsupported, before any write', async () => { - process.env['CHAT_HARNESS_RUNTIME'] = 'pi-rpc'; + it('refuses the legacy repository write when the router resolved pi-rpc, before any write', async () => { const { brain, addMessage } = brainWithMessageSpy(); - const controller = new ConversationsController(brain); + const controller = new ConversationsController(brain, routerFixedTo('pi-rpc')); await expect( controller.addMessage(CONVERSATION_ID, sendMessageDto(), USER), @@ -66,10 +70,9 @@ describe('conversations REST write path — Task 5 harness fence', () => { expect(addMessage).not.toHaveBeenCalled(); }); - it('writes through the repository in legacy mode (GREEN control)', async () => { - process.env['CHAT_HARNESS_RUNTIME'] = 'legacy'; + it('writes through the repository when the router resolved legacy (GREEN control)', async () => { const { brain, addMessage } = brainWithMessageSpy(); - const controller = new ConversationsController(brain); + const controller = new ConversationsController(brain, routerFixedTo('legacy')); const result = await controller.addMessage(CONVERSATION_ID, sendMessageDto(), USER); @@ -85,10 +88,26 @@ describe('conversations REST write path — Task 5 harness fence', () => { expect(result).toMatchObject({ id: 'message-1', conversationId: CONVERSATION_ID }); }); - it('writes through the repository when no runtime mode is set (defaults to legacy — GREEN control)', async () => { - delete process.env['CHAT_HARNESS_RUNTIME']; + it('keeps refusing under a pi-rpc router even when CHAT_HARNESS_RUNTIME is flipped to legacy after startup', async () => { + // The runtime mode is fixed at module init. A later env mutation must not reopen the fence: + // a request-time `resolveChatRuntimeMode(process.env)` read would see `legacy` and wrongly write. + process.env['CHAT_HARNESS_RUNTIME'] = 'legacy'; const { brain, addMessage } = brainWithMessageSpy(); - const controller = new ConversationsController(brain); + const controller = new ConversationsController(brain, routerFixedTo('pi-rpc')); + + await expect( + controller.addMessage(CONVERSATION_ID, sendMessageDto(), USER), + ).rejects.toMatchObject({ code: 'runtime_unsupported' }); + + expect(addMessage).not.toHaveBeenCalled(); + }); + + it('keeps writing under a legacy router even when CHAT_HARNESS_RUNTIME is flipped to pi-rpc after startup', async () => { + // Symmetric guard: a legacy-resolved router must keep writing regardless of the live env, so a + // request-time env read of `pi-rpc` cannot spuriously refuse a legitimate legacy write. + process.env['CHAT_HARNESS_RUNTIME'] = 'pi-rpc'; + const { brain, addMessage } = brainWithMessageSpy(); + const controller = new ConversationsController(brain, routerFixedTo('legacy')); await controller.addMessage(CONVERSATION_ID, sendMessageDto(), USER); diff --git a/apps/gateway/src/conversations/conversations.controller.ts b/apps/gateway/src/conversations/conversations.controller.ts index 393a7ef7..b6eaf480 100644 --- a/apps/gateway/src/conversations/conversations.controller.ts +++ b/apps/gateway/src/conversations/conversations.controller.ts @@ -20,7 +20,7 @@ import type { Brain } from '@mosaicstack/brain'; import { BRAIN } from '../brain/brain.tokens.js'; import { AuthGuard } from '../auth/auth.guard.js'; import { CurrentUser } from '../auth/current-user.decorator.js'; -import { resolveChatRuntimeMode } from '../chat/chat-runtime.js'; +import { ChatRuntimeRouter } from '../chat/chat-runtime-router.js'; import { CreateConversationDto, UpdateConversationDto, @@ -52,7 +52,17 @@ class HarnessRuntimeWriteUnsupportedException extends HttpException { @Controller('api/conversations') @UseGuards(AuthGuard) export class ConversationsController { - constructor(@Inject(BRAIN) private readonly brain: Brain) {} + /** + * `router` supplies the ONE immutable runtime mode resolved at module init (Task 5, item 3). + * The pre-write fence reads `router.runtimeMode`, never `resolveChatRuntimeMode(process.env)` at + * request time — a single source of truth, so the controller cannot disagree with the router + * about the live runtime if the environment is mutated after startup. Narrowed to `runtimeMode` + * so this class depends on nothing else the router exposes. + */ + constructor( + @Inject(BRAIN) private readonly brain: Brain, + @Inject(ChatRuntimeRouter) private readonly router: Pick, + ) {} @Get() async list(@CurrentUser() user: { id: string }) { @@ -118,8 +128,9 @@ export class ConversationsController { @CurrentUser() user: { id: string }, ) { // Fail the legacy repository write closed under pi-rpc BEFORE touching the repository — the - // harness path owns persistence there, so a direct write would duplicate the message. - if (resolveChatRuntimeMode(process.env) === 'pi-rpc') { + // harness path owns persistence there, so a direct write would duplicate the message. The mode + // comes from the router's init-time resolution, not a request-time env read. + if (this.router.runtimeMode === 'pi-rpc') { throw new HarnessRuntimeWriteUnsupportedException(); } diff --git a/apps/gateway/src/conversations/conversations.module.ts b/apps/gateway/src/conversations/conversations.module.ts index 19f3912f..ac56d566 100644 --- a/apps/gateway/src/conversations/conversations.module.ts +++ b/apps/gateway/src/conversations/conversations.module.ts @@ -1,7 +1,14 @@ import { Module } from '@nestjs/common'; +import { ChatModule } from '../chat/chat.module.js'; import { ConversationsController } from './conversations.controller.js'; +/** + * Imports {@link ChatModule} solely to inject its exported {@link ChatRuntimeRouter} into + * {@link ConversationsController}, so the REST write fence reads the same init-time runtime mode the + * router resolved — one source of truth, no duplicate provider, no global token, no AppModule edit. + */ @Module({ + imports: [ChatModule], controllers: [ConversationsController], }) export class ConversationsModule {} diff --git a/apps/gateway/src/plugin/discord-ingress.security.spec.ts b/apps/gateway/src/plugin/discord-ingress.security.spec.ts index 6698730b..066e811d 100644 --- a/apps/gateway/src/plugin/discord-ingress.security.spec.ts +++ b/apps/gateway/src/plugin/discord-ingress.security.spec.ts @@ -679,6 +679,105 @@ describe('Discord ingress security', () => { expect(harnessConversations.append).not.toHaveBeenCalled(); }); + it('a verified SEND that fails the configured service identity consumes no replay claim, so a corrected byte-identical retry dispatches/persists/acks exactly once and a later duplicate stays fail-closed (Task 5 item 4 — claim ordering)', 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(undefined); + 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-claim-ordering', + data: { discordService: true }, + emit: vi.fn(), + }; + const ackCount = (): number => + client.emit.mock.calls.filter((call) => call[0] === 'message:ack').length; + + // A single fully-valid signed envelope, reused byte-for-byte across all three deliveries. + const envelope = ingressEnvelope('verified once with late identity', 'discord-order-001', { + conversationId: 'Nova:discord:channel-001', + }); + + // (1) Configured service identity is MISSING. The envelope is validly signed and passes the + // binding + route checks, but the SEND must refuse at the identity gate BEFORE any claim + // or effect. If the claim fires ahead of that gate, this delivery silently burns the + // replay claim for `discord-order-001` even though nothing dispatched. + delete process.env['DISCORD_SERVICE_USER_ID']; + await gateway.handleMessage(client as never, envelope); + expect(createSession).toHaveBeenCalledTimes(0); + expect(prompt).toHaveBeenCalledTimes(0); + expect(addMessage).toHaveBeenCalledTimes(0); + expect(ackCount()).toBe(0); + + // (2) Identity is now configured; the operator resends the SAME envelope byte-for-byte. Because + // step (1) consumed no claim, this corrected retry claims once and runs the full embedded + // dispatch exactly once. (Under the pre-fix ordering the claim was already spent in step (1), + // so this retry is dropped as a replay and never dispatches — the RED this test drives.) + process.env['DISCORD_SERVICE_USER_ID'] = 'discord-service'; + 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 a committed turn stays fail-closed: the claim taken in step (2) + // blocks it, so every effect remains at exactly one. + 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('retains validated persisted attachments in resumed conversation history', async () => { const attachment = { id: 'attachment-history',