Merge branch 'main' into fix/special-char-passwords

Author: anandgupta42

packages/opencode/src/altimate/telemetry/index.ts

@@ -644,15 +644,35 @@ export namespace Telemetry {
         session_id: string
         /** skipped = no cache or stale, passed = valid SQL, blocked = invalid SQL caught, error = validation itself failed */
         outcome: "skipped" | "passed" | "blocked" | "error"
-        /** why: no_cache, stale_cache, empty_cache, valid, non_structural, structural_error, validation_exception */
+        /** why: no_cache, stale_cache, empty_cache, valid, non_structural, structural_error, dispatcher_failed, validation_exception */
         reason: string
+        /** warehouse driver type (postgres, snowflake, bigquery, ...) — enables per-warehouse catch-rate analysis */
+        warehouse_type: string
+        /** read / write / unknown — enables per-query-type analysis */
+        query_type: string
+        /** SHA-256 prefix of masked SQL — join key to sql_execute_failure events for same query */
+        masked_sql_hash: string
         schema_columns: number
         /** true when schema scan hit the column-scan cap — flags samples biased by large-warehouse truncation */
         schema_truncated: boolean
         duration_ms: number
         error_message?: string
       }
     // altimate_change end
+    // altimate_change start — plan-agent model tool-call refusal detection
+    | {
+        type: "plan_no_tool_generation"
+        timestamp: number
+        session_id: string
+        message_id: string
+        model_id: string
+        provider_id: string
+        /** "stop" finish_reason without any tool calls in the session — flags models that refuse to tool-call in plan mode */
+        finish_reason: string
+        /** output tokens on the stop-without-tools generation — helps distinguish "refused" (low) from "wrote a long text plan" (high) */
+        tokens_output: number
+      }
+    // altimate_change end
 
   /** SHA256 hash a masked error message for anonymous grouping. */
   export function hashError(maskedMessage: string): string {

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

@@ -43,7 +43,7 @@ export const SqlExecuteTool = Tool.define("sql_execute", {
     // 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).catch(() => {})
+    preValidateSql(args.query, args.warehouse, queryType).catch(() => {})
     // altimate_change end
 
     try {
@@ -115,22 +115,39 @@ interface PreValidationResult {
   error?: string
 }
 
-async function preValidateSql(sql: string, warehouse?: string): Promise<PreValidationResult> {
+async function preValidateSql(sql: string, warehouse: string | undefined, queryType: string): Promise<PreValidationResult> {
   const startTime = Date.now()
+  // Yield the event loop before heavy synchronous SQLite work so concurrent
+  // tasks aren't blocked. Bun's sqlite API is sync and listColumns can touch
+  // hundreds of thousands of rows for large warehouses.
+  await new Promise<void>((resolve) => setImmediate(resolve))
+
+  // Precompute correlation fields used in every telemetry event this function emits.
+  const maskedSqlHash = Telemetry.hashError(Telemetry.maskString(sql))
+
   try {
     // Resolve the warehouse the same way sql.execute's fallback path does:
     // when caller omits `warehouse`, sql.execute uses Registry.list()[0].
     // Matching that here keeps the shadow validation aligned with actual
     // execution (dbt-routed queries are a known gap — they short-circuit
     // before this fallback, so validation may use a different warehouse
     // than the one dbt selects).
+    const registered = Registry.list().warehouses
     let warehouseName = warehouse
     if (!warehouseName) {
-      const registered = Registry.list().warehouses
       warehouseName = registered[0]?.name
     }
+    const warehouseInfo = registered.find((w) => w.name === warehouseName)
+    const warehouseType = warehouseInfo?.type ?? "unknown"
+
+    const ctx: TrackCtx = {
+      warehouse_type: warehouseType,
+      query_type: queryType,
+      masked_sql_hash: maskedSqlHash,
+    }
+
     if (!warehouseName) {
-      trackPreValidation("skipped", "no_cache", 0, Date.now() - startTime, false)
+      trackPreValidation("skipped", "no_cache", 0, Date.now() - startTime, false, ctx)
       return { blocked: false }
     }
 
@@ -139,31 +156,39 @@ async function preValidateSql(sql: string, warehouse?: string): Promise<PreValid
 
     const warehouseStatus = status.warehouses.find((w) => w.name === warehouseName)
     if (!warehouseStatus?.last_indexed) {
-      trackPreValidation("skipped", "no_cache", 0, Date.now() - startTime, false)
+      trackPreValidation("skipped", "no_cache", 0, Date.now() - startTime, false, ctx)
       return { blocked: false }
     }
 
     // Check cache freshness
     const cacheAge = Date.now() - new Date(warehouseStatus.last_indexed).getTime()
     if (cacheAge > CACHE_TTL_MS) {
-      trackPreValidation("skipped", "stale_cache", 0, Date.now() - startTime, false)
+      trackPreValidation("skipped", "stale_cache", 0, Date.now() - startTime, false, ctx)
       return { blocked: false }
     }
 
     // Build schema context from cached columns
     const columns = cache.listColumns(warehouseName, COLUMN_SCAN_LIMIT)
     const schemaTruncated = columns.length >= COLUMN_SCAN_LIMIT
     if (columns.length === 0) {
-      trackPreValidation("skipped", "empty_cache", 0, Date.now() - startTime, false)
+      trackPreValidation("skipped", "empty_cache", 0, Date.now() - startTime, false, ctx)
       return { blocked: false }
     }
 
-    const schemaContext: Record<string, any> = {}
+    // Build schema context keyed by fully-qualified name (database.schema.table)
+    // so multi-database warehouses don't collide on schema+table alone.
+    // Dedupe columns per table to defend against residual collisions.
+    const schemaContext: Record<string, { name: string; type: string; nullable: boolean }[]> = {}
+    const seenColumns: Record<string, Set<string>> = {}
     for (const col of columns) {
-      const tableName = col.schema_name ? `${col.schema_name}.${col.table}` : col.table
+      const tableName = [col.database, col.schema_name, col.table].filter(Boolean).join(".")
+      if (!tableName) continue
       if (!schemaContext[tableName]) {
         schemaContext[tableName] = []
+        seenColumns[tableName] = new Set()
       }
+      if (seenColumns[tableName].has(col.name)) continue
+      seenColumns[tableName].add(col.name)
       schemaContext[tableName].push({
         name: col.name,
         type: col.data_type || "VARCHAR",
@@ -178,60 +203,61 @@ async function preValidateSql(sql: string, warehouse?: string): Promise<PreValid
       schema_context: schemaContext,
     })
 
+    // If the dispatcher itself failed, don't treat missing data as "valid".
+    if (!validationResult.success) {
+      const errMsg = typeof validationResult.error === "string" ? validationResult.error : undefined
+      trackPreValidation("error", "dispatcher_failed", 0, Date.now() - startTime, false, ctx, errMsg)
+      return { blocked: false }
+    }
+
     const data = (validationResult.data ?? {}) as Record<string, any>
     const errors = Array.isArray(data.errors) ? data.errors : []
     const isValid = data.valid !== false && errors.length === 0
 
     if (isValid) {
-      trackPreValidation("passed", "valid", columns.length, Date.now() - startTime, schemaTruncated)
+      trackPreValidation("passed", "valid", columns.length, Date.now() - startTime, schemaTruncated, ctx)
       return { blocked: false }
     }
 
     // Only block on high-confidence structural errors
     const structuralErrors = errors.filter((e: any) => {
       const msg = (e.message ?? "").toLowerCase()
-      return msg.includes("column") || msg.includes("table") || msg.includes("not found") || msg.includes("does not exist")
+      return /\b(column|table|view|relation|identifier|not found|does not exist)\b/.test(msg)
     })
 
     if (structuralErrors.length === 0) {
       // Non-structural errors (ambiguous cases) — let them through
-      trackPreValidation("passed", "non_structural", columns.length, Date.now() - startTime, schemaTruncated)
+      trackPreValidation("passed", "non_structural", columns.length, Date.now() - startTime, schemaTruncated, ctx)
       return { blocked: false }
     }
 
-    // Build helpful error with available columns
     const errorMsgs = structuralErrors.map((e: any) => e.message).join("\n")
-    const referencedTables = Object.keys(schemaContext).slice(0, 10)
-    const availableColumns = referencedTables
-      .map((t) => `${t}: ${schemaContext[t].map((c: any) => c.name).join(", ")}`)
-      .join("\n")
-
-    const errorOutput = [
-      `Pre-execution validation failed (validated against cached schema):`,
-      ``,
-      errorMsgs,
-      ``,
-      `Available tables and columns:`,
-      availableColumns,
-      ``,
-      `Fix the query and retry. If the schema cache is outdated, run schema_index to refresh it.`,
-    ].join("\n")
-
-    trackPreValidation("blocked", "structural_error", columns.length, Date.now() - startTime, schemaTruncated, errorMsgs)
-    return { blocked: true, error: errorOutput }
+    trackPreValidation("blocked", "structural_error", columns.length, Date.now() - startTime, schemaTruncated, ctx, errorMsgs)
+    // Shadow mode: caller discards the result. When blocking is enabled in the
+    // future, build errorOutput here with the structural errors and
+    // schemaContext keys for user-facing guidance.
+    return { blocked: false }
   } catch {
     // Validation failure should never block execution
-    trackPreValidation("error", "validation_exception", 0, Date.now() - startTime, false)
+    const ctx: TrackCtx = { warehouse_type: "unknown", query_type: queryType, masked_sql_hash: maskedSqlHash }
+    trackPreValidation("error", "validation_exception", 0, Date.now() - startTime, false, ctx)
     return { blocked: false }
   }
 }
 
+interface TrackCtx {
+  warehouse_type: string
+  query_type: string
+  masked_sql_hash: string
+}
+
 function trackPreValidation(
   outcome: "skipped" | "passed" | "blocked" | "error",
   reason: string,
   schema_columns: number,
   duration_ms: number,
   schema_truncated: boolean,
+  ctx: TrackCtx,
   error_message?: string,
 ) {
   // Mask schema identifiers (table / column names, paths, user IDs) from the
@@ -244,6 +270,9 @@ function trackPreValidation(
     session_id: Telemetry.getContext().sessionId,
     outcome,
     reason,
+    warehouse_type: ctx.warehouse_type,
+    query_type: ctx.query_type,
+    masked_sql_hash: ctx.masked_sql_hash,
     schema_columns,
     schema_truncated,
     duration_ms,

packages/opencode/src/session/processor.ts

@@ -49,6 +49,14 @@ export namespace SessionProcessor {
     // altimate_change start — per-step generation telemetry
     let stepStartTime = Date.now()
     // altimate_change end
+    // altimate_change start — plan-agent tool-call-refusal detection
+    // Some models (observed: qwen3-coder-next, occasionally gpt-5.4) end plan-agent
+    // steps with finish_reason=stop and never emit tool calls. User abandons the
+    // session thinking it's stuck. Track whether the session has ever produced a
+    // tool call; if plan agent finishes its first step with stop-no-tools, warn.
+    let sessionToolCallsMade = 0
+    let planNoToolWarningEmitted = false
+    // altimate_change end
 
     const result = {
       get message() {
@@ -162,6 +170,9 @@ export namespace SessionProcessor {
                       metadata: value.providerMetadata,
                     })
                     toolcalls[value.toolCallId] = part as MessageV2.ToolPart
+                    // altimate_change start — session has now tool-called; suppresses plan refusal warning
+                    sessionToolCallsMade++
+                    // altimate_change end
 
                     const parts = await MessageV2.parts(input.assistantMessage.id)
                     const lastThree = parts.slice(-DOOM_LOOP_THRESHOLD)
@@ -322,6 +333,52 @@ export namespace SessionProcessor {
                     ...(usage.tokens.cache.write > 0 && { tokens_cache_write: usage.tokens.cache.write }),
                   })
                   // altimate_change end
+                  // altimate_change start — detect plan-agent tool-call refusal
+                  // A plan-agent step that ends with finish=stop and NO tool calls
+                  // (ever) in the session means the model wrote text and gave up.
+                  // Users read the text, see no progress, and abandon. Surface a
+                  // warning + telemetry so the pattern is measurable and the user
+                  // knows to try a different model.
+                  if (
+                    input.assistantMessage.agent === "plan" &&
+                    value.finishReason === "stop" &&
+                    sessionToolCallsMade === 0 &&
+                    !planNoToolWarningEmitted
+                  ) {
+                    planNoToolWarningEmitted = true
+                    Telemetry.track({
+                      type: "plan_no_tool_generation",
+                      timestamp: Date.now(),
+                      session_id: input.sessionID,
+                      message_id: input.assistantMessage.id,
+                      model_id: input.model.id,
+                      provider_id: input.model.providerID,
+                      finish_reason: value.finishReason,
+                      tokens_output: usage.tokens.output,
+                    })
+                    log.warn("plan agent stopped without tool calls — model may not be tool-calling properly", {
+                      sessionID: input.sessionID,
+                      modelID: input.model.id,
+                      providerID: input.model.providerID,
+                      tokensOutput: usage.tokens.output,
+                    })
+                    // synthetic: true so this warning is shown in the TUI but
+                    // excluded when the transcript is replayed to the LLM next turn
+                    // (prompt.ts filters synthetic text parts — see lines 648, 795).
+                    await Session.updatePart({
+                      id: PartID.ascending(),
+                      messageID: input.assistantMessage.id,
+                      sessionID: input.assistantMessage.sessionID,
+                      type: "text",
+                      synthetic: true,
+                      text:
+                        `⚠️ altimate-code: the \`plan\` agent is running on \`${input.model.providerID}/${input.model.id}\`, ` +
+                        `which returned text without calling any tools. If you expected the plan agent to explore the ` +
+                        `codebase, try switching to a model with stronger tool-use via \`/model\`.`,
+                      time: { start: Date.now(), end: Date.now() },
+                    })
+                  }
+                  // altimate_change end
                   await Session.updatePart({
                     id: PartID.ascending(),
                     reason: value.finishReason,

packages/opencode/test/altimate/connections.test.ts

@@ -51,6 +51,36 @@ describe("ConnectionRegistry", () => {
     )
   })
 
+  test("cassandra gives helpful hint instead of generic unsupported error", async () => {
+    Registry.setConfigs({
+      mydb: { type: "cassandra", host: "localhost" },
+    })
+    await expect(Registry.get("mydb")).rejects.toThrow("not yet supported")
+    await expect(Registry.get("mydb")).rejects.toThrow("cqlsh")
+  })
+
+  test("cockroachdb suggests using postgres type", async () => {
+    Registry.setConfigs({
+      mydb: { type: "cockroachdb", host: "localhost" },
+    })
+    await expect(Registry.get("mydb")).rejects.toThrow("postgres")
+  })
+
+  test("timescaledb suggests using postgres type", async () => {
+    Registry.setConfigs({
+      mydb: { type: "timescaledb", host: "localhost" },
+    })
+    await expect(Registry.get("mydb")).rejects.toThrow("postgres")
+  })
+
+  test("truly unknown type gives generic unsupported error with supported list", async () => {
+    Registry.setConfigs({
+      mydb: { type: "neo4j", host: "localhost" },
+    })
+    await expect(Registry.get("mydb")).rejects.toThrow("Unsupported database type")
+    await expect(Registry.get("mydb")).rejects.toThrow("Supported:")
+  })
+
   test("getConfig returns config for known connection", () => {
     Registry.setConfigs({
       mydb: { type: "postgres", host: "localhost" },
@@ -608,6 +638,44 @@ trino_project:
       fs.rmSync(tmpDir, { recursive: true })
     }
   })
+
+  test("clickhouse adapter maps correctly from dbt profiles", async () => {
+    const fs = await import("fs")
+    const os = await import("os")
+    const path = await import("path")
+
+    const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "dbt-test-"))
+    const profilesPath = path.join(tmpDir, "profiles.yml")
+
+    fs.writeFileSync(
+      profilesPath,
+      `
+ch_project:
+  outputs:
+    dev:
+      type: clickhouse
+      host: clickhouse.example.com
+      port: 8443
+      user: default
+      password: secret
+      database: analytics
+      schema: default
+`,
+    )
+
+    try {
+      const connections = await parseDbtProfiles(profilesPath)
+      expect(connections).toHaveLength(1)
+      expect(connections[0].type).toBe("clickhouse")
+      expect(connections[0].config.type).toBe("clickhouse")
+      expect(connections[0].config.host).toBe("clickhouse.example.com")
+      expect(connections[0].config.port).toBe(8443)
+      expect(connections[0].config.user).toBe("default")
+      expect(connections[0].config.database).toBe("analytics")
+    } finally {
+      fs.rmSync(tmpDir, { recursive: true })
+    }
+  })
 })
 
 // ---------------------------------------------------------------------------