feat(chat.agent): delta-only wire and history reconstruction at run boot

The wire used to ship the full UIMessage array on every send. Long
chats with tool results crossed the 512 KiB body cap on
/realtime/v1/sessions/{id}/in/append around turn 10-30 and stalled.

Each .in/append now carries at most one new message. The agent rebuilds
prior history at run boot from a JSON snapshot in object storage plus
a wait=0 replay of the session.out tail. Snapshot writes are awaited
after every onTurnComplete so they survive idle suspend; reads happen
only at boot. Registering hydrateMessages short-circuits both paths.

SDK:
- ChatTaskWirePayload drops `messages`, adds `message` + `headStartMessages`
- run boot reconstructs the accumulator from snapshot + replay
- per-turn handling does delta merge, not full-history reset
- chat.headStart bridges the first-ever turn via headStartMessages
- onValidateMessages skipped when no incoming message
- transports, mid-stream pendingMessages, head-start route handler updated

Server:
- new /realtime/v1/sessions/:session/:io/records endpoint (wait=0 drain)
- ensureRunForSession sets continuation: true and previousRunId on
  cross-run continuations so the boot gate triggers snapshot read
- /api/v1/packets/ and /api/v2/packets/ exempt from customer rate limits
- s2Append retries transient failures with exponential backoff and
  surfaces undici cause codes in the diagnostic log

References + tests:
- 4 new SDK unit tests, 2 new webapp integration tests
- mockChatAgent harness updated to the new wire
- new reference tools: getCurrentTime, searchHackerNews, createGithubIssue
- chat.tsx bridge exposes regenerate and error for smoke-test capture
- chat-client-burst-test orchestrator for concurrent-send stress

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: ericallam

.changeset/chat-agent-delta-wire-snapshots.md

@@ -0,0 +1,8 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+`chat.agent` wire is now delta-only — clients ship at most one new message per `.in/append` instead of the full `UIMessage[]` history. The agent rebuilds prior history at run boot from a JSON snapshot in object storage plus a `wait=0` replay of the `session.out` tail. Long chats stop hitting the 512 KiB body cap on `/realtime/v1/sessions/{id}/in/append`. Snapshot writes happen after every `onTurnComplete`, awaited so they survive idle suspend; reads happen only at run boot. Registering a `hydrateMessages` hook short-circuits both the snapshot read/write and the replay — the customer is the source of truth for history.
+
+Custom transports that constructed `ChatTaskWirePayload` directly need to drop the `messages: UIMessage[]` field and use `message?: UIMessage` (singular). Built-in transports (`TriggerChatTransport`, `AgentChat`) handle the change below the customer-facing surface — most apps need no changes. Configure object-store env vars (`OBJECT_STORE_*`) on your webapp deployment if you haven't already; without an object store and without `hydrateMessages`, conversations don't survive run boundaries.

apps/webapp/app/routes/realtime.v1.sessions.$session.$io.records.ts

@@ -0,0 +1,97 @@
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import {
+  canonicalSessionAddressingKey,
+  isSessionFriendlyIdForm,
+  resolveSessionByIdOrExternalId,
+} from "~/services/realtime/sessions.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+const ParamsSchema = z.object({
+  session: z.string(),
+  io: z.enum(["out", "in"]),
+});
+
+const SearchSchema = z.object({
+  // S2 sequence number — same cursor format as the SSE Last-Event-ID
+  // (the SSE `id:` field on session-channel events is the seq_num,
+  // stringified). Records returned have `seqNum > afterEventId`.
+  afterEventId: z.string().regex(/^\d+$/).optional(),
+});
+
+// GET: non-SSE, `wait=0` drain of a session channel. Returns a JSON body
+// `{ records: StreamRecord[] }` with whatever records exist after
+// `afterEventId` (or from the head if absent) and closes immediately.
+//
+// Used by the SDK's `replaySessionOutTail` at run boot — the SSE long-poll
+// path costs ~1s per fresh chat (the timeout duration) regardless of stream
+// content, which is unacceptable on the first-message TTFC budget. This
+// route gives the agent a cheap "what's there right now" peek instead.
+//
+// Same row-optional addressing as the SSE GET route in `…$io.ts`: we
+// resolve via `resolveSessionByIdOrExternalId` and only 404 for opaque
+// `session_*` friendlyIds (which must reference a real row). External-id
+// form falls through with `row: null` so the boot path doesn't 404 on a
+// fresh chat that hasn't written its first chunk yet.
+const loader = createLoaderApiRoute(
+  {
+    params: ParamsSchema,
+    searchParams: SearchSchema,
+    allowJWT: true,
+    corsStrategy: "all",
+    findResource: async (params, auth) => {
+      const row = await resolveSessionByIdOrExternalId(
+        $replica,
+        auth.environment.id,
+        params.session
+      );
+      if (!row && isSessionFriendlyIdForm(params.session)) {
+        return undefined;
+      }
+      return {
+        row,
+        addressingKey: canonicalSessionAddressingKey(row, params.session),
+      };
+    },
+    authorization: {
+      action: "read",
+      resource: ({ row, addressingKey }) => {
+        const ids = new Set<string>([addressingKey]);
+        if (row) {
+          ids.add(row.friendlyId);
+          if (row.externalId) ids.add(row.externalId);
+        }
+        return { sessions: [...ids] };
+      },
+      superScopes: ["read:sessions", "read:all", "admin"],
+    },
+  },
+  async ({ params, authentication, resource, searchParams }) => {
+    const realtimeStream = getRealtimeStreamInstance(authentication.environment, "v2", {
+      session: resource.row,
+      organization: resource.row ? null : authentication.environment.organization,
+    });
+
+    if (!(realtimeStream instanceof S2RealtimeStreams)) {
+      return new Response("Session channels require the S2 realtime backend", {
+        status: 501,
+      });
+    }
+
+    const afterSeqNum =
+      searchParams.afterEventId !== undefined ? Number(searchParams.afterEventId) : undefined;
+
+    const records = await realtimeStream.readSessionStreamRecords(
+      resource.addressingKey,
+      params.io,
+      afterSeqNum
+    );
+
+    return json({ records });
+  }
+);
+
+export { loader };

apps/webapp/app/services/apiRateLimit.server.ts

@@ -63,6 +63,13 @@ export const apiRateLimiter = authorizationRateLimitMiddleware({
     /^\/api\/v1\/runs\/[^\/]+\/attempts$/, // /api/v1/runs/$runFriendlyId/attempts
     /^\/api\/v1\/waitpoints\/tokens\/[^\/]+\/callback\/[^\/]+$/, // /api/v1/waitpoints/tokens/$waitpointFriendlyId/callback/$hash
     /^\/api\/v\d+\/deployments/, // /api/v{1,2,3,n}/deployments/*
+    // Internal SDK plumbing — packets are presigned-URL handshakes for
+    // payload uploads (v2 PUT) and downloads (v1 GET), authenticated via
+    // run-scoped JWT, called once per task/turn boundary by the runtime.
+    // Same shape as `/api/v1/runs/$runFriendlyId/attempts` above; not a
+    // customer-facing surface so customer rate limits shouldn't apply.
+    /^\/api\/v1\/packets\//,
+    /^\/api\/v2\/packets\//,
   ],
   log: {
     rejections: env.API_RATE_LIMIT_REJECTION_LOGS_ENABLED === "1",

apps/webapp/app/services/realtime/s2realtimeStreams.server.ts

@@ -441,8 +441,16 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
 
   // ---------- Internals: S2 REST ----------
   private async s2Append(stream: string, body: S2AppendInput): Promise<S2AppendAck> {
-    // POST /v1/streams/{stream}/records (JSON)
-    const res = await fetch(`${this.baseUrl}/streams/${encodeURIComponent(stream)}/records`, {
+    // POST /v1/streams/{stream}/records (JSON).
+    //
+    // Retries transient failures (network errors and 5xx) up to 3 times with
+    // exponential backoff. Undici's "fetch failed" errors observed locally
+    // are pre-connection (DNS/TCP) so the request never reaches S2, making
+    // retry safe — the alternative is a 500 surfacing to the SDK transport,
+    // which then retries the whole `/in/append` round-trip and pollutes
+    // logs. 4xx are not retried (genuine client errors).
+    const url = `${this.baseUrl}/streams/${encodeURIComponent(stream)}/records`;
+    const init: RequestInit = {
       method: "POST",
       headers: {
         Authorization: `Bearer ${this.token}`,
@@ -451,12 +459,60 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
         "S2-Basin": this.basin,
       },
       body: JSON.stringify(body),
-    });
-    if (!res.ok) {
-      const text = await res.text().catch(() => "");
-      throw new Error(`S2 append failed: ${res.status} ${res.statusText} ${text}`);
+    };
+
+    const maxAttempts = 3;
+    const backoffsMs = [100, 250, 600];
+    let lastError: unknown;
+
+    for (let attempt = 0; attempt < maxAttempts; attempt++) {
+      // The `try` only wraps `fetch` — once we have a Response we handle status
+      // outside the catch, so a 4xx throw can't be swallowed and retried.
+      let res: Response | undefined;
+      try {
+        res = await fetch(url, init);
+      } catch (err) {
+        lastError = err;
+      }
+
+      if (res) {
+        if (res.ok) {
+          return (await res.json()) as S2AppendAck;
+        }
+        const text = await res.text().catch(() => "");
+        const httpError = new Error(
+          `S2 append failed: ${res.status} ${res.statusText} ${text}`
+        );
+        if (res.status >= 400 && res.status < 500) {
+          // 4xx — caller-side problem (auth, malformed body, closed stream).
+          // Retrying won't help.
+          throw httpError;
+        }
+        // 5xx — retryable.
+        lastError = httpError;
+      }
+
+      const isLastAttempt = attempt === maxAttempts - 1;
+      const diagnostics = describeFetchError(lastError);
+      if (isLastAttempt) {
+        this.logger.error("S2 append failed after retries", {
+          stream,
+          attempts: maxAttempts,
+          ...diagnostics,
+        });
+        break;
+      }
+
+      this.logger.warn("S2 append transient failure, retrying", {
+        stream,
+        attempt: attempt + 1,
+        nextDelayMs: backoffsMs[attempt],
+        ...diagnostics,
+      });
+      await new Promise((resolve) => setTimeout(resolve, backoffsMs[attempt]));
     }
-    return (await res.json()) as S2AppendAck;
+
+    throw lastError instanceof Error ? lastError : new Error(String(lastError));
   }
 
   private async getS2AccessToken(id: string): Promise<string> {
@@ -560,3 +616,40 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     return Number.isFinite(n) && n >= 0 ? n + 1 : undefined;
   }
 }
+
+// Pulls the underlying network error out of undici's generic "fetch failed".
+// undici sets `error.cause` to either a SystemError-shaped object with `code`
+// (e.g. `ECONNRESET`, `UND_ERR_SOCKET`, `ETIMEDOUT`), `errno`, and `syscall`,
+// or — for happy-eyeballs / multi-address connect attempts — an
+// `AggregateError` whose `errors[]` each carry their own code. Surfacing
+// those tells us whether failures are pre-connection (DNS / TCP), mid-stream
+// socket resets, or genuine S2 server errors.
+function describeFetchError(err: unknown): Record<string, unknown> {
+  if (!(err instanceof Error)) {
+    return { error: String(err) };
+  }
+  const out: Record<string, unknown> = {
+    error: err.message,
+    name: err.name,
+  };
+  const cause = (err as { cause?: unknown }).cause;
+  if (cause && typeof cause === "object") {
+    const c = cause as Record<string, unknown>;
+    if (typeof c.code === "string") out.causeCode = c.code;
+    if (typeof c.errno === "number" || typeof c.errno === "string") out.causeErrno = c.errno;
+    if (typeof c.syscall === "string") out.causeSyscall = c.syscall;
+    if (typeof c.message === "string") out.causeMessage = c.message;
+    if (Array.isArray(c.errors)) {
+      out.causeErrors = c.errors
+        .filter((e: unknown): e is Error => e instanceof Error)
+        .map((e) => ({
+          message: e.message,
+          code: (e as { code?: unknown }).code,
+          syscall: (e as { syscall?: unknown }).syscall,
+          address: (e as { address?: unknown }).address,
+          port: (e as { port?: unknown }).port,
+        }));
+    }
+  }
+  return out;
+}