stack/apps/gateway/src/agent/interaction.controller.ts

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<RuntimeStreamEvent> | undefined;
      let cancelled = false;
      void (async (): Promise<void> => {
        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;
  }
}