fix(chat): ignore stale turn completions after reconnect

.changeset/calm-chat-reconnects.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Browser chats now keep the active turn open across page reloads when older completion records are replayed.

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

@@ -13,6 +13,7 @@ import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.s
 import { stripClientWebhookActionSource } from "~/services/realtime/sanitizeSessionInput.server";
 import {
   claimSessionStreamPart,
+  commitSessionStreamPart,
   drainSessionStreamWaitpoints,
   releaseSessionStreamPart,
 } from "~/services/sessionStreamWaitpointCache.server";
@@ -147,22 +148,65 @@ const { action, loader } = createActionApiRoute(
     const partId = clientPartId ?? nanoid(7);
 
     // Idempotency on client-supplied part ids: atomically claim the id before
-    // appending. A concurrent or retried POST that loses the claim skips the
-    // append (no duplicate record) but still falls through to the drain below,
-    // so a retry whose first attempt died before waking the waitpoint can still
-    // recover it. The claim is released on append failure so a genuine retry
-    // can re-claim and proceed.
-    const wonClaim = clientPartId
+    // appending, then store the resulting sequence with the claim. A retry
+    // returns that original sequence and still falls through to the drain, so
+    // a first attempt that lost its HTTP response cannot lose turn correlation
+    // or strand a waitpoint. The claim is released on append failure.
+    let claim = clientPartId
       ? await claimSessionStreamPart(
           authentication.environment.id,
           addressingKey,
           params.io,
           clientPartId
         )
-      : true;
+      : ({ status: "claimed", claimValue: undefined } as const);
 
-    let appendSeq: number | undefined;
-    if (wonClaim) {
+    let appendSeq = claim.status === "committed" ? claim.seq : undefined;
+    if (claim.status === "pending") {
+      const pendingClaim = claim;
+      const [recoveryError, recoveredSeq] = await tryCatch(
+        realtimeStream.findSessionStreamPartSequence(
+          addressingKey,
+          params.io,
+          partId,
+          pendingClaim.claimedAt
+        )
+      );
+
+      if (recoveryError) {
+        logger.error("Failed to recover pending session stream append", {
+          sessionId: session.id,
+          io: params.io,
+          partId,
+          error: recoveryError,
+        });
+      } else if (recoveredSeq !== undefined) {
+        await commitSessionStreamPart(
+          authentication.environment.id,
+          addressingKey,
+          params.io,
+          partId,
+          pendingClaim.claimValue,
+          recoveredSeq
+        );
+        // The S2 record is durable even if Redis expired or evicted the
+        // pending claim before the best-effort sequence write-back.
+        appendSeq = recoveredSeq;
+        claim = { status: "committed", seq: recoveredSeq };
+      }
+
+      // `.out` data may have been trimmed after it was accepted, so a
+      // successful search with no record preserves the prior no-op response.
+      // A failed search proves nothing and remains retryable for both channels.
+      if (claim.status === "pending" && (params.io === "in" || recoveryError)) {
+        return json(
+          { ok: false, error: "This append is still in progress, please retry." },
+          { status: 409, headers: { "Retry-After": "1" } }
+        );
+      }
+    }
+
+    if (claim.status === "claimed") {
       const [appendError, seq] = await tryCatch(
         realtimeStream.appendPartToSessionStream(part, partId, addressingKey, params.io)
       );
@@ -174,7 +218,8 @@ const { action, loader } = createActionApiRoute(
             authentication.environment.id,
             addressingKey,
             params.io,
-            clientPartId
+            clientPartId,
+            claim.claimValue
           );
         }
         if (appendError instanceof ServiceValidationError) {
@@ -193,6 +238,17 @@ const { action, loader } = createActionApiRoute(
           { status: 500 }
         );
       }
+
+      if (clientPartId && appendSeq !== undefined) {
+        await commitSessionStreamPart(
+          authentication.environment.id,
+          addressingKey,
+          params.io,
+          clientPartId,
+          claim.claimValue,
+          appendSeq
+        );
+      }
     }
 
     // Fire any run-scoped waitpoints registered against this channel. Best

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

@@ -94,6 +94,18 @@ type S2AppendAck = {
   end: { seq_num: number; timestamp: number };
   tail: { seq_num: number; timestamp: number };
 };
+type S2ReadRecord = {
+  body: string;
+  seq_num: number;
+  timestamp: number;
+  headers?: Array<[string, string]>;
+};
+type S2TailResponse = {
+  tail: { seq_num: number; timestamp: number };
+};
+
+const PART_RECOVERY_CLOCK_SKEW_MS = 10_000;
+const PART_RECOVERY_PAGE_SIZE = 1_000;
 
 export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
   private readonly basin: string;
@@ -264,6 +276,102 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     return this.#readRecordsByName(this.toSessionStreamName(friendlyId, io), afterSeqNum);
   }
 
+  /**
+   * Find an append whose Redis claim was left pending after S2 accepted it.
+   * Snapshots the stream tail, then reads forward from the claim timestamp.
+   * The fixed tail keeps an active stream from extending the recovery search.
+   */
+  async findSessionStreamPartSequence(
+    friendlyId: string,
+    io: "out" | "in",
+    partId: string,
+    claimedAt?: number
+  ): Promise<number | undefined> {
+    const s2Stream = this.toSessionStreamName(friendlyId, io);
+    const tailResponse = await fetch(
+      `${this.baseUrl}/streams/${encodeURIComponent(s2Stream)}/records/tail`,
+      {
+        method: "GET",
+        headers: {
+          Authorization: `Bearer ${this.token}`,
+          Accept: "application/json",
+          "S2-Basin": this.basin,
+        },
+      }
+    );
+
+    if (!tailResponse.ok) {
+      if (tailResponse.status === 404) return undefined;
+      const text = await tailResponse.text().catch(() => "");
+      throw new Error(
+        `S2 findSessionStreamPartSequence tail check failed: ${tailResponse.status} ${tailResponse.statusText} ${text}`
+      );
+    }
+
+    const { tail } = (await tailResponse.json()) as S2TailResponse;
+    const snapshotTail = tail.seq_num;
+    if (snapshotTail === 0) return undefined;
+
+    let nextSeq: number | undefined;
+
+    while (true) {
+      const qs = new URLSearchParams();
+      if (nextSeq !== undefined) {
+        qs.set("seq_num", String(nextSeq));
+      } else if (claimedAt !== undefined) {
+        qs.set("timestamp", String(Math.max(0, claimedAt - PART_RECOVERY_CLOCK_SKEW_MS)));
+      } else {
+        qs.set("seq_num", "0");
+      }
+      qs.set("clamp", "true");
+      qs.set("count", String(PART_RECOVERY_PAGE_SIZE));
+      qs.set("bytes", String(S2_MAX_METERED_BYTES));
+      qs.set("wait", "0");
+
+      const res = await fetch(
+        `${this.baseUrl}/streams/${encodeURIComponent(s2Stream)}/records?${qs}`,
+        {
+          method: "GET",
+          headers: {
+            Authorization: `Bearer ${this.token}`,
+            Accept: "application/json",
+            "S2-Format": "raw",
+            "S2-Basin": this.basin,
+          },
+        }
+      );
+
+      if (!res.ok) {
+        if (res.status === 404 || res.status === 416) return undefined;
+        const text = await res.text().catch(() => "");
+        throw new Error(
+          `S2 findSessionStreamPartSequence failed: ${res.status} ${res.statusText} ${text}`
+        );
+      }
+
+      const json = (await res.json()) as {
+        records?: S2ReadRecord[];
+      };
+      const records = json.records ?? [];
+
+      for (const record of records) {
+        if (record.seq_num >= snapshotTail) break;
+        if (record.headers?.[0]?.[0] === "") continue;
+        try {
+          const envelope = JSON.parse(record.body) as { id?: unknown };
+          if (envelope.id === partId) return record.seq_num;
+        } catch {
+          // A malformed or control record cannot be the requested data part.
+        }
+      }
+
+      const lastRecord = records.at(-1);
+      if (!lastRecord) return undefined;
+      nextSeq = lastRecord.seq_num + 1;
+      if (nextSeq >= snapshotTail) return undefined;
+    }
+  }
+
   async #readRecordsByName(s2Stream: string, afterSeqNum?: number): Promise<StreamRecord[]> {
     const startSeq = afterSeqNum != null ? afterSeqNum + 1 : 0;
 
@@ -320,14 +428,7 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
 
       if (eventType === "batch" && data) {
         try {
-          const parsed = JSON.parse(data) as {
-            records: Array<{
-              body: string;
-              seq_num: number;
-              timestamp: number;
-              headers?: Array<[string, string]>;
-            }>;
-          };
+          const parsed = JSON.parse(data) as { records: S2ReadRecord[] };
 
           for (const record of parsed.records) {
             // S2 command records (trim/fence) have a single header with