fix(slack): roll expired native streams

Author: tdietert

.changeset/slack-stream-rollover.md

@@ -0,0 +1,5 @@
+---
+"@chat-adapter/slack": patch
+---
+
+Recover Slack native streams by rolling to a fresh streamed message when Slack expires the current stream.

packages/adapter-slack/package.json

@@ -45,7 +45,7 @@
   "dependencies": {
     "@chat-adapter/shared": "workspace:*",
     "@slack/socket-mode": "^2.0.5",
-    "@slack/web-api": "^7.14.0",
+    "@slack/web-api": "^7.17.0",
     "chat": "workspace:*"
   },
   "devDependencies": {

packages/adapter-slack/src/index.test.ts

@@ -17,6 +17,16 @@ import { createSlackAdapter, SlackAdapter } from "./index";
 
 const FILE_ID_PATTERN = /^file-/;
 
+function createStreamExpiredError(): Error & {
+  code: string;
+  data: { error: string; ok: false };
+} {
+  return Object.assign(new Error("message_not_in_streaming_state"), {
+    code: "slack_webapi_platform_error",
+    data: { error: "message_not_in_streaming_state", ok: false as const },
+  });
+}
+
 // Mock @slack/socket-mode
 const mockSocketStart = vi.fn().mockResolvedValue({});
 const mockSocketDisconnect = vi.fn().mockResolvedValue(undefined);
@@ -7977,6 +7987,7 @@ describe("stream with empty threadTs", () => {
     await adapter.stream("slack:D123:1234567890.000000", stream());
 
     expect(chatStream).toHaveBeenCalledWith({
+      buffer_size: 1,
       channel: "D123",
       thread_ts: "1234567890.000000",
     });
@@ -8061,6 +8072,196 @@ describe("stream with empty threadTs", () => {
       expect.objectContaining({ token: "xoxb-test-token" })
     );
   });
+
+  it("rolls to a fresh stream and retries only the pending markdown delta", async () => {
+    const adapter = createSlackAdapter({
+      botToken: "xoxb-test-token",
+      signingSecret: "test-signing-secret",
+      logger: mockLogger,
+    });
+    const firstAppend = vi
+      .fn()
+      .mockResolvedValueOnce({ ok: true })
+      .mockRejectedValueOnce(createStreamExpiredError());
+    const firstStop = vi.fn();
+    const secondAppend = vi.fn().mockResolvedValue({ ok: true });
+    const secondStop = vi.fn().mockResolvedValue({
+      ok: true,
+      ts: "1234567890.222222",
+    });
+    const chatStream = vi
+      .fn()
+      .mockReturnValueOnce({
+        append: firstAppend,
+        stop: firstStop,
+        ts: "1234567890.111111",
+      })
+      .mockReturnValueOnce({
+        append: secondAppend,
+        stop: secondStop,
+        ts: "1234567890.222222",
+      });
+    mockClientMethod(adapter, "chatStream", chatStream);
+
+    async function* textStream() {
+      yield "first\n";
+      yield "second\n";
+    }
+
+    const result = await adapter.stream(
+      "slack:C123:1234567890.000000",
+      textStream(),
+      {
+        recipientUserId: "U123",
+        recipientTeamId: "T123",
+      }
+    );
+
+    expect(chatStream).toHaveBeenCalledTimes(2);
+    expect(firstAppend).toHaveBeenNthCalledWith(
+      1,
+      expect.objectContaining({ markdown_text: "first\n" })
+    );
+    expect(firstAppend).toHaveBeenNthCalledWith(
+      2,
+      expect.objectContaining({ markdown_text: "second\n" })
+    );
+    expect(secondAppend).toHaveBeenCalledTimes(1);
+    expect(secondAppend).toHaveBeenCalledWith(
+      expect.objectContaining({ markdown_text: "second\n" })
+    );
+    expect(firstStop).not.toHaveBeenCalled();
+    expect(secondStop).toHaveBeenCalledTimes(1);
+    expect(result?.id).toBe("1234567890.222222");
+  });
+
+  it("re-seeds the latest structured chunks when rolling streams", async () => {
+    const adapter = createSlackAdapter({
+      botToken: "xoxb-test-token",
+      signingSecret: "test-signing-secret",
+      logger: mockLogger,
+    });
+    const taskChunk = {
+      details: "working",
+      id: "task-1",
+      status: "in_progress" as const,
+      title: "Task one",
+      type: "task_update" as const,
+    };
+    const firstAppend = vi
+      .fn()
+      .mockResolvedValueOnce({ ok: true })
+      .mockResolvedValueOnce({ ok: true })
+      .mockRejectedValueOnce(createStreamExpiredError());
+    const secondAppend = vi.fn().mockResolvedValue({ ok: true });
+    const secondStop = vi.fn().mockResolvedValue({
+      ok: true,
+      ts: "1234567890.222222",
+    });
+    mockClientMethod(
+      adapter,
+      "chatStream",
+      vi
+        .fn()
+        .mockReturnValueOnce({
+          append: firstAppend,
+          stop: vi.fn(),
+          ts: "1234567890.111111",
+        })
+        .mockReturnValueOnce({
+          append: secondAppend,
+          stop: secondStop,
+          ts: "1234567890.222222",
+        })
+    );
+
+    async function* chunkStream() {
+      yield taskChunk;
+      yield "hello\n";
+      yield "world\n";
+    }
+
+    await adapter.stream("slack:C123:1234567890.000000", chunkStream(), {
+      recipientUserId: "U123",
+      recipientTeamId: "T123",
+    });
+
+    expect(secondAppend).toHaveBeenNthCalledWith(
+      1,
+      expect.objectContaining({ chunks: [taskChunk] })
+    );
+    expect(secondAppend).toHaveBeenNthCalledWith(
+      2,
+      expect.objectContaining({ markdown_text: "world\n" })
+    );
+  });
+
+  it("does not reject when Slack expires the stream before stop", async () => {
+    const adapter = createSlackAdapter({
+      botToken: "xoxb-test-token",
+      signingSecret: "test-signing-secret",
+      logger: mockLogger,
+    });
+    const append = vi.fn().mockResolvedValue({ ok: true });
+    const stop = vi.fn().mockRejectedValue(createStreamExpiredError());
+    const update = vi.fn();
+    mockClientMethod(
+      adapter,
+      "chatStream",
+      vi.fn().mockReturnValue({
+        append,
+        stop,
+        ts: "1234567890.111111",
+      })
+    );
+    mockClientMethod(adapter, "chat.update", update);
+
+    async function* textStream() {
+      yield "hello\n";
+    }
+
+    const result = await adapter.stream(
+      "slack:C123:1234567890.000000",
+      textStream(),
+      {
+        recipientUserId: "U123",
+        recipientTeamId: "T123",
+      }
+    );
+
+    expect(stop).toHaveBeenCalledTimes(1);
+    expect(update).not.toHaveBeenCalled();
+    expect(result?.id).toBe("1234567890.111111");
+  });
+
+  it("still rejects non-expiry stream append errors", async () => {
+    const adapter = createSlackAdapter({
+      botToken: "xoxb-test-token",
+      signingSecret: "test-signing-secret",
+      logger: mockLogger,
+    });
+    const append = vi.fn().mockRejectedValue(new Error("boom"));
+    mockClientMethod(
+      adapter,
+      "chatStream",
+      vi.fn().mockReturnValue({
+        append,
+        stop: vi.fn(),
+        ts: "1234567890.111111",
+      })
+    );
+
+    async function* textStream() {
+      yield "hello\n";
+    }
+
+    await expect(
+      adapter.stream("slack:C123:1234567890.000000", textStream(), {
+        recipientUserId: "U123",
+        recipientTeamId: "T123",
+      })
+    ).rejects.toThrow("boom");
+  });
 });
 
 describe("scheduleMessage with empty threadTs", () => {