Fixes #756 (#763)

apps/gateway/src/chat/chat.dto.ts

@@ -1,3 +1,4 @@
+import type { ChannelAttachmentDto } from '@mosaicstack/types';
 import { IsOptional, IsString, IsUUID, MaxLength } from 'class-validator';
 
 export class ChatRequestDto {
@@ -32,4 +33,7 @@ export class ChatSocketMessageDto {
   @IsOptional()
   @IsUUID()
   agentId?: string;
+
+  /** Validated channel attachment references; binary content is not embedded. */
+  attachments?: readonly ChannelAttachmentDto[];
 }

apps/gateway/src/chat/chat.gateway-redaction.spec.ts

@@ -4,6 +4,10 @@ import { ChatGateway } from './chat.gateway.js';
 const CONVERSATION_ID = 'conversation-1';
 const CANARY = 'sk_canary12345678';
 
+function clientConversationKey(clientId: string, conversationId: string): string {
+  return `${clientId}\u0000${conversationId}`;
+}
+
 type GatewayInternals = {
   clientSessions: Map<string, unknown>;
   relayEvent(client: unknown, conversationId: string, event: unknown): void;
@@ -40,6 +44,7 @@ describe('ChatGateway redaction boundary', (): void => {
       emit: vi.fn(),
     };
     const session = {
+      clientId: client.id,
       conversationId: CONVERSATION_ID,
       cleanup: vi.fn(),
       assistantText: '',
@@ -47,7 +52,7 @@ describe('ChatGateway redaction boundary', (): void => {
       pendingToolCalls: new Map(),
       scope: { userId: 'user-1', tenantId: 'tenant-1' },
     };
-    gateway.clientSessions.set(client.id, session);
+    gateway.clientSessions.set(clientConversationKey(client.id, CONVERSATION_ID), session);
 
     gateway.relayEvent(client, CONVERSATION_ID, {
       type: 'message_update',
@@ -139,6 +144,51 @@ describe('ChatGateway redaction boundary', (): void => {
     });
   });
 
+  it('isolates concurrent conversation streams sharing one Discord socket', (): void => {
+    const { gateway } = buildGateway();
+    const client = {
+      connected: true,
+      id: 'discord-client',
+      data: { user: { id: 'user-1' } },
+      emit: vi.fn(),
+    };
+    const firstConversation = 'Nova:discord:thread-1';
+    const secondConversation = 'Nova:discord:thread-2';
+    const createSession = (conversationId: string) => ({
+      clientId: client.id,
+      conversationId,
+      cleanup: vi.fn(),
+      assistantText: '',
+      toolCalls: [],
+      pendingToolCalls: new Map(),
+      scope: { userId: 'user-1', tenantId: 'tenant-1' },
+    });
+    const firstSession = createSession(firstConversation);
+    const secondSession = createSession(secondConversation);
+    gateway.clientSessions.set(clientConversationKey(client.id, firstConversation), firstSession);
+    gateway.clientSessions.set(clientConversationKey(client.id, secondConversation), secondSession);
+
+    gateway.relayEvent(client, firstConversation, {
+      type: 'message_update',
+      assistantMessageEvent: { type: 'text_delta', delta: 'first response ' },
+    });
+    gateway.relayEvent(client, secondConversation, {
+      type: 'message_update',
+      assistantMessageEvent: { type: 'text_delta', delta: 'second response ' },
+    });
+
+    expect(firstSession.assistantText).toBe('first response ');
+    expect(secondSession.assistantText).toBe('second response ');
+    expect(client.emit).toHaveBeenCalledWith('agent:text', {
+      conversationId: firstConversation,
+      text: 'first response ',
+    });
+    expect(client.emit).toHaveBeenCalledWith('agent:text', {
+      conversationId: secondConversation,
+      text: 'second response ',
+    });
+  });
+
   it('persists only redacted assistant content with classifications', (): void => {
     const { gateway, brain } = buildGateway();
     const client = {
@@ -147,7 +197,8 @@ describe('ChatGateway redaction boundary', (): void => {
       data: { user: { id: 'user-1' } },
       emit: vi.fn(),
     };
-    gateway.clientSessions.set(client.id, {
+    gateway.clientSessions.set(clientConversationKey(client.id, CONVERSATION_ID), {
+      clientId: client.id,
       conversationId: CONVERSATION_ID,
       cleanup: vi.fn(),
       assistantText: CANARY,

apps/gateway/src/chat/chat.gateway.ts

@@ -17,6 +17,7 @@ import {
   parseDiscordInteractionBindings,
   resolveDiscordInteractionActorId,
   resolveDiscordInteractionBinding,
+  type DiscordAttachment,
   type DiscordIngressEnvelope,
   type DiscordIngressPayload,
 } from '@mosaicstack/discord-plugin';
@@ -30,6 +31,7 @@ import type {
   SystemReloadPayload,
   RoutingDecisionInfo,
   AbortPayload,
+  ChannelAttachmentDto,
 } from '@mosaicstack/types';
 import { AgentService, type ConversationHistoryMessage } from '../agent/agent.service.js';
 import {
@@ -56,6 +58,7 @@ import { DiscordReplayProtector } from '../plugin/discord-replay-protector.js';
 
 /** Per-client state tracking streaming accumulation for persistence. */
 interface ClientSession {
+  clientId: string;
   conversationId: string;
   cleanup: () => void;
   /** Accumulated assistant response text for the current turn. */
@@ -76,6 +79,68 @@ interface ClientSession {
  */
 const modelOverrides = new Map<string, string>();
 const MAX_REDACTION_BUFFER_LENGTH = 8_192;
+const MAX_CHANNEL_ATTACHMENTS = 10;
+const MAX_ATTACHMENT_METADATA_BYTES = 16_384;
+const MAX_ATTACHMENT_ID_LENGTH = 128;
+const MAX_ATTACHMENT_NAME_LENGTH = 255;
+const MAX_ATTACHMENT_URL_LENGTH = 2_048;
+const MAX_ATTACHMENT_MIME_LENGTH = 255;
+
+function isSafeAttachmentUrl(value: string): boolean {
+  if (value.length === 0 || value.length > MAX_ATTACHMENT_URL_LENGTH) return false;
+  try {
+    const url = new URL(value);
+    return (
+      url.protocol === 'https:' &&
+      !url.username &&
+      !url.password &&
+      !url.hash &&
+      url.search.length === 0
+    );
+  } catch {
+    return false;
+  }
+}
+
+function hasValidAttachmentBounds(value: {
+  id: string;
+  name: string;
+  url: string;
+  sizeBytes?: number;
+}): boolean {
+  return (
+    value.id.length > 0 &&
+    value.id.length <= MAX_ATTACHMENT_ID_LENGTH &&
+    value.name.length > 0 &&
+    value.name.length <= MAX_ATTACHMENT_NAME_LENGTH &&
+    isSafeAttachmentUrl(value.url) &&
+    (value.sizeBytes === undefined || (Number.isFinite(value.sizeBytes) && value.sizeBytes >= 0))
+  );
+}
+
+function isDiscordAttachment(value: unknown): value is DiscordAttachment {
+  if (typeof value !== 'object' || value === null) return false;
+  const attachment = value as Partial<DiscordAttachment>;
+  return (
+    typeof attachment.id === 'string' &&
+    typeof attachment.name === 'string' &&
+    typeof attachment.url === 'string' &&
+    (attachment.contentType === null ||
+      (typeof attachment.contentType === 'string' &&
+        attachment.contentType.length <= MAX_ATTACHMENT_MIME_LENGTH)) &&
+    (attachment.sizeBytes === undefined || typeof attachment.sizeBytes === 'number') &&
+    hasValidAttachmentBounds(attachment as DiscordAttachment)
+  );
+}
+
+function hasValidAttachmentArray(value: unknown, guard: (attachment: unknown) => boolean): boolean {
+  return (
+    Array.isArray(value) &&
+    value.length <= MAX_CHANNEL_ATTACHMENTS &&
+    JSON.stringify(value).length <= MAX_ATTACHMENT_METADATA_BYTES &&
+    value.every(guard)
+  );
+}
 
 function isDiscordIngressEnvelope(value: unknown): value is DiscordIngressEnvelope {
   if (typeof value !== 'object' || value === null) return false;
@@ -88,23 +153,49 @@ function isDiscordIngressEnvelope(value: unknown): value is DiscordIngressEnvelo
     return false;
   }
   const payload = envelope.payload as Record<string, unknown>;
-  return [
-    payload['correlationId'],
-    payload['messageId'],
-    payload['guildId'],
-    payload['channelId'],
-    payload['userId'],
-    payload['conversationId'],
-    payload['content'],
-  ].every((field: unknown): boolean => typeof field === 'string');
+  return (
+    [
+      payload['correlationId'],
+      payload['messageId'],
+      payload['guildId'],
+      payload['channelId'],
+      payload['userId'],
+      payload['conversationId'],
+      payload['content'],
+    ].every((field: unknown): boolean => typeof field === 'string') &&
+    (payload['threadId'] === undefined || typeof payload['threadId'] === 'string') &&
+    (payload['attachments'] === undefined ||
+      hasValidAttachmentArray(payload['attachments'], isDiscordAttachment))
+  );
+}
+
+function isChannelAttachment(value: unknown): value is ChannelAttachmentDto {
+  if (typeof value !== 'object' || value === null) return false;
+  const attachment = value as Partial<ChannelAttachmentDto>;
+  return (
+    typeof attachment.id === 'string' &&
+    typeof attachment.name === 'string' &&
+    typeof attachment.url === 'string' &&
+    (attachment.mimeType === null ||
+      (typeof attachment.mimeType === 'string' &&
+        attachment.mimeType.length <= MAX_ATTACHMENT_MIME_LENGTH)) &&
+    (attachment.sizeBytes === undefined || typeof attachment.sizeBytes === 'number') &&
+    hasValidAttachmentBounds(attachment as ChannelAttachmentDto)
+  );
 }
 
 function isChatSocketMessage(value: unknown): value is ChatSocketMessageDto {
   if (typeof value !== 'object' || value === null) return false;
-  const payload = value as { content?: unknown; conversationId?: unknown };
+  const payload = value as {
+    content?: unknown;
+    conversationId?: unknown;
+    attachments?: unknown;
+  };
   return (
     typeof payload.content === 'string' &&
-    (payload.conversationId === undefined || typeof payload.conversationId === 'string')
+    (payload.conversationId === undefined || typeof payload.conversationId === 'string') &&
+    (payload.attachments === undefined ||
+      hasValidAttachmentArray(payload.attachments, isChannelAttachment))
   );
 }
 
@@ -174,20 +265,24 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
 
   handleDisconnect(client: Socket): void {
     this.logger.log(`Client disconnected: ${client.id}`);
-    const session = this.clientSessions.get(client.id);
-    if (session) {
+    for (const [key, session] of this.clientSessions) {
+      if (session.clientId !== client.id) continue;
       session.cleanup();
       this.agentService.removeChannel(
         session.conversationId,
         `websocket:${client.id}`,
         session.scope,
       );
-      this.clientSessions.delete(client.id);
+      this.clientSessions.delete(key);
+      this.textEgressBuffers.delete(key);
+      this.thinkingEgressBuffers.delete(key);
+      this.overflowedEgress.delete(`${key}:agent:text`);
+      this.overflowedEgress.delete(`${key}:agent:thinking`);
     }
-    this.textEgressBuffers.delete(client.id);
-    this.thinkingEgressBuffers.delete(client.id);
-    this.overflowedEgress.delete(this.egressKey(client, 'agent:text'));
-    this.overflowedEgress.delete(this.egressKey(client, 'agent:thinking'));
+  }
+
+  private clientConversationKey(client: Pick<Socket, 'id'>, conversationId: string): string {
+    return `${client.id}\u0000${conversationId}`;
   }
 
   private getClientScope(client: Socket): ActorTenantScope | null {
@@ -218,7 +313,25 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
       }
       discordIngress = this.resolveDiscordIngress(client, rawData);
       if (!discordIngress) return;
-      data = { conversationId: discordIngress.conversationId, content: discordIngress.content };
+      data = {
+        conversationId: discordIngress.conversationId,
+        content: discordIngress.content,
+        ...(discordIngress.attachments
+          ? {
+              attachments: discordIngress.attachments.map(
+                (attachment): ChannelAttachmentDto => ({
+                  id: attachment.id,
+                  name: attachment.name,
+                  url: attachment.url,
+                  mimeType: attachment.contentType,
+                  ...(attachment.sizeBytes !== undefined
+                    ? { sizeBytes: attachment.sizeBytes }
+                    : {}),
+                }),
+              ),
+            }
+          : {}),
+      };
     } else {
       if (!isChatSocketMessage(rawData)) {
         this.logger.warn(`Rejected malformed chat message from ${client.id}`);
@@ -227,6 +340,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
       data = rawData;
     }
     const conversationId = data.conversationId ?? uuid();
+    const clientConversationKey = this.clientConversationKey(client, conversationId);
     const discordServiceUserId = process.env['DISCORD_SERVICE_USER_ID'];
     if (discordIngress && !discordServiceUserId) {
       this.logger.warn(
@@ -281,7 +395,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
           this.logger.log(
             `Using /model override "${modelOverride}" for conversation=${conversationId}`,
           );
-        } else if (!resolvedProvider && !resolvedModelId) {
+        } else if (!resolvedProvider && !resolvedModelId && !discordIngress) {
           // No explicit provider/model from client — use routing engine (M4-012)
           try {
             const routingDecision = await this.routingEngine.resolve(data.content, userId);
@@ -304,12 +418,24 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
           }
         }
 
+        let resolvedAgentConfigId = data.agentId;
+        if (discordIngress) {
+          const binding = this.discordBindingFor(discordIngress, 'send');
+          const agentConfig = binding
+            ? await this.brain.agents.findById(binding.agentConfigId)
+            : undefined;
+          if (!binding || !agentConfig || agentConfig.name !== binding.instanceId) {
+            throw new Error('Configured Discord logical agent is not provisioned');
+          }
+          resolvedAgentConfigId = agentConfig.id;
+        }
+
         // M5-004: Use existingSessionId as sessionId when available (session reuse)
         const sessionIdToCreate = existingSessionId ?? conversationId;
         agentSession = await this.agentService.createSession(sessionIdToCreate, {
           provider: resolvedProvider,
           modelId: resolvedModelId,
-          agentConfigId: data.agentId,
+          agentConfigId: resolvedAgentConfigId,
           userId,
           tenantId: scope.tenantId,
           conversationHistory: conversationHistory.length > 0 ? conversationHistory : undefined,
@@ -360,6 +486,17 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
                     discordUserId: discordIngress?.userId,
                   }
                 : {}),
+              ...(data.attachments && data.attachments.length > 0
+                ? {
+                    channelAttachments: data.attachments.map(
+                      (attachment): ChannelAttachmentDto => ({
+                        ...attachment,
+                        name: redactSensitiveContent(attachment.name).content,
+                        url: redactSensitiveContent(attachment.url).content,
+                      }),
+                    ),
+                  }
+                : {}),
               classifications: redactSensitiveContent(data.content).classifications,
             },
           },
@@ -374,7 +511,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     }
 
     // Always clean up previous listener to prevent leak
-    const existing = this.clientSessions.get(client.id);
+    const existing = this.clientSessions.get(clientConversationKey);
     if (existing) {
       existing.cleanup();
     }
@@ -389,10 +526,11 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     );
 
     // Preserve routing decision from the existing client session if we didn't get a new one
-    const prevClientSession = this.clientSessions.get(client.id);
+    const prevClientSession = this.clientSessions.get(clientConversationKey);
     const routingDecisionToStore = sessionRoutingDecision ?? prevClientSession?.lastRoutingDecision;
 
-    this.clientSessions.set(client.id, {
+    this.clientSessions.set(clientConversationKey, {
+      clientId: client.id,
       conversationId,
       cleanup,
       assistantText: '',
@@ -438,7 +576,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
 
     // Dispatch to agent
     try {
-      await this.agentService.prompt(conversationId, data.content, scope);
+      await this.agentService.prompt(conversationId, data.content, scope, data.attachments);
     } catch (err) {
       this.logger.error(
         `Agent prompt failed for client=${client.id}, conversation=${conversationId}`,
@@ -645,9 +783,9 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     };
 
     // Emit to all clients currently subscribed to this conversation
-    for (const [clientId, session] of this.clientSessions) {
+    for (const session of this.clientSessions.values()) {
       if (session.conversationId === conversationId && this.scopesEqual(session.scope, scope)) {
-        const socket = this.server.sockets.sockets.get(clientId);
+        const socket = this.server.sockets.sockets.get(session.clientId);
         if (socket?.connected) {
           socket.emit('session:info', payload);
         }
@@ -677,16 +815,10 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
       !this.durableSessions
     )
       return;
-    const binding = resolveDiscordInteractionBinding(
-      parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']),
-      ingress.guildId,
-      ingress.channelId,
-      ingress.userId,
-      'approve',
-    );
+    const binding = this.discordBindingFor(ingress, 'approve');
     const actorId = binding && resolveDiscordInteractionActorId(binding, ingress.userId);
-    const agentName = process.env['MOSAIC_AGENT_NAME']?.trim();
-    if (!actorId || !agentName || binding.instanceId !== agentName) {
+    const agentName = binding?.instanceId;
+    if (!actorId || !agentName) {
       this.logger.warn(
         `Rejected Discord approval without a matching runtime agent from ${client.id}`,
       );
@@ -774,13 +906,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     const tenantId = process.env['DISCORD_SERVICE_TENANT_ID']?.trim();
     if (!ingress || !approvalRef || !tenantId || !this.runtimeRegistry || !this.durableSessions)
       return;
-    const binding = resolveDiscordInteractionBinding(
-      parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']),
-      ingress.guildId,
-      ingress.channelId,
-      ingress.userId,
-      'stop',
-    );
+    const binding = this.discordBindingFor(ingress, 'stop');
     const actorId = binding && resolveDiscordInteractionActorId(binding, ingress.userId);
     if (!actorId) return;
 
@@ -854,17 +980,18 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
       return null;
     }
     try {
-      const binding = resolveDiscordInteractionBinding(
-        parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']),
-        payload.guildId,
-        payload.channelId,
-        payload.userId,
-        operation,
-      );
+      const binding = this.discordBindingFor(payload, operation);
       if (!binding) {
         this.logger.warn(`Rejected unpaired Discord ingress from ${client.id}`);
         return null;
       }
+      const expectedConversationId = `${binding.instanceId}:discord:${payload.threadId ?? payload.channelId}`;
+      if (payload.conversationId !== expectedConversationId) {
+        this.logger.warn(
+          `Rejected Discord ingress for a different logical agent from ${client.id}`,
+        );
+        return null;
+      }
     } catch {
       this.logger.warn(
         `Rejected Discord ingress without valid binding configuration from ${client.id}`,
@@ -880,6 +1007,19 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     return payload;
   }
 
+  private discordBindingFor(
+    payload: DiscordIngressPayload,
+    operation: 'send' | 'approve' | 'stop',
+  ) {
+    return resolveDiscordInteractionBinding(
+      parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']),
+      payload.guildId,
+      payload.channelId,
+      payload.userId,
+      operation,
+    );
+  }
+
   private readDiscordAllowlist(name: string): string[] {
     return (process.env[name] ?? '')
       .split(',')
@@ -958,11 +1098,15 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
       const messages = await this.brain.conversations.findMessages(conversationId, userId);
       if (messages.length === 0) return [];
 
-      return messages.map((msg) => ({
-        role: msg.role as 'user' | 'assistant' | 'system',
-        content: msg.content,
-        createdAt: msg.createdAt,
-      }));
+      return messages.map((msg) => {
+        const attachments = this.persistedChannelAttachments(msg.metadata);
+        return {
+          role: msg.role as 'user' | 'assistant' | 'system',
+          content: msg.content,
+          createdAt: msg.createdAt,
+          ...(attachments ? { attachments } : {}),
+        };
+      });
     } catch (err) {
       this.logger.error(
         `Failed to load conversation history for conversation=${conversationId}`,
@@ -972,6 +1116,14 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     }
   }
 
+  private persistedChannelAttachments(metadata: unknown): readonly ChannelAttachmentDto[] | null {
+    if (typeof metadata !== 'object' || metadata === null) return null;
+    const attachments = (metadata as { channelAttachments?: unknown }).channelAttachments;
+    return hasValidAttachmentArray(attachments, isChannelAttachment)
+      ? (attachments as readonly ChannelAttachmentDto[])
+      : null;
+  }
+
   private appendAndFlushRedactedEgress(
     client: Socket,
     conversationId: string,
@@ -979,18 +1131,19 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     buffers: Map<string, string>,
     delta: string,
   ): void {
-    const key = this.egressKey(client, eventName);
+    const sessionKey = this.clientConversationKey(client, conversationId);
+    const key = this.egressKey(client, conversationId, eventName);
     if (this.overflowedEgress.has(key)) return;
 
-    const buffered = `${buffers.get(client.id) ?? ''}${delta}`;
+    const buffered = `${buffers.get(sessionKey) ?? ''}${delta}`;
     if (buffered.length > MAX_REDACTION_BUFFER_LENGTH) {
-      buffers.delete(client.id);
+      buffers.delete(sessionKey);
       this.overflowedEgress.add(key);
       client.emit(eventName, { conversationId, text: '[REDACTED_STREAM_OVERFLOW]' });
       return;
     }
 
-    buffers.set(client.id, buffered);
+    buffers.set(sessionKey, buffered);
     this.flushRedactedEgress(client, conversationId, eventName, buffers, false);
   }
 
@@ -1006,21 +1159,22 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     buffers: Map<string, string>,
     final: boolean,
   ): void {
-    const key = this.egressKey(client, eventName);
+    const sessionKey = this.clientConversationKey(client, conversationId);
+    const key = this.egressKey(client, conversationId, eventName);
     if (this.overflowedEgress.has(key)) {
       if (final) this.overflowedEgress.delete(key);
       return;
     }
 
-    const buffered = buffers.get(client.id) ?? '';
+    const buffered = buffers.get(sessionKey) ?? '';
     const releaseLength = final ? buffered.length : this.safeRedactionPrefixLength(buffered);
     const released = buffered.slice(0, releaseLength);
     const pending = buffered.slice(releaseLength);
 
     if (pending) {
-      buffers.set(client.id, pending);
+      buffers.set(sessionKey, pending);
     } else {
-      buffers.delete(client.id);
+      buffers.delete(sessionKey);
     }
 
     if (released) {
@@ -1089,8 +1243,12 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     return retainedFrom;
   }
 
-  private egressKey(client: Socket, eventName: 'agent:text' | 'agent:thinking'): string {
-    return `${client.id}:${eventName}`;
+  private egressKey(
+    client: Socket,
+    conversationId: string,
+    eventName: 'agent:text' | 'agent:thinking',
+  ): string {
+    return `${this.clientConversationKey(client, conversationId)}:${eventName}`;
   }
 
   private relayEvent(client: Socket, conversationId: string, event: AgentSessionEvent): void {
@@ -1101,26 +1259,27 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
       return;
     }
 
+    const sessionKey = this.clientConversationKey(client, conversationId);
     switch (event.type) {
       case 'agent_start': {
         // Reset accumulation buffers for the new turn
-        const cs = this.clientSessions.get(client.id);
+        const cs = this.clientSessions.get(sessionKey);
         if (cs) {
           cs.assistantText = '';
           cs.toolCalls = [];
           cs.pendingToolCalls.clear();
         }
-        this.textEgressBuffers.set(client.id, '');
-        this.thinkingEgressBuffers.set(client.id, '');
-        this.overflowedEgress.delete(this.egressKey(client, 'agent:text'));
-        this.overflowedEgress.delete(this.egressKey(client, 'agent:thinking'));
+        this.textEgressBuffers.set(sessionKey, '');
+        this.thinkingEgressBuffers.set(sessionKey, '');
+        this.overflowedEgress.delete(this.egressKey(client, conversationId, 'agent:text'));
+        this.overflowedEgress.delete(this.egressKey(client, conversationId, 'agent:thinking'));
         client.emit('agent:start', { conversationId });
         break;
       }
 
       case 'agent_end': {
         // Gather usage stats from the Pi session
-        const activeClientSession = this.clientSessions.get(client.id);
+        const activeClientSession = this.clientSessions.get(sessionKey);
         const agentSession = activeClientSession
           ? this.agentService.getSession(conversationId, activeClientSession.scope)
           : undefined;
@@ -1173,7 +1332,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
         }
 
         // Persist the assistant message with metadata
-        const cs = this.clientSessions.get(client.id);
+        const cs = this.clientSessions.get(sessionKey);
         const userId = (client.data.user as { id: string } | undefined)?.id;
         if (cs && userId && cs.assistantText.trim().length > 0) {
           const metadata: Record<string, unknown> = {
@@ -1225,7 +1384,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
         const assistantEvent = event.assistantMessageEvent;
         if (assistantEvent.type === 'text_delta') {
           // Keep raw stream material in memory only; persist and emit only redacted text.
-          const cs = this.clientSessions.get(client.id);
+          const cs = this.clientSessions.get(sessionKey);
           if (cs) {
             cs.assistantText += assistantEvent.delta;
           }
@@ -1250,7 +1409,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
 
       case 'tool_execution_start': {
         // Track pending tool call for later recording
-        const cs = this.clientSessions.get(client.id);
+        const cs = this.clientSessions.get(sessionKey);
         if (cs) {
           cs.pendingToolCalls.set(event.toolCallId, {
             toolName: event.toolName,
@@ -1267,7 +1426,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
 
       case 'tool_execution_end': {
         // Finalise tool call record
-        const cs = this.clientSessions.get(client.id);
+        const cs = this.clientSessions.get(sessionKey);
         if (cs) {
           const pending = cs.pendingToolCalls.get(event.toolCallId);
           cs.toolCalls.push({