import { createHash } from 'node:crypto';
import { Inject, Logger, Optional } from '@nestjs/common';
import {
  WebSocketGateway,
  WebSocketServer,
  SubscribeMessage,
  OnGatewayConnection,
  OnGatewayDisconnect,
  type OnGatewayInit,
  ConnectedSocket,
  MessageBody,
} from '@nestjs/websockets';
import { Server, Socket } from 'socket.io';
import {
  verifyDiscordIngressEnvelope,
  parseDiscordInteractionBindings,
  resolveDiscordInteractionActorId,
  resolveDiscordInteractionBinding,
  type DiscordAttachment,
  type DiscordIngressEnvelope,
  type DiscordIngressPayload,
} from '@mosaicstack/discord-plugin';
import type { Auth } from '@mosaicstack/auth';
import type { Brain } from '@mosaicstack/brain';
import { redactSensitiveContent } from '@mosaicstack/log';
import type {
  SetThinkingPayload,
  SlashCommandApprovalResultPayload,
  SlashCommandPayload,
  SystemReloadPayload,
  RoutingDecisionInfo,
  AbortPayload,
  ChannelAttachmentDto,
} from '@mosaicstack/types';
import type { ConversationHistoryMessage } from '../agent/agent.service.js';
import {
  RUNTIME_PROVIDER_AUDIT_SINK,
  RuntimeProviderService,
  type RuntimeAuditSink,
} from '../agent/runtime-provider-registry.service.js';
import { DurableSessionService } from '../agent/durable-session.service.js';
import { AUTH } from '../auth/auth.tokens.js';
import {
  scopeFromUser,
  type ActorTenantScope,
  type AuthenticatedUserLike,
} from '../auth/session-scope.js';
import { BRAIN } from '../brain/brain.tokens.js';
import { CommandRegistryService } from '../commands/command-registry.service.js';
import { CommandExecutorService } from '../commands/command-executor.service.js';
import { CommandAuthorizationService } from '../commands/command-authorization.service.js';
import { RoutingEngineService } from '../agent/routing/routing-engine.service.js';
import { ChatRuntimeRouter } from './chat-runtime-router.js';
import {
  ownConversation,
  verifyDiscordIngress,
  type LegacyRuntimeEvent,
  type LegacyRuntimeStream,
  type LegacySocketTurnLease,
  type VerifiedDiscordTurnLease,
} from './chat-runtime.js';
import { v4 as uuid } from 'uuid';
import { ChatSocketMessageDto } from './chat.dto.js';
import { validateDiscordServiceToken, validateSocketSession } from './chat.gateway-auth.js';
import { DiscordReplayProtector } from '../plugin/discord-replay-protector.js';

/** Per-client state tracking streaming accumulation for persistence. */
interface ClientSession {
  clientId: string;
  conversationId: string;
  /** Server-derived egress channel id (`websocket:<socket-id>`) this turn streams over. */
  channelId: string;
  /** The prepared runtime turn; disposing it removes the listener and channel. */
  lease: LegacySocketTurnLease | VerifiedDiscordTurnLease;
  /** Accumulated assistant response text for the current turn. */
  assistantText: string;
  /** Tool calls observed during the current turn. */
  toolCalls: Array<{ toolCallId: string; toolName: string; args: unknown; isError: boolean }>;
  /** Tool calls in-flight (started but not ended yet). */
  pendingToolCalls: Map<string, { toolName: string; args: unknown }>;
  /** Server-derived owner/tenant scope for this socket's conversation attachment. */
  scope: ActorTenantScope;
}

/**
 * Per-conversation model overrides set via /model command (M4-007).
 * Keyed by conversationId, value is the model name to use.
 */
const modelOverrides = new Map<string, string>();
/**
 * Task 5 (G3): commands whose effect is runtime-independent — they operate on gateway/system
 * state rather than an embedded chat session — and therefore stay available under pi-rpc. Every
 * other command is an embedded-session command and is fixed-"unsupported" under pi-rpc, failing
 * closed before the executor. Kept as an explicit allowlist so adding a runtime-independent
 * command is a deliberate edit, never an accidental fall-through.
 */
const RUNTIME_INDEPENDENT_COMMANDS: ReadonlySet<string> = new Set(['reload']);
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;
  const envelope = value as { payload?: unknown; signature?: unknown };
  if (
    typeof envelope.signature !== 'string' ||
    typeof envelope.payload !== 'object' ||
    envelope.payload === null
  ) {
    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') &&
    (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;
    attachments?: unknown;
  };
  return (
    typeof payload.content === 'string' &&
    (payload.conversationId === undefined || typeof payload.conversationId === 'string') &&
    (payload.attachments === undefined ||
      hasValidAttachmentArray(payload.attachments, isChannelAttachment))
  );
}

@WebSocketGateway({
  cors: {
    origin: process.env['GATEWAY_CORS_ORIGIN'] ?? 'http://localhost:3000',
  },
  namespace: '/chat',
})
export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect {
  @WebSocketServer()
  server!: Server;

  private readonly logger = new Logger(ChatGateway.name);
  private readonly clientSessions = new Map<string, ClientSession>();
  /** Raw stream fragments are kept in memory only until they are safe to redact and emit. */
  private readonly textEgressBuffers = new Map<string, string>();
  private readonly thinkingEgressBuffers = new Map<string, string>();
  private readonly overflowedEgress = new Set<string>();
  private readonly discordReplayProtector = new DiscordReplayProtector();

  constructor(
    private readonly runtime: ChatRuntimeRouter,
    @Inject(AUTH) private readonly auth: Auth,
    @Inject(BRAIN) private readonly brain: Brain,
    @Inject(CommandRegistryService) private readonly commandRegistry: CommandRegistryService,
    @Inject(CommandExecutorService) private readonly commandExecutor: CommandExecutorService,
    // Vestigial arity-only slot: the router is the sole execution authority and the gateway
    // never routes. The union type erases to `Object`, so with `@Optional()` and no `@Inject`
    // this resolves to `null` in every graph (production and test) and is never invoked.
    @Optional() private readonly routingEngine: RoutingEngineService | null = null,
    @Optional()
    @Inject(CommandAuthorizationService)
    private readonly commandAuthorization: CommandAuthorizationService | null = null,
    @Optional()
    @Inject(RuntimeProviderService)
    private readonly runtimeRegistry: RuntimeProviderService | null = null,
    @Optional()
    @Inject(DurableSessionService)
    private readonly durableSessions: DurableSessionService | null = null,
    @Optional()
    @Inject(RUNTIME_PROVIDER_AUDIT_SINK)
    private readonly runtimeAudit: RuntimeAuditSink | null = null,
  ) {}

  afterInit(): void {
    this.logger.log('Chat WebSocket gateway initialized');
  }

  async handleConnection(client: Socket): Promise<void> {
    const serviceToken = client.handshake.auth['discordServiceToken'];
    if (validateDiscordServiceToken(serviceToken, process.env['DISCORD_SERVICE_TOKEN'])) {
      client.data.discordService = true;
      this.logger.log(`Authenticated Discord service connected: ${client.id}`);
      return;
    }

    const session = await validateSocketSession(client.handshake.headers, this.auth);
    if (!session) {
      this.logger.warn(`Rejected unauthenticated WebSocket client: ${client.id}`);
      client.disconnect();
      return;
    }

    client.data.user = session.user;
    client.data.session = session.session;
    this.logger.log(`Client connected: ${client.id}`);
    client.emit('commands:manifest', { manifest: this.commandRegistry.getManifest() });

    // Send-protocol advertisement (Task 5): a conversation id or harness selection never proves the
    // connected Gateway handles a given wire event, so after BetterAuth authentication and
    // user/session assignment advertise — exactly once, bound to this connection — which send event
    // the browser may use. Legacy mode handles `message` (`legacy-message`); `pi-rpc` fails that
    // handler closed and its authenticated `turn:send` handler lands in Task 15, so it advertises
    // `unavailable` and never `turn-send`. Capability is routing information, never authorization.
    client.emit('chat:send-capability', {
      protocol: this.runtime.runtimeMode === 'pi-rpc' ? 'unavailable' : 'legacy-message',
      connectionId: client.id,
    });
  }

  handleDisconnect(client: Socket): void {
    this.logger.log(`Client disconnected: ${client.id}`);
    for (const [key, session] of this.clientSessions) {
      if (session.clientId !== client.id) continue;
      // The lease owns listener + channel teardown; dispose is idempotent and non-throwing.
      void session.lease.dispose();
      this.clientSessions.delete(key);
      this.textEgressBuffers.delete(key);
      this.thinkingEgressBuffers.delete(key);
      this.overflowedEgress.delete(`${key}:agent:text`);
      this.overflowedEgress.delete(`${key}:agent:thinking`);
    }
  }

  private clientConversationKey(client: Pick<Socket, 'id'>, conversationId: string): string {
    return `${client.id}\u0000${conversationId}`;
  }

  private getClientScope(client: Socket): ActorTenantScope | null {
    const user = client.data.user as AuthenticatedUserLike | undefined;
    if (!user?.id) return null;
    return scopeFromUser(user);
  }

  private modelOverrideKey(conversationId: string, scope: ActorTenantScope): string {
    return `${scope.tenantId}:${scope.userId}:${conversationId}`;
  }

  private scopesEqual(a: ActorTenantScope, b: ActorTenantScope): boolean {
    return a.userId === b.userId && a.tenantId === b.tenantId;
  }

  @SubscribeMessage('message')
  async handleMessage(
    @ConnectedSocket() client: Socket,
    @MessageBody() rawData: unknown,
  ): Promise<void> {
    // Verified-Discord ingress and browser turns are distinct trust surfaces: the service flag
    // is set only after handshake-token auth at handleConnection. A forged envelope from a
    // non-service socket falls through to the browser path, where it is rejected as malformed.
    if (client.data.discordService) {
      await this.handleVerifiedDiscordSend(client, rawData);
      return;
    }
    await this.handleBrowserSend(client, rawData);
  }

  private async handleBrowserSend(client: Socket, rawData: unknown): Promise<void> {
    // Fail a legacy browser turn closed under pi-rpc BEFORE parsing the payload — never fall back.
    if (this.runtime.runtimeMode === 'pi-rpc') {
      const conversationId =
        typeof rawData === 'object' &&
        rawData !== null &&
        typeof (rawData as { conversationId?: unknown }).conversationId === 'string'
          ? (rawData as { conversationId: string }).conversationId
          : undefined;
      client.emit('error', {
        conversationId,
        code: 'runtime_unsupported',
        retryable: false,
        error: 'Browser chat is not available on this deployment.',
      });
      return;
    }

    if (!isChatSocketMessage(rawData)) {
      this.logger.warn(`Rejected malformed chat message from ${client.id}`);
      return;
    }
    const data = rawData;
    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
    // same server-derived channel id, so disposing after would tear down the new subscription.
    const key = this.clientConversationKey(client, conversationId);
    await this.disposeExistingSession(key);

    const stream: LegacyRuntimeStream = {
      channelId: `websocket:${client.id}`,
      onEvent: (event: LegacyRuntimeEvent): void => this.relayEvent(client, conversationId, event),
    };

    const prepared = await this.runtime.prepareLegacySocketTurn(
      ownConversation(conversationId, scope),
      {
        content: data.content,
        ...(data.provider ? { provider: data.provider } : {}),
        ...(data.modelId ? { modelId: data.modelId } : {}),
        ...(data.agentId ? { agentId: data.agentId } : {}),
        ...(data.attachments ? { attachments: data.attachments } : {}),
      },
      stream,
    );
    if (!prepared.ok) {
      client.emit('error', {
        conversationId,
        code: prepared.code,
        retryable: prepared.retryable,
        error: 'Failed to start agent session. Please try again.',
      });
      return;
    }

    this.registerClientSession(client, conversationId, stream.channelId, prepared.value, scope);
    // Persist the user turn BEFORE acknowledging or dispatching. If persistence fails, abort the
    // turn: tear down the just-prepared lease and surface the failure — never ack-then-lose.
    const persisted = await this.persistUserMessage(
      conversationId,
      scope.userId,
      data.content,
      data.attachments,
    );
    if (!persisted) {
      await this.disposeExistingSession(key);
      client.emit('error', {
        conversationId,
        code: 'persist_failed',
        retryable: true,
        error: 'Your message could not be saved. Please try again.',
      });
      return;
    }

    client.emit('session:info', { conversationId, ...prepared.value.presentation });
    client.emit('message:ack', { conversationId, messageId: uuid() });

    const dispatched = await prepared.value.dispatch();
    if (!dispatched.ok) {
      client.emit('error', {
        conversationId,
        error: 'The agent failed to process your message. Please try again.',
      });
    }
  }

  private async handleVerifiedDiscordSend(client: Socket, rawData: unknown): Promise<void> {
    if (!isDiscordIngressEnvelope(rawData)) {
      this.logger.warn(`Rejected malformed Discord ingress from ${client.id}`);
      return;
    }
    const ingress = this.resolveDiscordIngress(client, rawData, 'send', false);
    if (!ingress) return;

    const discordServiceUserId = process.env['DISCORD_SERVICE_USER_ID'];
    if (!discordServiceUserId) {
      this.logger.warn(
        `Rejected Discord ingress without configured service owner from ${client.id}`,
      );
      return;
    }
    const binding = this.discordBindingFor(ingress, 'send');
    if (!binding) {
      this.logger.warn(`Rejected unpaired Discord ingress from ${client.id}`);
      return;
    }

    const scope: ActorTenantScope = {
      userId: discordServiceUserId,
      tenantId: process.env['DISCORD_SERVICE_TENANT_ID'] ?? discordServiceUserId,
    };
    const conversationId = ingress.conversationId;
    const attachments = ingress.attachments?.map(
      (attachment): ChannelAttachmentDto => ({
        id: attachment.id,
        name: attachment.name,
        url: attachment.url,
        mimeType: attachment.contentType,
        ...(attachment.sizeBytes !== undefined ? { sizeBytes: attachment.sizeBytes } : {}),
      }),
    );

    // 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 must match the
    // binding EXACTLY on BOTH id and name — a record whose id differs from the binding's
    // agentConfigId (an aliased/substituted lookup) is rejected as firmly as a name mismatch. A
    // missing record, an id mismatch, 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.id !== binding.agentConfigId || 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
    // 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}`,
    );

    const key = this.clientConversationKey(client, conversationId);
    await this.disposeExistingSession(key);

    const stream: LegacyRuntimeStream = {
      channelId: `websocket:${client.id}`,
      onEvent: (event: LegacyRuntimeEvent): void => this.relayEvent(client, conversationId, event),
    };

    // 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,
      content: ingress.content,
      ...(attachments && attachments.length > 0 ? { attachments } : {}),
      correlationId: ingress.correlationId,
      discordMessageId: ingress.messageId,
      discordUserId: ingress.userId,
    });

    const prepared = await this.runtime.dispatchVerifiedDiscordIngress(context, stream);
    if (!prepared.ok) {
      client.emit('error', {
        conversationId,
        code: prepared.code,
        retryable: prepared.retryable,
        error: 'Failed to start agent session. Please try again.',
      });
      return;
    }

    this.registerClientSession(client, conversationId, stream.channelId, prepared.value, scope);
    // Persist BEFORE acknowledging or dispatching; on persistence failure abort the verified turn
    // (tear down the lease, surface the error) rather than ack-then-lose the Discord message.
    const persisted = await this.persistUserMessage(
      conversationId,
      scope.userId,
      ingress.content,
      attachments,
      {
        correlationId: ingress.correlationId,
        discordMessageId: ingress.messageId,
        discordUserId: ingress.userId,
      },
    );
    if (!persisted) {
      await this.disposeExistingSession(key);
      client.emit('error', {
        conversationId,
        code: 'persist_failed',
        retryable: true,
        correlationId: ingress.correlationId,
        discordMessageId: ingress.messageId,
        discordUserId: ingress.userId,
        error: 'Your message could not be saved. Please try again.',
      });
      return;
    }

    client.emit('session:info', { conversationId, ...prepared.value.presentation });
    client.emit('message:ack', {
      conversationId,
      messageId: uuid(),
      correlationId: ingress.correlationId,
      discordMessageId: ingress.messageId,
      discordUserId: ingress.userId,
    });

    const dispatched = await prepared.value.dispatch();
    if (!dispatched.ok) {
      client.emit('error', {
        conversationId,
        error: 'The agent failed to process your message. Please try again.',
      });
    }
  }

  private registerClientSession(
    client: Socket,
    conversationId: string,
    channelId: string,
    lease: LegacySocketTurnLease | VerifiedDiscordTurnLease,
    scope: ActorTenantScope,
  ): void {
    this.clientSessions.set(this.clientConversationKey(client, conversationId), {
      clientId: client.id,
      conversationId,
      channelId,
      lease,
      assistantText: '',
      toolCalls: [],
      pendingToolCalls: new Map(),
      scope,
    });
  }

  private async disposeExistingSession(key: string): Promise<void> {
    const existing = this.clientSessions.get(key);
    if (!existing) return;
    await existing.lease.dispose();
    this.clientSessions.delete(key);
  }

  private async persistUserMessage(
    conversationId: string,
    userId: string | undefined,
    content: string,
    attachments: readonly ChannelAttachmentDto[] | undefined,
    discord?: { correlationId: string; discordMessageId: string; discordUserId: string },
  ): Promise<boolean> {
    if (!userId) return true;
    await this.ensureConversation(conversationId, userId);
    try {
      const saved = await this.brain.conversations.addMessage(
        {
          conversationId,
          role: 'user',
          content: redactSensitiveContent(content).content,
          metadata: {
            timestamp: new Date().toISOString(),
            ...(discord
              ? {
                  correlationId: discord.correlationId,
                  discordMessageId: discord.discordMessageId,
                  discordUserId: discord.discordUserId,
                }
              : {}),
            ...(attachments && attachments.length > 0
              ? {
                  channelAttachments: attachments.map(
                    (attachment): ChannelAttachmentDto => ({
                      ...attachment,
                      name: redactSensitiveContent(attachment.name).content,
                      url: redactSensitiveContent(attachment.url).content,
                    }),
                  ),
                }
              : {}),
            classifications: redactSensitiveContent(content).classifications,
          },
        },
        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(
        `Failed to persist user message for conversation=${conversationId}`,
        err instanceof Error ? err.stack : String(err),
      );
      return false;
    }
  }

  @SubscribeMessage('set:thinking')
  handleSetThinking(
    @ConnectedSocket() client: Socket,
    @MessageBody() data: SetThinkingPayload,
  ): void {
    const scope = this.getClientScope(client);
    if (!scope) {
      client.emit('error', {
        conversationId: data.conversationId,
        error: 'Authenticated user scope is required.',
      });
      return;
    }

    const result = this.runtime.setLegacyThinking(
      ownConversation(data.conversationId, scope),
      data.level,
    );
    if (!result.ok) {
      if (result.code === 'thinking_level_invalid') {
        client.emit('error', {
          conversationId: data.conversationId,
          error: `Invalid thinking level "${data.level}". Available: ${result.availableThinkingLevels.join(', ')}`,
        });
      } else if (result.code === 'conversation_unavailable') {
        client.emit('error', {
          conversationId: data.conversationId,
          error: 'No active session for this conversation.',
        });
      } else {
        client.emit('error', {
          conversationId: data.conversationId,
          error: 'Failed to set thinking level.',
        });
      }
      return;
    }

    this.logger.log(
      `Thinking level set to "${data.level}" for conversation ${data.conversationId}`,
    );

    client.emit('session:info', { conversationId: data.conversationId, ...result.value });
  }

  @SubscribeMessage('abort')
  async handleAbort(
    @ConnectedSocket() client: Socket,
    @MessageBody() data: AbortPayload,
  ): Promise<void> {
    const conversationId = data.conversationId;
    this.logger.log(`Abort requested by ${client.id} for conversation ${conversationId}`);

    const scope = this.getClientScope(client);
    if (!scope) {
      client.emit('error', { conversationId, error: 'Authenticated user scope is required.' });
      return;
    }

    const result = await this.runtime.abortLegacyTurn(ownConversation(conversationId, scope));
    if (!result.ok) {
      client.emit('error', {
        conversationId,
        error:
          result.code === 'conversation_unavailable'
            ? 'No active session to abort.'
            : 'Failed to abort the agent operation.',
      });
      return;
    }
    this.logger.log(`Agent session ${conversationId} aborted successfully`);
  }

  @SubscribeMessage('command:execute')
  async handleCommandExecute(
    @ConnectedSocket() client: Socket,
    @MessageBody() payload: SlashCommandPayload,
  ): Promise<void> {
    const scope = this.getClientScope(client);
    if (!scope) {
      client.emit('command:result', {
        command: payload.command,
        conversationId: payload.conversationId,
        success: false,
        message: 'Authenticated user scope is required.',
      });
      return;
    }

    // Task 5 (G3): under pi-rpc there is no embedded chat session, so embedded slash-commands are
    // unsupported. Fail closed BEFORE the executor — never fall back to embedded execution — while
    // runtime-independent audited system commands (e.g. /reload) still pass through.
    if (
      this.runtime.runtimeMode === 'pi-rpc' &&
      !RUNTIME_INDEPENDENT_COMMANDS.has(payload.command)
    ) {
      client.emit('command:result', {
        command: payload.command,
        conversationId: payload.conversationId,
        success: false,
        message: 'Slash commands are not available on this deployment.',
      });
      return;
    }

    const result = await this.commandExecutor.execute(payload, scope);
    client.emit('command:result', result);
  }

  @SubscribeMessage('command:approve')
  async handleCommandApproval(
    @ConnectedSocket() client: Socket,
    @MessageBody() payload: SlashCommandPayload,
  ): Promise<void> {
    const scope = this.getClientScope(client);
    const approval = scope ? await this.commandExecutor.createApproval(payload, scope) : null;
    const result: SlashCommandApprovalResultPayload = approval
      ? {
          command: payload.command,
          conversationId: payload.conversationId,
          success: true,
          approvalId: approval.approvalId,
          expiresAt: approval.expiresAt,
        }
      : {
          command: payload.command,
          conversationId: payload.conversationId,
          success: false,
          message: 'Not authorized to approve this command.',
        };
    client.emit('command:approval', result);
  }

  broadcastReload(payload: SystemReloadPayload): void {
    this.server.emit('system:reload', payload);
    this.logger.log('Broadcasted system:reload to all connected clients');
  }

  /**
   * Set a per-conversation model override (M4-007 / M5-002).
   * When set, the routing engine is bypassed and the specified model is used.
   * Pass null to clear the override and resume automatic routing.
   * M5-005: Emits session:info to clients subscribed to this conversation when a model is set.
   * M5-007: Records a model switch in session metrics.
   */
  setModelOverride(
    conversationId: string,
    modelName: string | null,
    scope: ActorTenantScope,
  ): void {
    const key = this.modelOverrideKey(conversationId, scope);
    if (modelName) {
      modelOverrides.set(key, modelName);
      this.logger.log(`Model override set: conversation=${conversationId} model="${modelName}"`);

      // M5-002: Update the live session's modelId so session:info reflects the new model immediately
      this.runtime.applyLegacyModelOverride(ownConversation(conversationId, scope), modelName);

      // M5-005: Broadcast session:info to all clients subscribed to this conversation
      this.broadcastSessionInfo(conversationId, scope);
    } else {
      modelOverrides.delete(key);
      this.logger.log(`Model override cleared: conversation=${conversationId}`);
    }
  }

  /**
   * Return the active model override for a conversation, or undefined if none.
   */
  getModelOverride(conversationId: string, scope: ActorTenantScope): string | undefined {
    return modelOverrides.get(this.modelOverrideKey(conversationId, scope));
  }

  /**
   * M5-005: Broadcast session:info to all clients currently subscribed to a conversation.
   * Called on model or agent switch to ensure the TUI TopBar updates immediately.
   */
  broadcastSessionInfo(
    conversationId: string,
    scope: ActorTenantScope,
    extra?: { agentName?: string; routingDecision?: RoutingDecisionInfo },
  ): void {
    const result = this.runtime.readLegacySessionPresentation(
      ownConversation(conversationId, scope),
    );
    if (!result.ok) return;

    const resolvedAgentName = extra?.agentName ?? result.value.agentName;
    const payload = {
      conversationId,
      ...result.value,
      ...(resolvedAgentName ? { agentName: resolvedAgentName } : {}),
      ...(extra?.routingDecision ? { routingDecision: extra.routingDecision } : {}),
    };

    // Emit to all clients currently subscribed to this conversation
    for (const session of this.clientSessions.values()) {
      if (session.conversationId === conversationId && this.scopesEqual(session.scope, scope)) {
        const socket = this.server.sockets.sockets.get(session.clientId);
        if (socket?.connected) {
          socket.emit('session:info', payload);
        }
      }
    }
  }

  /**
   * Ensure a conversation record exists in the DB.
   * Creates it if absent — safe to call concurrently since a duplicate insert
   * would fail on the PK constraint and be caught here.
   */
  @SubscribeMessage('discord:approve')
  async handleDiscordApproval(
    @ConnectedSocket() client: Socket,
    @MessageBody() envelope: DiscordIngressEnvelope,
  ): Promise<void> {
    if (!client.data.discordService) return;
    const ingress = this.resolveDiscordIngress(client, envelope, 'approve');
    const isApprovalCommand = /^\/approve\s*$/i.test(ingress?.content ?? '');
    const tenantId = process.env['DISCORD_SERVICE_TENANT_ID']?.trim();
    if (
      !ingress ||
      !isApprovalCommand ||
      !tenantId ||
      !this.commandAuthorization ||
      !this.durableSessions
    )
      return;
    const binding = this.discordBindingFor(ingress, 'approve');
    const actorId = binding && resolveDiscordInteractionActorId(binding, ingress.userId);
    const agentName = binding?.instanceId;
    if (!actorId || !agentName) {
      this.logger.warn(
        `Rejected Discord approval without a matching runtime agent from ${client.id}`,
      );
      client.emit('discord:approval', {
        correlationId: ingress.correlationId,
        success: false,
        approvalId: undefined,
        expiresAt: undefined,
      });
      return;
    }
    let snapshot;
    try {
      snapshot = await this.durableSessions.getSnapshot(ingress.conversationId, {
        actorScope: { userId: actorId, tenantId },
        channelId: ingress.channelId,
        correlationId: ingress.correlationId,
      });
    } catch {
      client.emit('discord:approval', {
        correlationId: ingress.correlationId,
        success: false,
        approvalId: undefined,
        expiresAt: undefined,
      });
      return;
    }
    if (snapshot.identity.agentName !== agentName) {
      client.emit('discord:approval', {
        correlationId: ingress.correlationId,
        success: false,
        approvalId: undefined,
        expiresAt: undefined,
      });
      return;
    }
    const approval = await this.commandAuthorization.createRuntimeTerminationApproval({
      providerId: snapshot.identity.providerId,
      sessionId: snapshot.identity.runtimeSessionId,
      actorId,
      tenantId,
      channelId: ingress.channelId,
      correlationId: this.discordRuntimeActionCorrelation(
        binding.instanceId,
        ingress,
        snapshot.identity.providerId,
        snapshot.identity.runtimeSessionId,
      ),
      agentName,
    });
    if (!approval) {
      await this.runtimeAudit?.record({
        providerId: snapshot.identity.providerId,
        operation: 'session.terminate',
        outcome: 'denied',
        actorId,
        tenantId,
        channelId: ingress.channelId,
        correlationId: this.discordRuntimeActionCorrelation(
          binding.instanceId,
          ingress,
          snapshot.identity.providerId,
          snapshot.identity.runtimeSessionId,
        ),
        resourceId: snapshot.identity.runtimeSessionId,
        errorCode: 'policy_denied',
      });
    }
    client.emit('discord:approval', {
      correlationId: ingress.correlationId,
      success: approval !== null,
      approvalId: approval?.approvalId,
      expiresAt: approval?.expiresAt,
    });
  }

  @SubscribeMessage('discord:stop')
  async handleDiscordStop(
    @ConnectedSocket() client: Socket,
    @MessageBody() envelope: DiscordIngressEnvelope,
  ): Promise<void> {
    if (!client.data.discordService) return;
    const ingress = this.resolveDiscordIngress(client, envelope, 'stop');
    const approvalRef = /^\/stop\s+([^\s]+)$/i.exec(ingress?.content ?? '')?.[1];
    const tenantId = process.env['DISCORD_SERVICE_TENANT_ID']?.trim();
    if (!ingress || !approvalRef || !tenantId || !this.runtimeRegistry || !this.durableSessions)
      return;
    const binding = this.discordBindingFor(ingress, 'stop');
    const actorId = binding && resolveDiscordInteractionActorId(binding, ingress.userId);
    if (!actorId) return;

    try {
      const context = {
        actorScope: { userId: actorId, tenantId },
        channelId: ingress.channelId,
        correlationId: ingress.correlationId,
      };
      const snapshot = await this.durableSessions.getSnapshot(ingress.conversationId, context);
      if (snapshot.identity.agentName !== binding.instanceId) throw new Error('agent mismatch');
      // RuntimeProviderService consumes the durable approval exactly once using the
      // provisioned approving-admin identity, never the Discord service account.
      await this.runtimeRegistry.terminate(
        snapshot.identity.providerId,
        snapshot.identity.runtimeSessionId,
        approvalRef,
        {
          ...context,
          correlationId: this.discordRuntimeActionCorrelation(
            binding.instanceId,
            ingress,
            snapshot.identity.providerId,
            snapshot.identity.runtimeSessionId,
          ),
        },
      );
      client.emit('discord:stop', { correlationId: ingress.correlationId, success: true });
    } catch {
      client.emit('discord:stop', { correlationId: ingress.correlationId, success: false });
    }
  }

  /**
   * Correlates the immutable termination target rather than either Discord message.
   * Approval and stop are distinct ingress events, but must consume the same seven-field action.
   */
  private discordRuntimeActionCorrelation(
    instanceId: string,
    ingress: DiscordIngressPayload,
    providerId: string,
    sessionId: string,
  ): string {
    const target = [
      instanceId,
      ingress.guildId,
      ingress.channelId,
      ingress.conversationId,
      providerId,
      sessionId,
    ];
    return `discord-action:v1:${createHash('sha256').update(JSON.stringify(target)).digest('hex')}`;
  }

  private resolveDiscordIngress(
    client: Socket,
    envelope: DiscordIngressEnvelope,
    operation: 'send' | 'approve' | 'stop' = 'send',
    claimReplay = true,
  ): DiscordIngressPayload | null {
    const payload = verifyDiscordIngressEnvelope(
      envelope,
      process.env['DISCORD_SERVICE_TOKEN'] ?? '',
      {
        guildIds: this.readDiscordAllowlist('DISCORD_ALLOWED_GUILD_IDS'),
        channelIds: this.readDiscordAllowlist('DISCORD_ALLOWED_CHANNEL_IDS'),
        userIds: this.readDiscordAllowlist('DISCORD_ALLOWED_USER_IDS'),
      },
    );
    if (!payload) {
      this.logger.warn(`Rejected invalid Discord ingress envelope from ${client.id}`);
      return null;
    }
    try {
      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}`,
      );
      return null;
    }
    // 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}`,
      );
      return null;
    }
    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(',')
      .map((id: string): string => id.trim())
      .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<boolean> {
    try {
      if (suppliedConversationId !== undefined) {
        const owned = await this.brain.conversations.findById(suppliedConversationId, userId);
        return owned !== undefined;
      }
      // Minting a new conversation must actually yield the durable record we asked for before any
      // runtime effect runs (finding 1). A `create` that resolves nullish, or returns a record that
      // is not this exact id owned by this user, is a persistence failure — fail closed so the caller
      // never dispatches/persists against an unpersisted or mis-scoped conversation.
      const created = await this.brain.conversations.create({ id: conversationId, userId });
      return created != null && created.id === conversationId && created.userId === userId;
    } 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<void> {
    try {
      const existing = await this.brain.conversations.findById(conversationId, userId);
      if (!existing) {
        await this.brain.conversations.create({
          id: conversationId,
          userId,
        });
      }
    } catch (err) {
      this.logger.error(
        `Failed to ensure conversation record for conversation=${conversationId}`,
        err instanceof Error ? err.stack : String(err),
      );
    }
  }

  /**
   * Load prior conversation messages from DB for context injection on session resume (M1-004).
   * Returns an empty array when no history exists, the conversation is not owned by the user,
   * or userId is not provided.
   */
  private async loadConversationHistory(
    conversationId: string,
    userId: string | undefined,
  ): Promise<ConversationHistoryMessage[]> {
    if (!userId) return [];

    try {
      const messages = await this.brain.conversations.findMessages(conversationId, userId);
      if (messages.length === 0) return [];

      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}`,
        err instanceof Error ? err.stack : String(err),
      );
      return [];
    }
  }

  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,
    eventName: 'agent:text' | 'agent:thinking',
    buffers: Map<string, string>,
    delta: string,
  ): void {
    const sessionKey = this.clientConversationKey(client, conversationId);
    const key = this.egressKey(client, conversationId, eventName);
    if (this.overflowedEgress.has(key)) return;

    const buffered = `${buffers.get(sessionKey) ?? ''}${delta}`;
    if (buffered.length > MAX_REDACTION_BUFFER_LENGTH) {
      buffers.delete(sessionKey);
      this.overflowedEgress.add(key);
      client.emit(eventName, { conversationId, text: '[REDACTED_STREAM_OVERFLOW]' });
      return;
    }

    buffers.set(sessionKey, buffered);
    this.flushRedactedEgress(client, conversationId, eventName, buffers, false);
  }

  /**
   * Holds any suffix that could become a secret, email, or phone number after a
   * later stream chunk. This avoids relying on downstream redaction after data
   * has already reached the socket.
   */
  private flushRedactedEgress(
    client: Socket,
    conversationId: string,
    eventName: 'agent:text' | 'agent:thinking',
    buffers: Map<string, string>,
    final: boolean,
  ): void {
    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(sessionKey) ?? '';
    const releaseLength = final ? buffered.length : this.safeRedactionPrefixLength(buffered);
    const released = buffered.slice(0, releaseLength);
    const pending = buffered.slice(releaseLength);

    if (pending) {
      buffers.set(sessionKey, pending);
    } else {
      buffers.delete(sessionKey);
    }

    if (released) {
      client.emit(eventName, {
        conversationId,
        text: redactSensitiveContent(released).content,
      });
    }
  }

  private safeRedactionPrefixLength(content: string): number {
    let retainedFrom = content.length;

    // Retain the current token because it may become a split secret or email.
    const token = /(?:^|\s)(\S*)$/.exec(content);
    if (token) {
      const matched = token[0] ?? '';
      const trailingToken = token[1] ?? '';
      retainedFrom = token.index + matched.length - trailingToken.length;
    }

    // The secret classifier accepts whitespace around ':' and '=', so preserve
    // a pending label until its value and delimiter are both complete.
    const pendingSecretLabel =
      /(?:^|[^A-Za-z0-9_])((?:api[_-]?key|token|password|secret|bearer|authorization)\s*)$/i.exec(
        content,
      );
    if (pendingSecretLabel) {
      const label = pendingSecretLabel[1] ?? '';
      retainedFrom = Math.min(
        retainedFrom,
        pendingSecretLabel.index + pendingSecretLabel[0].length - label.length,
      );
    }

    const secretLabel = /(?:api[_-]?key|token|password|secret|authorization)\s*[:=]\s*$/i.exec(
      content,
    );
    if (secretLabel) {
      retainedFrom = Math.min(retainedFrom, secretLabel.index);
    }

    // Phone numbers can contain whitespace and punctuation; preserve the full
    // trailing numeric candidate until a non-phone character establishes a boundary.
    const phone = /(?:^|[^A-Za-z0-9_])(\+?\d[\d(). -]*)$/.exec(content);
    if (phone) {
      const matched = phone[0] ?? '';
      const trailingPhoneCandidate = phone[1] ?? '';
      retainedFrom = Math.min(
        retainedFrom,
        phone.index + matched.length - trailingPhoneCandidate.length,
      );
    }

    const privateKeyStart = content.lastIndexOf('-----BEGIN');
    if (privateKeyStart >= 0) {
      const privateKey = content.slice(privateKeyStart);
      if (/-----END(?: [A-Z]+)* KEY-----/.test(privateKey)) {
        // Release the complete block in one pass so the full-block classifier can redact it.
        retainedFrom = content.length;
      } else {
        retainedFrom = Math.min(retainedFrom, privateKeyStart);
      }
    }

    return retainedFrom;
  }

  private egressKey(
    client: Socket,
    conversationId: string,
    eventName: 'agent:text' | 'agent:thinking',
  ): string {
    return `${this.clientConversationKey(client, conversationId)}:${eventName}`;
  }

  /**
   * Relay one normalized {@link LegacyRuntimeEvent} to the socket, preserving the exact legacy
   * egress event names and shapes. The runtime owns session/token bookkeeping — the gateway never
   * reads a pi session or records usage here; usage arrives verbatim on the `settled` event.
   */
  private relayEvent(client: Socket, conversationId: string, event: LegacyRuntimeEvent): void {
    if (!client.connected) {
      this.logger.warn(
        `Dropping event ${event.type} for disconnected client=${client.id}, conversation=${conversationId}`,
      );
      return;
    }

    const sessionKey = this.clientConversationKey(client, conversationId);
    switch (event.type) {
      case 'started': {
        // Reset accumulation buffers for the new turn
        const cs = this.clientSessions.get(sessionKey);
        if (cs) {
          cs.assistantText = '';
          cs.toolCalls = [];
          cs.pendingToolCalls.clear();
        }
        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 'text_delta': {
        // Keep raw stream material in memory only; persist and emit only redacted text.
        const cs = this.clientSessions.get(sessionKey);
        if (cs) {
          cs.assistantText += event.text;
        }
        this.appendAndFlushRedactedEgress(
          client,
          conversationId,
          'agent:text',
          this.textEgressBuffers,
          event.text,
        );
        break;
      }

      case 'thinking_delta': {
        this.appendAndFlushRedactedEgress(
          client,
          conversationId,
          'agent:thinking',
          this.thinkingEgressBuffers,
          event.text,
        );
        break;
      }

      case 'tool_started': {
        // Track pending tool call for later recording
        const cs = this.clientSessions.get(sessionKey);
        if (cs) {
          cs.pendingToolCalls.set(event.toolCallId, {
            toolName: event.toolName,
            args: undefined,
          });
        }
        client.emit('agent:tool:start', {
          conversationId,
          toolCallId: event.toolCallId,
          toolName: event.toolName,
        });
        break;
      }

      case 'tool_finished': {
        // Finalise tool call record
        const cs = this.clientSessions.get(sessionKey);
        if (cs) {
          const pending = cs.pendingToolCalls.get(event.toolCallId);
          cs.toolCalls.push({
            toolCallId: event.toolCallId,
            toolName: event.toolName,
            args: pending?.args ?? null,
            isError: event.isError,
          });
          cs.pendingToolCalls.delete(event.toolCallId);
        }
        client.emit('agent:tool:end', {
          conversationId,
          toolCallId: event.toolCallId,
          toolName: event.toolName,
          isError: event.isError,
        });
        break;
      }

      case 'settled': {
        this.flushRedactedEgress(
          client,
          conversationId,
          'agent:text',
          this.textEgressBuffers,
          true,
        );
        this.flushRedactedEgress(
          client,
          conversationId,
          'agent:thinking',
          this.thinkingEgressBuffers,
          true,
        );
        client.emit('agent:end', { conversationId, usage: event.usage });

        // Persist the assistant message with metadata
        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> = {
            timestamp: new Date().toISOString(),
            model: event.usage?.modelId ?? 'unknown',
            provider: event.usage?.provider ?? 'unknown',
            toolCalls: cs.toolCalls,
            ...(event.usage?.tokens ? { tokenUsage: event.usage.tokens } : {}),
          };

          this.brain.conversations
            .addMessage(
              {
                conversationId,
                role: 'assistant',
                content: redactSensitiveContent(cs.assistantText).content,
                metadata: {
                  ...metadata,
                  classifications: redactSensitiveContent(cs.assistantText).classifications,
                },
              },
              userId,
            )
            .catch((err: unknown) => {
              this.logger.error(
                `Failed to persist assistant message for conversation=${conversationId}`,
                err instanceof Error ? err.stack : String(err),
              );
            });

          // Reset accumulation
          cs.assistantText = '';
          cs.toolCalls = [];
          cs.pendingToolCalls.clear();
        }
        break;
      }
    }
  }
}