refactor(chat): route browser chat through one runtime

Co-Authored-By: Claude Opus 4.8 <[email protected]>
Claude-Session: https://claude.ai/code/session_01ESFAnh2t9HmLwng8oW95St
This commit is contained in:
2026-08-12 12:05:05 -05:00
co-authored by Claude Opus 4.8
parent 472dcee7ed
commit 694f1a4199
9 changed files with 434 additions and 27 deletions
@@ -417,7 +417,7 @@ describe('ConversationsController — search endpoint', () => {
}, },
]; ];
brain = createMockBrain({ searchResults }); 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 () => { it('returns matching messages for a valid search query', async () => {
@@ -479,7 +479,7 @@ describe('ConversationsController — search endpoint', () => {
describe('ConversationsController — message CRUD', () => { describe('ConversationsController — message CRUD', () => {
it('listMessages returns 404 when conversation is not owned by user', async () => { it('listMessages returns 404 when conversation is not owned by user', async () => {
const brain = createMockBrain({ conversation: undefined }); 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( await expect(controller.listMessages(CONV_ID, { id: USER_ID })).rejects.toBeInstanceOf(
NotFoundException, NotFoundException,
@@ -489,7 +489,7 @@ describe('ConversationsController — message CRUD', () => {
it('listMessages returns the messages for an owned conversation', async () => { it('listMessages returns the messages for an owned conversation', async () => {
const msgs = [makeMessage('user', 'Test message'), makeMessage('assistant', 'Test reply')]; const msgs = [makeMessage('user', 'Test message'), makeMessage('assistant', 'Test reply')];
const brain = createMockBrain({ conversation: makeConversation(), messages: msgs }); 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 }); 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 () => { it('addMessage returns the persisted message', async () => {
const brain = createMockBrain({ conversation: makeConversation() }); 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( const result = await controller.addMessage(
CONV_ID, CONV_ID,
@@ -60,7 +60,7 @@ describe('Resource ownership checks', () => {
// The repo enforces ownership via the WHERE clause; it returns undefined when the // The repo enforces ownership via the WHERE clause; it returns undefined when the
// conversation does not belong to the requesting user. // conversation does not belong to the requesting user.
brain.conversations.findById.mockResolvedValue(undefined); 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( await expect(controller.findOne('conv-1', { id: 'user-1' })).rejects.toBeInstanceOf(
NotFoundException, NotFoundException,
@@ -19,8 +19,14 @@ import {
import { ChatRuntimeRouter } from './chat-runtime-router.js'; import { ChatRuntimeRouter } from './chat-runtime-router.js';
import { import {
ChatRuntimeUnavailableError, ChatRuntimeUnavailableError,
ownConversation,
type ChatRuntime, type ChatRuntime,
type ChatRuntimeMode, type ChatRuntimeMode,
type LegacyEmbeddedChatPort,
type LegacyRuntimeStream,
type LegacySessionPresentation,
type LegacySocketTurnLease,
type OwnedConversationContext,
} from './chat-runtime.js'; } from './chat-runtime.js';
import { AppModule } from '../app.module.js'; import { AppModule } from '../app.module.js';
import { ProviderService } from '../agent/provider.service.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.<op>` 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). * Task Five, Step Two — group 1 (real Nest module-graph readiness).
* *
+20 -2
View File
@@ -441,7 +441,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
this.logger.warn(`Rejected malformed Discord ingress from ${client.id}`); this.logger.warn(`Rejected malformed Discord ingress from ${client.id}`);
return; return;
} }
const ingress = this.resolveDiscordIngress(client, rawData); const ingress = this.resolveDiscordIngress(client, rawData, 'send', false);
if (!ingress) return; if (!ingress) return;
const discordServiceUserId = process.env['DISCORD_SERVICE_USER_ID']; 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( this.logger.log(
`Message from ${client.id} in conversation ${conversationId} correlation=${ingress.correlationId}`, `Message from ${client.id} in conversation ${conversationId} correlation=${ingress.correlationId}`,
); );
@@ -1008,6 +1020,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
client: Socket, client: Socket,
envelope: DiscordIngressEnvelope, envelope: DiscordIngressEnvelope,
operation: 'send' | 'approve' | 'stop' = 'send', operation: 'send' | 'approve' | 'stop' = 'send',
claimReplay = true,
): DiscordIngressPayload | null { ): DiscordIngressPayload | null {
const payload = verifyDiscordIngressEnvelope( const payload = verifyDiscordIngressEnvelope(
envelope, envelope,
@@ -1041,7 +1054,12 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
); );
return null; 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( this.logger.warn(
`Rejected replayed Discord message=${payload.messageId} correlation=${payload.correlationId}`, `Rejected replayed Discord message=${payload.messageId} correlation=${payload.correlationId}`,
); );
+1 -1
View File
@@ -54,6 +54,6 @@ import { HarnessChatRuntime } from './harness-chat.runtime.js';
], ],
}, },
], ],
exports: [ChatGateway], exports: [ChatGateway, ChatRuntimeRouter],
}) })
export class ChatModule {} export class ChatModule {}
@@ -1,4 +1,5 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import type { ChatRuntimeMode } from '../chat/chat-runtime.js';
import { ConversationsController } from './conversations.controller.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. * 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`. * 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 * Item 3 (single runtime-mode source of truth): the mode is the router's ONE init-time resolution,
* the controller's constructor signature does not change; the mode is read at request time. * 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
* RED today: the controller writes unconditionally, so the pi-rpc case both writes (repo spy > 0) * load-bearing guard: they pass only because the controller reads the fixed injected mode, and turn
* and returns a message instead of the typed refusal — a behavioural failure, not a DI/import one. * RED the instant the fence is reverted to `resolveChatRuntimeMode(process.env)`.
* GREEN (HELD until the RED checkpoint independently passes) adds the pre-write mode guard.
*/ */
const CONVERSATION_ID = '22222222-2222-4222-8222-222222222222'; const CONVERSATION_ID = '22222222-2222-4222-8222-222222222222';
const USER = { id: 'user-1' }; 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; let priorMode: string | undefined;
describe('conversations REST write path — Task 5 harness fence', () => { 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; 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 () => { it('refuses the legacy repository write when the router resolved pi-rpc, before any write', async () => {
process.env['CHAT_HARNESS_RUNTIME'] = 'pi-rpc';
const { brain, addMessage } = brainWithMessageSpy(); const { brain, addMessage } = brainWithMessageSpy();
const controller = new ConversationsController(brain); const controller = new ConversationsController(brain, routerFixedTo('pi-rpc'));
await expect( await expect(
controller.addMessage(CONVERSATION_ID, sendMessageDto(), USER), controller.addMessage(CONVERSATION_ID, sendMessageDto(), USER),
@@ -66,10 +70,9 @@ describe('conversations REST write path — Task 5 harness fence', () => {
expect(addMessage).not.toHaveBeenCalled(); expect(addMessage).not.toHaveBeenCalled();
}); });
it('writes through the repository in legacy mode (GREEN control)', async () => { it('writes through the repository when the router resolved legacy (GREEN control)', async () => {
process.env['CHAT_HARNESS_RUNTIME'] = 'legacy';
const { brain, addMessage } = brainWithMessageSpy(); 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); 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 }); 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 () => { it('keeps refusing under a pi-rpc router even when CHAT_HARNESS_RUNTIME is flipped to legacy after startup', async () => {
delete process.env['CHAT_HARNESS_RUNTIME']; // 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 { 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); await controller.addMessage(CONVERSATION_ID, sendMessageDto(), USER);
@@ -20,7 +20,7 @@ import type { Brain } from '@mosaicstack/brain';
import { BRAIN } from '../brain/brain.tokens.js'; import { BRAIN } from '../brain/brain.tokens.js';
import { AuthGuard } from '../auth/auth.guard.js'; import { AuthGuard } from '../auth/auth.guard.js';
import { CurrentUser } from '../auth/current-user.decorator.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 { import {
CreateConversationDto, CreateConversationDto,
UpdateConversationDto, UpdateConversationDto,
@@ -52,7 +52,17 @@ class HarnessRuntimeWriteUnsupportedException extends HttpException {
@Controller('api/conversations') @Controller('api/conversations')
@UseGuards(AuthGuard) @UseGuards(AuthGuard)
export class ConversationsController { 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<ChatRuntimeRouter, 'runtimeMode'>,
) {}
@Get() @Get()
async list(@CurrentUser() user: { id: string }) { async list(@CurrentUser() user: { id: string }) {
@@ -118,8 +128,9 @@ export class ConversationsController {
@CurrentUser() user: { id: string }, @CurrentUser() user: { id: string },
) { ) {
// Fail the legacy repository write closed under pi-rpc BEFORE touching the repository — the // 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. // harness path owns persistence there, so a direct write would duplicate the message. The mode
if (resolveChatRuntimeMode(process.env) === 'pi-rpc') { // comes from the router's init-time resolution, not a request-time env read.
if (this.router.runtimeMode === 'pi-rpc') {
throw new HarnessRuntimeWriteUnsupportedException(); throw new HarnessRuntimeWriteUnsupportedException();
} }
@@ -1,7 +1,14 @@
import { Module } from '@nestjs/common'; import { Module } from '@nestjs/common';
import { ChatModule } from '../chat/chat.module.js';
import { ConversationsController } from './conversations.controller.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({ @Module({
imports: [ChatModule],
controllers: [ConversationsController], controllers: [ConversationsController],
}) })
export class ConversationsModule {} export class ConversationsModule {}
@@ -679,6 +679,105 @@ describe('Discord ingress security', () => {
expect(harnessConversations.append).not.toHaveBeenCalled(); 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 () => { it('retains validated persisted attachments in resumed conversation history', async () => {
const attachment = { const attachment = {
id: 'attachment-history', id: 'attachment-history',