Per-Phase Provenance Logging: Designing Audit Schemas for Compliant Financial Agent Pipelines

import { z } from 'zod';
import { randomUUID } from 'crypto';

const PhaseProvenanceRecord = z.object({
  record_id: z.string().uuid(),
  correlation_id: z.string().uuid(),       // threads the full transaction
  transaction_id: z.string(),
  phase_name: z.string(),
  phase_schema_version: z.string(),        // e.g. 'eligibility_check_v2'
  entered_at: z.string().datetime(),
  exited_at: z.string().datetime().nullable(),
  input_snapshot: z.record(z.unknown()),   // frozen copy of inputs at phase entry
  model_id: z.string().nullable(),         // null if rule-based phase
  tool_calls: z.array(z.object({
    tool_name: z.string(),
    input: z.record(z.unknown()),
    output: z.record(z.unknown()),
    latency_ms: z.number()
  })),
  phase_outcome: z.enum(['pass', 'deny', 'escalate', 'error']),
  outcome_rationale: z.string(),           // human-readable, written by agent
  human_override: z.boolean().default(false),
  human_override_actor: z.string().nullable(),
  tamper_hash: z.string()                  // SHA-256 of record minus this field
});

type PhaseProvenanceRecord = z.infer<typeof PhaseProvenanceRecord>;

async function emitPhaseProvenance(
  partial: Omit<PhaseProvenanceRecord, 'record_id' | 'tamper_hash'>,
  store: ProvenanceStore
): Promise<PhaseProvenanceRecord> {
  const record_id = randomUUID();
  const withoutHash = { record_id, ...partial };
  const tamper_hash = await sha256(JSON.stringify(withoutHash));
  const record = PhaseProvenanceRecord.parse({ ...withoutHash, tamper_hash });
  await store.writeAtomic(record);  // write before acting on phase result
  return record;
}

// TypeScript: phase wrapper that guarantees provenance before state advance
async function runPhaseWithProvenance<TInput, TOutput>(
  phaseName: string,
  schemaVersion: string,
  input: TInput,
  executor: (input: TInput) => Promise<{
    outcome: 'pass' | 'deny' | 'escalate' | 'error';
    rationale: string;
    toolCalls: ToolCall[];
    result: TOutput;
  }>,
  ctx: { correlationId: string; transactionId: string; store: ProvenanceStore }
): Promise<TOutput> {
  const enteredAt = new Date().toISOString();
  let executorResult;

  try {
    executorResult = await executor(input);
  } catch (err) {
    // still emit provenance on error — do not swallow the audit trail
    await emitPhaseProvenance({
      correlation_id: ctx.correlationId,
      transaction_id: ctx.transactionId,
      phase_name: phaseName,
      phase_schema_version: schemaVersion,
      entered_at: enteredAt,
      exited_at: new Date().toISOString(),
      input_snapshot: input as Record<string, unknown>,
      model_id: null,
      tool_calls: [],
      phase_outcome: 'error',
      outcome_rationale: `Phase failed: ${(err as Error).message}`,
      human_override: false,
      human_override_actor: null
    }, ctx.store);
    throw err;
  }

  await emitPhaseProvenance({
    correlation_id: ctx.correlationId,
    transaction_id: ctx.transactionId,
    phase_name: phaseName,
    phase_schema_version: schemaVersion,
    entered_at: enteredAt,
    exited_at: new Date().toISOString(),
    input_snapshot: input as Record<string, unknown>,
    model_id: null,
    tool_calls: executorResult.toolCalls,
    phase_outcome: executorResult.outcome,
    outcome_rationale: executorResult.rationale,
    human_override: false,
    human_override_actor: null
  }, ctx.store);

  return executorResult.result;
}

interface AgentHandoffRecord {
  handoff_id: string;
  correlation_id: string;          // same root as phase records
  transaction_id: string;
  from_agent: string;
  to_agent: string;
  handoff_at: string;              // ISO datetime
  routing_criteria: {
    field: string;
    operator: string;
    threshold: unknown;
    observed_value: unknown;
  }[];
  routing_confidence: number | null; // null for deterministic rule-based routing
  human_override: boolean;
  payload_snapshot: Record<string, unknown>; // what was handed off
}

// Usage at handoff point inside a CrewAI task or LangGraph edge
const handoff: AgentHandoffRecord = {
  handoff_id: randomUUID(),
  correlation_id: ctx.correlationId,
  transaction_id: ctx.transactionId,
  from_agent: 'risk_scoring_agent',
  to_agent: 'human_review_queue',
  handoff_at: new Date().toISOString(),
  routing_criteria: [{
    field: 'risk_score',
    operator: 'gt',
    threshold: 80,
    observed_value: riskScore
  }],
  routing_confidence: null,        // deterministic rule, not probabilistic
  human_override: false,
  payload_snapshot: transactionSnapshot
};

await provenanceStore.writeHandoff(handoff);

// Inject infrastructure execution context into every phase provenance record
interface ExecutionContext {
  model_id: string;
  model_version: string;
  retrieval_index_commit: string;
  policy_document_hash: string;
  infrastructure_config_hash: string;  // SHA-256 of IaC-resolved config
  agent_image_digest: string;           // container image digest
}

async function resolveExecutionContext(
  configStore: ConfigStore
): Promise<ExecutionContext> {
  const [model, index, policy, infra, image] = await Promise.all([
    configStore.get('ACTIVE_MODEL_VERSION'),
    configStore.get('RETRIEVAL_INDEX_COMMIT'),
    configStore.getHash('POLICY_DOCUMENTS'),
    configStore.getInfraHash(),          // from IaC state API
    configStore.get('AGENT_IMAGE_DIGEST')
  ]);
  return {
    model_id: model.id,
    model_version: model.version,
    retrieval_index_commit: index,
    policy_document_hash: policy,
    infrastructure_config_hash: infra,
    agent_image_digest: image
  };
}

// Add to PhaseProvenanceRecord schema:
// execution_context: ExecutionContext  <-- frozen at phase entry, not phase exit

// TypeScript: provenance-safe redaction before write
const REDACT_PATTERNS: RegExp[] = [
  /\bsk-[A-Za-z0-9]{32,}\b/g,           // OpenAI-style API keys
  /\b[A-Z0-9]{20}:[A-Za-z0-9+/]{40}\b/g, // AWS-style credentials
  /\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b/g  // card numbers
];

function redactSensitiveFields(
  snapshot: Record<string, unknown>
): { sanitized: Record<string, unknown>; redacted_paths: string[] } {
  const redacted_paths: string[] = [];

  function walk(obj: Record<string, unknown>, path: string): Record<string, unknown> {
    return Object.fromEntries(
      Object.entries(obj).map(([k, v]) => {
        const currentPath = path ? `${path}.${k}` : k;
        if (typeof v === 'string') {
          const hadSecret = REDACT_PATTERNS.some(r => r.test(v));
          if (hadSecret) {
            redacted_paths.push(currentPath);
            return [k, '[REDACTED]'];
          }
        }
        if (v && typeof v === 'object' && !Array.isArray(v)) {
          return [k, walk(v as Record<string, unknown>, currentPath)];
        }
        return [k, v];
      })
    );
  }

  return { sanitized: walk(snapshot, ''), redacted_paths };
}

// Extend PhaseProvenanceRecord for loop-aware provenance
interface LoopAwarePhaseRecord extends PhaseProvenanceRecord {
  loop_sequence: number;              // 1-indexed iteration within this phase
  is_correction_iteration: boolean;   // true if triggered by human feedback
  prior_iteration_id: string | null;  // links to previous loop iteration record
  context_truncation_occurred: boolean;
  truncated_context_paths: string[];  // which context fields were dropped
}

// Utility to emit loop-aware record and link to predecessor
async function emitLoopPhaseProvenance(
  base: Omit<PhaseProvenanceRecord, 'record_id' | 'tamper_hash'>,
  loopMeta: {
    loop_sequence: number;
    is_correction_iteration: boolean;
    prior_iteration_id: string | null;
    context_truncation_occurred: boolean;
    truncated_context_paths: string[];
  },
  store: ProvenanceStore
): Promise<LoopAwarePhaseRecord> {
  const record_id = randomUUID();
  const merged = { record_id, ...base, ...loopMeta };
  const tamper_hash = await sha256(JSON.stringify(merged));
  const record = { ...merged, tamper_hash } as LoopAwarePhaseRecord;
  await store.writeAtomic(record);
  return record;
}

// Mark provenance records during canary deployment windows
interface DeploymentMarker {
  deployment_id: string;
  is_canary: boolean;
  canary_percentage: number | null;    // null when not canary
  infrastructure_config_hash: string;
  rollback_target_hash: string | null; // hash of the version to restore to
}

// Injected at startup from deployment config, shared across all phase records
const deploymentMarker: DeploymentMarker = await configStore.getDeploymentMarker();

// Divergence check: compare outcome distributions between canary and stable
async function detectProvenanceDivergence(
  store: ProvenanceStore,
  canaryHash: string,
  stableHash: string,
  windowMinutes: number = 30
): Promise<{ diverged: boolean; escalation_rate_delta: number }> {
  const since = new Date(Date.now() - windowMinutes * 60_000).toISOString();
  const [canary, stable] = await Promise.all([
    store.getOutcomeDistribution(canaryHash, since),
    store.getOutcomeDistribution(stableHash, since)
  ]);
  const delta = canary.escalation_rate - stable.escalation_rate;
  return { diverged: Math.abs(delta) > 0.05, escalation_rate_delta: delta };
}