feat(mcp): background MCP tasks → task_notification

packages/core/src/agent/engine.ts

@@ -2074,6 +2074,9 @@ export class AgentEngine {
     // builders and deliberately untouched): entity-escape a literal closing task-notification tag
     // (any casing/inner whitespace) so a hostile result can't close THIS block early — the real
     // closing tag below must stay the only one.
+    // Escape BOTH ends of the tag, not just the closing one: escaping only `</task-notification>`
+    // still lets a hostile result emit a bare `<task-notification>` that reads as a nested block.
+    // Neither can be produced literally by a server result any more.
     const clean = (s: string) => this.sanitizeForReminder(s).replace(/<\/\s*task-notification\s*>/gi, "&lt;/task-notification&gt;");
     const label = clean(e.name ?? e.agentId);
     const summary = e.status === "completed" ? `Agent "${label}" completed`
@@ -2098,6 +2101,39 @@ export class AgentEngine {
     void this.runTurn(sessionId).catch((err) => console.error("bg-notification turn failed:", err));
   }
 
+  /** MCP sibling of notifyBgCompletion. Persists a settled background MCP task as a
+   *  task_notification and wakes the session the same way. Exactly-once via the registry's
+   *  takeForNotification claim, which matters here because BOTH the abort path and the stream's
+   *  settle path call onTaskSettled for the same key.
+   *
+   *  SANITISATION IS NOT OPTIONAL HERE. `result` is authored by a third-party MCP SERVER — less
+   *  trusted than the subagent output notifyBgCompletion already sanitises — and this event is
+   *  durable, replayed user-role into every later turn. Same two passes: sanitizeForReminder,
+   *  then entity-escape any closing task-notification tag so a hostile result cannot end the
+   *  block early and inject structure after it. */
+  notifyMcpTaskCompletion(sessionId: string, key: string): void {
+    const e = this.cfg.mcp?.tasks.takeForNotification(key);
+    if (!e) return;
+    // Escape BOTH ends of the tag. Escaping only the closing one still lets a hostile result emit
+    // a bare `<task-notification>` that reads as a nested block; neither can now be produced
+    // literally by server output. (notifyBgCompletion escapes only the closing tag — the same
+    // hardening would suit it, but changing that path is out of scope here.)
+    const clean = (s: string) => this.sanitizeForReminder(s)
+      .replace(/<\/\s*task-notification\s*>/gi, "&lt;/task-notification&gt;")
+      .replace(/<\s*task-notification\s*>/gi, "&lt;task-notification&gt;");
+    const label = clean(`${e.tool} on ${e.server}`);
+    const summary = e.status === "completed" ? `MCP task "${label}" completed`
+      : e.status === "failed" ? `MCP task "${label}" failed`
+      : `MCP task "${label}" was cancelled`;
+    const result = e.result ? `\n<result>${clean(e.result)}</result>` : "";
+    const content = `<task-notification>\n<task-id>${clean(e.key)}</task-id>\n<status>${e.status}</status>\n<summary>${summary}</summary>${result}\n</task-notification>`;
+    this.cfg.hub.append(sessionId, { type: "task_notification", sessionId, threadId: MAIN_THREAD, content });
+    // Same wake discipline as notifyBgCompletion: defer while a turn is in flight (runTurn's
+    // finally drains it), otherwise the session is idle and a fresh turn replays the event.
+    if (this.isRunning(sessionId)) { this.retriggerPending.add(sessionId); return; }
+    void this.runTurn(sessionId).catch((err) => console.error("mcp-task-notification turn failed:", err));
+  }
+
   /** Task B2: persists a completed/failed WORKFLOW run as a task_notification history event —
    *  mirrors notifyBgCompletion just above (bg-agent completions), reusing the SAME event variant
    *  (no new SessionEvent — the union already has task_notification). Exactly-once via

packages/core/src/agent/mcp/client.ts

@@ -16,6 +16,21 @@ export type McpContentBlock =
   | { type: "resource_link"; uri: string; name?: string; mimeType?: string }
   | { type: "other"; raw: unknown };
 
+/** A task in flight: its server-assigned id, plus a promise that settles when the stream reaches
+ *  its terminal message. `settled` never rejects — a failed task resolves with `ok: false`. */
+export type McpTaskHandle = { taskId: string; settled: Promise<{ ok: boolean; blocks: McpContentBlock[] }> };
+
+/** Shared content-block mapper: one implementation for callToolContent and callToolTask. */
+function toBlocks(content: unknown): McpContentBlock[] {
+  return ((content as any[]) ?? []).map((c: any): McpContentBlock => {
+    if (c?.type === "text") return { type: "text", text: String(c.text ?? "") };
+    if (c?.type === "image") return { type: "image", data: String(c.data ?? ""), mimeType: String(c.mimeType ?? "image/png") };
+    if (c?.type === "audio") return { type: "audio", data: String(c.data ?? ""), mimeType: String(c.mimeType ?? "audio/wav") };
+    if (c?.type === "resource_link") return { type: "resource_link", uri: String(c.uri ?? ""), name: c.name !== undefined ? String(c.name) : undefined, mimeType: c.mimeType !== undefined ? String(c.mimeType) : undefined };
+    return { type: "other", raw: c };
+  });
+}
+
 /** Per-call ceiling. The SDK's own default is 60s (DEFAULT_REQUEST_TIMEOUT_MSEC), which would be a
  *  REGRESSION here: the hand-rolled client this replaces had no per-request timeout at all, so a
  *  tool legally running longer than a minute works today. 10 minutes preserves every realistic
@@ -118,6 +133,8 @@ export class McpStdioClient {
     return text || (res?.isError ? "[mcp tool error]" : "");
   }
 
+  /** Structured sibling of `callTool`. Same request, but the content blocks are returned intact so
+   *  a caller holding a ToolContext can attach images instead of dropping them. */
   /** Structured sibling of `callTool`. Same request, but the content blocks are returned intact so
    *  a caller holding a ToolContext can attach images instead of dropping them. */
   async callToolContent(name: string, args: unknown, signal?: AbortSignal): Promise<{ blocks: McpContentBlock[]; isError: boolean }> {
@@ -127,14 +144,62 @@ export class McpStdioClient {
       undefined,
       { signal, timeout: callTimeoutMs(), resetTimeoutOnProgress: true },
     );
-    const blocks: McpContentBlock[] = (res?.content ?? []).map((c: any): McpContentBlock => {
-      if (c?.type === "text") return { type: "text", text: String(c.text ?? "") };
-      if (c?.type === "image") return { type: "image", data: String(c.data ?? ""), mimeType: String(c.mimeType ?? "image/png") };
-      if (c?.type === "audio") return { type: "audio", data: String(c.data ?? ""), mimeType: String(c.mimeType ?? "audio/wav") };
-      if (c?.type === "resource_link") return { type: "resource_link", uri: String(c.uri ?? ""), name: c.name !== undefined ? String(c.name) : undefined, mimeType: c.mimeType !== undefined ? String(c.mimeType) : undefined };
-      return { type: "other", raw: c };
-    });
-    return { blocks, isError: !!res?.isError };
+    return { blocks: toBlocks(res?.content), isError: !!res?.isError };
+  }
+
+  /** Task-aware call. `client.experimental.tasks.callToolStream` decides FOR US whether this
+   *  becomes a task: it augments the request only when the tool declares `execution.taskSupport`
+   *  AND the server advertised a `tasks.requests.tools.call` capability (client's private
+   *  isToolTask). Otherwise it sends a plain request. So a non-task server takes the "sync" arm
+   *  below and behaves exactly as it did before tasks existed — Norma adds no branching of its
+   *  own and never modifies a server-authored schema.
+   *
+   *  The generator is guaranteed to end in `result` or `error`. On the task arm we return as soon
+   *  as `taskCreated` arrives, so the turn is NOT held open, and keep draining in the background
+   *  to settle the handle. Verified live sequence:
+   *    taskCreated -> taskStatus(working) -> taskStatus(completed) -> result
+   *  A FAILING task arrives as the stream's `error` message (not a `result` with isError), which
+   *  is why the error arm settles rather than throws once a task exists. */
+  async callToolTask(
+    name: string,
+    args: unknown,
+    signal?: AbortSignal,
+    onCreated?: (taskId: string) => void,
+  ): Promise<{ kind: "sync"; blocks: McpContentBlock[]; isError: boolean } | { kind: "task"; handle: McpTaskHandle }> {
+    if (this._dead) throw new Error("mcp server is not running");
+    const stream = this.client!.experimental.tasks.callToolStream(
+      { name, arguments: (args ?? {}) as Record<string, unknown> },
+      undefined,
+      { signal, timeout: callTimeoutMs(), resetTimeoutOnProgress: true },
+    );
+    const iter = stream[Symbol.asyncIterator]();
+
+    for (;;) {
+      const { value, done } = await iter.next();
+      if (done) return { kind: "sync", blocks: [], isError: true };
+      const msg: any = value;
+      if (msg.type === "taskCreated") {
+        const taskId = String(msg.task?.taskId ?? "");
+        onCreated?.(taskId);
+        const settled = (async (): Promise<{ ok: boolean; blocks: McpContentBlock[] }> => {
+          for (;;) {
+            const n = await iter.next();
+            if (n.done) return { ok: false, blocks: [{ type: "text", text: "[mcp task ended without a result]" }] };
+            const m: any = n.value;
+            if (m.type === "result") return { ok: !m.result?.isError, blocks: toBlocks(m.result?.content) };
+            if (m.type === "error") return { ok: false, blocks: [{ type: "text", text: String(m.error?.message ?? "mcp task failed") }] };
+            // 'taskStatus' — progress only; keep draining.
+          }
+        })().catch((e): { ok: boolean; blocks: McpContentBlock[] } => (
+          { ok: false, blocks: [{ type: "text", text: String((e as Error)?.message ?? "mcp task failed") }] }
+        ));
+        return { kind: "task", handle: { taskId, settled } };
+      }
+      if (msg.type === "result") return { kind: "sync", blocks: toBlocks(msg.result?.content), isError: !!msg.result?.isError };
+      // No task was created, so there is nothing to settle later — a hard failure of the call
+      // itself, which the caller expects as a throw (matching callTool/callToolContent).
+      if (msg.type === "error") throw new Error(String(msg.error?.message ?? "mcp tool error"));
+    }
   }
 
   async listResources(signal?: AbortSignal): Promise<McpResourceInfo[]> {
@@ -168,6 +233,13 @@ export class McpStdioClient {
     }));
   }
 
+  /** Upstream cancellation for a task in flight. Best-effort: a dead client is a no-op, since the
+   *  server is already gone. */
+  async cancelTask(taskId: string): Promise<void> {
+    if (this._dead) return;
+    await this.client!.experimental.tasks.cancelTask(taskId);
+  }
+
   stop(): void {
     try { void this.client?.close(); } catch { /* ignore */ }
     try { void this.transport?.close(); } catch { /* ignore */ }

packages/core/src/agent/mcp/manager.ts

@@ -3,6 +3,8 @@ import { readFileSync, realpathSync } from "node:fs";
 import { join } from "node:path";
 import { McpStdioClient } from "./client";
 import { attachImageGuarded } from "../tools/attach-image";
+import { McpTaskRegistry, taskKey } from "./task-registry";
+import type { McpContentBlock } from "./client";
 import type { ToolRegistry } from "../tools/registry";
 import type { TrustStore } from "../trust";
 
@@ -26,7 +28,22 @@ export class McpManager {
   private inFlight = new Map<string, Promise<void>>();
   private pluginState: Array<{ display: string; status: McpServerStatus["status"]; toolNames: string[]; client?: McpStdioClient }> = [];
   private pluginToolNames: string[] = []; // full `mcp__<plugin>_<server>__<tool>` names, for stopAll teardown
-  constructor(private readonly deps: { registry: ToolRegistry; trust: TrustStore; log?: (m: string) => void }) {}
+  /** Public so the engine can claim settled entries via takeForNotification. */
+  readonly tasks = new McpTaskRegistry();
+
+  /** Fired once per task reaching a terminal state. MUTABLE on purpose: the daemon constructs
+   *  McpManager before the engine exists, so it assigns this afterwards (daemon.ts's existing
+   *  later-assigned-closure pattern). Also accepted as a constructor dep for tests. */
+  onTaskSettled?: (sessionId: string, key: string) => void;
+
+  constructor(private readonly deps: {
+    registry: ToolRegistry;
+    trust: TrustStore;
+    log?: (m: string) => void;
+    onTaskSettled?: (sessionId: string, key: string) => void;
+  }) {
+    this.onTaskSettled = deps.onTaskSettled;
+  }
 
   /**
    * Shared per-server bring-up used by startAll/doEnsureProject/startPlugins: spawn the client,
@@ -45,32 +62,62 @@ export class McpManager {
    *    (this fail-fast semantics is pinned by an existing test: a server whose OWN tool list has
    *    an internal duplicate name is entirely marked failed, not partially registered).
    */
-  /** The `run` closure every registered MCP tool shares. Uses `callToolContent` rather than
-   *  `callTool` so a non-text block can be ATTACHED instead of flattened: an image goes through
-   *  `attachImageGuarded` — the same path `read_mcp_resource` uses, so the size guard and the
-   *  "[image omitted: …]" fallback behave identically — and only its failure text joins the
-   *  string result. `ToolRunResult` is unchanged (still a string); images travel out-of-band on
-   *  `ctx.attachImage`, which is why this needed no registry-contract change. */
-  private toolRunner(client: McpStdioClient, toolName: string) {
-    return async (a: unknown, ctx: { signal?: AbortSignal; attachImage?: (dataUrl: string) => void }): Promise<string> => {
-      const { blocks, isError } = await client.callToolContent(toolName, a, ctx.signal);
-      const parts: string[] = [];
-      for (const b of blocks) {
-        if (b.type === "text") { parts.push(b.text); continue; }
-        if (b.type === "image") {
-          const omitted = attachImageGuarded(ctx, { mime: b.mimeType, base64: b.data });
-          if (omitted) parts.push(omitted);
-          continue;
-        }
-        if (b.type === "resource_link") { parts.push(`[resource: ${b.uri}${b.mimeType ? ` (${b.mimeType})` : ""}]`); continue; }
-        if (b.type === "audio") { parts.push(`[audio omitted: ${b.mimeType}]`); continue; }
-        parts.push("[non-text content omitted]");
-      }
-      const text = parts.join("");
-      return text || (isError ? "[mcp tool error]" : "");
+  /** The `run` closure every registered MCP tool shares.
+   *
+   *  Routes through `callToolTask`, which lets the SDK decide per call whether this becomes a
+   *  task (only when the tool declares execution.taskSupport AND the server advertised the
+   *  capability). A non-task server therefore takes the `sync` arm and behaves exactly as before
+   *  tasks existed — this is why every pre-existing MCP test still passes untouched. */
+  private toolRunner(client: McpStdioClient, serverKey: string, toolName: string) {
+    return async (a: unknown, ctx: { signal?: AbortSignal; attachImage?: (dataUrl: string) => void; sessionId?: string }): Promise<string> => {
+      const out = await client.callToolTask(toolName, a, ctx.signal);
+      if (out.kind === "sync") return this.renderBlocks(out.blocks, out.isError, ctx);
+
+      const { taskId, settled } = out.handle;
+      const key = taskKey(serverKey, taskId);
+      const sessionId = ctx.sessionId ?? "";
+      this.tasks.register({ sessionId, server: serverKey, tool: toolName, taskId });
+
+      // ctx.signal aborts the CLIENT-side wait; tell the SERVER too so it stops working. The
+      // registry's first-terminal-state-wins rule settles the race between this and `settled`.
+      ctx.signal?.addEventListener("abort", () => {
+        void client.cancelTask(taskId).catch(() => {});
+        this.tasks.complete(key, { ok: false, result: "", status: "cancelled" });
+        this.onTaskSettled?.(sessionId, key);
+      }, { once: true });
+
+      void settled.then((s) => {
+        // renderBlocks WITHOUT ctx: the turn that owned ctx.attachImage has already ended, so an
+        // image cannot be attached and degrades to a labelled placeholder instead of vanishing.
+        this.tasks.complete(key, { ok: s.ok, result: this.renderBlocks(s.blocks, !s.ok) });
+        this.onTaskSettled?.(sessionId, key);
+      });
+
+      return `Task started: ${toolName} on ${serverKey} (task ${taskId}). It runs in the background; you will be notified when it finishes.`;
     };
   }
 
+  /** Shared block→string rendering. With a ctx, images attach out-of-band via the same
+   *  attachImageGuarded path read_mcp_resource uses; without one (a task settling after its turn
+   *  ended) they degrade to a labelled placeholder. */
+  private renderBlocks(blocks: McpContentBlock[], isError: boolean, ctx?: { attachImage?: (dataUrl: string) => void }): string {
+    const parts: string[] = [];
+    for (const b of blocks) {
+      if (b.type === "text") { parts.push(b.text); continue; }
+      if (b.type === "image") {
+        if (!ctx) { parts.push(`[image omitted: ${b.mimeType} — task completed after its turn]`); continue; }
+        const omitted = attachImageGuarded(ctx, { mime: b.mimeType, base64: b.data });
+        if (omitted) parts.push(omitted);
+        continue;
+      }
+      if (b.type === "resource_link") { parts.push(`[resource: ${b.uri}${b.mimeType ? ` (${b.mimeType})` : ""}]`); continue; }
+      if (b.type === "audio") { parts.push(`[audio omitted: ${b.mimeType}]`); continue; }
+      parts.push("[non-text content omitted]");
+    }
+    const text = parts.join("");
+    return text || (isError ? "[mcp tool error]" : "");
+  }
+
   private async startOne(
     serverKey: string,
     cfg: McpServerConfig,
@@ -95,7 +142,7 @@ export class McpManager {
               args: z.object({}).passthrough(),
               rawParameters: t.inputSchema,
               scope: opts?.scope,
-              run: this.toolRunner(client, t.name),
+              run: this.toolRunner(client, serverKey, t.name),
             });
           } catch (e) {
             this.deps.log?.(`mcp: register '${full}' failed: ${(e as Error).message}`);
@@ -108,7 +155,7 @@ export class McpManager {
             args: z.object({}).passthrough(),
             rawParameters: t.inputSchema,
             scope: opts?.scope,
-            run: this.toolRunner(client, t.name),
+            run: this.toolRunner(client, serverKey, t.name),
           });
         }
         toolNames.push(t.name);