flynn/src/gateway/handlers/system.ts

import type { GatewayRequest, OutboundMessage } from '../protocol.js';
import { makeResponse, makeError, ErrorCode } from '../protocol.js';
import type { MetricsSnapshot, EventEntry, ActiveRequestInfo } from '../metrics.js';
import type { ServiceInfo } from './services.js';
import type { NodeLocation, NodeStatus, NodePushToken } from './node.js';
import type { SessionAnalyticsSnapshot } from '../../session/index.js';
import type { LocalBackendAction, LocalBackendControlResult, LocalBackendStatus } from './localBackends.js';

/** Per-session token usage report returned by system.tokenUsage. */
export interface TokenUsageEntry {
  sessionId: string;
  primary: { inputTokens: number; outputTokens: number; calls: number };
  delegation: Record<string, { inputTokens: number; outputTokens: number; calls: number }>;
  total: { inputTokens: number; outputTokens: number; calls: number; estimatedCost: number };
}

/** Per-session context budget report returned by system.contextUsage. */
export interface ContextUsageEntry {
  sessionId: string;
  budget: {
    estimatedTokens: number;
    contextWindow: number;
    remainingTokens: number;
    usagePct: number;
    thresholdPct: number;
    thresholdTokens: number;
    shouldCompact: boolean;
  };
}

export interface PresenceEntry {
  channel: string;
  senderId: string;
  senderName?: string;
  firstSeenAt: number;
  lastSeenAt: number;
  messageCount: number;
  status: 'online' | 'offline';
}

export interface NodeLocationEntry {
  nodeId: string;
  role: string;
  connectionId: string;
  location: NodeLocation;
}

export interface NodeEntry {
  connectionId: string;
  nodeId: string;
  role: string;
  identity?: string;
  protocolVersion: number;
  capabilities: string[];
  registeredAt: number;
  location?: NodeLocation;
  status?: NodeStatus;
  push?: NodePushTokenSummary;
}

export interface NodePushTokenSummary {
  provider: NodePushToken['provider'];
  tokenPreview: string;
  topic?: string;
  environment?: NodePushToken['environment'];
  registeredAt: number;
}

export interface SystemHandlerDeps {
  startTime: number;
  version: string;
  getSessionCount: () => number;
  getToolCount: () => number;
  getConnectionCount: () => number;
  /** Optional callback to trigger a graceful restart. If not provided, system.restart returns an error. */
  restart?: () => Promise<void>;
  getChannels?: () => Array<{ name: string; status: string; error?: string; lastErrorAt?: number }>;
  getUsage?: () => { totalSessions: number; activeConnections: number };
  /** Optional callback to retrieve per-session token usage data. */
  getTokenUsage?: () => TokenUsageEntry[];
  /** Optional callback to retrieve per-session context budget data. */
  getContextUsage?: () => ContextUsageEntry[];
  /** Optional callback to retrieve aggregated metrics snapshot. */
  getMetrics?: () => MetricsSnapshot;
  /** Optional callback to retrieve session analytics. */
  getSessionAnalytics?: (opts?: { days?: number; topLimit?: number }) => SessionAnalyticsSnapshot;
  /** Optional callback to retrieve recent events. */
  getEvents?: (opts?: { level?: string; limit?: number }) => EventEntry[];
  /** Optional callback to retrieve active requests. */
  getActiveRequests?: () => ActiveRequestInfo[];
  /** Optional callback to retrieve all services (channels + automation). */
  getServices?: () => ServiceInfo[];
  /** Optional callback to retrieve tracked sender presence. */
  getPresence?: (opts?: { channel?: string; status?: 'online' | 'offline'; limit?: number }) => PresenceEntry[];
  /** Optional callback to retrieve latest node location data. */
  getNodeLocations?: (opts?: { role?: string; nodeId?: string; limit?: number }) => NodeLocationEntry[];
  /** Optional callback to retrieve registered node connection snapshots. */
  getNodes?: (opts?: { role?: string; platform?: string; limit?: number }) => NodeEntry[];
  /** Optional callback to retrieve provider model catalogs. */
  getModelCatalog?: (opts?: { provider?: string; forceRefresh?: boolean }) => Promise<Array<{
    provider: string;
    models: string[];
    source: 'api' | 'config' | 'unavailable';
    error?: string;
    fetchedAt: number;
  }>>;
  /** Optional callback to retrieve local backend daemon statuses. */
  getLocalBackends?: () => Promise<LocalBackendStatus[]> | LocalBackendStatus[];
  /** Optional callback to control local backend daemons. */
  controlLocalBackend?: (backend: string, action: string) => Promise<LocalBackendControlResult>;
}

function normalizeErrorMessage(error: unknown): string {
  if (error instanceof Error && error.message.trim().length > 0) {
    return error.message;
  }
  if (typeof error === 'string' && error.trim().length > 0) {
    return error;
  }
  return 'Unknown error';
}

export function createSystemHandlers(deps: SystemHandlerDeps) {
  return {
    'system.health': async (request: GatewayRequest): Promise<OutboundMessage> => {
      return makeResponse(request.id, {
        status: 'ok',
        uptime: Math.floor((Date.now() - deps.startTime) / 1000),
        version: deps.version,
        sessions: deps.getSessionCount(),
        tools: deps.getToolCount(),
        connections: deps.getConnectionCount(),
      });
    },

    'system.restart': async (request: GatewayRequest): Promise<OutboundMessage> => {
      const restart = deps.restart;
      if (!restart) {
        return makeError(request.id, ErrorCode.InternalError, 'Restart not available in this environment');
      }

      // Send response before initiating restart (client receives confirmation)
      const response = makeResponse(request.id, { restarting: true });

      // Schedule restart after response is sent (next tick)
      queueMicrotask(() => {
        restart().catch((err) => {
          console.error('Restart failed:', err);
        });
      });

      return response;
    },

    'system.channels': async (request: GatewayRequest): Promise<OutboundMessage> => {
      if (!deps.getChannels) {
        return makeResponse(request.id, { channels: [] });
      }
      return makeResponse(request.id, { channels: deps.getChannels() });
    },

    'system.services': async (request: GatewayRequest): Promise<OutboundMessage> => {
      if (!deps.getServices) {
        return makeResponse(request.id, { services: [] });
      }
      return makeResponse(request.id, { services: deps.getServices() });
    },

    'system.presence': async (request: GatewayRequest): Promise<OutboundMessage> => {
      if (!deps.getPresence) {
        return makeResponse(request.id, { presence: [], summary: { total: 0, online: 0, offline: 0 } });
      }

      const params = request.params as { channel?: string; status?: 'online' | 'offline'; limit?: number } | undefined;
      const presence = deps.getPresence({
        channel: params?.channel,
        status: params?.status,
        limit: params?.limit,
      });
      const online = presence.filter((entry) => entry.status === 'online').length;
      return makeResponse(request.id, {
        presence,
        summary: {
          total: presence.length,
          online,
          offline: presence.length - online,
        },
      });
    },

    'system.location': async (request: GatewayRequest): Promise<OutboundMessage> => {
      if (!deps.getNodeLocations) {
        return makeResponse(request.id, { locations: [], summary: { total: 0 } });
      }

      const params = request.params as { role?: string; nodeId?: string; limit?: number } | undefined;
      const locations = deps.getNodeLocations({
        role: params?.role,
        nodeId: params?.nodeId,
        limit: params?.limit,
      });
      return makeResponse(request.id, {
        locations,
        summary: {
          total: locations.length,
        },
      });
    },

    'system.nodes': async (request: GatewayRequest): Promise<OutboundMessage> => {
      if (!deps.getNodes) {
        return makeResponse(request.id, { nodes: [], summary: { total: 0 } });
      }

      const params = request.params as { role?: string; platform?: string; limit?: number } | undefined;
      const nodes = deps.getNodes({
        role: params?.role,
        platform: params?.platform,
        limit: params?.limit,
      });
      return makeResponse(request.id, {
        nodes,
        summary: {
          total: nodes.length,
        },
      });
    },

    'system.usage': async (request: GatewayRequest): Promise<OutboundMessage> => {
      const uptime = Math.floor((Date.now() - deps.startTime) / 1000);
      const usage = deps.getUsage?.() ?? { totalSessions: 0, activeConnections: 0 };
      return makeResponse(request.id, {
        uptime,
        ...usage,
        tools: deps.getToolCount(),
      });
    },

    'system.tokenUsage': async (request: GatewayRequest): Promise<OutboundMessage> => {
      const sessions = deps.getTokenUsage?.() ?? [];
      return makeResponse(request.id, { sessions });
    },

    'system.contextUsage': async (request: GatewayRequest): Promise<OutboundMessage> => {
      const sessions = deps.getContextUsage?.() ?? [];
      return makeResponse(request.id, { sessions });
    },

    'system.metrics': async (request: GatewayRequest): Promise<OutboundMessage> => {
      if (!deps.getMetrics) {
        return makeResponse(request.id, {});
      }
      return makeResponse(request.id, deps.getMetrics());
    },

    'system.sessionAnalytics': async (request: GatewayRequest): Promise<OutboundMessage> => {
      if (!deps.getSessionAnalytics) {
        return makeResponse(request.id, {
          daily: [],
          topSessions: [],
          topTools: [],
          topTopics: [],
          averageMessagesPerSession: 0,
          totalSessions: 0,
          totalMessages: 0,
        });
      }
      const params = request.params as { days?: number; topLimit?: number } | undefined;
      const snapshot = deps.getSessionAnalytics({
        days: params?.days,
        topLimit: params?.topLimit,
      });
      return makeResponse(request.id, snapshot);
    },

    'system.events': async (request: GatewayRequest): Promise<OutboundMessage> => {
      if (!deps.getEvents) {
        return makeResponse(request.id, { events: [] });
      }
      const params = request.params as { level?: string; limit?: number } | undefined;
      const events = deps.getEvents({ level: params?.level, limit: params?.limit });
      return makeResponse(request.id, { events });
    },

    'system.activeRequests': async (request: GatewayRequest): Promise<OutboundMessage> => {
      if (!deps.getActiveRequests) {
        return makeResponse(request.id, { requests: [] });
      }
      return makeResponse(request.id, { requests: deps.getActiveRequests() });
    },

    'system.modelCatalog': async (request: GatewayRequest): Promise<OutboundMessage> => {
      if (!deps.getModelCatalog) {
        return makeResponse(request.id, { providers: [] });
      }
      const params = request.params as { provider?: string; forceRefresh?: boolean } | undefined;
      const providers = await deps.getModelCatalog({
        provider: params?.provider,
        forceRefresh: params?.forceRefresh === true,
      });
      return makeResponse(request.id, { providers });
    },

    'system.localBackends': async (request: GatewayRequest): Promise<OutboundMessage> => {
      if (!deps.getLocalBackends) {
        return makeResponse(request.id, { backends: [] });
      }
      try {
        const backends = await deps.getLocalBackends();
        return makeResponse(request.id, { backends });
      } catch (error) {
        return makeError(request.id, ErrorCode.InternalError, `Failed to load local backends: ${normalizeErrorMessage(error)}`);
      }
    },

    'system.localBackendControl': async (request: GatewayRequest): Promise<OutboundMessage> => {
      if (!deps.controlLocalBackend) {
        return makeError(request.id, ErrorCode.InternalError, 'Local backend control is not available in this environment');
      }
      const params = request.params as { backend?: string; action?: LocalBackendAction } | undefined;
      if (!params?.backend || typeof params.backend !== 'string') {
        return makeError(request.id, ErrorCode.InvalidRequest, 'backend is required');
      }
      if (!params?.action || typeof params.action !== 'string') {
        return makeError(request.id, ErrorCode.InvalidRequest, 'action is required');
      }
      if (!['start', 'restart', 'stop', 'update'].includes(params.action)) {
        return makeError(request.id, ErrorCode.InvalidRequest, 'action must be one of: start, restart, stop, update');
      }

      try {
        const result = await deps.controlLocalBackend(params.backend, params.action);
        return makeResponse(request.id, result);
      } catch (error) {
        return makeError(request.id, ErrorCode.InternalError, `Local backend control failed: ${normalizeErrorMessage(error)}`);
      }
    },
  };
}