LWLT-AIBOT/control-plane/src/parser-orchestrator.ts

import { performance } from 'node:perf_hooks';
import type { BusinessRouteId, ParserMode } from './business-routes.js';
import { compareParserResults } from './parser-contract.js';
import { parseProgramInput, programFallbackAllowed, programParserVersion } from './program-parser.js';
import type { ParseDecisionInput, ParseTaskClaim } from './task-service.js';

export interface AiParser {
  parse(input: {
    rawText: string;
    taskId: string;
    receivedAt: string;
    turnNo?: number;
    sessionId?: string;
    recoveryCount?: number;
    history?: Array<{ turnNo: number; rawText: string; receivedAt?: string }>;
    signal?: AbortSignal;
  }): Promise<unknown>;
  checkConnection?: () => Promise<unknown>;
}

export interface ParserOrchestrationResult {
  result: unknown;
  decision: ParseDecisionInput;
}

function record(value: unknown): Record<string, unknown> {
  return value && typeof value === 'object' && !Array.isArray(value) ? value as Record<string, unknown> : {};
}

function elapsed(startedAt: number): number {
  return Math.max(0, Math.round(performance.now() - startedAt));
}

function promptVersion(result: unknown): string | null {
  const value = record(result);
  return String(value.parser_prompt_version || '').trim() || null;
}

function programVersion(result: unknown, routeId: BusinessRouteId | null): string | null {
  const value = record(result);
  return String(value.program_parser_version || '').trim() || programParserVersion(routeId);
}

export class ParserOrchestrator {
  constructor(private readonly aiParser: AiParser) {}

  private aiInput(claim: ParseTaskClaim, signal?: AbortSignal) {
    return {
      rawText: claim.messageText,
      taskId: claim.task.task_id,
      receivedAt: claim.task.created_at,
      turnNo: claim.turnNo,
      sessionId: claim.sessionId,
      recoveryCount: claim.recoveryCount,
      history: claim.history,
      signal
    };
  }

  private programInput(claim: ParseTaskClaim) {
    return {
      rawText: claim.messageText,
      receivedAt: claim.task.created_at,
      routeId: claim.task.parser.route_id,
      history: claim.history.map((item) => ({ text: item.rawText, receivedAt: item.receivedAt }))
    };
  }

  async parse(claim: ParseTaskClaim, signal?: AbortSignal): Promise<ParserOrchestrationResult> {
    const routeId = claim.task.parser.route_id;
    const configuredMode: ParserMode = claim.task.parser.configured_mode;
    const affinity = claim.task.parser.engine_affinity;

    // Unresolved/legacy traffic and tasks explicitly pinned to AI preserve the
    // original parser path. This is also the permanent affinity after an Auto
    // fallback or the one-time administrator AI reparse.
    if (!routeId || configuredMode === 'ai' || affinity === 'ai') {
      const started = performance.now();
      const aiResult = await this.aiParser.parse(this.aiInput(claim, signal));
      return {
        result: aiResult,
        decision: {
          routeId,
          configuredMode,
          authoritativeEngine: 'ai',
          aiPromptVersion: promptVersion(aiResult),
          aiResult,
          aiDurationMs: elapsed(started),
          fallbackReason: claim.task.parser.fallback_reason === 'manual_ai_reparse'
            ? 'manual_ai_reparse'
            : null
        }
      };
    }

    if (configuredMode === 'shadow') {
      const programStarted = performance.now();
      const programPromise = parseProgramInput(this.programInput(claim))
        .then((result) => ({ result, duration: elapsed(programStarted) }));
      const aiStarted = performance.now();
      const aiPromise = this.aiParser.parse(this.aiInput(claim, signal))
        .then((result) => ({ result, duration: elapsed(aiStarted) }));
      const [program, ai] = await Promise.all([programPromise, aiPromise]);
      const comparison = compareParserResults(ai.result, program.result);
      return {
        // AI is the sole authority in Shadow. The program candidate is only
        // persisted encrypted for comparison and cannot reach confirmation.
        result: ai.result,
        decision: {
          routeId,
          configuredMode,
          authoritativeEngine: 'ai',
          programVersion: programVersion(program.result, routeId),
          aiPromptVersion: promptVersion(ai.result),
          comparisonStatus: comparison.status,
          diffPaths: comparison.diff_paths,
          programResult: program.result,
          aiResult: ai.result,
          programDurationMs: program.duration,
          aiDurationMs: ai.duration
        }
      };
    }

    const programStarted = performance.now();
    const programResult = await parseProgramInput(this.programInput(claim));
    const programDuration = elapsed(programStarted);
    if (configuredMode === 'program' || !programFallbackAllowed(programResult)) {
      return {
        result: programResult,
        decision: {
          routeId,
          configuredMode,
          authoritativeEngine: 'program',
          programVersion: programVersion(programResult, routeId),
          programResult,
          programDurationMs: programDuration
        }
      };
    }

    // Auto may reach this branch only for the fixed technical allowlist. A
    // complete AI result becomes authoritative; candidates are never merged.
    const fallbackReason = String(record(programResult).error_code || 'program_internal_error');
    const aiStarted = performance.now();
    const aiResult = await this.aiParser.parse(this.aiInput(claim, signal));
    return {
      result: aiResult,
      decision: {
        routeId,
        configuredMode,
        authoritativeEngine: 'ai',
        programVersion: programVersion(programResult, routeId),
        aiPromptVersion: promptVersion(aiResult),
        fallbackReason,
        programResult,
        aiResult,
        programDurationMs: programDuration,
        aiDurationMs: elapsed(aiStarted)
      }
    };
  }
}