fix(react-hooks,sdk): review + CI fixes for session channels

- useSessionStreamChannel: only clear the abort-controller ref when it still
  points at the current request, so a finishing older request can't strand a
  newer subscription (CodeRabbit).
- Drop the unenforced keepLastN retention option until producer-side trim lands.
- Update the initializeSessionStream call-arity assertions in sessions.test.ts.
- Formatting.

Author: ericallam

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

@@ -100,7 +100,13 @@ const { action, loader } = createActionApiRoute(
     let appendSeq: number | undefined;
     if (wonClaim) {
       const [appendError, seq] = await tryCatch(
-        realtimeStream.appendPartToSessionStream(part, partId, addressingKey, params.io, params.channel)
+        realtimeStream.appendPartToSessionStream(
+          part,
+          partId,
+          addressingKey,
+          params.io,
+          params.channel
+        )
       );
       appendSeq = seq ?? undefined;
 
@@ -125,7 +131,10 @@ const { action, loader } = createActionApiRoute(
           channel: params.channel,
           error: appendError,
         });
-        return json({ ok: false, error: "Something went wrong, please try again." }, { status: 500 });
+        return json(
+          { ok: false, error: "Something went wrong, please try again." },
+          { status: 500 }
+        );
       }
     }
 

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

@@ -78,9 +78,7 @@ const { action } = createActionApiRoute(
 
     const addressingKey = canonicalSessionAddressingKey(maybeSession, params.session);
 
-    const maxAgeSeconds = parsePositiveIntHeader(
-      request.headers.get("x-channel-max-age-seconds")
-    );
+    const maxAgeSeconds = parsePositiveIntHeader(request.headers.get("x-channel-max-age-seconds"));
     const deleteOnEmptyMinAgeSeconds = parsePositiveIntHeader(
       request.headers.get("x-channel-delete-on-empty-seconds")
     );

packages/core/src/v3/sessionStreams/channels.ts

@@ -17,20 +17,13 @@ export type SessionChannel<
 export type AnySessionChannel = SessionChannel<string, SessionChannelShape>;
 
 /** Extract a channel's literal name, the analogue of `TaskIdentifier`. */
-export type SessionChannelName<C extends AnySessionChannel> = C extends SessionChannel<infer N, any>
-  ? N
-  : never;
+export type SessionChannelName<C extends AnySessionChannel> =
+  C extends SessionChannel<infer N, any> ? N : never;
 
 /** Extract the `.out` record type, the analogue of `TaskOutput`. */
-export type SessionChannelOut<C extends AnySessionChannel> = C extends SessionChannel<any, infer S>
-  ? S extends { out: infer O }
-    ? O
-    : unknown
-  : never;
+export type SessionChannelOut<C extends AnySessionChannel> =
+  C extends SessionChannel<any, infer S> ? (S extends { out: infer O } ? O : unknown) : never;
 
 /** Extract the `.in` record type, the analogue of `TaskPayload`. */
-export type SessionChannelIn<C extends AnySessionChannel> = C extends SessionChannel<any, infer S>
-  ? S extends { in: infer I }
-    ? I
-    : unknown
-  : never;
+export type SessionChannelIn<C extends AnySessionChannel> =
+  C extends SessionChannel<any, infer S> ? (S extends { in: infer I } ? I : unknown) : never;

packages/react-hooks/src/hooks/useSessionStreamChannel.ts

@@ -146,10 +146,9 @@ export function useSessionStreamChannel<
     }
   }, [idKey, sessionId, channelName, io, lastEventId]);
 
-  const { data: lastControl = undefined, mutate: setLastControl } = useSWR<undefined | ControlEvent>(
-    [idKey, sessionId, channelName, io, "lastControl"],
-    null
-  );
+  const { data: lastControl = undefined, mutate: setLastControl } = useSWR<
+    undefined | ControlEvent
+  >([idKey, sessionId, channelName, io, "lastControl"], null);
 
   const { data: _isComplete = false, mutate: setIsComplete } = useSWR<boolean>(
     [idKey, sessionId, channelName, io, "complete"],
@@ -207,12 +206,13 @@ export function useSessionStreamChannel<
   }, [maxRecords, mutateRecords]);
 
   const triggerRequest = useCallback(async () => {
+    let abortController: AbortController | null = null;
     try {
       if (!sessionId || !apiClient) {
         return;
       }
 
-      const abortController = new AbortController();
+      abortController = new AbortController();
       abortControllerRef.current = abortController;
 
       await processSessionChannelStream<TRecord>(
@@ -236,13 +236,12 @@ export function useSessionStreamChannel<
       );
     } catch (err) {
       if ((err as any).name === "AbortError") {
-        abortControllerRef.current = null;
         return;
       }
 
       setError(err as Error);
     } finally {
-      if (abortControllerRef.current) {
+      if (abortControllerRef.current === abortController) {
         abortControllerRef.current = null;
       }
 

packages/trigger-sdk/src/v3/ai.ts

@@ -114,10 +114,7 @@ import {
   sessions,
   type SessionSubscribeOptions,
 } from "./sessions.js";
-import type {
-  AnySessionChannel,
-  SessionChannelName,
-} from "@trigger.dev/core/v3";
+import type { AnySessionChannel, SessionChannelName } from "@trigger.dev/core/v3";
 import { createTask } from "./shared.js";
 import { markChatAgentRunForStreamsWarning } from "./streams.js";
 import { tracer } from "./tracer.js";

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

@@ -96,7 +96,7 @@ describe("SessionOutputChannel initializeSessionStream cache", () => {
     await Promise.all([p1.waitUntilComplete(), p2.waitUntilComplete(), p3.waitUntilComplete()]);
 
     expect(spy).toHaveBeenCalledTimes(1);
-    expect(spy).toHaveBeenCalledWith("session-1", "out", undefined);
+    expect(spy).toHaveBeenCalledWith("session-1", "out", undefined, undefined, undefined);
   });
 
   it("evicts on initialize failure so the next call retries instead of returning a poisoned entry", async () => {
@@ -150,8 +150,8 @@ describe("SessionOutputChannel initializeSessionStream cache", () => {
     ]);
 
     expect(spy).toHaveBeenCalledTimes(2);
-    expect(spy).toHaveBeenCalledWith("session-a", "out", undefined);
-    expect(spy).toHaveBeenCalledWith("session-b", "out", undefined);
+    expect(spy).toHaveBeenCalledWith("session-a", "out", undefined, undefined, undefined);
+    expect(spy).toHaveBeenCalledWith("session-b", "out", undefined, undefined, undefined);
   });
 
   it("evicts the cache when a writer's wait() rejects (simulated stale-token failure)", async () => {

packages/trigger-sdk/src/v3/sessions.ts

@@ -325,13 +325,11 @@ export type SessionPipeStreamOptions = Omit<PipeStreamOptions, "target">;
 /**
  * Retention for a named side channel. `maxAgeSeconds` and
  * `deleteOnEmptyMinAgeSeconds` are applied server-side as native S2 per-stream
- * config on first initialize; `keepLastN` is a producer-side trim floor the
- * writer enforces by appending an S2 trim command as records accumulate.
+ * config on first initialize.
  */
 export type SessionChannelRetention = {
   maxAgeSeconds?: number;
   deleteOnEmptyMinAgeSeconds?: number;
-  keepLastN?: number;
 };
 
 export type SessionChannelOptions = {
@@ -732,7 +730,13 @@ export class SessionInputChannel<TIn = unknown> {
       requestOptions
     );
 
-    await apiClient.appendToSessionStream(this.sessionId, "in", body, $requestOptions, this.channel);
+    await apiClient.appendToSessionStream(
+      this.sessionId,
+      "in",
+      body,
+      $requestOptions,
+      this.channel
+    );
   }
 
   /**