fix(ai): stop streams getting cut off and keep tool steps across messages

plugins/ai/tests/backend/engine.test.ts

@@ -200,4 +200,79 @@ describe("runAgent", () => {
     expect(events.at(-1)).toMatchObject({ type: "error" });
     expect(streamChat.mock.calls.length).toBeLessThanOrEqual(8);
   });
+
+  it("emits every message it adds so the caller can store the run", async () => {
+    handler.mockResolvedValue({ hosts: [] });
+    streamChat
+      .mockReturnValueOnce(
+        chunks(
+          { type: "text", text: "Looking" },
+          {
+            type: "tool_call",
+            call: { id: "c1", name: "list_hosts", arguments: {} },
+          },
+          { type: "done" },
+        ),
+      )
+      .mockReturnValueOnce(
+        chunks({ type: "text", text: "none" }, { type: "done" }),
+      );
+
+    const stored = (await collect())
+      .filter((e) => e.type === "message")
+      .map((e) => e.message);
+
+    expect(stored).toEqual([
+      {
+        role: "assistant",
+        content: "Looking",
+        toolCalls: [{ id: "c1", name: "list_hosts", arguments: {} }],
+      },
+      {
+        role: "tool",
+        content: JSON.stringify({ hosts: [] }),
+        toolCallId: "c1",
+        toolName: "list_hosts",
+      },
+      { role: "assistant", content: "none" },
+    ]);
+  });
+
+  it("explains a stream the provider cut off", async () => {
+    streamChat.mockReturnValueOnce(
+      (async function* () {
+        yield { type: "text", text: "par" } as ChatChunk;
+        throw Object.assign(new TypeError("terminated"), {
+          cause: Object.assign(new Error("Body Timeout Error"), {
+            code: "UND_ERR_BODY_TIMEOUT",
+          }),
+        });
+      })(),
+    );
+
+    const events = await collect();
+    expect(events.at(-1).type).toBe("error");
+    expect(events.at(-1).message).toMatch(/stopped sending data/);
+  });
+
+  it("stays quiet when the user stopped it", async () => {
+    const controller = new AbortController();
+    streamChat.mockReturnValueOnce(
+      (async function* () {
+        yield { type: "text", text: "" } as ChatChunk;
+        controller.abort();
+        throw new Error("This operation was aborted");
+      })(),
+    );
+
+    const events: any[] = [];
+    for await (const event of runAgent({
+      ...BASE,
+      history: [],
+      signal: controller.signal,
+    })) {
+      events.push(event);
+    }
+    expect(events.some((e) => e.type === "error")).toBe(false);
+  });
 });

plugins/ai/tests/backend/history.test.ts

@@ -0,0 +1,116 @@
+import { describe, expect, it } from "vitest";
+import { toChatHistory, toStoredMessage } from "../../src/backend/history.js";
+
+const calls = [
+  { id: "c1", name: "list_hosts", arguments: {} },
+  { id: "c2", name: "get_host", arguments: { hostId: 1 } },
+];
+
+describe("toStoredMessage", () => {
+  it("keeps the answered call on a tool row", () => {
+    expect(
+      toStoredMessage({
+        role: "tool",
+        content: "{}",
+        toolCallId: "c1",
+        toolName: "list_hosts",
+      }),
+    ).toEqual({
+      role: "tool",
+      content: "{}",
+      toolCalls: JSON.stringify([{ id: "c1", name: "list_hosts" }]),
+    });
+  });
+
+  it("stores no tool calls for a plain reply", () => {
+    expect(toStoredMessage({ role: "assistant", content: "hi" })).toEqual({
+      role: "assistant",
+      content: "hi",
+      toolCalls: null,
+    });
+  });
+});
+
+describe("toChatHistory", () => {
+  it("round-trips a full run", () => {
+    const rows = [
+      { role: "user", content: "check", toolCalls: null },
+      toStoredMessage({ role: "assistant", content: "", toolCalls: calls }),
+      toStoredMessage({
+        role: "tool",
+        content: '{"hosts":[]}',
+        toolCallId: "c1",
+        toolName: "list_hosts",
+      }),
+      toStoredMessage({
+        role: "tool",
+        content: '{"host":{}}',
+        toolCallId: "c2",
+        toolName: "get_host",
+      }),
+      { role: "assistant", content: "done", toolCalls: null },
+    ];
+
+    expect(toChatHistory(rows)).toEqual([
+      { role: "user", content: "check" },
+      { role: "assistant", content: "", toolCalls: calls },
+      {
+        role: "tool",
+        content: '{"hosts":[]}',
+        toolCallId: "c1",
+        toolName: "list_hosts",
+      },
+      {
+        role: "tool",
+        content: '{"host":{}}',
+        toolCallId: "c2",
+        toolName: "get_host",
+      },
+      { role: "assistant", content: "done" },
+    ]);
+  });
+
+  it("fills in results for calls a cut off run never answered", () => {
+    const rows = [
+      { role: "user", content: "check", toolCalls: null },
+      toStoredMessage({ role: "assistant", content: "", toolCalls: calls }),
+      toStoredMessage({
+        role: "tool",
+        content: "{}",
+        toolCallId: "c1",
+        toolName: "list_hosts",
+      }),
+      { role: "user", content: "continue", toolCalls: null },
+    ];
+
+    const history = toChatHistory(rows);
+    expect(history.map((m) => m.role)).toEqual([
+      "user",
+      "assistant",
+      "tool",
+      "tool",
+      "user",
+    ]);
+    expect(history[3]).toMatchObject({
+      toolCallId: "c2",
+      toolName: "get_host",
+    });
+    expect(JSON.parse(history[3].content).error).toMatch(/interrupted/);
+  });
+
+  it("repairs a legacy row saved with tool calls and no results", () => {
+    const history = toChatHistory([
+      { role: "assistant", content: "", toolCalls: JSON.stringify(calls) },
+    ]);
+    expect(history.filter((m) => m.role === "tool")).toHaveLength(2);
+  });
+
+  it("drops tool rows that answer nothing", () => {
+    expect(
+      toChatHistory([
+        { role: "tool", content: "{}", toolCalls: null },
+        { role: "user", content: "hi", toolCalls: null },
+      ]),
+    ).toEqual([{ role: "user", content: "hi" }]);
+  });
+});

plugins/ai/tests/backend/providers/provider-http.test.ts

@@ -1,5 +1,9 @@
 import { describe, expect, it, vi } from "vitest";
-import { createProviderFetch } from "../../../src/backend/providers/http.js";
+import {
+  createProviderFetch,
+  describeProviderError,
+} from "../../../src/backend/providers/http.js";
+import { AiProviderError } from "../../../src/backend/providers/types.js";
 import { PRIVATE_DESTINATION_MESSAGE } from "../../../src/backend/egress.js";
 
 describe("createProviderFetch", () => {
@@ -62,3 +66,53 @@ describe("createProviderFetch", () => {
     ).rejects.toThrow("socket hang up");
   });
 });
+
+describe("describeProviderError", () => {
+  const undiciError = (message: string, cause?: unknown) =>
+    Object.assign(new TypeError(message), { cause });
+
+  it("explains a body that went quiet instead of a bare terminated", () => {
+    const error = undiciError(
+      "terminated",
+      Object.assign(new Error("Body Timeout Error"), {
+        code: "UND_ERR_BODY_TIMEOUT",
+      }),
+    );
+    expect(describeProviderError(error, "fallback")).toMatch(
+      /stopped sending data/,
+    );
+  });
+
+  it("explains a refused connection", () => {
+    const error = undiciError(
+      "fetch failed",
+      Object.assign(new Error("connect ECONNREFUSED"), {
+        code: "ECONNREFUSED",
+      }),
+    );
+    expect(describeProviderError(error, "fallback")).toMatch(/refused/);
+  });
+
+  it("maps a private address refused after DNS to the allowlist hint", () => {
+    const error = undiciError(
+      "fetch failed",
+      new Error("Private destinations are not allowed"),
+    );
+    expect(describeProviderError(error, "fallback")).toBe(
+      PRIVATE_DESTINATION_MESSAGE,
+    );
+  });
+
+  it("names the cause of an unfamiliar fetch failure", () => {
+    const error = undiciError("fetch failed", new Error("something odd"));
+    expect(describeProviderError(error, "fallback")).toBe(
+      "The provider request failed: something odd",
+    );
+  });
+
+  it("keeps a provider error's own message", () => {
+    expect(
+      describeProviderError(new AiProviderError("bad key", 401), "fallback"),
+    ).toBe("bad key");
+  });
+});