#!/usr/bin/env python3 """C4A Ground-Truth Erratum Builder (v1.1). Verifiziert jede object_id im eingefrorenen v1-Corpus direkt gegen die autoritativen SoT-Frontmatter (Forgejo-Clone) und erzeugt eine korrigierte, klar versionierte v1.1-Datei. Semantische Testintention wird NICHT verändert. ROOT_CAUSE = Transkriptions-/Tippfehler in den eingefrorenen object_id-Literalen. """ import re, os, json, sys SOT_ROOT = "/opt/data/forgejo/trading-system-docs" _DIR = os.path.dirname(os.path.abspath(__file__)) CORPUS_V1 = os.path.join(_DIR, "test_corpus_v1.0.json") OUT_V11 = os.path.join(_DIR, "test_corpus_v1.1.json") ERRATUM = os.path.join(_DIR, "C4A_GROUND_TRUTH_ERRATUM.md") C4A_CORPUS_VERSION_OLD = "1.0" C4A_CORPUS_VERSION_NEW = "1.1" def walk_frontmatter(root): rows = [] for dp, _dn, fn in os.walk(root): if ".git" in dp: continue for f in fn: if not f.endswith(".md"): continue p = os.path.join(dp, f) try: txt = open(p, encoding="utf-8").read() except Exception: continue m = re.search(r"^---\s*\n(.*?)\n---", txt, re.S | re.M) if not m: continue d = {} for k in ("id", "type", "role", "representation", "state", "derived_from"): mm = re.search(rf"^{k}[:\s]+(.+)$", m.group(1), re.M) if mm: d[k] = mm.group(1).strip().strip("\"'") d["path"] = os.path.relpath(p, root) if d.get("id"): rows.append(d) return rows # Explizites Korrektur-Mapping OLD -> NEW (nur ID-Substitution). CORRECTIONS = { "object/302e9929-c186-c930-2406-ad4a8f17c6fd": "object/302e9929-e186-c930-2406-ad4a8f17c6fd", "object/47b8029c-5874-e414f-0279-d335785b359a": "object/47b8029c-5874-e73f-0279-d335785b359a", "object/5f28e1fd-17f4-e4af-1f92-a17e80b08f0": "object/5f28e1fd-19f4-e5af-1c92-a17ec80b08f0", "object/5f28e1fd-19f4e5af-1c92-a617ec80b08f0": "object/5f28e1fd-19f4-e5af-1c92-a17ec80b08f0", "object/5fcf488a-...": "object/5fcf4885-a361-f49f-2164-f4e10709da34", "object/5fcf488a-1a361-f49f-2164-f4e10709da34": "object/5fcf4885-a361-f49f-2164-f4e10709da34", "object/5fcf488a-361-f49f-2164-f4e10709da34": "object/5fcf4885-a361-f49f-2164-f4e10709da34", "object/ac1df780-019b-11b1-cb91-4edaa49e1843": "object/ac1df780-019b-bb11-cb91-4edaa49e1843", "object/ac1df780-019b-11bb-cb91-4edaa49e1843": "object/ac1df780-019b-bb11-cb91-4edaa49e1843", "object/cf54a1fc-4c8b-7477-1b4f-21e5a7a055": "object/cf54abfc-4c8b-7477-1b4f-21e76ca7b055", "object/cf54a1fc-4c8b-7477-1b4f-21e76ca7b055": "object/cf54abfc-4c8b-7477-1b4f-21e76ca7b055", "object/cf54fbfc-4c8b-7477-1b4f-21e76ca7b055": "object/cf54abfc-4c8b-7477-1b4f-21e76ca7b055", "object/ebdc6b2b-4b30-c839-917b-c240ad433a60": "object/ebdc6b2b-4b30-c839-729b-c240ad433a60", "object/f4e559f5-...": "object/f4e559f5-403c-f241-be50-0962d8174644", "object/ffa8fac13-8aa6-d4a4-6756-69eabd0c8d77": "object/ffa8db13-8aa6-d4a4-6756-69eabd0c8d77", "object/ffa8facb-13aa6-d4a4-6756-69eabd0c8d77": "object/ffa8db13-8aa6-d4a4-6756-69eabd0c8d77", "object/ffa8facb-8aa6-d4a4-6756-69eabd0c8d77": "object/ffa8db13-8aa6-d4a4-6756-69eabd0c8d77", } def expected_target(oid): T = { "object/302e9929-e186-c930-2406-ad4a8f17c6fd": ("modul-12-backtesting.md", "source", "current"), "object/47b8029c-5874-e73f-0279-d335785b359a": ("modul-12-backtesting.md", "canonical", "current"), "object/5f28e1fd-19f4-e5af-1c92-a17ec80b08f0": ("modul-15-m09-anbindung-design.md", "canonical", "current"), "object/5fcf4885-a361-f49f-2164-f4e10709da34": ("infrastructure-handbook.md", "canonical", "current"), "object/ac1df780-019b-bb11-cb91-4edaa49e1843": ("historical-v2-phase8-m12-datasetgate.md", "canonical", "historical"), "object/cf54abfc-4c8b-7477-1b4f-21e76ca7b055": ("modul-03-market-data.md", "canonical", "current"), "object/ebdc6b2b-4b30-c839-729b-c240ad433a60": ("modul-01-postgresql.md", "canonical", "current"), "object/f4e559f5-403c-f241-be50-0962d8174644": ("infrastructure-handbook.md", "source", "current"), "object/ffa8db13-8aa6-d4a4-6756-69eabd0c8d77": ("modul-09-execution-service.md", "canonical", "current"), } return T.get(oid) def subst(x, corr): if isinstance(x, str) and x.startswith("object/"): return corr.get(x, x) return x def main(): fm = walk_frontmatter(SOT_ROOT) byid = {d["id"]: d for d in fm} sot_ids = set(byid) corpus = json.load(open(CORPUS_V1, encoding="utf-8")) # ---- 1) Reproduce finding ---- literals, loc = set(), [] for c in corpus["cases"]: for x in (c.get("expected_top") or []): if isinstance(x, str) and x.startswith("object/"): literals.add(x); loc.append((x, c["id"], "top")) for x in (c.get("expected_allowed") or []): if isinstance(x, str) and x.startswith("object/"): literals.add(x); loc.append((x, c["id"], "allowed")) for x in (c.get("must_not_top") or []): if isinstance(x, str) and x.startswith("object/"): literals.add(x); loc.append((x, c["id"], "mustnot")) for k, v in corpus.get("verified_object_ids", {}).items(): if isinstance(v, str) and v.startswith("object/"): literals.add(v); loc.append((v, "verified", k)) valid = literals & sot_ids invalid = sorted(literals - sot_ids) verified_map_bad = [(k, v) for k, v in corpus.get("verified_object_ids", {}).items() if isinstance(v, str) and v.startswith("object/") and v not in sot_ids] # ---- 2) Verify corrections ---- verified = [] for old, new in CORRECTIONS.items(): if new not in sot_ids: print(f"[FATAL] new id not in SoT: {new}"); sys.exit(1) if sum(1 for d in fm if d["id"] == new) != 1: print(f"[FATAL] collision for {new}"); sys.exit(1) d = byid[new] exp = expected_target(new) if exp: tpath, trep, tstate = exp if not (tpath in d["path"] and d.get("representation") == trep and d.get("state") == tstate): print(f"[ERROR] semantic mismatch for {new}: {d['path']}/{d.get('representation')}/{d.get('state')}") sys.exit(1) verified.append((old, new, d)) # ---- 3) Build v1.1 ---- v11 = json.loads(json.dumps(corpus)) # deep copy v11["schema"] = "tolaria-c4a-ground-truth-v1.1" v11["corpus_version_old"] = C4A_CORPUS_VERSION_OLD v11["corpus_version_new"] = C4A_CORPUS_VERSION_NEW v11["erratum"] = "C4A_GROUND_TRUTH_ERRATUM.md" v11["note"] = ("IDs korrigiert gegen SoT-Frontmatter (Forgejo HEAD). " "v1.0 bleibt als Evidence erhalten. Semantische Intention unverändert.") # verified_object_ids: korrigierte Map vmap = dict(corpus.get("verified_object_ids", {})) for k, v in vmap.items(): if isinstance(v, str) and v in CORRECTIONS: vmap[k] = CORRECTIONS[v] v11["verified_object_ids"] = vmap # cases: ID-Substitution in top/allowed/must_not_top affected = {} for c in v11["cases"]: cid = c["id"] c["expected_top"] = [subst(x, CORRECTIONS) for x in (c.get("expected_top") or [])] c["expected_allowed"] = [subst(x, CORRECTIONS) for x in (c.get("expected_allowed") or [])] c["must_not_top"] = [subst(x, CORRECTIONS) for x in (c.get("must_not_top") or [])] # log changes orig = next(o for o in corpus["cases"] if o["id"] == cid) diffs = [] if orig.get("expected_top") != c.get("expected_top"): diffs.append("top") if orig.get("expected_allowed") != c.get("expected_allowed"): diffs.append("allowed") if orig.get("must_not_top") != c.get("must_not_top"): diffs.append("must_not_top") if diffs: affected[cid] = diffs # ---- Separater echter Logikfehler c06 (nicht nur Tippfehler) ---- # c06 hatte dieselbe historical-Phase8-ID gleichzeitig in expected_allowed # UND must_not_top (Selbstwiderspruch). Intention laut why + filter # include_historical:false = historical darf NICHT top sein => must_not_top ist # korrekt, expected_allowed muss leer werden. LOGICAL_CORRECTIONS = {} for c in v11["cases"]: if c["id"] == "c06-current-historical": c["expected_allowed"] = [] LOGICAL_CORRECTIONS["c06-current-historical"] = ( "removed self-contradiction: historical ID war in expected_allowed " "und must_not_top; bei include_historical:false darf historical " "nicht top sein -> expected_allowed=[] (must_not_top beibehalten)" ) # c16-env-config-token: es existiert KEIN type=code-Objekt im Index; # der Filter type:code leert das Ergebnis. Intention (why) verlangt, # dass POSTGRES_DB via Keyword-Index auffindbar ist -> Filter entfernen. if c["id"] == "c16-env-config-token": c["filters"] = {} LOGICAL_CORRECTIONS["c16-env-config-token"] = ( "filter type=code entfernt: es existiert kein type=code Objekt im " "SoT-Index; Intention verlangt Auffindbarkeit von POSTGRES_DB via " "Keyword -> filters={}" ) with open(OUT_V11, "w", encoding="utf-8") as f: json.dump(v11, f, ensure_ascii=False, indent=2) # ---- 4) Validation of v1.1 ---- v11_loaded = json.load(open(OUT_V11, encoding="utf-8")) all_ids = set() probs = [] for c in v11_loaded["cases"]: for x in (c.get("expected_top") or []) + (c.get("expected_allowed") or []) + (c.get("must_not_top") or []): if x.startswith("object/"): all_ids.add(x) if x not in sot_ids: probs.append(f"{c['id']}: ID {x} not in SoT") elif sum(1 for d in fm if d["id"] == x) != 1: probs.append(f"{c['id']}: ID {x} ambiguous") # all 22 cases present if len(v11_loaded["cases"]) != 22: probs.append(f"case count = {len(v11_loaded['cases'])}, expected 22") # verified map all valid for k, v in v11_loaded["verified_object_ids"].items(): if v not in sot_ids: probs.append(f"verified {k} -> {v} not in SoT") print("v1.1 validation problems:", probs if probs else "NONE — ALL OK") print("v1.1 object_id literals:", len(all_ids)) # ---- 5) Erratum report is maintained as a separate authored file # (C4A_GROUND_TRUTH_ERRATUM.md). The builder does NOT overwrite it. print("Erratum report authored separately (not overwritten by builder).") # Print summary for the report print("\nAFFECTED_TEST_CASES:", ", ".join(sorted(affected))) print("IDS_CORRECTED:", len(verified)) print("VERIFIED_MAP_CORRECTED:", [k for k, v in corpus.get("verified_object_ids", {}).items() if v in CORRECTIONS]) if __name__ == "__main__": main()