From 9aa2d5a70a035b9327a56714e0a4e077fc06e317 Mon Sep 17 00:00:00 2001 From: Ankit Bhardwaj <97785108+Ankit6149@users.noreply.github.com> Date: Fri, 25 Sep 2026 04:23:47 -0700 Subject: [PATCH 1/7] Add truthful generation progress contract --- frontend/lib/ai/generationProgress.mjs | 96 ++++++++++++++++++++++++++ 1 file changed, 96 insertions(+) create mode 100644 frontend/lib/ai/generationProgress.mjs diff --git a/frontend/lib/ai/generationProgress.mjs b/frontend/lib/ai/generationProgress.mjs new file mode 100644 index 00000000..b58a497c --- /dev/null +++ b/frontend/lib/ai/generationProgress.mjs @@ -0,0 +1,96 @@ +const TERMINAL_DESTINATION_STATES = new Set([ + "complete", + "needs_review", + "failed", + "cancelled", +]); + +function normalizedChannels(value) { + return Array.from(new Set((Array.isArray(value) ? value : []) + .map((channel) => String(channel || "").trim()) + .filter(Boolean))); +} + +export function createGenerationProgressReporter({ + channels = [], + onProgress = null, +} = {}) { + const selected = normalizedChannels(channels); + const destinations = Object.fromEntries(selected.map((channel) => [channel, "queued"])); + let strategy = "queued"; + let phase = "queued"; + let sequence = 0; + + function snapshot({ status = "", destination = "" } = {}) { + const completedDestinations = Object.values(destinations) + .filter((value) => TERMINAL_DESTINATION_STATES.has(value)) + .length; + return Object.freeze({ + schemaVersion: 1, + sequence, + phase, + status: String(status || ""), + destination: String(destination || ""), + strategy, + completedDestinations, + totalDestinations: selected.length, + destinations: Object.freeze({ ...destinations }), + }); + } + + function emit(details = {}) { + sequence += 1; + const event = snapshot(details); + if (typeof onProgress === "function") { + try { + onProgress(event); + } catch { + // Progress observers are advisory and must never break generation. + } + } + return event; + } + + function setStrategy(status) { + strategy = String(status || "queued"); + phase = "strategy"; + return emit({ status: strategy }); + } + + function queueDestinations() { + phase = "destinations"; + return emit({ status: "queued" }); + } + + function setDestination(channel, status) { + const key = String(channel || "").trim(); + if (!Object.prototype.hasOwnProperty.call(destinations, key)) return snapshot(); + destinations[key] = String(status || "queued"); + phase = "destinations"; + return emit({ status: destinations[key], destination: key }); + } + + function cancelOutstanding() { + for (const channel of selected) { + if (!TERMINAL_DESTINATION_STATES.has(destinations[channel])) { + destinations[channel] = "cancelled"; + } + } + phase = "cancelled"; + return emit({ status: "cancelled" }); + } + + function complete() { + phase = "complete"; + return emit({ status: "complete" }); + } + + return Object.freeze({ + setStrategy, + queueDestinations, + setDestination, + cancelOutstanding, + complete, + snapshot, + }); +} From 9515b31192a67d5474ff482c673f4c54501c52d6 Mon Sep 17 00:00:00 2001 From: Ankit Bhardwaj <97785108+Ankit6149@users.noreply.github.com> Date: Fri, 25 Sep 2026 04:24:08 -0700 Subject: [PATCH 2/7] Emit strategy and destination generation progress --- frontend/lib/ai/generateStudioPackage.js | 28 +++++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/frontend/lib/ai/generateStudioPackage.js b/frontend/lib/ai/generateStudioPackage.js index a0e6ab63..bb513a0f 100644 --- a/frontend/lib/ai/generateStudioPackage.js +++ b/frontend/lib/ai/generateStudioPackage.js @@ -15,6 +15,7 @@ import { estimateGenerationRequestBudget, generationRequestBudgetError, } from "./generationExecutionBudget.mjs"; +import { createGenerationProgressReporter } from "./generationProgress.mjs"; import { CHANNEL_CONTRACTS, assessChannelDraft, @@ -178,6 +179,7 @@ async function generateDestination({ }) { const packageKey = packageKeyForChannel(channel); const projectName = campaignBrief?.project?.name || generationInputs.projectName; + config.progressReporter?.setDestination?.(channel, "generating"); let firstDraft = null; let firstQuality = null; @@ -195,6 +197,7 @@ async function generateDestination({ }); if (firstQuality.valid) { + config.progressReporter?.setDestination?.(channel, "complete"); return { channel, packageKey, @@ -210,6 +213,7 @@ async function generateDestination({ }; } + config.progressReporter?.setDestination?.(channel, "revising"); const revisedRaw = await generateJSON({ provider, prompt: buildChannelPrompt({ @@ -231,6 +235,7 @@ async function generateDestination({ const selectedDraft = useRevision ? revisedDraft : firstDraft; const selectedQuality = useRevision ? revisedQuality : firstQuality; + config.progressReporter?.setDestination?.(channel, selectedQuality.valid ? "complete" : "needs_review"); return { channel, packageKey, @@ -250,6 +255,10 @@ async function generateDestination({ model: modelOverride || "", }); const safeError = providerErrorPayload(providerError); + config.progressReporter?.setDestination?.( + channel, + safeError.code === "provider_request_cancelled" ? "cancelled" : "failed", + ); return { channel, packageKey, @@ -282,6 +291,7 @@ async function reviseDuplicateDestination({ const current = generatedDestinations.find((item) => item.channel === target.channel); if (!current || current.status.status === "failed") return current; + config.progressReporter?.setDestination?.(target.channel, "revising"); try { const revisedRaw = await generateJSON({ provider, @@ -301,6 +311,7 @@ async function reviseDuplicateDestination({ sourceContext: context, }); + config.progressReporter?.setDestination?.(target.channel, quality.valid ? "complete" : "needs_review"); return { ...current, draft: revisedDraft, @@ -323,6 +334,10 @@ async function reviseDuplicateDestination({ model: modelOverride || "", }); const safeError = providerErrorPayload(providerError); + config.progressReporter?.setDestination?.( + target.channel, + safeError.code === "provider_request_cancelled" ? "cancelled" : "needs_review", + ); return { ...current, status: { @@ -387,6 +402,10 @@ export async function generateStudioPackage(inputs) { }; const channels = selectedDestinationList(selectedChannels); + const progressReporter = createGenerationProgressReporter({ + channels, + onProgress: config.onProgress, + }); const context = buildUnifiedContext(generationInputs); const contextWarnings = Array.isArray(context.warnings) ? context.warnings : []; const campaignBriefPrompt = buildCampaignBriefPrompt(context); @@ -429,8 +448,9 @@ export async function generateStudioPackage(inputs) { requestBudget = requestBudget || createGenerationExecutionBudget({ maxRequests: requestBudgetPlan.hardMaxRequests, }); - const executionConfig = { ...config, requestBudget }; + const executionConfig = { ...config, requestBudget, progressReporter }; + progressReporter.setStrategy("generating"); const rawBrief = await generateJSON({ provider: generator, prompt: campaignBriefPrompt, @@ -448,6 +468,7 @@ export async function generateStudioPackage(inputs) { }); if (strategyQuality.status !== STRATEGY_QUALITY_STATES.COMPLETE) { + progressReporter.setStrategy(strategyQuality.status); return { ok: false, code: "strategy_quality_blocked", @@ -474,6 +495,8 @@ export async function generateStudioPackage(inputs) { }; } + progressReporter.setStrategy("complete"); + progressReporter.queueDestinations(); pkg.posts = emptyPackagePosts(); let generatedDestinations = await mapWithConcurrency(channels, (channel) => generateDestination({ @@ -487,6 +510,7 @@ export async function generateStudioPackage(inputs) { }), executionConfig.destinationConcurrency, { signal: executionConfig.signal }); if (executionConfig.signal?.aborted) { + progressReporter.cancelOutstanding(); const cancelled = new Error("Generation request was cancelled."); cancelled.code = "provider_request_cancelled"; cancelled.status = 499; @@ -516,6 +540,7 @@ export async function generateStudioPackage(inputs) { config: executionConfig, }), executionConfig.destinationConcurrency, { signal: executionConfig.signal }); if (executionConfig.signal?.aborted) { + progressReporter.cancelOutstanding(); const cancelled = new Error("Generation request was cancelled."); cancelled.code = "provider_request_cancelled"; cancelled.status = 499; @@ -571,6 +596,7 @@ export async function generateStudioPackage(inputs) { throw new Error(`Every selected destination failed: ${failedDestinations.map((item) => item.channel).join(", ")}.`); } + progressReporter.complete(); const generationExecution = { plan: requestBudgetPlan, actual: requestBudget.snapshot(), From 6283ea64597b1cc02dcca7ba5e02eef9aba69d66 Mon Sep 17 00:00:00 2001 From: Ankit Bhardwaj <97785108+Ankit6149@users.noreply.github.com> Date: Fri, 25 Sep 2026 04:24:47 -0700 Subject: [PATCH 3/7] Stream live generation progress to Studio clients --- frontend/app/api/launch_kit/route.js | 116 ++++++++++++++++++++------- 1 file changed, 89 insertions(+), 27 deletions(-) diff --git a/frontend/app/api/launch_kit/route.js b/frontend/app/api/launch_kit/route.js index a6316604..f0748432 100644 --- a/frontend/app/api/launch_kit/route.js +++ b/frontend/app/api/launch_kit/route.js @@ -18,6 +18,72 @@ import { readGenerationRequestBody } from "../../../lib/server/generationRequest const OWNER_ONLY_ENDPOINT_PROVIDERS = new Set(["custom", "ollama", "lmstudio"]); +function safeGenerationFailure(error) { + if (error instanceof ProviderError) { + const providerError = providerErrorPayload(error); + return { + ok: false, + error: providerError.message, + providerError, + warnings: [providerError.message], + }; + } + return { + ok: false, + error: "SignalFlow could not complete campaign generation.", + warnings: ["Campaign generation failed unexpectedly. Retry deliberately or inspect server diagnostics with the correlation context."], + }; +} + +function streamGeneration({ generationInput, generationConfig, warnings }) { + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + let closed = false; + const write = (value) => { + if (closed) return; + try { + controller.enqueue(encoder.encode(`${JSON.stringify(value)}\n`)); + } catch { + closed = true; + } + }; + Promise.resolve().then(async () => { + try { + const result = await generateStudioPackage({ + ...generationInput, + config: { + ...generationConfig, + onProgress: (progress) => write({ type: "progress", progress }), + }, + }); + const allWarnings = Array.from(new Set([...warnings, ...(result.warnings || [])])); + write({ type: "result", data: { ...result, warnings: allWarnings } }); + } catch (error) { + write({ type: "error", data: safeGenerationFailure(error) }); + } finally { + if (!closed) { + closed = true; + try { + controller.close(); + } catch { + // The browser may have cancelled the stream already. + } + } + } + }); + }, + }); + + return new Response(stream, { + status: 200, + headers: { + "Content-Type": "application/x-ndjson; charset=utf-8", + "Cache-Control": "no-store", + }, + }); +} + export const maxDuration = 60; export async function POST(request) { @@ -214,7 +280,7 @@ export async function POST(request) { } void enableAutoCapture; - const result = await generateStudioPackage({ + const generationInput = { projectName, notes, audience, @@ -228,13 +294,22 @@ export async function POST(request) { model_name: providerModelName || modelName, model_endpoint: providerBaseUrl || modelEndpoint, appUrl, - config: { - apiKey: providerApiKey, - baseUrl: providerBaseUrl, - modelName: providerModelName, - allowServerKey: isOwner, - signal: request.signal, - }, + }; + const generationConfig = { + apiKey: providerApiKey, + baseUrl: providerBaseUrl, + modelName: providerModelName, + allowServerKey: isOwner, + signal: request.signal, + }; + + if (request.headers.get("accept")?.includes("application/x-ndjson")) { + return streamGeneration({ generationInput, generationConfig, warnings }); + } + + const result = await generateStudioPackage({ + ...generationInput, + config: generationConfig, }); const allWarnings = Array.from(new Set([...warnings, ...(result.warnings || [])])); @@ -243,25 +318,12 @@ export async function POST(request) { headers: { "Content-Type": "application/json" }, }); } catch (error) { - if (error instanceof ProviderError) { - const providerError = providerErrorPayload(error); - return new Response(JSON.stringify({ - ok: false, - error: providerError.message, - providerError, - warnings: [providerError.message], - }), { - status: providerError.httpStatus && providerError.httpStatus >= 400 ? providerError.httpStatus : 502, - headers: { "Content-Type": "application/json" }, - }); - } - - return new Response(JSON.stringify({ - ok: false, - error: "SignalFlow could not complete campaign generation.", - warnings: ["Campaign generation failed unexpectedly. Retry deliberately or inspect server diagnostics with the correlation context."], - }), { - status: 500, + const failure = safeGenerationFailure(error); + const status = error instanceof ProviderError + ? (error.httpStatus && error.httpStatus >= 400 ? error.httpStatus : 502) + : 500; + return new Response(JSON.stringify(failure), { + status, headers: { "Content-Type": "application/json" }, }); } From 49835b40244657cb4ee30259e54e2c8ff00b0399 Mon Sep 17 00:00:00 2001 From: Ankit Bhardwaj <97785108+Ankit6149@users.noreply.github.com> Date: Fri, 25 Sep 2026 04:25:34 -0700 Subject: [PATCH 4/7] Render live destination generation progress --- frontend/app/page.js | 110 +++++++++++++++++++++++++++++++++++++++---- 1 file changed, 102 insertions(+), 8 deletions(-) diff --git a/frontend/app/page.js b/frontend/app/page.js index c62b8946..8eaf007d 100644 --- a/frontend/app/page.js +++ b/frontend/app/page.js @@ -207,6 +207,58 @@ async function readJsonResponse(response, fallbackMessage) { throw new Error(response.ok ? fallbackMessage : `${fallbackMessage} (HTTP ${response.status})`); } +async function readGenerationResponse(response, onProgress, fallbackMessage) { + const contentType = String(response.headers.get("content-type") || "").toLowerCase(); + if (!contentType.includes("application/x-ndjson") || !response.body) { + return readJsonResponse(response, fallbackMessage); + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + let finalData = null; + + const processLine = (line) => { + if (!line.trim()) return; + const event = safeJsonParse(line, null); + if (!event || typeof event !== "object") throw new Error(fallbackMessage); + if (event.type === "progress" && event.progress) { + onProgress?.(event.progress); + return; + } + if ((event.type === "result" || event.type === "error") && event.data) { + finalData = event.data; + } + }; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split("\n"); + buffer = lines.pop() || ""; + for (const line of lines) processLine(line); + } + buffer += decoder.decode(); + if (buffer.trim()) processLine(buffer); + + if (finalData && typeof finalData === "object") return finalData; + throw new Error(fallbackMessage); +} + +function generationProgressLabel(status) { + const labels = { + queued: "Queued", + generating: "Generating", + revising: "Revising", + complete: "Complete", + needs_review: "Needs review", + failed: "Failed", + cancelled: "Cancelled", + }; + return labels[String(status || "")] || "Preparing"; +} + function downloadText(filename, value, type = "text/plain") { const blob = new Blob([value], { type }); const url = URL.createObjectURL(blob); @@ -394,6 +446,7 @@ export default function Home() { const [files, setFiles] = useState([]); const [documentText, setDocumentText] = useState([]); const [busy, setBusy] = useState(false); + const [generationProgress, setGenerationProgress] = useState(null); const [message, setMessage] = useState(null); const [strategyReview, setStrategyReview] = useState(null); const [library, setLibrary] = useState([]); @@ -729,6 +782,7 @@ const sourceAndChannelsReady = sourceSignals > 0 && channels.length > 0; setFiles([]); setDocumentText([]); setPublishOptions({ reddit: { subreddit: "", title: "" } }); + setGenerationProgress(null); setMessage(null); navigateSection("studio"); } @@ -899,7 +953,10 @@ ${extractedText}`); }); const response = await fetch("/api/launch_kit", { method: "POST", - headers: authHeaders({ "Content-Type": "application/json" }), + headers: authHeaders({ + "Content-Type": "application/json", + Accept: "application/x-ndjson", + }), signal, body: JSON.stringify({ project_name: form.projectName.trim() || "Untitled campaign", @@ -930,7 +987,11 @@ ${extractedText}`); }), }); - const data = await readJsonResponse(response, "SignalFlow returned an unreadable generation response."); + const data = await readGenerationResponse( + response, + setGenerationProgress, + "SignalFlow returned an unreadable generation response.", + ); if (data.code === "strategy_quality_blocked" && data.strategy_review) { return { strategyBlocked: true, data }; } @@ -950,6 +1011,7 @@ ${extractedText}`); } function beginGenerationRequest() { + setGenerationProgress(null); const controller = new AbortController(); generationAbortRef.current = controller; return controller; @@ -963,6 +1025,9 @@ ${extractedText}`); const controller = generationAbortRef.current; if (!controller || controller.signal.aborted) return; controller.abort(); + setGenerationProgress((previous) => previous + ? { ...previous, phase: "cancelled", status: "cancelled" } + : previous); setMessage({ type: "warning", text: "Cancelling generation. Existing drafts will remain unchanged." }); } @@ -2301,12 +2366,41 @@ async function exportZip() {