fix(mastra): support local abort handling

.codegraph/.gitignore

@@ -0,0 +1,5 @@
+# CodeGraph data files — local to each machine, not for committing.
+# Ignore everything in .codegraph/ except this file itself, so transient
+# files (the database, daemon.pid, sockets, logs) never show up in git.
+*
+!.gitignore

integrations/mastra/typescript/src/__tests__/abort.test.ts

@@ -0,0 +1,52 @@
+import { describe, expect, it } from "vitest";
+import { MastraAgent } from "../mastra";
+import { FakeLocalAgent, makeInput } from "./helpers";
+
+describe("abortRun()", () => {
+  it("aborts the per-run signal passed to a local Mastra agent stream", async () => {
+    const fakeAgent = new FakeLocalAgent();
+    let capturedSignal: AbortSignal | undefined;
+    let resolveStreamStarted!: () => void;
+    let resolveStream!: () => void;
+    const streamStarted = new Promise<void>((resolve) => {
+      resolveStreamStarted = resolve;
+    });
+    const releaseStream = new Promise<void>((resolve) => {
+      resolveStream = resolve;
+    });
+
+    fakeAgent.stream = async (_messages: any, opts?: any) => {
+      capturedSignal = opts?.abortSignal;
+      resolveStreamStarted();
+      return {
+        fullStream: (async function* () {
+          await releaseStream;
+          yield { type: "finish", payload: {} };
+        })(),
+      };
+    };
+
+    const agent = new MastraAgent({
+      agentId: "test-agent",
+      agent: fakeAgent as any,
+      resourceId: "resource-1",
+    });
+
+    const runFinished = new Promise<void>((resolve, reject) => {
+      agent.run(makeInput()).subscribe({
+        error: reject,
+        complete: resolve,
+      });
+    });
+
+    await streamStarted;
+    expect(capturedSignal).toBeInstanceOf(AbortSignal);
+    expect(capturedSignal?.aborted).toBe(false);
+
+    agent.abortRun();
+
+    expect(capturedSignal?.aborted).toBe(true);
+    resolveStream();
+    await runFinished;
+  });
+});

integrations/mastra/typescript/src/__tests__/edge-cases.test.ts

@@ -681,6 +681,43 @@ describe("event emission details (fake-only)", () => {
       warnSpy.mockRestore();
     });
 
+    it("treats abort chunks as a terminal stream boundary without warning", async () => {
+      const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
+
+      const agent = makeLocalMastraAgent({
+        streamChunks: [
+          { type: "text-delta", payload: { text: "before" } },
+          { type: "abort" },
+          { type: "text-delta", payload: { text: "after" } },
+          { type: "finish", payload: { finishReason: "stop" } },
+        ],
+      });
+
+      const events = await collectEvents(agent, makeInput());
+
+      expect(events.some((e) => e.type === EventType.RUN_FINISHED)).toBe(true);
+      const joinedText = events
+        .filter(
+          (e): e is TextMessageChunkEvent =>
+            e.type === EventType.TEXT_MESSAGE_CHUNK,
+        )
+        .map((e) => e.delta ?? "")
+        .join("");
+      expect(joinedText).toContain("before");
+      expect(joinedText).not.toContain("after");
+
+      const warnedTypes = warnSpy.mock.calls.map((c) => String(c[0]));
+      expect(
+        warnedTypes.some(
+          (m) =>
+            m.includes("Unrecognized stream chunk type") &&
+            m.includes("abort"),
+        ),
+      ).toBe(false);
+
+      warnSpy.mockRestore();
+    });
+
     it("warns at most once per payload-less chunk type (no log flood)", async () => {
       const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
 

integrations/mastra/typescript/src/__tests__/helpers.ts

@@ -79,6 +79,10 @@ export class FakeLocalAgent {
     return this.memory;
   }
 
+  async listTools(_opts?: any) {
+    return {};
+  }
+
   async stream(messages: any, opts?: any) {
     this.lastStreamMessages = messages;
     this.lastStreamOpts = opts;

integrations/mastra/typescript/src/__tests__/local-stream-hooks.test.ts

@@ -0,0 +1,99 @@
+import { describe, expect, it, vi } from "vitest";
+import { MastraAgent } from "../mastra";
+import {
+  FakeLocalAgent,
+  FakeRemoteAgent,
+  collectEvents,
+  makeInput,
+} from "./helpers";
+import { getLocalAgent, getLocalAgents } from "../utils";
+
+const finishChunks = [{ type: "finish", payload: {} }];
+
+describe("local Mastra stream hooks", () => {
+  it("forwards Mastra prepareStep and lifecycle hooks", async () => {
+    const prepareStep = vi.fn();
+    const onFinish = vi.fn();
+    const onStepFinish = vi.fn();
+    const fake = new FakeLocalAgent({ streamChunks: finishChunks });
+    const agent = new MastraAgent({
+      agentId: "test-agent",
+      agent: fake as any,
+      prepareStep,
+      onFinish,
+      onStepFinish,
+    });
+
+    await collectEvents(agent, makeInput());
+
+    expect(fake.lastStreamOpts?.prepareStep).toBe(prepareStep);
+    expect(fake.lastStreamOpts?.onFinish).toBe(onFinish);
+    expect(fake.lastStreamOpts?.onStepFinish).toBe(onStepFinish);
+  });
+
+  it("does not forward in-process callbacks to remote agents", async () => {
+    const fake = new FakeRemoteAgent({ streamChunks: finishChunks });
+    const agent = new MastraAgent({
+      agentId: "test-agent",
+      agent: fake as any,
+      prepareStep: vi.fn(),
+      onFinish: vi.fn(),
+      onStepFinish: vi.fn(),
+    });
+
+    await collectEvents(agent, makeInput());
+
+    expect(fake.lastStreamOpts).not.toHaveProperty("prepareStep");
+    expect(fake.lastStreamOpts).not.toHaveProperty("onFinish");
+    expect(fake.lastStreamOpts).not.toHaveProperty("onStepFinish");
+  });
+
+  it("getLocalAgent exposes and forwards the hooks", async () => {
+    const prepareStep = vi.fn();
+    const onFinish = vi.fn();
+    const onStepFinish = vi.fn();
+    const fake = new FakeLocalAgent({ streamChunks: finishChunks });
+    const mastra = { getAgent: vi.fn(() => fake) } as any;
+    const agent = getLocalAgent({
+      mastra,
+      agentId: "test-agent",
+      resourceId: "resource-1",
+      prepareStep,
+      onFinish,
+      onStepFinish,
+    }) as MastraAgent;
+
+    await collectEvents(agent, makeInput());
+
+    expect(fake.lastStreamOpts?.prepareStep).toBe(prepareStep);
+    expect(fake.lastStreamOpts?.onFinish).toBe(onFinish);
+    expect(fake.lastStreamOpts?.onStepFinish).toBe(onStepFinish);
+  });
+
+  it("getLocalAgents exposes and forwards the hooks to each agent", async () => {
+    const prepareStep = vi.fn();
+    const onFinish = vi.fn();
+    const onStepFinish = vi.fn();
+    const first = new FakeLocalAgent({ streamChunks: finishChunks });
+    const second = new FakeLocalAgent({ streamChunks: finishChunks });
+    const mastra = {
+      listAgents: vi.fn(() => ({ first, second })),
+    } as any;
+    const agents = getLocalAgents({
+      mastra,
+      resourceId: "resource-1",
+      prepareStep,
+      onFinish,
+      onStepFinish,
+    });
+
+    await collectEvents(agents.first as MastraAgent, makeInput());
+    await collectEvents(agents.second as MastraAgent, makeInput());
+
+    for (const fake of [first, second]) {
+      expect(fake.lastStreamOpts?.prepareStep).toBe(prepareStep);
+      expect(fake.lastStreamOpts?.onFinish).toBe(onFinish);
+      expect(fake.lastStreamOpts?.onStepFinish).toBe(onStepFinish);
+    }
+  });
+});