feat(harness): tht sql preview --json + --offset for AGGrid paging (BE-2)

Add inject_limit_offset (tht/execute/limit.py) — pure subquery wrapper that
applies LIMIT/OFFSET non-destructively without clobbering user-supplied LIMITs.
Wire offset param into do_run (pre-processing when offset>0) and add --offset /
--json flags to preview_cmd; JSON mode emits pristine stdout with columns, rows,
execution_ms, truncated, limit, offset. 5 new tests (4 unit + 1 JSON-purity),
219/219 total passing (no regressions).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

harness/tests/test_sql_preview_json.py

@@ -0,0 +1,65 @@
+"""Task 6: `tht sql preview --json` + `--offset` paging.
+
+Pure-logic tests (no DB needed):
+  1. inject_limit_offset — subquery wrapping, non-destructive.
+  2. preview_cmd JSON-purity — monkeypatched do_run, verifies stdout is valid JSON.
+"""
+import json
+
+import pytest
+
+from tht.execute.limit import inject_limit_offset
+
+
+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
+    # original query stays as a subquery (no clobber of an existing user LIMIT)
+    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
+
+
+def test_inject_limit_offset_strips_trailing_semicolon():
+    out = inject_limit_offset("SELECT 1;", limit=3, offset=0)
+    assert "LIMIT 3" in out
+    # semicolon should not appear in the inner subquery
+    assert ";" not in out
+
+
+def test_inject_limit_offset_default_offset_zero():
+    out = inject_limit_offset("SELECT x FROM y", limit=7)
+    assert "LIMIT 7" in out
+    assert "OFFSET" not in out
+
+
+def test_preview_json_pure_stdout(monkeypatch, tmp_path, capsys):
+    from types import SimpleNamespace
+
+    from tht.cli import sql_cmd
+
+    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)  # must parse: stdout pristine
+    assert data["columns"] == ["a"]
+    assert data["rows"] == [[1], [2]]
+    assert data["execution_ms"] == 3
+    assert data["truncated"] is False
+    assert "limit" in data
+    assert data["offset"] == 0

harness/tht/cli/sql_cmd.py

@@ -102,8 +102,15 @@ def do_explain(cfg, sql: str):
     return explain(_ro_engine(cfg), sql, timeout_ms=cfg.execution.statement_timeout_ms)
 
 
-def do_run(cfg, sql: str, *, limit: int):
+def do_run(cfg, sql: str, *, limit: int, offset: int = 0):
     """Esecuzione controllata secondo il transport configurato (direct|rest)."""
+    from tht.execute.limit import inject_limit_offset
+
+    # Inject OFFSET (and an outer LIMIT) via subquery wrapping when offset > 0.
+    # When offset == 0 we let the inner runners apply LIMIT directly (existing path).
+    if offset:
+        sql = inject_limit_offset(sql, limit=limit, offset=offset)
+
     if cfg.database.transport == "rest":
         from tht.rest.execute import run_controlled_rest
 
@@ -157,15 +164,13 @@ def explain_cmd(
 def preview_cmd(
     file: Path = typer.Argument(...),
     limit: int = typer.Option(None, "--limit", help="Default: execution.max_preview_rows."),
+    offset: int = typer.Option(0, "--offset", help="Riga di partenza (0-based) per il paging."),
     session: str = typer.Option(None, "--session"),
+    json_out: bool = typer.Option(False, "--json", help="Output JSON puro per il backend (sopprime tabella rich)."),
     config: Path = CONFIG_OPT,
 ) -> None:
     """Esecuzione controllata con LIMIT iniettato; aggregati mostrati per interi."""
-    from rich.console import Console
-    from rich.table import Table
-
     from tht.execute import ExecutionError
-    from tht.execute.warnings import runtime_warnings, static_warnings
 
     cfg = _load_config_or_exit(config)
     require_action(cfg, "preview")
@@ -173,11 +178,30 @@ def preview_cmd(
     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)
+        result = do_run(cfg, sql, limit=effective_limit, offset=offset)
     except ExecutionError as e:
         typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True)
         raise typer.Exit(code=1)
 
+    if json_out:
+        typer.echo(json.dumps(
+            {
+                "columns": 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
+
+    from rich.console import Console
+    from rich.table import Table
+
+    from tht.execute.warnings import runtime_warnings, static_warnings
+
     cells = len(result.rows) * len(result.columns)
     is_aggregate = (
         "aggregate" in cfg.execution.allow

harness/tht/execute/limit.py

@@ -0,0 +1,21 @@
+"""Pure subquery LIMIT/OFFSET injection for paging (non-destructive).
+
+Unlike `_inject_limit` (which uses AST rewriting and adds +1 for truncation detection),
+`inject_limit_offset` wraps the query as a subquery and applies LIMIT/OFFSET verbatim.
+This preserves any existing LIMIT that was part of the user's query (it becomes part
+of the inner subquery), and adds paging on top.
+"""
+
+
+def inject_limit_offset(sql: str, *, limit: int, offset: int = 0) -> str:
+    """Wrap query as subquery and apply LIMIT/OFFSET non-destructively.
+
+    The original query (including any user LIMIT) becomes the inner subquery,
+    so no existing LIMIT clause is overwritten. OFFSET clause is omitted when
+    offset == 0 to keep the SQL clean.
+    """
+    inner = sql.strip().rstrip(";")
+    clause = f"LIMIT {int(limit)}"
+    if offset:
+        clause += f" OFFSET {int(offset)}"
+    return f"SELECT * FROM (\n{inner}\n) AS _tht_page {clause}"