#!/usr/bin/env python3 """Evaluate TELEN's immutable version-index contract on FiscalQA Pro. This is a retrieval-provenance evaluation, not an answer-generation score. It uses the released expert-reviewed R3 questions and their gold article-version identifiers, and deliberately keeps deterministic temporal filtering separate from dense semantic retrieval. """ from __future__ import annotations import argparse import ast import hashlib import json import platform import re import subprocess import sys import unicodedata from collections import defaultdict from datetime import datetime, timezone from pathlib import Path from typing import Iterable import numpy as np import pandas as pd import torch REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "src")) from scripts.run_static_benchmark import load_or_encode, resolve_device # noqa: E402 from telen.static_benchmark import sha256_file # noqa: E402 UPSTREAM_REPOSITORY = "https://github.com/rosecymbler/fiscal-fr-bench.git" UPSTREAM_COMMIT = "6ca3807d46926f6f5b355d2d435b4baa05300bdf" DATA_LICENSE = "CC-BY-4.0" CODE_LICENSE = "MIT" MODEL_NAME = "BAAI/bge-m3" MODEL_REVISION = "5617a9f61b028005a4858fdac845db406aefb181" EXPECTED_SCORED_QUESTIONS = 209 EXPECTED_EXCLUDED_QUESTIONS = 12 EXPECTED_R3_QUESTIONS = 221 BOOTSTRAP_SEED = 20260829 CGI_PRINCIPAL_CODE_ID = "LEGITEXT000006069577" def _sha256_text(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() def _git_head(path: Path) -> str: completed = subprocess.run( ["git", "-C", str(path), "rev-parse", "HEAD"], check=True, capture_output=True, text=True, ) return completed.stdout.strip() def _read_id_file(path: Path) -> set[str]: return { line.strip() for line in path.read_text(encoding="utf-8").splitlines() if line.strip() and not line.lstrip().startswith("#") } def _gold_dict(value: object) -> dict: if isinstance(value, dict): return value if isinstance(value, str): parsed = ast.literal_eval(value) if isinstance(parsed, dict): return parsed raise ValueError("FiscalQA gold must be a dictionary or dictionary literal") def _load_scored_questions(root: Path) -> list[dict]: benchmark = root / "data/benchmark" raw = json.loads((benchmark / "questions.json").read_text(encoding="utf-8")) # Never expose or propagate the upstream contamination canary. r3 = [ row for row in raw if row.get("qid") != "_CANARY_" and row.get("regime") == "R3" ] scored = _read_id_file(benchmark / "scored209_qids.txt") excluded = _read_id_file(benchmark / "excluded_qids_scope.txt") observed_ids = {str(row["qid"]) for row in r3} if len(r3) != EXPECTED_R3_QUESTIONS: raise ValueError(f"expected {EXPECTED_R3_QUESTIONS} R3 questions") if len(scored) != EXPECTED_SCORED_QUESTIONS: raise ValueError(f"expected {EXPECTED_SCORED_QUESTIONS} scored qids") if len(excluded) != EXPECTED_EXCLUDED_QUESTIONS: raise ValueError(f"expected {EXPECTED_EXCLUDED_QUESTIONS} excluded qids") if scored != observed_ids.difference(excluded): raise ValueError("scored qids are not exactly R3 minus scope exclusions") result = [] for row in r3: if row["qid"] not in scored: continue gold = _gold_dict(row["gold"]) required = {"version_id", "article_cid", "canonical_answer", "code"} if not required.issubset(gold): raise ValueError(f"question {row['qid']} has incomplete gold metadata") result.append( { "qid": str(row["qid"]), "prompt": str(row["prompt"]), "date_anchor": pd.Timestamp(row["date_anchor"]), "gold_version_id": str(gold["version_id"]), "gold_cid": str(gold["article_cid"]), "gold_code": str(gold["code"]), } ) if len(result) != EXPECTED_SCORED_QUESTIONS: raise ValueError("scored question load is incomplete") return result def _load_corpus(path: Path) -> pd.DataFrame: corpus = pd.read_parquet(path).copy() required = { "id", "cid", "num", "texte_clean", "date_debut", "date_fin", "code_id", } missing = required.difference(corpus.columns) if missing: raise ValueError(f"FiscalQA corpus is missing columns: {sorted(missing)}") if corpus["id"].duplicated().any(): raise ValueError("FiscalQA corpus has duplicate version ids") if corpus[["id", "cid", "num", "date_debut", "date_fin"]].isna().any().any(): raise ValueError("FiscalQA corpus has null version identity or interval fields") corpus["id"] = corpus["id"].astype(str) corpus["cid"] = corpus["cid"].astype(str) corpus["num"] = corpus["num"].astype(str) corpus["code_id"] = corpus["code_id"].astype(str) corpus["texte_clean"] = corpus["texte_clean"].fillna("").astype(str) # Légifrance contains MODIFIE_MORT_NE/ABROGE records whose end is not after # their start. Preserve them in the faithful all-version corpus but # quarantine them from every snapshot rather than inventing a validity span. corpus["snapshot_eligible"] = corpus["date_fin"] > corpus["date_debut"] corpus = corpus.sort_values("id", kind="mergesort").reset_index(drop=True) return corpus def _resolved_snapshot_mask(corpus: pd.DataFrame, as_of: pd.Timestamp) -> np.ndarray: """Choose the latest-starting valid version for every article identity. Three FiscalQA scored dates overlap two upstream intervals for their gold article. Choosing the latest valid start is deterministic and recovers the upstream gold without reading a question label. """ active = ( corpus["snapshot_eligible"] & (corpus["date_debut"] <= as_of) & (as_of < corpus["date_fin"]) ) eligible = corpus.loc[active, ["cid", "date_debut", "id"]].copy() if eligible.empty: return np.zeros(len(corpus), dtype=bool) eligible["position"] = eligible.index selected = ( eligible.sort_values( ["cid", "date_debut", "id"], ascending=[True, True, True], kind="mergesort", ) .groupby("cid", sort=False, as_index=False) .tail(1)["position"] .to_numpy(dtype=np.int64) ) mask = np.zeros(len(corpus), dtype=bool) mask[selected] = True return mask def _interval_mask(corpus: pd.DataFrame, as_of: pd.Timestamp) -> np.ndarray: return np.asarray( corpus["snapshot_eligible"] & (corpus["date_debut"] <= as_of) & (as_of < corpus["date_fin"]), dtype=bool, ) def _normalize_tokens(value: str) -> tuple[str, ...]: normalized = unicodedata.normalize("NFKC", value).casefold() return tuple(re.findall(r"[0-9]+|[^\W\d_]+", normalized, flags=re.UNICODE)) def _cited_article_numbers( prompt: str, known_numbers: Iterable[str], *, max_number_tokens: int = 6, ) -> tuple[str, ...]: """Extract explicit ``article``/``art.`` citations without gold metadata.""" lookup: dict[tuple[str, ...], set[str]] = defaultdict(set) for number in known_numbers: tokens = _normalize_tokens(number) if tokens: lookup[tokens].add(number) tokens = _normalize_tokens(prompt) candidates: set[str] = set() for position, token in enumerate(tokens): if token not in {"article", "articles", "art"}: continue longest = 0 matches: set[str] = set() for width in range(1, max_number_tokens + 1): chunk = tokens[position + 1 : position + 1 + width] if len(chunk) != width: break observed = lookup.get(chunk, set()) if observed: longest = width matches = set(observed) if longest: candidates.update(matches) return tuple(sorted(candidates)) def _rank_observation( scores: np.ndarray, candidate_mask: np.ndarray, version_ids: np.ndarray, cids: np.ndarray, *, gold_version_id: str, gold_cid: str, ) -> dict: positions = np.flatnonzero(candidate_mask) if not len(positions): return { "gold_rank": None, "gold_article_rank": None, "top1_version_id": None, "top1_cid": None, "ndcg_at_10": 0.0, "mrr": 0.0, "hit_at_1": 0.0, "hit_at_5": 0.0, "hit_at_10": 0.0, "article_hit_at_1": 0.0, "article_hit_at_5": 0.0, "article_hit_at_10": 0.0, } order_local = np.lexsort( (version_ids[positions], -np.asarray(scores[positions], dtype=np.float64)) ) ranking = positions[order_local] ranked_versions = version_ids[ranking] ranked_cids = cids[ranking] gold_positions = np.flatnonzero(ranked_versions == gold_version_id) article_positions = np.flatnonzero(ranked_cids == gold_cid) gold_rank = int(gold_positions[0]) + 1 if len(gold_positions) else None article_rank = int(article_positions[0]) + 1 if len(article_positions) else None return { "gold_rank": gold_rank, "gold_article_rank": article_rank, "top1_version_id": str(ranked_versions[0]), "top1_cid": str(ranked_cids[0]), "ndcg_at_10": ( float(1.0 / np.log2(gold_rank + 1)) if gold_rank is not None and gold_rank <= 10 else 0.0 ), "mrr": float(1.0 / gold_rank) if gold_rank is not None else 0.0, "hit_at_1": float(gold_rank == 1), "hit_at_5": float(gold_rank is not None and gold_rank <= 5), "hit_at_10": float(gold_rank is not None and gold_rank <= 10), "article_hit_at_1": float(article_rank == 1), "article_hit_at_5": float(article_rank is not None and article_rank <= 5), "article_hit_at_10": float(article_rank is not None and article_rank <= 10), } METRIC_FIELDS = ( "ndcg_at_10", "mrr", "hit_at_1", "hit_at_5", "hit_at_10", "article_hit_at_1", "article_hit_at_5", "article_hit_at_10", ) def _aggregate(rows: list[dict]) -> dict: by_cid: dict[str, list[dict]] = defaultdict(list) for row in rows: by_cid[row["gold_cid"]].append(row) aggregate = {field: float(np.mean([row[field] for row in rows])) for field in METRIC_FIELDS} aggregate.update( { "query_count": len(rows), "article_cluster_count": len(by_cid), "article_macro_ndcg_at_10": float( np.mean( [ np.mean([row["ndcg_at_10"] for row in cluster]) for cluster in by_cid.values() ] ) ), "gold_missing_from_candidates": int( sum(row["gold_rank"] is None for row in rows) ), "top1_temporal_misgrounding": float( np.mean( [ row["top1_cid"] == row["gold_cid"] and row["top1_version_id"] != row["gold_version_id"] for row in rows ] ) ), } ) return aggregate def _clustered_difference( candidate_rows: list[dict], baseline_rows: list[dict], *, field: str = "ndcg_at_10", samples: int = 10_000, ) -> dict: if [row["qid"] for row in candidate_rows] != [row["qid"] for row in baseline_rows]: raise ValueError("paired rows are not aligned") grouped: dict[str, list[float]] = defaultdict(list) for candidate, baseline in zip(candidate_rows, baseline_rows, strict=True): if candidate["gold_cid"] != baseline["gold_cid"]: raise ValueError("paired rows have different article clusters") grouped[candidate["gold_cid"]].append(candidate[field] - baseline[field]) cluster_values = np.asarray( [np.mean(grouped[cid]) for cid in sorted(grouped)], dtype=np.float64 ) rng = np.random.default_rng(BOOTSTRAP_SEED) draws = rng.choice( cluster_values, size=(samples, len(cluster_values)), replace=True, ).mean(axis=1) return { "metric": field, "estimand": "paired article-macro candidate minus baseline", "article_cluster_count": int(len(cluster_values)), "observed_difference": float(cluster_values.mean()), "bootstrap_samples": samples, "bootstrap_seed": BOOTSTRAP_SEED, "ci_95": [float(np.quantile(draws, 0.025)), float(np.quantile(draws, 0.975))], } def run(args: argparse.Namespace) -> dict: upstream_head = _git_head(args.upstream_root) if upstream_head != UPSTREAM_COMMIT: raise ValueError( f"FiscalQA upstream commit drifted: expected {UPSTREAM_COMMIT}, got {upstream_head}" ) corpus_path = args.upstream_root / "data/articles_cgi_versioned.parquet" question_path = args.upstream_root / "data/benchmark/questions.json" scored_path = args.upstream_root / "data/benchmark/scored209_qids.txt" excluded_path = args.upstream_root / "data/benchmark/excluded_qids_scope.txt" license_path = args.upstream_root / "LICENSE-DATA" corpus = _load_corpus(corpus_path) questions = _load_scored_questions(args.upstream_root) version_ids = corpus["id"].to_numpy(dtype=str) cids = corpus["cid"].to_numpy(dtype=str) numbers = corpus["num"].to_numpy(dtype=str) code_ids = corpus["code_id"].to_numpy(dtype=str) id_to_position = {version_id: position for position, version_id in enumerate(version_ids)} if len(id_to_position) != len(version_ids): raise ValueError("corpus version ids are not unique after normalization") for question in questions: position = id_to_position.get(question["gold_version_id"]) if position is None: raise ValueError(f"missing gold version {question['gold_version_id']}") if cids[position] != question["gold_cid"]: raise ValueError(f"gold CID mismatch for {question['qid']}") if not _interval_mask(corpus, question["date_anchor"])[position]: raise ValueError(f"gold version is not date-valid for {question['qid']}") if {question["gold_code"] for question in questions} != {"CGI"}: raise ValueError("the frozen R3 scored slice is no longer CGI-principal only") version_texts = tuple( f"Article {row.num}\n{row.texte_clean}" for row in corpus.itertuples(index=False) ) query_ids = tuple(question["qid"] for question in questions) query_texts = tuple(question["prompt"] for question in questions) device = resolve_device(args.device) version_embeddings, version_cache = load_or_encode( version_texts, tuple(version_ids), role="fiscalqa-pro-article-versions", cache_dir=args.cache_dir, model_name=args.model_name, revision=args.model_revision, device=device, max_length=args.max_length, batch_size=args.corpus_batch_size, ) query_embeddings, query_cache = load_or_encode( query_texts, query_ids, role="fiscalqa-pro-r3-scored-queries", cache_dir=args.cache_dir, model_name=args.model_name, revision=args.model_revision, device=device, max_length=args.max_length, batch_size=args.query_batch_size, ) scores = np.asarray(query_embeddings, dtype=np.float32) @ np.asarray( version_embeddings, dtype=np.float32 ).T current_mask = _resolved_snapshot_mask(corpus, pd.Timestamp(args.current_snapshot)) known_numbers = tuple(sorted(set(numbers))) conditions: dict[str, list[dict]] = defaultdict(list) citation_audit = [] all_mask = np.ones(len(corpus), dtype=bool) for row_index, question in enumerate(questions): date = question["date_anchor"] interval_mask = _interval_mask(corpus, date) resolved_mask = _resolved_snapshot_mask(corpus, date) cited_numbers = _cited_article_numbers(question["prompt"], known_numbers) cited_mask = resolved_mask & np.isin(numbers, np.asarray(cited_numbers, dtype=str)) declared_cgi_mask = resolved_mask & (code_ids == CGI_PRINCIPAL_CODE_ID) cited_cgi_mask = declared_cgi_mask & np.isin( numbers, np.asarray(cited_numbers, dtype=str) ) citation_or_dense_cgi_mask = ( cited_cgi_mask if cited_numbers else declared_cgi_mask ) oracle_mask = resolved_mask & (cids == question["gold_cid"]) masks = { "all_versions_dense": all_mask, "current_snapshot_dense": current_mask, "date_interval_dense": interval_mask, "date_resolved_dense": resolved_mask, "explicit_citation_date_resolved_dense": cited_mask, "declared_r3_cgi_scope_date_resolved_dense": declared_cgi_mask, "citation_or_dense_cgi_date_resolved": citation_or_dense_cgi_mask, "oracle_cid_date_resolved_dense": oracle_mask, } citation_audit.append( { "qid": question["qid"], "cited_number_count": len(cited_numbers), "gold_number_detected": bool( numbers[id_to_position[question["gold_version_id"]]] in cited_numbers ), } ) for condition, mask in masks.items(): observation = _rank_observation( scores[row_index], mask, version_ids, cids, gold_version_id=question["gold_version_id"], gold_cid=question["gold_cid"], ) observation.update( { "qid": question["qid"], "gold_cid": question["gold_cid"], "gold_version_id": question["gold_version_id"], "date_anchor": date.date().isoformat(), } ) conditions[condition].append(observation) aggregates = {name: _aggregate(rows) for name, rows in conditions.items()} comparisons = { "date_resolved_vs_current": _clustered_difference( conditions["date_resolved_dense"], conditions["current_snapshot_dense"], ), "date_resolved_vs_all_versions": _clustered_difference( conditions["date_resolved_dense"], conditions["all_versions_dense"], ), "citation_resolver_vs_date_resolved": _clustered_difference( conditions["explicit_citation_date_resolved_dense"], conditions["date_resolved_dense"], ), "declared_cgi_scope_vs_date_resolved": _clustered_difference( conditions["declared_r3_cgi_scope_date_resolved_dense"], conditions["date_resolved_dense"], ), "citation_or_dense_cgi_vs_declared_cgi_scope": _clustered_difference( conditions["citation_or_dense_cgi_date_resolved"], conditions["declared_r3_cgi_scope_date_resolved_dense"], ), "citation_or_dense_cgi_vs_date_resolved": _clustered_difference( conditions["citation_or_dense_cgi_date_resolved"], conditions["date_resolved_dense"], ), "citation_or_dense_cgi_vs_current": _clustered_difference( conditions["citation_or_dense_cgi_date_resolved"], conditions["current_snapshot_dense"], ), "citation_or_dense_cgi_vs_all_versions": _clustered_difference( conditions["citation_or_dense_cgi_date_resolved"], conditions["all_versions_dense"], ), } receipt = { "schema_version": 1, "artifact_type": "telen_fiscalqa_external_retrieval_validation", "generated_at": datetime.now(timezone.utc).isoformat(), "claim_scope": { "retrieval_provenance_evaluated": True, "answer_generation_evaluated": False, "learned_graph_residual_evaluated": False, "vietnamese_external_validity_evaluated": False, "interpretation": ( "External expert-reviewed French temporal retrieval diagnostic; " "it validates version resolution but cannot replace Vietnamese gold qrels." ), }, "upstream": { "repository": UPSTREAM_REPOSITORY, "commit": upstream_head, "data_license": DATA_LICENSE, "code_license": CODE_LICENSE, "contamination_canary_policy": "header skipped and never copied to outputs", "files": { "corpus": {"path": str(corpus_path), "sha256": sha256_file(corpus_path)}, "questions": {"path": str(question_path), "sha256": sha256_file(question_path)}, "scored_qids": {"path": str(scored_path), "sha256": sha256_file(scored_path)}, "excluded_qids": {"path": str(excluded_path), "sha256": sha256_file(excluded_path)}, "data_license": {"path": str(license_path), "sha256": sha256_file(license_path)}, }, }, "dataset": { "corpus_version_count": len(corpus), "article_identity_count": int(corpus["cid"].nunique()), "snapshot_ineligible_interval_count": int( (~corpus["snapshot_eligible"]).sum() ), "scored_question_count": len(questions), "scored_article_cluster_count": len({q["gold_cid"] for q in questions}), "date_range": [ min(q["date_anchor"] for q in questions).date().isoformat(), max(q["date_anchor"] for q in questions).date().isoformat(), ], "gold_interval_checks_passed": len(questions), "explicit_citation_gold_number_coverage": float( np.mean([row["gold_number_detected"] for row in citation_audit]) ), }, "model": { "name": args.model_name, "revision": args.model_revision, "device": str(device), "max_length": args.max_length, "version_embedding_cache": str(version_cache), "version_embedding_cache_sha256": sha256_file(version_cache), "query_embedding_cache": str(query_cache), "query_embedding_cache_sha256": sha256_file(query_cache), "version_text_order_sha256": _sha256_text("\n".join(version_ids)), "query_order_sha256": _sha256_text("\n".join(query_ids)), }, "protocol": { "current_snapshot": args.current_snapshot, "declared_r3_code_scope": { "label": "CGI", "code_id": CGI_PRINCIPAL_CODE_ID, "source": "upstream R3 benchmark definition", }, "validity_interval": "date_debut <= date_anchor < date_fin", "overlap_resolution": "latest date_debut, then stable version id", "ranking_ties": "stable version id ascending", "cluster_unit": "gold article CID", "bootstrap_samples": 10_000, "bootstrap_seed": BOOTSTRAP_SEED, "analysis_status": "external_post_hoc_diagnostic_not_confirmatory", "protocol_amendment": ( "Declared-CGI-scope and citation-or-dense baselines were added " "after the first date-resolved result was visible, but before " "testing any learned external residual; they are reported as " "post-hoc transparent strong baselines." ), }, "results": aggregates, "comparisons": comparisons, "environment": { "python": platform.python_version(), "numpy": np.__version__, "pandas": pd.__version__, "torch": torch.__version__, "cuda_available": torch.cuda.is_available(), }, "limitations": [ "Retrieval provenance is not FiscalQA strict answer accuracy.", "Most R3 prompts explicitly cite the target article, so the citation resolver is a necessary strong rule baseline.", "The benchmark is French tax law and does not establish Vietnamese legal relevance.", "This run does not train or credit the learned local graph residual.", "Validity intervals are supplied by the benchmark; uncertainty in event extraction is not tested.", ], } args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text( json.dumps(receipt, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) return receipt def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument( "--upstream-root", type=Path, default=REPO_ROOT / "tmp/deep-research/fiscal-fr-bench", ) parser.add_argument( "--cache-dir", type=Path, default=REPO_ROOT / "tmp/telen/static-benchmarks/cache/fiscalqa", ) parser.add_argument( "--output", type=Path, default=REPO_ROOT / "research/results/telen_fiscalqa_external_validation_2026-08-29.json", ) parser.add_argument("--model-name", default=MODEL_NAME) parser.add_argument("--model-revision", default=MODEL_REVISION) parser.add_argument("--device", default="auto") parser.add_argument("--max-length", type=int, default=2048) parser.add_argument("--corpus-batch-size", type=int, default=8) parser.add_argument("--query-batch-size", type=int, default=32) parser.add_argument("--current-snapshot", default="2026-08-05") return parser.parse_args() def main() -> None: receipt = run(parse_args()) print(json.dumps(receipt, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()