stack/apps/orchestrator/src/api/mission-control/mission-control.service.ts

import { Injectable, NotFoundException } from "@nestjs/common";
import type { AgentMessage, AgentSession, IAgentProvider, InjectResult } from "@mosaic/shared";
import type { Prisma } from "@prisma/client";
import { PrismaService } from "../../prisma/prisma.service";
import { AgentProviderRegistry } from "../agents/agent-provider.registry";

type MissionControlAction = "inject" | "pause" | "resume" | "kill";

const DEFAULT_OPERATOR_ID = "mission-control";

export interface AuditLogEntry {
  id: string;
  userId: string;
  sessionId: string;
  provider: string;
  action: string;
  content: string | null;
  metadata: Prisma.JsonValue;
  createdAt: Date;
}

export interface MissionControlAuditLogPage {
  items: AuditLogEntry[];
  total: number;
  page: number;
  pages: number;
}

@Injectable()
export class MissionControlService {
  constructor(
    private readonly registry: AgentProviderRegistry,
    private readonly prisma: PrismaService
  ) {}

  listSessions(): Promise<AgentSession[]> {
    return this.registry.listAllSessions();
  }

  async getSession(sessionId: string): Promise<AgentSession> {
    const resolved = await this.registry.getProviderForSession(sessionId);
    if (!resolved) {
      throw new NotFoundException(`Session ${sessionId} not found`);
    }

    return resolved.session;
  }

  async getMessages(sessionId: string, limit?: number, before?: string): Promise<AgentMessage[]> {
    const { provider } = await this.getProviderForSessionOrThrow(sessionId);
    return provider.getMessages(sessionId, limit, before);
  }

  async getAuditLog(
    sessionId: string | undefined,
    page: number,
    limit: number
  ): Promise<MissionControlAuditLogPage> {
    const normalizedSessionId = sessionId?.trim();
    const where: Prisma.OperatorAuditLogWhereInput =
      normalizedSessionId && normalizedSessionId.length > 0
        ? { sessionId: normalizedSessionId }
        : {};

    const [total, items] = await this.prisma.$transaction([
      this.prisma.operatorAuditLog.count({ where }),
      this.prisma.operatorAuditLog.findMany({
        where,
        orderBy: { createdAt: "desc" },
        skip: (page - 1) * limit,
        take: limit,
      }),
    ]);

    return {
      items,
      total,
      page,
      pages: total === 0 ? 0 : Math.ceil(total / limit),
    };
  }

  async injectMessage(
    sessionId: string,
    message: string,
    operatorId = DEFAULT_OPERATOR_ID
  ): Promise<InjectResult> {
    const { provider } = await this.getProviderForSessionOrThrow(sessionId);
    const result = await provider.injectMessage(sessionId, message);

    await this.writeOperatorAuditLog({
      sessionId,
      providerId: provider.providerId,
      operatorId,
      action: "inject",
      content: message,
      payload: { message },
    });

    return result;
  }

  async pauseSession(sessionId: string, operatorId = DEFAULT_OPERATOR_ID): Promise<void> {
    const { provider } = await this.getProviderForSessionOrThrow(sessionId);
    await provider.pauseSession(sessionId);

    await this.writeOperatorAuditLog({
      sessionId,
      providerId: provider.providerId,
      operatorId,
      action: "pause",
      payload: {},
    });
  }

  async resumeSession(sessionId: string, operatorId = DEFAULT_OPERATOR_ID): Promise<void> {
    const { provider } = await this.getProviderForSessionOrThrow(sessionId);
    await provider.resumeSession(sessionId);

    await this.writeOperatorAuditLog({
      sessionId,
      providerId: provider.providerId,
      operatorId,
      action: "resume",
      payload: {},
    });
  }

  async killSession(
    sessionId: string,
    force = true,
    operatorId = DEFAULT_OPERATOR_ID
  ): Promise<void> {
    const { provider } = await this.getProviderForSessionOrThrow(sessionId);
    await provider.killSession(sessionId, force);

    await this.writeOperatorAuditLog({
      sessionId,
      providerId: provider.providerId,
      operatorId,
      action: "kill",
      payload: { force },
    });
  }

  async streamMessages(sessionId: string): Promise<AsyncIterable<AgentMessage>> {
    const { provider } = await this.getProviderForSessionOrThrow(sessionId);
    return provider.streamMessages(sessionId);
  }

  private async getProviderForSessionOrThrow(
    sessionId: string
  ): Promise<{ provider: IAgentProvider; session: AgentSession }> {
    const resolved = await this.registry.getProviderForSession(sessionId);

    if (!resolved) {
      throw new NotFoundException(`Session ${sessionId} not found`);
    }

    return resolved;
  }

  private toJsonValue(value: Record<string, unknown>): Prisma.InputJsonValue {
    return value as Prisma.InputJsonValue;
  }

  private async writeOperatorAuditLog(params: {
    sessionId: string;
    providerId: string;
    operatorId: string;
    action: MissionControlAction;
    content?: string;
    payload: Record<string, unknown>;
  }): Promise<void> {
    await this.prisma.operatorAuditLog.create({
      data: {
        sessionId: params.sessionId,
        userId: params.operatorId,
        provider: params.providerId,
        action: params.action,
        ...(params.content !== undefined ? { content: params.content } : {}),
        metadata: this.toJsonValue({ payload: params.payload }),
      },
    });
  }
}