stack/apps/gateway/src/coord/interaction-coordination.service.ts

import { Inject, Injectable } from '@nestjs/common';
import {
  InteractionCoordinationClient,
  type CoordinationObservation,
  type CoordinationResult,
  type CoordinationScope,
  type InteractionCoordinationIdentity,
  type InteractionCoordinationPort,
  type HandoffReceipt,
} from '@mosaicstack/coord';
import type { RuntimeProviderRequestContext } from '../agent/runtime-provider-registry.service.js';
import type { CreateHandoffDto } from './interaction-coordination.dto.js';

export const COORDINATION_PORT = Symbol('COORDINATION_PORT');
export const COORDINATION_CONFIG = Symbol('COORDINATION_CONFIG');

const HANDOFF_TRACKING_TTL_MS = 60 * 60 * 1_000;
const MAX_TRACKED_HANDOFFS = 1_000;
const MAX_IDEMPOTENCY_KEY_LENGTH = 128;
const MAX_SUMMARY_LENGTH = 2_048;
const MAX_CONTEXT_LENGTH = 8_192;
const MAX_MISSION_ID_LENGTH = 128;

export interface InteractionCoordinationConfig {
  interactionAgentId?: string;
  orchestrationAgentId?: string;
}

interface HandoffOwner {
  actorId: string;
  tenantId: string;
  requesterAgentId: string;
  correlationId: string;
  expiresAt: number;
}

interface NormalizedHandoffRequest {
  idempotencyKey: string;
  summary: string;
  context?: string;
  missionId?: string;
}

interface TrackedHandoff {
  request: NormalizedHandoffRequest;
  receipt: Promise<HandoffReceipt>;
  expiresAt: number;
}

/**
 * Gateway authority boundary for the interaction agent. It derives requester,
 * actor, and tenant from trusted server configuration and authentication; no
 * channel request can name a target or gain orchestrator-owned orchestration verbs.
 */
@Injectable()
export class InteractionCoordinationService {
  private readonly owners = new Map<string, HandoffOwner>();
  private readonly handoffsByIdempotencyKey = new Map<string, TrackedHandoff>();

  constructor(
    @Inject(COORDINATION_PORT) private readonly port: InteractionCoordinationPort,
    @Inject(COORDINATION_CONFIG) private readonly config: InteractionCoordinationConfig,
    private readonly handoffIdFactory: () => string = (): string => crypto.randomUUID(),
  ) {}

  async handoff(
    request: CreateHandoffDto,
    context: RuntimeProviderRequestContext,
  ): Promise<HandoffReceipt> {
    this.pruneExpiredTracking();
    const normalized = this.normalizeRequest(request);
    const scope = this.scope(context);
    const idempotencyKey = this.idempotencyKey(normalized.idempotencyKey, scope);
    const existing = this.handoffsByIdempotencyKey.get(idempotencyKey);
    if (existing !== undefined) {
      if (!sameRequest(existing.request, normalized)) {
        throw new InteractionCoordinationGatewayError(
          'handoff_conflict',
          'Handoff idempotency key is already bound to different immutable input',
        );
      }
      return existing.receipt;
    }

    const pending = this.deliverHandoff(this.handoffIdFactory(), normalized, scope);
    const tracked: TrackedHandoff = {
      request: normalized,
      receipt: pending,
      expiresAt: this.expiresAt(),
    };
    this.handoffsByIdempotencyKey.set(idempotencyKey, tracked);
    this.enforceTrackingLimit(this.handoffsByIdempotencyKey);
    try {
      return await pending;
    } catch (error: unknown) {
      if (this.handoffsByIdempotencyKey.get(idempotencyKey) === tracked) {
        this.handoffsByIdempotencyKey.delete(idempotencyKey);
      }
      throw error;
    }
  }

  async observe(
    handoffId: string,
    context: RuntimeProviderRequestContext,
  ): Promise<CoordinationObservation> {
    this.pruneExpiredTracking();
    const scope = this.scope(context);
    const owner = this.ownerFor(handoffId, scope);
    return this.client().observe(handoffId, { ...scope, correlationId: owner.correlationId });
  }

  async result(
    handoffId: string,
    context: RuntimeProviderRequestContext,
  ): Promise<CoordinationResult> {
    this.pruneExpiredTracking();
    const scope = this.scope(context);
    const owner = this.ownerFor(handoffId, scope);
    return this.client().result(handoffId, { ...scope, correlationId: owner.correlationId });
  }

  private async deliverHandoff(
    handoffId: string,
    request: NormalizedHandoffRequest,
    scope: CoordinationScope,
  ): Promise<HandoffReceipt> {
    const receipt = await this.client((): string => handoffId).handoff(request, scope);
    const owner: HandoffOwner = {
      actorId: scope.actorId,
      tenantId: scope.tenantId,
      requesterAgentId: scope.requesterAgentId,
      correlationId: scope.correlationId,
      expiresAt: this.expiresAt(),
    };
    const existing = this.owners.get(receipt.handoffId);
    if (existing !== undefined && !sameOwner(existing, owner)) {
      throw new InteractionCoordinationGatewayError(
        'handoff_conflict',
        'Handoff ID is already bound to a different authenticated scope',
      );
    }
    this.owners.set(receipt.handoffId, owner);
    this.enforceTrackingLimit(this.owners);
    return receipt;
  }

  private client(handoffIdFactory?: () => string): InteractionCoordinationClient {
    return new InteractionCoordinationClient(this.identity(), this.port, handoffIdFactory);
  }

  private identity(): InteractionCoordinationIdentity {
    const interactionAgentId = this.config.interactionAgentId?.trim();
    const orchestrationAgentId = this.config.orchestrationAgentId?.trim();
    if (!interactionAgentId) {
      throw new InteractionCoordinationGatewayError(
        'unconfigured_requester',
        'Interaction agent identity is not configured',
      );
    }
    if (!orchestrationAgentId) {
      throw new InteractionCoordinationGatewayError(
        'unconfigured_target',
        'Orchestration agent identity is not configured',
      );
    }
    return { interactionAgentId, orchestrationAgentId };
  }

  private scope(context: RuntimeProviderRequestContext): CoordinationScope {
    const identity = this.identity();
    return Object.freeze({
      actorId: context.actorScope.userId,
      tenantId: context.actorScope.tenantId,
      correlationId: context.correlationId,
      requesterAgentId: identity.interactionAgentId,
    });
  }

  private normalizeRequest(request: CreateHandoffDto): NormalizedHandoffRequest {
    if (typeof request !== 'object' || request === null) {
      throw new InteractionCoordinationGatewayError(
        'invalid_request',
        'Handoff request is invalid',
      );
    }
    const idempotencyKey = this.requiredString(
      request.idempotencyKey,
      'idempotency key',
      MAX_IDEMPOTENCY_KEY_LENGTH,
    );
    const summary = this.requiredString(request.summary, 'summary', MAX_SUMMARY_LENGTH);
    const context = this.optionalString(request.context, 'context', MAX_CONTEXT_LENGTH);
    const missionId = this.optionalString(request.missionId, 'mission ID', MAX_MISSION_ID_LENGTH);
    return Object.freeze({
      idempotencyKey,
      summary,
      ...(context === undefined ? {} : { context }),
      ...(missionId === undefined ? {} : { missionId }),
    });
  }

  private idempotencyKey(requestKey: string, scope: CoordinationScope): string {
    return `${scope.tenantId}\u0000${scope.actorId}\u0000${scope.requesterAgentId}\u0000${requestKey}`;
  }

  private requiredString(value: unknown, field: string, maximumLength: number): string {
    if (typeof value !== 'string') {
      throw new InteractionCoordinationGatewayError(
        'invalid_request',
        `Handoff ${field} must be a string`,
      );
    }
    const normalized = value.trim();
    if (normalized.length === 0 || normalized.length > maximumLength) {
      throw new InteractionCoordinationGatewayError(
        'invalid_request',
        `Handoff ${field} is invalid`,
      );
    }
    return normalized;
  }

  private optionalString(value: unknown, field: string, maximumLength: number): string | undefined {
    if (value === undefined) return undefined;
    return this.requiredString(value, field, maximumLength);
  }

  private expiresAt(): number {
    return Date.now() + HANDOFF_TRACKING_TTL_MS;
  }

  private pruneExpiredTracking(): void {
    const now = Date.now();
    for (const [key, tracked] of this.handoffsByIdempotencyKey) {
      if (tracked.expiresAt <= now) this.handoffsByIdempotencyKey.delete(key);
    }
    for (const [key, owner] of this.owners) {
      if (owner.expiresAt <= now) this.owners.delete(key);
    }
  }

  private enforceTrackingLimit<T>(entries: Map<string, T>): void {
    while (entries.size > MAX_TRACKED_HANDOFFS) {
      const oldest = entries.keys().next().value;
      if (typeof oldest !== 'string') return;
      entries.delete(oldest);
    }
  }

  private ownerFor(handoffId: string, scope: CoordinationScope): HandoffOwner {
    const owner = this.owners.get(handoffId);
    if (owner === undefined) {
      throw new InteractionCoordinationGatewayError('not_found', 'Handoff was not found');
    }
    if (
      owner.tenantId !== scope.tenantId ||
      owner.actorId !== scope.actorId ||
      owner.requesterAgentId !== scope.requesterAgentId
    ) {
      throw new InteractionCoordinationGatewayError(
        'cross_tenant_forbidden',
        'Handoff is outside the authenticated scope',
      );
    }
    return owner;
  }
}

export type InteractionCoordinationGatewayErrorCode =
  | 'cross_tenant_forbidden'
  | 'handoff_conflict'
  | 'invalid_request'
  | 'not_found'
  | 'unconfigured_requester'
  | 'unconfigured_target';

function sameOwner(left: HandoffOwner, right: HandoffOwner): boolean {
  return (
    left.actorId === right.actorId &&
    left.tenantId === right.tenantId &&
    left.requesterAgentId === right.requesterAgentId
  );
}

function sameRequest(left: NormalizedHandoffRequest, right: NormalizedHandoffRequest): boolean {
  return (
    left.idempotencyKey === right.idempotencyKey &&
    left.summary === right.summary &&
    left.context === right.context &&
    left.missionId === right.missionId
  );
}

export class InteractionCoordinationGatewayError extends Error {
  constructor(
    readonly code: InteractionCoordinationGatewayErrorCode,
    message: string,
  ) {
    super(message);
    this.name = InteractionCoordinationGatewayError.name;
  }
}