memind/services/orchestrator/contracts.mjs

import crypto from 'node:crypto';

export const ORCHESTRATOR_MODE = Object.freeze({
  OFF: 'off',
  SHADOW: 'shadow',
  CANARY: 'canary',
  ACTIVE: 'active',
});

export const WORKFLOW_ENGINE = Object.freeze({
  NATIVE: 'native',
  LANGGRAPH: 'langgraph',
});

export const DEFAULT_ORCHESTRATED_WORKFLOWS = Object.freeze([
  'code-run-v1',
]);

export const PAGE_DATA_VALIDATION_OBSERVATION_VERSION =
  'page-data-validation-observation-v1';

const ENGINE_ID_PATTERN = /^[a-z][a-z0-9_-]{0,63}$/;
const WORKFLOW_NAME_PATTERN = /^[a-z][a-z0-9_-]{0,127}$/;
const VALIDATION_CHECK_ID_PATTERN = /^[a-z][a-z0-9_-]{0,127}$/;
const VALIDATION_CHECK_STATUSES = new Set(['passed', 'failed', 'skipped']);
const REQUIRED_PAGE_DATA_CHECKS = Object.freeze([
  'agent_run_completion',
  'page_data_binding',
  'page_data_storage_policy',
]);

function normalizeIdentifier(value, pattern, fallback) {
  const normalized = String(value ?? '').trim().toLowerCase();
  return pattern.test(normalized) ? normalized : fallback;
}

export function normalizeWorkflowEngineId(value, fallback = WORKFLOW_ENGINE.NATIVE) {
  return normalizeIdentifier(value, ENGINE_ID_PATTERN, fallback);
}

export function normalizeWorkflowName(value, fallback = '') {
  return normalizeIdentifier(value, WORKFLOW_NAME_PATTERN, fallback);
}

export function normalizeStringList(value, {
  limit = 200,
  itemLimit = 128,
  normalize = (item) => String(item ?? '').trim(),
} = {}) {
  if (!Array.isArray(value)) return [];
  return [...new Set(value
    .map((item) => normalize(item).slice(0, itemLimit))
    .filter(Boolean))]
    .slice(0, limit);
}

export function normalizeRunSpec(input = {}) {
  const source = input && typeof input === 'object' && !Array.isArray(input) ? input : {};
  const requestId = String(source.requestId ?? '').trim();
  const runId = String(source.runId ?? crypto.randomUUID()).trim();
  const workflowName = normalizeWorkflowName(
    source.workflow?.name ?? source.workflowName,
  );
  if (!requestId) throw new Error('RunSpec requires requestId');
  if (!runId) throw new Error('RunSpec requires runId');
  if (!workflowName) throw new Error('RunSpec requires workflow.name');
  return {
    version: 'orchestrator-run-v1',
    runId,
    requestId,
    workflow: {
      name: workflowName,
      version: Math.max(1, Number(source.workflow?.version ?? 1) || 1),
    },
    subject: {
      tenantId: String(source.subject?.tenantId ?? '').trim() || null,
      userId: String(source.subject?.userId ?? '').trim() || null,
    },
    input: source.input && typeof source.input === 'object' ? structuredClone(source.input) : {},
    policy: source.policy && typeof source.policy === 'object' ? structuredClone(source.policy) : {},
    limits: source.limits && typeof source.limits === 'object' ? structuredClone(source.limits) : {},
    metadata: source.metadata && typeof source.metadata === 'object'
      ? structuredClone(source.metadata)
      : {},
  };
}

export function buildRunEvent({
  eventId = crypto.randomUUID(),
  runId,
  sequence,
  type,
  timestamp = Date.now(),
  data = null,
} = {}) {
  const normalizedRunId = String(runId ?? '').trim();
  const normalizedType = String(type ?? '').trim();
  if (!normalizedRunId) throw new Error('RunEvent requires runId');
  if (!normalizedType) throw new Error('RunEvent requires type');
  return {
    version: 'orchestrator-event-v1',
    eventId: String(eventId),
    runId: normalizedRunId,
    sequence: Math.max(0, Number(sequence ?? 0) || 0),
    type: normalizedType,
    timestamp: Number(timestamp) || Date.now(),
    data: data == null ? null : structuredClone(data),
  };
}

function validationContractError(message) {
  const error = new Error(message);
  error.code = 'WORKFLOW_VALIDATION_INVALID';
  error.status = 422;
  return error;
}

function normalizeValidationCount(value) {
  const parsed = Number(value);
  if (!Number.isFinite(parsed) || parsed < 0) return 0;
  return Math.min(1_000_000, Math.floor(parsed));
}

export function normalizePageDataValidationObservation(input = {}) {
  const source = input && typeof input === 'object' && !Array.isArray(input) ? input : {};
  const idempotencyKey = String(source.idempotencyKey ?? '').trim().slice(0, 200);
  if (!idempotencyKey) {
    throw validationContractError('Page Data validation observation requires idempotencyKey');
  }
  if (!Array.isArray(source.checks) || source.checks.length === 0) {
    throw validationContractError('Page Data validation observation requires checks');
  }
  const checks = source.checks.slice(0, 32).map((item) => {
    const check = item && typeof item === 'object' && !Array.isArray(item) ? item : {};
    const id = normalizeIdentifier(check.id, VALIDATION_CHECK_ID_PATTERN, '');
    const status = String(check.status ?? '').trim().toLowerCase();
    if (!id || !VALIDATION_CHECK_STATUSES.has(status)) {
      throw validationContractError('Page Data validation observation contains an invalid check');
    }
    return {
      id,
      status,
      codes: normalizeStringList(check.codes, {
        limit: 32,
        itemLimit: 128,
        normalize: (value) => String(value ?? '').trim().toUpperCase(),
      }),
    };
  });
  const byId = new Map(checks.map((check) => [check.id, check]));
  const failed = checks.some((check) => check.status === 'failed');
  const required = source.required === true;
  const requiredChecksPassed = REQUIRED_PAGE_DATA_CHECKS.every(
    (id) => byId.get(id)?.status === 'passed',
  );
  const verdict = failed
    ? 'failed'
    : required && !requiredChecksPassed
      ? 'inconclusive'
      : checks.some((check) => check.status === 'passed')
        ? 'passed'
        : 'inconclusive';
  const taskType = normalizeWorkflowName(source.taskType, '');
  const observedAt = Number(source.observedAt);
  return {
    version: PAGE_DATA_VALIDATION_OBSERVATION_VERSION,
    kind: 'page-data-delivery',
    idempotencyKey,
    taskType: taskType || null,
    required,
    verdict,
    checks,
    metrics: {
      pageCount: normalizeValidationCount(source.metrics?.pageCount),
      publicationCount: normalizeValidationCount(source.metrics?.publicationCount),
    },
    source: normalizeIdentifier(
      source.source,
      VALIDATION_CHECK_ID_PATTERN,
      'portal-agent-run',
    ),
    dataPolicy: 'control-plane-only-v1',
    observedAt: Number.isFinite(observedAt) && observedAt > 0 ? observedAt : Date.now(),
  };
}

export function stableRolloutBucket(value) {
  const hash = crypto.createHash('sha256').update(String(value ?? '')).digest();
  return hash.readUInt32BE(0) % 100;
}