feat(runtime, react): Explicit error handling and retires for IntelligenceAgent* (CopilotKit#3379)

<!--
Thank you for sending the PR! We appreciate you spending the time to
work on these changes.

Help us understand your motivation by explaining why you decided to make
this change.


**Please PLEASE reach out to us first before starting any significant
work on new or existing features.**

By the time you've gotten here, you're looking at creating a pull
request so hopefully we're not too late.

We love community contributions! That said, we want to make sure we're
all on the same page before you start.
Investing a lot of time and effort just to find out it doesn't align
with the upstream project feels awful, and we don't want that to happen.
It also helps to make sure the work you're planning isn't already in
progress.

As described in our contributing guide, please file an issue first:
https://github.com/ag-ui-protocol/ag-ui/issues
Or, reach out to us on Discord: https://discord.com/invite/6dffbvGU3D


You can learn more about contributing to copilotkit here:
https://github.com/copilotkit/copilotkit/blob/master/CONTRIBUTING.md

Happy contributing!

-->

## What does this PR do?

This PR expands on reconnection and error handling for the
IntelligenceAgent and IntelligenceAgentRunner interactions with the
Phoenix lib/server, like timeouts and disconnects.

It explicitly defines exponential backoff behavior (Phoenix has default
settings, but they are silently disabled if you add error callbacks.
While that was only done in the AgentRunner to begin with, it felt
strange to leave it as a foot gun elsewhere).

There is also added logic to ensure cleanup messages are sent (at best
effort) before channels are closed.

Additionally:
* each agent run through the IntelligenceAgentRunner now gets its own
web socket connection (before, it only got its own _channel_)
* the IntelligenceAgent now stops a running agent by publishing a stop
(custom AG-UI) message to Phoenix, which is handled by the
IntelligenceAgentRunner.

## Related PRs and Issues

- https://linear.app/copilotkit/issue/ENT-54
- https://linear.app/copilotkit/issue/ENT-59

## Checklist

- [ ] I have read the [Contribution
Guide](https://github.com/copilotkit/copilotkit/blob/master/CONTRIBUTING.md)
- [ ] If the PR changes or adds functionality, I have updated the
relevant documentation

Author: BenTaylorDev

packages/v2/core/src/__tests__/intelligence-agent.test.ts

@@ -257,6 +257,113 @@ describe("IntelligenceAgent", () => {
     });
   });
 
+  describe("mid-run disconnect", () => {
+    it("does not error the observable on a single channel crash (Phoenix retries)", () => {
+      const agent = createAgent();
+      let error: Error | null = null;
+      agent.run(defaultInput).subscribe({
+        next: () => {},
+        error: (err) => {
+          error = err;
+        },
+      });
+
+      const channel = getChannel(agent);
+      channel.triggerJoin("ok");
+
+      // A single channel error should be a no-op — Phoenix handles rejoin.
+      channel.triggerError("server crash");
+
+      expect(error).toBeNull();
+    });
+
+    it("does not error the observable on a single socket error (Phoenix retries)", () => {
+      const agent = createAgent();
+      let error: Error | null = null;
+      agent.run(defaultInput).subscribe({
+        next: () => {},
+        error: (err) => {
+          error = err;
+        },
+      });
+
+      const channel = getChannel(agent);
+      channel.triggerJoin("ok");
+
+      const socket = getSocket(agent);
+      socket.triggerError(new Error("network failure"));
+
+      // A single socket error should not kill the connection — Phoenix retries.
+      expect(error).toBeNull();
+    });
+
+    it("errors the observable after MAX_CONSECUTIVE_ERRORS socket errors", async () => {
+      const agent = createAgent();
+      const promise = collectEvents(agent);
+
+      const socket = getSocket(agent);
+      const channel = getChannel(agent);
+      channel.triggerJoin("ok");
+
+      // Fire 5 consecutive errors (the threshold)
+      for (let i = 0; i < 5; i++) {
+        socket.triggerError(new Error("network failure"));
+      }
+
+      const result = await promise;
+      expect(result.completed).toBe(false);
+      expect(result.error).toBeInstanceOf(Error);
+      expect(result.error!.message).toContain("5 consecutive errors");
+    });
+
+    it("cleans up socket and channel after reaching the error threshold", async () => {
+      const agent = createAgent();
+      const promise = collectEvents(agent);
+
+      const socket = getSocket(agent);
+      const channel = getChannel(agent);
+      channel.triggerJoin("ok");
+
+      for (let i = 0; i < 5; i++) {
+        socket.triggerError();
+      }
+
+      await promise;
+      expect(channel.left).toBe(true);
+      expect(socket.disconnected).toBe(true);
+    });
+
+    it("resets the error counter on successful reconnection", () => {
+      const agent = createAgent();
+      let error: Error | null = null;
+      agent.run(defaultInput).subscribe({
+        next: () => {},
+        error: (err) => {
+          error = err;
+        },
+      });
+
+      const socket = getSocket(agent);
+      const channel = getChannel(agent);
+      channel.triggerJoin("ok");
+
+      // 4 errors (just below threshold)
+      for (let i = 0; i < 4; i++) {
+        socket.triggerError();
+      }
+      expect(error).toBeNull();
+
+      // Successful reconnect resets counter
+      socket.triggerOpen();
+
+      // 4 more errors — still below threshold because counter was reset
+      for (let i = 0; i < 4; i++) {
+        socket.triggerError();
+      }
+      expect(error).toBeNull();
+    });
+  });
+
   describe("join failures", () => {
     it("errors the observable on join error", async () => {
       const agent = createAgent();
@@ -286,66 +393,102 @@ describe("IntelligenceAgent", () => {
   });
 
   describe("abortRun", () => {
-    it("fires a REST POST to /agent/{agentId}/stop/{threadId} and cleans up", async () => {
+    it("pushes a CUSTOM stop event to the channel", () => {
       const agent = createAgent();
       agent.run(defaultInput).subscribe({ next: () => {}, error: () => {} });
 
       const channel = getChannel(agent);
       channel.triggerJoin("ok");
 
-      // Reset fetch mock to only track abort calls
-      mockFetch.mockClear();
-      mockFetch.mockResolvedValue({ ok: true });
-
       agent.abortRun();
 
-      expect(mockFetch).toHaveBeenCalledTimes(1);
-      const [url, options] = mockFetch.mock.calls[0];
-      expect(url).toContain("/agent/my-agent/stop/thread-1");
-      expect(options.method).toBe("POST");
-      expect(options.headers).toMatchObject({
-        "Content-Type": "application/json",
-        Authorization: "Bearer abc",
+      const stopPush = channel.pushLog.find(
+        (c) =>
+          c.payload?.type === EventType.CUSTOM && c.payload?.name === "stop",
+      );
+      expect(stopPush).toBeDefined();
+      expect(stopPush!.payload).toMatchObject({
+        type: EventType.CUSTOM,
+        name: "stop",
+        value: { threadId: "thread-1" },
       });
+    });
+
+    it("defers cleanup until the push is acknowledged (ok)", () => {
+      const agent = createAgent();
+      agent.run(defaultInput).subscribe({ next: () => {}, error: () => {} });
+
+      const channel = getChannel(agent);
+      channel.triggerJoin("ok");
+
+      agent.abortRun();
+
+      // Cleanup has NOT happened yet — waiting for push ACK
+      expect(channel.left).toBe(false);
+
+      // Server acknowledges the stop push
+      const stopEntry = channel.pushLog.find(
+        (c) => c.payload?.name === "stop",
+      )!;
+      stopEntry.push.trigger("ok");
 
-      // Channel should be cleaned up
       expect(channel.left).toBe(true);
     });
 
-    it("is a no-op when fetch is unavailable (SSR)", () => {
+    it("cleans up on push error reply", () => {
       const agent = createAgent();
       agent.run(defaultInput).subscribe({ next: () => {}, error: () => {} });
 
       const channel = getChannel(agent);
       channel.triggerJoin("ok");
+      agent.abortRun();
+
+      const stopEntry = channel.pushLog.find(
+        (c) => c.payload?.name === "stop",
+      )!;
+      stopEntry.push.trigger("error");
 
-      vi.stubGlobal("fetch", undefined);
-      expect(() => agent.abortRun()).not.toThrow();
       expect(channel.left).toBe(true);
     });
 
-    it("does not push any CUSTOM events to the channel", () => {
+    it("cleans up on push timeout reply", () => {
       const agent = createAgent();
       agent.run(defaultInput).subscribe({ next: () => {}, error: () => {} });
 
       const channel = getChannel(agent);
       channel.triggerJoin("ok");
+      agent.abortRun();
+
+      const stopEntry = channel.pushLog.find(
+        (c) => c.payload?.name === "stop",
+      )!;
+      stopEntry.push.trigger("timeout");
 
-      // Clear the push log (run may have pushed something)
-      channel.pushLog.length = 0;
+      expect(channel.left).toBe(true);
+    });
+
+    it("cleans up immediately via fallback timer when socket is down", () => {
+      vi.useFakeTimers();
+      const agent = createAgent();
+      agent.run(defaultInput).subscribe({ next: () => {}, error: () => {} });
 
+      const channel = getChannel(agent);
+      channel.triggerJoin("ok");
       agent.abortRun();
 
-      // No CUSTOM stop pushes
-      const stopPush = channel.pushLog.find(
-        (c) => c.event === EventType.CUSTOM && c.payload?.name === "stop",
-      );
-      expect(stopPush).toBeUndefined();
+      // No ACK will arrive (socket is down) — channel still open
+      expect(channel.left).toBe(false);
+
+      // Fallback fires after 5 seconds
+      vi.advanceTimersByTime(5_000);
+      expect(channel.left).toBe(true);
+
+      vi.useRealTimers();
     });
 
-    it("is a no-op when no run is active", () => {
+    it("cleans up immediately when no run is active", () => {
       const agent = createAgent();
-      // Should not throw.
+      // Should not throw and should not push anything
       expect(() => agent.abortRun()).not.toThrow();
       expect(mockFetch).not.toHaveBeenCalled();
     });
@@ -387,27 +530,6 @@ describe("IntelligenceAgent", () => {
       expect(options.credentials).toBe("include");
     });
 
-    it("forwards credentials on abortRun fetch when configured", async () => {
-      const agent = new IntelligenceAgent({
-        url: "ws://localhost:4000/socket",
-        runtimeUrl: "http://localhost:4000",
-        agentId: "my-agent",
-        credentials: "include",
-      });
-      agent.run(defaultInput).subscribe({ next: () => {}, error: () => {} });
-
-      const channel = getChannel(agent);
-      channel.triggerJoin("ok");
-
-      mockFetch.mockClear();
-      mockFetch.mockResolvedValue({ ok: true });
-
-      agent.abortRun();
-
-      const [, options] = mockFetch.mock.calls[0];
-      expect(options.credentials).toBe("include");
-    });
-
     it("omits credentials when not configured", async () => {
       const agent = createAgent();
       agent.run(defaultInput).subscribe({ next: () => {}, error: () => {} });
@@ -534,6 +656,42 @@ describe("IntelligenceAgent", () => {
       expect(result.error).toBeInstanceOf(Error);
       expect(result.error!.message).toContain("Failed to join channel");
     });
+
+    it("does not error the observable on a single channel crash (Phoenix retries)", () => {
+      const agent = createAgent();
+      let error: Error | null = null;
+      (agent as any).connect(defaultInput).subscribe({
+        next: () => {},
+        error: (err: Error) => {
+          error = err;
+        },
+      });
+
+      const channel = getChannel(agent);
+      channel.triggerJoin("ok");
+      channel.triggerError("server crash");
+
+      // Channel errors are handled by Phoenix auto-rejoin — no observer error.
+      expect(error).toBeNull();
+    });
+
+    it("errors the observable after MAX_CONSECUTIVE_ERRORS socket errors", async () => {
+      const agent = createAgent();
+      const promise = connectAgent(agent);
+
+      const socket = getSocket(agent);
+      const channel = getChannel(agent);
+      channel.triggerJoin("ok");
+
+      for (let i = 0; i < 5; i++) {
+        socket.triggerError(new Error("network failure"));
+      }
+
+      const result = await promise;
+      expect(result.completed).toBe(false);
+      expect(result.error).toBeInstanceOf(Error);
+      expect(result.error!.message).toContain("5 consecutive errors");
+    });
   });
 
   describe("clone", () => {

packages/v2/core/src/__tests__/test-utils.ts

@@ -285,11 +285,12 @@ export class MockPush {
 export class MockChannel {
   public topic: string;
   public params: Record<string, any>;
-  public pushLog: Array<{ event: string; payload: any }> = [];
+  public pushLog: Array<{ event: string; payload: any; push: MockPush }> = [];
   public left = false;
 
   private handlers = new Map<string, Array<(payload: any) => void>>();
   private joinPush = new MockPush();
+  private errorHandlers: Array<(reason?: any) => void> = [];
 
   constructor(topic: string = "", params: Record<string, any> = {}) {
     this.topic = topic;
@@ -304,13 +305,18 @@ export class MockChannel {
     return this.handlers.get(event)!.length;
   }
 
+  onError(callback: (reason?: any) => void): void {
+    this.errorHandlers.push(callback);
+  }
+
   join(): MockPush {
     return this.joinPush;
   }
 
   push(event: string, payload: any): MockPush {
-    this.pushLog.push({ event, payload });
-    return new MockPush();
+    const mockPush = new MockPush();
+    this.pushLog.push({ event, payload, push: mockPush });
+    return mockPush;
   }
 
   leave(): void {
@@ -328,6 +334,11 @@ export class MockChannel {
       handler(payload);
     }
   }
+
+  /** Test helper — simulate the channel crashing server-side. */
+  triggerError(reason?: string): void {
+    for (const handler of this.errorHandlers) handler(reason);
+  }
 }
 
 export class MockSocket {
@@ -337,6 +348,9 @@ export class MockSocket {
   public disconnected = false;
   public channels: MockChannel[] = [];
 
+  private errorHandlers: Array<(error?: any) => void> = [];
+  private openHandlers: Array<() => void> = [];
+
   constructor(url: string = "", opts: Record<string, any> = {}) {
     this.url = url;
     this.opts = opts;
@@ -350,9 +364,27 @@ export class MockSocket {
     this.disconnected = true;
   }
 
+  onError(callback: (error?: any) => void): void {
+    this.errorHandlers.push(callback);
+  }
+
+  onOpen(callback: () => void): void {
+    this.openHandlers.push(callback);
+  }
+
   channel(topic: string, params: Record<string, any> = {}): MockChannel {
     const ch = new MockChannel(topic, params);
     this.channels.push(ch);
     return ch;
   }
+
+  /** Test helper — simulate the WebSocket transport erroring. */
+  triggerError(error?: any): void {
+    for (const handler of this.errorHandlers) handler(error);
+  }
+
+  /** Test helper — simulate a successful (re)connection. */
+  triggerOpen(): void {
+    for (const handler of this.openHandlers) handler();
+  }
 }