refactor(sql-check): drop predicate_effect; keep read-only, bind and permission fixes

packages/drivers/src/duckdb.ts

@@ -166,12 +166,13 @@ export async function connect(
     params: any[] | undefined,
     timeoutMs: number | undefined,
     maxResultBytes?: number,
+    readOnly?: boolean,
     onDispatch?: () => void,
     signal?: AbortSignal,
   ): Promise<any[]> {
     if (!shared) throw new Error("DuckDB connection not open")
     try {
-      return await shared.engine.call<any[]>("all", { sql, params, connectionId, maxResultBytes }, timeoutMs, onDispatch, signal)
+      return await shared.engine.call<any[]>("all", { sql, params, connectionId, maxResultBytes, readOnly }, timeoutMs, onDispatch, signal)
     } catch (err) {
       throw err instanceof Error ? wrapDuckDBError(err) : err
     }
@@ -180,7 +181,7 @@ export async function connect(
 
   const query = async (sql: string, signal?: AbortSignal): Promise<any[]> => {
     signal?.throwIfAborted()
-    return run(sql, undefined, resolveQueryTimeout(config).ms, undefined, undefined, signal)
+    return run(sql, undefined, resolveQueryTimeout(config).ms, undefined, undefined, undefined, signal)
   }
   const queryWithParams = (sql: string, params: any[]): Promise<any[]> =>
     run(sql, params, resolveQueryTimeout(config).ms)
@@ -332,6 +333,7 @@ export async function connect(
         binds?.length ? binds : undefined,
         resolveQueryTimeout(config, options).ms,
         options?.maxResultBytes,
+        options?.readOnly,
         options?.onDispatch,
         options?.signal,
       )

packages/drivers/src/local-engine-child-source.ts

@@ -222,10 +222,41 @@ async function close() {
   return null;
 }
 
+// Guard setup, result consumption and cleanup share one serialized request.
+// Separate requests would let another caller run between setup and cleanup.
+async function withReadOnly(p, execute) {
+  if (engine === "sqlite") {
+    const previous = sqliteDb.query("PRAGMA query_only").get().query_only;
+    sqliteDb.exec("PRAGMA query_only = ON");
+    try {
+      return await execute();
+    } finally {
+      sqliteDb.exec("PRAGMA query_only = " + (previous ? "ON" : "OFF"));
+    }
+  }
+  const control = (sql) => all({ sql, connectionId: p.connectionId });
+  // The legacy binding has no autocommit getter. Implicit transactions receive
+  // different IDs; an explicit transaction keeps the same ID across statements.
+  // Refuse before BEGIN: a nested BEGIN would abort the caller's transaction.
+  const first = await control("SELECT txid_current() AS id");
+  const second = await control("SELECT txid_current() AS id");
+  if (first[0].id === second[0].id) {
+    throw new Error("DuckDB read-only execution cannot run inside an active transaction; finish it before running the query.");
+  }
+  await control("BEGIN TRANSACTION READ ONLY");
+  try {
+    return await execute();
+  } finally {
+    await control("ROLLBACK");
+  }
+}
+
 async function handle(req) {
   if (req.op === "open") return open(req.payload);
-  if (req.op === "all") return all(req.payload);
-  if (req.op === "run") return run(req.payload);
+  if (req.op === "all" || req.op === "run") {
+    const execute = () => req.op === "all" ? all(req.payload) : run(req.payload);
+    return req.payload.readOnly ? withReadOnly(req.payload, execute) : execute();
+  }
   if (req.op === "disconnect") return disconnect(req.payload);
   if (req.op === "close") return close();
   throw new Error("unknown op " + req.op);

packages/drivers/src/postgres.ts

@@ -95,7 +95,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
       pool = new Pool(poolConfig)
     },
 
-    async execute(sql: string, limit?: number, _binds?: any[], options?: ExecuteOptions): Promise<ConnectorResult> {
+    async execute(sql: string, limit?: number, binds?: any[], options?: ExecuteOptions): Promise<ConnectorResult> {
       const client = await acquireClient(pool, options?.signal)
       // Tighter of the explicit statement_timeout and the common execution
       // deadline: a configured limit stricter than the default is preserved.
@@ -216,7 +216,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
           const consume = createResultByteBudget(options.maxResultBytes)
           const Query = pg.default?.Query ?? pg.Query
           return await runQuery(() => new Promise<ConnectorResult>((resolve, reject) => {
-            const stream = new Query({ text: query, rowMode: "array" })
+            const stream = new Query({ text: query, values: binds, rowMode: "array" })
             const rows: unknown[][] = []
             let columns: string[] = []
             let truncated = false
@@ -253,7 +253,9 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
           }))
         }
         // altimate_change end
-        const result = await runQuery<any>(() => client.query(query))
+        // altimate_change start — preserve bound catalog names and other query parameters
+        const result = await runQuery<any>(() => client.query(query, binds))
+        // altimate_change end
         const columns = result.fields?.map((f: any) => f.name) ?? []
         const truncated = effectiveLimit > 0 && result.rows.length > effectiveLimit
         const rows = truncated

packages/drivers/src/sqlite.ts

@@ -77,7 +77,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
         // altimate_change start — report dispatch so callers can order follow-up work
         const info = await engine.call<{ changes: number; lastInsertRowid: number | bigint }>(
           "run",
-          { sql },
+          { sql, readOnly: options?.readOnly },
           timeoutMs,
           options?.onDispatch,
           options?.signal,
@@ -92,7 +92,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
       }
 
       // altimate_change start — forward the budget before the helper collects rows
-      const rows = await engine.call<any[]>("all", { sql: query, maxResultBytes: options?.maxResultBytes }, timeoutMs, options?.onDispatch, options?.signal)
+      const rows = await engine.call<any[]>("all", { sql: query, maxResultBytes: options?.maxResultBytes, readOnly: options?.readOnly }, timeoutMs, options?.onDispatch, options?.signal)
       // altimate_change end
       const columns = rows.length > 0 ? Object.keys(rows[0]) : []
       const truncated = effectiveLimit > 0 && rows.length > effectiveLimit

packages/drivers/src/types.ts

@@ -31,7 +31,7 @@ export interface ExecuteOptions {
   maxResultBytes?: number
   // altimate_change end
   // altimate_change start — PostgreSQL source extraction uses a pinned, rolled-back transaction
-  /** PostgreSQL only: execute in a READ ONLY transaction and roll it back before releasing the client. */
+  /** PostgreSQL/DuckDB: rolled-back READ ONLY transaction; SQLite: scoped query_only pragma. */
   readOnly?: boolean
   // altimate_change end
   /** Cancel pending or active execution when the driver supports cancellation. */

packages/drivers/test/postgres-readonly.test.ts

@@ -1,4 +1,5 @@
 import { expect, mock, test } from "bun:test"
+import { EventEmitter } from "node:events"
 import { connect } from "../src/postgres"
 import { QueryTimeoutError } from "../src/types"
 import { restoreModuleMocksAfterTests } from "./module-mock"
@@ -226,6 +227,64 @@ test("Postgres never starts user SQL when cancellation arrives during session se
   }
 })
 
+for (const maxResultBytes of [undefined, 1024]) {
+  test(`Postgres forwards bound catalog arrays in ${maxResultBytes === undefined ? "regular" : "byte-limited"} read-only queries`, async () => {
+    const calls: { sql: string; values?: unknown[] }[] = []
+    const releases: unknown[] = []
+    class Query extends EventEmitter {
+      constructor(readonly config: { text: string; values?: unknown[]; rowMode: string }) {
+        super()
+      }
+    }
+    mock.module("pg", () => ({
+      default: {
+        Query,
+        Pool: class {
+          async connect() {
+            return {
+              query(query: string | Query, values?: unknown[]) {
+                const sql = typeof query === "string" ? query : query.config.text
+                calls.push({ sql, values: typeof query === "string" ? values : query.config.values })
+                const fields = [{ name: "proname" }]
+                if (query instanceof Query) {
+                  queueMicrotask(() => {
+                    query.emit("row", ["lower"], { fields })
+                    query.emit("end", { fields })
+                  })
+                  return query
+                }
+                return Promise.resolve({ fields, rows: [{ proname: "lower" }] })
+              },
+              release(discard?: boolean) { releases.push(discard) },
+            }
+          }
+          async end() {}
+        },
+      },
+    }))
+    const connector = await connect({ type: "postgres", query_timeout_ms: 0 })
+    await connector.connect()
+    const sql = "SELECT proname FROM pg_catalog.pg_proc WHERE proname = ANY($1)"
+    const values = [["lower", "upper"]]
+    try {
+      const result = await connector.execute(sql, undefined, values, {
+        noLimit: true,
+        readOnly: true,
+        maxResultBytes,
+      })
+      expect(result.rows).toEqual([["lower"]])
+      expect(calls).toEqual([
+        { sql: "BEGIN READ ONLY", values: undefined },
+        { sql, values },
+        { sql: "ROLLBACK", values: undefined },
+      ])
+      expect(releases).toEqual([undefined])
+    } finally {
+      await connector.close()
+    }
+  })
+}
+
 async function probeDatabase(config: Record<string, unknown> = { query_timeout_ms: 1000 }) {
   const queries: string[] = []
   const releases: unknown[] = []

packages/opencode/src/altimate/tools/sql-check.ts

@@ -5,6 +5,8 @@ import * as Precedence from "../workspace/precedence"
 import * as Registry from "../native/connections/registry"
 import { classifyAndCheck, isSingleQuery } from "./sql-classify"
 import { maskSql, tokenizeSql } from "./sql-tokenize"
+import { sqlFunctionCalls } from "./sql-functions"
+import { hasSideEffectFunction } from "./sql-function-policy"
 import { executeQuery, type QueryContext } from "../native/query-execution"
 
 const column = z
@@ -120,9 +122,11 @@ export function readSql(sql: string, dialect?: string): string {
   // embedding user SQL, even when the parser reports a query category.
   const { tokens, complete } = tokenizeSql(sql, dialect)
   const masked = maskSql(tokens)
-  // SELECT alone does not make sequence setters, extension loaders or file writers read-only.
-  // Preserve quoted function identifiers while hiding string literals and comments.
-  const functions = maskSql(tokens, true)
+  // Keep sql_check's existing stricter ban on dynamic query(), even when the
+  // classifier can prove a literal body read-only. All calls use real code '(' tokens.
+  const unsafeFunction = sqlFunctionCalls(sql, dialect).some(
+    (call) => call.escaped || call.name === "query" || hasSideEffectFunction(call.name),
+  )
   const body = masked.trim().replace(/;$/, "").trim()
   let depth = 0
   let balanced = true
@@ -135,24 +139,15 @@ export function readSql(sql: string, dialect?: string): string {
     queryType !== "read" ||
     !isSingleQuery(sql, dialect) ||
     !complete ||
+    unsafeFunction ||
     !balanced ||
     depth !== 0 ||
     !/^(SELECT|WITH)\b/i.test(body) ||
     body.includes(";") ||
     /\b(INSERT|UPDATE|DELETE|MERGE|DROP|ALTER|CREATE|TRUNCATE|COPY|ATTACH|DETACH|PRAGMA|VACUUM|CALL|DO|INTO|INSTALL|LOAD|EXPORT|IMPORT|GRANT|REVOKE)\b/i.test(
       body,
     ) ||
-    /\bREPLACE\b(?!\s*\()/i.test(body) ||
-    /\b(nextval|setval|set_config|load_extension|writefile|dblink_exec|lo_import|lo_export|lo_create|lo_unlink|lo_put|pg_write_file|pg_append_file|pg_file_write|pg_file_rename|pg_file_unlink|pg_advisory_lock|pg_advisory_xact_lock)\s*\(/i.test(
-      functions,
-    ) ||
-    // PostgreSQL Unicode-escaped identifiers can spell a denied function without its literal name.
-    /\bU&/i.test(functions) ||
-    /\bSYSTEM\$\w+\s*\(/i.test(functions) ||
-    // Session-state, backend-signalling and replication functions are callable from a plain SELECT.
-    /\b(pg_try_advisory\w*|pg_advisory_unlock\w*|pg_advisory_lock\w*|pg_advisory_xact\w*|pg_terminate_backend|pg_cancel_backend|pg_reload_conf|pg_rotate_logfile|pg_promote|pg_switch_wal|pg_logical_emit_message|pg_create_\w*_replication_slot|pg_drop_replication_slot|pg_replication_slot_advance|pg_replication_origin_\w+|dblink|dblink_\w+|pg_notify|pg_stat_reset\w*|pg_wal_replay_\w+|pg_logical_slot_\w+|pg_replication_slot_\w+|get_lock|release_lock|release_all_locks|is_free_lock|is_used_lock|master_pos_wait|source_pos_wait|sleep|benchmark|pg_xlog_replay_\w+|pg_log_backend_memory_contexts|pg_create_restore_point|pg_backup_\w+|pg_start_backup|pg_stop_backup|lo_from_bytea)\s*\(/i.test(
-      functions,
-    )
+    /\bREPLACE\b(?!\s*\()/i.test(body)
   ) {
     throw new Error(
       "sql_check only accepts a single read-only SELECT (or WITH ... SELECT); it never writes or fixes data.",
@@ -174,7 +169,13 @@ export function checkSql(input: SqlCheckPublicInput, context?: QueryContext) {
 
 async function checkWithinDeadline(input: SqlCheckPublicInput, context: QueryContext) {
   context?.signal?.throwIfAborted()
-  const args = parameters.parse(dropEmpty(input))
+  const parsed = parameters.safeParse(dropEmpty(input))
+  if (!parsed.success) {
+    throw new Error(
+      `sql_check invalid input: ${parsed.error.issues.map((issue) => `${issue.path.join(".")}: ${issue.message}`).join("; ")}`,
+    )
+  }
+  const args = parsed.data
   if ([args.left_warehouse, args.right_warehouse].some((w) => w !== undefined && w !== args.warehouse)) {
     throw new Error("sql_check runs in one warehouse; use sql_materialize to put both sides in scratch first.")
   }
@@ -197,8 +198,8 @@ async function checkWithinDeadline(input: SqlCheckPublicInput, context: QueryCon
   async function run(query: string): Promise<Row[]> {
     const response = await executeQuery(Registry.getConfig(args.warehouse), {
       noLimit: true,
+      readOnly: true,
       queryTimeoutMs: args.query_timeout_ms,
-      ...(dialect === "postgres" ? { readOnly: true } : {}),
     }, context, (options) => connector.execute(query, undefined, undefined, options))
     if (response.truncated) throw new Error("sql_check received truncated aggregates; no verdict is available")
     return response.rows.map((row) => Object.fromEntries(response.columns.map((col, i) => [col, row[i]])))

packages/opencode/src/altimate/tools/sql-classify.ts

@@ -6,6 +6,8 @@
 
 import { hasMongoWriteStage, normalizeMqlCommand } from "@altimateai/drivers"
 import { maskSql, statementsOf, tokenizeSql } from "./sql-tokenize"
+import { sqlFunctionCalls, sqlLiteral } from "./sql-functions"
+import { hasSideEffectFunction, isReadOnlyNestedFunction } from "./sql-function-policy"
 
 // Safe import: napi binary may not be available on all platforms
 let getStatementTypes: ((sql: string, dialect?: string | null) => any) | null = null
@@ -165,6 +167,55 @@ export function classifyMulti(sql: string, dialect?: string): "read" | "write" {
   return classify(sql, dialect)
 }
 
+// query_table and *_scan take table identifiers, not executable SQL. Only these
+// known SQL arguments can establish read safety recursively in their target dialect.
+const SQL_ARGUMENTS: Record<string, { index: number; dialect: string }> = {
+  query: { index: 0, dialect: "duckdb" },
+  to_query: { index: 0, dialect: "snowflake" },
+  postgres_query: { index: 1, dialect: "postgres" },
+  mysql_query: { index: 1, dialect: "mysql" },
+  sqlite_query: { index: 1, dialect: "sqlite" },
+  postgres_execute: { index: 1, dialect: "postgres" },
+  mysql_execute: { index: 1, dialect: "mysql" },
+  sqlite_execute: { index: 1, dialect: "sqlite" },
+  eval: { index: 0, dialect: "sqlite" },
+}
+
+// Nested execution lacks the outer driver's read-only boundary. Apply sql_check's
+// conservative function policy there without extending its PG allowlist to ordinary sql_execute reads.
+function functionSafety(sql: string, dialect: string | undefined, depth: number, strict = depth > 0) {
+  let write = false
+  let blocked = false
+  for (const call of sqlFunctionCalls(sql, dialect)) {
+    // ODBC and other opaque remote engines do not reveal the target SQL dialect.
+    if (
+      call.escaped ||
+      hasSideEffectFunction(call.name, strict) ||
+      /^(odbc_query|odbc_copy|odbc_commit|odbc_rollback|openquery|external_query)$/.test(call.name)
+    ) write = true
+    // Nested and remote SQL cannot be proven read-only by a denylist: a user-defined
+    // function may write. Require every call to be on the read-only allowlist.
+    // SQL-string executors (query, postgres_query, ...) are exempt because their literal SQL is
+    // classified recursively below; dynamic arguments and executors with side effects still fail closed.
+    const nestedExecutor =
+      Object.hasOwn(SQL_ARGUMENTS, call.name) && call.qualifier === undefined && !hasSideEffectFunction(call.name)
+    if (depth > 0 && !nestedExecutor && !isReadOnlyNestedFunction(call.name, call.qualifier)) write = true
+    // SQL Server's XML .query() takes XQuery, not executable SQL.
+    if (call.name === "query" && /^(sqlserver|tsql|mssql)$/i.test(dialect ?? "")) continue
+    const argument = Object.hasOwn(SQL_ARGUMENTS, call.name) ? SQL_ARGUMENTS[call.name] : undefined
+    if (!argument) continue
+    const inner = sqlLiteral(call.args[argument.index], dialect)
+    if (!inner?.trim() || depth >= 8) {
+      write = true
+      continue
+    }
+    const result = classifySql(inner, argument.dialect, depth + 1)
+    if (result.queryType !== "read" || !isSingleQuery(inner, argument.dialect)) write = true
+    if (result.blocked) blocked = true
+  }
+  return { write, blocked }
+}
+
 /**
  * Single-pass: classify and check for hard-denied statement types.
  * Returns both the overall query type and whether a hard-deny pattern was found.
@@ -173,6 +224,14 @@ export function classifyAndCheck(sql: string, dialect?: string): {
   queryType: "read" | "write"
   blocked: boolean
   parseError?: string
+} {
+  return classifySql(sql, dialect, 0)
+}
+
+function classifySql(sql: string, dialect: string | undefined, depth: number): {
+  queryType: "read" | "write"
+  blocked: boolean
+  parseError?: string
 } {
   if (!sql || typeof sql !== "string") return { queryType: "read", blocked: false }
   if (!tokenizeSql(sql, dialect).complete) {
@@ -187,13 +246,18 @@ export function classifyAndCheck(sql: string, dialect?: string): {
 
     // A generic parser can treat MySQL arithmetic (--2) as a comment and omit
     // trailing statements. Keep dialect-aware lexical hard denies authoritative.
+    const functions = functionSafety(sql, dialect, depth)
     const blocked =
+      functions.blocked ||
       classifyFallback(sql, dialect).blocked ||
       result.statements.some(
         (s: { statement_type: string }) => s.statement_type && HARD_DENY_TYPES.has(s.statement_type.toUpperCase()),
       )
 
-    return { queryType: otherIsReadOnly(sql, result.statements, dialect) ? "read" : "write", blocked }
+    return {
+      queryType: !functions.write && !blocked && otherIsReadOnly(sql, result.statements, dialect) ? "read" : "write",
+      blocked,
+    }
   } catch (error) {
     return { ...classifyFallback(sql, dialect), parseError: error instanceof Error ? error.message : String(error) }
   }
@@ -286,6 +350,8 @@ export function isScratchSafeWrite(sql: string): boolean {
   const dialect = "duckdb"
   if (!getStatementTypes || !sql || typeof sql !== "string") return false
   if (!tokenizeSql(sql, dialect).complete) return false
+  const functions = functionSafety(sql, dialect, 0, true)
+  if (functions.write || functions.blocked) return false
   try {
     const result = getStatementTypes(sql, dialect)
     const statements: Array<{ statement_type?: string }> = result?.statements ?? []