From e077f6458fcf1d75ab79d1de51bedcba84967bbb Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Sun, 12 Jul 2026 15:43:15 +0530 Subject: [PATCH 1/9] feat(execute_sql): extract Executor port + guarded envelope (AH-012 slice 1) Split execute_sql into a shared, un-bypassable guarded envelope (execute_guarded: read-only guard -> model-safety -> resolve datasource -> executor.execute) and a swappable Executor port (ports.Executor, 5th port). The built-in executor stays the default connect-per-query path; per-engine _run_ now return native-typed rows (ExecResult) instead of writing CSV, and the subprocess/CLI _execute_ wrappers serialize to stdout CSV at the edge (byte-identical). This is the seam a hosted consumer injects a pooled/RBAC/tunnel executor behind, no fork (REQ-002/REQ-014). Spec: AH-012 --- packages/agami-core/src/execute_sql.py | 489 +++++++++++++++++-------- packages/agami-core/src/ports.py | 45 ++- plugins/agami/lib/execute_sql.py | 489 +++++++++++++++++-------- 3 files changed, 711 insertions(+), 312 deletions(-) diff --git a/packages/agami-core/src/execute_sql.py b/packages/agami-core/src/execute_sql.py index 44222f5e..e791f660 100644 --- a/packages/agami-core/src/execute_sql.py +++ b/packages/agami-core/src/execute_sql.py @@ -46,11 +46,19 @@ import stat import sys import urllib.parse +from collections.abc import Callable +from dataclasses import dataclass from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any import agami_paths +if TYPE_CHECKING: + # ``Executor`` is the 5th port; imported only for type-checkers. At runtime ``execute_sql`` never + # imports ``ports`` (it ships in the stdlib-lean plugin mirror without it), so the annotation on + # ``execute_guarded`` stays a lazy string (``from __future__ import annotations``). + from ports import Executor + # Credentials + config now live under /local/ (the consolidated, # gitignored replacement for ~/.agami). The path is stable regardless of migration # timing — bootstrap() just moves the files into it. See agami_paths. @@ -87,6 +95,46 @@ def _err(msg: str, *, code: int = 2) -> int: return code +@dataclass(frozen=True) +class ExecResult: + """What an executor returns: columns + rows with **native Python types preserved** (ints, + Decimals, datetimes, ``None``), not stringified. Serializing to text — CSV for the subprocess + wire, JSON at the MCP-tool edge — is the *caller's* single, final step, so an in-process + executor never pays a serialize→re-parse round-trip and never loses a type or confuses NULL + with "". ``truncated`` mirrors the ``fetchmany(cap + 1)`` bound: True when the result was capped. + + This lives here (not in ``ports``) because ``execute_sql`` ships in the stdlib-lean plugin + mirror, which does not include ``ports``; ``ports.Executor`` references it under TYPE_CHECKING. + """ + + columns: list[str] + rows: list[tuple] + truncated: bool = False + + +class ExecutorError(Exception): + """A connect / credential / run failure raised by the built-in executor. Carries the exact + stderr message and exit code the subprocess CLI emits, so ``main`` reproduces today's bytes and + the in-process caller gets a catchable error instead of a process exit. Replaces the old + ``return _err(...)`` returns inside the per-engine run functions.""" + + def __init__(self, msg: str, *, code: int) -> None: + super().__init__(msg) + self.msg = msg + self.code = code + + +class GuardRefused(Exception): + """A guard refusal short-circuiting the envelope. ``envelope`` is the JSON error object the + caller must emit (the read-only guard's ``permission`` refusal), or ``None`` when the refusal + JSON was already written to stderr by ``_model_safety`` (carry only the exit ``code``).""" + + def __init__(self, envelope: dict | None, *, code: int) -> None: + super().__init__() + self.envelope = envelope + self.code = code + + def _env_token(profile: str) -> str: """The env-var suffix for a datasource: the profile id upper-cased with every non-alphanumeric char folded to `_` (so `sales-pg` → `SALES_PG`, used as `DATASOURCE_URL__SALES_PG`).""" @@ -337,20 +385,23 @@ def _parse_dsn(dsn: str) -> dict[str, str]: def _require(creds: dict[str, str], *fields: str) -> None: + """Raise ``ExecutorError`` (not ``sys.exit``) when a required credential field is missing, so the + same check is safe in-process (a bad profile can't kill the server) and the subprocess ``main`` + still surfaces the identical stderr message + exit code 2.""" missing = [f for f in fields if not creds.get(f)] if missing: - sys.stderr.write( + raise ExecutorError( f"Credentials profile is missing required fields: {missing}. " - f"Edit /local/credentials and add them.\n" + f"Edit /local/credentials and add them.", + code=2, ) - sys.exit(2) -def _execute_postgres(creds: dict[str, str], sql: str) -> int: +def _run_postgres(creds: dict[str, str], sql: str) -> ExecResult: try: import psycopg2 # type: ignore except ImportError: - return _err("psycopg2 not installed. Run: pip install psycopg2-binary", code=3) + raise ExecutorError("psycopg2 not installed. Run: pip install psycopg2-binary", code=3) _require(creds, "host", "port", "user", "password", "database") try: conn = psycopg2.connect( @@ -363,7 +414,7 @@ def _execute_postgres(creds: dict[str, str], sql: str) -> int: connect_timeout=10, ) except Exception as e: - return _err(f"Postgres connect failed: {e}", code=4) + raise ExecutorError(f"Postgres connect failed: {e}", code=4) try: with conn: # A server-side (named) cursor so the row cap bounds TRANSFER, not just what we write: @@ -374,19 +425,19 @@ def _execute_postgres(creds: dict[str, str], sql: str) -> int: with conn.cursor(name="agami_bounded") as cur: cur.itersize = _resolve_row_cap() + 1 # server fetch batch = the bounded window cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"Postgres execution error: {e}", code=5) + raise ExecutorError(f"Postgres execution error: {e}", code=5) finally: conn.close() - return 0 + return result -def _execute_mysql(creds: dict[str, str], sql: str) -> int: +def _run_mysql(creds: dict[str, str], sql: str) -> ExecResult: try: import pymysql # type: ignore except ImportError: - return _err("pymysql not installed. Run: pip install pymysql", code=3) + raise ExecutorError("pymysql not installed. Run: pip install pymysql", code=3) _require(creds, "host", "port", "user", "password", "database") try: conn = pymysql.connect( @@ -400,31 +451,31 @@ def _execute_mysql(creds: dict[str, str], sql: str) -> int: autocommit=True, ) except Exception as e: - return _err(f"MySQL connect failed: {e}", code=4) + raise ExecutorError(f"MySQL connect failed: {e}", code=4) try: with conn.cursor() as cur: cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"MySQL execution error: {e}", code=5) + raise ExecutorError(f"MySQL execution error: {e}", code=5) finally: conn.close() - return 0 + return result -def _execute_snowflake(creds: dict[str, str], sql: str) -> int: +def _run_snowflake(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for Snowflake using snowflake-connector-python.""" try: import snowflake.connector # type: ignore except ImportError: - return _err( + raise ExecutorError( "snowflake-connector-python not installed. " "Run: pip install snowflake-connector-python", code=3, ) _require(creds, "account", "user") if not (creds.get("password") or creds.get("authenticator")): - return _err( + raise ExecutorError( "Snowflake profile is missing 'password' or 'authenticator'. " "Add one to /local/credentials.", code=2, @@ -441,22 +492,22 @@ def _execute_snowflake(creds: dict[str, str], sql: str) -> int: try: conn = snowflake.connector.connect(**conn_kwargs) except Exception as e: - return _err(f"Snowflake connect failed: {e}", code=4) + raise ExecutorError(f"Snowflake connect failed: {e}", code=4) try: cur = conn.cursor() cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"Snowflake execution error: {e}", code=5) + raise ExecutorError(f"Snowflake execution error: {e}", code=5) finally: try: conn.close() except Exception: pass - return 0 + return result -def _execute_bigquery(creds: dict[str, str], sql: str) -> int: +def _run_bigquery(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for BigQuery using google-cloud-bigquery. Required: `project`. One of: `service_account_path` (path to a JSON key @@ -469,7 +520,7 @@ def _execute_bigquery(creds: dict[str, str], sql: str) -> int: from google.cloud import bigquery # type: ignore from google.oauth2 import service_account # type: ignore except ImportError: - return _err( + raise ExecutorError( "google-cloud-bigquery not installed. " "Run: pip install google-cloud-bigquery", code=3, @@ -487,7 +538,7 @@ def _execute_bigquery(creds: dict[str, str], sql: str) -> int: if sa_path: sa_path_expanded = os.path.expanduser(sa_path) if not os.path.exists(sa_path_expanded): - return _err( + raise ExecutorError( f"service_account_path '{sa_path}' doesn't exist. " f"Point at the JSON key file you downloaded from GCP.", code=2, @@ -508,12 +559,12 @@ def _execute_bigquery(creds: dict[str, str], sql: str) -> int: ) client_kwargs["credentials"] = creds_obj except Exception as e: - return _err(f"BigQuery credentials load failed: {e}", code=2) + raise ExecutorError(f"BigQuery credentials load failed: {e}", code=2) try: client = bigquery.Client(**client_kwargs) except Exception as e: - return _err(f"BigQuery client init failed: {e}", code=4) + raise ExecutorError(f"BigQuery client init failed: {e}", code=4) # If `dataset` was set, prefix unqualified table references via the # default_dataset job config so the SQL can omit `..` @@ -531,55 +582,52 @@ def _execute_bigquery(creds: dict[str, str], sql: str) -> int: job = client.query(sql, job_config=job_config) else: job = client.query(sql) - # BigQuery has no DB-API cursor, so it can't funnel through `_write_cursor_csv`; apply the + # BigQuery has no DB-API cursor, so it can't funnel through `_collect_cursor`; apply the # same bounded-fetch cap here. `max_results=cap+1` bounds what the API returns (transfer), # and the (cap+1)th row flags truncation — the never-silent guarantee holds for BigQuery too. results = job.result(max_results=cap + 1) # waits for completion; raises on error except Exception as e: - return _err(f"BigQuery execution error: {e}", code=5) - - writer = csv.writer(sys.stdout) - if results.schema: - writer.writerow([f.name for f in results.schema]) - written = 0 - truncated = False - for row in results: - if written >= cap: - truncated = True - break - writer.writerow([row[i] for i in range(len(results.schema))]) - written += 1 - if truncated: - _flag_truncated(cap) - - return 0 - - -def _execute_sqlite(creds: dict[str, str], sql: str) -> int: + raise ExecutorError(f"BigQuery execution error: {e}", code=5) + + if not results.schema: + return ExecResult(columns=[], rows=[], truncated=False) + columns = [f.name for f in results.schema] + ncols = len(results.schema) + rows: list[tuple] = [] + truncated = False + for row in results: + if len(rows) >= cap: + truncated = True + break + rows.append(tuple(row[i] for i in range(ncols))) + return ExecResult(columns=columns, rows=rows, truncated=truncated) + + +def _run_sqlite(creds: dict[str, str], sql: str) -> ExecResult: import sqlite3 # always available in stdlib _require(creds, "path") path = os.path.expanduser(creds["path"]) try: conn = sqlite3.connect(path) except Exception as e: - return _err(f"SQLite connect failed: {e}", code=4) + raise ExecutorError(f"SQLite connect failed: {e}", code=4) try: cur = conn.cursor() cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"SQLite execution error: {e}", code=5) + raise ExecutorError(f"SQLite execution error: {e}", code=5) finally: conn.close() - return 0 + return result -def _execute_sqlserver(creds: dict[str, str], sql: str) -> int: +def _run_sqlserver(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for SQL Server / Azure SQL using pymssql.""" try: import pymssql # type: ignore except ImportError: - return _err("pymssql not installed. Run: pip install pymssql", code=3) + raise ExecutorError("pymssql not installed. Run: pip install pymssql", code=3) _require(creds, "host", "user", "password") try: conn = pymssql.connect( @@ -588,27 +636,27 @@ def _execute_sqlserver(creds: dict[str, str], sql: str) -> int: database=creds.get("database", ""), login_timeout=15, ) except Exception as e: - return _err(f"SQL Server connect failed: {e}", code=4) + raise ExecutorError(f"SQL Server connect failed: {e}", code=4) try: cur = conn.cursor() cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"SQL Server execution error: {e}", code=5) + raise ExecutorError(f"SQL Server execution error: {e}", code=5) finally: try: conn.close() except Exception: pass - return 0 + return result -def _execute_oracle(creds: dict[str, str], sql: str) -> int: +def _run_oracle(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for Oracle using python-oracledb (thin mode — no client libs).""" try: import oracledb # type: ignore except ImportError: - return _err("python-oracledb not installed. Run: pip install oracledb", code=3) + raise ExecutorError("python-oracledb not installed. Run: pip install oracledb", code=3) _require(creds, "user", "password") dsn = creds.get("dsn") or creds.get("url") if not dsn: @@ -618,27 +666,27 @@ def _execute_oracle(creds: dict[str, str], sql: str) -> int: try: conn = oracledb.connect(user=creds["user"], password=creds["password"], dsn=dsn) except Exception as e: - return _err(f"Oracle connect failed: {e}", code=4) + raise ExecutorError(f"Oracle connect failed: {e}", code=4) try: cur = conn.cursor() cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"Oracle execution error: {e}", code=5) + raise ExecutorError(f"Oracle execution error: {e}", code=5) finally: try: conn.close() except Exception: pass - return 0 + return result -def _execute_databricks(creds: dict[str, str], sql: str) -> int: +def _run_databricks(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for Databricks SQL warehouses using databricks-sql-connector.""" try: from databricks import sql as dbsql # type: ignore except ImportError: - return _err( + raise ExecutorError( "databricks-sql-connector not installed. Run: pip install databricks-sql-connector", code=3, ) @@ -649,27 +697,27 @@ def _execute_databricks(creds: dict[str, str], sql: str) -> int: access_token=creds["token"], ) except Exception as e: - return _err(f"Databricks connect failed: {e}", code=4) + raise ExecutorError(f"Databricks connect failed: {e}", code=4) try: cur = conn.cursor() cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"Databricks execution error: {e}", code=5) + raise ExecutorError(f"Databricks execution error: {e}", code=5) finally: try: conn.close() except Exception: pass - return 0 + return result -def _execute_trino(creds: dict[str, str], sql: str) -> int: +def _run_trino(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for Trino / Presto using the trino python client.""" try: import trino # type: ignore except ImportError: - return _err("trino not installed. Run: pip install trino", code=3) + raise ExecutorError("trino not installed. Run: pip install trino", code=3) _require(creds, "host", "user") try: auth = None @@ -681,43 +729,43 @@ def _execute_trino(creds: dict[str, str], sql: str) -> int: http_scheme="https" if creds.get("password") else "http", auth=auth, ) except Exception as e: - return _err(f"Trino connect failed: {e}", code=4) + raise ExecutorError(f"Trino connect failed: {e}", code=4) try: cur = conn.cursor() cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"Trino execution error: {e}", code=5) + raise ExecutorError(f"Trino execution error: {e}", code=5) finally: try: conn.close() except Exception: pass - return 0 + return result -def _execute_duckdb(creds: dict[str, str], sql: str) -> int: +def _run_duckdb(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for DuckDB using the duckdb python module (file or in-memory).""" try: import duckdb # type: ignore except ImportError: - return _err("duckdb not installed. Run: pip install duckdb", code=3) + raise ExecutorError("duckdb not installed. Run: pip install duckdb", code=3) path = creds.get("path") or creds.get("database") or ":memory:" try: conn = duckdb.connect(path, read_only=True) except Exception as e: - return _err(f"DuckDB open failed: {e}", code=4) + raise ExecutorError(f"DuckDB open failed: {e}", code=4) try: cur = conn.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"DuckDB execution error: {e}", code=5) + raise ExecutorError(f"DuckDB execution error: {e}", code=5) finally: try: conn.close() except Exception: pass - return 0 + return result _DEFAULT_MAX_ROWS = 1000 # rows materialized per result before truncation (ACE-038) @@ -746,19 +794,41 @@ def _flag_truncated(cap: int) -> None: sys.stderr.write(json.dumps({"truncated": {"row_cap": cap}}) + "\n") -def _write_cursor_csv(cur: Any) -> None: - """Stream at most the row cap to stdout as CSV. `fetchmany(cap + 1)` — never `fetchall` — so a - huge result can't be buffered whole; the (cap+1)th row means the result was truncated, flagged - on stderr. The SQL itself is untouched (no injected LIMIT).""" +def _collect_cursor(cur: Any) -> ExecResult: + """Fetch at most the row cap from a DB-API cursor into an ``ExecResult`` with **native types**. + `fetchmany(cap + 1)` — never `fetchall` — so a huge result can't be buffered whole; a (cap+1)th + row means the result was truncated. The SQL itself is untouched (no injected LIMIT). This is the + single bounded-fetch implementation both the CSV wire (`_write_cursor_csv`) and the in-process + executor path share, so the row cap is enforced once, identically, for every caller.""" cap = _resolve_row_cap() + if cur.description is None: + return ExecResult(columns=[], rows=[], truncated=False) + columns = [d[0] for d in cur.description] + fetched = cur.fetchmany(cap + 1) + truncated = len(fetched) > cap + return ExecResult(columns=columns, rows=[tuple(r) for r in fetched[:cap]], truncated=truncated) + + +def _emit_result_csv(result: ExecResult) -> None: + """Serialize an ``ExecResult`` to stdout as CSV — the subprocess/CLI wire. Byte-for-byte what the + old inline cursor→CSV writer produced: header row then data rows, and a truncation marker on + stderr when capped. This is the *single, final* text serialization for the fork path; the + in-process path skips it and returns the native rows straight to the tool edge.""" + if not result.columns: # cursor had no description → wrote nothing (e.g. a non-row statement) + return writer = csv.writer(sys.stdout) - if cur.description is not None: - writer.writerow([d[0] for d in cur.description]) - rows = cur.fetchmany(cap + 1) - for row in rows[:cap]: - writer.writerow(row) - if len(rows) > cap: - _flag_truncated(cap) + writer.writerow(result.columns) + for row in result.rows: + writer.writerow(row) + if result.truncated: + _flag_truncated(_resolve_row_cap()) + + +def _write_cursor_csv(cur: Any) -> None: + """Collect the bounded result and write it to stdout as CSV — the per-engine sink the subprocess + path uses. Kept as the thin composition ``_emit_result_csv(_collect_cursor(cur))`` so the fetch + bound and the CSV shape stay single-sourced (and the existing bounded-fetch tests still pin it).""" + _emit_result_csv(_collect_cursor(cur)) def _hosted() -> bool: @@ -909,6 +979,159 @@ def _model_safety(sql: str, profile: str, area: str | None): return sql, None +# --------------------------------------------------------------------------- +# Executor seam (AH-012): one guarded envelope, a swappable connect-and-run step +# --------------------------------------------------------------------------- +# +# `execute_guarded` is the single execution chokepoint: guard -> resolve datasource -> +# executor.execute(vetted_sql) -> return native rows. The built-in executor (`BUILTIN_EXECUTOR`) is +# the default connect-per-query path, unchanged; a consumer injects its own `ports.Executor` +# (pooled / RBAC / tunnelled) *behind* the same guard — no fork of the guard, per REQ-002/REQ-014. +# The subprocess `main` and the in-process MCP handler both go through `execute_guarded`, so the +# guard is applied identically and can't be bypassed. The per-engine `_execute_` CSV wrappers +# below are the subprocess/CLI adapter (they emit CSV + return an exit code); `_run_` is the +# shared connect-and-run that returns native rows to either caller. + + +def _emit_or_err(run: Callable[[], ExecResult]) -> int: + """Subprocess/CLI adapter over a ``_run_`` function: write its result to stdout as CSV and + return exit code 0, or translate an ``ExecutorError`` into the stderr message + exit code the CLI + contract documents (byte-identical to what the old ``_execute_`` emitted).""" + try: + _emit_result_csv(run()) + except ExecutorError as e: + return _err(e.msg, code=e.code) + return 0 + + +def _execute_postgres(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_postgres(creds, sql)) + + +def _execute_mysql(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_mysql(creds, sql)) + + +def _execute_snowflake(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_snowflake(creds, sql)) + + +def _execute_bigquery(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_bigquery(creds, sql)) + + +def _execute_sqlite(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_sqlite(creds, sql)) + + +def _execute_sqlserver(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_sqlserver(creds, sql)) + + +def _execute_oracle(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_oracle(creds, sql)) + + +def _execute_databricks(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_databricks(creds, sql)) + + +def _execute_trino(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_trino(creds, sql)) + + +def _execute_duckdb(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_duckdb(creds, sql)) + + +def _builtin_execute(vetted_sql: str, creds: dict[str, str], *, profile: str) -> ExecResult: + """The built-in connect-and-run: dispatch on the datasource type and return native rows. Same + per-engine behaviour as before (redshift/supabase ride the Postgres wire); only the row-emit + moved to the caller. Raises ``ExecutorError`` on an unknown/missing type or a driver/connect/run + failure. This is what ``BUILTIN_EXECUTOR.execute`` calls.""" + db_type = creds.get("type", "").lower() + if not db_type: + raise ExecutorError(f"Credentials profile [{profile}] is missing the 'type' field.", code=2) + if db_type == "postgres": + return _run_postgres(creds, vetted_sql) + if db_type == "redshift": + # Redshift speaks the Postgres wire protocol; psycopg2 connects fine. `_run_postgres` reads + # host/port/etc. directly, so the type field doesn't matter — only sslmode defaulting does. + if "sslmode" not in creds: + creds = {**creds, "sslmode": "require"} + return _run_postgres(creds, vetted_sql) + if db_type == "mysql": + return _run_mysql(creds, vetted_sql) + if db_type == "sqlite": + return _run_sqlite(creds, vetted_sql) + if db_type == "snowflake": + return _run_snowflake(creds, vetted_sql) + if db_type == "bigquery": + return _run_bigquery(creds, vetted_sql) + if db_type in ("sqlserver", "mssql"): + return _run_sqlserver(creds, vetted_sql) + if db_type == "oracle": + return _run_oracle(creds, vetted_sql) + if db_type == "databricks": + return _run_databricks(creds, vetted_sql) + if db_type in ("trino", "presto"): + return _run_trino(creds, vetted_sql) + if db_type == "duckdb": + return _run_duckdb(creds, vetted_sql) + if db_type == "supabase": + # Supabase is hosted Postgres. + return _run_postgres(creds, vetted_sql) + raise ExecutorError( + f"Unsupported db type {db_type!r}. Supported: postgres, supabase, redshift, " + f"mysql, sqlite, snowflake, bigquery, sqlserver, oracle, databricks, trino, duckdb.", + code=2, + ) + + +class _BuiltinExecutor: + """The default ``ports.Executor``: wraps the connect-per-query dispatch as an object so it + satisfies the port by shape (method-style, like the other four ports). Stateless — one shared + ``BUILTIN_EXECUTOR`` instance.""" + + def execute(self, vetted_sql: str, creds: dict[str, str], *, profile: str) -> ExecResult: + return _builtin_execute(vetted_sql, creds, profile=profile) + + +BUILTIN_EXECUTOR = _BuiltinExecutor() + + +def execute_guarded( + sql: str, + profile: str, + area: str | None, + *, + executor: Executor, + no_safety: bool = False, +) -> ExecResult: + """The un-bypassable guarded envelope — the single execution chokepoint (REQ-002/REQ-014). + + In fixed order: read-only / dangerous-SQL guard (the hard security gate — NOT bypassable via + ``no_safety``, which skips only the semantic-model pass, never write/RCE/DoS protection) -> + semantic-model safety pass (fan/chasm pre-flight + scope + PII + ``default_filters`` rewrite) -> + resolve the datasource -> ``executor.execute(vetted_sql, …)``. The executor only ever receives + SQL both guards have passed. Raises ``GuardRefused`` on a refusal (the read-only refusal carries + its JSON envelope for the caller to emit; a model-safety refusal already wrote its JSON to stderr + and carries only the exit code) and ``ExecutorError`` on a connect/run failure — so the + subprocess ``main`` and the in-process MCP handler apply the same guard and surface errors + identically. The row cap rides the ``_max_rows_override`` module global the caller sets.""" + import sql_guard + + reason = sql_guard.check_read_only(sql) + if reason is not None: + raise GuardRefused({"error": {"kind": "permission", "remediation": reason}}, code=1) + if not no_safety: + sql, rc = _model_safety(sql, profile, area) + if rc is not None: + raise GuardRefused(None, code=rc) + creds = _load_credentials(profile) + return executor.execute(sql, creds, profile=profile) + + def main() -> int: # One-shot migration of a legacy /local into /local/, then re-resolve # the paths (the migration can set the artifacts-dir pointer to a custom location). @@ -946,64 +1169,26 @@ def main() -> int: profile = args.profile or _resolve_default_profile() - # Read-only / dangerous-SQL guard — the hard security gate, at the shared executor - # chokepoint so EVERY caller (both MCP servers, the agami-query skill, cron) is - # protected, not just whichever path happened to pre-check. This is NOT bypassable - # via --no-safety: that flag skips only the *semantic-model* pass (fan/chasm + - # default_filters), never write / RCE / DoS protection. Same gate the MCP tool layer - # fail-fast pre-checks (tools.check_read_only -> sql_guard). - import sql_guard - - guard_reason = sql_guard.check_read_only(sql) - if guard_reason is not None: - json.dump({"error": {"kind": "permission", "remediation": guard_reason}}, sys.stderr) - sys.stderr.write("\n") - return 1 - - # Semantic-model safety pass (fan/chasm pre-flight + default_filters). Inert when - # there's no model for the profile, so this is safe for every caller. - if not args.no_safety: - sql, _rc = _model_safety(sql, profile, args.area) - if _rc is not None: - return _rc - creds = _load_credentials(profile) - db_type = creds.get("type", "").lower() - if not db_type: - return _err(f"Credentials profile [{profile}] is missing the 'type' field.") - if db_type == "postgres": - return _execute_postgres(creds, sql) - if db_type == "redshift": - # Redshift speaks Postgres wire protocol; psycopg2 connects fine. - # The credentials dict has type=redshift, but _execute_postgres reads - # host/port/etc. directly so the type field doesn't matter. - if "sslmode" not in creds: - creds = {**creds, "sslmode": "require"} - return _execute_postgres(creds, sql) - if db_type == "mysql": - return _execute_mysql(creds, sql) - if db_type == "sqlite": - return _execute_sqlite(creds, sql) - if db_type == "snowflake": - return _execute_snowflake(creds, sql) - if db_type == "bigquery": - return _execute_bigquery(creds, sql) - if db_type in ("sqlserver", "mssql"): - return _execute_sqlserver(creds, sql) - if db_type == "oracle": - return _execute_oracle(creds, sql) - if db_type == "databricks": - return _execute_databricks(creds, sql) - if db_type in ("trino", "presto"): - return _execute_trino(creds, sql) - if db_type == "duckdb": - return _execute_duckdb(creds, sql) - if db_type == "supabase": - # Supabase is hosted Postgres. - return _execute_postgres(creds, sql) - return _err( - f"Unsupported db type {db_type!r}. Supported: postgres, supabase, redshift, " - f"mysql, sqlite, snowflake, bigquery, sqlserver, oracle, databricks, trino, duckdb." - ) + # Route through the single guarded envelope with the built-in executor: guard -> model-safety -> + # resolve -> connect-and-run, returning native rows we then serialize to stdout as CSV (the + # subprocess wire). Same guard, same verdicts, same connect-per-query behaviour as before — the + # split just makes the connect-and-run step swappable in-process (AH-012). The guard is the hard + # security gate for EVERY caller (both MCP servers, the agami-query skill, cron), NOT bypassable + # via --no-safety (which skips only the semantic-model pass, never write/RCE/DoS protection). + try: + result = execute_guarded( + sql, profile, args.area, executor=BUILTIN_EXECUTOR, no_safety=args.no_safety + ) + except GuardRefused as refusal: + if refusal.envelope is not None: # read-only refusal: emit its JSON (model-safety already did) + json.dump(refusal.envelope, sys.stderr) + sys.stderr.write("\n") + return refusal.code + except ExecutorError as exc: + sys.stderr.write(f"{exc.msg}\n") + return exc.code + _emit_result_csv(result) + return 0 if __name__ == "__main__": diff --git a/packages/agami-core/src/ports.py b/packages/agami-core/src/ports.py index aff2d65d..cffb3983 100644 --- a/packages/agami-core/src/ports.py +++ b/packages/agami-core/src/ports.py @@ -1,4 +1,4 @@ -"""The four port Protocols — the seams adapters plug into. +"""The five port Protocols — the seams adapters plug into. agami-core keeps one MCP implementation across deployments; deployment-specific behavior is swapped at the composition root through these ports, never by forking a tool: @@ -7,6 +7,8 @@ - ``OrgResolver`` — single vs multi tenancy as a config flag, not a schema fork - ``AuthProvider`` — bearer token → principal (presence by default) - ``GovernancePolicy`` — warn-only by default; enforcement is a paid concern + - ``Executor`` — the connect-and-run step, *behind* the shared guard (built-in by default; + a consumer injects a pooled/RBAC/tunnel executor without forking the guard) These are **interfaces only** — `typing.Protocol`, so an adapter satisfies a port by shape, with no import coupling back to core. The OSS default adapters live in ``oss_adapters`` (so the local @@ -31,6 +33,12 @@ # @runtime_checkable only checks method *names*, so isinstance() works without these. from contracts import QueryExecutionRecord + # ``ExecResult`` is defined in ``execute_sql`` (not here): it is the executor's result type and + # ``execute_sql`` ships in the stdlib-lean plugin mirror that does NOT include this module, so it + # cannot import ``ports`` at runtime. Referencing it under TYPE_CHECKING keeps the ``Executor`` + # annotation resolvable for type-checkers without a runtime import cycle. + from execute_sql import ExecResult + # --------------------------------------------------------------------------- # Seam value types (minimal — a consumer extends them when it needs more) # --------------------------------------------------------------------------- @@ -63,7 +71,7 @@ class GovernanceVerdict: # --------------------------------------------------------------------------- -# The four ports +# The five ports # --------------------------------------------------------------------------- @@ -104,8 +112,23 @@ class GovernancePolicy(Protocol): def evaluate(self, ctx: object | None = None) -> GovernanceVerdict: ... +@runtime_checkable +class Executor(Protocol): + """Connect to a datasource and run **already-vetted** SQL — the only swappable part of the + execution path. It runs *inside* the guarded envelope (guard → executor → shape/log), so it + **only ever receives SQL the guard already passed**, never raw user input; it does no guarding, + logging, or governance itself. This is the seam a hosted consumer overrides to supply a + pooled / per-user-RBAC / SSH-tunnel executor **behind agami-core's one guard** — no fork. + + ``profile`` is the datasource identity a pooling executor keys its reused connection on. The + built-in OSS default (subprocess/direct connect-per-query) implements this same shape, so a + plain deploy is unchanged.""" + + def execute(self, vetted_sql: str, creds: dict[str, str], *, profile: str) -> ExecResult: ... + + # --------------------------------------------------------------------------- -# Composition-root container — the four adapters passed as one argument +# Composition-root container — the adapters passed as one argument # --------------------------------------------------------------------------- @@ -113,13 +136,19 @@ def evaluate(self, ctx: object | None = None) -> GovernanceVerdict: ... class Adapters: """The four port adapters, bundled so ``mcp_http.create_app`` takes them as one argument. - A consumer builds this with its own implementations of the four ports (its own ``OrgResolver``, - ``AuthProvider``, ``ActivitySink``, and ``GovernancePolicy``); passing ``adapters=None`` to - ``create_app`` uses the OSS defaults (``mcp_http.default_adapters``). Today ``create_app`` wires - ``auth_provider`` + ``org_resolver`` into the request path; ``activity_sink`` + ``governance`` - are carried here for consumers and not yet referenced by a core call site.""" + A consumer builds this with its own implementations of the ports (its own ``OrgResolver``, + ``AuthProvider``, ``ActivitySink``, ``GovernancePolicy``, and optionally an ``Executor``); + passing ``adapters=None`` to ``create_app`` uses the OSS defaults (``mcp_http.default_adapters``). + Today ``create_app`` wires ``auth_provider`` + ``org_resolver`` into the request path; + ``activity_sink`` + ``governance`` are carried here for consumers and not yet referenced by a + core call site. + + ``executor`` is optional and defaults to ``None`` — meaning "use the built-in executor" (the + subprocess/direct connect-per-query path, byte-identical to today). A consumer sets it to run + execution in-process behind the shared guard (see ``tools.tool_execute_sql``).""" activity_sink: ActivitySink org_resolver: OrgResolver auth_provider: AuthProvider governance: GovernancePolicy + executor: Executor | None = None diff --git a/plugins/agami/lib/execute_sql.py b/plugins/agami/lib/execute_sql.py index 44222f5e..e791f660 100644 --- a/plugins/agami/lib/execute_sql.py +++ b/plugins/agami/lib/execute_sql.py @@ -46,11 +46,19 @@ import stat import sys import urllib.parse +from collections.abc import Callable +from dataclasses import dataclass from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any import agami_paths +if TYPE_CHECKING: + # ``Executor`` is the 5th port; imported only for type-checkers. At runtime ``execute_sql`` never + # imports ``ports`` (it ships in the stdlib-lean plugin mirror without it), so the annotation on + # ``execute_guarded`` stays a lazy string (``from __future__ import annotations``). + from ports import Executor + # Credentials + config now live under /local/ (the consolidated, # gitignored replacement for ~/.agami). The path is stable regardless of migration # timing — bootstrap() just moves the files into it. See agami_paths. @@ -87,6 +95,46 @@ def _err(msg: str, *, code: int = 2) -> int: return code +@dataclass(frozen=True) +class ExecResult: + """What an executor returns: columns + rows with **native Python types preserved** (ints, + Decimals, datetimes, ``None``), not stringified. Serializing to text — CSV for the subprocess + wire, JSON at the MCP-tool edge — is the *caller's* single, final step, so an in-process + executor never pays a serialize→re-parse round-trip and never loses a type or confuses NULL + with "". ``truncated`` mirrors the ``fetchmany(cap + 1)`` bound: True when the result was capped. + + This lives here (not in ``ports``) because ``execute_sql`` ships in the stdlib-lean plugin + mirror, which does not include ``ports``; ``ports.Executor`` references it under TYPE_CHECKING. + """ + + columns: list[str] + rows: list[tuple] + truncated: bool = False + + +class ExecutorError(Exception): + """A connect / credential / run failure raised by the built-in executor. Carries the exact + stderr message and exit code the subprocess CLI emits, so ``main`` reproduces today's bytes and + the in-process caller gets a catchable error instead of a process exit. Replaces the old + ``return _err(...)`` returns inside the per-engine run functions.""" + + def __init__(self, msg: str, *, code: int) -> None: + super().__init__(msg) + self.msg = msg + self.code = code + + +class GuardRefused(Exception): + """A guard refusal short-circuiting the envelope. ``envelope`` is the JSON error object the + caller must emit (the read-only guard's ``permission`` refusal), or ``None`` when the refusal + JSON was already written to stderr by ``_model_safety`` (carry only the exit ``code``).""" + + def __init__(self, envelope: dict | None, *, code: int) -> None: + super().__init__() + self.envelope = envelope + self.code = code + + def _env_token(profile: str) -> str: """The env-var suffix for a datasource: the profile id upper-cased with every non-alphanumeric char folded to `_` (so `sales-pg` → `SALES_PG`, used as `DATASOURCE_URL__SALES_PG`).""" @@ -337,20 +385,23 @@ def _parse_dsn(dsn: str) -> dict[str, str]: def _require(creds: dict[str, str], *fields: str) -> None: + """Raise ``ExecutorError`` (not ``sys.exit``) when a required credential field is missing, so the + same check is safe in-process (a bad profile can't kill the server) and the subprocess ``main`` + still surfaces the identical stderr message + exit code 2.""" missing = [f for f in fields if not creds.get(f)] if missing: - sys.stderr.write( + raise ExecutorError( f"Credentials profile is missing required fields: {missing}. " - f"Edit /local/credentials and add them.\n" + f"Edit /local/credentials and add them.", + code=2, ) - sys.exit(2) -def _execute_postgres(creds: dict[str, str], sql: str) -> int: +def _run_postgres(creds: dict[str, str], sql: str) -> ExecResult: try: import psycopg2 # type: ignore except ImportError: - return _err("psycopg2 not installed. Run: pip install psycopg2-binary", code=3) + raise ExecutorError("psycopg2 not installed. Run: pip install psycopg2-binary", code=3) _require(creds, "host", "port", "user", "password", "database") try: conn = psycopg2.connect( @@ -363,7 +414,7 @@ def _execute_postgres(creds: dict[str, str], sql: str) -> int: connect_timeout=10, ) except Exception as e: - return _err(f"Postgres connect failed: {e}", code=4) + raise ExecutorError(f"Postgres connect failed: {e}", code=4) try: with conn: # A server-side (named) cursor so the row cap bounds TRANSFER, not just what we write: @@ -374,19 +425,19 @@ def _execute_postgres(creds: dict[str, str], sql: str) -> int: with conn.cursor(name="agami_bounded") as cur: cur.itersize = _resolve_row_cap() + 1 # server fetch batch = the bounded window cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"Postgres execution error: {e}", code=5) + raise ExecutorError(f"Postgres execution error: {e}", code=5) finally: conn.close() - return 0 + return result -def _execute_mysql(creds: dict[str, str], sql: str) -> int: +def _run_mysql(creds: dict[str, str], sql: str) -> ExecResult: try: import pymysql # type: ignore except ImportError: - return _err("pymysql not installed. Run: pip install pymysql", code=3) + raise ExecutorError("pymysql not installed. Run: pip install pymysql", code=3) _require(creds, "host", "port", "user", "password", "database") try: conn = pymysql.connect( @@ -400,31 +451,31 @@ def _execute_mysql(creds: dict[str, str], sql: str) -> int: autocommit=True, ) except Exception as e: - return _err(f"MySQL connect failed: {e}", code=4) + raise ExecutorError(f"MySQL connect failed: {e}", code=4) try: with conn.cursor() as cur: cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"MySQL execution error: {e}", code=5) + raise ExecutorError(f"MySQL execution error: {e}", code=5) finally: conn.close() - return 0 + return result -def _execute_snowflake(creds: dict[str, str], sql: str) -> int: +def _run_snowflake(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for Snowflake using snowflake-connector-python.""" try: import snowflake.connector # type: ignore except ImportError: - return _err( + raise ExecutorError( "snowflake-connector-python not installed. " "Run: pip install snowflake-connector-python", code=3, ) _require(creds, "account", "user") if not (creds.get("password") or creds.get("authenticator")): - return _err( + raise ExecutorError( "Snowflake profile is missing 'password' or 'authenticator'. " "Add one to /local/credentials.", code=2, @@ -441,22 +492,22 @@ def _execute_snowflake(creds: dict[str, str], sql: str) -> int: try: conn = snowflake.connector.connect(**conn_kwargs) except Exception as e: - return _err(f"Snowflake connect failed: {e}", code=4) + raise ExecutorError(f"Snowflake connect failed: {e}", code=4) try: cur = conn.cursor() cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"Snowflake execution error: {e}", code=5) + raise ExecutorError(f"Snowflake execution error: {e}", code=5) finally: try: conn.close() except Exception: pass - return 0 + return result -def _execute_bigquery(creds: dict[str, str], sql: str) -> int: +def _run_bigquery(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for BigQuery using google-cloud-bigquery. Required: `project`. One of: `service_account_path` (path to a JSON key @@ -469,7 +520,7 @@ def _execute_bigquery(creds: dict[str, str], sql: str) -> int: from google.cloud import bigquery # type: ignore from google.oauth2 import service_account # type: ignore except ImportError: - return _err( + raise ExecutorError( "google-cloud-bigquery not installed. " "Run: pip install google-cloud-bigquery", code=3, @@ -487,7 +538,7 @@ def _execute_bigquery(creds: dict[str, str], sql: str) -> int: if sa_path: sa_path_expanded = os.path.expanduser(sa_path) if not os.path.exists(sa_path_expanded): - return _err( + raise ExecutorError( f"service_account_path '{sa_path}' doesn't exist. " f"Point at the JSON key file you downloaded from GCP.", code=2, @@ -508,12 +559,12 @@ def _execute_bigquery(creds: dict[str, str], sql: str) -> int: ) client_kwargs["credentials"] = creds_obj except Exception as e: - return _err(f"BigQuery credentials load failed: {e}", code=2) + raise ExecutorError(f"BigQuery credentials load failed: {e}", code=2) try: client = bigquery.Client(**client_kwargs) except Exception as e: - return _err(f"BigQuery client init failed: {e}", code=4) + raise ExecutorError(f"BigQuery client init failed: {e}", code=4) # If `dataset` was set, prefix unqualified table references via the # default_dataset job config so the SQL can omit `..` @@ -531,55 +582,52 @@ def _execute_bigquery(creds: dict[str, str], sql: str) -> int: job = client.query(sql, job_config=job_config) else: job = client.query(sql) - # BigQuery has no DB-API cursor, so it can't funnel through `_write_cursor_csv`; apply the + # BigQuery has no DB-API cursor, so it can't funnel through `_collect_cursor`; apply the # same bounded-fetch cap here. `max_results=cap+1` bounds what the API returns (transfer), # and the (cap+1)th row flags truncation — the never-silent guarantee holds for BigQuery too. results = job.result(max_results=cap + 1) # waits for completion; raises on error except Exception as e: - return _err(f"BigQuery execution error: {e}", code=5) - - writer = csv.writer(sys.stdout) - if results.schema: - writer.writerow([f.name for f in results.schema]) - written = 0 - truncated = False - for row in results: - if written >= cap: - truncated = True - break - writer.writerow([row[i] for i in range(len(results.schema))]) - written += 1 - if truncated: - _flag_truncated(cap) - - return 0 - - -def _execute_sqlite(creds: dict[str, str], sql: str) -> int: + raise ExecutorError(f"BigQuery execution error: {e}", code=5) + + if not results.schema: + return ExecResult(columns=[], rows=[], truncated=False) + columns = [f.name for f in results.schema] + ncols = len(results.schema) + rows: list[tuple] = [] + truncated = False + for row in results: + if len(rows) >= cap: + truncated = True + break + rows.append(tuple(row[i] for i in range(ncols))) + return ExecResult(columns=columns, rows=rows, truncated=truncated) + + +def _run_sqlite(creds: dict[str, str], sql: str) -> ExecResult: import sqlite3 # always available in stdlib _require(creds, "path") path = os.path.expanduser(creds["path"]) try: conn = sqlite3.connect(path) except Exception as e: - return _err(f"SQLite connect failed: {e}", code=4) + raise ExecutorError(f"SQLite connect failed: {e}", code=4) try: cur = conn.cursor() cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"SQLite execution error: {e}", code=5) + raise ExecutorError(f"SQLite execution error: {e}", code=5) finally: conn.close() - return 0 + return result -def _execute_sqlserver(creds: dict[str, str], sql: str) -> int: +def _run_sqlserver(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for SQL Server / Azure SQL using pymssql.""" try: import pymssql # type: ignore except ImportError: - return _err("pymssql not installed. Run: pip install pymssql", code=3) + raise ExecutorError("pymssql not installed. Run: pip install pymssql", code=3) _require(creds, "host", "user", "password") try: conn = pymssql.connect( @@ -588,27 +636,27 @@ def _execute_sqlserver(creds: dict[str, str], sql: str) -> int: database=creds.get("database", ""), login_timeout=15, ) except Exception as e: - return _err(f"SQL Server connect failed: {e}", code=4) + raise ExecutorError(f"SQL Server connect failed: {e}", code=4) try: cur = conn.cursor() cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"SQL Server execution error: {e}", code=5) + raise ExecutorError(f"SQL Server execution error: {e}", code=5) finally: try: conn.close() except Exception: pass - return 0 + return result -def _execute_oracle(creds: dict[str, str], sql: str) -> int: +def _run_oracle(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for Oracle using python-oracledb (thin mode — no client libs).""" try: import oracledb # type: ignore except ImportError: - return _err("python-oracledb not installed. Run: pip install oracledb", code=3) + raise ExecutorError("python-oracledb not installed. Run: pip install oracledb", code=3) _require(creds, "user", "password") dsn = creds.get("dsn") or creds.get("url") if not dsn: @@ -618,27 +666,27 @@ def _execute_oracle(creds: dict[str, str], sql: str) -> int: try: conn = oracledb.connect(user=creds["user"], password=creds["password"], dsn=dsn) except Exception as e: - return _err(f"Oracle connect failed: {e}", code=4) + raise ExecutorError(f"Oracle connect failed: {e}", code=4) try: cur = conn.cursor() cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"Oracle execution error: {e}", code=5) + raise ExecutorError(f"Oracle execution error: {e}", code=5) finally: try: conn.close() except Exception: pass - return 0 + return result -def _execute_databricks(creds: dict[str, str], sql: str) -> int: +def _run_databricks(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for Databricks SQL warehouses using databricks-sql-connector.""" try: from databricks import sql as dbsql # type: ignore except ImportError: - return _err( + raise ExecutorError( "databricks-sql-connector not installed. Run: pip install databricks-sql-connector", code=3, ) @@ -649,27 +697,27 @@ def _execute_databricks(creds: dict[str, str], sql: str) -> int: access_token=creds["token"], ) except Exception as e: - return _err(f"Databricks connect failed: {e}", code=4) + raise ExecutorError(f"Databricks connect failed: {e}", code=4) try: cur = conn.cursor() cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"Databricks execution error: {e}", code=5) + raise ExecutorError(f"Databricks execution error: {e}", code=5) finally: try: conn.close() except Exception: pass - return 0 + return result -def _execute_trino(creds: dict[str, str], sql: str) -> int: +def _run_trino(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for Trino / Presto using the trino python client.""" try: import trino # type: ignore except ImportError: - return _err("trino not installed. Run: pip install trino", code=3) + raise ExecutorError("trino not installed. Run: pip install trino", code=3) _require(creds, "host", "user") try: auth = None @@ -681,43 +729,43 @@ def _execute_trino(creds: dict[str, str], sql: str) -> int: http_scheme="https" if creds.get("password") else "http", auth=auth, ) except Exception as e: - return _err(f"Trino connect failed: {e}", code=4) + raise ExecutorError(f"Trino connect failed: {e}", code=4) try: cur = conn.cursor() cur.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"Trino execution error: {e}", code=5) + raise ExecutorError(f"Trino execution error: {e}", code=5) finally: try: conn.close() except Exception: pass - return 0 + return result -def _execute_duckdb(creds: dict[str, str], sql: str) -> int: +def _run_duckdb(creds: dict[str, str], sql: str) -> ExecResult: """Tier-3 path for DuckDB using the duckdb python module (file or in-memory).""" try: import duckdb # type: ignore except ImportError: - return _err("duckdb not installed. Run: pip install duckdb", code=3) + raise ExecutorError("duckdb not installed. Run: pip install duckdb", code=3) path = creds.get("path") or creds.get("database") or ":memory:" try: conn = duckdb.connect(path, read_only=True) except Exception as e: - return _err(f"DuckDB open failed: {e}", code=4) + raise ExecutorError(f"DuckDB open failed: {e}", code=4) try: cur = conn.execute(sql) - _write_cursor_csv(cur) + result = _collect_cursor(cur) except Exception as e: - return _err(f"DuckDB execution error: {e}", code=5) + raise ExecutorError(f"DuckDB execution error: {e}", code=5) finally: try: conn.close() except Exception: pass - return 0 + return result _DEFAULT_MAX_ROWS = 1000 # rows materialized per result before truncation (ACE-038) @@ -746,19 +794,41 @@ def _flag_truncated(cap: int) -> None: sys.stderr.write(json.dumps({"truncated": {"row_cap": cap}}) + "\n") -def _write_cursor_csv(cur: Any) -> None: - """Stream at most the row cap to stdout as CSV. `fetchmany(cap + 1)` — never `fetchall` — so a - huge result can't be buffered whole; the (cap+1)th row means the result was truncated, flagged - on stderr. The SQL itself is untouched (no injected LIMIT).""" +def _collect_cursor(cur: Any) -> ExecResult: + """Fetch at most the row cap from a DB-API cursor into an ``ExecResult`` with **native types**. + `fetchmany(cap + 1)` — never `fetchall` — so a huge result can't be buffered whole; a (cap+1)th + row means the result was truncated. The SQL itself is untouched (no injected LIMIT). This is the + single bounded-fetch implementation both the CSV wire (`_write_cursor_csv`) and the in-process + executor path share, so the row cap is enforced once, identically, for every caller.""" cap = _resolve_row_cap() + if cur.description is None: + return ExecResult(columns=[], rows=[], truncated=False) + columns = [d[0] for d in cur.description] + fetched = cur.fetchmany(cap + 1) + truncated = len(fetched) > cap + return ExecResult(columns=columns, rows=[tuple(r) for r in fetched[:cap]], truncated=truncated) + + +def _emit_result_csv(result: ExecResult) -> None: + """Serialize an ``ExecResult`` to stdout as CSV — the subprocess/CLI wire. Byte-for-byte what the + old inline cursor→CSV writer produced: header row then data rows, and a truncation marker on + stderr when capped. This is the *single, final* text serialization for the fork path; the + in-process path skips it and returns the native rows straight to the tool edge.""" + if not result.columns: # cursor had no description → wrote nothing (e.g. a non-row statement) + return writer = csv.writer(sys.stdout) - if cur.description is not None: - writer.writerow([d[0] for d in cur.description]) - rows = cur.fetchmany(cap + 1) - for row in rows[:cap]: - writer.writerow(row) - if len(rows) > cap: - _flag_truncated(cap) + writer.writerow(result.columns) + for row in result.rows: + writer.writerow(row) + if result.truncated: + _flag_truncated(_resolve_row_cap()) + + +def _write_cursor_csv(cur: Any) -> None: + """Collect the bounded result and write it to stdout as CSV — the per-engine sink the subprocess + path uses. Kept as the thin composition ``_emit_result_csv(_collect_cursor(cur))`` so the fetch + bound and the CSV shape stay single-sourced (and the existing bounded-fetch tests still pin it).""" + _emit_result_csv(_collect_cursor(cur)) def _hosted() -> bool: @@ -909,6 +979,159 @@ def _model_safety(sql: str, profile: str, area: str | None): return sql, None +# --------------------------------------------------------------------------- +# Executor seam (AH-012): one guarded envelope, a swappable connect-and-run step +# --------------------------------------------------------------------------- +# +# `execute_guarded` is the single execution chokepoint: guard -> resolve datasource -> +# executor.execute(vetted_sql) -> return native rows. The built-in executor (`BUILTIN_EXECUTOR`) is +# the default connect-per-query path, unchanged; a consumer injects its own `ports.Executor` +# (pooled / RBAC / tunnelled) *behind* the same guard — no fork of the guard, per REQ-002/REQ-014. +# The subprocess `main` and the in-process MCP handler both go through `execute_guarded`, so the +# guard is applied identically and can't be bypassed. The per-engine `_execute_` CSV wrappers +# below are the subprocess/CLI adapter (they emit CSV + return an exit code); `_run_` is the +# shared connect-and-run that returns native rows to either caller. + + +def _emit_or_err(run: Callable[[], ExecResult]) -> int: + """Subprocess/CLI adapter over a ``_run_`` function: write its result to stdout as CSV and + return exit code 0, or translate an ``ExecutorError`` into the stderr message + exit code the CLI + contract documents (byte-identical to what the old ``_execute_`` emitted).""" + try: + _emit_result_csv(run()) + except ExecutorError as e: + return _err(e.msg, code=e.code) + return 0 + + +def _execute_postgres(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_postgres(creds, sql)) + + +def _execute_mysql(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_mysql(creds, sql)) + + +def _execute_snowflake(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_snowflake(creds, sql)) + + +def _execute_bigquery(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_bigquery(creds, sql)) + + +def _execute_sqlite(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_sqlite(creds, sql)) + + +def _execute_sqlserver(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_sqlserver(creds, sql)) + + +def _execute_oracle(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_oracle(creds, sql)) + + +def _execute_databricks(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_databricks(creds, sql)) + + +def _execute_trino(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_trino(creds, sql)) + + +def _execute_duckdb(creds: dict[str, str], sql: str) -> int: + return _emit_or_err(lambda: _run_duckdb(creds, sql)) + + +def _builtin_execute(vetted_sql: str, creds: dict[str, str], *, profile: str) -> ExecResult: + """The built-in connect-and-run: dispatch on the datasource type and return native rows. Same + per-engine behaviour as before (redshift/supabase ride the Postgres wire); only the row-emit + moved to the caller. Raises ``ExecutorError`` on an unknown/missing type or a driver/connect/run + failure. This is what ``BUILTIN_EXECUTOR.execute`` calls.""" + db_type = creds.get("type", "").lower() + if not db_type: + raise ExecutorError(f"Credentials profile [{profile}] is missing the 'type' field.", code=2) + if db_type == "postgres": + return _run_postgres(creds, vetted_sql) + if db_type == "redshift": + # Redshift speaks the Postgres wire protocol; psycopg2 connects fine. `_run_postgres` reads + # host/port/etc. directly, so the type field doesn't matter — only sslmode defaulting does. + if "sslmode" not in creds: + creds = {**creds, "sslmode": "require"} + return _run_postgres(creds, vetted_sql) + if db_type == "mysql": + return _run_mysql(creds, vetted_sql) + if db_type == "sqlite": + return _run_sqlite(creds, vetted_sql) + if db_type == "snowflake": + return _run_snowflake(creds, vetted_sql) + if db_type == "bigquery": + return _run_bigquery(creds, vetted_sql) + if db_type in ("sqlserver", "mssql"): + return _run_sqlserver(creds, vetted_sql) + if db_type == "oracle": + return _run_oracle(creds, vetted_sql) + if db_type == "databricks": + return _run_databricks(creds, vetted_sql) + if db_type in ("trino", "presto"): + return _run_trino(creds, vetted_sql) + if db_type == "duckdb": + return _run_duckdb(creds, vetted_sql) + if db_type == "supabase": + # Supabase is hosted Postgres. + return _run_postgres(creds, vetted_sql) + raise ExecutorError( + f"Unsupported db type {db_type!r}. Supported: postgres, supabase, redshift, " + f"mysql, sqlite, snowflake, bigquery, sqlserver, oracle, databricks, trino, duckdb.", + code=2, + ) + + +class _BuiltinExecutor: + """The default ``ports.Executor``: wraps the connect-per-query dispatch as an object so it + satisfies the port by shape (method-style, like the other four ports). Stateless — one shared + ``BUILTIN_EXECUTOR`` instance.""" + + def execute(self, vetted_sql: str, creds: dict[str, str], *, profile: str) -> ExecResult: + return _builtin_execute(vetted_sql, creds, profile=profile) + + +BUILTIN_EXECUTOR = _BuiltinExecutor() + + +def execute_guarded( + sql: str, + profile: str, + area: str | None, + *, + executor: Executor, + no_safety: bool = False, +) -> ExecResult: + """The un-bypassable guarded envelope — the single execution chokepoint (REQ-002/REQ-014). + + In fixed order: read-only / dangerous-SQL guard (the hard security gate — NOT bypassable via + ``no_safety``, which skips only the semantic-model pass, never write/RCE/DoS protection) -> + semantic-model safety pass (fan/chasm pre-flight + scope + PII + ``default_filters`` rewrite) -> + resolve the datasource -> ``executor.execute(vetted_sql, …)``. The executor only ever receives + SQL both guards have passed. Raises ``GuardRefused`` on a refusal (the read-only refusal carries + its JSON envelope for the caller to emit; a model-safety refusal already wrote its JSON to stderr + and carries only the exit code) and ``ExecutorError`` on a connect/run failure — so the + subprocess ``main`` and the in-process MCP handler apply the same guard and surface errors + identically. The row cap rides the ``_max_rows_override`` module global the caller sets.""" + import sql_guard + + reason = sql_guard.check_read_only(sql) + if reason is not None: + raise GuardRefused({"error": {"kind": "permission", "remediation": reason}}, code=1) + if not no_safety: + sql, rc = _model_safety(sql, profile, area) + if rc is not None: + raise GuardRefused(None, code=rc) + creds = _load_credentials(profile) + return executor.execute(sql, creds, profile=profile) + + def main() -> int: # One-shot migration of a legacy /local into /local/, then re-resolve # the paths (the migration can set the artifacts-dir pointer to a custom location). @@ -946,64 +1169,26 @@ def main() -> int: profile = args.profile or _resolve_default_profile() - # Read-only / dangerous-SQL guard — the hard security gate, at the shared executor - # chokepoint so EVERY caller (both MCP servers, the agami-query skill, cron) is - # protected, not just whichever path happened to pre-check. This is NOT bypassable - # via --no-safety: that flag skips only the *semantic-model* pass (fan/chasm + - # default_filters), never write / RCE / DoS protection. Same gate the MCP tool layer - # fail-fast pre-checks (tools.check_read_only -> sql_guard). - import sql_guard - - guard_reason = sql_guard.check_read_only(sql) - if guard_reason is not None: - json.dump({"error": {"kind": "permission", "remediation": guard_reason}}, sys.stderr) - sys.stderr.write("\n") - return 1 - - # Semantic-model safety pass (fan/chasm pre-flight + default_filters). Inert when - # there's no model for the profile, so this is safe for every caller. - if not args.no_safety: - sql, _rc = _model_safety(sql, profile, args.area) - if _rc is not None: - return _rc - creds = _load_credentials(profile) - db_type = creds.get("type", "").lower() - if not db_type: - return _err(f"Credentials profile [{profile}] is missing the 'type' field.") - if db_type == "postgres": - return _execute_postgres(creds, sql) - if db_type == "redshift": - # Redshift speaks Postgres wire protocol; psycopg2 connects fine. - # The credentials dict has type=redshift, but _execute_postgres reads - # host/port/etc. directly so the type field doesn't matter. - if "sslmode" not in creds: - creds = {**creds, "sslmode": "require"} - return _execute_postgres(creds, sql) - if db_type == "mysql": - return _execute_mysql(creds, sql) - if db_type == "sqlite": - return _execute_sqlite(creds, sql) - if db_type == "snowflake": - return _execute_snowflake(creds, sql) - if db_type == "bigquery": - return _execute_bigquery(creds, sql) - if db_type in ("sqlserver", "mssql"): - return _execute_sqlserver(creds, sql) - if db_type == "oracle": - return _execute_oracle(creds, sql) - if db_type == "databricks": - return _execute_databricks(creds, sql) - if db_type in ("trino", "presto"): - return _execute_trino(creds, sql) - if db_type == "duckdb": - return _execute_duckdb(creds, sql) - if db_type == "supabase": - # Supabase is hosted Postgres. - return _execute_postgres(creds, sql) - return _err( - f"Unsupported db type {db_type!r}. Supported: postgres, supabase, redshift, " - f"mysql, sqlite, snowflake, bigquery, sqlserver, oracle, databricks, trino, duckdb." - ) + # Route through the single guarded envelope with the built-in executor: guard -> model-safety -> + # resolve -> connect-and-run, returning native rows we then serialize to stdout as CSV (the + # subprocess wire). Same guard, same verdicts, same connect-per-query behaviour as before — the + # split just makes the connect-and-run step swappable in-process (AH-012). The guard is the hard + # security gate for EVERY caller (both MCP servers, the agami-query skill, cron), NOT bypassable + # via --no-safety (which skips only the semantic-model pass, never write/RCE/DoS protection). + try: + result = execute_guarded( + sql, profile, args.area, executor=BUILTIN_EXECUTOR, no_safety=args.no_safety + ) + except GuardRefused as refusal: + if refusal.envelope is not None: # read-only refusal: emit its JSON (model-safety already did) + json.dump(refusal.envelope, sys.stderr) + sys.stderr.write("\n") + return refusal.code + except ExecutorError as exc: + sys.stderr.write(f"{exc.msg}\n") + return exc.code + _emit_result_csv(result) + return 0 if __name__ == "__main__": From 5b15f9cd7fafd3e2e183b26958e73e1648c3ddb4 Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Sun, 12 Jul 2026 15:44:53 +0530 Subject: [PATCH 2/9] =?UTF-8?q?test(execute=5Fsql):=20AH-012=20slice=201?= =?UTF-8?q?=20=E2=80=94=20guard-before-executor,=20vetted-only,=20native?= =?UTF-8?q?=20rows?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pin the seam invariants: read-only + model-safety guards refuse before the executor is reached (un-bypassable), the executor only ever gets vetted SQL + resolved creds, --no-safety skips only the model pass, and the built-in executor returns native-typed rows (NULL as None) while the CLI wire still emits byte-identical CSV. Spec: AH-012 --- tests/test_ah012_executor_seam.py | 163 ++++++++++++++++++++++++++++++ 1 file changed, 163 insertions(+) create mode 100644 tests/test_ah012_executor_seam.py diff --git a/tests/test_ah012_executor_seam.py b/tests/test_ah012_executor_seam.py new file mode 100644 index 00000000..8cb638ca --- /dev/null +++ b/tests/test_ah012_executor_seam.py @@ -0,0 +1,163 @@ +"""AH-012 — the executor seam. `execute_sql` is split into a single un-bypassable guarded envelope +(`execute_guarded`: read-only guard -> semantic-model safety -> resolve datasource -> +`executor.execute(vetted_sql)`) and a swappable `Executor` port. These tests pin the load-bearing +invariants: the guard runs BEFORE the executor, the executor only ever sees already-vetted SQL, there +is no path to an executor around the guard (fail-closed, REQ-002/REQ-014), and the built-in executor +returns NATIVE-typed rows while the subprocess wire still serializes byte-identical CSV. +""" + +from __future__ import annotations + +import sqlite3 +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parent.parent +PKG_SRC = REPO_ROOT / "packages" / "agami-core" / "src" +if str(PKG_SRC) not in sys.path: + sys.path.insert(0, str(PKG_SRC)) + +import execute_sql # noqa: E402 + + +@pytest.fixture(autouse=True) +def _reset_override(): + # _max_rows_override is a module global (ACE-044); isolate every test from it. + execute_sql._max_rows_override = None + yield + execute_sql._max_rows_override = None + + +class _SpyExecutor: + """Records every call so a test can assert what SQL/creds/profile reached the connect-and-run + step — and that it was reached at all (or not, when a guard refuses first).""" + + def __init__(self, result: execute_sql.ExecResult | None = None): + self.calls: list[tuple[str, dict, str]] = [] + self._result = result or execute_sql.ExecResult(columns=["c"], rows=[(1,)], truncated=False) + + def execute(self, vetted_sql: str, creds: dict, *, profile: str) -> execute_sql.ExecResult: + self.calls.append((vetted_sql, creds, profile)) + return self._result + + +# --- the guard is un-bypassable: no path to the executor around it ----------------------------- + + +def test_readonly_guard_refuses_before_the_executor_is_reached(): + # A write statement is refused by the hard read-only gate; the executor is NEVER constructed a + # query for. This is the "no public path reaches an executor without the guard" invariant. + spy = _SpyExecutor() + with pytest.raises(execute_sql.GuardRefused) as ei: + execute_sql.execute_guarded("DELETE FROM t", "acme", None, executor=spy) + assert ei.value.code == 1 + assert ei.value.envelope["error"]["kind"] == "permission" + assert spy.calls == [] # executor never reached + + +def test_readonly_guard_still_fires_under_no_safety(): + # --no-safety skips ONLY the semantic-model pass, never the write/RCE/DoS read-only gate. + spy = _SpyExecutor() + with pytest.raises(execute_sql.GuardRefused) as ei: + execute_sql.execute_guarded("DROP TABLE t", "acme", None, executor=spy, no_safety=True) + assert ei.value.code == 1 + assert spy.calls == [] + + +def test_model_safety_refusal_short_circuits_before_the_executor(monkeypatch): + # A model-safety refusal already wrote its JSON to stderr, so the envelope is None and only the + # exit code is carried; the executor must not run. + monkeypatch.setattr(execute_sql, "_model_safety", lambda s, p, a: (s, 1)) + spy = _SpyExecutor() + with pytest.raises(execute_sql.GuardRefused) as ei: + execute_sql.execute_guarded("SELECT 1", "acme", None, executor=spy) + assert ei.value.code == 1 and ei.value.envelope is None + assert spy.calls == [] + + +# --- the executor only ever receives already-vetted SQL ---------------------------------------- + + +def test_executor_receives_vetted_sql_and_resolved_creds(monkeypatch): + # The default_filters rewrite happens in the model pass; the executor sees the POST-guard SQL and + # the resolved datasource creds — never raw user input, never an unresolved profile. + monkeypatch.setattr(execute_sql, "_load_credentials", lambda p: {"type": "sqlite", "path": ":memory:"}) + monkeypatch.setattr(execute_sql, "_model_safety", lambda s, p, a: ("SELECT 1 AS c /*vetted*/", None)) + spy = _SpyExecutor() + + result = execute_sql.execute_guarded("SELECT 1 AS c", "acme", "sales", executor=spy) + + assert spy.calls == [("SELECT 1 AS c /*vetted*/", {"type": "sqlite", "path": ":memory:"}, "acme")] + assert result.rows == [(1,)] + + +def test_no_safety_bypasses_the_model_pass_but_still_runs(monkeypatch): + monkeypatch.setattr(execute_sql, "_load_credentials", lambda p: {"type": "sqlite", "path": ":memory:"}) + + def _boom(*a, **k): + raise AssertionError("_model_safety must be skipped when no_safety=True") + + monkeypatch.setattr(execute_sql, "_model_safety", _boom) + spy = _SpyExecutor() + + result = execute_sql.execute_guarded("SELECT 1 AS c", "acme", None, executor=spy, no_safety=True) + + assert spy.calls[0][0] == "SELECT 1 AS c" # raw SQL passed straight to the executor, unrewritten + assert result.rows == [(1,)] + + +# --- the built-in executor: native rows in, byte-identical CSV out ------------------------------ + + +def test_builtin_executor_satisfies_the_executor_port(): + import ports + + assert isinstance(execute_sql.BUILTIN_EXECUTOR, ports.Executor) # 5th port, by shape + + +def test_builtin_executor_returns_native_typed_rows_and_emits_identical_csv(tmp_path, monkeypatch, capsys): + db = tmp_path / "t.db" + con = sqlite3.connect(db) + con.execute("CREATE TABLE t (n INTEGER, s TEXT)") + con.executemany("INSERT INTO t (n, s) VALUES (?, ?)", [(1, "a"), (2, None)]) + con.commit() + con.close() + monkeypatch.setattr(execute_sql, "_load_credentials", lambda p: {"type": "sqlite", "path": str(db)}) + + result = execute_sql.execute_guarded( + "SELECT n, s FROM t ORDER BY n", "acme", None, + executor=execute_sql.BUILTIN_EXECUTOR, no_safety=True, + ) + + # Native fidelity (Sandeep's concern): ints stay ints, SQL NULL stays None — NOT "" and not "2". + assert result.columns == ["n", "s"] + assert result.rows == [(1, "a"), (2, None)] + + # The subprocess/CLI wire still serializes byte-identical CSV at the edge (NULL renders as an + # empty field there — the ambiguity lives only in the text wire, not in the native rows). + execute_sql._emit_result_csv(result) + assert capsys.readouterr().out == "n,s\r\n1,a\r\n2,\r\n" + + +def test_collect_cursor_bounds_and_preserves_native_types(monkeypatch): + monkeypatch.setenv("AGAMI_SQL_MAX_ROWS", "2") + + class _Cur: + description = [("n",), ("s",)] + + def fetchmany(self, k): + return [(1, "a"), (2, None), (3, "c")][:k] + + r = execute_sql._collect_cursor(_Cur()) + assert r.columns == ["n", "s"] + assert r.rows == [(1, "a"), (2, None)] # cap 2, native None preserved + assert r.truncated is True # a (cap+1)th row was available + + +def test_builtin_executor_raises_executor_error_on_unknown_db(): + with pytest.raises(execute_sql.ExecutorError) as ei: + execute_sql._builtin_execute("SELECT 1", {"type": "nosuchdb"}, profile="acme") + assert ei.value.code == 2 + assert "Unsupported db type" in ei.value.msg From 5ce5bbf42efc021f29be3a754f53df3297e2af5a Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Sun, 12 Jul 2026 15:51:12 +0530 Subject: [PATCH 3/9] feat(tools,mcp_http): wire injectable executor in-process behind the guard (AH-012 slice 2) create_app registers adapters.executor via tools.set_injected_executor. When set, tool_execute_sql runs through execute_guarded in-process (no subprocess fork, no CSV round-trip) and builds the result from native rows; when None (the default) it forks the execute_sql subprocess exactly as before (byte-identical). Both paths funnel through the shared _finalize_execution so the returned envelope is identical. In-process rows are textualized to match the CSV wire for now (native-typed rows are the deferred decision). Spec: AH-012 --- packages/agami-core/src/mcp_http.py | 4 + packages/agami-core/src/tools.py | 165 +++++++++++++++++++++------- 2 files changed, 130 insertions(+), 39 deletions(-) diff --git a/packages/agami-core/src/mcp_http.py b/packages/agami-core/src/mcp_http.py index bc9e0572..a8f3ab9d 100644 --- a/packages/agami-core/src/mcp_http.py +++ b/packages/agami-core/src/mcp_http.py @@ -51,6 +51,7 @@ bootstrap_paths, record_tool_call, server_version, + set_injected_executor, ) _log = logging.getLogger(__name__) @@ -367,6 +368,9 @@ def create_app(extra_tools: dict | None = None, adapters: Adapters | None = None bootstrap_paths() adapters = adapters or default_adapters() auth_provider = adapters.auth_provider + # AH-012: register the composition-root executor (None = the default subprocess path). Behind the + # shared guard in `tool_execute_sql`; a hosted consumer injects a pooled/RBAC/tunnel executor here. + set_injected_executor(adapters.executor) # Validate consumer-supplied tools up front so a malformed entry fails at construction with a # clear error, not later as a KeyError/500 inside tools/list or tools/call. for tool_name, meta in (extra_tools or {}).items(): diff --git a/packages/agami-core/src/tools.py b/packages/agami-core/src/tools.py index 02efdd61..fbb90742 100644 --- a/packages/agami-core/src/tools.py +++ b/packages/agami-core/src/tools.py @@ -871,12 +871,118 @@ def _executor_truncated(stderr: str | None) -> bool: return False +# The composition-root executor (AH-012). ``None`` (the default) means "fork the execute_sql +# subprocess" — the byte-identical local/single-user path. A consumer injects a ``ports.Executor`` +# via ``create_app(adapters=…)`` to run execution IN-PROCESS behind the same guard (no fork, native +# rows). Process-global on purpose: the executor is a composition-root singleton, not per-request. +_INJECTED_EXECUTOR: Any | None = None + + +def set_injected_executor(executor: Any | None) -> None: + """Register (or clear) the composition-root executor. Called once by ``mcp_http.create_app`` from + ``adapters.executor``; ``None`` keeps the default subprocess path.""" + global _INJECTED_EXECUTOR + _INJECTED_EXECUTOR = executor + + +def _finalize_execution( + columns: list, data_rows: list, truncated: bool, *, profile: str, sql: str, + execution_ms: int, args: dict[str, Any], +) -> str: + """Shape a successful result (units + exact-render markdown + trust receipt), log the execution + through the single sink, and return the tool JSON. Shared by both execution paths — the subprocess + fork and the in-process executor — so a query returns the identical envelope whichever ran it.""" + # Deterministic, exact rendering — so the numbers a user verifies don't depend on + # how the host LLM chooses to format them. `markdown` is the table to display + # verbatim; `rows` stays raw (exact CSV values) for charting / programmatic use. + unit_map = _resolve_units(profile, sql) + try: + from semantic_model import units # stdlib-only; safe even without model deps + + markdown = units.format_table(columns, data_rows, unit_map) + except Exception: + markdown = None + + result = { + "columns": columns, + "rows": data_rows, + "row_count": len(data_rows), + "truncated": truncated, + "units": unit_map, + "markdown": markdown, # exact, full numbers (currency symbol + grouping) — render as-is + "sql": sql, + "execution_ms": execution_ms, + # Trust receipt — provenance + anything unapproved this answer used. Same assembler + # the agami-query skill renders, so Desktop gets the same trust panel. Clients should + # surface receipt.warnings and any receipt.metrics whose review_state != "approved" + # (offer to approve/correct via the save_correction tool). + "receipt": _resolve_receipt(profile, sql), + } + + # Log the execution through the single chokepoint: the DB sink when AGAMI_DB_URL is set (one + # query_executions row), else the local jsonl the skills use. Best-effort either way. + _record_query( + { + "ts": _now_iso(), + "profile": profile, + "question": args.get("raw_query"), + "sql": sql, + "row_count": len(data_rows), + "source": "mcp_server", + } + ) + return json.dumps(result, indent=2, default=str) + + +def _run_in_process( + sql: str, profile: str, area: str | None, max_rows: int | None, executor: Any +) -> tuple[list, list, bool] | dict: + """Run through the in-process executor behind the shared guarded envelope (no subprocess, no CSV + round-trip). Returns ``(columns, data_rows, truncated)`` on success, or an error dict on a guard + refusal / execution failure — the same error shape the subprocess branch produces. + + Rows are textualized to match the subprocess CSV wire (``None`` → ``""``, else ``str``) so the + two paths return observably identical JSON. Native-typed rows are a deliberately deferred decision + (see the AH-012 spec); flipping this one coercion is the follow-up once that's settled.""" + import execute_sql + + # The per-call cap rides execute_sql's module global (ACE-044). ACE-028, which makes in-process + # the real serving path, must thread the cap per-call instead — this global is not safe under + # concurrent in-process queries with different caps. Save/restore keeps it inert by default. + prev_cap = execute_sql._max_rows_override + execute_sql._max_rows_override = max_rows + try: + result = execute_sql.execute_guarded(sql, profile, area, executor=executor) + except execute_sql.GuardRefused as refusal: + if refusal.envelope is not None: + return {"error": refusal.envelope["error"]} + # A model-safety refusal wrote its detail to the server log (stderr); surface a clean refusal. + return {"error": {"kind": "permission", + "remediation": "Query refused by the semantic-model safety pass."}} + except execute_sql.ExecutorError as exc: + return {"error": {"kind": _classify_exit(exc.code), "remediation": exc.msg}} + finally: + execute_sql._max_rows_override = prev_cap + + columns = list(result.columns) + data_rows = [["" if v is None else str(v) for v in row] for row in result.rows] + truncated = result.truncated + if max_rows is not None and len(data_rows) > max_rows: # backstop, matches the subprocess branch + data_rows = data_rows[:max_rows] + truncated = True + return columns, data_rows, truncated + + def tool_execute_sql(args: dict[str, Any]) -> str: """Local analog of Ask Agami `execute_sql`: run a read-only SELECT locally. Routes through the sibling execute_sql.py (Tier-3 Python executor) so all DB types are handled identically and nothing but the rows leaves the process. Enforces the same read-only guarantee as the hosted connector. + + Two execution paths behind the same guard: the default forks the execute_sql subprocess + (isolation, byte-identical local/single-user); an injected executor (AH-012) runs in-process with + native rows. Both funnel through `_finalize_execution` so the returned envelope is identical. """ sql = args.get("sql") if not isinstance(sql, str) or not sql.strip(): @@ -903,6 +1009,23 @@ def tool_execute_sql(args: dict[str, Any]) -> str: if max_rows is not None: max_rows = max(1, min(max_rows, 10_000)) + area = str(args["area"]) if args.get("area") else None + + # In-process path (AH-012): a consumer injected an executor, so run behind the shared guarded + # envelope with no subprocess and no CSV round-trip. Falls through to the subprocess fork below + # when no executor is injected (the default) — that path stays byte-identical. + if _INJECTED_EXECUTOR is not None: + started = time.monotonic() + outcome = _run_in_process(sql, profile, area, max_rows, _INJECTED_EXECUTOR) + execution_ms = int((time.monotonic() - started) * 1000) + if isinstance(outcome, dict): # guard refusal / execution error + return json.dumps({**outcome, "sql": sql, "execution_ms": execution_ms}, indent=2) + columns, data_rows, truncated = outcome + return _finalize_execution( + columns, data_rows, truncated, + profile=profile, sql=sql, execution_ms=execution_ms, args=args, + ) + # The model safety pass (fan/chasm pre-flight + default_filters) runs inside # execute_sql.py; pass the subject area so default_filters scope correctly. # Route through the unified executor as a module (the package is installed alongside @@ -954,46 +1077,10 @@ def tool_execute_sql(args: dict[str, Any]) -> str: data_rows = data_rows[:max_rows] truncated = True - # Deterministic, exact rendering — so the numbers a user verifies don't depend on - # how the host LLM chooses to format them. `markdown` is the table to display - # verbatim; `rows` stays raw (exact CSV values) for charting / programmatic use. - unit_map = _resolve_units(profile, sql) - try: - from semantic_model import units # stdlib-only; safe even without model deps - - markdown = units.format_table(columns, data_rows, unit_map) - except Exception: - markdown = None - - result = { - "columns": columns, - "rows": data_rows, - "row_count": len(data_rows), - "truncated": truncated, - "units": unit_map, - "markdown": markdown, # exact, full numbers (currency symbol + grouping) — render as-is - "sql": sql, - "execution_ms": execution_ms, - # Trust receipt — provenance + anything unapproved this answer used. Same assembler - # the agami-query skill renders, so Desktop gets the same trust panel. Clients should - # surface receipt.warnings and any receipt.metrics whose review_state != "approved" - # (offer to approve/correct via the save_correction tool). - "receipt": _resolve_receipt(profile, sql), - } - - # Log the execution through the single chokepoint: the DB sink when AGAMI_DB_URL is set (one - # query_executions row), else the local jsonl the skills use. Best-effort either way. - _record_query( - { - "ts": _now_iso(), - "profile": profile, - "question": args.get("raw_query"), - "sql": sql, - "row_count": len(data_rows), - "source": "mcp_server", - } + return _finalize_execution( + columns, data_rows, truncated, + profile=profile, sql=sql, execution_ms=execution_ms, args=args, ) - return json.dumps(result, indent=2, default=str) def _now_iso() -> str: From fd3163fcfab28fd5ad60bf14f22301d4cbbbc4c3 Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Sun, 12 Jul 2026 15:53:03 +0530 Subject: [PATCH 4/9] =?UTF-8?q?test(tools):=20AH-012=20slice=202=20?= =?UTF-8?q?=E2=80=94=20injection=20via=20create=5Fapp,=20in-process=20rout?= =?UTF-8?q?ing,=20un-bypassable?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit create_app(adapters=Adapters(executor=…)) registers the executor and default clears it; an injected executor runs in-process on vetted SQL with NO subprocess fork; a write is refused before the executor is ever reached; an ExecutorError maps to the same error envelope; and the default (no executor) still forks the CLI subprocess. Spec: AH-012 --- tests/test_ah012_executor_seam.py | 107 ++++++++++++++++++++++++++++++ 1 file changed, 107 insertions(+) diff --git a/tests/test_ah012_executor_seam.py b/tests/test_ah012_executor_seam.py index 8cb638ca..72cc1c28 100644 --- a/tests/test_ah012_executor_seam.py +++ b/tests/test_ah012_executor_seam.py @@ -8,6 +8,7 @@ from __future__ import annotations +import json import sqlite3 import sys from pathlib import Path @@ -161,3 +162,109 @@ def test_builtin_executor_raises_executor_error_on_unknown_db(): execute_sql._builtin_execute("SELECT 1", {"type": "nosuchdb"}, profile="acme") assert ei.value.code == 2 assert "Unsupported db type" in ei.value.msg + + +# --- Slice 2: injection through create_app + the in-process branch in tool_execute_sql ---------- + + +@pytest.fixture(autouse=True) +def _reset_injected_executor(): + import tools + + tools.set_injected_executor(None) + yield + tools.set_injected_executor(None) + + +def test_create_app_registers_and_clears_the_injected_executor(monkeypatch): + pytest.importorskip("starlette") + pytest.importorskip("mcp") + monkeypatch.setenv("PUBLIC_BASE_URL", "https://agami.example.test") + import mcp_http + import tools + from ports import Adapters + + base = mcp_http.default_adapters() + fake = _SpyExecutor() + adapters = Adapters( + activity_sink=base.activity_sink, org_resolver=base.org_resolver, + auth_provider=base.auth_provider, governance=base.governance, executor=fake, + ) + mcp_http.create_app(adapters=adapters) + assert tools._INJECTED_EXECUTOR is fake # wired from adapters.executor + + mcp_http.create_app() # default adapters carry no executor + assert tools._INJECTED_EXECUTOR is None # default path stays the subprocess fork + + +def test_injected_executor_runs_in_process_with_vetted_sql_and_no_fork(monkeypatch): + import tools + + monkeypatch.setattr(tools, "resolve_profile", lambda ds: "acme") + monkeypatch.setattr(execute_sql, "_load_credentials", lambda p: {"type": "sqlite", "path": ":memory:"}) + monkeypatch.setattr(execute_sql, "_model_safety", lambda s, p, a: (s + " /*vetted*/", None)) + # a fork here would be the REQ-002 violation the seam prevents — fail loudly if it happens. + monkeypatch.setattr(tools.subprocess, "run", lambda *a, **k: pytest.fail("must not fork a subprocess")) + + fake = _SpyExecutor(result=execute_sql.ExecResult(columns=["n"], rows=[(1,), (2,)], truncated=False)) + tools.set_injected_executor(fake) + out = json.loads(tools.tool_execute_sql({"sql": "SELECT n FROM t", "datasource": "acme"})) + + assert fake.calls[0][0] == "SELECT n FROM t /*vetted*/" # executor saw POST-guard SQL only + assert out["columns"] == ["n"] and out["rows"] == [["1"], ["2"]] and out["row_count"] == 2 + + +def test_injected_executor_is_unreachable_for_a_write(monkeypatch): + import tools + + monkeypatch.setattr(tools, "resolve_profile", lambda ds: "acme") + monkeypatch.setattr(tools.subprocess, "run", lambda *a, **k: pytest.fail("must not fork a subprocess")) + + fake = _SpyExecutor() + tools.set_injected_executor(fake) + out = json.loads(tools.tool_execute_sql({"sql": "DELETE FROM t"})) + + assert out["error"]["kind"] == "permission" # refused by the read-only guard + assert fake.calls == [] # the injected executor was never reached — un-bypassable + + +def test_injected_executor_error_maps_to_the_same_envelope(monkeypatch): + import tools + + monkeypatch.setattr(tools, "resolve_profile", lambda ds: "acme") + monkeypatch.setattr(execute_sql, "_load_credentials", lambda p: {"type": "sqlite", "path": ":memory:"}) + monkeypatch.setattr(execute_sql, "_model_safety", lambda s, p, a: (s, None)) + + class _Boom: + def execute(self, vetted_sql, creds, *, profile): + raise execute_sql.ExecutorError("Postgres connect failed: refused", code=4) + + tools.set_injected_executor(_Boom()) + out = json.loads(tools.tool_execute_sql({"sql": "SELECT 1", "datasource": "acme"})) + + assert out["error"]["kind"] == tools._classify_exit(4) + assert "connect failed" in out["error"]["remediation"] + + +def test_default_no_injected_executor_forks_the_subprocess(monkeypatch): + import tools + + monkeypatch.setattr(tools, "resolve_profile", lambda ds: "acme") + captured: dict = {} + + class _Proc: + returncode = 0 + stdout = "n\r\n1\r\n" + stderr = "" + + def _fake_run(cmd, **kw): + captured["cmd"] = cmd + return _Proc() + + monkeypatch.setattr(tools.subprocess, "run", _fake_run) + tools.set_injected_executor(None) # the default + + out = json.loads(tools.tool_execute_sql({"sql": "SELECT n FROM t", "datasource": "acme"})) + + assert "-m" in captured["cmd"] and "execute_sql" in captured["cmd"] # forked the CLI executor + assert out["rows"] == [["1"]] From 3f9dce5dafb6900e95425877ea67aab382ce0338 Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Sun, 12 Jul 2026 15:58:04 +0530 Subject: [PATCH 5/9] test(ah012): cover main() envelope translation + in-process refusal/backstop Add tests for the load-bearing new branches coverage flagged: main() translating a read-only refusal (JSON + exit 1), an ExecutorError (stderr message + exit code), and a success (byte-identical stdout CSV); plus the in-process model-safety refusal returning a clean error and the max_rows backstop trim. Spec: AH-012 --- tests/test_ah012_executor_seam.py | 75 +++++++++++++++++++++++++++++++ 1 file changed, 75 insertions(+) diff --git a/tests/test_ah012_executor_seam.py b/tests/test_ah012_executor_seam.py index 72cc1c28..9ee91b92 100644 --- a/tests/test_ah012_executor_seam.py +++ b/tests/test_ah012_executor_seam.py @@ -268,3 +268,78 @@ def _fake_run(cmd, **kw): assert "-m" in captured["cmd"] and "execute_sql" in captured["cmd"] # forked the CLI executor assert out["rows"] == [["1"]] + + +def test_injected_executor_model_safety_refusal_returns_clean_error(monkeypatch): + import tools + + monkeypatch.setattr(tools, "resolve_profile", lambda ds: "acme") + monkeypatch.setattr(execute_sql, "_load_credentials", lambda p: {"type": "sqlite", "path": ":memory:"}) + monkeypatch.setattr(execute_sql, "_model_safety", lambda s, p, a: (s, 1)) # refuse + fake = _SpyExecutor() + tools.set_injected_executor(fake) + + out = json.loads(tools.tool_execute_sql({"sql": "SELECT 1", "datasource": "acme"})) + + assert out["error"]["kind"] == "permission" + assert "semantic-model safety pass" in out["error"]["remediation"] + assert fake.calls == [] # refused before the executor + + +def test_injected_executor_backstop_trims_to_max_rows(monkeypatch): + import tools + + monkeypatch.setattr(tools, "resolve_profile", lambda ds: "acme") + monkeypatch.setattr(execute_sql, "_load_credentials", lambda p: {"type": "sqlite", "path": ":memory:"}) + monkeypatch.setattr(execute_sql, "_model_safety", lambda s, p, a: (s, None)) + fake = _SpyExecutor(result=execute_sql.ExecResult(columns=["n"], rows=[(1,), (2,), (3,)], truncated=False)) + tools.set_injected_executor(fake) + + out = json.loads(tools.tool_execute_sql({"sql": "SELECT n FROM t", "datasource": "acme", "max_rows": 2})) + + assert out["rows"] == [["1"], ["2"]] and out["truncated"] is True + + +# --- main() (the subprocess CLI entry) translates the envelope's outcomes byte-identically -------- + + +def _raise(exc): + raise exc + + +def test_main_read_only_refusal_writes_json_and_returns_1(tmp_path, monkeypatch, capsys): + monkeypatch.setenv("AGAMI_ARTIFACTS_DIR", str(tmp_path)) + monkeypatch.setattr(sys, "argv", ["execute_sql", "--profile", "acme", "--sql", "DELETE FROM t"]) + + rc = execute_sql.main() + + assert rc == 1 + assert json.loads(capsys.readouterr().err.strip())["error"]["kind"] == "permission" + + +def test_main_executor_error_writes_message_and_returns_code(tmp_path, monkeypatch, capsys): + monkeypatch.setenv("AGAMI_ARTIFACTS_DIR", str(tmp_path)) + monkeypatch.setattr( + execute_sql, "execute_guarded", + lambda *a, **k: _raise(execute_sql.ExecutorError("Postgres connect failed: refused", code=4)), + ) + monkeypatch.setattr(sys, "argv", ["execute_sql", "--profile", "acme", "--sql", "SELECT 1"]) + + rc = execute_sql.main() + + assert rc == 4 + assert capsys.readouterr().err.strip() == "Postgres connect failed: refused" + + +def test_main_success_serializes_result_to_stdout_csv(tmp_path, monkeypatch, capsys): + monkeypatch.setenv("AGAMI_ARTIFACTS_DIR", str(tmp_path)) + monkeypatch.setattr( + execute_sql, "execute_guarded", + lambda *a, **k: execute_sql.ExecResult(columns=["n"], rows=[(1,)], truncated=False), + ) + monkeypatch.setattr(sys, "argv", ["execute_sql", "--profile", "acme", "--sql", "SELECT n FROM t"]) + + rc = execute_sql.main() + + assert rc == 0 + assert capsys.readouterr().out == "n\r\n1\r\n" From 0e55b66e3fcecb246c8a629352dd8648cbb8793d Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Sun, 12 Jul 2026 16:04:18 +0530 Subject: [PATCH 6/9] =?UTF-8?q?docs+test(ah012):=20review=20nits=20?= =?UTF-8?q?=E2=80=94=20scope=20fidelity=20claim,=20clarify=20dead-branch?= =?UTF-8?q?=20+=20edge?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the review panel's non-blocking findings (0 security/correctness must-fix): - ports.Adapters docstring stale 'four' count -> 'the port adapters' - test header now scopes native fidelity to the ExecResult/CSV wire; the MCP tool edge textualizes (deferred decision) so the tool-edge assertions read stringified on purpose - comment: in-process read-only refusal is caught upstream (defence-in-depth branch) - comment: description None-vs-empty divergence is unreachable (guard admits only SELECT) - new test: tool edge renders SQL NULL as '' (not 'None') — pins the deferred coercion Spec: AH-012 --- packages/agami-core/src/execute_sql.py | 4 ++++ packages/agami-core/src/ports.py | 2 +- packages/agami-core/src/tools.py | 3 +++ plugins/agami/lib/execute_sql.py | 4 ++++ tests/test_ah012_executor_seam.py | 27 +++++++++++++++++++++++--- 5 files changed, 36 insertions(+), 4 deletions(-) diff --git a/packages/agami-core/src/execute_sql.py b/packages/agami-core/src/execute_sql.py index e791f660..41bf174d 100644 --- a/packages/agami-core/src/execute_sql.py +++ b/packages/agami-core/src/execute_sql.py @@ -802,6 +802,10 @@ def _collect_cursor(cur: Any) -> ExecResult: executor path share, so the row cap is enforced once, identically, for every caller.""" cap = _resolve_row_cap() if cur.description is None: + # No result set (a non-row statement). `_emit_result_csv` writes nothing for empty columns, + # matching the old sink. A description that is an *empty list* (a zero-column result set) + # would diverge from the old bare-header line, but the read-only guard admits only + # SELECT/WITH…SELECT, which always project >= 1 column — so that case can't reach here. return ExecResult(columns=[], rows=[], truncated=False) columns = [d[0] for d in cur.description] fetched = cur.fetchmany(cap + 1) diff --git a/packages/agami-core/src/ports.py b/packages/agami-core/src/ports.py index cffb3983..4e7f6d48 100644 --- a/packages/agami-core/src/ports.py +++ b/packages/agami-core/src/ports.py @@ -134,7 +134,7 @@ def execute(self, vetted_sql: str, creds: dict[str, str], *, profile: str) -> Ex @dataclass(frozen=True) class Adapters: - """The four port adapters, bundled so ``mcp_http.create_app`` takes them as one argument. + """The port adapters, bundled so ``mcp_http.create_app`` takes them as one argument. A consumer builds this with its own implementations of the ports (its own ``OrgResolver``, ``AuthProvider``, ``ActivitySink``, ``GovernancePolicy``, and optionally an ``Executor``); diff --git a/packages/agami-core/src/tools.py b/packages/agami-core/src/tools.py index fbb90742..c89681ed 100644 --- a/packages/agami-core/src/tools.py +++ b/packages/agami-core/src/tools.py @@ -954,6 +954,9 @@ def _run_in_process( try: result = execute_sql.execute_guarded(sql, profile, area, executor=executor) except execute_sql.GuardRefused as refusal: + # A read-only refusal (envelope present) is already caught by tool_execute_sql's upstream + # check_read_only fast-fail, so in practice only the model-safety branch (envelope None) is + # reached here; both are handled for defence-in-depth. if refusal.envelope is not None: return {"error": refusal.envelope["error"]} # A model-safety refusal wrote its detail to the server log (stderr); surface a clean refusal. diff --git a/plugins/agami/lib/execute_sql.py b/plugins/agami/lib/execute_sql.py index e791f660..41bf174d 100644 --- a/plugins/agami/lib/execute_sql.py +++ b/plugins/agami/lib/execute_sql.py @@ -802,6 +802,10 @@ def _collect_cursor(cur: Any) -> ExecResult: executor path share, so the row cap is enforced once, identically, for every caller.""" cap = _resolve_row_cap() if cur.description is None: + # No result set (a non-row statement). `_emit_result_csv` writes nothing for empty columns, + # matching the old sink. A description that is an *empty list* (a zero-column result set) + # would diverge from the old bare-header line, but the read-only guard admits only + # SELECT/WITH…SELECT, which always project >= 1 column — so that case can't reach here. return ExecResult(columns=[], rows=[], truncated=False) columns = [d[0] for d in cur.description] fetched = cur.fetchmany(cap + 1) diff --git a/tests/test_ah012_executor_seam.py b/tests/test_ah012_executor_seam.py index 9ee91b92..3175b4f4 100644 --- a/tests/test_ah012_executor_seam.py +++ b/tests/test_ah012_executor_seam.py @@ -1,9 +1,14 @@ """AH-012 — the executor seam. `execute_sql` is split into a single un-bypassable guarded envelope (`execute_guarded`: read-only guard -> semantic-model safety -> resolve datasource -> `executor.execute(vetted_sql)`) and a swappable `Executor` port. These tests pin the load-bearing -invariants: the guard runs BEFORE the executor, the executor only ever sees already-vetted SQL, there -is no path to an executor around the guard (fail-closed, REQ-002/REQ-014), and the built-in executor -returns NATIVE-typed rows while the subprocess wire still serializes byte-identical CSV. +invariants: the guard runs BEFORE the executor, the executor only ever sees already-vetted SQL, and +there is no path to an executor around the guard (fail-closed, REQ-002/REQ-014). + +On result fidelity: the built-in executor returns NATIVE-typed rows (`ExecResult`) — NULL as None, +ints as int — and the subprocess wire still serializes byte-identical CSV. The in-process TOOL edge +(`tools._run_in_process`) currently textualizes those rows to match the CSV wire so both execution +paths return identical JSON; native-typed rows at the MCP JSON edge are a deliberately deferred +decision (see the AH-012 spec), so the tool-edge tests assert the stringified form on purpose. """ from __future__ import annotations @@ -286,6 +291,22 @@ def test_injected_executor_model_safety_refusal_returns_clean_error(monkeypatch) assert fake.calls == [] # refused before the executor +def test_injected_executor_textualizes_null_as_empty_at_the_tool_edge(monkeypatch): + # The deferred-decision contract: at the MCP JSON edge the in-process path renders SQL NULL as + # "" (matching the CSV wire), NOT "None". Pins the one coercion a future native-typed switch flips. + import tools + + monkeypatch.setattr(tools, "resolve_profile", lambda ds: "acme") + monkeypatch.setattr(execute_sql, "_load_credentials", lambda p: {"type": "sqlite", "path": ":memory:"}) + monkeypatch.setattr(execute_sql, "_model_safety", lambda s, p, a: (s, None)) + fake = _SpyExecutor(result=execute_sql.ExecResult(columns=["n", "s"], rows=[(1, None)], truncated=False)) + tools.set_injected_executor(fake) + + out = json.loads(tools.tool_execute_sql({"sql": "SELECT n, s FROM t", "datasource": "acme"})) + + assert out["rows"] == [["1", ""]] # int -> "1", NULL -> "" (never "None") + + def test_injected_executor_backstop_trims_to_max_rows(monkeypatch): import tools From 817de6675295ebd7ef407f1f653980ac0f6f58ee Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Sun, 12 Jul 2026 16:18:09 +0530 Subject: [PATCH 7/9] fix(tools): fail-closed net for in-process SystemExit + validate executor shape (Copilot) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address Copilot review on PR #116: - _run_in_process now catches SystemExit — a deep sys.exit(2) in _load_credentials/_parse_dsn (bad profile/DSN) can no longer escape in-process and take down the host; it becomes a fail-closed tool error. The subprocess/CLI path keeps sys.exit for byte-identical exit codes; converting those helpers to raise is a follow-up (they have SystemExit-asserting tests). - set_injected_executor validates the object satisfies ports.Executor at registration, so a malformed adapter fails fast at app construction instead of AttributeError at query time. Spec: AH-012 --- packages/agami-core/src/tools.py | 21 ++++++++++++++++++++- tests/test_ah012_executor_seam.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/packages/agami-core/src/tools.py b/packages/agami-core/src/tools.py index c89681ed..deb9cee2 100644 --- a/packages/agami-core/src/tools.py +++ b/packages/agami-core/src/tools.py @@ -880,8 +880,18 @@ def _executor_truncated(stderr: str | None) -> bool: def set_injected_executor(executor: Any | None) -> None: """Register (or clear) the composition-root executor. Called once by ``mcp_http.create_app`` from - ``adapters.executor``; ``None`` keeps the default subprocess path.""" + ``adapters.executor``; ``None`` keeps the default subprocess path. Validates the shape at + registration so a malformed adapter fails fast at app construction, not as an ``AttributeError`` + at query time.""" global _INJECTED_EXECUTOR + if executor is not None: + import ports + + if not isinstance(executor, ports.Executor): # runtime_checkable: has execute(...) + raise TypeError( + "injected executor must satisfy ports.Executor " + "(an execute(vetted_sql, creds, *, profile) method)" + ) _INJECTED_EXECUTOR = executor @@ -964,6 +974,15 @@ def _run_in_process( "remediation": "Query refused by the semantic-model safety pass."}} except execute_sql.ExecutorError as exc: return {"error": {"kind": _classify_exit(exc.code), "remediation": exc.msg}} + except SystemExit as exc: + # Fail-closed net: some deep helpers (_load_credentials / _parse_dsn) still `sys.exit(2)` on a + # bad profile/DSN. In-process that SystemExit would escape the tool envelope and could take + # down the host, so convert it to a tool error. (Converting those helpers to raise is a + # follow-up; the subprocess/CLI path keeps sys.exit for byte-identical exit codes. The + # detailed message already went to the server log via their stderr write.) + code = exc.code if isinstance(exc.code, int) else 2 + return {"error": {"kind": _classify_exit(code), + "remediation": "Credentials or datasource configuration error."}} finally: execute_sql._max_rows_override = prev_cap diff --git a/tests/test_ah012_executor_seam.py b/tests/test_ah012_executor_seam.py index 3175b4f4..7928098e 100644 --- a/tests/test_ah012_executor_seam.py +++ b/tests/test_ah012_executor_seam.py @@ -251,6 +251,35 @@ def execute(self, vetted_sql, creds, *, profile): assert "connect failed" in out["error"]["remediation"] +def test_set_injected_executor_rejects_a_bad_shape(): + import tools + + class _NotAnExecutor: + pass # no .execute method + + with pytest.raises(TypeError): + tools.set_injected_executor(_NotAnExecutor()) + assert tools._INJECTED_EXECUTOR is None # rejected, nothing stored + + +def test_injected_executor_systemexit_is_caught_not_fatal(monkeypatch): + # A deep sys.exit (a bad profile/DSN in _load_credentials/_parse_dsn) must NOT escape in-process + # and kill the host — it becomes a fail-closed tool error. + import tools + + monkeypatch.setattr(tools, "resolve_profile", lambda ds: "acme") + + def _exit(*a, **k): + raise SystemExit(2) + + monkeypatch.setattr(execute_sql, "execute_guarded", _exit) + tools.set_injected_executor(_SpyExecutor()) + + out = json.loads(tools.tool_execute_sql({"sql": "SELECT 1", "datasource": "acme"})) + + assert "error" in out # a tool error envelope, not a process exit + + def test_default_no_injected_executor_forks_the_subprocess(monkeypatch): import tools From 89f8c55d7ed1b6153c9fd77d7bcdc02770c82044 Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Sun, 12 Jul 2026 16:44:19 +0530 Subject: [PATCH 8/9] =?UTF-8?q?fix(execute=5Fsql):=20convert=20credential/?= =?UTF-8?q?DSN=20sys.exit=20to=20ExecutorError=20=E2=80=94=20parity=20in-p?= =?UTF-8?q?rocess=20(Copilot)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address Copilot's remediation-drift finding on PR #116: the SystemExit net discarded the detailed message _load_credentials/_parse_dsn wrote, so in-process errors got a generic remediation while the subprocess path surfaced the detail. Rather than capture stderr (thread-unsafe global swap in a server), convert the 4 sys.exit(2) sites to ExecutorError(code=2). Now the detailed message rides the exception -> in-process remediation matches the subprocess path; main() still emits byte-identical CLI stderr+exit. - execute_sql: _load_credentials (missing creds / chmod / bad profile) + _parse_dsn (unsupported scheme) raise ExecutorError instead of sys.exit - setup_pgauth.py: translate ExecutorError -> stderr + exit 2 (keeps its CLI UX) - tests: dsn/env-credentials updated (SystemExit -> ExecutorError); new AH-012 test pins detailed in-process remediation; SystemExit net kept as defence-in-depth Spec: AH-012 --- packages/agami-core/src/execute_sql.py | 24 ++++++++++++------------ packages/agami-core/src/tools.py | 11 +++++------ plugins/agami/lib/execute_sql.py | 24 ++++++++++++------------ plugins/agami/scripts/setup_pgauth.py | 10 ++++++++-- tests/test_ah012_executor_seam.py | 22 ++++++++++++++++++++++ tests/test_dsn_parsing.py | 11 ++++++----- tests/test_env_credentials.py | 11 ++++++----- 7 files changed, 71 insertions(+), 42 deletions(-) diff --git a/packages/agami-core/src/execute_sql.py b/packages/agami-core/src/execute_sql.py index 41bf174d..2c4b7935 100644 --- a/packages/agami-core/src/execute_sql.py +++ b/packages/agami-core/src/execute_sql.py @@ -195,13 +195,13 @@ def _load_credentials(profile: str) -> dict[str, str]: return _parse_dsn(dsn) if not CREDENTIALS_PATH.exists(): - sys.stderr.write( + raise ExecutorError( f"No warehouse credentials for profile [{profile}]. Set DATASOURCE_URL " f"(or DATASOURCE_URL__{_env_token(profile)}) " "in the environment, or create /local/credentials via the agami `init` skill.\n" - "Never type credentials into chat — they belong in the environment or the file.\n" + "Never type credentials into chat — they belong in the environment or the file.", + code=2, ) - sys.exit(2) # chmod check: refuse if too permissive. POSIX only — Windows file modes don't # map to Unix permission bits (NTFS ACLs guard the file; a stat() there reports @@ -209,11 +209,11 @@ def _load_credentials(profile: str) -> dict[str, str]: if os.name == "posix": mode = stat.S_IMODE(CREDENTIALS_PATH.stat().st_mode) if mode not in ALLOWED_PERMS: - sys.stderr.write( + raise ExecutorError( f"/local/credentials must be chmod 600 (currently {oct(mode)[2:]})\n" - f"Run: chmod 600 /local/credentials\n" + f"Run: chmod 600 /local/credentials", + code=2, ) - sys.exit(2) # IMPORTANT: enable inline-comment stripping for both `#` and `;`. Without # this, a credentials line like `account = xy12345 # locator + region` @@ -223,11 +223,11 @@ def _load_credentials(profile: str) -> dict[str, str]: cfg = configparser.ConfigParser(inline_comment_prefixes=("#", ";")) cfg.read(CREDENTIALS_PATH) if profile not in cfg: - sys.stderr.write( + raise ExecutorError( f"Profile [{profile}] not found in /local/credentials. " - f"Sections present: {cfg.sections()}\n" + f"Sections present: {cfg.sections()}", + code=2, ) - sys.exit(2) section = {k: (v.strip() if isinstance(v, str) else v) for k, v in cfg[profile].items()} @@ -336,12 +336,12 @@ def _parse_dsn(dsn: str) -> dict[str, str]: result = {"type": "sqlite", "path": path or u.path.lstrip("/")} return result else: - sys.stderr.write( + raise ExecutorError( f"Unsupported scheme {raw_scheme!r}. " f"Supported: postgresql[+driver], postgres[+driver], redshift, " - f"mysql[+driver], mariadb, snowflake, sqlite.\n" + f"mysql[+driver], mariadb, snowflake, sqlite.", + code=2, ) - sys.exit(2) # Snowflake's URL is account-shaped, not host:port. The "hostname" portion # of `snowflake://user:pw@xy12345.us-east-1.aws/MYDB/PUBLIC` is the account diff --git a/packages/agami-core/src/tools.py b/packages/agami-core/src/tools.py index deb9cee2..86268e49 100644 --- a/packages/agami-core/src/tools.py +++ b/packages/agami-core/src/tools.py @@ -975,14 +975,13 @@ def _run_in_process( except execute_sql.ExecutorError as exc: return {"error": {"kind": _classify_exit(exc.code), "remediation": exc.msg}} except SystemExit as exc: - # Fail-closed net: some deep helpers (_load_credentials / _parse_dsn) still `sys.exit(2)` on a - # bad profile/DSN. In-process that SystemExit would escape the tool envelope and could take - # down the host, so convert it to a tool error. (Converting those helpers to raise is a - # follow-up; the subprocess/CLI path keeps sys.exit for byte-identical exit codes. The - # detailed message already went to the server log via their stderr write.) + # Defence-in-depth. The known credential/DSN failures now raise ExecutorError (handled above, + # carrying their detailed message), so this net catches only a residual/future sys.exit deep + # in a driver — ensuring an in-process query can never take down the host; it becomes a + # fail-closed tool error instead. code = exc.code if isinstance(exc.code, int) else 2 return {"error": {"kind": _classify_exit(code), - "remediation": "Credentials or datasource configuration error."}} + "remediation": "Datasource configuration error."}} finally: execute_sql._max_rows_override = prev_cap diff --git a/plugins/agami/lib/execute_sql.py b/plugins/agami/lib/execute_sql.py index 41bf174d..2c4b7935 100644 --- a/plugins/agami/lib/execute_sql.py +++ b/plugins/agami/lib/execute_sql.py @@ -195,13 +195,13 @@ def _load_credentials(profile: str) -> dict[str, str]: return _parse_dsn(dsn) if not CREDENTIALS_PATH.exists(): - sys.stderr.write( + raise ExecutorError( f"No warehouse credentials for profile [{profile}]. Set DATASOURCE_URL " f"(or DATASOURCE_URL__{_env_token(profile)}) " "in the environment, or create /local/credentials via the agami `init` skill.\n" - "Never type credentials into chat — they belong in the environment or the file.\n" + "Never type credentials into chat — they belong in the environment or the file.", + code=2, ) - sys.exit(2) # chmod check: refuse if too permissive. POSIX only — Windows file modes don't # map to Unix permission bits (NTFS ACLs guard the file; a stat() there reports @@ -209,11 +209,11 @@ def _load_credentials(profile: str) -> dict[str, str]: if os.name == "posix": mode = stat.S_IMODE(CREDENTIALS_PATH.stat().st_mode) if mode not in ALLOWED_PERMS: - sys.stderr.write( + raise ExecutorError( f"/local/credentials must be chmod 600 (currently {oct(mode)[2:]})\n" - f"Run: chmod 600 /local/credentials\n" + f"Run: chmod 600 /local/credentials", + code=2, ) - sys.exit(2) # IMPORTANT: enable inline-comment stripping for both `#` and `;`. Without # this, a credentials line like `account = xy12345 # locator + region` @@ -223,11 +223,11 @@ def _load_credentials(profile: str) -> dict[str, str]: cfg = configparser.ConfigParser(inline_comment_prefixes=("#", ";")) cfg.read(CREDENTIALS_PATH) if profile not in cfg: - sys.stderr.write( + raise ExecutorError( f"Profile [{profile}] not found in /local/credentials. " - f"Sections present: {cfg.sections()}\n" + f"Sections present: {cfg.sections()}", + code=2, ) - sys.exit(2) section = {k: (v.strip() if isinstance(v, str) else v) for k, v in cfg[profile].items()} @@ -336,12 +336,12 @@ def _parse_dsn(dsn: str) -> dict[str, str]: result = {"type": "sqlite", "path": path or u.path.lstrip("/")} return result else: - sys.stderr.write( + raise ExecutorError( f"Unsupported scheme {raw_scheme!r}. " f"Supported: postgresql[+driver], postgres[+driver], redshift, " - f"mysql[+driver], mariadb, snowflake, sqlite.\n" + f"mysql[+driver], mariadb, snowflake, sqlite.", + code=2, ) - sys.exit(2) # Snowflake's URL is account-shaped, not host:port. The "hostname" portion # of `snowflake://user:pw@xy12345.us-east-1.aws/MYDB/PUBLIC` is the account diff --git a/plugins/agami/scripts/setup_pgauth.py b/plugins/agami/scripts/setup_pgauth.py index b6b67939..bb39d9cb 100644 --- a/plugins/agami/scripts/setup_pgauth.py +++ b/plugins/agami/scripts/setup_pgauth.py @@ -54,7 +54,7 @@ ensure_importable() import agami_paths # noqa: E402 -from execute_sql import _parse_dsn # reuse DSN parsing logic # noqa: E402 +from execute_sql import ExecutorError, _parse_dsn # reuse DSN parsing logic # noqa: E402 # NOTE: never bootstrap() at import — this module is imported by build_duckdb_attach and # tests. The one-shot legacy migration runs only from main() (and the other entry points). @@ -96,7 +96,13 @@ def _load_section(profile: str) -> dict[str, str]: section = {k: (v.strip() if isinstance(v, str) else v) for k, v in cfg[profile].items()} if "url" in section and section["url"]: - from_dsn = _parse_dsn(section["url"]) + # `_parse_dsn` now raises ExecutorError (not sys.exit) on a bad scheme so it's safe in-process; + # this script keeps its clean CLI UX (message on stderr, exit 2) by translating it here. + try: + from_dsn = _parse_dsn(section["url"]) + except ExecutorError as e: + sys.stderr.write(e.msg + "\n") + sys.exit(2) merged = dict(from_dsn) for k, v in section.items(): if k == "url": diff --git a/tests/test_ah012_executor_seam.py b/tests/test_ah012_executor_seam.py index 7928098e..0a8c82c9 100644 --- a/tests/test_ah012_executor_seam.py +++ b/tests/test_ah012_executor_seam.py @@ -262,6 +262,28 @@ class _NotAnExecutor: assert tools._INJECTED_EXECUTOR is None # rejected, nothing stored +def test_injected_executor_credential_error_surfaces_detailed_remediation(monkeypatch): + # Parity with the subprocess path: a bad-profile ExecutorError carries its detailed message, so + # the in-process tool envelope surfaces the SAME remediation the CLI stderr would (not a generic + # string). This is why _load_credentials/_parse_dsn raise instead of sys.exit. + import tools + + monkeypatch.setattr(tools, "resolve_profile", lambda ds: "acme") + + def _bad(profile): + raise execute_sql.ExecutorError( + "No warehouse credentials for profile [acme]. Set DATASOURCE_URL ...", code=2 + ) + + monkeypatch.setattr(execute_sql, "_load_credentials", _bad) + monkeypatch.setattr(execute_sql, "_model_safety", lambda s, p, a: (s, None)) + tools.set_injected_executor(_SpyExecutor()) + + out = json.loads(tools.tool_execute_sql({"sql": "SELECT 1", "datasource": "acme"})) + + assert "DATASOURCE_URL" in out["error"]["remediation"] # detailed, not the generic net string + + def test_injected_executor_systemexit_is_caught_not_fatal(monkeypatch): # A deep sys.exit (a bad profile/DSN in _load_credentials/_parse_dsn) must NOT escape in-process # and kill the host — it becomes a fail-closed tool error. diff --git a/tests/test_dsn_parsing.py b/tests/test_dsn_parsing.py index e1f54fed..7139b5ec 100644 --- a/tests/test_dsn_parsing.py +++ b/tests/test_dsn_parsing.py @@ -19,7 +19,7 @@ REPO_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO_ROOT / "plugins" / "agami" / "scripts")) -from execute_sql import _parse_dsn # noqa: E402 +from execute_sql import ExecutorError, _parse_dsn # noqa: E402 # --- Postgres family ------------------------------------------------------- @@ -251,12 +251,13 @@ def test_bigquery_no_credentials_means_adc(): # --- Rejection ------------------------------------------------------------- -def test_unsupported_scheme_exits(): - with pytest.raises(SystemExit) as exc: +def test_unsupported_scheme_raises(): + # _parse_dsn now raises ExecutorError (code 2), not sys.exit, so it's safe to call in-process. + with pytest.raises(ExecutorError) as exc: _parse_dsn("redis://u:p@host:6379/0") assert exc.value.code == 2 -def test_random_string_exits(): - with pytest.raises(SystemExit): +def test_random_string_raises(): + with pytest.raises(ExecutorError): _parse_dsn("not a url at all") diff --git a/tests/test_env_credentials.py b/tests/test_env_credentials.py index a3b62938..8b188d37 100644 --- a/tests/test_env_credentials.py +++ b/tests/test_env_credentials.py @@ -155,10 +155,11 @@ def test_bigquery_empty_alias_falls_through_to_adc(monkeypatch, tmp_path): # --- neither source → an error that names both ----------------------------- -def test_missing_both_sources_names_env_and_file(monkeypatch, tmp_path, capsys): +def test_missing_both_sources_names_env_and_file(monkeypatch, tmp_path): + # _load_credentials now raises ExecutorError (safe in-process); the detailed message that used to + # go to stderr now rides the exception, so callers (main() / the in-process tool) can surface it. monkeypatch.setattr(execute_sql, "CREDENTIALS_PATH", tmp_path / "absent") - with pytest.raises(SystemExit): + with pytest.raises(execute_sql.ExecutorError) as ei: _load_credentials("default") - err = capsys.readouterr().err - assert "DATASOURCE_URL" in err - assert "credentials" in err + assert "DATASOURCE_URL" in ei.value.msg + assert "credentials" in ei.value.msg From eb83acc02de4a663688b838a0f57e80644f1ebab Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Sun, 12 Jul 2026 17:00:13 +0530 Subject: [PATCH 9/9] docs(tools): scope the cross-path 'identical envelope' claim to successful results (Copilot) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Copilot correctly flagged that model-safety refusals aren't identical across paths: the subprocess surfaces execute_sql's stderr JSON as remediation (kind=_classify_exit(1)), while in-process returns a clean generic refusal. Successful RESULTS are identical (the valuable guarantee); refusal presentation differs because the structured detail is only on the subprocess's captured stderr — in-process it went to the server log. True structured-refusal parity needs _model_safety to RETURN its envelope (it writes to stderr today, pinned by the fail-closed guard tests in test_ace051) plus a change to the default subprocess path's refusal output — a guard-contract change separable from this seam, tracked as a follow-up. Fix the overclaiming docstrings + clarify the in-process remediation points to the server log; behaviour unchanged. Spec: AH-012 --- packages/agami-core/src/tools.py | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/packages/agami-core/src/tools.py b/packages/agami-core/src/tools.py index 86268e49..57df3eb6 100644 --- a/packages/agami-core/src/tools.py +++ b/packages/agami-core/src/tools.py @@ -901,7 +901,12 @@ def _finalize_execution( ) -> str: """Shape a successful result (units + exact-render markdown + trust receipt), log the execution through the single sink, and return the tool JSON. Shared by both execution paths — the subprocess - fork and the in-process executor — so a query returns the identical envelope whichever ran it.""" + fork and the in-process executor — so a **successful** query returns the identical result envelope + whichever ran it. (A guard *refusal* is not yet identical across paths: the subprocess surfaces + execute_sql's stderr JSON as the remediation, while the in-process path returns a clean generic + refusal with the structured detail in the server log. Full structured-refusal parity needs + `_model_safety` to *return* its envelope — it currently writes it to stderr, pinned by the + fail-closed guard tests — so it's tracked as a follow-up, not folded into this seam.)""" # Deterministic, exact rendering — so the numbers a user verifies don't depend on # how the host LLM chooses to format them. `markdown` is the table to display # verbatim; `rows` stays raw (exact CSV values) for charting / programmatic use. @@ -969,9 +974,12 @@ def _run_in_process( # reached here; both are handled for defence-in-depth. if refusal.envelope is not None: return {"error": refusal.envelope["error"]} - # A model-safety refusal wrote its detail to the server log (stderr); surface a clean refusal. + # A model-safety refusal wrote its structured detail to the server log (stderr); surface a + # clean refusal here. (The subprocess path instead surfaces that stderr JSON as remediation — + # the not-yet-identical refusal envelope tracked as a follow-up.) return {"error": {"kind": "permission", - "remediation": "Query refused by the semantic-model safety pass."}} + "remediation": "Query refused by the semantic-model safety pass " + "(see server logs for the specific rule)."}} except execute_sql.ExecutorError as exc: return {"error": {"kind": _classify_exit(exc.code), "remediation": exc.msg}} except SystemExit as exc: @@ -1003,7 +1011,9 @@ def tool_execute_sql(args: dict[str, Any]) -> str: Two execution paths behind the same guard: the default forks the execute_sql subprocess (isolation, byte-identical local/single-user); an injected executor (AH-012) runs in-process with - native rows. Both funnel through `_finalize_execution` so the returned envelope is identical. + native rows. Both funnel through `_finalize_execution`, so a **successful** query's result + envelope is identical either way (a guard refusal's envelope is not yet identical across paths — + see `_finalize_execution` and the tracked structured-refusal-parity follow-up). """ sql = args.get("sql") if not isinstance(sql, str) or not sql.strip():