Skip to content
3 changes: 2 additions & 1 deletion docs/docs/data-engineering/tools/sql-tools.md
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,8 @@ Run SQL queries against your connected warehouse.
```

**Parameters:**
- `query` (required): SQL to execute
- `query` (required unless `query_from` is set): SQL to execute
- `query_from` (optional): `{warehouse, sql}` - a single read-only SELECT, run on `warehouse`, that returns exactly one row and one non-empty SQL string column (max 1 MiB). That SQL is validated as a read-only SELECT and run on the target `warehouse` without passing through the conversation. `query_from.warehouse` is the source; the top-level `warehouse` is then required and names the target; provide exactly one of `query` or `query_from`
- `warehouse` (optional): Connection name from config. Uses default if omitted
- `limit` (optional, default: 100): Max rows returned

Expand Down
7 changes: 7 additions & 0 deletions packages/opencode/src/altimate/tools/sql-check.ts
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,13 @@ export function readSql(sql: string, dialect?: string): string {
/\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) ||
Comment thread
anandgupta42 marked this conversation as resolved.
/\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,
Comment thread
anandgupta42 marked this conversation as resolved.
)
) {
throw new Error(
Expand Down
136 changes: 123 additions & 13 deletions packages/opencode/src/altimate/tools/sql-execute.ts
Original file line number Diff line number Diff line change
Expand Up @@ -39,12 +39,33 @@ import * as Precedence from "../workspace/precedence"
// altimate_change start — never render a warehouse failure as an empty result
import { normalizeError } from "./response-normalization"
// altimate_change end
import { readSql } from "./sql-check"

const MAX_SQL_ROWS = 10_000
const LIMIT_ERROR = `limit must be a finite positive integer between 1 and ${MAX_SQL_ROWS}`
const MAX_GENERATED_SQL_BYTES = 1024 * 1024
// Below the default tool-output cap, so a redirect never spills the SQL into a truncated file.
const MAX_REDIRECT_SQL_BYTES = 40 * 1024

function generatedRead(sql: string, warehouse: string, stage: "source" | "result"): string {
try {
return readSql(sql, Registry.canonicalType(Registry.getConfig(warehouse)?.type) ?? undefined)
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
} catch {
throw new Error(`query_from ${stage} must be a single read-only SELECT (or WITH ... SELECT).`)
}
}

const parameters = z.object({
query: z.string().describe("SQL query to execute"),
query: z.string().nullish().describe("SQL query to execute; omit when using query_from"),
query_from: z
.object({
warehouse: z.string().describe("Warehouse containing the SQL generator or saved SQL"),
sql: z.string().describe("Single read-only SELECT returning exactly one row and one SQL string column"),
})
.nullish()
.describe(
"Fetch SQL internally instead of copying generated SQL through the conversation; requires an explicit target warehouse",
),
warehouse: z.string().optional().describe("Warehouse connection name"),
limit: z
.number({ error: LIMIT_ERROR })
Expand Down Expand Up @@ -73,10 +94,79 @@ export const SqlExecuteTool = Tool.define(
Effect.gen(function* () {
const bridge = yield* EffectBridge.make()
return yield* legacyToInit({
description: "Execute SQL against a connected data warehouse. Returns results as a formatted table.",
description:
"Execute SQL against a connected data warehouse. Returns results as a formatted table. " +
"Provide exactly one of query or query_from. query_from runs a read-only generator SELECT and executes its single SQL string result (up to 1 MiB) as a read-only SELECT on warehouse, without previewing or retyping the generated SQL. " +
"For DuckDB table sets, generate query_table([...], true) with escaped table-name string literals. For thousands of tables, UNION ALL BY NAME batches of at most 128 names to stay within the default expression depth.",
parameters,
formatValidationError: (error) => error.issues?.map((issue) => issue.message).join("; ") || LIMIT_ERROR,
async execute(args, ctx) {
let sql = args.query ?? ""
// Strict-schema clients send null or empty placeholders for unused optional fields.
const queryFrom = args.query_from?.warehouse.trim() || args.query_from?.sql.trim() ? args.query_from : undefined
if (Boolean(sql.trim()) === Boolean(queryFrom)) {
throw new Error("Provide exactly one of query or query_from.")
}
let sourcePrecedence: Precedence.Verdict | undefined
if (queryFrom) {
if (!queryFrom.warehouse.trim() || !queryFrom.sql.trim()) {
throw new Error("query_from requires non-empty warehouse and sql.")
}
if (!args.warehouse?.trim()) throw new Error("query_from requires an explicit target warehouse.")
const source = queryFrom
const sourceSql = generatedRead(source.sql, source.warehouse, "source")
await ctx.ask({
permission: "sql_execute",
patterns: [sourceSql],
always: ["*"],
metadata: { warehouse: source.warehouse },
})
sourcePrecedence = await Precedence.check(ctx.sessionID, "sql_execute", source.warehouse)
if (sourcePrecedence.redirect) return sourcePrecedence.redirect
try {
const result = await Dispatcher.call("sql.execute", {
sql: sourceSql,
warehouse: source.warehouse,
limit: 2,
query_timeout_ms: args.query_timeout_ms,
Comment thread
anandgupta42 marked this conversation as resolved.
...(Registry.canonicalType(Registry.getConfig(source.warehouse)?.type) === "postgres"
? { read_only: true }
: {}),
}, { signal: ctx.abort, sessionID: ctx.sessionID })
const error = normalizeError(result.error)
if (error !== undefined) {
throw new Error(`query_from source failed: ${error.trim() || "SQL execution failed."}`)
}
const value = result.rows[0]?.[0]
if (
result.truncated ||
result.row_count !== 1 ||
result.columns.length !== 1 ||
result.rows.length !== 1 ||
result.rows[0].length !== 1 ||
typeof value !== "string" ||
!value.trim()
) {
throw new Error(
"query_from must return exactly one row and one non-empty SQL string column, without truncation.",
)
}
if (Buffer.byteLength(value, "utf8") > MAX_GENERATED_SQL_BYTES) {
throw new Error("query_from generated SQL exceeds the 1 MiB limit.")
}
sql = generatedRead(value, args.warehouse, "result")
} catch (error) {
if (!sourcePrecedence.notice) throw error
const message = error instanceof Error ? error.message : String(error)
throw new Error(`${sourcePrecedence.notice}\n\n${message}`)
}
await ctx.ask({
permission: "sql_execute",
patterns: [sql],
always: ["*"],
metadata: { warehouse: args.warehouse },
})
Comment thread
anandgupta42 marked this conversation as resolved.
}
// altimate_change start - SQL write access control
// Permission checks OUTSIDE try/catch so denial errors propagate to the framework
// MongoDB warehouses take JSON command documents; the SQL classifier would read
Expand All @@ -87,9 +177,9 @@ export const SqlExecuteTool = Tool.define(
const dialect = args.warehouse
? (Registry.canonicalType(Registry.getConfig(args.warehouse)?.type) ?? undefined)
: undefined
const mongo = isMongoWarehouse(warehouse) ? classifyMongoCommand(args.query) : null
const { queryType, blocked } = mongo ?? classifyAndCheck(args.query, dialect)
const query = mongo?.normalized ?? args.query
const mongo = isMongoWarehouse(warehouse) ? classifyMongoCommand(sql) : null
const { queryType, blocked } = mongo ?? classifyAndCheck(sql, dialect)
const query = mongo?.normalized ?? sql
if (blocked) {
throw new Error(
"DROP DATABASE, DROP SCHEMA, TRUNCATE, and executable SQL comments are blocked for safety. This cannot be overridden.",
Expand All @@ -111,10 +201,10 @@ export const SqlExecuteTool = Tool.define(
// writes to), so table DDL/row DML there needs no write permission. Statements that can
// reach outside the engine (COPY TO, ATTACH, INSTALL/LOAD, EXPORT, SET) still ask, and
// the hard-deny list above still applies.
if (queryType === "write" && !(isScratch(args.warehouse) && isScratchSafeWrite(args.query))) {
if (queryType === "write" && !(isScratch(args.warehouse) && isScratchSafeWrite(sql))) {
await ctx.ask({
permission: "sql_execute_write",
patterns: [args.query.slice(0, 200)],
patterns: [sql.slice(0, 200)],
always: ["*"],
metadata: { queryType },
})
Expand All @@ -130,16 +220,36 @@ export const SqlExecuteTool = Tool.define(
// the same statement needed a moment ago. Approving and then redirecting is not a
// wasted prompt: the write still happens, through the engine, and what the user
// authorised is the write — not which connection carries it.
const precedence = await Precedence.check(ctx.sessionID, "sql_execute", args.warehouse)
if (precedence.redirect) return precedence.redirect
const precedence = { ...(await Precedence.check(ctx.sessionID, "sql_execute", args.warehouse)) }
if (precedence.redirect) {
// The workspace tool needs the SQL; a generated statement exists only in this
// frame, so hand it to the caller or the redirect cannot be followed.
if (queryFrom && Buffer.byteLength(sql, "utf8") > MAX_REDIRECT_SQL_BYTES) {
throw new Error(
"query_from generated SQL is too large to hand to the workspace tool; run it on a local connection.",
)
}
const redirect = queryFrom
? {
...precedence.redirect,
output: `${precedence.redirect.output}\n\nSQL to run:\n${sql}`,
Comment thread
anandgupta42 marked this conversation as resolved.
Comment thread
anandgupta42 marked this conversation as resolved.
metadata: { ...precedence.redirect.metadata, generated_sql: true },
}
: precedence.redirect
return sourcePrecedence ? Precedence.annotate(sourcePrecedence, redirect) : redirect
}
if (sourcePrecedence?.notice) {
precedence.notice = [sourcePrecedence.notice, precedence.notice].filter(Boolean).join("\n")
precedence.precedence ??= sourcePrecedence.precedence
}
// altimate_change end

// altimate_change start — shadow-mode pre-execution SQL validation
// Runs validation against cached schema and emits sql_pre_validation telemetry,
// but does NOT block execution. Used to measure catch rate before deciding
// whether to enable blocking in a future release. Fire-and-forget so it
// doesn't add latency to the sql_execute hot path.
preValidateSql(args.query, args.warehouse, queryType).catch(() => {})
preValidateSql(sql, args.warehouse, queryType).catch(() => {})
// altimate_change end

try {
Expand Down Expand Up @@ -187,7 +297,7 @@ export const SqlExecuteTool = Tool.define(
// than after formatting) means a formatting failure below still leaves this
// execution counted, instead of silently dropping it from the telemetry.
try {
const fp = computeSqlFingerprint(args.query, dialect)
const fp = computeSqlFingerprint(sql, dialect)
if (fp) {
Telemetry.track({
type: "sql_fingerprint",
Expand Down Expand Up @@ -224,7 +334,7 @@ export const SqlExecuteTool = Tool.define(
maxLines: Math.max(0, limits.maxLines - extra.split("\n").length + 1),
write: (text) => bridge.promise(truncate.write(text)),
})
const lint = await lintSqlResult(args.query, args.warehouse, ctx.sessionID)
const lint = await lintSqlResult(sql, args.warehouse, ctx.sessionID)
let output = formatted.output + (lint ? "\n\n" + lint.output : "")
// altimate_change start — progressive disclosure suggestions
if (suggestion) {
Expand All @@ -239,7 +349,7 @@ export const SqlExecuteTool = Tool.define(
// altimate_change — carries the fail-open notice when the target could not be
// attributed to the workspace; a no-op otherwise.
const response = Precedence.annotate(precedence, {
title: `SQL: ${args.query.slice(0, 60)}${args.query.length > 60 ? "..." : ""}`,
title: queryFrom ? "SQL: generated read" : `SQL: ${sql.slice(0, 60)}${sql.length > 60 ? "..." : ""}`,
metadata: { ...formatted.metadata, ...(lint && { lint_rules: lint.rules }) },
output,
})
Expand Down
Loading
Loading