stack/packages/agent/src/tmux-fleet-runtime-provider.ts

import { randomUUID } from 'node:crypto';
import type {
  AgentRuntimeProvider,
  RuntimeAttachHandle,
  RuntimeAttachMode,
  RuntimeCapabilitySet,
  RuntimeHealth,
  RuntimeMessage,
  RuntimeScope,
  RuntimeSession,
  RuntimeSessionTree,
  RuntimeStreamEvent,
} from '@mosaicstack/types';
const FLEET_PROVIDER_ID = 'fleet.tmux';
const ATTACHMENT_TTL_MS = 5 * 60 * 1_000;

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

/** A roster-bound target verified by the concrete fleet transport. */
export interface FleetRuntimeTarget {
  id: string;
  runtimeId: string;
  socketName: string;
}

/**
 * Narrow transport boundary implemented by the Mosaic tmux adapter. Keeping it
 * here prevents the runtime package from depending on the Mosaic CLI package.
 */
export interface FleetRuntimeTransport {
  verifySession(sessionId: string): Promise<FleetRuntimeTarget>;
  listSessions(): Promise<FleetRuntimeTarget[]>;
  sendMessage(sessionId: string, message: string, sourceLabel: string): Promise<void>;
  terminate(sessionId: string): Promise<void>;
}

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

export interface FleetReadAuthorization {
  operation: FleetReadOperation;
  scope: RuntimeScope;
  sessionId?: string;
}

/** Authorization for fleet inspection and read-only attachments. */
export interface FleetReadAuthority {
  canRead(authorization: FleetReadAuthorization): Promise<boolean>;
}

export interface FleetWriteAuthorization {
  operation: 'session.send' | 'session.terminate';
  sessionId: string;
  scope: RuntimeScope;
  /** Present only for terminate; authority adapters bind it to the exact action. */
  approvalRef?: string;
}

/**
 * The orchestrator is the only authority that may permit interaction-plane write/control requests to a
 * fleet peer. Gateway records the request and denial/success around provider
 * invocation; the default authority prevents direct interaction-plane writes by design.
 */
export interface FleetWriteAuthority {
  /** Non-consuming preflight used before probing the fleet transport. */
  canWrite(authorization: FleetWriteAuthorization): Promise<boolean>;
  /** Final exact-target authorization; may consume an orchestrator grant. */
  assertAuthorized(authorization: FleetWriteAuthorization): Promise<void>;
}

export interface TmuxFleetRuntimeProviderOptions {
  transport: FleetRuntimeTransport;
  readAuthority?: FleetReadAuthority;
  writeAuthority?: FleetWriteAuthority;
  sourceLabel?: string;
  attachmentIdFactory?: () => string;
  now?: () => Date;
  attachmentTtlMs?: number;
}

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

/** A typed, fail-closed provider error that callers can normalize at the boundary. */
export class FleetRuntimeProviderError extends Error {
  constructor(
    readonly code: FleetRuntimeProviderErrorCode,
    message: string,
  ) {
    super(message);
    this.name = FleetRuntimeProviderError.name;
  }
}

class DenyFleetReadAuthority implements FleetReadAuthority {
  async canRead(_authorization: FleetReadAuthorization): Promise<boolean> {
    return false;
  }
}

class DenyFleetWriteAuthority implements FleetWriteAuthority {
  async canWrite(_authorization: FleetWriteAuthorization): Promise<boolean> {
    return false;
  }

  async assertAuthorized(_authorization: FleetWriteAuthorization): Promise<void> {
    throw new FleetRuntimeProviderError(
      'forbidden',
      'Fleet writes require an explicit orchestrator authority decision',
    );
  }
}

/**
 * A capability-limited provider for rostered local fleet peers. It never
 * permits raw tmux socket/target selection, interactive control attach, or
 * direct interaction-plane writes; all side effects pass through exact transport checks.
 */
export class TmuxFleetRuntimeProvider implements AgentRuntimeProvider {
  readonly id = FLEET_PROVIDER_ID;
  private readonly attachments = new Map<string, FleetAttachment>();
  private readonly readAuthority: FleetReadAuthority;
  private readonly writeAuthority: FleetWriteAuthority;
  private readonly sourceLabel: string;
  private readonly attachmentIdFactory: () => string;
  private readonly now: () => Date;
  private readonly attachmentTtlMs: number;

  constructor(private readonly options: TmuxFleetRuntimeProviderOptions) {
    this.readAuthority = options.readAuthority ?? new DenyFleetReadAuthority();
    this.writeAuthority = options.writeAuthority ?? new DenyFleetWriteAuthority();
    this.sourceLabel = options.sourceLabel ?? 'interaction';
    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.send',
        'session.attach',
        'session.terminate',
      ],
    };
  }

  async health(scope: RuntimeScope): Promise<RuntimeHealth> {
    const targets = await this.readTargets('runtime.health', scope);
    return {
      status: targets.length > 0 ? 'healthy' : 'down',
      checkedAt: this.now().toISOString(),
      detail:
        targets.length > 0
          ? 'Authorized rostered fleet peers are reachable'
          : 'No authorized rostered fleet peers are reachable',
    };
  }

  async listSessions(scope: RuntimeScope): Promise<RuntimeSession[]> {
    const targets = await this.readTargets('session.list', scope);
    return this.toRuntimeSessions(targets);
  }

  async getSessionTree(scope: RuntimeScope): Promise<RuntimeSessionTree[]> {
    const targets = await this.readTargets('session.tree', scope);
    return this.toRuntimeSessions(targets).map(
      (session): RuntimeSessionTree => ({
        session,
        children: [],
      }),
    );
  }

  async *streamSession(
    _sessionId: string,
    _cursor: string | undefined,
    _scope: RuntimeScope,
  ): AsyncIterable<RuntimeStreamEvent> {
    throw new FleetRuntimeProviderError(
      'capability_unsupported',
      'Fleet session streaming is not supported by the tmux provider',
    );
  }

  async sendMessage(
    sessionId: string,
    message: RuntimeMessage,
    scope: RuntimeScope,
  ): Promise<void> {
    if (message.content.length === 0) {
      throw new FleetRuntimeProviderError('invalid_request', 'Fleet message content is required');
    }
    await this.assertWritePermitted('session.send', sessionId, scope);
    const target = await this.options.transport.verifySession(sessionId);
    await this.assertWriteAuthorized('session.send', target.id, scope);
    await this.options.transport.sendMessage(target.id, message.content, this.sourceLabel);
  }

  async attach(
    sessionId: string,
    mode: RuntimeAttachMode,
    scope: RuntimeScope,
  ): Promise<RuntimeAttachHandle> {
    if (mode !== 'read') {
      throw new FleetRuntimeProviderError('forbidden', 'Fleet control attach is not permitted');
    }
    await this.assertReadAuthorized('session.attach', sessionId, scope);
    const target = await this.options.transport.verifySession(sessionId);
    await this.assertReadAuthorized('session.attach', target.id, scope);
    const attachmentId = this.attachmentIdFactory();
    const nowMs = this.now().getTime();
    this.pruneExpiredAttachments(nowMs);
    const expiresAtMs = nowMs + this.attachmentTtlMs;
    this.attachments.set(attachmentId, {
      sessionId: target.id,
      scope: snapshotScope(scope),
      expiresAtMs,
    });
    return {
      attachmentId,
      sessionId: target.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 FleetRuntimeProviderError('not_found', 'Fleet attachment is not active');
    }
    if (this.now().getTime() >= attachment.expiresAtMs) {
      this.attachments.delete(attachmentId);
      throw new FleetRuntimeProviderError('forbidden', 'Fleet attachment has expired');
    }
    if (!sameScope(attachment.scope, scope)) {
      throw new FleetRuntimeProviderError('forbidden', 'Fleet attachment scope does not match');
    }
    this.attachments.delete(attachmentId);
  }

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

  private async readTargets(
    operation: FleetReadOperation,
    scope: RuntimeScope,
  ): Promise<FleetRuntimeTarget[]> {
    await this.assertReadAuthorized(operation, undefined, scope);
    const targets = await this.options.transport.listSessions();
    const authorization = await Promise.all(
      targets.map(
        async (target): Promise<boolean> =>
          this.readAuthority.canRead({ operation, sessionId: target.id, scope }),
      ),
    );
    return targets.filter((_target, index): boolean => authorization[index] === true);
  }

  private toRuntimeSessions(targets: FleetRuntimeTarget[]): RuntimeSession[] {
    const timestamp = this.now().toISOString();
    return targets.map(
      (target): RuntimeSession => ({
        id: target.id,
        providerId: this.id,
        runtimeId: target.runtimeId,
        state: 'active',
        createdAt: timestamp,
        updatedAt: timestamp,
      }),
    );
  }

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

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

  private async assertWritePermitted(
    operation: FleetWriteAuthorization['operation'],
    sessionId: string,
    scope: RuntimeScope,
    approvalRef?: string,
  ): Promise<void> {
    const permitted = await this.writeAuthority.canWrite({
      operation,
      sessionId,
      scope,
      ...(approvalRef ? { approvalRef } : {}),
    });
    if (!permitted) {
      throw new FleetRuntimeProviderError('forbidden', 'Fleet write is not authorized');
    }
  }

  private async assertWriteAuthorized(
    operation: FleetWriteAuthorization['operation'],
    sessionId: string,
    scope: RuntimeScope,
    approvalRef?: string,
  ): Promise<void> {
    await this.writeAuthority.assertAuthorized({
      operation,
      sessionId,
      scope,
      ...(approvalRef ? { approvalRef } : {}),
    });
  }
}

function snapshotScope(scope: RuntimeScope): RuntimeScope {
  return Object.freeze({
    actorId: scope.actorId,
    tenantId: scope.tenantId,
    channelId: scope.channelId,
    correlationId: scope.correlationId,
  });
}

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