Assessor_2.0/server/compliance/compliance-engine-runtime.ts

import { join } from 'node:path';
import { Pool } from 'pg';
import type { AiCompletionClient } from '../ai/litellm-client';
import { repositoryConnectionString, sslConfiguration, type PersistenceKind, type RepositoryEnvironment } from '../repository-factory';
import type { AssessorRepository } from '../../src/persistence/assessor-repository';
import { complianceReadPortFrom, type ComplianceStore } from '../../src/compliance/compliance-store';
import { LocalComplianceStore } from '../../src/compliance/local-compliance-store';
import { PostgresComplianceStore } from './postgres-compliance-store';
import { createRegulatoryRagRuntime } from './regulatory-rag-runtime';
import { RegulatoryMappingService } from './regulatory-mapping-service';
import { RegulatoryCorpusUploadService } from './regulatory-corpus-upload-service';
import { RegulatoryAutoMapping } from './regulatory-auto-mapping';
import { AiOverviewService, type AiOverviewServiceOptions } from './ai-overview-service';
import { BsiCatalogService } from './bsi-catalog-service';

export interface ComplianceEngineRuntime {
  mappingService: RegulatoryMappingService;
  corpusUploadService: RegulatoryCorpusUploadService;
  aiOverviewService: AiOverviewService;
  bsiCatalogService: BsiCatalogService;
  store: ComplianceStore;
  /** Stops running work; ASSESSOR keeps its own repository open. */
  stop(): Promise<void>;
  close(): Promise<void>;
}

export interface ComplianceEngineRuntimeOptions {
  repository: AssessorRepository;
  persistence: PersistenceKind;
  aiClient?: AiCompletionClient;
  artifactRoot: string;
  environment?: RepositoryEnvironment;
  aiOverview?: Omit<AiOverviewServiceOptions, 'client'>;
}

/**
 * Starts the Compliance Engine beside ASSESSOR. It reads the Core only through its read-only port and
 * writes only its own tables through a separate small pool. If retrieval cannot start, mapping is
 * refused with a reason while ASSESSOR keeps running.
 */
export async function createComplianceEngineRuntime(options: ComplianceEngineRuntimeOptions): Promise<ComplianceEngineRuntime> {
  const pool = options.persistence === 'postgresql' ? compliancePool(options.environment ?? process.env as RepositoryEnvironment) : undefined;
  const store: ComplianceStore = pool ? new PostgresComplianceStore(pool) : new LocalComplianceStore();
  const read = complianceReadPortFrom(options.repository);
  const corpusRoot = join(options.artifactRoot, 'regulatory-corpora');
  // The active BSI catalog and ISO mapping; the pinned snapshots until Update from BSI stores a newer state.
  const bsiCatalogService = new BsiCatalogService(join(corpusRoot, 'bsi', 'active-catalog.json'));
  const bsiCatalogs = () => bsiCatalogService.catalogs();
  let mappingService: RegulatoryMappingService;
  let rag: ReturnType<typeof createRegulatoryRagRuntime> | undefined;
  let disabledReason: string | undefined;
  try {
    rag = createRegulatoryRagRuntime();
    mappingService = new RegulatoryMappingService(read, store, {
      findingRetrieval: rag.findingRetrieval,
      index: rag.index,
      embeddingModel: rag.embeddingModel,
      client: options.aiClient,
      corpusId: process.env.ASSESSOR_RAG_ACTIVE_CORPUS_ID?.trim(),
      corpusVersion: process.env.ASSESSOR_RAG_ACTIVE_CORPUS_VERSION?.trim(),
      bsiCatalogs,
      ...(!options.aiClient ? { disabledReason: 'Work LiteLLM is not configured.' } : {})
    });
  } catch (error) {
    disabledReason = error instanceof Error ? error.message : 'Regulatory mapping is unavailable.';
    mappingService = new RegulatoryMappingService(read, store, { disabledReason, bsiCatalogs });
  }
  await mappingService.recoverInterruptedRuns();
  const autoMapping = new RegulatoryAutoMapping({ settingsPath: join(corpusRoot, 'auto-mapping.json'), read, mapping: mappingService });
  const corpusUploadService = new RegulatoryCorpusUploadService({
    root: corpusRoot,
    ingestion: rag?.ingestion,
    countIndexed: rag ? (corpusId, corpusVersion) => rag!.index.count({ corpusId, corpusVersion }) : undefined,
    pruneIndexed: rag ? (corpusId, corpusVersion) => rag!.index.deleteOtherCorpusVersions(corpusId, corpusVersion) : undefined,
    mapping: mappingService,
    autoMapping,
    embeddingModel: rag?.embeddingModel,
    disabledReason
  });
  await corpusUploadService.initialize();
  await autoMapping.start();
  const aiOverviewService = new AiOverviewService(read, store, { ...options.aiOverview, client: options.aiClient });
  await aiOverviewService.recoverInterruptedRuns();
  return {
    mappingService,
    corpusUploadService,
    aiOverviewService,
    bsiCatalogService,
    store,
    stop: () => Promise.all([autoMapping.stop(), corpusUploadService.stop().then(() => mappingService.close()), aiOverviewService.close()]).then(() => undefined),
    close: async () => { await pool?.end(); }
  };
}

function compliancePool(environment: RepositoryEnvironment): Pool {
  const connectionString = repositoryConnectionString(environment);
  if (!connectionString) throw new Error('The Compliance Engine needs the existing PostgreSQL connection.');
  const hostname = new URL(connectionString).hostname;
  const sslMode = environment.ASSESSOR_DATABASE_SSL?.trim()
    || (['localhost', '127.0.0.1', '[::1]'].includes(hostname) ? 'disable' : 'require');
  return new Pool({
    connectionString,
    ssl: sslConfiguration(sslMode),
    max: 4,
    connectionTimeoutMillis: 5000,
    idleTimeoutMillis: 30000,
    application_name: 'assessor-compliance-engine'
  });
}