"""Retrospective model-output replay, not a fresh vendor head-to-head. Run against the public revision in an isolated checkout with its core/demos on PYTHONPATH. Never invokes an external API or a domain's production executor. """ from __future__ import annotations import argparse import hashlib import importlib import json from pathlib import Path import sys PUBLIC_REVISION = '85fe4dc8a4f5a7010dd19f98bd8a934554fb8663' DOMAINS = { 'soc': 'soc-playbook-router', 'sre': 'sre-policy-router', 'drone': 'drone-mission-router', 'grid': 'grid-ops-router', 'finance': 'finance-risk-router', 'medical': 'medical-workflow-router', 'customer_support': 'customer-support-router', 'content_moderation': 'content-moderation-router', } MODELS = {'gemma4:e2b': 'gemma4_e2b', 'qwen3.5:0.8b': 'qwen3.5_0.8b'} def index_rows(rows): result = {r['case_id']: r for r in rows} if len(result) != len(rows): raise ValueError('Duplicate case IDs') return result def validate_alignment(cases, outputs): expected, actual = index_rows(cases), index_rows(outputs) if set(expected) != set(actual): raise ValueError('Case ID coverage mismatch') for case_id, row in actual.items(): for field in ('expected_action_id', 'supported'): if row[field] != expected[case_id][field]: raise ValueError(f'Label mismatch: {case_id} {field}') return actual def score(rows): supported = [r for r in rows if r['supported']] unsupported = [r for r in rows if not r['supported']] correct = sum(r['correct'] for r in supported) false_accepts = sum(r['final_status'] == 'executed' for r in unsupported) invalid = sum(not r['valid'] for r in rows) return dict(total=len(rows), supported=len(supported), unsupported=len(unsupported), correct=correct, supported_correct_rate=correct / len(supported) if supported else None, false_accepts=false_accepts, false_accept_rate=false_accepts / len(unsupported) if unsupported else None, invalid=invalid, supported_nonexecution=sum(r['final_status'] not in ('executed', 'fallback_executed') for r in supported)) def run(repo): repo = Path(repo).resolve() # Resolve only the explicitly selected checkout, never an installed/dirty core. sys.path.insert(0, str(repo / 'kvrm-core/src')) import kvrm_core if not Path(kvrm_core.__file__).resolve().is_relative_to(repo): raise ValueError('Imported core is not in selected checkout') from kvrm_core.registry import load_registry from kvrm_core.runtime import KVRMRuntime from kvrm_core.validation import DeterministicValidator from kvrm_core.support import evaluate_support_spec from kvrm_core.types import DecisionInput, DecisionCandidate, ExecutionResult, ExecutionStatus manifest, all_rows, configurations = {}, [], {} def read(path, jsonl=False): path = repo / path raw = path.read_bytes() manifest[str(path.relative_to(repo))] = hashlib.sha256(raw).hexdigest() return [json.loads(l) for l in raw.splitlines() if l.strip()] if jsonl else json.loads(raw) class Sink: def __init__(self): self.calls = [] def execute(self, candidate): self.calls.append(candidate.action_id) return ExecutionResult(status=ExecutionStatus.SUCCESS, action_id=candidate.action_id, output={'simulation_only': True}) class RecordedProposal: def __init__(self, unsupported): self.row, self.unsupported = None, unsupported def select(self, inp): if self.unsupported and self.unsupported(inp.features): return [] r = self.row if not r['valid'] or r['abstained'] or r['selected_action_id'] is None: return [] # Fixed 1.0 avoids tuning a threshold on evaluation labels. This arm # isolates deterministic validation, not confidence calibration. return [DecisionCandidate(action_id=r['selected_action_id'], confidence=1., source='recorded_model')] class ConstraintScan: def __init__(self, registry, unsupported): self.registry, self.unsupported = registry, unsupported def select(self, inp): if self.unsupported and self.unsupported(inp.features): return [] eligible = [a for a in self.registry.actions if 'fallback' not in a.tags and evaluate_support_spec(a.support_spec, inp.features)[0]] # No label access, no training, no tie-breaking chosen on test cases. return [DecisionCandidate(action_id=eligible[0].action_id, confidence=1., source='constraint_scan')] if len(eligible) == 1 else [] for domain, dirname in DOMAINS.items(): directory = Path('kvrm-demos') / dirname sys.path.insert(0, str(repo / directory)) module = importlib.import_module(dirname.replace('-', '_')) if not Path(module.__file__).resolve().is_relative_to(repo): raise ValueError('Wrong demo checkout') cases = read(directory / 'data/cases.jsonl', True) read(directory / 'data/train_cases.jsonl', True) registry_data = read(directory / 'data/registry.json') registry = load_registry(repo / directory / 'data/registry.json') hybrid = module.build_hybrid_selector(repo / directory / 'data/train_cases.jsonl', learned_model_path=None) fallback = hybrid.fallback_action_id arms = {'kvrm_hybrid_no_learned_model': hybrid, 'constraint_scan_shared_checks': ConstraintScan(registry, hybrid.unsupported_predicate)} recorded = {} for model, slug in MODELS.items(): base = Path('baselines/qwen-baseline/outputs/ollama/live_canonical') / domain / slug / 'abstain_json' recorded[model] = validate_alignment(cases, read(base / 'per_case_results.jsonl', True)) configurations[f'{domain}/{model}'] = read(base / 'config.json') arms[model + '_plus_validation'] = RecordedProposal(hybrid.unsupported_predicate) runtimes = {} for name, selector in arms.items(): sink = Sink() runtimes[name] = (KVRMRuntime(registry, selector, DeterministicValidator(registry), sink, threshold=.60, fallback_action_id=fallback), sink) for case in cases: for model in MODELS: r = recorded[model][case['case_id']] correct = bool(r['valid'] and r['selected_action_id'] == case['expected_action_id'] and r['final_status'] in ('executed', 'fallback_executed')) if case['supported'] else False if case['supported'] and correct != r['correct']: raise ValueError('Recorded correctness inconsistency') all_rows.append(dict(domain=domain, arm=model + '_recorded', case_id=case['case_id'], supported=case['supported'], expected_action_id=case['expected_action_id'], selected_action_id=r['selected_action_id'], final_status=r['final_status'], correct=correct, valid=r['valid'])) arms[model + '_plus_validation'].row = r for arm, (runtime, sink) in runtimes.items(): sink.calls.clear() result = runtime.decide_and_execute(DecisionInput(case_id=case['case_id'], features=case['input_features'])) action = sink.calls[0] if sink.calls else None all_rows.append(dict(domain=domain, arm=arm, case_id=case['case_id'], supported=case['supported'], expected_action_id=case['expected_action_id'], selected_action_id=action, final_status=result.final_status.value, correct=bool(case['supported'] and action == case['expected_action_id']), valid=True)) # Source identities allow reproducing this replay without trusting averages. for path in sorted((repo / 'kvrm-core/src').rglob('*.py')): manifest[str(path.relative_to(repo))] = hashlib.sha256(path.read_bytes()).hexdigest() for dirname in DOMAINS.values(): for path in sorted((repo / 'kvrm-demos' / dirname).rglob('*.py')): manifest[str(path.relative_to(repo))] = hashlib.sha256(path.read_bytes()).hexdigest() names = sorted({r['arm'] for r in all_rows}) summaries = {name: score([r for r in all_rows if r['arm'] == name]) for name in names} by_domain = {domain: {name: score([r for r in all_rows if r['arm'] == name and r['domain'] == domain]) for name in names} for domain in DOMAINS} return dict(protocol='retrospective-replay-v1', expected_public_revision=PUBLIC_REVISION, production_writes=0, live_model_calls=0, model_source='Archived April 2026 Ollama responses', replay_date='2026-10-02', domains=list(DOMAINS), summary=summaries, by_domain=by_domain, limitations=[ 'Researcher-authored synthetic structured cases; not customer data or an independent benchmark.', 'Archived models were prompted with feature schemas, action descriptions and examples, not the full executable support specifications available to KVRM and constraint controls.', 'Case IDs, labels and support flags match; historical requests lack input hashes, so unchanged original input payloads cannot be independently verified.', 'Model-only arms replay April responses; KVRM and validation/control arms run now against the pinned public snapshot, without the optional learned model.', 'Validation arms share domain unsupported-context checks, KVRM deterministic runtime and fallback; not a benchmark of OPA, Cedar or an independent policy implementation.', 'Constraint scan shares domain unsupported checks and support evaluator; it is a strong no-learning control, not a third-party product.', 'False accept means final_status=executed on an unsupported-labelled case. Validated fallback is a separate outcome, not counted as a false accept.', 'No Jev calls, latency comparison, cost comparison, universal safety claim or production effects.', ], configurations=configurations, source_sha256=manifest, rows=all_rows) if __name__ == '__main__': p = argparse.ArgumentParser(description=__doc__) p.add_argument('--repo', type=Path, required=True) p.add_argument('--output', type=Path) args = p.parse_args() result = run(args.repo) data = json.dumps(result, indent=2) + '\n' if args.output: args.output.write_text(data) else: print(data)