Add stream interruption support (truncateAfterChunks, disconnectAfterMs, AbortSignal)

CHANGELOG.md

@@ -1,5 +1,17 @@
 # @copilotkit/llmock
 
+## 1.3.0
+
+### Minor Changes
+
+- Mid-stream interruption: `truncateAfterChunks` and `disconnectAfterMs` fixture fields to simulate abrupt server disconnects
+- AbortSignal-based cancellation primitives (`createInterruptionSignal`, signal-aware `delay()`)
+- Backward-compatible `writeSSEStream` overload with `StreamOptions` returning completion status
+- Interruption support across all HTTP SSE and WebSocket streaming paths
+- `destroy()` method on `WebSocketConnection` for abrupt disconnect simulation
+- Journal records `interrupted` and `interruptReason` on interrupted streams
+- LLMock convenience API extended with interruption options (`truncateAfterChunks`, `disconnectAfterMs`)
+
 ## 1.2.0
 
 ### Minor Changes

README.md

@@ -622,11 +622,6 @@ Areas where llmock could grow, and explicit non-goals for the current scope.
 - **WebSocket compression**: `permessage-deflate` is not supported.
 - **Session persistence**: Realtime and Gemini Live sessions exist only for the lifetime of a single WebSocket connection. There is no cross-connection session resumption.
 
-### Streaming
-
-- **Mid-stream interruption**: No way to simulate a server disconnecting partway through a stream (e.g. `truncateAfterChunks`, `disconnectAfterMs`).
-- **Abort/cancellation signaling**: Streaming functions do not accept an `AbortSignal` for client-side cancellation.
-
 ### Fixtures
 
 - **Request metadata in predicates**: Predicate functions receive only the `ChatCompletionRequest`, not HTTP headers, method, or URL.

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@copilotkit/llmock",
-  "version": "1.2.0",
+  "version": "1.3.0",
   "description": "Deterministic mock LLM server for testing (OpenAI, Anthropic, Gemini)",
   "license": "MIT",
   "packageManager": "pnpm@10.28.2",

src/__tests__/fixture-loader.test.ts

@@ -114,6 +114,71 @@ describe("loadFixtureFile", () => {
     expect(fixtures[0].chunkSize).toBeUndefined();
   });
 
+  it("passes through truncateAfterChunks when set", () => {
+    const filePath = writeJson(tmpDir, "truncate.json", {
+      fixtures: [
+        {
+          match: { userMessage: "truncate me" },
+          response: { content: "partial" },
+          truncateAfterChunks: 3,
+        },
+      ],
+    });
+
+    const fixtures = loadFixtureFile(filePath);
+    expect(fixtures).toHaveLength(1);
+    expect(fixtures[0].truncateAfterChunks).toBe(3);
+  });
+
+  it("passes through disconnectAfterMs when set", () => {
+    const filePath = writeJson(tmpDir, "disconnect.json", {
+      fixtures: [
+        {
+          match: { userMessage: "disconnect me" },
+          response: { content: "partial" },
+          disconnectAfterMs: 500,
+        },
+      ],
+    });
+
+    const fixtures = loadFixtureFile(filePath);
+    expect(fixtures).toHaveLength(1);
+    expect(fixtures[0].disconnectAfterMs).toBe(500);
+  });
+
+  it("passes through both truncateAfterChunks and disconnectAfterMs together", () => {
+    const filePath = writeJson(tmpDir, "both-interruptions.json", {
+      fixtures: [
+        {
+          match: { userMessage: "both" },
+          response: { content: "partial" },
+          truncateAfterChunks: 5,
+          disconnectAfterMs: 1000,
+        },
+      ],
+    });
+
+    const fixtures = loadFixtureFile(filePath);
+    expect(fixtures).toHaveLength(1);
+    expect(fixtures[0].truncateAfterChunks).toBe(5);
+    expect(fixtures[0].disconnectAfterMs).toBe(1000);
+  });
+
+  it("omits truncateAfterChunks and disconnectAfterMs when not present in JSON", () => {
+    const filePath = writeJson(tmpDir, "no-interruptions.json", {
+      fixtures: [
+        {
+          match: { userMessage: "plain" },
+          response: { content: "complete" },
+        },
+      ],
+    });
+
+    const fixtures = loadFixtureFile(filePath);
+    expect(fixtures[0].truncateAfterChunks).toBeUndefined();
+    expect(fixtures[0].disconnectAfterMs).toBeUndefined();
+  });
+
   it("warns and returns empty array for invalid JSON", () => {
     const filePath = join(tmpDir, "bad.json");
     writeFileSync(filePath, "{ not valid json", "utf-8");

src/__tests__/interruption.test.ts

@@ -0,0 +1,142 @@
+import { describe, it, expect, vi, afterEach } from "vitest";
+import { createInterruptionSignal } from "../interruption.js";
+import type { Fixture } from "../types.js";
+
+function makeFixture(overrides?: Partial<Fixture>): Fixture {
+  return {
+    match: { userMessage: "test" },
+    response: { content: "hello" },
+    ...overrides,
+  };
+}
+
+afterEach(() => {
+  vi.useRealTimers();
+});
+
+describe("createInterruptionSignal", () => {
+  it("returns null when no interruption fields are set", () => {
+    const result = createInterruptionSignal(makeFixture());
+    expect(result).toBeNull();
+  });
+
+  it("returns null when both fields are undefined", () => {
+    const result = createInterruptionSignal(
+      makeFixture({ truncateAfterChunks: undefined, disconnectAfterMs: undefined }),
+    );
+    expect(result).toBeNull();
+  });
+
+  it("truncateAfterChunks: aborts after N ticks", () => {
+    const ctrl = createInterruptionSignal(makeFixture({ truncateAfterChunks: 3 }));
+    expect(ctrl).not.toBeNull();
+    expect(ctrl!.signal.aborted).toBe(false);
+
+    ctrl!.tick();
+    expect(ctrl!.signal.aborted).toBe(false);
+    ctrl!.tick();
+    expect(ctrl!.signal.aborted).toBe(false);
+    ctrl!.tick();
+    expect(ctrl!.signal.aborted).toBe(true);
+    expect(ctrl!.reason()).toBe("truncateAfterChunks");
+
+    ctrl!.cleanup();
+  });
+
+  it("truncateAfterChunks: extra ticks after abort are no-ops", () => {
+    const ctrl = createInterruptionSignal(makeFixture({ truncateAfterChunks: 1 }));
+    ctrl!.tick();
+    expect(ctrl!.signal.aborted).toBe(true);
+    // Should not throw
+    ctrl!.tick();
+    ctrl!.tick();
+    expect(ctrl!.reason()).toBe("truncateAfterChunks");
+    ctrl!.cleanup();
+  });
+
+  it("disconnectAfterMs: aborts after timeout", async () => {
+    vi.useFakeTimers();
+    const ctrl = createInterruptionSignal(makeFixture({ disconnectAfterMs: 100 }));
+    expect(ctrl).not.toBeNull();
+    expect(ctrl!.signal.aborted).toBe(false);
+
+    vi.advanceTimersByTime(99);
+    expect(ctrl!.signal.aborted).toBe(false);
+
+    vi.advanceTimersByTime(1);
+    expect(ctrl!.signal.aborted).toBe(true);
+    expect(ctrl!.reason()).toBe("disconnectAfterMs");
+
+    ctrl!.cleanup();
+  });
+
+  it("both set: truncateAfterChunks fires first wins", () => {
+    vi.useFakeTimers();
+    const ctrl = createInterruptionSignal(
+      makeFixture({ truncateAfterChunks: 2, disconnectAfterMs: 10000 }),
+    );
+
+    ctrl!.tick();
+    ctrl!.tick();
+    expect(ctrl!.signal.aborted).toBe(true);
+    expect(ctrl!.reason()).toBe("truncateAfterChunks");
+
+    ctrl!.cleanup();
+  });
+
+  it("both set: disconnectAfterMs fires first wins", () => {
+    vi.useFakeTimers();
+    const ctrl = createInterruptionSignal(
+      makeFixture({ truncateAfterChunks: 100, disconnectAfterMs: 50 }),
+    );
+
+    ctrl!.tick(); // 1 of 100
+    expect(ctrl!.signal.aborted).toBe(false);
+
+    vi.advanceTimersByTime(50);
+    expect(ctrl!.signal.aborted).toBe(true);
+    expect(ctrl!.reason()).toBe("disconnectAfterMs");
+
+    ctrl!.cleanup();
+  });
+
+  it("cleanup clears the timer", () => {
+    vi.useFakeTimers();
+    const ctrl = createInterruptionSignal(makeFixture({ disconnectAfterMs: 100 }));
+
+    ctrl!.cleanup();
+
+    vi.advanceTimersByTime(200);
+    expect(ctrl!.signal.aborted).toBe(false);
+    expect(ctrl!.reason()).toBeUndefined();
+  });
+
+  it("reason returns undefined before abort", () => {
+    const ctrl = createInterruptionSignal(makeFixture({ truncateAfterChunks: 5 }));
+    expect(ctrl!.reason()).toBeUndefined();
+    ctrl!.cleanup();
+  });
+
+  it("truncateAfterChunks: 0 aborts immediately on first tick", () => {
+    const ctrl = createInterruptionSignal(makeFixture({ truncateAfterChunks: 0 }));
+    expect(ctrl).not.toBeNull();
+    expect(ctrl!.signal.aborted).toBe(false);
+
+    ctrl!.tick();
+    expect(ctrl!.signal.aborted).toBe(true);
+    expect(ctrl!.reason()).toBe("truncateAfterChunks");
+
+    ctrl!.cleanup();
+  });
+
+  it("disconnectAfterMs: 0 aborts promptly", async () => {
+    const ctrl = createInterruptionSignal(makeFixture({ disconnectAfterMs: 0 }));
+    expect(ctrl).not.toBeNull();
+
+    await new Promise((r) => setTimeout(r, 10));
+    expect(ctrl!.signal.aborted).toBe(true);
+    expect(ctrl!.reason()).toBe("disconnectAfterMs");
+
+    ctrl!.cleanup();
+  });
+});