Build durable agent collaboration, replies, and direct messages

packages/app/src/electro-bridge/ipc/agent-host-api.ts

@@ -0,0 +1,46 @@
+import type {
+  AgentHostEvent,
+  AgentHostRendererRequest,
+  IAgentHostAPI,
+} from "@/shared/types/agent-host";
+import type { LocalAIRendererIPC } from "./local-ai-api";
+
+export const AGENT_HOST_CHANNELS = {
+  ENQUEUE: "agent-host:enqueue",
+  READY: "agent-host:ready",
+  LIST_JOBS: "agent-host:list-jobs",
+  CANCEL: "agent-host:cancel",
+  RESPOND: "agent-host:respond",
+  REQUEST: "agent-host:request",
+  EVENT: "agent-host:event",
+} as const;
+
+export function createAgentHostAPI(
+  rendererIPC: LocalAIRendererIPC,
+): IAgentHostAPI {
+  const invoke = rendererIPC.invoke.bind(rendererIPC) as <T>(
+    channel: string,
+    ...args: unknown[]
+  ) => Promise<T>;
+  return {
+    ready: () => invoke(AGENT_HOST_CHANNELS.READY),
+    enqueue: (dispatch) => invoke(AGENT_HOST_CHANNELS.ENQUEUE, dispatch),
+    listJobs: () => invoke(AGENT_HOST_CHANNELS.LIST_JOBS),
+    cancel: (jobId) => invoke(AGENT_HOST_CHANNELS.CANCEL, jobId),
+    respond: (response) => invoke(AGENT_HOST_CHANNELS.RESPOND, response),
+    onRequest: (callback) => {
+      const handler = (_event: unknown, request: AgentHostRendererRequest) =>
+        callback(request);
+      rendererIPC.on(AGENT_HOST_CHANNELS.REQUEST, handler);
+      return () =>
+        rendererIPC.removeListener(AGENT_HOST_CHANNELS.REQUEST, handler);
+    },
+    onEvent: (callback) => {
+      const handler = (_event: unknown, event: AgentHostEvent) =>
+        callback(event);
+      rendererIPC.on(AGENT_HOST_CHANNELS.EVENT, handler);
+      return () =>
+        rendererIPC.removeListener(AGENT_HOST_CHANNELS.EVENT, handler);
+    },
+  };
+}

packages/app/src/electro-bridge/ipc/agent-host-context.test.ts

@@ -0,0 +1,132 @@
+import { EventEmitter } from "node:events";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import type { AgentHost } from "@/electron/agent-host/host";
+import type { AgentHostRendererBridge } from "@/electron/agent-host/renderer-bridge";
+import type { AgentHostDispatch } from "@/shared/types/agent-host";
+
+vi.mock("electron", () => ({
+  ipcMain: { handle: vi.fn(), removeHandler: vi.fn() },
+}));
+
+import { AGENT_HOST_CHANNELS } from "./agent-host-api";
+import { setupAgentHostIPC } from "./agent-host-context";
+
+type Handler = (event: FakeInvokeEvent, ...args: never[]) => unknown;
+
+class FakeWebContents extends EventEmitter {
+  readonly mainFrame = {};
+  private destroyed = false;
+
+  isDestroyed() {
+    return this.destroyed;
+  }
+}
+
+interface FakeInvokeEvent {
+  sender: FakeWebContents;
+  senderFrame: object;
+}
+
+function event(sender: FakeWebContents): FakeInvokeEvent {
+  return { sender, senderFrame: sender.mainFrame };
+}
+
+function mainIPC() {
+  const handlers = new Map<string, Handler>();
+  return {
+    handlers,
+    ipc: {
+      handle: (channel: string, handler: Handler) =>
+        handlers.set(channel, handler),
+      removeHandler: (channel: string) => handlers.delete(channel),
+    },
+  };
+}
+
+const dispatch: AgentHostDispatch = {
+  channelId: "channel",
+  conversationId: "conversation",
+  triggerMessageId: "message",
+  contextMessageIds: ["message"],
+  mode: "direct",
+  offeredAgentMemberIds: ["agent:fizz"],
+  targets: [{ agentId: "fizz", memberId: "agent:fizz" }],
+  chain: { hops: 0, invoked: ["agent:fizz"] },
+};
+
+describe("Agent Host IPC", () => {
+  beforeEach(() => vi.clearAllMocks());
+
+  it("starts only after the renderer is ready and forwards durable dispatches", async () => {
+    const sender = new FakeWebContents();
+    const host = {
+      start: vi.fn(),
+      enqueue: vi.fn(async () => [{ id: "job" }]),
+      listJobs: vi.fn(async () => []),
+      cancel: vi.fn(async () => true),
+    } as unknown as AgentHost;
+    const bridge = {
+      respond: vi.fn(() => true),
+    } as unknown as AgentHostRendererBridge;
+    const { handlers, ipc } = mainIPC();
+    setupAgentHostIPC(
+      {
+        host,
+        bridge,
+        getAllowedWebContents: () => sender as never,
+      },
+      ipc as never,
+    );
+
+    expect(
+      await handlers.get(AGENT_HOST_CHANNELS.READY)?.(event(sender)),
+    ).toEqual({ success: true });
+    expect(host.start).toHaveBeenCalledOnce();
+    expect(
+      await handlers.get(AGENT_HOST_CHANNELS.ENQUEUE)?.(
+        event(sender),
+        dispatch as never,
+      ),
+    ).toEqual({ success: true, jobs: [{ id: "job" }] });
+    expect(host.enqueue).toHaveBeenCalledWith(dispatch);
+  });
+
+  it("rejects another renderer and does not accept forged responses", async () => {
+    const allowed = new FakeWebContents();
+    const attacker = new FakeWebContents();
+    const host = {
+      enqueue: vi.fn(),
+    } as unknown as AgentHost;
+    const bridge = {
+      respond: vi.fn(),
+    } as unknown as AgentHostRendererBridge;
+    const { handlers, ipc } = mainIPC();
+    setupAgentHostIPC(
+      {
+        host,
+        bridge,
+        getAllowedWebContents: () => allowed as never,
+      },
+      ipc as never,
+    );
+
+    expect(
+      await handlers.get(AGENT_HOST_CHANNELS.ENQUEUE)?.(
+        event(attacker),
+        dispatch as never,
+      ),
+    ).toMatchObject({
+      success: false,
+      error: expect.stringContaining("allowed"),
+    });
+    expect(
+      await handlers.get(AGENT_HOST_CHANNELS.RESPOND)?.(event(attacker), {
+        requestId: "request",
+        success: true,
+        data: {},
+      } as never),
+    ).toMatchObject({ success: false });
+    expect(host.enqueue).not.toHaveBeenCalled();
+    expect(bridge.respond).not.toHaveBeenCalled();
+  });
+});

packages/app/src/electro-bridge/ipc/agent-host-context.ts

@@ -0,0 +1,130 @@
+import type {
+  AgentHostDispatch,
+  AgentHostRendererResponse,
+} from "@/shared/types/agent-host";
+import type { AgentHost } from "@/electron/agent-host/host";
+import type { AgentHostRendererBridge } from "@/electron/agent-host/renderer-bridge";
+import {
+  ipcMain,
+  type IpcMain,
+  type IpcMainInvokeEvent,
+  type WebContents,
+} from "electron";
+import { AGENT_HOST_CHANNELS } from "./agent-host-api";
+import { isAllowedLocalAISender } from "./local-ai-context";
+
+export interface AgentHostIPCOptions {
+  host?: AgentHost;
+  bridge?: AgentHostRendererBridge;
+  getAllowedWebContents: () => WebContents | WebContents[] | null;
+}
+
+function errorMessage(error: unknown): string {
+  return error instanceof Error ? error.message : String(error);
+}
+
+function allowed(
+  event: IpcMainInvokeEvent,
+  options: AgentHostIPCOptions,
+): boolean {
+  return isAllowedLocalAISender(event, options.getAllowedWebContents());
+}
+
+export function setupAgentHostIPC(
+  options: AgentHostIPCOptions,
+  mainIPC: Pick<IpcMain, "handle" | "removeHandler"> = ipcMain,
+): void {
+  for (const channel of [
+    AGENT_HOST_CHANNELS.ENQUEUE,
+    AGENT_HOST_CHANNELS.READY,
+    AGENT_HOST_CHANNELS.LIST_JOBS,
+    AGENT_HOST_CHANNELS.CANCEL,
+    AGENT_HOST_CHANNELS.RESPOND,
+  ]) {
+    mainIPC.removeHandler(channel);
+  }
+
+  mainIPC.handle(AGENT_HOST_CHANNELS.READY, async (event) => {
+    if (!allowed(event, options)) {
+      return { success: false, error: "Agent Host IPC sender is not allowed." };
+    }
+    if (!options.host) {
+      return { success: false, error: "Agent Host is unavailable." };
+    }
+    options.host.start();
+    return { success: true };
+  });
+
+  mainIPC.handle(
+    AGENT_HOST_CHANNELS.ENQUEUE,
+    async (event, dispatch: AgentHostDispatch) => {
+      if (!allowed(event, options)) {
+        return {
+          success: false,
+          error: "Agent Host IPC sender is not allowed.",
+        };
+      }
+      if (!options.host) {
+        return { success: false, error: "Agent Host is unavailable." };
+      }
+      try {
+        return { success: true, jobs: await options.host.enqueue(dispatch) };
+      } catch (error) {
+        return { success: false, error: errorMessage(error) };
+      }
+    },
+  );
+
+  mainIPC.handle(AGENT_HOST_CHANNELS.LIST_JOBS, async (event) => {
+    if (!allowed(event, options)) {
+      return { success: false, error: "Agent Host IPC sender is not allowed." };
+    }
+    if (!options.host) {
+      return { success: false, error: "Agent Host is unavailable." };
+    }
+    try {
+      return { success: true, jobs: await options.host.listJobs() };
+    } catch (error) {
+      return { success: false, error: errorMessage(error) };
+    }
+  });
+
+  mainIPC.handle(AGENT_HOST_CHANNELS.CANCEL, async (event, jobId: unknown) => {
+    if (!allowed(event, options)) {
+      return { success: false, error: "Agent Host IPC sender is not allowed." };
+    }
+    if (!options.host) {
+      return { success: false, error: "Agent Host is unavailable." };
+    }
+    if (typeof jobId !== "string" || !jobId) {
+      return { success: false, error: "A job id is required." };
+    }
+    try {
+      return {
+        success: true,
+        cancelled: await options.host.cancel(jobId),
+      };
+    } catch (error) {
+      return { success: false, error: errorMessage(error) };
+    }
+  });
+
+  mainIPC.handle(
+    AGENT_HOST_CHANNELS.RESPOND,
+    async (event, response: AgentHostRendererResponse) => {
+      if (!allowed(event, options)) {
+        return {
+          success: false,
+          error: "Agent Host IPC sender is not allowed.",
+        };
+      }
+      if (!options.bridge) {
+        return { success: false, error: "Agent Host bridge is unavailable." };
+      }
+      return {
+        success: true,
+        accepted: options.bridge.respond(response),
+      };
+    },
+  );
+}

packages/app/src/electro-bridge/ipc/listeners-register.ts

@@ -2,6 +2,8 @@
 import { WindowSizeConfig } from "@/electron/windows/window-size";
 import { ThemeMode } from "@/shared/types/electron";
 import type { LocalAIRuntimeService } from "@/shared/types/local-ai";
+import type { AgentHost } from "@/electron/agent-host/host";
+import type { AgentHostRendererBridge } from "@/electron/agent-host/renderer-bridge";
 import {
   BrowserWindow,
   ipcMain,
@@ -34,6 +36,7 @@ import {
 import { setupLoggerIPC } from "./logger-context";
 import { setupLocalAIIPC } from "./local-ai-context";
 import { setupMCPIPC } from "./mcp-context";
+import { setupAgentHostIPC } from "./agent-host-context";
 
 // Extended interface that includes additional methods beyond IPCServer
 interface ElectronAPI extends IPCServer {
@@ -97,6 +100,8 @@ export interface ListenerOptions {
   mainWindow?: () => BrowserWindow | null;
   registerGlobalShortcuts?: () => void;
   localAIRuntime?: LocalAIRuntimeService;
+  agentHost?: AgentHost;
+  agentHostBridge?: AgentHostRendererBridge;
   /** Extra senders (currently the web bridge) allowed to drive local AI. */
   extraLocalAISenders?: () => WebContents[];
   /** Registration target; the web bridge swaps in a recording proxy. */
@@ -257,5 +262,18 @@ export function registerListeners(options: ListenerOptions = {}) {
     },
     ipc,
   );
+  setupAgentHostIPC(
+    {
+      host: options.agentHost,
+      bridge: options.agentHostBridge,
+      getAllowedWebContents: () => {
+        const window = options.mainWindow?.();
+        const renderer =
+          window && !window.isDestroyed() ? [window.webContents] : [];
+        return [...renderer, ...(options.extraLocalAISenders?.() ?? [])];
+      },
+    },
+    ipc,
+  );
   console.log("All IPC listeners registered successfully");
 }