stack/packages/agent/src/matrix-native-runtime-provider.ts

import { randomUUID } from 'node:crypto';
import type {
  AgentRuntimeProvider,
  RuntimeAttachHandle,
  RuntimeAttachMode,
  RuntimeCapabilitySet,
  RuntimeHealth,
  RuntimeMessage,
  RuntimeScope,
  RuntimeSession,
  RuntimeSessionTree,
  RuntimeStreamEvent,
} from '@mosaicstack/types';

const MATRIX_PROVIDER_ID = 'runtime.matrix';
const ATTACHMENT_TTL_MS = 5 * 60 * 1_000;

/** A verified native runtime session. Matrix room and event details remain transport-local. */
export interface MatrixRuntimeSession {
  id: string;
  runtimeId: string;
  parentSessionId?: string;
  state: RuntimeSession['state'];
  createdAt: string;
  updatedAt: string;
}

/**
 * Narrow native Matrix boundary. The concrete Mosaic transport owns homeserver
 * authentication, exact room mapping, Matrix identity checks, and replay cursors.
 */
export interface MatrixRuntimeTransport {
  health(scope: RuntimeScope): Promise<RuntimeHealth>;
  listSessions(scope: RuntimeScope): Promise<MatrixRuntimeSession[]>;
  verifySession(sessionId: string, scope: RuntimeScope): Promise<MatrixRuntimeSession>;
  stream(
    sessionId: string,
    cursor: string | undefined,
    scope: RuntimeScope,
  ): AsyncIterable<RuntimeStreamEvent>;
  send(sessionId: string, message: RuntimeMessage, scope: RuntimeScope): Promise<void>;
  terminate(sessionId: string, approvalRef: string, scope: RuntimeScope): Promise<void>;
}

export type MatrixRuntimeProviderErrorCode =
  | 'capability_unsupported'
  | 'forbidden'
  | 'invalid_request'
  | 'not_found';

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

export type MatrixReadOperation =
  | 'runtime.health'
  | 'session.list'
  | 'session.tree'
  | 'session.stream'
  | 'session.attach';

export interface MatrixReadAuthority {
  canRead(input: {
    operation: MatrixReadOperation;
    scope: RuntimeScope;
    sessionId?: string;
  }): Promise<boolean>;
}

export interface MatrixWriteAuthority {
  canWrite(input: {
    operation: 'session.send' | 'session.terminate';
    sessionId: string;
    scope: RuntimeScope;
    approvalRef?: string;
  }): Promise<boolean>;
  assertAuthorized(input: {
    operation: 'session.send' | 'session.terminate';
    sessionId: string;
    scope: RuntimeScope;
    approvalRef?: string;
  }): Promise<void>;
}

export interface MatrixNativeRuntimeProviderOptions {
  transport: MatrixRuntimeTransport;
  readAuthority?: MatrixReadAuthority;
  writeAuthority?: MatrixWriteAuthority;
  attachmentIdFactory?: () => string;
  now?: () => Date;
  attachmentTtlMs?: number;
}

interface Attachment {
  sessionId: string;
  scope: RuntimeScope;
  expiresAtMs: number;
}

class DenyMatrixReadAuthority implements MatrixReadAuthority {
  async canRead(): Promise<boolean> {
    return false;
  }
}

class DenyMatrixWriteAuthority implements MatrixWriteAuthority {
  async canWrite(): Promise<boolean> {
    return false;
  }

  async assertAuthorized(): Promise<void> {
    throw new MatrixRuntimeProviderError(
      'forbidden',
      'Matrix runtime writes require orchestrator authority',
    );
  }
}

/**
 * Native Matrix adapter behind the Mosaic runtime contract. It accepts only
 * stable session IDs; room identifiers, Matrix event schemas, and credentials
 * are deliberately confined to the concrete transport implementation.
 */
export class MatrixNativeRuntimeProvider implements AgentRuntimeProvider {
  readonly id = MATRIX_PROVIDER_ID;
  private readonly readAuthority: MatrixReadAuthority;
  private readonly writeAuthority: MatrixWriteAuthority;
  private readonly attachmentIdFactory: () => string;
  private readonly now: () => Date;
  private readonly attachmentTtlMs: number;
  private readonly attachments = new Map<string, Attachment>();

  constructor(private readonly options: MatrixNativeRuntimeProviderOptions) {
    this.readAuthority = options.readAuthority ?? new DenyMatrixReadAuthority();
    this.writeAuthority = options.writeAuthority ?? new DenyMatrixWriteAuthority();
    this.attachmentIdFactory = options.attachmentIdFactory ?? randomUUID;
    this.now = options.now ?? (() => new Date());
    this.attachmentTtlMs = options.attachmentTtlMs ?? ATTACHMENT_TTL_MS;
  }

  async capabilities(_scope: RuntimeScope): Promise<RuntimeCapabilitySet> {
    return {
      supported: [
        'session.list',
        'session.tree',
        'session.stream',
        'session.send',
        'session.attach',
        'session.terminate',
      ],
    };
  }

  async health(scope: RuntimeScope): Promise<RuntimeHealth> {
    await this.assertRead('runtime.health', undefined, scope);
    return this.options.transport.health(scope);
  }

  async listSessions(scope: RuntimeScope): Promise<RuntimeSession[]> {
    await this.assertRead('session.list', undefined, scope);
    const sessions = await this.options.transport.listSessions(scope);
    const visible = await Promise.all(
      sessions.map((session) =>
        this.readAuthority.canRead({ operation: 'session.list', sessionId: session.id, scope }),
      ),
    );
    return sessions
      .filter((_session, index) => visible[index] === true)
      .map((session) => this.runtimeSession(session));
  }

  async getSessionTree(scope: RuntimeScope): Promise<RuntimeSessionTree[]> {
    await this.assertRead('session.tree', undefined, scope);
    const sessions = await this.options.transport.listSessions(scope);
    const visible = await Promise.all(
      sessions.map((session) =>
        this.readAuthority.canRead({ operation: 'session.tree', sessionId: session.id, scope }),
      ),
    );
    const runtimeSessions = sessions
      .filter((_session, index) => visible[index] === true)
      .map((session) => this.runtimeSession(session));
    const nodes = new Map<string, RuntimeSessionTree>(
      runtimeSessions.map((session) => [session.id, { session, children: [] }]),
    );
    const roots: RuntimeSessionTree[] = [];
    for (const session of runtimeSessions) {
      const node = nodes.get(session.id)!;
      const parent = session.parentSessionId ? nodes.get(session.parentSessionId) : undefined;
      if (parent) parent.children.push(node);
      else roots.push(node);
    }
    return roots;
  }

  async *streamSession(
    sessionId: string,
    cursor: string | undefined,
    scope: RuntimeScope,
  ): AsyncIterable<RuntimeStreamEvent> {
    await this.assertRead('session.stream', sessionId, scope);
    const session = await this.options.transport.verifySession(sessionId, scope);
    await this.assertRead('session.stream', session.id, scope);
    yield* this.options.transport.stream(session.id, cursor, scope);
  }

  async sendMessage(
    sessionId: string,
    message: RuntimeMessage,
    scope: RuntimeScope,
  ): Promise<void> {
    if (!message.content.trim()) {
      throw new MatrixRuntimeProviderError(
        'invalid_request',
        'Matrix runtime message content is required',
      );
    }
    await this.assertWritePermitted('session.send', sessionId, scope);
    const session = await this.options.transport.verifySession(sessionId, scope);
    await this.assertWriteAuthorized('session.send', session.id, scope);
    await this.options.transport.send(session.id, message, scope);
  }

  async attach(
    sessionId: string,
    mode: RuntimeAttachMode,
    scope: RuntimeScope,
  ): Promise<RuntimeAttachHandle> {
    if (mode !== 'read') {
      throw new MatrixRuntimeProviderError('forbidden', 'Matrix control attach is not permitted');
    }
    await this.assertRead('session.attach', sessionId, scope);
    const session = await this.options.transport.verifySession(sessionId, scope);
    await this.assertRead('session.attach', session.id, scope);
    const nowMs = this.now().getTime();
    this.pruneExpired(nowMs);
    const attachmentId = this.attachmentIdFactory();
    const expiresAtMs = nowMs + this.attachmentTtlMs;
    this.attachments.set(attachmentId, {
      sessionId: session.id,
      scope: snapshotScope(scope),
      expiresAtMs,
    });
    return {
      attachmentId,
      sessionId: session.id,
      mode,
      expiresAt: new Date(expiresAtMs).toISOString(),
    };
  }

  async detach(attachmentId: string, scope: RuntimeScope): Promise<void> {
    const attachment = this.attachments.get(attachmentId);
    if (!attachment)
      throw new MatrixRuntimeProviderError('not_found', 'Matrix attachment is not active');
    if (this.now().getTime() >= attachment.expiresAtMs) {
      this.attachments.delete(attachmentId);
      throw new MatrixRuntimeProviderError('forbidden', 'Matrix attachment has expired');
    }
    if (!sameScope(attachment.scope, scope)) {
      throw new MatrixRuntimeProviderError('forbidden', 'Matrix attachment scope does not match');
    }
    this.attachments.delete(attachmentId);
  }

  async terminate(sessionId: string, approvalRef: string, scope: RuntimeScope): Promise<void> {
    if (!approvalRef.trim()) {
      throw new MatrixRuntimeProviderError(
        'invalid_request',
        'Matrix termination approval is required',
      );
    }
    await this.assertWritePermitted('session.terminate', sessionId, scope, approvalRef);
    const session = await this.options.transport.verifySession(sessionId, scope);
    await this.assertWriteAuthorized('session.terminate', session.id, scope, approvalRef);
    await this.options.transport.terminate(session.id, approvalRef, scope);
  }

  private runtimeSession(session: MatrixRuntimeSession): RuntimeSession {
    return { ...session, providerId: this.id };
  }

  private async assertRead(
    operation: MatrixReadOperation,
    sessionId: string | undefined,
    scope: RuntimeScope,
  ): Promise<void> {
    const allowed = await this.readAuthority.canRead({
      operation,
      scope,
      ...(sessionId ? { sessionId } : {}),
    });
    if (!allowed)
      throw new MatrixRuntimeProviderError('forbidden', 'Matrix runtime read is not authorized');
  }

  private async assertWritePermitted(
    operation: 'session.send' | 'session.terminate',
    sessionId: string,
    scope: RuntimeScope,
    approvalRef?: string,
  ): Promise<void> {
    const allowed = await this.writeAuthority.canWrite({
      operation,
      sessionId,
      scope,
      ...(approvalRef ? { approvalRef } : {}),
    });
    if (!allowed)
      throw new MatrixRuntimeProviderError('forbidden', 'Matrix runtime write is not authorized');
  }

  private async assertWriteAuthorized(
    operation: 'session.send' | 'session.terminate',
    sessionId: string,
    scope: RuntimeScope,
    approvalRef?: string,
  ): Promise<void> {
    await this.writeAuthority.assertAuthorized({
      operation,
      sessionId,
      scope,
      ...(approvalRef ? { approvalRef } : {}),
    });
  }

  private pruneExpired(nowMs: number): void {
    for (const [id, attachment] of this.attachments) {
      if (attachment.expiresAtMs <= nowMs) this.attachments.delete(id);
    }
  }
}

function snapshotScope(scope: RuntimeScope): RuntimeScope {
  return Object.freeze({ ...scope });
}

function sameScope(left: RuntimeScope, right: RuntimeScope): boolean {
  return (
    left.actorId === right.actorId &&
    left.tenantId === right.tenantId &&
    left.channelId === right.channelId &&
    left.correlationId === right.correlationId
  );
}