feat(FN-4824): forward task assignment events across remote runtime

Fusion-Task-Id: FN-4824
Fusion-Task-Lineage: 65f53a33-0939-4b6f-b62b-4a255ad14d9e

packages/engine/src/runtimes/__tests__/remote-node-client.test.ts

@@ -123,6 +123,25 @@ describe("RemoteNodeClient", () => {
     );
   });
 
+  it("pollPendingAssignments() sends since cursor query", async () => {
+    const fetchMock = vi.fn().mockResolvedValue(
+      new Response(JSON.stringify([{ taskId: "KB-1", agentId: "agent-1", assignedAt: "2026-04-08T00:00:00.000Z" }]), {
+        status: 200,
+        headers: { "content-type": "application/json" },
+      })
+    );
+    globalThis.fetch = fetchMock as unknown as typeof fetch;
+
+    const client = new RemoteNodeClient({ baseUrl: BASE_URL, apiKey: API_KEY });
+    const result = await client.pollPendingAssignments({ since: "2026-04-08T00:00:00.000Z" });
+
+    expect(result).toEqual([{ taskId: "KB-1", agentId: "agent-1", assignedAt: "2026-04-08T00:00:00.000Z" }]);
+    expect(fetchMock).toHaveBeenCalledWith(
+      `${BASE_URL}/api/events/assignments?since=2026-04-08T00%3A00%3A00.000Z`,
+      expect.objectContaining({ method: "GET" })
+    );
+  });
+
   it("executeTask() posts to execute endpoint", async () => {
     const fetchMock = vi.fn().mockResolvedValue(
       new Response(JSON.stringify({ acknowledged: true }), {

packages/engine/src/runtimes/__tests__/remote-node-runtime.test.ts

@@ -7,6 +7,7 @@ const mockClientConstructor = vi.hoisted(() => vi.fn());
 const mockHealth = vi.hoisted(() => vi.fn());
 const mockGetMetrics = vi.hoisted(() => vi.fn());
 const mockStreamEvents = vi.hoisted(() => vi.fn());
+const mockPollPendingAssignments = vi.hoisted(() => vi.fn());
 
 vi.mock("../remote-node-client.js", () => ({
   RemoteNodeClient: vi.fn().mockImplementation((options: unknown) => {
@@ -15,6 +16,7 @@ vi.mock("../remote-node-client.js", () => ({
       health: mockHealth,
       getMetrics: mockGetMetrics,
       streamEvents: mockStreamEvents,
+      pollPendingAssignments: mockPollPendingAssignments,
     };
   }),
 }));
@@ -60,6 +62,7 @@ describe("RemoteNodeRuntime", () => {
     mockHealth.mockReset();
     mockGetMetrics.mockReset();
     mockStreamEvents.mockReset();
+    mockPollPendingAssignments.mockReset();
 
     mockHealth.mockResolvedValue({ status: "ok", version: "1.0.0", uptime: 100 });
     mockGetMetrics.mockResolvedValue({
@@ -70,6 +73,7 @@ describe("RemoteNodeRuntime", () => {
     mockStreamEvents.mockImplementation(({ signal }: { signal?: AbortSignal } = {}) =>
       idleStream(signal)
     );
+    mockPollPendingAssignments.mockResolvedValue([]);
   });
 
   afterEach(async () => {
@@ -173,6 +177,7 @@ describe("RemoteNodeRuntime", () => {
     const createdHandler = vi.fn();
     const movedHandler = vi.fn();
     const updatedHandler = vi.fn();
+    const assignedHandler = vi.fn();
     const errorHandler = vi.fn();
 
     mockStreamEvents.mockImplementation(({ signal }: { signal?: AbortSignal } = {}) =>
@@ -193,6 +198,11 @@ describe("RemoteNodeRuntime", () => {
             payload: { id: "KB-1", column: "done" },
             timestamp: NOW,
           },
+          {
+            type: "task:assigned",
+            payload: { taskId: "KB-1", agentId: "agent-1", assignedAt: NOW },
+            timestamp: NOW,
+          },
           {
             type: "error",
             payload: { message: "boom" },
@@ -212,6 +222,7 @@ describe("RemoteNodeRuntime", () => {
     runtime.on("task:created", createdHandler);
     runtime.on("task:moved", movedHandler);
     runtime.on("task:updated", updatedHandler);
+    runtime.on("task:assigned", assignedHandler);
     runtime.on("error", errorHandler);
 
     await runtime.start();
@@ -224,6 +235,12 @@ describe("RemoteNodeRuntime", () => {
         to: "in-progress",
       });
       expect(updatedHandler).toHaveBeenCalledWith({ id: "KB-1", column: "done" });
+      expect(assignedHandler).toHaveBeenCalledWith({
+        taskId: "KB-1",
+        agentId: "agent-1",
+        assignedAt: NOW,
+        source: "cross-node-push",
+      });
       expect(errorHandler).toHaveBeenCalledWith(expect.any(Error));
     });
 
@@ -256,6 +273,40 @@ describe("RemoteNodeRuntime", () => {
     await runtime.stop();
   });
 
+  it("polls pending assignments while reconnecting and deduplicates by assignedAt", async () => {
+    vi.useFakeTimers();
+
+    mockStreamEvents.mockImplementation(async function* () {
+      throw new Error("stream down");
+    });
+
+    mockPollPendingAssignments
+      .mockResolvedValueOnce([{ taskId: "KB-2", agentId: "agent-2", assignedAt: NOW }])
+      .mockResolvedValue([{ taskId: "KB-2", agentId: "agent-2", assignedAt: NOW }]);
+
+    const runtime = new RemoteNodeRuntime({
+      nodeConfig: createNode(),
+      projectId: "proj_poll",
+      projectName: "Project Poll",
+    });
+
+    (runtime as unknown as { reconnectBaseDelayMs: number }).reconnectBaseDelayMs = 1;
+    (runtime as unknown as { maxReconnectDelayMs: number }).maxReconnectDelayMs = 1;
+    (runtime as unknown as { maxReconnectAttempts: number }).maxReconnectAttempts = 2;
+
+    const assignedHandler = vi.fn();
+    runtime.on("task:assigned", assignedHandler);
+
+    await runtime.start();
+    await vi.advanceTimersByTimeAsync(5);
+
+    await vi.waitFor(() => {
+      expect(assignedHandler).toHaveBeenCalledTimes(1);
+    });
+
+    await runtime.stop();
+  });
+
   it("validates remote node config on start", async () => {
     const runtime = new RemoteNodeRuntime({
       nodeConfig: createNode({ type: "local", url: undefined, apiKey: undefined }),

packages/engine/src/runtimes/remote-node-client.ts

@@ -2,8 +2,25 @@ import type { Task, TaskCreateInput } from "@fusion/core";
 import type { RuntimeMetrics } from "../project-runtime.js";
 import { remoteNodeLog } from "../logger.js";
 
+export type RemoteNodeEventType =
+  | "task:created"
+  | "task:moved"
+  | "task:updated"
+  | "task:assigned"
+  | "error"
+  | (string & {});
+
+export interface RemoteNodeTaskAssignedPayload {
+  taskId: string;
+  agentId: string;
+  fromNodeId?: string;
+  toNodeId?: string;
+  leaseEpoch?: number;
+  assignedAt: string;
+}
+
 export interface RemoteNodeEvent {
-  type: string;
+  type: RemoteNodeEventType;
   payload: unknown;
   timestamp: string;
 }
@@ -95,6 +112,21 @@ export class RemoteNodeClient {
     );
   }
 
+  async pollPendingAssignments(options?: { since?: string }): Promise<RemoteNodeTaskAssignedPayload[]> {
+    const query = new URLSearchParams();
+    if (options?.since) {
+      query.set("since", options.since);
+    }
+    const path = query.size > 0
+      ? `/api/events/assignments?${query.toString()}`
+      : "/api/events/assignments";
+    return this.withRetry(() =>
+      this.requestJson<RemoteNodeTaskAssignedPayload[]>(path, {
+        method: "GET",
+      })
+    );
+  }
+
   async *streamEvents(options?: { signal?: AbortSignal }): AsyncIterable<RemoteNodeEvent> {
     const response = await this.withRetry(
       () => this.openStream("/api/events/stream", options?.signal),

packages/engine/src/runtimes/remote-node-runtime.ts

@@ -8,7 +8,11 @@ import type {
   RuntimeStatus,
 } from "../project-runtime.js";
 import { remoteNodeLog } from "../logger.js";
-import { RemoteNodeClient, type RemoteNodeEvent } from "./remote-node-client.js";
+import {
+  RemoteNodeClient,
+  type RemoteNodeEvent,
+  type RemoteNodeTaskAssignedPayload,
+} from "./remote-node-client.js";
 
 export interface RemoteNodeRuntimeConfig {
   nodeConfig: NodeConfig;
@@ -32,6 +36,8 @@ export class RemoteNodeRuntime
   private reconnectBaseDelayMs = 5_000;
   private maxReconnectDelayMs = 60_000;
   private maxReconnectAttempts = 10;
+  private lastAssignmentCursor: string | null = null;
+  private readonly seenAssignmentKeys = new Set<string>();
 
   constructor(private config: RemoteNodeRuntimeConfig) {
     super();
@@ -222,6 +228,8 @@ export class RemoteNodeRuntime
             `(attempt ${reconnectAttempts}/${this.maxReconnectAttempts})`
         );
 
+        await this.pollPendingAssignments("cross-node-poll");
+
         await this.sleep(delayMs, signal);
         if (signal.aborted) {
           return;
@@ -253,6 +261,9 @@ export class RemoteNodeRuntime
       case "task:updated":
         this.emit("task:updated", event.payload as Task);
         break;
+      case "task:assigned":
+        this.emitAssignmentWake(event.payload as RemoteNodeTaskAssignedPayload, "cross-node-push");
+        break;
       case "error": {
         const payload = event.payload;
         if (payload instanceof Error) {
@@ -272,6 +283,46 @@ export class RemoteNodeRuntime
     }
   }
 
+  async reconcileAssignments(assignments: RemoteNodeTaskAssignedPayload[]): Promise<void> {
+    for (const assignment of assignments) {
+      this.emitAssignmentWake(assignment, "cross-node-reconcile");
+    }
+  }
+
+  private async pollPendingAssignments(source: "cross-node-poll" | "cross-node-reconcile"): Promise<void> {
+    try {
+      const assignments = await this.client.pollPendingAssignments({
+        since: this.lastAssignmentCursor ?? undefined,
+      });
+      for (const assignment of assignments) {
+        this.emitAssignmentWake(assignment, source);
+      }
+    } catch (error) {
+      this.emitRuntimeError(this.toError(error));
+    }
+  }
+
+  private emitAssignmentWake(
+    assignment: RemoteNodeTaskAssignedPayload,
+    source: "cross-node-push" | "cross-node-poll" | "cross-node-reconcile"
+  ): void {
+    const key = `${assignment.taskId}:${assignment.agentId}:${assignment.assignedAt}`;
+    if (this.seenAssignmentKeys.has(key)) {
+      return;
+    }
+    this.seenAssignmentKeys.add(key);
+    this.lastAssignmentCursor = assignment.assignedAt;
+    remoteNodeLog.log(
+      `[wake-trigger-diagnostics] source=${source} taskId=${assignment.taskId} agentId=${assignment.agentId}`
+    );
+    this.emit("task:assigned", {
+      taskId: assignment.taskId,
+      agentId: assignment.agentId,
+      assignedAt: assignment.assignedAt,
+      source,
+    });
+  }
+
   private async refreshMetrics(): Promise<RuntimeMetrics> {
     try {
       const metrics = await this.client.getMetrics();