feat(tht): sync-schema-linking projects F4 ledger into schema_linking.json

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

harness/tests/test_sync_schema_linking.py

@@ -0,0 +1,63 @@
+"""Tests for the F4 ledger -> schema_linking.json deterministic projection."""
+import json
+
+from tht.config import DatabaseConfig
+from tht.decisions import append_decision
+from tht.session.store import create_session, set_schema_linking, sync_schema_linking
+
+
+def _db():
+    return DatabaseConfig(
+        database="testdb", user="u", password="p",  # noqa: S106
+        **{"schema": "public"},
+    )
+
+
+def test_projection_from_ledger(tmp_path):
+    m = create_session("domanda X", _db(), tmp_path)
+    sid = m.id
+    sdir = tmp_path / sid
+    # seed a prior schema_linking with a join to prove joins are preserved
+    set_schema_linking(sid, {"question": "domanda X", "candidates": [], "joins": [
+        {"from": "dim_patient.cod_paz", "to": "fact_x.cod_paz"}], "excluded": []}, tmp_path)
+    append_decision(sdir, type="table_promoted", subject="dim_patient")
+    append_decision(sdir, type="column_promoted", subject="dim_patient.cod_paz")
+    append_decision(sdir, type="column_promoted", subject="dim_patient.nome")
+    append_decision(sdir, type="table_excluded", subject="fact_sost")
+
+    sync_schema_linking(sid, tmp_path)
+
+    data = json.loads((sdir / "schema_linking.json").read_text())
+    tabs = {(c["kind"], c["name"], c["decision"]) for c in data["candidates"]}
+    assert ("table", "dim_patient", "promoted") in tabs
+    assert ("column", "dim_patient.cod_paz", "promoted") in tabs
+    assert ("column", "dim_patient.nome", "promoted") in tabs
+    assert {"kind": "table", "name": "fact_sost"} in [
+        {"kind": e["kind"], "name": e["name"]} for e in data["excluded"]]
+    assert data["joins"], "existing joins must be preserved"
+
+
+def test_projection_last_decision_wins(tmp_path):
+    m = create_session("domanda Y", _db(), tmp_path)
+    sid = m.id
+    sdir = tmp_path / sid
+    append_decision(sdir, type="table_promoted", subject="dim_x")
+    append_decision(sdir, type="table_excluded", subject="dim_x")
+
+    sync_schema_linking(sid, tmp_path)
+
+    data = json.loads((sdir / "schema_linking.json").read_text())
+    assert data["candidates"] == []
+    assert [e["name"] for e in data["excluded"]] == ["dim_x"]
+
+
+def test_projection_defaults_question_from_manifest(tmp_path):
+    m = create_session("domanda Z", _db(), tmp_path)
+    sid = m.id
+
+    path = sync_schema_linking(sid, tmp_path)
+
+    data = json.loads(path.read_text())
+    assert data["question"] == "domanda Z"
+    assert data["candidates"] == []
+    assert data["excluded"] == []