Merge branch 'main' into sentry-db-unreachable-fingerprint

Author: d-cs

.changeset/cli-deploy-skip-rewrite-timestamp.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Add `TRIGGER_BUILD_SKIP_REWRITE_TIMESTAMP=1` escape hatch for local self-hosted builds whose buildx driver doesn't support `rewrite-timestamp` alongside push (e.g. orbstack's default `docker` driver).

.claude/REVIEW.md

@@ -11,6 +11,7 @@ Reserve 🔴 for things that would page someone or block a rollback. In this cod
   - A Redis data shape used by both versions changes in place. New shapes need a new key namespace.
   - A migration is not backward-compatible with the prior image.
 - **Schema / migration safety.** Prisma migrations must be backward-compatible with the prior deploy. Adding NOT NULL without a default, dropping a column an old image still reads, renaming a column — all 🔴.
+- **ClickHouse migration ordering + idempotency.** Goose runs in strict mode in the deploy pipeline and refuses to apply a missing version below the current version — slotting a new file in below the latest already-applied version blocks the deploy. New ClickHouse migration files MUST use the next available number (`max(files in internal-packages/clickhouse/schema/) + 1`); if main has added migrations while you've been on a branch, renumber yours. DDL must also be idempotent (`ADD COLUMN IF NOT EXISTS`, `DROP COLUMN IF EXISTS`, `CREATE TABLE IF NOT EXISTS`, `ADD INDEX IF NOT EXISTS`) so a partial / `--allow-missing` apply elsewhere doesn't fail on retry. Either fault is 🔴 — both break test/prod deploys. Rules live in `internal-packages/clickhouse/CLAUDE.md`.
 - **Queue / concurrency correctness.** RunQueue, MarQS (V1, legacy), redis-worker — any change to enqueue / dequeue / locking semantics. Re-derive the invariant on paper before flagging or accepting.
 - **Missing index on a hot table.** New Prisma queries against `TaskRun`, `TaskRunExecutionSnapshot`, `JobRun`, `Project`, etc. must use an existing index. Check `internal-packages/database/prisma/schema.prisma` for the relevant `@@index` lines — don't guess and don't propose `EXPLAIN`.
 - **Recovery-path queries.** Any `TaskRun.findFirst` / `findMany` added to a schedule, run-recovery, or restart loop. Recovery fan-outs (Redis crash, restart storms) turn "rare indexed query" into a DB incident. 🔴 even if indexed.

.github/workflows/pr_checks.yml

@@ -17,16 +17,21 @@ jobs:
     name: Detect changes
     runs-on: ubuntu-latest
     outputs:
-      code: ${{ steps.filter.outputs.code }}
+      code: ${{ steps.code_filter.outputs.code }}
+      typecheck_self: ${{ steps.filter.outputs.typecheck_self }}
       webapp: ${{ steps.filter.outputs.webapp }}
       packages: ${{ steps.filter.outputs.packages }}
       internal: ${{ steps.filter.outputs.internal }}
       cli: ${{ steps.filter.outputs.cli }}
       sdk: ${{ steps.filter.outputs.sdk }}
     steps:
+      # `code` uses `every` semantics so the negation patterns actually subtract.
+      # With the default `some` quantifier, `**` matches every file and the
+      # subsequent `!...` patterns are no-ops (each pattern is OR'd, not AND'd).
       - uses: dorny/paths-filter@fbd0ab8f3e69293af611ebaee6363fc25e6d187d # v4.0.1
-        id: filter
+        id: code_filter
         with:
+          predicate-quantifier: every
           filters: |
             code:
               - '**'
@@ -37,6 +42,11 @@ jobs:
               - '!references/**'
               - '!**/*.md'
               - '!**/.env.example'
+      - uses: dorny/paths-filter@fbd0ab8f3e69293af611ebaee6363fc25e6d187d # v4.0.1
+        id: filter
+        with:
+          filters: |
+            typecheck_self:
               - '.github/workflows/pr_checks.yml'
               - '.github/workflows/typecheck.yml'
             webapp:
@@ -95,7 +105,7 @@ jobs:
 
   typecheck:
     needs: changes
-    if: needs.changes.outputs.code == 'true'
+    if: needs.changes.outputs.code == 'true' || needs.changes.outputs.typecheck_self == 'true'
     uses: ./.github/workflows/typecheck.yml
 
   webapp:

internal-packages/clickhouse/CLAUDE.md

@@ -4,18 +4,46 @@
 
 ## Migrations
 
-Goose-format SQL migrations live in `schema/`. Create new numbered files:
+Goose-format SQL migrations live in `schema/`. Two rules below are load-bearing — both can block a deploy.
+
+### Rule 1: number to `max + 1`, never slot in
+
+Goose runs in strict mode in the deploy pipeline. If a migration file numbered *below* the version currently recorded in `goose_db_version` ever shows up, goose refuses to apply it and the deploy fails:
+
+```
+goose run: error: found 1 missing migrations before current version 30:
+  version 29: 029_add_task_kind_to_task_runs_v2.sql
+```
+
+When adding a migration:
+
+1. Look at `schema/` and take the largest existing number, call it `N`.
+2. Name your file `0(N+1)_descriptive_name.sql`.
+3. If you've been on a branch while main added migrations, **rebase and renumber** before opening the PR — a file numbered below the new max will block the next deploy after your PR merges.
+
+### Rule 2: DDL must be idempotent
+
+Migrations can be applied out of order in some environments (`goose up --allow-missing` for local recovery, manual fixups, etc.) and may be retried. Always use idempotent forms so a re-apply is a no-op:
 
 ```sql
 -- +goose Up
 ALTER TABLE trigger_dev.your_table
-ADD COLUMN new_column String DEFAULT '';
+  ADD COLUMN IF NOT EXISTS new_column String DEFAULT '';
 
 -- +goose Down
 ALTER TABLE trigger_dev.your_table
-DROP COLUMN new_column;
+  DROP COLUMN IF EXISTS new_column;
 ```
 
+Equivalent forms for other DDL:
+
+- `CREATE TABLE IF NOT EXISTS …`
+- `DROP TABLE IF EXISTS …`
+- `ADD INDEX IF NOT EXISTS …` / `DROP INDEX IF EXISTS …`
+- `CREATE MATERIALIZED VIEW IF NOT EXISTS …` / `DROP VIEW IF EXISTS …`
+
+ClickHouse supports `IF [NOT] EXISTS` on all of the above. Older migrations in this directory predate the rule and are not idempotent — leave them as-is unless you're explicitly hardening one.
+
 ## Naming Conventions
 
 - `raw_` prefix for input tables (where data lands first)

internal-packages/clickhouse/schema/029_add_task_kind_to_task_runs_v2.sql

@@ -0,0 +1,12 @@
+-- +goose Up
+-- IF NOT EXISTS is required because this migration was previously numbered
+-- 029 and may have been applied in environments where goose accepted it
+-- before 030_create_sessions_v1 advanced the version counter. Renaming to
+-- 031 makes goose treat this as new everywhere, so the DDL must tolerate
+-- the column already being present.
+ALTER TABLE trigger_dev.task_runs_v2
+  ADD COLUMN IF NOT EXISTS task_kind LowCardinality(String) DEFAULT '';
+
+-- +goose Down
+ALTER TABLE trigger_dev.task_runs_v2
+  DROP COLUMN IF EXISTS task_kind;

internal-packages/clickhouse/schema/031_add_task_kind_to_task_runs_v2.sql

@@ -1152,7 +1152,14 @@ function getOutputOptions({
     return outputOptions;
   }
 
-  const outputOptions: string[] = ["type=image", "oci-mediatypes=true", "rewrite-timestamp=true"];
+  // `rewrite-timestamp` is incompatible with the buildx docker driver's
+  // implicit `unpack=true` on push (used by e.g. orbstack's default builder).
+  // Provide an env-var escape hatch so local-dev deploys can opt out.
+  const skipRewriteTimestamp = process.env.TRIGGER_BUILD_SKIP_REWRITE_TIMESTAMP === "1";
+  const outputOptions: string[] = ["type=image", "oci-mediatypes=true"];
+  if (!skipRewriteTimestamp) {
+    outputOptions.push("rewrite-timestamp=true");
+  }
 
   if (imageTag) {
     outputOptions.push(`name=${imageTag}`);

packages/cli-v3/src/deploy/buildImage.ts

@@ -568,6 +568,40 @@ describe("TriggerChatTransport", () => {
       expect(subscribe!).toContain("/realtime/v1/sessions/chat-by-chatid/out");
     });
 
+    it("routes .out SSE through streamBaseURL while appends stay on baseURL", async () => {
+      const requests: string[] = [];
+      global.fetch = vi.fn().mockImplementation(async (url: string | URL) => {
+        const urlStr = typeof url === "string" ? url : url.toString();
+        requests.push(urlStr);
+        if (isSessionStreamAppendUrl(urlStr)) return defaultAppendResponse();
+        if (isSessionOutSubscribeUrl(urlStr)) return defaultSseResponse();
+        throw new Error(`Unexpected URL: ${urlStr}`);
+      });
+
+      const transport = new TriggerChatTransport({
+        task: "my-chat-task",
+        accessToken: () => "pat",
+        baseURL: "https://api.test.trigger.dev",
+        streamBaseURL: "https://chat-proxy.example.com",
+        sessions: { "chat-split": { publicAccessToken: "p" } },
+      });
+
+      const stream = await transport.sendMessages({
+        trigger: "submit-message",
+        chatId: "chat-split",
+        messageId: undefined,
+        messages: [createUserMessage("Hi")],
+        abortSignal: undefined,
+      });
+      await drainChunks(stream);
+
+      const append = requests.find(isSessionStreamAppendUrl);
+      const subscribe = requests.find(isSessionOutSubscribeUrl);
+      expect(append!.startsWith("https://api.test.trigger.dev/")).toBe(true);
+      expect(subscribe!.startsWith("https://chat-proxy.example.com/")).toBe(true);
+      expect(subscribe!).toContain("/realtime/v1/sessions/chat-split/out");
+    });
+
     it("for submit-message, only the latest message is delivered to .in", async () => {
       // Slim wire: each `.in/append` carries at most ONE new message in
       // `payload.message` (singular). Even if the caller hands sendMessages

packages/trigger-sdk/src/v3/chat.test.ts

@@ -225,6 +225,15 @@ export type TriggerChatTransportOptions<TClientData = unknown> = {
   /** Base URL for the Trigger.dev API. @default "https://api.trigger.dev" */
   baseURL?: string;
 
+  /**
+   * Base URL for the SSE stream subscription only (`GET .../sessions/{chatId}/out`).
+   * Falls back to `baseURL` when unset. Set this to route the long-lived
+   * stream through a custom proxy (e.g. a Cloudflare worker capturing JA4
+   * fingerprints for bot detection) while keeping append POSTs direct to
+   * `baseURL` to avoid an extra hop on every user message.
+   */
+  streamBaseURL?: string;
+
   /** Additional headers included in every API request. */
   headers?: Record<string, string>;
 
@@ -346,6 +355,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     | ((params: StartSessionParams<Record<string, unknown>>) => Promise<StartSessionResult>)
     | undefined;
   private readonly baseURL: string;
+  private readonly streamBaseURL: string;
   private readonly extraHeaders: Record<string, string>;
   private readonly streamTimeoutSeconds: number;
   private defaultMetadata: Record<string, unknown> | undefined;
@@ -367,6 +377,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       | ((params: StartSessionParams<Record<string, unknown>>) => Promise<StartSessionResult>)
       | undefined;
     this.baseURL = options.baseURL ?? DEFAULT_BASE_URL;
+    this.streamBaseURL = options.streamBaseURL ?? this.baseURL;
     this.extraHeaders = options.headers ?? {};
     this.streamTimeoutSeconds = options.streamTimeoutSeconds ?? DEFAULT_STREAM_TIMEOUT_SECONDS;
     this.defaultMetadata = options.clientData;
@@ -1021,7 +1032,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       );
     }
 
-    const streamUrl = `${this.baseURL}/realtime/v1/sessions/${encodeURIComponent(chatId)}/out`;
+    const streamUrl = `${this.streamBaseURL}/realtime/v1/sessions/${encodeURIComponent(chatId)}/out`;
 
     return new ReadableStream<UIMessageChunk>({
       start: async (controller) => {