shaggyandmos-dt-0
216cd72226
ci/woodpecker/push/publish Pipeline failed
Co-authored-by: shaggy <[email protected]>
1524 lines
56 KiB
TypeScript
1524 lines
56 KiB
TypeScript
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;
|
|
}
|
|
}
|
|
}
|
|
}
|