From 445ac22908a7455145b85124e3049378cd5a4035 Mon Sep 17 00:00:00 2001 From: Ankit Bhardwaj <97785108+Ankit6149@users.noreply.github.com> Date: Mon, 21 Sep 2026 13:24:24 -0700 Subject: [PATCH 1/7] Add MCP campaign execution registry --- mcp/lib/executionRegistry.mjs | 149 ++++++++++++++++++++++++++++++++++ 1 file changed, 149 insertions(+) create mode 100644 mcp/lib/executionRegistry.mjs diff --git a/mcp/lib/executionRegistry.mjs b/mcp/lib/executionRegistry.mjs new file mode 100644 index 00000000..22a7c3f4 --- /dev/null +++ b/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(); From 3b26c934dfca9a8c3b65d987b32dc1daed536da5 Mon Sep 17 00:00:00 2001 From: Ankit Bhardwaj <97785108+Ankit6149@users.noreply.github.com> Date: Mon, 21 Sep 2026 13:24:58 -0700 Subject: [PATCH 2/7] Allow MCP campaign requests to be cancelled --- mcp/lib/httpClient.mjs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/mcp/lib/httpClient.mjs b/mcp/lib/httpClient.mjs index 2a5a752e..dcdd2583 100644 --- a/mcp/lib/httpClient.mjs +++ b/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); } } From b04160cb3a8179c40e1d96492428dac6413b2b4c Mon Sep 17 00:00:00 2001 From: Ankit Bhardwaj <97785108+Ankit6149@users.noreply.github.com> Date: Mon, 21 Sep 2026 13:27:36 -0700 Subject: [PATCH 3/7] Add trackable MCP campaign workflow tools --- mcp/lib/tools.mjs | 122 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 122 insertions(+) diff --git a/mcp/lib/tools.mjs b/mcp/lib/tools.mjs index 7419b78e..070e136f 100644 --- a/mcp/lib/tools.mjs +++ b/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"); From 8102bec71c6e96f288ddcd071dfe662bdc9f382f Mon Sep 17 00:00:00 2001 From: Ankit Bhardwaj <97785108+Ankit6149@users.noreply.github.com> Date: Mon, 21 Sep 2026 13:28:24 -0700 Subject: [PATCH 4/7] Keep MCP protocol requests responsive during generation --- mcp/server.mjs | 27 +++++++++++++++------------ 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/mcp/server.mjs b/mcp/server.mjs index 20eff055..b81131b1 100644 --- a/mcp/server.mjs +++ b/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()); From 2b58925e25aa991d5d23b017ecdf2a418b04e313 Mon Sep 17 00:00:00 2001 From: Ankit Bhardwaj <97785108+Ankit6149@users.noreply.github.com> Date: Mon, 21 Sep 2026 13:29:43 -0700 Subject: [PATCH 5/7] Test MCP campaign execution registry --- mcp/tests/executionRegistry.test.mjs | 77 ++++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) create mode 100644 mcp/tests/executionRegistry.test.mjs diff --git a/mcp/tests/executionRegistry.test.mjs b/mcp/tests/executionRegistry.test.mjs new file mode 100644 index 00000000..b5b849fe --- /dev/null +++ b/mcp/tests/executionRegistry.test.mjs @@ -0,0 +1,77 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { createExecutionRegistry } from "../lib/executionRegistry.mjs"; + +function tick() { + return new Promise((resolve) => setImmediate(resolve)); +} + +test("execution registry tracks completion without blocking the caller", async () => { + let release; + const gate = new Promise((resolve) => { release = resolve; }); + const registry = createExecutionRegistry({ + idFactory: () => "campaign-test-1", + }); + + const started = registry.start(async ({ reportProgress }) => { + reportProgress({ phase: "generating", completedDestinations: 0, totalDestinations: 2 }); + await gate; + return { ok: true, drafts: 2 }; + }, { + metadata: { projectName: "SignalFlow", provider: "gemini", channels: ["linkedin", "x"] }, + }); + + assert.equal(started.id, "campaign-test-1"); + assert.equal(started.status, "queued"); + + await tick(); + const running = registry.get(started.id); + assert.equal(running.status, "running"); + assert.equal(running.phase, "generating"); + assert.equal(running.progress.totalDestinations, 2); + + release(); + await tick(); + await tick(); + + const completed = registry.get(started.id); + assert.equal(completed.status, "completed"); + assert.deepEqual(completed.result, { ok: true, drafts: 2 }); +}); + +test("execution registry cancellation aborts active work and preserves cancelled truth", async () => { + const registry = createExecutionRegistry({ + idFactory: () => "campaign-test-cancel", + }); + + const started = registry.start(({ signal }) => new Promise((resolve, reject) => { + signal.addEventListener("abort", () => { + const error = new Error("cancelled"); + error.name = "AbortError"; + reject(error); + }, { once: true }); + })); + + await tick(); + assert.equal(registry.get(started.id).status, "running"); + + const cancelling = registry.cancel(started.id); + assert.equal(cancelling.cancellationRequested, true); + assert.equal(cancelling.phase, "cancelling"); + + await tick(); + await tick(); + + const cancelled = registry.get(started.id); + assert.equal(cancelled.status, "cancelled"); + assert.equal(cancelled.phase, "cancelled"); + assert.equal(cancelled.result, null); + assert.equal(cancelled.error, null); +}); + +test("execution registry returns null for unknown jobs", () => { + const registry = createExecutionRegistry(); + assert.equal(registry.get("missing"), null); + assert.equal(registry.cancel("missing"), null); +}); From 2a4a703528e2812571ad5785c04de5a41f849046 Mon Sep 17 00:00:00 2001 From: Ankit Bhardwaj <97785108+Ankit6149@users.noreply.github.com> Date: Mon, 21 Sep 2026 13:30:54 -0700 Subject: [PATCH 6/7] Test trackable MCP campaign tools --- mcp/tests/tools.test.mjs | 77 +++++++++++++++++++++++++++++++++++++++- 1 file changed, 76 insertions(+), 1 deletion(-) diff --git a/mcp/tests/tools.test.mjs b/mcp/tests/tools.test.mjs index 6c7a1af1..dd460515 100644 --- a/mcp/tests/tools.test.mjs +++ b/mcp/tests/tools.test.mjs @@ -4,13 +4,16 @@ import assert from "node:assert/strict"; import { createCapabilitySnapshot } from "../../frontend/lib/capabilities/capabilityContract.mjs"; import { executeTool, TOOL_DEFINITIONS } from "../lib/tools.mjs"; -test("MCP exposes capabilities, provider status, provider test, and campaign creation tools", () => { +test("MCP exposes blocking compatibility plus trackable campaign workflow tools", () => { assert.deepEqual( TOOL_DEFINITIONS.map((tool) => tool.name), [ "signalflow_capabilities", "signalflow_provider_status", "signalflow_test_provider", + "signalflow_start_campaign", + "signalflow_campaign_status", + "signalflow_cancel_campaign", "signalflow_create_campaign", ], ); @@ -152,3 +155,75 @@ test("API failures become structured MCP errors instead of fake campaign output" assert.equal(result.structuredContent.ok, false); assert.equal("posts" in result.structuredContent, false); }); + + +test("trackable campaign tools start, inspect, and cancel through the shared execution registry", async () => { + const calls = []; + const fakeRegistry = { + start(run, options) { + calls.push({ type: "start", run, options }); + return { + id: "campaign-123", + status: "queued", + phase: "queued", + metadata: options.metadata, + }; + }, + get(jobId) { + calls.push({ type: "get", jobId }); + return { + id: jobId, + status: "running", + phase: "generating", + progress: { completedDestinations: 1, totalDestinations: 2 }, + }; + }, + cancel(jobId) { + calls.push({ type: "cancel", jobId }); + return { + id: jobId, + status: "running", + phase: "cancelling", + cancellationRequested: true, + }; + }, + }; + + const started = await executeTool("signalflow_start_campaign", { + projectName: "SignalFlow", + notes: "Evidence", + provider: "gemini", + channels: ["linkedin", "x"], + }, { executionRegistry: fakeRegistry }); + + assert.equal(started.isError, false); + assert.equal(started.structuredContent.job.id, "campaign-123"); + assert.deepEqual(calls[0].options.metadata.channels, ["linkedin", "x"]); + + const status = await executeTool("signalflow_campaign_status", { + jobId: "campaign-123", + }, { executionRegistry: fakeRegistry }); + assert.equal(status.structuredContent.job.status, "running"); + assert.equal(status.structuredContent.job.progress.completedDestinations, 1); + + const cancelled = await executeTool("signalflow_cancel_campaign", { + jobId: "campaign-123", + }, { executionRegistry: fakeRegistry }); + assert.equal(cancelled.structuredContent.job.phase, "cancelling"); + assert.equal(cancelled.structuredContent.job.cancellationRequested, true); +}); + +test("trackable campaign status fails safely for unknown job IDs", async () => { + const result = await executeTool("signalflow_campaign_status", { + jobId: "missing", + }, { + executionRegistry: { + get() { return null; }, + cancel() { return null; }, + start() { throw new Error("not used"); }, + }, + }); + + assert.equal(result.isError, true); + assert.equal(result.structuredContent.code, "campaign_job_not_found"); +}); From 5d27e443fceb05e7a9754c50331c8004d6587d45 Mon Sep 17 00:00:00 2001 From: Ankit Bhardwaj <97785108+Ankit6149@users.noreply.github.com> Date: Mon, 21 Sep 2026 13:32:19 -0700 Subject: [PATCH 7/7] Prove MCP protocol responsiveness during generation --- mcp/tests/server.test.mjs | 82 ++++++++++++++++++++++++++++++++++++++- 1 file changed, 80 insertions(+), 2 deletions(-) diff --git a/mcp/tests/server.test.mjs b/mcp/tests/server.test.mjs index fa4dcb20..e7d7f9f8 100644 --- a/mcp/tests/server.test.mjs +++ b/mcp/tests/server.test.mjs @@ -4,6 +4,7 @@ import { spawn } from "node:child_process"; import readline from "node:readline"; import { fileURLToPath } from "node:url"; import path from "node:path"; +import { createServer } from "node:http"; const currentDir = path.dirname(fileURLToPath(import.meta.url)); const serverPath = path.resolve(currentDir, "../server.mjs"); @@ -59,9 +60,10 @@ test("stdio server enforces lifecycle, initializes, and lists SignalFlow tools", send(child, { jsonrpc: "2.0", id: 2, method: "tools/list", params: {} }); const toolList = await waitForLine(lines, (message) => message.id === 2); - assert.equal(toolList.result.tools.length, 4); + assert.equal(toolList.result.tools.length, 7); assert.equal(toolList.result.tools[0].name, "signalflow_capabilities"); - assert.equal(toolList.result.tools[3].name, "signalflow_create_campaign"); + assert.equal(toolList.result.tools[3].name, "signalflow_start_campaign"); + assert.equal(toolList.result.tools[6].name, "signalflow_create_campaign"); send(child, { jsonrpc: "2.0", @@ -73,3 +75,79 @@ test("stdio server enforces lifecycle, initializes, and lists SignalFlow tools", assert.equal(unknownTool.error.code, -32602); assert.match(unknownTool.error.message, /unknown tool/i); }); + + +test("ping remains responsive while a blocking campaign tool is awaiting generation", async (t) => { + let releaseGeneration; + const generationGate = new Promise((resolve) => { releaseGeneration = resolve; }); + const backend = createServer(async (request, response) => { + if (request.url === "/api/launch_kit" && request.method === "POST") { + for await (const _chunk of request) { + // Consume request body before deliberately holding the response. + } + await generationGate; + response.writeHead(200, { "Content-Type": "application/json" }); + response.end(JSON.stringify({ + ok: true, + providerUsed: "gemini", + generation_status: { linkedin: { status: "generated", qualityStatus: "complete" } }, + posts: { linkedin: "Generated draft" }, + })); + return; + } + response.writeHead(404, { "Content-Type": "application/json" }); + response.end(JSON.stringify({ ok: false, error: "not found" })); + }); + await new Promise((resolve) => backend.listen(0, "127.0.0.1", resolve)); + t.after(() => backend.close()); + const address = backend.address(); + + const child = spawn(process.execPath, [serverPath], { + stdio: ["pipe", "pipe", "pipe"], + env: { + ...process.env, + SIGNALFLOW_BASE_URL: `http://127.0.0.1:${address.port}`, + SIGNALFLOW_GEMINI_API_KEY: "test-provider-key", + }, + }); + t.after(() => child.kill("SIGTERM")); + + const lines = []; + const output = readline.createInterface({ input: child.stdout }); + output.on("line", (line) => lines.push(JSON.parse(line))); + + send(child, { + jsonrpc: "2.0", + id: 10, + method: "initialize", + params: { protocolVersion: "2025-11-25", capabilities: {}, clientInfo: { name: "test", version: "1" } }, + }); + await waitForLine(lines, (message) => message.id === 10); + send(child, { jsonrpc: "2.0", method: "notifications/initialized", params: {} }); + + send(child, { + jsonrpc: "2.0", + id: 11, + method: "tools/call", + params: { + name: "signalflow_create_campaign", + arguments: { + projectName: "SignalFlow", + notes: "Evidence", + provider: "gemini", + channels: ["linkedin"], + }, + }, + }); + + await new Promise((resolve) => setTimeout(resolve, 50)); + send(child, { jsonrpc: "2.0", id: 12, method: "ping", params: {} }); + + const ping = await waitForLine(lines, (message) => message.id === 12, 1000); + assert.deepEqual(ping.result, {}); + assert.equal(lines.some((message) => message.id === 11), false); + + releaseGeneration(); + const generation = await waitForLine(lines, (message) => message.id === 11, 3000); + assert.equal(generation.result.isError, false); +});