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" }, }); } 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() {
-
- {sourceSignals} source signal{sourceSignals === 1 ? "" : "s"} - - {channels.length} destinations - - {provider.label} +
+ {busy && generationProgress ? ( + <> + + {generationProgress.phase === "strategy" + ? `Strategy · ${generationProgressLabel(generationProgress.strategy)}` + : generationProgress.phase === "cancelled" + ? "Generation · Cancelling" + : `Destinations · ${generationProgress.completedDestinations || 0}/${generationProgress.totalDestinations || channels.length} complete`} + + + {provider.label} +
+ {Object.entries(generationProgress.destinations || {}).map(([channelId, status]) => ( + + + {channelMeta(channelId).label} · {generationProgressLabel(status)} + + ))} +
+ + ) : ( + <> + {sourceSignals} source signal{sourceSignals === 1 ? "" : "s"} + + {channels.length} destinations + + {provider.label} + + )}
{busy && ( diff --git a/frontend/app/studio-product.css b/frontend/app/studio-product.css index ae8b130b..9cc8c811 100644 --- a/frontend/app/studio-product.css +++ b/frontend/app/studio-product.css @@ -495,6 +495,50 @@ background: var(--app-surface); } +.app-shell .studio-actionbar__summary.has-progress { + flex: 1 1 32rem; + flex-wrap: wrap; + min-width: 0; +} + +.app-shell .generation-progress-list { + flex: 1 1 100%; + display: flex; + flex-wrap: wrap; + gap: 0.35rem; + min-width: 0; +} + +.app-shell .generation-progress-chip { + display: inline-flex; + align-items: center; + gap: 0.3rem; + min-width: 0; + padding: 0.3rem 0.5rem; + border: 1px solid var(--app-line); + border-radius: 999px; + background: var(--app-surface-muted); + color: var(--app-muted); + font-size: 0.68rem; + line-height: 1; +} + +.app-shell .generation-progress-chip.is-generating, +.app-shell .generation-progress-chip.is-revising { + color: var(--app-ink); +} + +.app-shell .generation-progress-chip.is-complete { + color: var(--app-ink); + background: var(--app-surface); +} + +.app-shell .generation-progress-chip.is-failed, +.app-shell .generation-progress-chip.is-cancelled, +.app-shell .generation-progress-chip.is-needs_review { + font-weight: 700; +} + @media (max-width: 68rem) { .app-shell .studio-page[data-stage="source"], .app-shell .studio-page[data-stage="destinations"] { 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(), 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, + }); +} diff --git a/frontend/tests/generationLimits.test.mjs b/frontend/tests/generationLimits.test.mjs index 4030294d..25d703ce 100644 --- a/frontend/tests/generationLimits.test.mjs +++ b/frontend/tests/generationLimits.test.mjs @@ -121,10 +121,12 @@ test("launch kit applies body and field limits before provider generation", asyn const route = await readFile(new URL("../app/api/launch_kit/route.js", import.meta.url), "utf8"); const readIndex = route.indexOf("readGenerationRequestBody(request)"); const validateIndex = route.indexOf("validateGenerationInputs(body)"); - const generateIndex = route.indexOf("generateStudioPackage({"); + const streamedGenerationIndex = route.indexOf("return streamGeneration({", validateIndex); + const jsonGenerationIndex = route.indexOf("const result = await generateStudioPackage({", validateIndex); assert.ok(readIndex >= 0); assert.ok(validateIndex > readIndex); - assert.ok(generateIndex > validateIndex); + assert.ok(streamedGenerationIndex > validateIndex); + assert.ok(jsonGenerationIndex > validateIndex); assert.match(route, /status: parsedRequest\.status/); assert.match(route, /limitIssues: validation\.limitIssues/); }); diff --git a/frontend/tests/generationProgress.test.mjs b/frontend/tests/generationProgress.test.mjs new file mode 100644 index 00000000..80bec17c --- /dev/null +++ b/frontend/tests/generationProgress.test.mjs @@ -0,0 +1,111 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { readFile } from "node:fs/promises"; + +import { createGenerationProgressReporter } from "../lib/ai/generationProgress.mjs"; + +test("generation progress reports truthful strategy and destination transitions", () => { + const events = []; + const reporter = createGenerationProgressReporter({ + channels: ["linkedin", "x", "reddit"], + onProgress: (event) => events.push(event), + }); + + reporter.setStrategy("generating"); + reporter.setStrategy("complete"); + reporter.queueDestinations(); + reporter.setDestination("linkedin", "generating"); + reporter.setDestination("x", "generating"); + reporter.setDestination("linkedin", "complete"); + reporter.setDestination("x", "revising"); + reporter.setDestination("x", "needs_review"); + reporter.setDestination("reddit", "failed"); + reporter.complete(); + + assert.equal(events[0].phase, "strategy"); + assert.equal(events[0].strategy, "generating"); + assert.equal(events[2].destinations.linkedin, "queued"); + const last = events.at(-1); + assert.equal(last.phase, "complete"); + assert.equal(last.completedDestinations, 3); + assert.equal(last.totalDestinations, 3); + assert.deepEqual(last.destinations, { + linkedin: "complete", + x: "needs_review", + reddit: "failed", + }); + assert.deepEqual(events.map((event) => event.sequence), events.map((_, index) => index + 1)); +}); + +test("cancellation marks only unfinished destinations cancelled", () => { + const reporter = createGenerationProgressReporter({ + channels: ["linkedin", "x", "reddit"], + }); + reporter.setStrategy("complete"); + reporter.queueDestinations(); + reporter.setDestination("linkedin", "complete"); + reporter.setDestination("x", "generating"); + const cancelled = reporter.cancelOutstanding(); + + assert.equal(cancelled.phase, "cancelled"); + assert.deepEqual(cancelled.destinations, { + linkedin: "complete", + x: "cancelled", + reddit: "cancelled", + }); + assert.equal(cancelled.completedDestinations, 3); +}); + +test("progress observers cannot break generation state transitions", () => { + const reporter = createGenerationProgressReporter({ + channels: ["linkedin"], + onProgress() { + throw new Error("observer unavailable"); + }, + }); + assert.doesNotThrow(() => reporter.setStrategy("generating")); + assert.doesNotThrow(() => reporter.setDestination("linkedin", "complete")); + assert.equal(reporter.snapshot().destinations.linkedin, "complete"); +}); + +test("progress contract contains no draft, prompt, source, or provider response content", () => { + const reporter = createGenerationProgressReporter({ channels: ["linkedin"] }); + reporter.setDestination("linkedin", "generating"); + const serialized = JSON.stringify(reporter.snapshot()); + assert.doesNotMatch(serialized, /prompt|draft|source|responseBody|repository/i); +}); + +test("orchestration emits progress from strategy through destination repair and cancellation", async () => { + const source = await readFile(new URL("../lib/ai/generateStudioPackage.js", import.meta.url), "utf8"); + assert.match(source, /createGenerationProgressReporter/); + assert.match(source, /progressReporter\.setStrategy\("generating"\)/); + assert.match(source, /progressReporter\.queueDestinations\(\)/); + assert.match(source, /setDestination\?\.\(channel, "generating"\)/); + assert.match(source, /setDestination\?\.\(channel, "revising"\)/); + assert.match(source, /progressReporter\.cancelOutstanding\(\)/); + assert.match(source, /progressReporter\.complete\(\)/); +}); + +test("launch kit keeps JSON compatibility and streams progress only when requested", async () => { + const route = await readFile(new URL("../app/api/launch_kit/route.js", import.meta.url), "utf8"); + assert.match(route, /request\.headers\.get\("accept"\)\?\.includes\("application\/x-ndjson"\)/); + assert.match(route, /onProgress: \(progress\) => write\(\{ type: "progress", progress \}\)/); + assert.match(route, /write\(\{ type: "result", data:/); + assert.match(route, /write\(\{ type: "error", data: safeGenerationFailure\(error\) \}\)/); + assert.match(route, /"Content-Type": "application\/x-ndjson; charset=utf-8"/); + assert.match(route, /const result = await generateStudioPackage/); + assert.match(route, /"Content-Type": "application\/json"/); +}); + +test("Studio consumes live progress with JSON fallback and accessible status semantics", async () => { + const page = await readFile(new URL("../app/page.js", import.meta.url), "utf8"); + assert.match(page, /Accept: "application\/x-ndjson"/); + assert.match(page, /readGenerationResponse\(/); + assert.match(page, /response\.body\.getReader\(\)/); + assert.match(page, /event\.type === "progress"/); + assert.match(page, /setGenerationProgress/); + assert.match(page, /className="generation-progress-list"/); + assert.match(page, /aria-label="Destination generation progress"/); + assert.match(page, /aria-live=\{busy && generationProgress \? "polite" : undefined\}/); + assert.match(page, /Cancel generation/); +});