569 lines
29 KiB
TypeScript
569 lines
29 KiB
TypeScript
import { createHash, randomUUID } from 'node:crypto';
|
||
import type { AiCompletionClient } from '../ai/litellm-client';
|
||
import { LiteLlmClientError } from '../ai/litellm-client';
|
||
import type { AssessorRepository } from '../../src/persistence/assessor-repository';
|
||
import { RepositoryConflictError, RepositoryNotFoundError, RepositoryValidationError } from '../../src/persistence/repository-errors';
|
||
import {
|
||
REGULATORY_MAPPING_OUTPUT_SCHEMA_VERSION,
|
||
REGULATORY_MAPPING_PROMPT_VERSION,
|
||
REGULATORY_RETRIEVAL_VERSION,
|
||
emptyRegulatoryMappingBatchCounts,
|
||
groundRegulatoryMappingResult,
|
||
normalizeRegulatoryMappingBatchScope,
|
||
regulatoryMappingResultValidationErrors,
|
||
type RegulatoryMappingBatchCreateRequest,
|
||
type RegulatoryMappingBatchResponse,
|
||
type RegulatoryMappingBatchRun,
|
||
type RegulatoryMappingBatchScope,
|
||
type RegulatoryMappingCandidate,
|
||
type RegulatoryMappingCreateRequest,
|
||
type RegulatoryMappingDecision,
|
||
type RegulatoryMappingDecisionCreateRequest,
|
||
type RegulatoryMappingEvidence,
|
||
type RegulatoryMappingResponse,
|
||
type RegulatoryMappingResult,
|
||
type RegulatoryMappingRun
|
||
} from '../../src/domain/regulatory-mapping';
|
||
import type { RegulatoryRetrievalCandidate } from '../../src/domain/regulatory-rag';
|
||
import type { UnifiedResultProjection } from '../../src/domain/unified-result';
|
||
import { privacySafeFinding } from '../../src/domain/privacy-safe-finding';
|
||
import type { RegulatoryFindingRetrievalService } from './regulatory-finding-retrieval';
|
||
import type { RegulatoryChunkIndex } from './qdrant-regulatory-index';
|
||
import { buildRegulatoryFindingQuery, isRegulatoryMappableResult } from '../../src/profiles/public-search/regulatory-query';
|
||
import { canonicalJson } from '../../src/shared/canonical-json';
|
||
|
||
const POLL_AFTER_MS = 1_500;
|
||
const UNSUPPORTED_KIND_MESSAGE = 'Only canonical scanner findings and GPO baseline comparisons can be mapped in this version.';
|
||
const MAXIMUM_PROMPT_CHARACTERS = 48_000;
|
||
const BATCH_PAGE_SIZE = 500;
|
||
const DEFAULT_MAXIMUM_CONCURRENT_RUNS = 2;
|
||
|
||
export interface RegulatoryMappingServiceOptions {
|
||
findingRetrieval?: RegulatoryFindingRetrievalService;
|
||
index?: RegulatoryChunkIndex;
|
||
client?: AiCompletionClient;
|
||
corpusId?: string;
|
||
corpusVersion?: string;
|
||
embeddingModel?: string;
|
||
disabledReason?: string;
|
||
now?: () => Date;
|
||
maximumConcurrentRuns?: number;
|
||
}
|
||
|
||
export class RegulatoryMappingService {
|
||
private readonly activeTasks = new Set<Promise<void>>();
|
||
private readonly now: () => Date;
|
||
private readonly maximumConcurrentRuns: number;
|
||
private readonly workQueue: Array<{
|
||
task: () => Promise<'succeeded' | 'failed'>;
|
||
resolve: (status: 'succeeded' | 'failed') => void;
|
||
}> = [];
|
||
private activeWorkerCount = 0;
|
||
private activationPending = false;
|
||
|
||
constructor(private readonly repository: AssessorRepository, private readonly options: RegulatoryMappingServiceOptions = {}) {
|
||
this.now = options.now ?? (() => new Date());
|
||
this.maximumConcurrentRuns = options.maximumConcurrentRuns ?? DEFAULT_MAXIMUM_CONCURRENT_RUNS;
|
||
if (!Number.isInteger(this.maximumConcurrentRuns) || this.maximumConcurrentRuns < 1 || this.maximumConcurrentRuns > 8) {
|
||
throw new Error('Regulatory mapping concurrency must be an integer between 1 and 8.');
|
||
}
|
||
}
|
||
|
||
get generationEnabled(): boolean {
|
||
return !this.activationPending && Boolean(this.options.findingRetrieval && this.options.index && this.options.client
|
||
&& this.options.corpusId && this.options.corpusVersion && this.options.embeddingModel);
|
||
}
|
||
|
||
async activateCorpus(corpusId: string, corpusVersion: string): Promise<void> {
|
||
this.activationPending = true;
|
||
try {
|
||
await this.close();
|
||
this.options.corpusId = corpusId;
|
||
this.options.corpusVersion = corpusVersion;
|
||
} finally {
|
||
this.activationPending = false;
|
||
}
|
||
}
|
||
|
||
async recoverInterruptedRuns(): Promise<void> {
|
||
const completedAt = this.now().toISOString();
|
||
for (const run of (await this.repository.listRegulatoryMappingRuns()).filter(item => ['queued', 'running'].includes(item.status))) {
|
||
await this.repository.saveRegulatoryMappingRun({
|
||
...run,
|
||
status: 'failed',
|
||
updatedAt: completedAt,
|
||
completedAt,
|
||
errorCode: 'regulatory_mapping_interrupted'
|
||
});
|
||
}
|
||
for (const project of await this.repository.listProjects()) {
|
||
for (const run of (await this.repository.listRegulatoryMappingBatchRuns(project.id)).filter(item => ['queued', 'running'].includes(item.status))) {
|
||
await this.repository.updateRegulatoryMappingBatchRun({
|
||
...run,
|
||
status: 'failed',
|
||
updatedAt: completedAt,
|
||
completedAt,
|
||
errorCode: 'regulatory_mapping_batch_interrupted',
|
||
errorDetail: 'Batch processing was interrupted before completion. Start a new batch to retry remaining rules.'
|
||
});
|
||
}
|
||
}
|
||
}
|
||
|
||
async get(projectId: string, resultProjectionId: string): Promise<RegulatoryMappingResponse> {
|
||
const finding = await this.finding(projectId, resultProjectionId);
|
||
const supported = Boolean(buildRegulatoryFindingQuery(finding));
|
||
const availability = supported ? await this.availability() : { enabled: false };
|
||
const enabled = availability.enabled && supported;
|
||
const ruleKey = supported ? this.ruleKey(finding) : undefined;
|
||
const run = ruleKey && this.options.client ? await this.repository.getRegulatoryMappingRun(this.mappingId(ruleKey)) : null;
|
||
const decisions = ruleKey ? await this.repository.listRegulatoryMappingDecisions(ruleKey) : [];
|
||
return {
|
||
version: 1,
|
||
projectId,
|
||
resultProjectionId,
|
||
generationEnabled: enabled,
|
||
state: run?.status ?? (enabled ? 'empty' : 'unsupported'),
|
||
...(!supported ? { disabledReason: !isRegulatoryMappableResult(finding)
|
||
? UNSUPPORTED_KIND_MESSAGE
|
||
: 'This finding has insufficient public context for regulatory mapping.' }
|
||
: !availability.enabled ? { disabledReason: availability.reason ?? this.options.disabledReason ?? 'Regulatory mapping is not configured.' } : {}),
|
||
...(run && ['queued', 'running'].includes(run.status) ? { pollAfterMs: POLL_AFTER_MS } : {}),
|
||
...(run ? { run: withoutEvidenceText(run) } : {}),
|
||
decisions
|
||
};
|
||
}
|
||
|
||
async start(projectId: string, request: RegulatoryMappingCreateRequest): Promise<RegulatoryMappingResponse> {
|
||
if (!this.generationEnabled) throw new RegulatoryMappingServiceError('regulatory_mapping_disabled', this.options.disabledReason ?? 'Regulatory mapping is not configured.', 503);
|
||
const finding = await this.finding(projectId, request.resultProjectionId);
|
||
if (!isRegulatoryMappableResult(finding)) throw new RegulatoryMappingServiceError('unsupported_finding_kind', UNSUPPORTED_KIND_MESSAGE, 422);
|
||
if (!buildRegulatoryFindingQuery(finding)) throw new RegulatoryMappingServiceError('insufficient_public_context', 'This finding has insufficient public context for regulatory mapping.', 422);
|
||
const claimed = await this.claim(finding, true);
|
||
if (claimed.claimed) this.track(this.enqueue(() => this.process(claimed.run, finding)).then(() => undefined));
|
||
return this.get(projectId, request.resultProjectionId);
|
||
}
|
||
|
||
async getBatch(projectId: string, requestedScope: RegulatoryMappingBatchScope): Promise<RegulatoryMappingBatchResponse> {
|
||
const scope = await this.validateBatchScope(projectId, requestedScope);
|
||
const scopeKey = this.batchScopeKey(projectId, scope);
|
||
const run = (await this.repository.listRegulatoryMappingBatchRuns(projectId, scopeKey))[0];
|
||
const availability = await this.availability();
|
||
return {
|
||
version: 1,
|
||
projectId,
|
||
scope,
|
||
generationEnabled: availability.enabled,
|
||
state: run?.status ?? 'empty',
|
||
...(!availability.enabled ? { disabledReason: availability.reason ?? this.options.disabledReason ?? 'Regulatory mapping is not configured.' } : {}),
|
||
...(run && ['queued', 'running'].includes(run.status) ? { pollAfterMs: POLL_AFTER_MS } : {}),
|
||
...(run ? { run } : {})
|
||
};
|
||
}
|
||
|
||
async startBatch(projectId: string, request: RegulatoryMappingBatchCreateRequest): Promise<RegulatoryMappingBatchResponse> {
|
||
if (!this.generationEnabled) throw new RegulatoryMappingServiceError('regulatory_mapping_disabled', this.options.disabledReason ?? 'Regulatory mapping is not configured.', 503);
|
||
if (typeof request.idempotencyKey !== 'string' || !/^[A-Za-z0-9._:-]{8,128}$/.test(request.idempotencyKey)) {
|
||
throw new RepositoryValidationError('idempotencyKey must contain 8 to 128 safe characters.');
|
||
}
|
||
const scope = await this.validateBatchScope(projectId, request);
|
||
const timestamp = this.now().toISOString();
|
||
const candidate: RegulatoryMappingBatchRun = {
|
||
version: 1,
|
||
id: `regulatory-mapping-batch-${randomUUID()}`,
|
||
projectId,
|
||
scopeKey: this.batchScopeKey(projectId, scope),
|
||
filterHash: sha256(canonicalJson(scope.filters)),
|
||
assessmentIds: scope.assessmentIds,
|
||
scanRunIds: scope.scanRunIds,
|
||
filters: scope.filters,
|
||
idempotencyKey: request.idempotencyKey,
|
||
status: 'queued',
|
||
counts: emptyRegulatoryMappingBatchCounts(),
|
||
createdAt: timestamp,
|
||
updatedAt: timestamp
|
||
};
|
||
const run = await this.repository.createRegulatoryMappingBatchRun(candidate);
|
||
if (run.id === candidate.id) this.track(this.processBatch(run));
|
||
return this.getBatch(projectId, {
|
||
assessmentIds: run.assessmentIds,
|
||
scanRunIds: run.scanRunIds,
|
||
filters: run.filters
|
||
});
|
||
}
|
||
|
||
async decide(projectId: string, request: RegulatoryMappingDecisionCreateRequest): Promise<RegulatoryMappingResponse> {
|
||
if (!this.generationEnabled) throw new RegulatoryMappingServiceError('regulatory_mapping_disabled', this.options.disabledReason ?? 'Regulatory mapping is not configured.', 503);
|
||
const finding = await this.finding(projectId, request.resultProjectionId);
|
||
if (!isRegulatoryMappableResult(finding)) throw new RegulatoryMappingServiceError('unsupported_finding_kind', UNSUPPORTED_KIND_MESSAGE, 422);
|
||
const ruleKey = this.ruleKey(finding);
|
||
const comment = request.comment?.trim();
|
||
const decision: RegulatoryMappingDecision = {
|
||
version: 1,
|
||
id: `regulatory-decision-${randomUUID()}`,
|
||
ruleKey,
|
||
sourceTool: finding.sourceTool ?? 'unknown',
|
||
...(finding.sourceRuleId ? { sourceRuleId: finding.sourceRuleId } : {}),
|
||
mappingId: this.mappingId(ruleKey),
|
||
controlId: request.controlId,
|
||
decision: request.decision,
|
||
...(comment ? { comment } : {}),
|
||
createdAt: this.now().toISOString()
|
||
};
|
||
await this.repository.createRegulatoryMappingDecision(decision, request.expectedLatestDecisionId);
|
||
return this.get(projectId, request.resultProjectionId);
|
||
}
|
||
|
||
// The global sharing key: the rule identity `tool:ruleId`, shared across every tenant.
|
||
// Rules without an id fall back to a stable hash of the non-personal check title so
|
||
// identical findings still share, while never keying on tenant or occurrence.
|
||
private ruleKey(finding: UnifiedResultProjection): string {
|
||
const tool = finding.sourceTool?.trim() || 'unknown';
|
||
const ruleId = finding.sourceRuleId?.trim();
|
||
if (ruleId) return `${tool}:${ruleId}`.slice(0, 200);
|
||
const hash = createHash('sha256').update(privacySafeFinding(finding).title.trim(), 'utf8').digest('hex').slice(0, 24);
|
||
return `${tool}#title:${hash}`.slice(0, 200);
|
||
}
|
||
|
||
// Deterministic id per (ruleKey, pipeline versions) so one row exists per rule per
|
||
// version tuple: regeneration reuses it, and a version bump yields a new mapping.
|
||
private mappingId(ruleKey: string): string {
|
||
const tuple = [ruleKey, this.options.corpusVersion, REGULATORY_RETRIEVAL_VERSION, REGULATORY_MAPPING_PROMPT_VERSION, this.options.client!.model].join(' |