feat(execute_sql): Executor seam — guarded envelope + swappable port (AH-012) - #116
Conversation
…lice 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_<db> now return native-typed rows (ExecResult) instead of writing CSV, and the subprocess/CLI _execute_<db> 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
…y, native rows 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
…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
…uting, un-bypassable 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
…ackstop 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
…anch + edge 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
There was a problem hiding this comment.
Pull request overview
This PR introduces an executor seam for SQL execution by splitting execute_sql into a single un-bypassable guarded envelope (execute_guarded) and a swappable Executor port, allowing hosted deployments to inject pooled/RBAC/tunneled executors behind the same guard while keeping the default subprocess/CSV path byte-identical.
Changes:
- Add a new
ports.ExecutorProtocol and plumb it throughAdaptersandmcp_http.create_app. - Refactor
execute_sqlto return native-typedExecResultviaexecute_guarded, with CSV serialization kept at the subprocess edge. - Add injected-executor support to
tool_execute_sql(in-process path) and add a comprehensive AH-012 test suite.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/test_ah012_executor_seam.py | Adds coverage for guard-before-executor invariants, fidelity, injection wiring, and parity between subprocess vs in-process paths. |
| packages/agami-core/src/execute_sql.py | Implements ExecResult, guarded envelope (execute_guarded), and executor dispatch separation (_run_<db> + thin CSV wrappers). |
| plugins/agami/lib/execute_sql.py | Auto-synced mirror of the execute_sql refactor for the stdlib-lean plugin distribution. |
| packages/agami-core/src/tools.py | Adds injected-executor plumbing and an in-process execution branch sharing _finalize_execution. |
| packages/agami-core/src/ports.py | Introduces the new Executor port and adds it to Adapters. |
| packages/agami-core/src/mcp_http.py | Registers adapters.executor into the tools layer at app construction. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| 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. | ||
| 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}} |
There was a problem hiding this comment.
Addressed in 817de66 with the fail-closed net you suggested: _run_in_process now catches SystemExit, so a deep sys.exit(2) from _load_credentials/_parse_dsn (bad profile/DSN) can no longer escape in-process and take down the host — it becomes a tool error envelope instead.
The deeper refactor (convert those helpers' sys.exit(2) → ExecutorError) is a deliberate follow-up: _parse_dsn/_load_credentials have existing tests asserting SystemExit (test_unsupported_scheme_exits, test_random_string_exits) and sit on the byte-identical subprocess/CLI path, so that change ripples beyond this seam — tracking it with ACE-028 (which makes in-process the real serving path). The net makes the seam safe today regardless.
| 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) |
There was a problem hiding this comment.
Same finding as the tools.py thread — addressed in 817de66 (fail-closed SystemExit net in _run_in_process); converting these helpers' sys.exit(2) to raises is the tracked follow-up (they have SystemExit-asserting tests + are on the byte-identical CLI path).
| 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) |
There was a problem hiding this comment.
Same finding — see the tools.py thread. Fail-closed net landed in 817de66; the mirror's sys.exit→raise conversion is the tracked follow-up.
| 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 |
There was a problem hiding this comment.
Done in 817de66 — set_injected_executor now isinstance-checks against ports.Executor (runtime_checkable) at registration, so a malformed adapter fails fast at create_app construction instead of an AttributeError at query time.
| 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) |
There was a problem hiding this comment.
All of these. Are there separate things happening per DB type? We need separate functions because of the dialect?
There was a problem hiding this comment.
Yes — the per-DB _run_<db> functions are irreducibly separate because each wraps a different driver with a genuinely different connect/execute API:
- Different drivers +
connect()signatures + imports: psycopg2, pymysql, snowflake.connector, google.cloud.bigquery, sqlite3, pymssql, oracledb, databricks.sql, trino, duckdb. - Different connection semantics: Postgres uses a server-side named cursor (
cursor(name="agami_bounded")+itersize) to bound transfer; MySQL setsautocommit; Snowflake takeswarehouse/role/authenticatorkwargs; BigQuery has no DB-API cursor at all (its ownjob.result(max_results=)loop); SQLite is file-path based; Oracle builds a DSN; Redshift = Postgres wire +sslmodedefault. - Different required fields per DB (
_require(...)).
What AH-012 made shared (it wasn't before): the bounded fetch (_collect_cursor), the CSV emit (_emit_result_csv), the guard (execute_guarded), and error handling (ExecutorError). So the only thing left per-DB is the dialect-specific connect+execute — which can't be collapsed without a driver-abstraction layer (that's what SQLAlchemy is; agami-core deliberately stays stdlib + driver-only, no heavy ORM dep).
Note the 10 thin _execute_<db> one-liners (return _emit_or_err(lambda: _run_<db>(...))) are collapsible into a dispatch table — they exist only so the CLI/tests call them by name. Happy to collapse those if you'd prefer less boilerplate; the _run_<db> bodies themselves must stay separate.
…utor shape (Copilot) 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
| 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. | ||
| 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}} | ||
| 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 | ||
|
|
There was a problem hiding this comment.
Good catch — fixed properly in 89f8c55, and I went a different way than stderr-capture on purpose.
Capturing stderr around execute_guarded would require swapping the process-global sys.stderr, which is not thread-safe in the server (handlers run in worker threads via run_blocking; concurrent requests would interleave/steal each other's captured output). So instead I removed the drift at the source: _load_credentials and _parse_dsn now raise ExecutorError(code=2) instead of sys.exit(2). The detailed message rides the exception, so:
- In-process:
_run_in_processcatchesExecutorError→remediation = exc.msg— the exact detailed message, parity with the subprocess path. - Subprocess/CLI:
main()catchesExecutorError→ writesmsg + "\n"and returns code 2 — byte-identical stderr + exit to the oldsys.exit(2).
Ripple handled: setup_pgauth.py (the one external _parse_dsn caller) translates ExecutorError back to its stderr+exit-2 CLI UX; the SystemExit-asserting tests were updated to assert ExecutorError; and the SystemExit net stays as pure defence-in-depth. New test test_injected_executor_credential_error_surfaces_detailed_remediation pins the parity. This also makes the executor library sys.exit-free (only the __main__ guard remains), which is the cleaner end state you flagged in the other threads.
…parity in-process (Copilot) 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
| 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. | ||
| return {"error": {"kind": "permission", | ||
| "remediation": "Query refused by the semantic-model safety pass."}} |
There was a problem hiding this comment.
Correct — and I've fixed the overclaim in eb83acc, but I'm deliberately not folding the full parity fix into this PR. Here's the reasoning:
Successful results are identical across paths (the valuable guarantee). The divergence is only for guard refusals, and it's structural: _model_safety writes its structured envelope to sys.stderr and returns (sql, rc). On the subprocess path tool_execute_sql reads that stderr and surfaces it (kind _classify_exit(1)); in-process, that JSON went to the server log (not captured — capturing it means a thread-unsafe global sys.stderr swap), so I return a clean generic refusal.
True structured-refusal parity requires _model_safety to return its envelope (so execute_guarded can carry it in GuardRefused for both callers), which ripples into:
test_ace051— the fail-closed guard tests pin both the stderr JSON and the(sql, rc)2-tuple across ~8 sites, and- the default subprocess path's refusal output (parsing the stderr envelope) — i.e. no longer byte-identical.
That's a change to the security guard's error-reporting contract, separable from the executor seam. Folding it into this PR would expand a security-sensitive, byte-identical-by-design change into the guard internals. So I've corrected the docstrings to scope the "identical envelope" claim to successful results, clarified the in-process remediation points to the server log, and I'm tracking structured-refusal parity as a follow-up (natural fit with the _model_safety-returns-envelope refactor). Both paths still correctly fail closed — only the refusal's presentation differs.
…ssful results (Copilot) 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
Spec: AH-012 — Executor seam in agami-core (contract: F1-consume-core-foundations · REQ-002).
Summary
Split
execute_sqlinto a single un-bypassable guarded envelope and a swappableExecutorport, so a hosted consumer can inject a pooled / per-user-RBAC / SSH-tunnel executor behind agami-core's one guard — no fork (REQ-002/REQ-014). The built-in executor stays the default and the local/single-user path is byte-identical (subprocess fork + CSV). This is the prerequisite for ACE-028 (the OSS in-process pooled executor) and for the hosted pooled executor (ported fromagami-data-agent) to plug in behind the shared guard.Changes
ports.py— newExecutorProtocol (5th port) +Adapters.executorfield (defaultNone).ExecResultlives inexecute_sql(the stdlib-lean plugin mirror can't importports) and is referenced underTYPE_CHECKING.execute_sql.py— the guarded envelopeexecute_guarded(read-only guard →_model_safety→ resolve creds →executor.execute(vetted_sql)). Per-engine_execute_<db>became_run_<db>returning a native-typedExecResult(raisingExecutorError/GuardRefused); thin_execute_<db>CSV wrappers are kept for the subprocess/CLI path.main()routes through the envelope withBUILTIN_EXECUTOR, then serializes byte-identical CSV._requireraises instead ofsys.exit(safe in-process; same stderr + exit 2 in the subprocess).tools.py—set_injected_executor+_INJECTED_EXECUTOR.tool_execute_sqlruns in-process viaexecute_guardedwhen an executor is injected (native rows, no subprocess, no CSV round-trip), else forks the subprocess (unchanged default). Both funnel through a shared_finalize_execution.mcp_http.create_app— registersadapters.executor.Fidelity (Sandeep's concern)
The CSV round-trip existed only because of the subprocess boundary and stringifies every value (NULL vs
"", Decimals/datetimes lose type). This change adds no new CSV conversion:ExecResultcarries native rows; the subprocess wire serializes byte-identical CSV at the edge. The in-process tool edge currently textualizes to match the CSV wire so both paths return identical JSON — native-typed rows at the MCP JSON edge are a deliberately deferred decision (flip one coercion once settled).Scope notes
execute_sql.pyinsertions are largely mechanical (engine rename + wrappers + moved dispatch);plugins/agami/lib/execute_sql.pyis the auto-synced mirror (dev.py sync-lib), not hand-written.max_rowsrides the_max_rows_overridemodule global — not concurrency-safe under injected executors. ACE-028 must thread it per-call before making in-process the real serving path.Out of scope
Any pooled/reusing executor (OSS = ACE-028; hosted = private
agamirepo). No change to guard/governance/logging logic — same checks, same verdicts. Single-user CSV wire untouched (documented public CLI contract).Test plan / checklist
tests/test_ah012_executor_seam.py): guard-before-executor (un-bypassable), executor sees only vetted SQL,--no-safetyscope, native rows + byte-identical CSV, injection viacreate_app, in-process error/refusal parity,main()byte-identical stderr/exit/CSV.uv run dev.py checkgreen — ruff + 1483 tests + gitleaks; plugin mirror synced.test_privacy_no_networkgreen); no customer data/secrets.