import { Body, Controller, ForbiddenException, Get, Headers, Sse, Inject, Param, Post, Query, UseGuards, UseFilters, } from '@nestjs/common'; import type { RuntimeAttachMode, RuntimeStreamEvent } from '@mosaicstack/types'; import { Observable } from 'rxjs'; import { AuthGuard } from '../auth/auth.guard.js'; import { CurrentUser } from '../auth/current-user.decorator.js'; import { scopeFromUser, type AuthenticatedUserLike } from '../auth/session-scope.js'; import { DurableSessionService } from './durable-session.service.js'; import { RuntimeApprovalDeniedFilter } from './runtime-approval-denied.filter.js'; import { RuntimeProviderService, type RuntimeProviderRequestContext, } from './runtime-provider-registry.service.js'; /** * Authenticated HTTP boundary for operator interaction clients. Identity is * selected from deployment configuration, never a client-side command name. */ @Controller('api/interaction/:agentName') @UseGuards(AuthGuard) @UseFilters(RuntimeApprovalDeniedFilter) export class InteractionController { constructor( @Inject(RuntimeProviderService) private readonly runtime: RuntimeProviderService, @Inject(DurableSessionService) private readonly durable: DurableSessionService, ) {} @Get('sessions') async sessions( @Param('agentName') agentName: string, @Query('provider') providerId: string, @CurrentUser() user: AuthenticatedUserLike, @Headers('x-correlation-id') correlationId?: string, ) { this.assertConfiguredAgent(agentName); return this.runtime.listSessions( this.requiredProvider(providerId), this.context(user, correlationId), ); } @Get('transitional-capabilities') async transitionalCapabilities( @Param('agentName') agentName: string, @Query('provider') providerId: string, @CurrentUser() user: AuthenticatedUserLike, @Headers('x-correlation-id') correlationId?: string, ) { this.assertConfiguredAgent(agentName); return this.runtime.transitionalCapabilityMatrix( this.requiredProvider(providerId), this.context(user, correlationId), ); } @Get('tree') async tree( @Param('agentName') agentName: string, @Query('provider') providerId: string, @CurrentUser() user: AuthenticatedUserLike, @Headers('x-correlation-id') correlationId?: string, ) { this.assertConfiguredAgent(agentName); return this.runtime.getSessionTree( this.requiredProvider(providerId), this.context(user, correlationId), ); } /** * Bind an existing, authorized runtime session to the stable conversation ID. * This is the lifecycle boundary where both runtime identifiers are known. */ @Post('sessions/:sessionId/enroll') async enroll( @Param('agentName') agentName: string, @Param('sessionId') sessionId: string, @Body() body: { providerId?: string; runtimeSessionId?: string } = {}, @CurrentUser() user: AuthenticatedUserLike, @Headers('x-correlation-id') correlationId?: string, ) { this.assertConfiguredAgent(agentName); const providerId = this.requiredProvider(body.providerId ?? ''); const runtimeSessionId = body.runtimeSessionId?.trim(); if (!runtimeSessionId) throw new ForbiddenException('Runtime session identity is required'); const context = this.context(user, correlationId); const sessions = await this.runtime.listSessions(providerId, context); if (!sessions.some((session): boolean => session.id === runtimeSessionId)) { throw new ForbiddenException('Runtime session is not visible to this actor'); } await this.durable.enroll( { agentName, sessionId, tenantId: context.actorScope.tenantId, ownerId: context.actorScope.userId, providerId, runtimeSessionId, }, context, ); return { status: 'enrolled', sessionId }; } @Post('sessions/:sessionId/attach') async attach( @Param('agentName') agentName: string, @Param('sessionId') sessionId: string, @Body() body: { mode?: RuntimeAttachMode } = {}, @CurrentUser() user: AuthenticatedUserLike, @Headers('x-correlation-id') correlationId?: string, ) { this.assertConfiguredAgent(agentName); const context = this.context(user, correlationId); const mode = body.mode ?? 'read'; if (mode !== 'read' && mode !== 'control') { throw new ForbiddenException('Interaction attach mode is invalid'); } const snapshot = await this.durable.getSnapshot(sessionId, context); this.assertSessionAgent(snapshot.identity.agentName, agentName); return this.runtime.attach( snapshot.identity.providerId, snapshot.identity.runtimeSessionId, mode, context, ); } @Sse('sessions/:sessionId/stream') stream( @Param('agentName') agentName: string, @Param('sessionId') sessionId: string, @Query('cursor') cursor: string | undefined, @CurrentUser() user: AuthenticatedUserLike, @Headers('x-correlation-id') correlationId?: string, ): Observable<{ data: RuntimeStreamEvent }> { this.assertConfiguredAgent(agentName); const context = this.context(user, correlationId); return new Observable((subscriber) => { let iterator: AsyncIterator | undefined; let cancelled = false; void (async (): Promise => { try { const snapshot = await this.durable.getSnapshot(sessionId, context); if (cancelled || subscriber.closed) return; this.assertSessionAgent(snapshot.identity.agentName, agentName); iterator = this.runtime .streamSession( snapshot.identity.providerId, snapshot.identity.runtimeSessionId, cursor?.trim() || undefined, context, ) [Symbol.asyncIterator](); if (cancelled || subscriber.closed) { await iterator.return?.(); return; } while (!cancelled && !subscriber.closed) { const next = await iterator.next(); if (next.done || cancelled || subscriber.closed) break; subscriber.next({ data: next.value }); } if (!subscriber.closed) subscriber.complete(); } catch (error: unknown) { if (!subscriber.closed) subscriber.error(error); } })(); return (): void => { cancelled = true; void iterator?.return?.(); }; }); } @Post('sessions/:sessionId/send') async send( @Param('agentName') agentName: string, @Param('sessionId') sessionId: string, @Body() body: { content?: string; idempotencyKey?: string } = {}, @CurrentUser() user: AuthenticatedUserLike, @Headers('x-correlation-id') correlationId?: string, ) { this.assertConfiguredAgent(agentName); if (!body.content?.trim() || !body.idempotencyKey?.trim()) { throw new ForbiddenException('Content and idempotency key are required'); } const context = this.context(user, correlationId); const snapshot = await this.durable.getSnapshot(sessionId, context); this.assertSessionAgent(snapshot.identity.agentName, agentName); const input = { sessionId, content: body.content, idempotencyKey: body.idempotencyKey, correlationId: context.correlationId, context, }; await this.durable.queueProviderSend(input); await this.durable.dispatchProviderOutbox(sessionId, input); return { status: 'queued', sessionId }; } @Post('sessions/:sessionId/stop') async stop( @Param('agentName') agentName: string, @Param('sessionId') sessionId: string, @Body() body: { approvalRef?: string } = {}, @CurrentUser() user: AuthenticatedUserLike, @Headers('x-correlation-id') correlationId?: string, ) { this.assertConfiguredAgent(agentName); if (!body.approvalRef?.trim()) throw new ForbiddenException('Exact-action approval is required'); const context = this.context(user, correlationId); const snapshot = await this.durable.getSnapshot(sessionId, context); this.assertSessionAgent(snapshot.identity.agentName, agentName); await this.runtime.terminate( snapshot.identity.providerId, snapshot.identity.runtimeSessionId, body.approvalRef, context, ); return { status: 'stopped', sessionId }; } @Post('sessions/:sessionId/recover') async recover( @Param('agentName') agentName: string, @Param('sessionId') sessionId: string, @CurrentUser() user: AuthenticatedUserLike, @Headers('x-correlation-id') correlationId?: string, ) { this.assertConfiguredAgent(agentName); const context = this.context(user, correlationId); const snapshot = await this.durable.getSnapshot(sessionId, context); this.assertSessionAgent(snapshot.identity.agentName, agentName); await this.durable.recoverProviderSession(sessionId, { sessionId, content: '', idempotencyKey: `recovery:${context.correlationId}`, correlationId: context.correlationId, context, }); return { status: 'recovered', sessionId }; } private context( user: AuthenticatedUserLike, correlationId?: string, ): RuntimeProviderRequestContext { const requestCorrelationId = correlationId?.trim(); // This non-simple request header is mandatory for mutations. Browser // cross-origin requests cannot set it without a CORS preflight, and the // gateway's allowlist rejects untrusted origins before the handler runs. if (!requestCorrelationId) { throw new ForbiddenException('X-Correlation-Id is required'); } return { actorScope: scopeFromUser(user), channelId: 'cli', correlationId: requestCorrelationId, }; } private assertConfiguredAgent(agentName: string): void { const configured = process.env['MOSAIC_AGENT_NAME']?.trim(); if (!configured || configured !== agentName) { throw new ForbiddenException('Interaction agent is not configured for this request'); } } private assertSessionAgent(sessionAgentName: string, agentName: string): void { if (sessionAgentName !== agentName) throw new ForbiddenException('Interaction session identity mismatch'); } private requiredProvider(providerId: string): string { if (!providerId?.trim()) throw new ForbiddenException('Runtime provider is required'); return providerId; } }