feat(workspace): route warehouse tools through the bound workspace's engine

packages/core/src/flag/flag.ts

@@ -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

packages/opencode/src/altimate/native/connections/register.ts

@@ -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(
@@ -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
@@ -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
 }
 
 // ---------------------------------------------------------------------------
@@ -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
+    // 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",

packages/opencode/src/altimate/native/connections/registry.ts

@@ -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) {

packages/opencode/src/altimate/tools/schema-inspect.ts

@@ -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.",
@@ -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,
@@ -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>
@@ -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)
     }
   },
 })

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

@@ -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.",
@@ -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)
+    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
@@ -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.`,
-      }
+      })
     }
   },
 })