Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions packages/core/src/flag/flag.ts
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,14 @@ export const Flag = {
get ALTIMATE_WORKSPACE() {
return truthy("ALTIMATE_WORKSPACE")
},
/**
* Workspace precedence escape hatch, set by `--integrations=local`. When on, the
* native warehouse tools serve every local connection themselves and nothing is
* redirected to the bound workspace's integration engine, for the whole session.
*/
get ALTIMATE_INTEGRATIONS_LOCAL() {
return process.env["ALTIMATE_INTEGRATIONS"]?.toLowerCase() === "local"
},
// altimate_change end

// Evaluated at access time (not module load) because tests, the CLI, and
Expand Down
166 changes: 138 additions & 28 deletions packages/opencode/src/altimate/native/connections/register.ts
Original file line number Diff line number Diff line change
Expand Up @@ -43,22 +43,23 @@ import { Telemetry } from "../../../telemetry"
/** Cached dbt adapter (lazily created on first use). */
let dbtAdapter: any | null | undefined = undefined

// altimate_change start — single-flight adapter creation.
// Two concurrent `warehouse`-less calls used to construct the adapter twice, and
// construction is expensive: it spawns a detached Python bridge, rebuilds the
// manifest and starts file watchers. Share one in-flight promise instead. The
// permanent negative cache (`dbtAdapter === null`) is unchanged — a project that
// becomes valid mid-session is still not retried.
let dbtAdapterInflight: Promise<any | null> | undefined

/**
* Try to execute SQL via dbt's adapter (which uses profiles.yml for connection).
* Returns null if dbt is not available or not configured — caller should fall back
* to native driver.
*
* This is the preferred path when working in a dbt project: dbt already knows
* how to connect, so users don't need to configure a separate connection.
* Resolve the dbt adapter for this project, or null when there is no usable dbt
* project. Idempotent, single-flight, and permanently negative once it has failed.
*/
async function tryExecuteViaDbt(
sql: string,
limit?: number,
): Promise<SqlExecuteResult | null> {
// Only attempt dbt once — if it's not configured, don't retry on every query
if (dbtAdapter === null) return null
async function ensureDbtAdapter(): Promise<any | null> {
if (dbtAdapter !== undefined) return dbtAdapter
if (dbtAdapterInflight) return dbtAdapterInflight

if (dbtAdapter === undefined) {
dbtAdapterInflight = (async () => {
try {
// Check if dbt config exists
const { read: readDbtConfig } = await import(
Expand All @@ -83,12 +84,106 @@ async function tryExecuteViaDbt(
// Create the adapter
const { create } = await import("../../../../../dbt-tools/src/adapter")
dbtAdapter = await create(dbtConfig)
return dbtAdapter
} catch {
// dbt-tools not available or config invalid — fall back to native
dbtAdapter = null
return null
} finally {
dbtAdapterInflight = undefined
}
})()
return dbtAdapterInflight
}

/** Where a `warehouse`-less call would actually go. */
export type DefaultTarget =
| {
source: "dbt"
type?: string
/** Where execution actually lands if the dbt attempt yields nothing. `sql.execute`
* falls back to the registry not only when dbt is absent, but whenever
* `tryExecuteViaDbt` returns null — an unrecognised result shape, or any throw.
* A caller deciding anything about this call has to consider both targets. */
fallback?: { type: string; name: string }
}
| { source: "registry"; type: string; name: string }
| { source: "none" }

/**
* Resolve the target a call with no `warehouse` would reach, mirroring the resolution
* the handler for `op` performs itself — so a caller inspecting the target ahead of
* time cannot disagree with where execution actually lands.
*
* Only `sql.execute` consults dbt. `sql.explain` and `schema.inspect` are
* registry-only, and must stay that way: resolving them through dbt would drag
* adapter construction (Python bridge, manifest rebuild, file watchers) onto paths
* that never touch dbt today.
*
* For the dbt path the reported `type` is the project's adapter type, which is what
* decides *which* warehouse the profile reaches. It is left undefined when it cannot
* be established — the adapter coalesces an unknown type to the string "unknown", and
* the call can throw before initialisation completes.
*/
export async function resolveDefaultTarget(
op: "sql.execute" | "sql.explain" | "schema.inspect",
): Promise<DefaultTarget> {
if (op === "sql.execute") {
const adapter = await ensureDbtAdapter()
if (adapter) {
let type: string | undefined
try {
const reported = adapter.getAdapterType?.()
if (typeof reported === "string" && reported && reported.toLowerCase() !== "unknown") type = reported
} catch {
// Adapter not initialised far enough to answer; leave the type undetermined.
}
if (!type) type = await adapterTypeFromManifest()
const warehouses = Registry.list().warehouses
const fallback = warehouses.length > 0 ? { type: warehouses[0].type, name: warehouses[0].name } : undefined
return { source: "dbt", type, fallback }
}
}

const warehouses = Registry.list().warehouses
if (warehouses.length === 0) return { source: "none" }
return { source: "registry", type: warehouses[0].type, name: warehouses[0].name }
}

/** Fallback adapter type: the dbt manifest records it as `metadata.adapter_type`. */
async function adapterTypeFromManifest(): Promise<string | undefined> {
try {
const { read: readDbtConfig } = await import("../../../../../dbt-tools/src/config")
const dbtConfig = await readDbtConfig()
if (!dbtConfig) return undefined
const fs = await import("fs")
const path = await import("path")
const manifestPath = path.join(dbtConfig.projectRoot, "target", "manifest.json")
if (!fs.existsSync(manifestPath)) return undefined
const raw = JSON.parse(fs.readFileSync(manifestPath, "utf8"))
const adapter = String(raw?.metadata?.adapter_type ?? "").toLowerCase()
return adapter || undefined
} catch {
return undefined
}
}
// altimate_change end

/**
* Try to execute SQL via dbt's adapter (which uses profiles.yml for connection).
* Returns null if dbt is not available or not configured — caller should fall back
* to native driver.
*
* This is the preferred path when working in a dbt project: dbt already knows
* how to connect, so users don't need to configure a separate connection.
*/
async function tryExecuteViaDbt(
sql: string,
limit?: number,
): Promise<SqlExecuteResult | null> {
// altimate_change start — share the single-flight creation path with resolveDefaultTarget
if (!(await ensureDbtAdapter())) return null
// altimate_change end

try {
const raw = limit
Expand Down Expand Up @@ -146,6 +241,9 @@ async function tryExecuteViaDbt(
/** Reset dbt adapter (for testing). */
export function resetDbtAdapter(): void {
dbtAdapter = undefined
// altimate_change — drop any in-flight creation too, or a test that resets mid-flight
// would still receive the previous adapter.
dbtAdapterInflight = undefined
}

// ---------------------------------------------------------------------------
Expand Down Expand Up @@ -369,29 +467,41 @@ register("sql.execute", async (params: SqlExecuteParams): Promise<SqlExecuteResu
const startTime = Date.now()
const warehouseType = getWarehouseType(params.warehouse)
try {
// altimate_change start — resolve the fallback connection before the dbt attempt.
// `tryExecuteViaDbt` awaits, and the registry is mutable: re-reading it afterwards
// could pick a different connection than the one the caller's routing decision was
// computed against (a concurrent `warehouse.add` can change which name sorts first).
// Reading once here makes the decided connection and the executed connection the
// same by construction. The dbt-first ordering below is unchanged.
const fallbackName = params.warehouse || Registry.list().warehouses[0]?.name

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

MAJOR — time-of-check/time-of-use between the routing decision and the executed target

This pin, and the check at :497-501, close the window across the dbt await. But the routing decision was made earlier and elsewhere: Precedence.check()resolveDefaultTarget (register.ts:139-160) does its own Registry.list().warehouses[0] read from inside the tool body, and the handler then resolves the target again, independently. The await Dispatcher.call(...) boundary and the handler's own awaits are enough for a queued concurrent mutation to land in between, so the comment's claim that this makes the decided and executed connection "the same by construction" is stronger than what the pin actually does.

Concretely:

  • the guard sees an unserved DuckDB default; a concurrent warehouse.remove drops it; sql.explain or schema.inspect then picks the newly-first Snowflake connection and executes it locally, despite Snowflake being shadowed — unaudited execution on a served connection, the exact outcome this design exists to prevent;
  • for an explicit name, a concurrent warehouse.add can replace that name with a served type after check() read it. The handler pins the already-replaced type and sees no subsequent change, so this check cannot detect that window.

Note also that this pin exists only in register("sql.execute")sql.explain (:552-570) and schema.inspect (:678-691) have no equivalent guard at all.

Fix: make the decision and the target acquisition atomic — move the precedence check into the handler after it pins the target (passing sessionID through), or return a lease {name, canonicalType, generation} that handlers must revalidate. Apply it to all three ops, explicit names included.

Related, same seam: Precedence.check()'s await import("../native/connections/register") (precedence.ts:586-588) has no try/catch, and check() is called outside the surrounding try in all three tool bodies — so a throw there takes out sql_execute, sql_explain and schema_inspect together instead of failing open.

default-target.test.ts:123-151 does not prove its stated invariant: it calls the dispatcher directly, omitting the preceding precedence decision, which is where the race actually is.

// Pinning the name is not enough on its own: the same name can be re-added against a
// different warehouse while this call is suspended, and the routing decision made for
// it is a function of that connection's canonical type. Pin the type too, so a
// replacement is caught rather than executed under a decision that never covered it.
const fallbackType = fallbackName ? Registry.canonicalType(Registry.getConfig(fallbackName)?.type) : undefined
// altimate_change end

// Strategy: try dbt adapter first (if in a dbt project), then fall back to native driver.
// dbt knows how to connect using profiles.yml — no separate connection config needed.
if (!params.warehouse) {
const dbtResult = await tryExecuteViaDbt(params.sql, params.limit)
if (dbtResult) return dbtResult
}

const warehouseName = params.warehouse
let result: SqlExecuteResult
if (!warehouseName) {
const warehouses = Registry.list().warehouses
if (warehouses.length === 0) {
throw new Error(
"No warehouse configured. Use warehouse.add, set ALTIMATE_CODE_CONN_* env vars, or configure a dbt profile.",
)
}
// Use the first warehouse as default
const connector = await Registry.get(warehouses[0].name)
result = await connector.execute(params.sql, params.limit)
} else {
const connector = await Registry.get(warehouseName)
result = await connector.execute(params.sql, params.limit)
if (!fallbackName) {
throw new Error(
"No warehouse configured. Use warehouse.add, set ALTIMATE_CODE_CONN_* env vars, or configure a dbt profile.",
)
}
// altimate_change start — refuse rather than execute under a stale decision.
if (Registry.canonicalType(Registry.getConfig(fallbackName)?.type) !== fallbackType) {
throw new Error(
`Connection "${fallbackName}" changed while this query was being prepared, so the routing decided for it no longer applies. Re-run the query.`,
)
}
// altimate_change end
const connector = await Registry.get(fallbackName)
const result: SqlExecuteResult = await connector.execute(params.sql, params.limit)
try {
Telemetry.track({
type: "warehouse_query",
Expand Down
23 changes: 23 additions & 0 deletions packages/opencode/src/altimate/native/connections/registry.ts
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,29 @@ const DRIVER_MAP: Record<string, string> = {
trino: "@altimateai/drivers/trino",
}

// altimate_change start — canonical driver identity for workspace precedence.
/**
* Collapse a `config.type` onto the canonical name of the driver that serves it, so
* callers reasoning about "which database is this really" cannot be fooled by an
* alias: `postgresql` and `postgres` are one driver, as are `mariadb`/`mysql`,
* `mssql`/`fabric`/`sqlserver`, and `mongo`/`mongodb`.
*
* Derived by inverting `DRIVER_MAP` rather than restating it, so a type added there
* cannot silently desync from everything keyed on driver identity. Returns null for a
* type no driver serves.
*
* `redshift` maps to its own driver and therefore stays distinct from `postgres`: a
* different service with different credentials and endpoints, where Postgres
* wire-compatibility is an implementation detail rather than an identity.
*/
export function canonicalType(type: string | undefined | null): string | null {
if (!type) return null
const driverPath = DRIVER_MAP[type.toLowerCase()]
if (!driverPath) return null
return driverPath.slice(driverPath.lastIndexOf("/") + 1)
}
// altimate_change end

async function createConnector(name: string, config: ConnectionConfig): Promise<Connector> {
const driverPath = DRIVER_MAP[config.type.toLowerCase()]
if (!driverPath) {
Expand Down
22 changes: 17 additions & 5 deletions packages/opencode/src/altimate/tools/schema-inspect.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,9 @@ import type { SchemaInspectResult } from "../native/types"
import { PostConnectSuggestions } from "./post-connect-suggestions"
// altimate_change end
import { isRecord, normalizeError } from "./response-normalization"
// altimate_change start — workspace precedence
import * as Precedence from "../workspace/precedence"
// altimate_change end

export const SchemaInspectTool = Tool.define("schema_inspect", {
description: "Inspect database schema — list columns, types, and constraints for a table.",
Expand All @@ -15,6 +18,14 @@ export const SchemaInspectTool = Tool.define("schema_inspect", {
warehouse: z.string().optional().describe("Warehouse connection name"),
}),
async execute(args, ctx) {
// altimate_change start — workspace precedence
const precedence = await Precedence.check(ctx.sessionID, "schema_inspect", args.warehouse)
if (precedence.redirect) return precedence.redirect
// Every failure exit goes through here, so a fail-open notice cannot be dropped by
// one path being overlooked — three of the four exits below are errors, and the
// marker is most needed on exactly those.
const failed = (message: string) => Precedence.annotate(precedence, schemaError(message))
// altimate_change end
try {
const result = (await Dispatcher.call("schema.inspect", {
table: args.table,
Expand All @@ -23,12 +34,12 @@ export const SchemaInspectTool = Tool.define("schema_inspect", {
})) as unknown

if (!isRecord(result)) {
return schemaError("Invalid schema response from dispatcher.")
return failed("Invalid schema response from dispatcher.")
}

const responseError = normalizeError(result.error)
if (result.success === false || responseError !== undefined) {
return schemaError(responseError?.trim() || "Schema inspection failed.")
return failed(responseError?.trim() || "Schema inspection failed.")
}

const schemaResult = (isRecord(result.data) ? result.data : result) as Partial<SchemaInspectResult>
Expand All @@ -45,14 +56,15 @@ export const SchemaInspectTool = Tool.define("schema_inspect", {
})
}
// altimate_change end
return {
// altimate_change — attaches the fail-open notice when present; no-op otherwise.
return Precedence.annotate(precedence, {
title: `Schema: ${schemaResult.table ?? args.table}`,
metadata: { success: true, columnCount: (schemaResult.columns ?? []).length, rowCount: schemaResult.row_count },
output,
}
})
} catch (e) {
const msg = e instanceof Error ? e.message : String(e)
return schemaError(msg)
return failed(msg)
}
},
})
Expand Down
30 changes: 26 additions & 4 deletions packages/opencode/src/altimate/tools/sql-execute.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,9 @@ import { PostConnectSuggestions } from "./post-connect-suggestions"
import { getCache } from "../native/schema/cache"
import * as Registry from "../native/connections/registry"
// altimate_change end
// altimate_change start — workspace precedence
import * as Precedence from "../workspace/precedence"
// altimate_change end

export const SqlExecuteTool = Tool.define("sql_execute", {
description: "Execute SQL against a connected data warehouse. Returns results as a formatted table.",
Expand All @@ -38,6 +41,19 @@ export const SqlExecuteTool = Tool.define("sql_execute", {
}
// altimate_change end

// altimate_change start — workspace precedence.
// Last, after BOTH native safety checks. A redirect returns early, so anything
// above it stops running — and neither check has an equivalent on the other side:
// the engine's execution tools apply no hard-deny list, and an engine tool key is
// matched by the builder's `"*": "allow"` rule while `sql_execute_write` is "ask".
// Redirecting first would let a write reach the warehouse without the confirmation
// 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)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

MAJOR — the approval is not bound to the call that eventually executes

The ordering here is correct, and precedence-guard-order.test.ts genuinely proves it: a hard-denied statement throws before any redirect, and a write is confirmed before it is redirected. Claim 3 holds as written.

What is not enforced is the invariant the ordering is for. The redirect is model-facing text. The model then composes a new engine call with new arguments. The engine wrapper checks only the generic engine-tool permission with pattern "*", which the builder's "*": "allow" rule matches; it does not re-run classifyAndCheck and does not raise sql_execute_write. So approving

UPDATE orders SET ...

does not guarantee the engine call carries that statement — it could carry a different write, or one of the supposedly un-overridable DROP DATABASE / DROP SCHEMA / TRUNCATE forms, since the hard-deny list has no engine-side equivalent. The comment above states the asymmetry plainly; what it does not say is that the confirmation therefore authorises a turn, not a statement.

The model could already call the engine tool before this PR — what changes is that for served connections the unguarded path becomes the recommended one, and E2E row 7 shows the model does follow redirects unprompted.

Fix: forward the already-validated arguments server-side to the selected engine tool, or wrap each mapped engine execute tool with the same classifier, hard deny and confirmation applied to the engine call's own SQL. At minimum this belongs in the review log as a disclosed residual — a reader of Claim 3 today would reasonably conclude the gates still cover the query.

if (precedence.redirect) return precedence.redirect
// 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
Expand Down Expand Up @@ -87,18 +103,24 @@ export const SqlExecuteTool = Tool.define("sql_execute", {
})
}
// altimate_change end
return {
// altimate_change — carries the fail-open notice when the target could not be
// attributed to the workspace; a no-op otherwise.
return Precedence.annotate(precedence, {
title: `SQL: ${args.query.slice(0, 60)}${args.query.length > 60 ? "..." : ""}`,
metadata: { rowCount: result.row_count, truncated: result.truncated },
output,
}
})
} catch (e) {
const msg = e instanceof Error ? e.message : String(e)
return {
// altimate_change — annotate the failure too. A fail-open notice that only rides
// on success is worse than none: the reason vanishes exactly when the call went
// wrong, and the `precedence` marker under-counts fail-open in precisely the
// cases most likely to fail.
return Precedence.annotate(precedence, {
title: "SQL: ERROR",
metadata: { rowCount: 0, truncated: false, error: msg },
output: `Failed to execute SQL: ${msg}\n\nEnsure the dispatcher is running and a warehouse connection is configured.`,
}
})
}
},
})
Expand Down
Loading
Loading