stack/apps/gateway/src/agent/tess-durable-session.service.ts

import { ForbiddenException, Inject, Injectable } from '@nestjs/common';
import { DurableSessionCoordinator, type DurableSessionIdentity } from '@mosaicstack/agent';
import type { TessProviderOutboxDto } from './tess-durable-session.dto.js';
import { TessDurableSessionRepository } from './tess-durable-session.repository.js';
import {
  RuntimeProviderService,
  type RuntimeProviderRequestContext,
} from './runtime-provider-registry.service.js';

/**
 * Scoped gateway boundary for the canonical Tess state machine. It deliberately
 * uses composition: raw state methods cannot be injected into channel, CLI, or
 * MCP adapters without a server-derived actor/tenant/correlation context.
 */
@Injectable()
export class TessDurableSessionService {
  private readonly coordinator: DurableSessionCoordinator;

  constructor(
    @Inject(TessDurableSessionRepository) repository: TessDurableSessionRepository,
    @Inject(RuntimeProviderService) private readonly runtimeProviders: RuntimeProviderService,
  ) {
    this.coordinator = new DurableSessionCoordinator(repository);
  }

  /** Enroll a verified runtime session under the stable cross-surface conversation handle. */
  async enroll(
    identity: DurableSessionIdentity,
    context: RuntimeProviderRequestContext,
  ): Promise<void> {
    if (
      identity.ownerId !== context.actorScope.userId ||
      identity.tenantId !== context.actorScope.tenantId
    ) {
      throw new ForbiddenException('Durable Tess enrollment scope mismatch');
    }
    await this.coordinator.create(identity);
  }

  async queueProviderSend(input: TessProviderOutboxDto): Promise<void> {
    const snapshot = await this.coordinator.snapshot(input.sessionId);
    this.assertScope(snapshot.identity.ownerId, snapshot.identity.tenantId, input);
    await this.coordinator.enqueueOutbox({
      sessionId: input.sessionId,
      idempotencyKey: input.idempotencyKey,
      correlationId: input.correlationId,
      channelId: input.context.channelId,
      kind: 'provider.send',
      content: input.content,
    });
  }

  async dispatchProviderOutbox(sessionId: string, input: TessProviderOutboxDto): Promise<void> {
    if (sessionId !== input.sessionId) {
      throw new ForbiddenException('Durable Tess outbox session mismatch');
    }
    const snapshot = await this.coordinator.snapshot(sessionId);
    this.assertScope(snapshot.identity.ownerId, snapshot.identity.tenantId, input);
    const pendingEntry = snapshot.outbox.find(
      (entry): boolean => entry.idempotencyKey === input.idempotencyKey,
    );
    if (!pendingEntry) return;
    // Validate immutable routing before claiming. A caller with a mismatched
    // correlation/channel must not strand a pending external side effect.
    this.assertOutboxScope(pendingEntry, input);
    await this.coordinator.dispatchOutboxEntry(
      sessionId,
      input.idempotencyKey,
      async (entry): Promise<void> => {
        this.assertOutboxScope(entry, input);
        await this.runtimeProviders.sendMessage(
          snapshot.identity.providerId,
          snapshot.identity.runtimeSessionId,
          { content: entry.content, idempotencyKey: entry.idempotencyKey },
          input.context,
        );
      },
    );
  }

  /** Read durable identity/state only after deriving and checking the server-side actor scope. */
  async getSnapshot(sessionId: string, context: RuntimeProviderRequestContext) {
    const snapshot = await this.coordinator.snapshot(sessionId);
    this.assertScope(snapshot.identity.ownerId, snapshot.identity.tenantId, {
      sessionId,
      content: '',
      idempotencyKey: 'read-only',
      correlationId: context.correlationId,
      context,
    });
    return snapshot;
  }

  /** Startup/recovery-only path; normal queue/dispatch methods never requeue live work. */
  async recoverProviderSession(sessionId: string, input: TessProviderOutboxDto): Promise<void> {
    const snapshot = await this.coordinator.snapshot(sessionId);
    this.assertScope(snapshot.identity.ownerId, snapshot.identity.tenantId, input);
    await this.coordinator.recover(sessionId);
  }

  private assertOutboxScope(
    entry: { kind: string; correlationId: string; channelId: string },
    input: TessProviderOutboxDto,
  ): void {
    if (
      entry.kind !== 'provider.send' ||
      entry.correlationId !== input.correlationId ||
      entry.channelId !== input.context.channelId
    ) {
      throw new ForbiddenException('Durable Tess outbox scope or correlation mismatch');
    }
  }

  private assertScope(ownerId: string, tenantId: string, input: TessProviderOutboxDto): void {
    if (
      input.context.actorScope.userId !== ownerId ||
      input.context.actorScope.tenantId !== tenantId ||
      input.context.correlationId !== input.correlationId
    ) {
      throw new ForbiddenException('Durable Tess session scope or correlation mismatch');
    }
  }
}