ThothII/docs/superpowers/plans/2026-06-27-harness-rpc-readiness.md

# Harness RPC-readiness Implementation Plan

> **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.

**Goal:** Rendere l'harness `tht` pienamente pilotabile da un client RPC esterno (il backend): il gate funziona in `pi --mode rpc` (kickoff + round-trip widget), il backend possiede l'id di sessione, il CLI espone le uscite `--json` necessarie, ed esistono i due test-double che chiudono il loop in CI.

**Architecture:** Si parte da uno **spike** che osserva il comportamento reale del gate dentro `pi --mode rpc` (mai testato finora, cf. `docs/l2-run-report-2026-06-27.md`). Le scoperte dello spike fissano l'esatta forma del wire e guidano l'adattamento del gate. Tutto l'adattamento del gate è coperto da un **fake-pi-runtime** (mock dell'API estensione `pi`) eseguibile in CI; le aggiunte al CLI sono TDD deterministiche; un **fake-pi-rpc** (processo che parla il protocollo JSONL su stdio) viene consegnato qui come asset condiviso per il Piano Backend, con un golden test di contratto (D10).

**Tech Stack:** Python 3.13 + Typer + pytest (CLI `tht`); JavaScript ESM/CJS + `node --test` (gate + test-double); Pi = `@earendil-works/pi-coding-agent` (binario `pi`, modalità `--mode rpc`).

## Global Constraints

- **Node ≥ 20**; JS test runner: `node --test` (come `harness/package.json` → `npm test`).
- **Python 3.13**, ambiente in `harness/.venv` (`pip install -e ".[dev]"`); test: `pytest` da `harness/`.
- **Framing RPC: LF-only JSONL** — serializzazione `JSON.stringify(value) + "\n"`; lettura: split su `\n`, strip di un eventuale `\r` finale. MAI `readline` (spezza su separatori Unicode validi dentro le stringhe JSON). Riferimento canonico: `@earendil-works/pi-coding-agent/dist/modes/rpc/jsonl.js`.
- **`--json` mantiene stdout puro**: in modalità JSON nessun warning/tabella umana su stdout (pattern già in `tht/cli/search_cmd.py`). Output: `typer.echo(json.dumps(data, ensure_ascii=False, indent=2))`.
- **Nessun segreto nel codice/test**: le credenziali stanno solo in `harness/.env` (gitignored). I task L2/spike che toccano Pi reale richiedono `.env` + VPN e NON girano in CI.
- **Il gate resta load-bearing**: gli invarianti verbatim del gate (anti-bypass `tool_call` hook, input-lock, no-limbo) non vanno indeboliti dagli adattamenti RPC.

---

### Task 1: Spike — comportamento del gate in `pi --mode rpc`

> Spike di osservazione (non TDD): mai verificato finora. L'esito fissa la forma esatta del wire e decide i Task 3–4. Richiede Pi reale + `.env` + VPN.

**Files:**
- Create: `harness/scripts/rpc_probe.mjs` (driver manuale usa-e-getta, committato come strumento)
- Create: `harness/docs/rpc-readiness-findings.md` (referto delle osservazioni + decisioni)

**Interfaces:**
- Produces: `harness/docs/rpc-readiness-findings.md` con le risposte alle 4 domande sotto, citate dai Task 3 e 4.

- [ ] **Step 1: Scrivere il driver di probe**

`harness/scripts/rpc_probe.mjs` fa spawn di Pi in RPC, manda l'avvio del workflow come comando `prompt`, stampa ogni riga JSONL ricevuta con un prefisso, e quando arriva un `extension_ui_request` risponde con un `extension_ui_response` correlato per `id`.

```javascript
// rpc_probe.mjs — manual probe: drive `pi --mode rpc` and observe the gate.
// Usage (from harness/, with .env loaded + VPN up):  node scripts/rpc_probe.mjs
import { spawn } from "node:child_process";

const pi = spawn("pi", ["--mode", "rpc"], { cwd: process.cwd(), env: process.env });

const send = (obj) => {
  const line = JSON.stringify(obj) + "\n";
  process.stdout.write(`>>> SEND ${line}`);
  pi.stdin.write(line);
};

let buf = "";
pi.stdout.on("data", (chunk) => {
  buf += chunk.toString("utf8");
  for (let nl; (nl = buf.indexOf("\n")) !== -1; ) {
    const line = buf.slice(0, nl).replace(/\r$/, "");
    buf = buf.slice(nl + 1);
    if (!line) continue;
    console.log(`<<< RECV ${line}`);
    let msg;
    try { msg = JSON.parse(line); } catch { continue; }
    if (msg.type === "extension_ui_request") {
      // Reply in BOTH the gate's expected shape and Pi's native shape; observe which one unblocks.
      const id = msg.id ?? msg.ui_request?.id;
      send({ type: "extension_ui_response", id, control: "freetext", text: "PROBE-ANSWER" });
    }
  }
});
pi.stderr.on("data", (d) => process.stdout.write(`!!! STDERR ${d}`));
pi.on("exit", (code) => console.log(`### pi exited ${code}`));

// Kick off the workflow via an RPC prompt command (NOT interactive keystrokes).
setTimeout(() => send({ type: "prompt", message: '/nuova-domanda "quante cardioversioni nel 2024"' }), 500);
setTimeout(() => { pi.stdin.end(); }, 60000);
```

- [ ] **Step 2: Eseguire il probe e catturare l'output**

Run (da `harness/`, con `.env` caricato e VPN attiva):
```bash
set -a; . ./.env; set +a
node scripts/rpc_probe.mjs | tee /tmp/rpc_probe.log
```
Expected: una sequenza di righe `<<< RECV {...}`. Osservare in particolare se compare `text_delta`/eventi del modello e se compare una riga `extension_ui_request`.

- [ ] **Step 3: Registrare le 4 osservazioni decisive in `rpc-readiness-findings.md`**

Documentare con evidenza (righe del log) le risposte a:
1. **Kickoff**: inviando il workflow come comando `prompt`, il gate attiva kickoff + input-lock? (cioè: il modello riceve le istruzioni operative e parte dalla Fase 1, oppure l'`input` hook — che filtra `event.source === "interactive"` — non scatta?)
2. **Emissione widget**: quando il gate chiama `emitAndWait`, su stdout appare `{"type":"extension_ui_request","ui_request":{...}}` (envelope custom del gate) oppure no?
3. **Routing risposta**: inviando `extension_ui_response` con `id` correlato, il gate prosegue (la `handleUiResponse` risolve la promise) oppure la risposta viene assorbita da `rpc-mode` (`pendingExtensionRequests`) e il gate resta appeso?
4. **Source dell'input `!`**: lo steering (`prompt`/`steer` con testo `!...`) raggiunge il modello?

- [ ] **Step 4: Decisione esplicita per i Task 3–4**

In coda al referto, scrivere la decisione: per il **kickoff** (Task 3) e per il **round-trip** (Task 4), indicare se basta confermare il meccanismo esistente o serve adattarlo, e come. Casi attesi:
- Se (1) è NO → il gate deve riconoscere l'avvio del workflow anche per `event.source !== "interactive"` (Task 3).
- Se (3) è "assorbita" → il gate deve emettere il widget tramite il meccanismo nativo che registra in `pendingExtensionRequests` (Task 4, variante B), invece del `sendRaw` custom (variante A).

- [ ] **Step 5: Commit**

```bash
git add harness/scripts/rpc_probe.mjs harness/docs/rpc-readiness-findings.md
git commit -m "spike(harness): probe gate behavior in pi --mode rpc + findings"
```

---

### Task 2: fake-pi-runtime — mock dell'API estensione `pi` per i test del gate

**Files:**
- Create: `harness/.pi/extensions/gate/__tests__/fake_pi_runtime.js`
- Test: `harness/.pi/extensions/gate/__tests__/fake_pi_runtime.test.js`

**Interfaces:**
- Produces: `createFakePi()` → `{ pi, ctx, sent, emit, lastSent() }` dove
  - `pi.on(event, handler)` registra handler; `pi.registerTool(def, fn)` li memorizza.
  - `pi.emit(event, payload)` invoca i handler registrati (await se async) e ritorna il loro valore (per testare i ritorni `{block, action}` dei hook).
  - `ctx.sendRaw(obj)` accoda in `sent[]`; `ctx.ui.notify(msg, level)` accoda in `ctx.notifications[]`; `ctx.hasUI = true`; `ctx.cwd = "<tmp>"`.
  - `lastSent()` ritorna l'ultimo oggetto passato a `sendRaw`.
- Consumes: nulla.

- [ ] **Step 1: Scrivere il test del runtime mock**

```javascript
const test = require("node:test");
const assert = require("node:assert");
const { createFakePi } = require("./fake_pi_runtime.js");

test("pi.on + emit invoca il handler e ne ritorna il valore", async () => {
  const { pi } = createFakePi();
  pi.on("tool_call", (e) => (e.toolName === "x" ? { block: true } : undefined));
  assert.deepEqual(await pi.emit("tool_call", { toolName: "x" }), { block: true });
  assert.equal(await pi.emit("tool_call", { toolName: "y" }), undefined);
});

test("ctx.sendRaw registra e lastSent ritorna l'ultimo", () => {
  const { ctx, lastSent } = createFakePi();
  ctx.sendRaw({ a: 1 });
  ctx.sendRaw({ a: 2 });
  assert.deepEqual(lastSent(), { a: 2 });
});
```

- [ ] **Step 2: Eseguire il test (deve fallire)**

Run: `cd harness && node --test .pi/extensions/gate/__tests__/fake_pi_runtime.test.js`
Expected: FAIL — `Cannot find module './fake_pi_runtime.js'`.

- [ ] **Step 3: Implementare il runtime mock**

```javascript
// fake_pi_runtime.js — minimal mock of the Pi extension runtime for gate tests.
function createFakePi() {
  const handlers = new Map();
  const tools = new Map();
  const sent = [];
  const ctx = {
    hasUI: true,
    cwd: "/tmp/fake-pi-session",
    notifications: [],
    sendRaw: (obj) => sent.push(obj),
    ui: { notify: async (message, level = "info") => ctx.notifications.push({ message, level }) },
  };
  const pi = {
    on: (event, handler) => {
      if (!handlers.has(event)) handlers.set(event, []);
      handlers.get(event).push(handler);
    },
    registerTool: (def, fn) => tools.set(def?.name ?? def, { def, fn }),
    emit: async (event, payload) => {
      let result;
      for (const h of handlers.get(event) ?? []) result = await h(payload, ctx);
      return result;
    },
  };
  return { pi, ctx, sent, emit: pi.emit, lastSent: () => sent[sent.length - 1] };
}
module.exports = { createFakePi };
```

- [ ] **Step 4: Eseguire il test (deve passare)**

Run: `cd harness && node --test .pi/extensions/gate/__tests__/fake_pi_runtime.test.js`
Expected: PASS (2 test).

- [ ] **Step 5: Commit**

```bash
git add harness/.pi/extensions/gate/__tests__/fake_pi_runtime.js harness/.pi/extensions/gate/__tests__/fake_pi_runtime.test.js
git commit -m "test(harness): fake-pi-runtime mock for gate CI tests"
```

---

### Task 3: Gate — kickoff + input-lock all'avvio del workflow in RPC mode

> Applica la decisione del Task 1 (osservazione #1). Il gate deve attivare kickoff + input-lock quando il workflow parte via comando RPC `prompt`, non solo da input interattivo.

**Files:**
- Modify: `harness/.pi/extensions/tht-gate.js` (handler `pi.on("input", …)`, ~riga 220-235)
- Test: `harness/.pi/extensions/gate/__tests__/gate_entry.test.js`

**Interfaces:**
- Consumes: `createFakePi()` (Task 2); il `default export` di `tht-gate.js` (la funzione `(pi) => {…}`).
- Produces: invariante "dopo un input di avvio workflow, `lockActive` è attivo e `pendingKickoff` è impostato", indipendentemente dal `source`.

- [ ] **Step 1: Scrivere il test di entry RPC**

```javascript
const test = require("node:test");
const assert = require("node:assert");
const { createFakePi } = require("./fake_pi_runtime.js");
const installGate = require("../../tht-gate.js").default ?? require("../../tht-gate.js");

test("avvio workflow via input non-interattivo attiva il lock (free text bloccato)", async () => {
  const { pi } = createFakePi();
  installGate(pi);
  // entry del workflow con source 'rpc' (come un comando prompt RPC)
  await pi.emit("input", { source: "rpc", text: '/nuova-domanda "x"' });
  // dopo l'entry, un testo libero senza '!' deve essere bloccato (lock attivo)
  const res = await pi.emit("input", { source: "rpc", text: "promuovi la tabella pazienti" });
  assert.equal(res.action, "handled");
});

test("testo con '!' passa al modello (steer) anche con lock attivo", async () => {
  const { pi } = createFakePi();
  installGate(pi);
  await pi.emit("input", { source: "rpc", text: "/nuova-domanda \"x\"" });
  const res = await pi.emit("input", { source: "rpc", text: "!considera solo il 2024" });
  assert.deepEqual(res, { action: "transform", text: "considera solo il 2024" });
});
```

- [ ] **Step 2: Eseguire il test (deve fallire)**

Run: `cd harness && node --test .pi/extensions/gate/__tests__/gate_entry.test.js`
Expected: FAIL — l'entry detection filtra `source === "interactive"`, quindi il lock non si attiva e il primo `emit` ritorna `{action:"continue"}` (non `handled`).

- [ ] **Step 3: Adattare l'entry detection nel gate**

In `tht-gate.js`, nel handler `pi.on("input", …)`: l'entry del workflow (`/nuova-domanda` | `/riprendi-sessione`) deve essere riconosciuta a prescindere dal `source`; il blocco del free-text resta valido per gli input dell'utente in sessione (interattivi o via RPC `prompt`). Sostituire la condizione di entry e quella di filtro:

```javascript
// entry detection: workflow-start funziona sia da TUI sia da comando RPC `prompt`.
if (/^\/(nuova-domanda|riprendi-sessione)\b/.test(raw)) {
  lockActive = true;
  lastSteered = false;
  pendingKickoff = /^\/nuova-domanda\b/.test(raw) ? NUOVA_DOMANDA_KICKOFF : RIPRENDI_KICKOFF;
}
// free-input block: attivo quando il lock è su, per qualsiasi input utente (non solo interattivo).
if (!lockActive) return { action: "continue" };
```

(Rimuovere i due `event.source === "interactive"` su entry e filtro. Conservare invariati: passthrough `/…`, canale `!` → `transform`, notify.)

- [ ] **Step 4: Eseguire i test (devono passare)**

Run: `cd harness && node --test .pi/extensions/gate/__tests__/gate_entry.test.js`
Expected: PASS (2 test). Poi `npm test` per assicurare nessuna regressione sui builder.
Expected: tutti verdi.

- [ ] **Step 5: Commit**

```bash
git add harness/.pi/extensions/tht-gate.js harness/.pi/extensions/gate/__tests__/gate_entry.test.js
git commit -m "fix(harness): gate kickoff/lock entry works in RPC mode (not only interactive)"
```

---

### Task 4: Gate — round-trip del widget in RPC mode

> Applica la decisione del Task 1 (osservazioni #2/#3). Garantisce che un widget emesso dal gate riceva la risposta dell'utente. Variante A (default): si conferma che `sendRaw` + `pi.on("extension_ui_response")` funziona. Variante B (se lo spike mostra che la risposta viene assorbita): emettere via il meccanismo nativo che registra in `pendingExtensionRequests`.

**Files:**
- Modify: `harness/.pi/extensions/tht-gate.js` (`emitAndWait` / registrazione `extension_ui_response`, ~riga 148-175 / 289)
- Test: `harness/.pi/extensions/gate/__tests__/gate_roundtrip.test.js`

**Interfaces:**
- Consumes: `createFakePi()` (Task 2). Estensione necessaria del mock: capacità di consegnare una `extension_ui_response` al gate — il test la inietta chiamando l'handler registrato dal gate.
- Produces: invariante "una `ui_response` con `id` correlato e `control !== 'cancel'` risolve `emitAndWait`; una `cancel`/`undefined` ri-emette lo stesso widget (no-limbo)".

- [ ] **Step 1: Estendere il fake-pi-runtime per consegnare risposte**

In `fake_pi_runtime.js` aggiungere, dentro `createFakePi`, un helper che invoca i handler registrati su un evento con payload (già coperto da `pi.emit`). Nessuna modifica se `pi.emit("extension_ui_response", resp)` raggiunge l'handler del gate; verificarlo nel test sotto.

- [ ] **Step 2: Scrivere il test del round-trip**

```javascript
const test = require("node:test");
const assert = require("node:assert");
const { createFakePi } = require("./fake_pi_runtime.js");
const installGate = require("../../tht-gate.js").default ?? require("../../tht-gate.js");

test("una ui_response correlata risolve l'attesa del widget", async () => {
  const { pi, lastSent } = createFakePi();
  installGate(pi);
  // accede a emitAndWait tramite un widget reale: si avvia un reviewer tool che emette un select.
  // Qui si testa il contratto osservabile: dopo l'emit, sendRaw contiene un extension_ui_request;
  // inviando la risposta correlata, la promise si risolve (nessun ri-invio).
  // (Il tool che emette è invocato dal gate; vedi nota implementativa nel piano.)
  // ... arrange: invoca il path che chiama emitAndWait con descriptor.id = "u1"
  // act: consegna la risposta
  await pi.emit("extension_ui_response", { id: "u1", control: "freetext", text: "ok" });
  const sent = lastSent();
  assert.equal(sent.type, "extension_ui_request");
});
```

> Nota implementativa: il widget è emesso da `emitAndWait`, chiamata dai reviewer tool (`reviewer_select`/`reviewer_confirm`). Il test invoca il tool registrato via `tools` del mock (Task 2 espone `registerTool`) e poi consegna la risposta. Concretizzare l'arrange invocando `tools.get("reviewer_select").fn(...)` con un descriptor a `id` noto.

- [ ] **Step 3: Eseguire il test (deve fallire o appendersi)**

Run: `cd harness && node --test .pi/extensions/gate/__tests__/gate_roundtrip.test.js`
Expected: FAIL (assert) o timeout (se la risposta non raggiunge `handleUiResponse`).

- [ ] **Step 4: Applicare la variante decisa dallo spike**

- **Variante A (sendRaw funziona):** nessuna modifica al meccanismo; assicurarsi solo che `handleUiResponse` sia registrato (`pi.on("extension_ui_response", handleUiResponse)`) e che `emitAndWait` correli per `descriptor.id`. Il test passa così com'è.
- **Variante B (risposta assorbita da rpc-mode):** modificare `emitAndWait` per emettere il widget tramite il meccanismo UI nativo che registra in `pendingExtensionRequests` (così la `extension_ui_response` viene instradata al gate), trasportando il widget-descriptor nel payload nativo (es. `method:"input"` con il descriptor serializzato in `title`/campo dedicato) e decodificando la risposta nativa (`{value}`) nel formato `ui_response` interno. Mantenere invariata la firma di `emitAndWait` e l'invariante no-limbo.

```javascript
// Variante B (estratto): emit via meccanismo nativo, decodifica la risposta nativa.
ctx.sendRaw({ type: "extension_ui_request", id: descriptor.id, method: "input",
              title: JSON.stringify({ widget_descriptor: descriptor }) });
// la risposta nativa { id, value } viene normalizzata in { id, ...JSON.parse(value) }
```

- [ ] **Step 5: Eseguire i test (devono passare) + no-limbo**

Run: `cd harness && node --test .pi/extensions/gate/__tests__/gate_roundtrip.test.js`
Expected: PASS. Aggiungere un caso che invia `{ id:"u1", control:"cancel" }` e verifica che il widget venga ri-emesso (secondo `sendRaw` con lo stesso `id`).

- [ ] **Step 6: Commit**

```bash
git add harness/.pi/extensions/tht-gate.js harness/.pi/extensions/gate/__tests__/
git commit -m "fix(harness): gate widget round-trip verified/adapted for RPC mode (no-limbo preserved)"
```

---

### Task 5: Sessione con id fornito dall'esterno (BE-5)

> Il backend pre-crea la sessione e ne possiede l'id; il gate deve USARE quell'id invece di istruire il modello a crearne uno. Veicolo: variabile d'ambiente `THT_SESSION` (già referenziata dal gate per `/torna`).

**Files:**
- Modify: `harness/.pi/extensions/tht-gate.js` (payload `NUOVA_DOMANDA_KICKOFF` + selezione kickoff, ~riga 44-62, 226-231)
- Test: `harness/.pi/extensions/gate/__tests__/gate_provided_session.test.js`

**Interfaces:**
- Consumes: `process.env.THT_SESSION` (impostata dal backend allo spawn).
- Produces: quando `THT_SESSION` è valorizzata, il kickoff iniettato istruisce il modello a USARE quell'id (niente `tht session new`); altrimenti comportamento attuale (il modello crea la sessione).

- [ ] **Step 1: Scrivere il test**

```javascript
const test = require("node:test");
const assert = require("node:assert");
const { createFakePi } = require("./fake_pi_runtime.js");
const installGate = require("../../tht-gate.js").default ?? require("../../tht-gate.js");

test("con THT_SESSION il kickoff usa l'id fornito e NON crea la sessione", async () => {
  process.env.THT_SESSION = "2026-06-27-100000-test";
  try {
    const { pi, ctx } = createFakePi();
    installGate(pi);
    await pi.emit("input", { source: "rpc", text: '/nuova-domanda "x"' });
    const injected = await pi.emit("before_agent_start", { });
    const text = injected?.appendMessage ?? injected?.text ?? "";
    assert.match(text, /2026-06-27-100000-test/);
    assert.doesNotMatch(text, /tht session new/);
  } finally { delete process.env.THT_SESSION; }
});
```

> Nota: adeguare il nome del campo ritornato da `before_agent_start` a come il gate inietta il kickoff (vedi handler ~riga 250). Il test asserisce il contenuto del testo iniettato.

- [ ] **Step 2: Eseguire il test (deve fallire)**

Run: `cd harness && node --test .pi/extensions/gate/__tests__/gate_provided_session.test.js`
Expected: FAIL — il kickoff contiene sempre `tht session new`.

- [ ] **Step 3: Implementare il kickoff a id fornito**

Aggiungere un secondo payload e selezionarlo quando `THT_SESSION` è presente:

```javascript
const NUOVA_DOMANDA_KICKOFF_PROVIDED = (sessionId) =>
  "Istruzioni operative — sessione ThothII (workflow human-in-the-middle). " +
  `La sessione è GIÀ creata: usa l'id \`${sessionId}\` in OGNI comando \`tht\`. ` +
  "NON eseguire `tht session new`.\n" +
  "1. Carica la skill leggendo `.pi/skills/tht-sessione/SKILL.md` con il tool `read`, " +
  `poi segui il workflow dalla Fase 1 usando l'id \`${sessionId}\`.\n` +
  "Regole non negoziabili: una domanda al reviewer per volta; mai promuovere/escludere/" +
  "correggere senza conferma; le interazioni passano dai tool reviewer_*; testo libero col prefisso '!'. " +
  "MAI `tht phase advance|reopen` né `tht decision add` da shell.";

// nella selezione del kickoff (input hook):
pendingKickoff = /^\/nuova-domanda\b/.test(raw)
  ? (process.env.THT_SESSION ? NUOVA_DOMANDA_KICKOFF_PROVIDED(process.env.THT_SESSION) : NUOVA_DOMANDA_KICKOFF)
  : RIPRENDI_KICKOFF;
```

- [ ] **Step 4: Eseguire i test (devono passare)**

Run: `cd harness && node --test .pi/extensions/gate/__tests__/gate_provided_session.test.js && npm test`
Expected: PASS, nessuna regressione.

- [ ] **Step 5: Commit**

```bash
git add harness/.pi/extensions/tht-gate.js harness/.pi/extensions/gate/__tests__/gate_provided_session.test.js
git commit -m "feat(harness): gate uses externally-provided THT_SESSION id (BE-5)"
```

---

### Task 6: CLI — `tht sql preview --json` + `--offset`

> Alimenta la paginazione AGGrid del backend. `do_run` ritorna già `rows/columns/execution_ms/truncated`; serve l'uscita JSON e l'iniezione di OFFSET.

**Files:**
- Modify: `harness/tht/cli/sql_cmd.py` (`preview_cmd`, ~riga 157; `do_run`, ~riga 105)
- Modify: `harness/tht/execute/__init__.py` e `harness/tht/rest/execute.py` (firma `run_controlled*` con `offset`)
- Test: `harness/tests/test_sql_preview_json.py`

**Interfaces:**
- Consumes: `do_run(cfg, sql, *, limit, offset=0)`.
- Produces: `tht sql preview <file> --json [--limit N] [--offset M] [--session S]` stampa su stdout `{"columns": [...], "rows": [[...]], "execution_ms": int, "truncated": bool, "limit": int, "offset": int}` e nient'altro.

- [ ] **Step 1: Scrivere il test (offset injection + json shape)**

```python
# harness/tests/test_sql_preview_json.py
import json
from tht.execute.limit import inject_limit_offset  # helper puro da creare

def test_inject_limit_offset_wraps_query():
    sql = "SELECT a FROM t ORDER BY a"
    out = inject_limit_offset(sql, limit=10, offset=20)
    assert "LIMIT 10" in out and "OFFSET 20" in out
    # la query originale resta una sottoquery (niente clobber di un LIMIT esistente)
    assert "SELECT a FROM t ORDER BY a" in out

def test_inject_limit_offset_zero_offset_no_offset_clause():
    out = inject_limit_offset("SELECT 1", limit=5, offset=0)
    assert "LIMIT 5" in out
    assert "OFFSET" not in out
```

- [ ] **Step 2: Eseguire il test (deve fallire)**

Run: `cd harness && pytest tests/test_sql_preview_json.py -v`
Expected: FAIL — `ModuleNotFoundError: tht.execute.limit`.

- [ ] **Step 3: Implementare l'helper di iniezione**

```python
# harness/tht/execute/limit.py
def inject_limit_offset(sql: str, *, limit: int, offset: int = 0) -> str:
    """Wrappa la query come sottoquery e applica LIMIT/OFFSET in modo non distruttivo.
    Evita di sovrascrivere un LIMIT già presente nella query dell'utente."""
    inner = sql.strip().rstrip(";")
    clause = f"LIMIT {int(limit)}" + (f" OFFSET {int(offset)}" if offset else "")
    return f"SELECT * FROM (\n{inner}\n) AS _tht_page {clause}"
```

- [ ] **Step 4: Eseguire il test (deve passare)**

Run: `cd harness && pytest tests/test_sql_preview_json.py -v`
Expected: PASS (2 test).

- [ ] **Step 5: Cablare `--json`/`--offset` in `preview_cmd` e `do_run`**

In `do_run` aggiungere `offset: int = 0` e usare `inject_limit_offset` quando `offset > 0` (altrimenti il path attuale a solo LIMIT). In `preview_cmd` aggiungere `offset` e `json_out`; in modalità JSON sopprimere tabella rich e warning, stampare solo il dict:

```python
@sql_app.command("preview")
def preview_cmd(
    file: Path = typer.Argument(...),
    limit: int = typer.Option(None, "--limit"),
    offset: int = typer.Option(0, "--offset"),
    session: str = typer.Option(None, "--session"),
    json_out: bool = typer.Option(False, "--json", help="Output JSON puro per il backend."),
    config: Path = CONFIG_OPT,
) -> None:
    cfg = _load_config_or_exit(config)
    require_action(cfg, "preview")
    sql = _read_sql(file)
    check = validate_or_exit(cfg, sql, session)
    effective_limit = limit if limit is not None else cfg.execution.max_preview_rows
    try:
        result = do_run(cfg, sql, limit=effective_limit, offset=offset)
    except ExecutionError as e:
        if json_out:
            typer.echo(json.dumps({"error": str(e)}, ensure_ascii=False)); raise typer.Exit(code=1)
        typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True); raise typer.Exit(code=1)
    if json_out:
        typer.echo(json.dumps({
            "columns": list(result.columns),
            "rows": [list(r) for r in result.rows],
            "execution_ms": result.execution_ms,
            "truncated": result.truncated,
            "limit": effective_limit, "offset": offset,
        }, ensure_ascii=False))
        return
    # ... (path umano esistente invariato)
```

- [ ] **Step 6: Test del contratto JSON (fixture senza DB)**

Aggiungere a `test_sql_preview_json.py` un test che invoca `preview_cmd` con `do_run` monkeypatchato a un risultato fittizio e verifica che stdout sia JSON puro con le chiavi attese.

```python
def test_preview_json_pure_stdout(monkeypatch, tmp_path, capsys):
    from tht.cli import sql_cmd
    from types import SimpleNamespace
    fake = SimpleNamespace(columns=["a"], rows=[[1],[2]], execution_ms=3, truncated=False)
    monkeypatch.setattr(sql_cmd, "do_run", lambda *a, **k: fake)
    monkeypatch.setattr(sql_cmd, "validate_or_exit", lambda *a, **k: SimpleNamespace(ast=None))
    monkeypatch.setattr(sql_cmd, "require_action", lambda *a, **k: None)
    monkeypatch.setattr(sql_cmd, "_load_config_or_exit",
                        lambda *a, **k: SimpleNamespace(execution=SimpleNamespace(max_preview_rows=100)))
    f = tmp_path / "q.sql"; f.write_text("SELECT 1")
    sql_cmd.preview_cmd(file=f, limit=None, offset=0, session=None, json_out=True, config=None)
    out = capsys.readouterr().out.strip()
    data = json.loads(out)  # deve parsare: stdout puro
    assert data["columns"] == ["a"] and data["rows"] == [[1],[2]]
```

Run: `cd harness && pytest tests/test_sql_preview_json.py -v`
Expected: PASS (tutti).

- [ ] **Step 7: Commit**

```bash
git add harness/tht/execute/limit.py harness/tht/cli/sql_cmd.py harness/tht/execute/__init__.py harness/tht/rest/execute.py harness/tests/test_sql_preview_json.py
git commit -m "feat(harness): tht sql preview --json + --offset for AGGrid paging (BE-2)"
```

---

### Task 7: CLI — `tht session list --json` + `tht session show --json`

> Alimenta la lista e il dettaglio sessioni nel FE. `list` è nuovo; `show` oggi stampa testo umano.

**Files:**
- Modify: `harness/tht/cli/session_cmd.py` (nuovo `list_cmd`; `show_cmd` con `--json`)
- Test: `harness/tests/test_session_list_json.py`

**Interfaces:**
- Produces:
  - `tht session list --json` → `[{"id","status","question","summary","created_at","updated_at","author"}, ...]` ordinato per `created_at` desc.
  - `tht session show <id> --json` → manifest completo + `{"phase": <int derivata>, "has_schema_linking": bool}`.

- [ ] **Step 1: Scrivere il test**

```python
# harness/tests/test_session_list_json.py
import json
from tht.session.store import create_session
from tht.session.models import SessionManifest

def test_list_json_lists_created_sessions(tmp_path):
    from tht.config import DatabaseConfig
    db = DatabaseConfig(database="d", schema="s", transport="rest")  # adattare ai campi reali
    m1 = create_session("prima domanda", db, tmp_path)
    m2 = create_session("seconda domanda", db, tmp_path)
    from tht.cli.session_cmd import _list_sessions  # helper puro
    rows = _list_sessions(tmp_path)
    ids = [r["id"] for r in rows]
    assert m1.id in ids and m2.id in ids
    assert set(["id","status","question","created_at"]).issubset(rows[0].keys())
```

- [ ] **Step 2: Eseguire il test (deve fallire)**

Run: `cd harness && pytest tests/test_session_list_json.py -v`
Expected: FAIL — `_list_sessions` non esiste.

- [ ] **Step 3: Implementare helper + comandi**

```python
def _list_sessions(sessions_root: Path) -> list[dict]:
    out = []
    for d in sorted([p for p in sessions_root.iterdir() if (p / "session_manifest.yaml").exists()]):
        m = SessionManifest.from_yaml(d / "session_manifest.yaml")
        out.append({"id": m.id, "status": m.status, "question": m.question,
                    "summary": m.summary, "created_at": m.created_at.isoformat(),
                    "updated_at": m.updated_at.isoformat() if m.updated_at else None,
                    "author": m.author})
    out.sort(key=lambda r: r["created_at"], reverse=True)
    return out

@session_app.command("list")
def list_cmd(json_out: bool = typer.Option(False, "--json"), config: Path = CONFIG_OPT) -> None:
    cfg = _load_config_or_exit(config)
    rows = _list_sessions(cfg.paths.sessions)
    if json_out:
        typer.echo(json.dumps(rows, ensure_ascii=False, indent=2)); return
    for r in rows:
        typer.echo(f"{r['id']}  [{r['status']}]  {r['summary']}")
```

E in `show_cmd` aggiungere `json_out: bool = typer.Option(False, "--json")`; in modalità JSON stampare il manifest (`model_dump(mode="json", by_alias=True)`) + `phase` (da `current_phase`) + `has_schema_linking`.

- [ ] **Step 4: Eseguire il test (deve passare)**

Run: `cd harness && pytest tests/test_session_list_json.py -v`
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add harness/tht/cli/session_cmd.py harness/tests/test_session_list_json.py
git commit -m "feat(harness): tht session list/show --json for FE session list"
```

---

### Task 8: Manifest — campi provider/model/thinking/name + opzioni di `tht session new`

> Persistono la scelta di modello/thinking/provider e il nome, riapplicati al resume dal backend (BE-6/BE-7).

**Files:**
- Modify: `harness/tht/session/models.py` (`SessionManifest`)
- Modify: `harness/tht/session/store.py` (`create_session`)
- Modify: `harness/tht/cli/session_cmd.py` (`new_cmd` opzioni)
- Test: `harness/tests/test_manifest_pi_fields.py`

**Interfaces:**
- Consumes: `create_session(question, db, sessions_root, *, author=None, summary=None, provider=None, model=None, thinking=None, name=None)`.
- Produces: manifest con campi opzionali `provider`, `model`, `thinking`, `name`; `tht session new <q> [--provider P --model M --thinking T --name N] [--json]` (con `--json` stampa `{"id": ...}` su stdout puro).

- [ ] **Step 1: Scrivere il test**

```python
# harness/tests/test_manifest_pi_fields.py
from tht.session.store import create_session
from tht.session.models import SessionManifest

def test_manifest_persists_pi_fields(tmp_path):
    from tht.config import DatabaseConfig
    db = DatabaseConfig(database="d", schema="s", transport="rest")
    m = create_session("q", db, tmp_path, provider="zai", model="glm-5.2",
                        thinking="medium", name="sessione test")
    reload = SessionManifest.from_yaml(tmp_path / m.id / "session_manifest.yaml")
    assert reload.provider == "zai" and reload.model == "glm-5.2"
    assert reload.thinking == "medium" and reload.name == "sessione test"

def test_manifest_pi_fields_optional(tmp_path):
    from tht.config import DatabaseConfig
    db = DatabaseConfig(database="d", schema="s", transport="rest")
    m = create_session("q", db, tmp_path)
    assert m.provider is None and m.model is None and m.thinking is None and m.name is None
```

- [ ] **Step 2: Eseguire il test (deve fallire)**

Run: `cd harness && pytest tests/test_manifest_pi_fields.py -v`
Expected: FAIL — `create_session` non accetta `provider`, ecc.

- [ ] **Step 3: Aggiungere i campi al modello e a create_session**

In `SessionManifest` (dopo `schema_version`):
```python
    provider: str | None = None
    model: str | None = None
    thinking: str | None = None
    name: str | None = None
```
In `create_session` aggiungere i parametri keyword e passarli al costruttore del manifest:
```python
def create_session(question, db, sessions_root, *, author=None, summary=None,
                   provider=None, model=None, thinking=None, name=None):
    ...
    manifest = SessionManifest(
        id=session_id, created_at=now, question=question,
        database=db.database, schema=db.db_schema,
        author=who, summary=summary or _summarize(question),
        updated_at=now, updated_by=who, schema_version=schema_version,
        provider=provider, model=model, thinking=thinking, name=name,
    )
```

- [ ] **Step 4: Eseguire il test (deve passare)**

Run: `cd harness && pytest tests/test_manifest_pi_fields.py -v`
Expected: PASS (2 test).

- [ ] **Step 5: Aggiungere le opzioni a `tht session new` + `--json`**

```python
@session_app.command("new")
def new_cmd(
    question: str = typer.Argument(...),
    provider: str = typer.Option(None, "--provider"),
    model: str = typer.Option(None, "--model"),
    thinking: str = typer.Option(None, "--thinking"),
    name: str = typer.Option(None, "--name"),
    json_out: bool = typer.Option(False, "--json"),
    config: Path = CONFIG_OPT,
) -> None:
    from tht.session.store import create_session
    cfg = _load_config_or_exit(config)
    manifest = create_session(question, cfg.database, cfg.paths.sessions,
                              provider=provider, model=model, thinking=thinking, name=name)
    if json_out:
        typer.echo(json.dumps({"id": manifest.id}, ensure_ascii=False)); return
    typer.secho(f"OK: sessione creata in {session_dir(cfg, manifest.id)}", fg=typer.colors.GREEN)
    typer.echo(manifest.id)
```

Run: `cd harness && pytest tests/test_manifest_pi_fields.py tests/test_session_list_json.py -v`
Expected: PASS (regressione esistente verde).

- [ ] **Step 6: Commit**

```bash
git add harness/tht/session/models.py harness/tht/session/store.py harness/tht/cli/session_cmd.py harness/tests/test_manifest_pi_fields.py
git commit -m "feat(harness): manifest provider/model/thinking/name + session new options (BE-6/7)"
```

---

### Task 9: `.pi/settings.json` — quietStartup + trust

> Correttezza dello spawn RPC: niente rumore di avvio su stdout (sporcherebbe il JSONL), file project-local fidati (niente prompt di trust che appende il loop).

**Files:**
- Modify: `harness/.pi/settings.json`
- Test: `harness/.pi/extensions/gate/__tests__/settings.test.js`

**Interfaces:**
- Produces: `.pi/settings.json` contiene almeno `{"theme": "thothii-mono", "quietStartup": true}`; il trust dei file project-local è documentato (verifica empirica nello spike/Task 10).

- [ ] **Step 1: Scrivere il test (forma del settings)**

```javascript
const test = require("node:test");
const assert = require("node:assert");
const fs = require("node:fs");
const path = require("node:path");

test("settings.json abilita quietStartup", () => {
  const s = JSON.parse(fs.readFileSync(path.join(__dirname, "../../settings.json"), "utf8"));
  assert.equal(s.quietStartup, true);
  assert.equal(s.theme, "thothii-mono");
});
```

- [ ] **Step 2: Eseguire il test (deve fallire)**

Run: `cd harness && node --test .pi/extensions/gate/__tests__/settings.test.js`
Expected: FAIL — `quietStartup` assente.

- [ ] **Step 3: Aggiornare settings.json**

```json
{
  "theme": "thothii-mono",
  "quietStartup": true
}
```

- [ ] **Step 4: Eseguire il test (deve passare)**

Run: `cd harness && node --test .pi/extensions/gate/__tests__/settings.test.js`
Expected: PASS.

- [ ] **Step 5: Documentare il trust + commit**

Aggiungere a `harness/docs/rpc-readiness-findings.md` una nota: come è stato concesso il trust dei file project-local allo spawn RPC (verificato che NON compaia un prompt di trust che blocca il loop — confermato nel Task 10 / spike).

```bash
git add harness/.pi/settings.json harness/.pi/extensions/gate/__tests__/settings.test.js harness/docs/rpc-readiness-findings.md
git commit -m "chore(harness): quietStartup + project-local trust for clean RPC spawn"
```

---

### Task 10: fake-pi-rpc — test-double del protocollo RPC + golden test di contratto (D10)

> Asset condiviso consegnato qui per il Piano Backend: un processo che parla il protocollo JSONL su stdio, scriptabile per emettere sequenze di eventi e accettare comandi. Un golden test fissa il contratto del widget-descriptor (D10).

**Files:**
- Create: `harness/tests/fake_pi/fake_pi_rpc.mjs`
- Create: `harness/tests/fake_pi/scripts/f1_disambiguation.json` (scenario scriptato)
- Test: `harness/tests/fake_pi/test_fake_pi_contract.mjs`

**Interfaces:**
- Produces: `fake_pi_rpc.mjs` — eseguibile con `node fake_pi_rpc.mjs <script.json>`:
  - legge comandi JSONL su stdin (LF-only); su `{type:"prompt"}` emette la sequenza di eventi dello script (es. un `extension_ui_request`); su `{type:"extension_ui_response", id}` correla e emette l'evento successivo dello script; risponde a `{type:"get_available_models"}` con un set fisso; eco di `{type:"response", command, success:true}` per i comandi che lo richiedono.
  - framing su stdout: `JSON.stringify(evt) + "\n"`.
- Consumes: lo schema widget-descriptor (architettura §4) per gli eventi di esempio.

- [ ] **Step 1: Definire lo scenario scriptato (golden)**

```json
// harness/tests/fake_pi/scripts/f1_disambiguation.json
{
  "on_prompt": [
    { "type": "extension_ui_request", "id": "u1",
      "ui_request": { "type": "ui_request", "id": "u1", "schema_version": 1,
        "phase": "F1_chiarimento", "title": "Disambigua",
        "widget": "select",
        "options": [ {"id":"a","label":"interpretazione A"}, {"id":"b","label":"interpretazione B"} ],
        "reserved": ["back","exit","other"] } }
  ],
  "on_response": { "u1": [ { "type": "agent_end" } ] },
  "available_models": [ {"provider":"zai","id":"glm-5.2"} ]
}
```

- [ ] **Step 2: Scrivere il test di contratto**

```javascript
// harness/tests/fake_pi/test_fake_pi_contract.mjs  — run: node --test
import test from "node:test";
import assert from "node:assert";
import { spawn } from "node:child_process";
import path from "node:path";

function drive(scriptPath, commands) {
  return new Promise((resolve) => {
    const fp = spawn("node", [path.join(import.meta.dirname, "fake_pi_rpc.mjs"), scriptPath]);
    const events = []; let buf = "";
    fp.stdout.on("data", (c) => {
      buf += c.toString("utf8");
      for (let nl; (nl = buf.indexOf("\n")) !== -1; ) {
        const line = buf.slice(0, nl).replace(/\r$/, ""); buf = buf.slice(nl + 1);
        if (line) events.push(JSON.parse(line));
      }
    });
    fp.on("exit", () => resolve(events));
    for (const cmd of commands) fp.stdin.write(JSON.stringify(cmd) + "\n");
    setTimeout(() => fp.stdin.end(), 300);
  });
}

test("on prompt emette il widget F1; on response avanza", async () => {
  const sp = path.join(import.meta.dirname, "scripts/f1_disambiguation.json");
  const events = await drive(sp, [
    { type: "prompt", message: "/nuova-domanda \"x\"" },
    { type: "extension_ui_response", id: "u1", choices: ["a"], decision: { type: "concept_clarified" } },
  ]);
  const widget = events.find((e) => e.type === "extension_ui_request");
  assert.equal(widget.ui_request.widget, "select");
  assert.equal(widget.ui_request.id, "u1");
  assert.ok(events.some((e) => e.type === "agent_end"));
});
```

- [ ] **Step 3: Eseguire il test (deve fallire)**

Run: `cd harness && node --test tests/fake_pi/test_fake_pi_contract.mjs`
Expected: FAIL — `fake_pi_rpc.mjs` non esiste.

- [ ] **Step 4: Implementare il fake-pi-rpc**

```javascript
// harness/tests/fake_pi/fake_pi_rpc.mjs — scripted RPC test double (LF-only JSONL).
import fs from "node:fs";
const script = JSON.parse(fs.readFileSync(process.argv[2], "utf8"));
const out = (evt) => process.stdout.write(JSON.stringify(evt) + "\n");

let buf = "";
process.stdin.on("data", (chunk) => {
  buf += chunk.toString("utf8");
  for (let nl; (nl = buf.indexOf("\n")) !== -1; ) {
    const line = buf.slice(0, nl).replace(/\r$/, ""); buf = buf.slice(nl + 1);
    if (!line) continue;
    let cmd; try { cmd = JSON.parse(line); } catch { continue; }
    if (cmd.type === "prompt") {
      for (const evt of script.on_prompt ?? []) out(evt);
    } else if (cmd.type === "extension_ui_response") {
      for (const evt of (script.on_response ?? {})[cmd.id] ?? []) out(evt);
    } else if (cmd.type === "get_available_models") {
      out({ type: "response", command: "get_available_models", id: cmd.id, success: true,
            data: { models: script.available_models ?? [] } });
    } else if (cmd.type === "steer") {
      out({ type: "response", command: "steer", id: cmd.id, success: true });
    } else if (cmd.type === "get_state") {
      out({ type: "response", command: "get_state", id: cmd.id, success: true,
            data: { sessionId: "fake", thinkingLevel: "medium", isStreaming: false } });
    }
  }
});
process.stdin.on("end", () => process.exit(0));
```

- [ ] **Step 5: Eseguire il test (deve passare)**

Run: `cd harness && node --test tests/fake_pi/test_fake_pi_contract.mjs`
Expected: PASS.

- [ ] **Step 6: Validazione end-to-end con Pi reale (L2, informativo, non-CI)**

Con `.env` + VPN, ri-eseguire `node scripts/rpc_probe.mjs` (Task 1) e confermare che, dopo i Task 3–5, il gate: (a) parte sul `prompt`, (b) emette il widget, (c) riceve la risposta e avanza. Annotare l'esito in `rpc-readiness-findings.md`. Questo chiude il rischio "path RPC mai testato".

- [ ] **Step 7: Commit**

```bash
git add harness/tests/fake_pi/
git commit -m "test(harness): fake-pi-rpc protocol double + F1 widget contract golden (D10)"
```

---

## Self-Review

**Spec coverage** (vs `2026-06-27-backend-design.md` §7 + decisioni BE):
- BE-5 (id fornito): Task 5 ✓
- BE-6/BE-7 (model/thinking/provider/name nel manifest): Task 8 ✓; settings spawn (quietStartup/trust): Task 9 ✓
- §7.1 (preview --json/--offset): Task 6 ✓
- §7.2 (kickoff con id): Task 5 ✓
- §7.3 (campi manifest): Task 8 ✓
- §7.4 (settings.json): Task 9 ✓
- §7.5 (session list/show --json): Task 7 ✓
- §7.6 (fake-Pi condiviso): Task 10 (fake-pi-rpc) + Task 2 (fake-pi-runtime) ✓
- Rischio "gate RPC mai testato": Task 1 (spike) + Task 3/4 (adattamento) + Task 10 Step 6 (validazione reale) ✓

**Placeholder scan:** Task 1 è uno spike dichiarato (osservazione, non TDD) — i suoi "step" sono azioni concrete con output atteso. Le varianti A/B del Task 4 sono entrambe specificate; la scelta è guidata dall'evidenza dello spike, non un TBD.

**Type consistency:** `create_session(..., provider, model, thinking, name)` (Task 8) coerente con i campi del manifest (Task 8) e con l'uso del backend (Piano 2). `inject_limit_offset(sql, *, limit, offset)` (Task 6) usato da `do_run(..., offset=)`. `_list_sessions` (Task 7) ritorna le chiavi usate dal FE. Envelope `{"type":"extension_ui_request","ui_request":{...}}` (Task 10) coerente con l'emissione del gate (Task 4) e con ciò che il backend tradurrà (Piano 2).

**Nota di sequenza:** il Task 1 (spike) richiede Pi reale + VPN; se non disponibile al momento dell'esecuzione, i Task 2 e 6–9 (CI puri) possono procedere in parallelo; i Task 3–4 (adattamento gate) richiedono la decisione dello spike e vanno dopo.