Make MCP campaign execution trackable and non-blocking

mcp/lib/executionRegistry.mjs

@@ -0,0 +1,149 @@
+import { randomUUID } from "node:crypto";
+
+function nowIso(now = () => new Date()) {
+  return now().toISOString();
+}
+
+function clone(value) {
+  if (value === undefined) return undefined;
+  return JSON.parse(JSON.stringify(value));
+}
+
+export function createExecutionRegistry({ now = () => new Date(), idFactory = () => `campaign-${randomUUID()}` } = {}) {
+  const jobs = new Map();
+
+  function snapshot(job) {
+    if (!job) return null;
+    return clone({
+      id: job.id,
+      status: job.status,
+      phase: job.phase,
+      createdAt: job.createdAt,
+      updatedAt: job.updatedAt,
+      startedAt: job.startedAt,
+      completedAt: job.completedAt,
+      cancelledAt: job.cancelledAt,
+      cancellationRequested: job.cancellationRequested,
+      metadata: job.metadata,
+      progress: job.progress,
+      result: job.result,
+      error: job.error,
+    });
+  }
+
+  function update(job, patch = {}) {
+    Object.assign(job, patch, { updatedAt: nowIso(now) });
+    return snapshot(job);
+  }
+
+  function start(run, { metadata = {} } = {}) {
+    if (typeof run !== "function") throw new TypeError("Execution registry requires a run function.");
+    const id = idFactory();
+    const controller = new AbortController();
+    const createdAt = nowIso(now);
+    const job = {
+      id,
+      status: "queued",
+      phase: "queued",
+      createdAt,
+      updatedAt: createdAt,
+      startedAt: null,
+      completedAt: null,
+      cancelledAt: null,
+      cancellationRequested: false,
+      metadata: clone(metadata),
+      progress: null,
+      result: null,
+      error: null,
+      controller,
+    };
+    jobs.set(id, job);
+
+    const reportProgress = (progress = {}) => {
+      if (!jobs.has(id) || job.status === "cancelled") return;
+      update(job, {
+        progress: clone(progress),
+        phase: String(progress.phase || job.phase || "running"),
+      });
+    };
+
+    queueMicrotask(async () => {
+      if (job.cancellationRequested) {
+        update(job, {
+          status: "cancelled",
+          phase: "cancelled",
+          cancelledAt: nowIso(now),
+        });
+        return;
+      }
+
+      update(job, {
+        status: "running",
+        phase: "generating",
+        startedAt: nowIso(now),
+      });
+
+      try {
+        const result = await run({ signal: controller.signal, reportProgress });
+        if (job.cancellationRequested || controller.signal.aborted) {
+          update(job, {
+            status: "cancelled",
+            phase: "cancelled",
+            cancelledAt: nowIso(now),
+            result: null,
+          });
+          return;
+        }
+        update(job, {
+          status: "completed",
+          phase: "completed",
+          completedAt: nowIso(now),
+          result: clone(result),
+          error: null,
+        });
+      } catch (error) {
+        if (job.cancellationRequested || controller.signal.aborted || error?.name === "AbortError") {
+          update(job, {
+            status: "cancelled",
+            phase: "cancelled",
+            cancelledAt: nowIso(now),
+            result: null,
+            error: null,
+          });
+          return;
+        }
+        update(job, {
+          status: "failed",
+          phase: "failed",
+          completedAt: nowIso(now),
+          result: null,
+          error: {
+            message: String(error?.message || "SignalFlow campaign execution failed."),
+          },
+        });
+      }
+    });
+
+    return snapshot(job);
+  }
+
+  function get(id) {
+    return snapshot(jobs.get(String(id || "")));
+  }
+
+  function cancel(id) {
+    const job = jobs.get(String(id || ""));
+    if (!job) return null;
+    if (["completed", "failed", "cancelled"].includes(job.status)) return snapshot(job);
+    job.cancellationRequested = true;
+    job.controller.abort();
+    return update(job, {
+      phase: "cancelling",
+      cancellationRequested: true,
+    });
+  }
+
+  return { start, get, cancel };
+}
+
+export const campaignExecutionRegistry = createExecutionRegistry();

mcp/lib/httpClient.mjs

@@ -21,6 +21,7 @@ export async function signalFlowRequest(path, {
   env = process.env,
   fetchImpl = globalThis.fetch,
   timeoutMs = 120000,
+  signal,
 } = {}) {
   if (typeof fetchImpl !== "function") {
     throw new Error("This Node runtime does not provide fetch(). Use Node 20 or newer.");
@@ -29,6 +30,9 @@ export async function signalFlowRequest(path, {
   const baseUrl = getSignalFlowBaseUrl(env);
   const accessKey = getSignalFlowAccessKey(env);
   const controller = new AbortController();
+  const abortFromCaller = () => controller.abort(signal?.reason);
+  if (signal?.aborted) abortFromCaller();
+  else signal?.addEventListener?.("abort", abortFromCaller, { once: true });
   const timeout = setTimeout(() => controller.abort(), timeoutMs);
 
   const headers = { Accept: "application/json" };
@@ -82,5 +86,6 @@ export async function signalFlowRequest(path, {
     throw error;
   } finally {
     clearTimeout(timeout);
+    signal?.removeEventListener?.("abort", abortFromCaller);
   }
 }

mcp/lib/tools.mjs

@@ -4,6 +4,7 @@ import {
   validateSourceGraph,
 } from "../../frontend/lib/domain/sourceArtifacts.mjs";
 import { signalFlowRequest } from "./httpClient.mjs";
+import { campaignExecutionRegistry } from "./executionRegistry.mjs";
 
 const CHANNELS = [
   "linkedin",
@@ -55,6 +56,55 @@ export const TOOL_DEFINITIONS = [
       additionalProperties: false,
     },
   },
+  {
+    name: "signalflow_start_campaign",
+    description: "Start campaign generation as trackable MCP work and return immediately with a job ID. Use campaign status and cancel tools while generation continues.",
+    inputSchema: {
+      type: "object",
+      required: ["projectName", "notes", "provider", "channels"],
+      properties: {
+        projectName: { type: "string", minLength: 1 },
+        notes: { type: "string", minLength: 1 },
+        audience: { type: "string" },
+        links: { type: "string" },
+        repository: { type: "string" },
+        provider: { type: "string", enum: PROVIDERS },
+        modelName: { type: "string" },
+        baseUrl: { type: "string" },
+        channels: {
+          type: "array",
+          minItems: 1,
+          uniqueItems: true,
+          items: { type: "string", enum: CHANNELS },
+        },
+        documentText: { type: "array", items: { type: "string" } },
+        assets: { type: "array", items: { type: "object", additionalProperties: true } },
+        sourceArtifacts: { type: "array", items: { type: "object", additionalProperties: true } },
+        processingRecords: { type: "array", items: { type: "object", additionalProperties: true } },
+      },
+      additionalProperties: false,
+    },
+  },
+  {
+    name: "signalflow_campaign_status",
+    description: "Inspect a trackable SignalFlow MCP campaign job without blocking unrelated MCP requests.",
+    inputSchema: {
+      type: "object",
+      required: ["jobId"],
+      properties: { jobId: { type: "string", minLength: 1 } },
+      additionalProperties: false,
+    },
+  },
+  {
+    name: "signalflow_cancel_campaign",
+    description: "Request cancellation of queued or active SignalFlow MCP campaign work.",
+    inputSchema: {
+      type: "object",
+      required: ["jobId"],
+      properties: { jobId: { type: "string", minLength: 1 } },
+      additionalProperties: false,
+    },
+  },
   {
     name: "signalflow_create_campaign",
     description: "Create a staged, destination-specific SignalFlow campaign from product evidence. This requires a real model provider and never uses local template copy.",
@@ -186,6 +236,78 @@ export async function executeTool(name, args = {}, options = {}) {
     };
   }
 
+  if (name === "signalflow_start_campaign") {
+    const projectName = requireString(args.projectName, "projectName");
+    const notes = requireString(args.notes, "notes");
+    const provider = requireProvider(args.provider);
+    const channels = requireChannels(args.channels);
+    const registry = options.executionRegistry || campaignExecutionRegistry;
+    const job = registry.start(async ({ signal, reportProgress }) => {
+      reportProgress({
+        phase: "generating",
+        completedDestinations: 0,
+        totalDestinations: channels.length,
+      });
+      return executeTool("signalflow_create_campaign", {
+        ...args,
+        projectName,
+        notes,
+        provider,
+        channels,
+      }, {
+        ...options,
+        signal,
+      });
+    }, {
+      metadata: {
+        projectName,
+        provider,
+        channels,
+      },
+    });
+    return {
+      content: textContent(`SignalFlow started campaign job ${job.id} for ${projectName}.`),
+      structuredContent: { ok: true, job },
+      isError: false,
+    };
+  }
+
+  if (name === "signalflow_campaign_status") {
+    const jobId = requireString(args.jobId, "jobId");
+    const registry = options.executionRegistry || campaignExecutionRegistry;
+    const job = registry.get(jobId);
+    if (!job) {
+      return {
+        content: textContent(`Unknown SignalFlow campaign job: ${jobId}.`),
+        structuredContent: { ok: false, code: "campaign_job_not_found", jobId },
+        isError: true,
+      };
+    }
+    return {
+      content: textContent(`SignalFlow campaign job ${job.id} is ${job.status} (${job.phase}).`),
+      structuredContent: { ok: true, job },
+      isError: job.status === "failed",
+    };
+  }
+
+  if (name === "signalflow_cancel_campaign") {
+    const jobId = requireString(args.jobId, "jobId");
+    const registry = options.executionRegistry || campaignExecutionRegistry;
+    const job = registry.cancel(jobId);
+    if (!job) {
+      return {
+        content: textContent(`Unknown SignalFlow campaign job: ${jobId}.`),
+        structuredContent: { ok: false, code: "campaign_job_not_found", jobId },
+        isError: true,
+      };
+    }
+    return {
+      content: textContent(`SignalFlow cancellation requested for campaign job ${job.id}.`),
+      structuredContent: { ok: true, job },
+      isError: false,
+    };
+  }
+
   if (name === "signalflow_create_campaign") {
     const projectName = requireString(args.projectName, "projectName");
     const notes = requireString(args.notes, "notes");

mcp/server.mjs

@@ -97,26 +97,29 @@ async function handleRequest(message) {
 }
 
 const input = readline.createInterface({ input: process.stdin, crlfDelay: Infinity });
-let chain = Promise.resolve();
+const activeRequests = new Set();
 
 input.on("line", (line) => {
   if (!line.trim()) return;
-  chain = chain.then(async () => {
-    let message;
-    try {
-      message = JSON.parse(line);
-    } catch {
-      error(null, -32700, "Parse error.");
-      return;
-    }
-    await handleRequest(message);
-  }).catch((unexpectedError) => {
+
+  let message;
+  try {
+    message = JSON.parse(line);
+  } catch {
+    error(null, -32700, "Parse error.");
+    return;
+  }
+
+  const request = handleRequest(message).catch((unexpectedError) => {
     console.error("SignalFlow MCP request failure:", unexpectedError);
+  }).finally(() => {
+    activeRequests.delete(request);
   });
+  activeRequests.add(request);
 });
 
 input.on("close", () => {
-  void chain.finally(() => process.exit(0));
+  void Promise.allSettled([...activeRequests]).finally(() => process.exit(0));
 });
 
 process.on("SIGINT", () => input.close());