fix(trigger): let workers see that Trigger.dev is available (#6869)

* fix(trigger): let workers see that Trigger.dev is available

Workers run Trigger.dev by definition, but the flag saying so is read from the
environment and had only ever been set on the app container. isTriggerAvailable()
was therefore false inside every task run, so work a task dispatched silently
took the in-process fallback instead of the queue it was written for.

Document processing is where this showed: a connector sync chunked and embedded
its documents itself, five at a time, rather than handing them to the
document-processing queue. The queue's concurrency limit, the per-document task's
machine, retry policy and duration budget all sat unused, and a sync with
thousands of documents ran until it hit its own max duration. It also explains
why that task has no runs for connector-synced knowledge bases at all.

Asserting the flag here is safe because the same check still requires
TRIGGER_SECRET_KEY, which only the Trigger.dev runtime provides: anywhere
dispatching is not actually possible the flag stays ineffective and behaviour is
unchanged.

Dispatch failure is now recoverable rather than silent. Only a total failure
raised before, so one failed batch left its documents at pending with nothing
recording why. Those are processed in-process instead, which costs the caller
the time it hoped to hand to the queue but does not drop the work. That path was
unreachable from a worker until this change made dispatching happen there.

* refactor: tighten the comments on this change and the quota classification

Both sets explained the incident that motivated the code rather than the code
itself. That kind of narrative stops being true as the surrounding system moves
and starts misleading instead, so each is cut back to the reason a reader needs.

Author: waleedlatif1

apps/sim/lib/embeddings/client.test.ts

@@ -823,11 +823,8 @@ describe('knowledge embedding transport fallback', () => {
   })
 
   /**
-   * OpenAI returns 429 for an exhausted balance as well as for a rate limit, but
-   * only one of them reopens. Retrying a spent account cannot succeed, and since
-   * the sweep re-queues failed documents every sync it turns into permanent load
-   * — this was observed burning every attempt on thousands of documents for
-   * weeks against an account with no credit.
+   * A spent account never reopens, and the sweep re-queues failed documents every
+   * sync — so retrying one burns the budget per document, indefinitely.
    */
   it('does not retry a 429 that reports an exhausted balance', async () => {
     vi.useFakeTimers()

apps/sim/lib/embeddings/client.ts

@@ -79,10 +79,7 @@ const EMBEDDING_RETRY_BUDGET_MS = EMBEDDING_MAX_RETRIES * EMBEDDING_MAX_RETRY_DE
 export class EmbeddingAPIError extends Error {
   public status: number
 
-  /**
-   * The provider rejected this for an exhausted balance rather than a rate that
-   * will recover. Both arrive as 429.
-   */
+  /** Rejected for an exhausted balance rather than a recoverable rate. Both are 429. */
   public quotaExhausted?: boolean
 
   /**
@@ -100,13 +97,8 @@ export class EmbeddingAPIError extends Error {
 
 /**
  * True when a rejection body reports an exhausted balance rather than a rate
- * limit.
- *
- * OpenAI returns 429 for both, but only one of them reopens. `insufficient_quota`
- * stands until somebody adds credit, so retrying it cannot succeed no matter how
- * long the loop waits — and because a failed document is re-queued by the sweep
- * on every sync, an account that has run out turns into a permanent load: the
- * budget is spent per document, per attempt, forever.
+ * limit. OpenAI returns 429 for both, but only a rate limit reopens: a spent
+ * account stands until someone adds credit, so retrying it cannot succeed.
  */
 function isQuotaExhaustionBody(errorText: string): boolean {
   try {
@@ -148,13 +140,10 @@ function statedWaitOutlastsBudget(error: unknown): boolean {
 }
 
 /**
- * Whether another attempt against the same provider could plausibly succeed.
- *
- * Deliberately narrower than {@link isTransientEmbeddingError}, which also
- * decides whether the fallback chain should try a *different* provider. Those
- * two questions differ: an exhausted balance rules out the key we just used, but
- * says nothing about the next one in the chain, so a quota rejection stops the
- * retries here while remaining eligible for failover.
+ * Whether another attempt against the *same* provider could succeed. Narrower
+ * than {@link isTransientEmbeddingError}, which decides whether to fail over to a
+ * different one: an exhausted balance rules out the key just used but says
+ * nothing about the next in the chain.
  */
 function isWorthRetrying(error: unknown): boolean {
   if (!isTransientEmbeddingError(error)) return false

apps/sim/lib/knowledge/documents/service.ts

@@ -721,6 +721,7 @@ async function dispatchViaBatchTrigger(
 ): Promise<number> {
   let dispatched = 0
   const batchIds: string[] = []
+  const undispatched: DocumentProcessingPayload[] = []
   const region = await resolveTriggerRegion()
   for (let i = 0; i < jobPayloads.length; i += TRIGGER_BATCH_SIZE) {
     const chunk = jobPayloads.slice(i, i + TRIGGER_BATCH_SIZE)
@@ -747,11 +748,25 @@ async function dispatchViaBatchTrigger(
       logger.error(`[${requestId}] Failed to batchTrigger ${chunk.length} document jobs`, {
         error: getErrorMessage(error),
       })
+      undispatched.push(...chunk)
     }
   }
   if (batchIds.length > 0) {
     logger.info(`[${requestId}] Trigger.dev batches dispatched`, { batchIds })
   }
+
+  /**
+   * Only a total dispatch failure raises, so a chunk failing alone would leave its
+   * documents at `pending` with nothing recording why. Processing them here is
+   * slower than the queue but does not drop the work.
+   */
+  if (undispatched.length > 0) {
+    logger.warn(
+      `[${requestId}] Processing ${undispatched.length} documents in-process after failed enqueue`
+    )
+    dispatched += await dispatchInProcess(undispatched, requestId)
+  }
+
   return dispatched
 }
 

apps/sim/trigger.config.ts

@@ -73,7 +73,17 @@ export default defineConfig({
       '@daytona/sdk',
     ],
     extensions: [
-      syncEnvVars(() => [{ name: 'DB_APP_NAME', value: 'sim-trigger' }]),
+      syncEnvVars(() => [
+        { name: 'DB_APP_NAME', value: 'sim-trigger' },
+        /**
+         * Workers run Trigger.dev by definition, but the flag saying so was only
+         * set on the app container, so `isTriggerAvailable()` was false in every
+         * task run and dispatched work silently took the in-process fallback.
+         * Ineffective where dispatching is impossible: the check also requires
+         * TRIGGER_SECRET_KEY, which only the Trigger.dev runtime provides.
+         */
+        { name: 'TRIGGER_DEV_ENABLED', value: 'TRUE' },
+      ]),
       additionalFiles({
         files: [
           './lib/execution/isolated-vm-worker.cjs',