diff --git a/harness/tests/test_sql_preview_json.py b/harness/tests/test_sql_preview_json.py
index 99252d8f..f26371e8 100644
--- a/harness/tests/test_sql_preview_json.py
+++ b/harness/tests/test_sql_preview_json.py
@@ -38,6 +38,75 @@ def test_inject_limit_offset_default_offset_zero():
     assert "OFFSET" not in out
 
 
+def test_do_run_offset_truncated_when_extra_row(monkeypatch):
+    """Real do_run offset>0 path: runner returns N+1 rows → truncated True, exactly N returned.
+
+    Stubs the transport layer (_run_transport) so do_run's own offset logic
+    (probe_limit=N+1, compute truncated, slice to N) is exercised for real.
+    """
+    from types import SimpleNamespace
+
+    from tht.cli import sql_cmd
+    from tht.execute import ExecResult
+
+    captured = {}
+
+    def fake_transport(cfg, sql, *, limit):
+        captured["sql"] = sql
+        captured["limit"] = limit
+        # runner sliced to probe_limit (N+1=3): it found the sentinel extra row.
+        return ExecResult(columns=["a"], rows=[(1,), (2,), (3,)], execution_ms=5, truncated=False)
+
+    monkeypatch.setattr(sql_cmd, "_run_transport", fake_transport)
+    result = sql_cmd.do_run(SimpleNamespace(), "SELECT a FROM t", limit=2, offset=10)
+
+    # do_run must have probed for N+1 rows and wrapped with OFFSET.
+    assert captured["limit"] == 3
+    assert "OFFSET 10" in captured["sql"] and "LIMIT 3" in captured["sql"]
+    # extra row detected → truncated True, sliced back to exactly N=2.
+    assert result.truncated is True
+    assert result.rows == [(1,), (2,)]
+
+
+def test_do_run_offset_not_truncated_when_exactly_n(monkeypatch):
+    """Real do_run offset>0 path: runner returns exactly N rows → truncated False."""
+    from types import SimpleNamespace
+
+    from tht.cli import sql_cmd
+    from tht.execute import ExecResult
+
+    def fake_transport(cfg, sql, *, limit):
+        return ExecResult(columns=["a"], rows=[(1,), (2,)], execution_ms=5, truncated=False)
+
+    monkeypatch.setattr(sql_cmd, "_run_transport", fake_transport)
+    result = sql_cmd.do_run(SimpleNamespace(), "SELECT a FROM t", limit=2, offset=10)
+
+    assert result.truncated is False
+    assert result.rows == [(1,), (2,)]
+
+
+def test_do_run_offset_zero_path_unchanged(monkeypatch):
+    """offset == 0 bypasses the wrapper entirely: SQL and limit reach the runner verbatim."""
+    from types import SimpleNamespace
+
+    from tht.cli import sql_cmd
+    from tht.execute import ExecResult
+
+    captured = {}
+
+    def fake_transport(cfg, sql, *, limit):
+        captured["sql"] = sql
+        captured["limit"] = limit
+        return ExecResult(columns=["a"], rows=[(1,)], execution_ms=5, truncated=False)
+
+    monkeypatch.setattr(sql_cmd, "_run_transport", fake_transport)
+    sql_cmd.do_run(SimpleNamespace(), "SELECT a FROM t", limit=2, offset=0)
+
+    # no wrapper applied: original SQL and original limit passed straight through.
+    assert captured["sql"] == "SELECT a FROM t"
+    assert captured["limit"] == 2
+
+
 def test_preview_json_pure_stdout(monkeypatch, tmp_path, capsys):
     from types import SimpleNamespace
 
diff --git a/harness/tht/cli/sql_cmd.py b/harness/tht/cli/sql_cmd.py
index 191c7fa5..40c5f6b6 100644
--- a/harness/tht/cli/sql_cmd.py
+++ b/harness/tht/cli/sql_cmd.py
@@ -102,15 +102,8 @@ 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, 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)
-
+def _run_transport(cfg, sql: str, *, limit: int):
+    """Dispatch all'esecutore controllato secondo il transport (direct|rest)."""
     if cfg.database.transport == "rest":
         from tht.rest.execute import run_controlled_rest
 
@@ -122,6 +115,34 @@ def do_run(cfg, sql: str, *, limit: int, offset: int = 0):
     )
 
 
+def do_run(cfg, sql: str, *, limit: int, offset: int = 0):
+    """Esecuzione controllata secondo il transport configurato (direct|rest).
+
+    Per offset == 0: path invariato (LIMIT iniettato dall'esecutore via AST, +1 per
+    rilevare il troncamento).
+
+    Per offset > 0: la query viene wrappata in `SELECT * FROM (...) LIMIT (N+1) OFFSET M`.
+    Il +1 e' essenziale: l'esecutore vede gia' un LIMIT esterno, quindi il suo
+    `_inject_limit` non inietta nulla (bail perche' un LIMIT E' PRESENTE, non assente) e
+    non rileverebbe mai il troncamento. Recuperando N+1 righe qui calcoliamo noi
+    `truncated = len(rows) > N` e ritagliamo a N. NON rimuovere il LIMIT del wrapper
+    pensando sia ridondante: e' l'unico cap effettivo per il path con offset.
+    """
+    if offset == 0:
+        return _run_transport(cfg, sql, limit=limit)
+
+    from dataclasses import replace
+
+    from tht.execute.limit import inject_limit_offset
+
+    probe_limit = limit + 1
+    wrapped = inject_limit_offset(sql, limit=probe_limit, offset=offset)
+    # limit=probe_limit cosi' l'esecutore ritaglia a N+1 (non a N) e ci lascia la riga sonda.
+    result = _run_transport(cfg, wrapped, limit=probe_limit)
+    truncated = len(result.rows) > limit
+    return replace(result, rows=result.rows[:limit], truncated=truncated)
+
+
 @sql_app.command("validate")
 def validate_cmd(
     file: Path = typer.Argument(..., help="File SQL da validare."),