fix(local-mcp): harden and test the MCP relay authorization boundary

designateRelayedMcpServers/handleMcpRelayRequest, the requestId dedupe, and
the expiry check had no coverage, so a regression in the per-run relay
authorization check would have shipped silently. Added tests driving the
real watch()/SSE entrypoint for: undesignated server, duplicate requestId,
expired request, and fire-and-forget execution.

Also fixed relayDesignations leaking one entry per relayed run for the life
of the app session (never deleted); it now evicts once a run reaches a
terminal status. Cleaned up stopWatcher's redundant double map lookup and
named the dedupe LRU's magic 1000 cap.

Author: richardsolomou

packages/core/src/cloud-task/cloud-task.test.ts

@@ -3076,3 +3076,211 @@ describe("CloudTaskService", () => {
     ).toHaveLength(3);
   });
 });
+
+describe("CloudTaskService MCP relay", () => {
+  let relayService: CloudTaskService;
+  let mcpRelayExecutor: {
+    execute: ReturnType<typeof vi.fn>;
+    closeRun: ReturnType<typeof vi.fn>;
+  };
+
+  beforeEach(() => {
+    const scopedLog = {
+      debug: vi.fn(),
+      info: vi.fn(),
+      warn: vi.fn(),
+      error: vi.fn(),
+    };
+    const loggerMock = { ...scopedLog, scope: vi.fn(() => scopedLog) };
+    const analyticsMock = { track: vi.fn() };
+    mcpRelayExecutor = {
+      execute: vi.fn(async () => ({
+        payload: { jsonrpc: "2.0", id: 1, result: {} },
+      })),
+      closeRun: vi.fn(async () => {}),
+    };
+    relayService = new CloudTaskService(
+      mockAuthService as never,
+      analyticsMock as never,
+      loggerMock,
+      mcpRelayExecutor as never,
+    );
+
+    mockNetFetch.mockReset();
+    mockStreamFetch.mockReset();
+    mockStreamTokenFetch.mockReset();
+    mockStreamTokenFetch.mockImplementation(() =>
+      Promise.resolve(
+        createJsonResponse({ token: "test-token", stream_base_url: null }),
+      ),
+    );
+    mockAuthService.authenticatedFetch.mockReset();
+    vi.stubGlobal("fetch", fetchRouter);
+    mockAuthService.authenticatedFetch.mockImplementation(
+      async (input: string | Request, init?: RequestInit) => {
+        return fetchRouter(input, {
+          ...init,
+          headers: {
+            ...(init?.headers ?? {}),
+            Authorization: "Bearer token",
+          },
+        });
+      },
+    );
+  });
+
+  afterEach(() => {
+    relayService.unwatchAll();
+    vi.unstubAllGlobals();
+  });
+
+  function mcpRequestSseLine(
+    overrides: Partial<{
+      requestId: string;
+      server: string;
+      expiresAt: string;
+    }> = {},
+  ): string {
+    const event = {
+      type: "mcp_request",
+      requestId: overrides.requestId ?? "req-1",
+      server: overrides.server ?? "slack",
+      payload: { jsonrpc: "2.0", id: 1, method: "initialize" },
+      expiresAt:
+        overrides.expiresAt ?? new Date(Date.now() + 60_000).toISOString(),
+    };
+    return `data: ${JSON.stringify(event)}\n\n`;
+  }
+
+  function watchRun(runId: string): void {
+    mockNetFetch.mockResolvedValueOnce(
+      createJsonResponse({
+        id: runId,
+        status: "in_progress",
+        stage: null,
+        output: null,
+        error_message: null,
+        branch: "main",
+        updated_at: "2026-01-01T00:00:00Z",
+      }),
+    );
+    relayService.watch({
+      taskId: "task-1",
+      runId,
+      apiHost: "https://app.example.com",
+      teamId: 2,
+      resumeFromEntryCount: 0,
+    });
+  }
+
+  it("drops a relay request for a server the run never designated", async () => {
+    mockStreamFetch.mockResolvedValueOnce(
+      createOpenSseResponse(mcpRequestSseLine({ server: "slack" })),
+    );
+    relayService.designateRelayedMcpServers("run-1", ["grafana"]);
+    watchRun("run-1");
+
+    await waitFor(() => mockStreamFetch.mock.calls.length > 0);
+    await new Promise((resolve) => setTimeout(resolve, 20));
+
+    expect(mcpRelayExecutor.execute).not.toHaveBeenCalled();
+  });
+
+  it("executes a designated relay request exactly once and posts mcp_response", async () => {
+    mockStreamFetch.mockResolvedValueOnce(
+      createOpenSseResponse(
+        mcpRequestSseLine({ requestId: "req-1", server: "slack" }) +
+          mcpRequestSseLine({ requestId: "req-1", server: "slack" }),
+      ),
+    );
+    mockNetFetch.mockResolvedValueOnce(createJsonResponse({ result: {} }));
+    relayService.designateRelayedMcpServers("run-1", ["slack"]);
+    watchRun("run-1");
+
+    await waitFor(() => mcpRelayExecutor.execute.mock.calls.length > 0);
+    // The duplicate line above must not trigger a second execution.
+    await new Promise((resolve) => setTimeout(resolve, 20));
+
+    expect(mcpRelayExecutor.execute).toHaveBeenCalledOnce();
+    expect(mcpRelayExecutor.execute).toHaveBeenCalledWith(
+      "run-1",
+      "slack",
+      expect.objectContaining({ method: "initialize" }),
+    );
+
+    await waitFor(() =>
+      mockNetFetch.mock.calls.some(([url]) =>
+        (url as string).includes("/command/"),
+      ),
+    );
+    const commandCall = mockNetFetch.mock.calls.find(([url]) =>
+      (url as string).includes("/command/"),
+    );
+    const body = JSON.parse((commandCall?.[1] as RequestInit).body as string);
+    expect(body).toEqual(
+      expect.objectContaining({
+        method: "mcp_response",
+        params: {
+          requestId: "req-1",
+          server: "slack",
+          payload: { jsonrpc: "2.0", id: 1, result: {} },
+        },
+      }),
+    );
+  });
+
+  it("evicts a run's relay designation once the run reaches a terminal status", async () => {
+    mockStreamFetch.mockResolvedValueOnce(
+      createOpenSseResponse(
+        `data: ${JSON.stringify({
+          type: "task_run_state",
+          status: "completed",
+          updated_at: "2026-01-01T00:00:01Z",
+        })}\n\n${mcpRequestSseLine({ server: "slack" })}`,
+      ),
+    );
+    relayService.designateRelayedMcpServers("run-1", ["slack"]);
+    watchRun("run-1");
+
+    await waitFor(() => mockStreamFetch.mock.calls.length > 0);
+    await new Promise((resolve) => setTimeout(resolve, 20));
+
+    expect(mcpRelayExecutor.execute).not.toHaveBeenCalled();
+  });
+
+  it("drops an expired relay request without executing it", async () => {
+    mockStreamFetch.mockResolvedValueOnce(
+      createOpenSseResponse(
+        mcpRequestSseLine({
+          server: "slack",
+          expiresAt: new Date(Date.now() - 60_000).toISOString(),
+        }),
+      ),
+    );
+    relayService.designateRelayedMcpServers("run-1", ["slack"]);
+    watchRun("run-1");
+
+    await waitFor(() => mockStreamFetch.mock.calls.length > 0);
+    await new Promise((resolve) => setTimeout(resolve, 20));
+
+    expect(mcpRelayExecutor.execute).not.toHaveBeenCalled();
+  });
+
+  it("sends no mcp_response for a fire-and-forget relay execution", async () => {
+    mcpRelayExecutor.execute.mockResolvedValueOnce({});
+    mockStreamFetch.mockResolvedValueOnce(
+      createOpenSseResponse(mcpRequestSseLine({ server: "slack" })),
+    );
+    relayService.designateRelayedMcpServers("run-1", ["slack"]);
+    watchRun("run-1");
+
+    await waitFor(() => mcpRelayExecutor.execute.mock.calls.length > 0);
+    await new Promise((resolve) => setTimeout(resolve, 20));
+
+    expect(
+      mockNetFetch.mock.calls.some(([url]) =>
+        (url as string).includes("/command/"),
+      ),
+    ).toBe(false);
+  });
+});

packages/core/src/cloud-task/cloud-task.ts

@@ -42,6 +42,7 @@ const SSE_HEALTHY_CONNECTION_MS = 60_000;
 const EVENT_BATCH_FLUSH_MS = 16;
 const EVENT_BATCH_MAX_SIZE = 50;
 const SESSION_LOG_PAGE_LIMIT = 5_000;
+const MAX_HANDLED_RELAY_REQUEST_IDS = 1_000;
 
 // Authoritative end-of-stream sentinel, matched on the SSE event name (event.event, not data.type).
 // The client stops on it without consulting run status.
@@ -400,7 +401,7 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
   private markRelayRequestHandled(requestId: string): void {
     this.handledRelayRequestIds.add(requestId);
     this.handledRelayRequestOrder.push(requestId);
-    if (this.handledRelayRequestOrder.length > 1000) {
+    if (this.handledRelayRequestOrder.length > MAX_HANDLED_RELAY_REQUEST_IDS) {
       const evicted = this.handledRelayRequestOrder.shift();
       if (evicted) this.handledRelayRequestIds.delete(evicted);
     }
@@ -771,15 +772,15 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
   }
 
   private stopWatcher(key: string): void {
-    const stopping = this.watchers.get(key);
-    if (stopping && this.relayDesignations.has(stopping.runId)) {
+    const watcher = this.watchers.get(key);
+    if (!watcher) return;
+
+    if (this.relayDesignations.has(watcher.runId)) {
       // No watcher → no relay events → nothing executes; release the run's
       // live server connections (stdio children included). They reopen
       // lazily if the run is watched again.
-      void this.mcpRelayExecutor?.closeRun?.(stopping.runId).catch(() => {});
+      void this.mcpRelayExecutor?.closeRun?.(watcher.runId).catch(() => {});
     }
-    const watcher = this.watchers.get(key);
-    if (!watcher) return;
 
     watcher.sseAbortController?.abort();
 
@@ -1801,6 +1802,12 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
       watcher.lastStatusUpdatedAt = updatedAt;
     }
 
+    // A terminal run gets no further relay requests; drop its designation so
+    // the map doesn't grow for the lifetime of the app session.
+    if (isTerminalStatus(watcher.lastStatus)) {
+      this.relayDesignations.delete(watcher.runId);
+    }
+
     return changed;
   }