fix: expose chat input kind to fetch callbacks

Author: gtremper

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

@@ -2,6 +2,7 @@ import { createServer, type Server, type ServerResponse } from "node:http";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import {
   TriggerChatTransport,
+  type ChatTransportEndpointContext,
   type ChatTransportEvent,
   type TriggerChatTransportOptions,
 } from "./chat.js";
@@ -31,6 +32,7 @@ describe("Stop delivery outcomes", () => {
   let inputs: InputRequest[];
   let outputs: ServerResponse[];
   let events: ChatTransportEvent[];
+  let fetchContexts: ChatTransportEndpointContext[];
   let respond: (input: InputRequest) => void;
   let inputSeq: number;
 
@@ -41,6 +43,10 @@ describe("Stop delivery outcomes", () => {
       accessToken: () => "test-token",
       sessions: { chat: { publicAccessToken: "test-token", lastEventId: "1" } },
       onEvent: (event) => events.push(event),
+      fetch: (url, init, context) => {
+        fetchContexts.push({ ...context });
+        return globalThis.fetch(url, init);
+      },
       ...options,
     });
   }
@@ -55,6 +61,7 @@ describe("Stop delivery outcomes", () => {
     inputs = [];
     outputs = [];
     events = [];
+    fetchContexts = [];
     inputSeq = 10;
     respond = (input) => appendResponse(input);
     server = createServer(async (request, response) => {
@@ -128,6 +135,7 @@ describe("Stop delivery outcomes", () => {
       await expect(stop(throwOnError)).resolves.toBe(true);
       expect(inputs).toHaveLength(1);
       expect(inputs[0]?.kind).toBe("stop");
+      expect(fetchContexts).toEqual([{ endpoint: "in", chatId: "chat", inputKind: "stop" }]);
       expect(sendFailures()).toHaveLength(0);
     }
   );
@@ -199,6 +207,10 @@ describe("Stop delivery outcomes", () => {
         "Bearer test-token",
         "Bearer fresh-token",
       ]);
+      expect(fetchContexts).toEqual([
+        { endpoint: "in", chatId: "chat", inputKind: "stop" },
+        { endpoint: "in", chatId: "chat", inputKind: "stop" },
+      ]);
       expect(sendFailures()).toHaveLength(0);
     }
   );
@@ -315,6 +327,29 @@ describe("Stop delivery outcomes", () => {
     }
   );
 
+  it.each(["message", "action"] as const)(
+    "identifies %s input and consumer Stop without an output input kind",
+    async (kind) => {
+      const abort = new AbortController();
+      const reader = (await send(kind, abort.signal)).getReader();
+      await vi.waitFor(() => expect(outputs).toHaveLength(1));
+      expect(fetchContexts).toEqual([
+        { endpoint: "in", chatId: "chat", inputKind: "message" },
+        { endpoint: "out", chatId: "chat" },
+      ]);
+      abort.abort();
+      await expect(reader.read()).resolves.toEqual({ done: true, value: undefined });
+      await vi.waitFor(() =>
+        expect(inputs.map((input) => input.kind)).toEqual(["message", "stop"])
+      );
+      expect(fetchContexts).toEqual([
+        { endpoint: "in", chatId: "chat", inputKind: "message" },
+        { endpoint: "out", chatId: "chat" },
+        { endpoint: "in", chatId: "chat", inputKind: "stop" },
+      ]);
+    }
+  );
+
   describe.each(["message", "action"] as const)("%s cancellation", (kind) => {
     it("does not hold a tab claim during cancelled session creation", async () => {
       const token = deferred<string>();

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

@@ -190,6 +190,8 @@ export type ChatTransportEndpoint = "in" | "out";
 export type ChatTransportEndpointContext = {
   endpoint: ChatTransportEndpoint;
   chatId: string;
+  /** Input control kind before serialization. Output requests omit this field. */
+  inputKind?: "message" | "stop";
 };
 
 /** Resolver form of `baseURL` — return the base for the given endpoint. */
@@ -1407,7 +1409,8 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
 
     const partId = crypto.randomUUID();
     const serializedBody = this.serializeInputChunk({ kind: "stop" });
-    const send = (token: string) => this.appendInputChunk(chatId, token, serializedBody, partId);
+    const send = (token: string) =>
+      this.appendInputChunk(chatId, token, serializedBody, partId, "stop");
     try {
       const inSeq = await this.sendWithEvents(
         chatId,
@@ -2147,9 +2150,10 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     chatId: string,
     token: string,
     body: string,
-    partId?: string
+    partId?: string,
+    inputKind: "message" | "stop" = "message"
   ): Promise<number | undefined> {
-    const ctx: ChatTransportEndpointContext = { endpoint: "in", chatId };
+    const ctx: ChatTransportEndpointContext = { endpoint: "in", chatId, inputKind };
     const url = `${this.resolveBaseURL(ctx)}/realtime/v1/sessions/${encodeURIComponent(chatId)}/in/append`;
     // extraHeaders first so the fixed headers below win — a transport-wide
     // X-Part-Id must not override the per-append idempotency key.
@@ -2349,7 +2353,9 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
             this.appendInputChunk(
               chatId,
               state.publicAccessToken,
-              this.serializeInputChunk({ kind: "stop" })
+              this.serializeInputChunk({ kind: "stop" }),
+              undefined,
+              "stop"
             )
               .then((inSeq) => this.recordStoppedInput(chatId, state, stoppedBoundary, inSeq))
               .catch(() => {});