diff --git a/docs/superpowers/plans/2026-07-06-f4-schema-linking-column-curation-harness.md b/docs/superpowers/plans/2026-07-06-f4-schema-linking-column-curation-harness.md
new file mode 100644
index 00000000..a7e38fae
--- /dev/null
+++ b/docs/superpowers/plans/2026-07-06-f4-schema-linking-column-curation-harness.md
@@ -0,0 +1,852 @@
+# F4 schema-linking column curation — Plan 2 (harness + persistence + live)
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Depends on Plan 1** (the `schema-linking` frontend widget + `UiResponse.tables` contract) being merged.
+
+**Goal:** Make the F4 gate emit the `schema-linking` descriptor (real catalog columns, suggested pre-flagged), record the reviewer's per-table column curation as authoritative ledger decisions, project them deterministically into `schema_linking.json`, and instruct the model to honor the curated columns downstream — verified end-to-end on the live stack.
+
+**Architecture:** A new Pi reviewer tool `reviewer_schema_linking` (in `tht-gate.js`) enriches the model's table proposal with catalog columns (`tht schema columns`), emits the descriptor via the existing `emitAndWait`, then records `table_promoted`/`table_excluded` + `column_promoted`/`column_excluded` decisions and calls a deterministic projection (`tht session sync-schema-linking`) that rebuilds `schema_linking.json` from the ledger. The backend is already response-shape-agnostic (verify only). Downstream honoring is soft (Option 1): SKILL.md guidance, no hard SQL validator.
+
+**Tech Stack:** Python 3 (typer CLI, pydantic models), Node ESM Pi extension (`node --test`), Fastify backend (vitest). Live stack via `./scripts/run-stack.sh` (needs VPN + `harness/.env` + `pi` on PATH).
+
+## Global Constraints
+
+- `tht`'s `-c/--config` is a **per-command** option — it follows the subcommand, never precedes it.
+- `--json` output must be **pristine** (only valid JSON on stdout) — it is a machine contract consumed by the extension.
+- A decision `type` is only accepted in a phase listed in that phase's `emits` in `workflow.yaml` (drives `decision_min_phase` / `require_phase_or_exit`). New types **must** be added to F4 `emits` or they are rejected.
+- The extension's privileged `tht` calls go through `tht(ctx, args)` = `execFileSync("tht", args, {cwd: ctx.cwd})`; the anti-bypass hook blocks only the *model's* `bash` calls, not the extension.
+- `SchemaLinking` pydantic model has `extra:"forbid"` — the projected dict must match its fields exactly.
+- UI chrome English; data content (table/column names, Italian descriptions) as-is. No em dashes in copy you write.
+- Harness Python tests: `harness/.venv/bin/pytest -q` (mark `l2` for live GLM+DB is opt-in). JS gate tests: `cd harness && npm test` (`node --test .pi/extensions/gate/__tests__/*.test.js`). Backend: `cd backend && npx vitest run` + `npx tsc --noEmit -p .`.
+
+---
+
+## File Structure
+
+| File | Change |
+|---|---|
+| `harness/tht/cli/schema_cmd.py` (modify) | New `tht schema columns
--json` reader over `physical.yaml`. |
+| `harness/tht/decisions.py` (modify) | Add `column_promoted`, `column_excluded` to `DecisionType`. |
+| `harness/workflow.yaml` (modify) | Add the two types to F4 `emits`. |
+| `harness/tht/session/store.py` (modify) | `sync_schema_linking(session_id, sessions_root)` projection from the ledger. |
+| `harness/tht/cli/session_cmd.py` (modify) | `tht session sync-schema-linking ` command. |
+| `harness/.pi/extensions/gate/builders.js` (modify) | `buildSchemaLinkingRequest(...)`. |
+| `harness/.pi/extensions/tht-gate.js` (modify) | `reviewer_schema_linking` tool; extend `prepareReviewerArguments` for stringified `tables`. |
+| `harness/.pi/skills/tht-sessione/SKILL.md` (modify) | Phase 4: use `reviewer_schema_linking`; honor curated columns downstream. |
+| `harness/tht/cli/sql_cmd.py` (modify) | `promoted_columns_for` helper (persisted set; surfaced, not enforced). |
+| `harness/tests/…` + `backend/test/…` (tests) | pytest + node:test + a backend passthrough test. |
+
+**Non-goal (documented, not built):** an AST-level "output column diverges from curated set" warning in `sqlcheck`. Rationale: the final projection references CTE-qualified columns, not base `table.column`, so mapping back through CTEs is fragile; the human F7 gate is the backstop. We persist and surface the curated set instead.
+
+---
+
+## Task 1: Catalog reader — `tht schema columns`
+
+**Files:**
+- Modify: `harness/tht/cli/schema_cmd.py`
+- Test: `harness/tests/test_schema_columns_cmd.py` (create)
+
+**Interfaces:**
+- Produces: `tht schema columns --json` printing `{"table": str, "description": str, "columns": [{"name": str, "description": str, "type": str, "pk": bool}]}`.
+
+- [ ] **Step 1: Write the failing test**
+
+Create `harness/tests/test_schema_columns_cmd.py`:
+
+```python
+import json
+from datetime import datetime
+from typer.testing import CliRunner
+from tht.cli import app # the root Typer app that mounts schema_app as "schema"
+from tht.mschema.models import PhysicalSchema, TablePhysical, ColumnPhysical
+
+
+def _write_catalog(tmp_path):
+ phys = PhysicalSchema(
+ database="d", schema="s", introspected_at=datetime(2026, 1, 1),
+ tables={
+ "dim_patient": TablePhysical(
+ comment="Anagrafica",
+ columns={
+ "cod_paz": ColumnPhysical(type="bigint", pk=True, comment="Codice paziente"),
+ "nome": ColumnPhysical(type="text", comment="Nome"),
+ },
+ )
+ },
+ )
+ out = tmp_path / "artifacts" / "mschema" / "physical.yaml"
+ phys.to_yaml(out)
+ return out
+
+
+def test_schema_columns_json(tmp_path, monkeypatch):
+ _write_catalog(tmp_path)
+ cfg = tmp_path / "workspace.yaml"
+ cfg.write_text(
+ "database: {transport: none, database: d, schema: s}\n"
+ f"paths: {{artifacts: {tmp_path/'artifacts'}, indexes: {tmp_path/'i'}, sessions: {tmp_path/'s'}}}\n"
+ )
+ res = CliRunner().invoke(app, ["schema", "columns", "dim_patient", "--json", "-c", str(cfg)])
+ assert res.exit_code == 0, res.output
+ data = json.loads(res.output)
+ assert data["table"] == "dim_patient"
+ assert data["description"] == "Anagrafica"
+ assert {"name": "cod_paz", "description": "Codice paziente", "type": "bigint", "pk": True} in data["columns"]
+```
+
+Note: the config file shape must match `load_config`. Read one existing `harness/workspaces/*.yaml` and an existing schema test (e.g. any test invoking `schema render`) first and mirror their config fixture exactly; adjust the `cfg.write_text` block to the real minimal config. If a shared config fixture helper exists in `harness/tests/conftest.py`, use it.
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `cd harness && .venv/bin/pytest tests/test_schema_columns_cmd.py -q`
+Expected: FAIL — no such command `columns`.
+
+- [ ] **Step 3: Implement the command**
+
+In `harness/tht/cli/schema_cmd.py`, add after `render_cmd`:
+
+```python
+@schema_app.command("columns")
+def columns_cmd(
+ table: str = typer.Argument(..., help="Nome tabella (chiave in physical.yaml)."),
+ json_out: bool = typer.Option(False, "--json", help="Emetti JSON puro su stdout."),
+ config: Path = CONFIG_OPT,
+) -> None:
+ """Elenca nome/descrizione/tipo/pk delle colonne di una tabella dal catalogo."""
+ import json as _json
+
+ from tht.mschema.models import PhysicalSchema
+
+ cfg = _load_config_or_exit(config)
+ phys_file = physical_path(cfg)
+ if not phys_file.exists():
+ typer.secho(
+ f"ERRORE: {phys_file} non trovato. Esegui prima `tht schema introspect`.",
+ fg=typer.colors.RED, err=True,
+ )
+ raise typer.Exit(code=1)
+ physical = PhysicalSchema.from_yaml(phys_file)
+ tbl = physical.tables.get(table)
+ if tbl is None:
+ typer.secho(f"ERRORE: tabella non nel catalogo: {table}", fg=typer.colors.RED, err=True)
+ raise typer.Exit(code=1)
+ payload = {
+ "table": table,
+ "description": tbl.comment,
+ "columns": [
+ {"name": name, "description": col.comment, "type": col.type, "pk": col.pk}
+ for name, col in tbl.columns.items()
+ ],
+ }
+ if json_out:
+ typer.echo(_json.dumps(payload, ensure_ascii=False))
+ return
+ typer.echo(f"{table}: {tbl.comment}")
+ for c in payload["columns"]:
+ typer.echo(f" {'*' if c['pk'] else ' '} {c['name']} ({c['type']}) — {c['description']}")
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `cd harness && .venv/bin/pytest tests/test_schema_columns_cmd.py -q`
+Expected: PASS.
+
+- [ ] **Step 5: Lint + commit**
+
+Run: `cd harness && .venv/bin/ruff check tht/cli/schema_cmd.py`
+
+```bash
+git add harness/tht/cli/schema_cmd.py harness/tests/test_schema_columns_cmd.py
+git commit -m "feat(tht): schema columns reader (name/description/type/pk) from catalog"
+```
+
+---
+
+## Task 2: Ledger decision types + workflow emits
+
+**Files:**
+- Modify: `harness/tht/decisions.py`
+- Modify: `harness/workflow.yaml`
+- Test: `harness/tests/test_column_decisions.py` (create)
+
+**Interfaces:**
+- Produces: `column_promoted`, `column_excluded` accepted by `append_decision` and by F4's phase-eligibility.
+
+- [ ] **Step 1: Write the failing test**
+
+Create `harness/tests/test_column_decisions.py`:
+
+```python
+from tht.decisions import DecisionType
+import typing
+
+
+def test_column_decision_types_exist():
+ allowed = set(typing.get_args(DecisionType))
+ assert "column_promoted" in allowed
+ assert "column_excluded" in allowed
+
+
+def test_f4_emits_column_types():
+ import yaml
+ from pathlib import Path
+ wf = yaml.safe_load(Path("workflow.yaml").read_text())
+ f4 = next(p for p in wf["phases"] if p["id"] == "F4")
+ assert "column_promoted" in f4["emits"]
+ assert "column_excluded" in f4["emits"]
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `cd harness && .venv/bin/pytest tests/test_column_decisions.py -q`
+Expected: FAIL.
+
+- [ ] **Step 3: Add the types**
+
+In `harness/tht/decisions.py`, add to the `DecisionType` Literal (next to `column_corrected`):
+
+```python
+ "column_promoted",
+ "column_excluded",
+```
+
+In `harness/workflow.yaml`, F4 `emits` — append the two types:
+
+```yaml
+ emits: [table_promoted, table_excluded, column_promoted, column_excluded, column_corrected, join_modified,
+ evidence_accepted, evidence_rejected, value_grounded,
+ concept_formula_approved, concept_formula_rejected]
+```
+
+- [ ] **Step 4: Run to verify it passes + full suite unaffected**
+
+Run: `cd harness && .venv/bin/pytest tests/test_column_decisions.py -q && .venv/bin/pytest -q`
+Expected: new tests PASS; suite green (excluding opt-in `l2`).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add harness/tht/decisions.py harness/workflow.yaml harness/tests/test_column_decisions.py
+git commit -m "feat(tht): column_promoted/column_excluded decision types (F4)"
+```
+
+---
+
+## Task 3: Deterministic projection — `tht session sync-schema-linking`
+
+**Files:**
+- Modify: `harness/tht/session/store.py`
+- Modify: `harness/tht/cli/session_cmd.py`
+- Test: `harness/tests/test_sync_schema_linking.py` (create)
+
+**Interfaces:**
+- Consumes: `effective_decisions` (`tht.phase`), `set_schema_linking` (Task uses existing), `DecisionRecord`.
+- Produces: `sync_schema_linking(session_id, sessions_root) -> Path` and `tht session sync-schema-linking `.
+
+- [ ] **Step 1: Write the failing test**
+
+Create `harness/tests/test_sync_schema_linking.py`:
+
+```python
+import json
+from tht.session.store import sync_schema_linking, set_schema_linking
+from tht.decisions import append_decision
+
+
+def _new_session(tmp_path):
+ # Mirror the minimal session fixture used by other store tests (conftest helper
+ # if present). Must create session_manifest.yaml with a `question`.
+ from tht.session.store import create_session
+ from tht.config import DatabaseConfig # adjust import to the real DatabaseConfig
+ db = DatabaseConfig(transport="none", database="d", db_schema="s")
+ m = create_session("domanda X", db, tmp_path)
+ return m.id
+
+
+def test_projection_from_ledger(tmp_path):
+ sid = _new_session(tmp_path)
+ 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"
+```
+
+Note: read `harness/tests/conftest.py` and an existing store test (e.g. `test_session_mutations.py`) first; reuse their session-creation fixture and the real `DatabaseConfig` constructor signature instead of the placeholder import above.
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `cd harness && .venv/bin/pytest tests/test_sync_schema_linking.py -q`
+Expected: FAIL — `sync_schema_linking` undefined.
+
+- [ ] **Step 3: Implement the projection**
+
+In `harness/tht/session/store.py`, add:
+
+```python
+def sync_schema_linking(session_id: str, sessions_root: Path) -> Path:
+ """Project the effective F4 ledger decisions into schema_linking.json.
+
+ candidates/excluded are rebuilt from table_promoted/table_excluded +
+ column_promoted/column_excluded (last decision per subject wins). question,
+ joins, concept_formulas and open_questions are preserved from the existing
+ file when present. The reviewer's curation is thus authoritative and
+ deterministic (no model transcription)."""
+ from tht.phase import effective_decisions
+
+ session_dir = sessions_root / session_id
+ manifest = load_session(session_id, sessions_root)
+
+ existing: dict = {}
+ sl_path = session_dir / "schema_linking.json"
+ if sl_path.exists():
+ existing = json.loads(sl_path.read_text())
+
+ # last decision per subject wins (handles a re-run of the gate).
+ latest: dict[str, str] = {}
+ for d in effective_decisions(session_dir):
+ if d.type in ("table_promoted", "table_excluded", "column_promoted", "column_excluded"):
+ latest[d.subject] = d.type
+
+ candidates: list[dict] = []
+ excluded: list[dict] = []
+ for subject, dtype in latest.items():
+ is_column = "." in subject
+ kind = "column" if is_column else "table"
+ if dtype in ("table_promoted", "column_promoted"):
+ candidates.append({"kind": kind, "name": subject, "decision": "promoted"})
+ else:
+ excluded.append({"kind": kind, "name": subject})
+
+ data = {
+ "question": existing.get("question") or manifest.question,
+ "candidates": candidates,
+ "joins": existing.get("joins", []),
+ "excluded": excluded,
+ "open_questions": existing.get("open_questions", []),
+ "concept_formulas": existing.get("concept_formulas", []),
+ }
+ return set_schema_linking(session_id, data, sessions_root)
+```
+
+In `harness/tht/cli/session_cmd.py`, add a command (near `set_schema_linking_cmd`):
+
+```python
+@session_app.command("sync-schema-linking")
+def sync_schema_linking_cmd(
+ session_id: str = typer.Argument(...),
+ config: Path = CONFIG_OPT,
+) -> None:
+ """Riproietta schema_linking.json dalle decisioni F4 del ledger (deterministico)."""
+ from tht.session.store import sync_schema_linking
+
+ cfg = _load_config_or_exit(config)
+ load_session_or_exit(cfg, session_id)
+ path = sync_schema_linking(session_id, cfg.paths.sessions)
+ typer.secho(f"OK: schema_linking.json riproiettato ({path}).", fg=typer.colors.GREEN)
+```
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `cd harness && .venv/bin/pytest tests/test_sync_schema_linking.py -q`
+Expected: PASS.
+
+- [ ] **Step 5: Lint + commit**
+
+Run: `cd harness && .venv/bin/ruff check tht/session/store.py tht/cli/session_cmd.py`
+
+```bash
+git add harness/tht/session/store.py harness/tht/cli/session_cmd.py harness/tests/test_sync_schema_linking.py
+git commit -m "feat(tht): sync-schema-linking projects F4 ledger into schema_linking.json"
+```
+
+---
+
+## Task 4: Descriptor builder — `buildSchemaLinkingRequest`
+
+**Files:**
+- Modify: `harness/.pi/extensions/gate/builders.js`
+- Test: `harness/.pi/extensions/gate/__tests__/builders.test.js` (append)
+
+**Interfaces:**
+- Produces: `buildSchemaLinkingRequest({ id, phase, title, tables }) -> { type:"ui_request", widget:"schema-linking", tables, reserved }`.
+
+- [ ] **Step 1: Write the failing test**
+
+Append to `harness/.pi/extensions/gate/__tests__/builders.test.js` (match its existing `node:test` import style):
+
+```js
+test("buildSchemaLinkingRequest carries tables + reserved", () => {
+ const { buildSchemaLinkingRequest } = require("../builders.js");
+ const out = buildSchemaLinkingRequest({
+ id: "u7", phase: "F4", title: "F4 — Schema linking",
+ tables: [{ id: "t1", name: "dim_patient", kind: "promote", columns: [] }],
+ });
+ assert.equal(out.widget, "schema-linking");
+ assert.equal(out.id, "u7");
+ assert.equal(out.tables.length, 1);
+ assert.deepEqual(out.reserved, ["back", "exit", "other"]);
+});
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `cd harness && node --test .pi/extensions/gate/__tests__/builders.test.js`
+Expected: FAIL — `buildSchemaLinkingRequest is not a function`.
+
+- [ ] **Step 3: Implement the builder**
+
+In `harness/.pi/extensions/gate/builders.js`, add before `module.exports`:
+
+```js
+// Build a `schema-linking` ui_request: table rows (promote/exclude) each carrying
+// their full catalog columns; the frontend renders rows + a per-table columns
+// modal (suggested pre-selected). `tables` is validated shallowly here.
+function buildSchemaLinkingRequest({ id, phase, title, tables }) {
+ requireString(title, "title", "schema-linking");
+ const tabs = requireArray(tables, "tables", "schema-linking");
+ return {
+ type: "ui_request",
+ id,
+ phase,
+ schema_version: SCHEMA_VERSION,
+ widget: "schema-linking",
+ title,
+ tables: [...tabs],
+ reserved: RESERVED,
+ };
+}
+```
+
+Add `buildSchemaLinkingRequest,` to the `module.exports` object.
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `cd harness && node --test .pi/extensions/gate/__tests__/builders.test.js`
+Expected: PASS.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add harness/.pi/extensions/gate/builders.js harness/.pi/extensions/gate/__tests__/builders.test.js
+git commit -m "feat(gate): buildSchemaLinkingRequest descriptor builder"
+```
+
+---
+
+## Task 5: Reviewer tool `reviewer_schema_linking` + stringified `tables`
+
+**Files:**
+- Modify: `harness/.pi/extensions/tht-gate.js`
+- Test: `harness/.pi/extensions/gate/__tests__/gate_stringified_params.test.js` (append) + a new roundtrip test file.
+
+**Interfaces:**
+- Consumes: `buildSchemaLinkingRequest` (Task 4), `emitAndWait`, `decisionAddArgs`, `tht`, `relayIfThtFails` (all in `tht-gate.js`), `tht schema columns --json` (Task 1), `tht session sync-schema-linking` (Task 3).
+- Produces: Pi tool `reviewer_schema_linking`; `prepareReviewerArguments` parses a stringified `tables` array.
+
+- [ ] **Step 1: Extend `prepareReviewerArguments` (stringified `tables`)**
+
+In `harness/.pi/extensions/tht-gate.js`, inside `prepareReviewerArguments`, add next to the `options` block:
+
+```js
+ if (typeof args.tables === "string") {
+ try {
+ const parsed = JSON.parse(args.tables);
+ if (Array.isArray(parsed)) args.tables = parsed;
+ } catch {
+ /* not JSON */
+ }
+ }
+```
+
+- [ ] **Step 2: Write the failing stringified test**
+
+Append to `harness/.pi/extensions/gate/__tests__/gate_stringified_params.test.js` (mirror its existing imports of `prepareReviewerArguments`):
+
+```js
+test("prepareReviewerArguments parses stringified tables array", () => {
+ const out = prepareReviewerArguments({ tables: JSON.stringify([{ id: "t1", name: "x" }]) });
+ assert.ok(Array.isArray(out.tables));
+ assert.equal(out.tables[0].id, "t1");
+});
+```
+
+Run: `cd harness && node --test .pi/extensions/gate/__tests__/gate_stringified_params.test.js`
+Expected: FAIL before Step 1's edit, PASS after. (Apply Step 1, then rerun to confirm PASS.)
+
+- [ ] **Step 3: Import the builder**
+
+In `tht-gate.js`, add `buildSchemaLinkingRequest,` to the existing `import { … } from "./gate/builders.js";` block (alongside `buildMultiselectRequest`).
+
+- [ ] **Step 4: Register the tool**
+
+In `tht-gate.js`, after the `reviewer_decide` `pi.registerTool({...})` block, add:
+
+```js
+ pi.registerTool({
+ name: "reviewer_schema_linking",
+ label: "Schema linking: tabelle/colonne/esclusioni (reviewer)",
+ description:
+ "F4: presenta al reviewer le tabelle da promuovere/escludere con la loro descrizione " +
+ "e, per ogni tabella, le colonne dal catalogo (le suggerite pre-selezionate). Il reviewer " +
+ "cura le colonne di ogni tabella promossa. PERSISTE table_promoted/table_excluded e " +
+ "column_promoted/column_excluded, poi riproietta schema_linking.json in modo deterministico " +
+ "(tht session sync-schema-linking). `tables[]`: {id, name, kind: 'promote'|'exclude', " +
+ "rationale?, suggested_columns?: string[]}. Le colonne complete arrivano dal catalogo, non dal modello.",
+ parameters: Type.Object({
+ session: Type.String(),
+ title: Type.String(),
+ tables: Type.Array(
+ Type.Object({
+ id: Type.String(),
+ name: Type.String(),
+ kind: Type.Union([Type.Literal("promote"), Type.Literal("exclude")]),
+ rationale: Type.Optional(Type.String()),
+ suggested_columns: Type.Optional(Type.Array(Type.String())),
+ recommended: Type.Optional(Type.Boolean()),
+ }),
+ ),
+ advance: Type.Optional(Type.Boolean()),
+ }),
+ prepareArguments: prepareReviewerArguments,
+ async execute(_id, params, _signal, _onUpdate, ctx) {
+ lockActive = true;
+ const { session, title, tables, advance } = params;
+ const phase = phaseId(ctx, currentPhase(ctx, session));
+
+ // Enrich each table with its full catalog columns (deterministic source).
+ const enriched = tables.map((t) => {
+ const cat = JSON.parse(tht(ctx, ["schema", "columns", t.name, "--json"]));
+ const suggested = new Set(t.suggested_columns ?? []);
+ return {
+ id: t.id,
+ name: t.name,
+ kind: t.kind,
+ recommended: t.recommended ?? true,
+ description: cat.description ?? "",
+ rationale: t.rationale ?? "",
+ columns: cat.columns.map((c) => ({ ...c, suggested: suggested.has(c.name) })),
+ };
+ });
+
+ const widget = buildSchemaLinkingRequest({
+ id: `u${Date.now()}`,
+ phase,
+ title,
+ tables: enriched,
+ });
+ const resp = await emitAndWait(ctx, widget);
+ if (resp.control === "back") return textResult("Il reviewer vuole tornare indietro.");
+ if (resp.control === "exit") return textResult("Il reviewer vuole uscire.");
+ if (resp.control === "freetext")
+ return textResult(`Altro (reviewer): ${resp.text}. Riformula tenendone conto.`);
+
+ const byId = new Map(enriched.map((t) => [t.id, t]));
+ let n = 0;
+ for (const rt of resp.tables ?? []) {
+ const t = byId.get(rt.id);
+ if (!t || !rt.enacted) continue;
+ if (t.kind === "promote") {
+ const e1 = relayIfThtFails(ctx, decisionAddArgs(session, {
+ type: "table_promoted", subject: t.name, detail: t.description, rationale: t.rationale,
+ }), "");
+ if (e1) return e1;
+ n++;
+ const sel = new Set(rt.columns ?? []);
+ for (const c of t.columns) {
+ const type = sel.has(c.name)
+ ? "column_promoted"
+ : (c.suggested ? "column_excluded" : null);
+ if (!type) continue;
+ const e2 = relayIfThtFails(ctx, decisionAddArgs(session, {
+ type, subject: `${t.name}.${c.name}`, detail: c.description ?? "",
+ }), "");
+ if (e2) return e2;
+ }
+ } else {
+ const e3 = relayIfThtFails(ctx, decisionAddArgs(session, {
+ type: "table_excluded", subject: t.name, detail: t.description, rationale: t.rationale,
+ }), "");
+ if (e3) return e3;
+ n++;
+ }
+ }
+
+ // Deterministic projection of the ledger into schema_linking.json.
+ const eSync = relayIfThtFails(ctx, ["session", "sync-schema-linking", session], "");
+ if (eSync) return eSync;
+
+ if (advance) advanceIfReady(ctx, session);
+ return textResult(`Schema linking registrato dal reviewer (${n} tabelle + colonne curate).`);
+ },
+ });
+```
+
+- [ ] **Step 5: Write a roundtrip test**
+
+Create `harness/.pi/extensions/gate/__tests__/gate_schema_linking.test.js`, modeled on `gate_roundtrip.test.js` (read it first for the `fake_pi_runtime` harness + how it stubs `ctx.ui.input`, `execFileSync`/`tht`). The test must:
+- stub `tht schema columns dim_patient --json` → `{"table":"dim_patient","description":"Anagrafica","columns":[{"name":"cod_paz","description":"Codice","type":"bigint","pk":true},{"name":"nome","description":"Nome","type":"text","pk":false}]}`;
+- stub `ctx.ui.input` to return `JSON.stringify({ id:, kind:"schema-linking", tables:[{ id:"t-pat", enacted:true, columns:["cod_paz"] }] })`;
+- assert the recorded `tht decision add` calls include `table_promoted dim_patient`, `column_promoted dim_patient.cod_paz`, `column_excluded dim_patient.nome`, and that `session sync-schema-linking` was invoked.
+
+Use the same capture mechanism `gate_roundtrip.test.js` uses to intercept `tht(...)` argv.
+
+- [ ] **Step 6: Run the gate tests**
+
+Run: `cd harness && npm test`
+Expected: all gate tests PASS, including the new roundtrip + stringified cases.
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add harness/.pi/extensions/tht-gate.js harness/.pi/extensions/gate/__tests__/gate_stringified_params.test.js harness/.pi/extensions/gate/__tests__/gate_schema_linking.test.js
+git commit -m "feat(gate): reviewer_schema_linking tool with per-column curation + deterministic sync"
+```
+
+---
+
+## Task 6: SKILL.md Phase 4 guidance
+
+**Files:**
+- Modify: `harness/.pi/skills/tht-sessione/SKILL.md`
+
+- [ ] **Step 1: Read the current Phase 4 section**
+
+Read the `## Phase 4 — Schema linking` section (after Phase 3, ~line 200+) and the phase-map row (line 40). Anchor edits there.
+
+- [ ] **Step 2: Replace the tables/columns/exclusions step**
+
+Where Phase 4 currently instructs a `reviewer_decide` for tables/columns/exclusions, replace with (keep the joins gate and `reviewer_confirm kind:"phase"` as they are):
+
+```markdown
+1. Propose tables to promote/exclude with **`reviewer_schema_linking`**: pass
+ `tables[]` as `{ id, name, kind: "promote"|"exclude", rationale, suggested_columns }`.
+ Do NOT list every column yourself — the gate loads the full column set (with
+ descriptions) from the catalog and pre-selects your `suggested_columns`. The
+ reviewer curates the columns per promoted table. The tool records
+ `table_promoted`/`table_excluded` + `column_promoted`/`column_excluded` and
+ re-projects `schema_linking.json` deterministically (you do NOT hand-write the
+ tables/columns part with `write_schema_linking`).
+```
+
+- [ ] **Step 3: Add the downstream honoring rule**
+
+In the phase-map row for F4 and in the Phase 6/7 sections, add:
+
+```markdown
+The promoted columns in `schema_linking.json` are the reviewer-approved OUTPUT
+columns: project exactly those in the final SELECT. You remain free to reference
+other columns as join keys or filter predicates when the query requires them.
+```
+
+- [ ] **Step 4: Sanity + commit**
+
+No automated test (prose). Re-read the edited section for consistency with the tool's actual behavior.
+
+```bash
+git add harness/.pi/skills/tht-sessione/SKILL.md
+git commit -m "docs(skill): Phase 4 uses reviewer_schema_linking; honor curated output columns"
+```
+
+---
+
+## Task 7: Backend passthrough verification (no prod change)
+
+**Files:**
+- Test: `backend/test/schema-linking-passthrough.test.ts` (create)
+
+**Rationale:** `SessionBridge.respond(uiResponse)` sends `{ type:"extension_ui_response", id, value: JSON.stringify(uiResponse) }` — it is already shape-agnostic, so the structured `{ tables: [...] }` reaches Pi untouched. This task pins that with a test; no production code changes.
+
+- [ ] **Step 1: Write the test**
+
+Create `backend/test/schema-linking-passthrough.test.ts`, modeled on the existing bridge test (read `backend/test/` for the `RpcClient` stub pattern):
+
+```ts
+import { describe, it, expect, vi } from "vitest";
+import { SessionBridge } from "../src/bridge/session-bridge";
+
+describe("schema-linking response passthrough", () => {
+ it("forwards the structured tables payload to Pi verbatim", () => {
+ const send = vi.fn();
+ const rpc = { on: vi.fn(), send } as any;
+ const bridge = new SessionBridge(rpc);
+ // simulate a pending Pi ui request id
+ (bridge as any).pendingPiId = "pi-123";
+ const resp = { id: "u7", kind: "schema-linking", tables: [{ id: "t1", enacted: true, columns: ["cod_paz"] }] };
+ bridge.respond(resp);
+ expect(send).toHaveBeenCalledWith({
+ type: "extension_ui_response",
+ id: "pi-123",
+ value: JSON.stringify(resp),
+ });
+ });
+});
+```
+
+Note: adjust the private-field priming (`pendingPiId`) to how `respond` actually reads the pending id (read `session-bridge.ts:respond`); if `respond` no-ops without a pending id, set it as the real code expects.
+
+- [ ] **Step 2: Run + typecheck + commit**
+
+Run: `cd backend && npx vitest run test/schema-linking-passthrough.test.ts && npx tsc --noEmit -p .`
+
+```bash
+git add backend/test/schema-linking-passthrough.test.ts
+git commit -m "test(backend): pin schema-linking structured response passthrough"
+```
+
+---
+
+## Task 8: `promoted_columns_for` helper (surface, not enforce)
+
+**Files:**
+- Modify: `harness/tht/cli/sql_cmd.py`
+- Test: `harness/tests/test_promoted_columns_for.py` (create)
+
+**Rationale:** expose the curated column set as a persisted, queryable helper (used by SKILL surfacing / future work). The AST divergence warning is a documented non-goal (fragile through CTEs).
+
+- [ ] **Step 1: Write the failing test**
+
+Create `harness/tests/test_promoted_columns_for.py`:
+
+```python
+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"}
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `cd harness && .venv/bin/pytest tests/test_promoted_columns_for.py -q`
+Expected: FAIL — `promoted_columns_for` undefined.
+
+- [ ] **Step 3: Implement**
+
+In `harness/tht/cli/sql_cmd.py`, add next to `promoted_tables_for`:
+
+```python
+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"
+ }
+```
+
+- [ ] **Step 4: Run + lint + commit**
+
+Run: `cd harness && .venv/bin/pytest tests/test_promoted_columns_for.py -q && .venv/bin/ruff check tht/cli/sql_cmd.py`
+
+```bash
+git add harness/tht/cli/sql_cmd.py harness/tests/test_promoted_columns_for.py
+git commit -m "feat(tht): promoted_columns_for helper (curated column set)"
+```
+
+---
+
+## Task 9: Live end-to-end verification
+
+**Files:** none (verification procedure). Requires VPN + `harness/.env` + `pi` on PATH.
+
+- [ ] **Step 1: Rebuild + start the full stack**
+
+Run: `./scripts/run-stack.sh` (frontend :5173 → backend :8787). Confirm both are up.
+
+- [ ] **Step 2: Drive an F4 session to the schema-linking gate**
+
+In the browser (:5173), start a new session with a question that promotes several tables (reuse the psd ablation question). Advance F1 to F4. At F4 the gate now renders as the `schema-linking` widget (table rows + description + rationale + "Colonne k/n").
+
+- [ ] **Step 3: Curate columns and confirm**
+
+- [ ] For a promoted table, open the modal: suggested columns are pre-checked + bold.
+- [ ] Deselect one suggested column and select one previously-unsuggested column; confirm the bold updates and the row count changes.
+- [ ] Confirm the gate.
+
+- [ ] **Step 4: Verify persistence on disk**
+
+Run (psd sessions path):
+
+```bash
+SID=
+cd /Users/mp/projects/tht-workspace-psd/sessions/$SID
+grep -E '"type": ?"(table_promoted|table_excluded|column_promoted|column_excluded)"' review_decisions.jsonl
+node -e 'const j=require("./schema_linking.json");console.log(j.candidates.filter(c=>c.kind==="column"&&c.decision==="promoted").map(c=>c.name))'
+```
+
+Expected: the ledger shows the per-column decisions matching your clicks; `schema_linking.json` promoted columns equal exactly the curated set (deselected suggested column absent, newly selected column present).
+
+- [ ] **Step 5: Verify downstream honoring (soft)**
+
+Continue the session through F5/F6 to F7 (final SQL). Inspect `sql_final.sql`:
+
+- [ ] The final SELECT projects the curated output columns (the deselected column is not in the output; the added column is).
+- [ ] Join keys / filter predicates outside the curated set are still allowed (no rejection).
+
+Record the outcome. If the model ignores the curated set, tighten the Task 6 SKILL.md wording and re-run (this is the only step that exercises the model's honoring).
+
+- [ ] **Step 6: Regression sweep**
+
+Run all offline suites:
+
+```bash
+cd harness && .venv/bin/pytest -q && npm test
+cd ../backend && npx vitest run && npx tsc --noEmit -p .
+cd ../frontend && npx vitest run && npx tsc -b
+```
+
+Expected: all green.
+
+---
+
+## Self-Review
+
+**Spec coverage (harness slice of the design doc):**
+- §1a descriptor built by harness (catalog merge + suggested) → Tasks 1, 4, 5. ✅
+- §1b structured response consumed → Task 5 handler; §4 backend passthrough → Task 7. ✅
+- §4 ledger types → Task 2; deterministic `schema_linking.json` reconcile → Task 3; catalog reader → Task 1; SKILL.md → Task 6. ✅
+- §5 downstream soft (Option 1) → Task 6 guidance + Task 8 helper; hard enforcement remains out of scope (documented non-goal). ✅
+- §6 live verification (now possible with VPN) → Task 9. ✅
+
+**Placeholder scan:** code is concrete. Three tasks (1, 3, 5) explicitly instruct reading a neighboring fixture/test first to mirror exact config/harness shapes (test scaffolding that genuinely depends on existing conventions) rather than guessing them; the production code in every task is complete.
+
+**Type consistency:** decision types `column_promoted`/`column_excluded` identical across decisions.py, workflow.yaml, the gate handler, and sync projection. `tht schema columns` JSON shape (`{table, description, columns:[{name,description,type,pk}]}`) produced in Task 1 and consumed verbatim in Task 5. Response shape `{ id, kind:"schema-linking", tables:[{id,enacted,columns?}] }` matches Plan 1 and the Task 5/7 consumers. `sync_schema_linking(session_id, sessions_root)` signature identical in store.py, the CLI command, and the gate's `session sync-schema-linking` shell call.
+
+**Ordering note for implementer:** Task 5's gate handler depends on Tasks 1 (`schema columns`), 3 (`sync-schema-linking`), and 4 (builder) existing; keep the task order. Task 9 depends on Plan 1 being merged so the frontend can render the descriptor.