fix(copilot): make createRunSegment idempotent on stream_id, surface swallowed PG errors

apps/sim/lib/copilot/async-runs/repository.ts

@@ -87,7 +87,12 @@ export async function createRunSegment(input: CreateRunSegmentInput) {
       [TraceAttr.CopilotRunStatus]: input.status ?? 'active',
     },
     async () => {
-      const [run] = await db
+      // stream_id is UNIQUE and the headless path derives it from the
+      // message id with a fresh run id per attempt, so a client retry of
+      // the same message replays this insert. Treat the replay as "this
+      // segment already exists": hand back the original row so the retry
+      // continues under the same run identity instead of failing.
+      const [inserted] = await db
         .insert(copilotRuns)
         .values({
           ...(input.id ? { id: input.id } : {}),
@@ -104,8 +109,15 @@ export async function createRunSegment(input: CreateRunSegmentInput) {
           requestContext: input.requestContext ?? {},
           status: input.status ?? 'active',
         })
+        .onConflictDoNothing({ target: copilotRuns.streamId })
         .returning()
-      return run
+      if (inserted) return inserted
+      const [existing] = await db
+        .select()
+        .from(copilotRuns)
+        .where(eq(copilotRuns.streamId, input.streamId))
+        .limit(1)
+      return existing
     }
   )
 }

apps/sim/lib/copilot/generated/metrics-v1.ts

@@ -14,6 +14,7 @@
 // union is queryable as one series set.
 
 export const Metric = {
+  CopilotAuthRejections: 'copilot.auth.rejections',
   CopilotCacheAttempted: 'copilot.cache.attempted',
   CopilotCacheHit: 'copilot.cache.hit',
   CopilotCacheWrite: 'copilot.cache.write',
@@ -39,6 +40,7 @@ export type MetricValue = (typeof Metric)[MetricKey]
 
 /** Readonly sorted list of every canonical mothership metric name. */
 export const MetricValues: readonly MetricValue[] = [
+  'copilot.auth.rejections',
   'copilot.cache.attempted',
   'copilot.cache.hit',
   'copilot.cache.write',

apps/sim/lib/copilot/generated/trace-attribute-values-v1.ts

@@ -34,6 +34,17 @@ export const AuthKeyMatch = {
 export type AuthKeyMatchKey = keyof typeof AuthKeyMatch
 export type AuthKeyMatchValue = (typeof AuthKeyMatch)[AuthKeyMatchKey]
 
+export const AuthRejectReason = {
+  Forbidden: 'forbidden',
+  InvalidKey: 'invalid_key',
+  Other: 'other',
+  RateLimited: 'rate_limited',
+  UsageLimit: 'usage_limit',
+} as const
+
+export type AuthRejectReasonKey = keyof typeof AuthRejectReason
+export type AuthRejectReasonValue = (typeof AuthRejectReason)[AuthRejectReasonKey]
+
 export const BillingAnalyticsOutcome = {
   Duplicate: 'duplicate',
   RetriesExhausted: 'retries_exhausted',

apps/sim/lib/copilot/generated/trace-attributes-v1.ts

@@ -52,6 +52,7 @@ export const TraceAttr = {
   AuthKeySource: 'auth.key.source',
   AuthKeyType: 'auth.key.type',
   AuthProvider: 'auth.provider',
+  AuthRejectReason: 'auth.reject_reason',
   AuthValidateStatusCode: 'auth.validate.status_code',
   AwsRegion: 'aws.region',
   BillingAttempts: 'billing.attempts',
@@ -679,6 +680,7 @@ export const TraceAttrValues: readonly TraceAttrValue[] = [
   'auth.key.source',
   'auth.key.type',
   'auth.provider',
+  'auth.reject_reason',
   'auth.validate.status_code',
   'aws.region',
   'billing.attempts',

apps/sim/lib/copilot/request/lifecycle/run.ts

@@ -1,6 +1,6 @@
 import type { Context } from '@opentelemetry/api'
 import { createLogger } from '@sim/logger'
-import { toError } from '@sim/utils/errors'
+import { getErrorMessage, toError } from '@sim/utils/errors'
 import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { isWorkspaceOnEnterprisePlan } from '@/lib/billing/core/subscription'
@@ -949,7 +949,7 @@ async function ensureHeadlessRunIdentity(input: {
   const runId = generateId()
 
   try {
-    await createRunSegment({
+    const run = await createRunSegment({
       id: runId,
       executionId,
       chatId: input.chatId,
@@ -964,12 +964,18 @@ async function ensureHeadlessRunIdentity(input: {
         source: 'headless_lifecycle',
       },
     })
-    return { executionId, runId }
+    // A retried message resolves to the pre-existing segment (stream_id is
+    // unique per message) — continue under ITS identity, not the fresh ids.
+    return { executionId: run?.executionId ?? executionId, runId: run?.id ?? runId }
   } catch (error) {
+    // Drizzle's "Failed query" message drops the Postgres error — the
+    // violated constraint / detail lives on `cause`.
+    const cause = toError(error).cause
     logger.warn('Failed to create headless run identity', {
       chatId: input.chatId,
       messageId: input.messageId,
       error: toError(error).message,
+      cause: cause === undefined ? undefined : getErrorMessage(cause),
     })
     return {}
   }

apps/sim/lib/copilot/request/lifecycle/start.ts

@@ -2,7 +2,7 @@ import { type Context, context as otelContextApi } from '@opentelemetry/api'
 import { db } from '@sim/db'
 import { copilotChats } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { getErrorMessage } from '@sim/utils/errors'
+import { getErrorMessage, toError } from '@sim/utils/errors'
 import { eq } from 'drizzle-orm'
 import { createRunSegment } from '@/lib/copilot/async-runs/repository'
 import { chatPubSub } from '@/lib/copilot/chat-status'
@@ -203,8 +203,12 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
               provider: (requestPayload.provider as string | undefined) || null,
               requestContext: { requestId },
             }).catch((error) => {
+              // Drizzle's "Failed query" message drops the Postgres error —
+              // the violated constraint / detail lives on `cause`.
+              const cause = toError(error).cause
               logger.warn(`[${requestId}] Failed to create copilot run segment`, {
                 error: getErrorMessage(error),
+                cause: cause === undefined ? undefined : getErrorMessage(cause),
               })
             })
           }