feat(tht): promoted_columns_for helper (curated column set)

harness/tests/test_promoted_columns_for.py

@@ -0,0 +1,22 @@
+import json
+from tht.cli.sql_cmd import promoted_columns_for
+
+
+def test_promoted_columns_for(tmp_path):
+    sid = "sess1"
+    sdir = tmp_path / sid
+    sdir.mkdir(parents=True)
+    (sdir / "schema_linking.json").write_text(json.dumps({
+        "question": "q",
+        "candidates": [
+            {"kind": "table", "name": "dim_patient", "decision": "promoted"},
+            {"kind": "column", "name": "dim_patient.cod_paz", "decision": "promoted"},
+            {"kind": "column", "name": "dim_patient.nome", "decision": "excluded"},
+        ],
+        "joins": [], "excluded": [],
+    }))
+
+    class Cfg:
+        class paths:  # noqa: N801
+            sessions = tmp_path
+    assert promoted_columns_for(Cfg, sid) == {"dim_patient.cod_paz"}

harness/tht/cli/sql_cmd.py

@@ -54,6 +54,21 @@ def promoted_tables_for(cfg, session_id: str | None) -> set[str] | None:
     }
 
 
+def promoted_columns_for(cfg, session_id: str | None) -> set[str] | None:
+    if session_id is None:
+        return None
+    linking_path = cfg.paths.sessions / session_id / "schema_linking.json"
+    if not linking_path.exists():
+        return None
+    from tht.session.models import SchemaLinking
+
+    linking = SchemaLinking.model_validate(json.loads(linking_path.read_text()))
+    return {
+        c.name for c in linking.candidates
+        if c.kind == "column" and c.decision == "promoted"
+    }
+
+
 def validate_or_exit(cfg, sql: str, session_id: str | None):
     """Validazione statica; stampa errori/warning. Exit 1 sugli errori."""
     from tht.sqlcheck import validate_sql