diff --git a/frontend/lib/ai/generateStudioPackage.js b/frontend/lib/ai/generateStudioPackage.js index bb513a0f..ca6f8afb 100644 --- a/frontend/lib/ai/generateStudioPackage.js +++ b/frontend/lib/ai/generateStudioPackage.js @@ -9,7 +9,7 @@ import { assertModelGenerationProvider } from "./generationPolicy.mjs"; import { evaluateStrategyQuality, STRATEGY_QUALITY_STATES } from "./strategyQuality.mjs"; import { duplicateRevisionTargets } from "./crossChannelQuality.mjs"; import { normalizeProviderError, providerErrorPayload } from "./providerErrors.mjs"; -import { mapWithConcurrency } from "./generationConcurrency.mjs"; +import { mapDestinationsWithPolicy } from "./generationConcurrency.mjs"; import { createGenerationExecutionBudget, estimateGenerationRequestBudget, @@ -499,7 +499,7 @@ export async function generateStudioPackage(inputs) { progressReporter.queueDestinations(); pkg.posts = emptyPackagePosts(); - let generatedDestinations = await mapWithConcurrency(channels, (channel) => generateDestination({ + let generatedDestinations = await mapDestinationsWithPolicy(channels, (channel) => generateDestination({ channel, context, campaignBrief: pkg, @@ -507,7 +507,11 @@ export async function generateStudioPackage(inputs) { provider: generator, modelOverride, config: executionConfig, - }), executionConfig.destinationConcurrency, { signal: executionConfig.signal }); + }), { + isLocalProvider: localProvider, + configuredConcurrency: executionConfig.destinationConcurrency, + signal: executionConfig.signal, + }); if (executionConfig.signal?.aborted) { progressReporter.cancelOutstanding(); @@ -529,7 +533,7 @@ export async function generateStudioPackage(inputs) { }); if (duplicateTargets.length) { - const revised = await mapWithConcurrency(duplicateTargets, (target) => reviseDuplicateDestination({ + const revised = await mapDestinationsWithPolicy(duplicateTargets, (target) => reviseDuplicateDestination({ target, generatedDestinations, context, @@ -538,7 +542,12 @@ export async function generateStudioPackage(inputs) { provider: generator, modelOverride, config: executionConfig, - }), executionConfig.destinationConcurrency, { signal: executionConfig.signal }); + }), { + isLocalProvider: localProvider, + configuredConcurrency: executionConfig.destinationConcurrency, + signal: executionConfig.signal, + channelOf: (target) => target.channel, + }); if (executionConfig.signal?.aborted) { progressReporter.cancelOutstanding(); const cancelled = new Error("Generation request was cancelled."); diff --git a/frontend/lib/ai/generationConcurrency.mjs b/frontend/lib/ai/generationConcurrency.mjs index 6e9146e1..fa449cc1 100644 --- a/frontend/lib/ai/generationConcurrency.mjs +++ b/frontend/lib/ai/generationConcurrency.mjs @@ -1,4 +1,33 @@ export const DEFAULT_DESTINATION_CONCURRENCY = 2; +export const LOCAL_DESTINATION_CONCURRENCY = 1; +export const LONG_FORM_DESTINATION_CONCURRENCY = 1; + +export const LONG_FORM_DESTINATIONS = Object.freeze([ + "blog", + "newsletter", + "youtube", + "reddit", +]); + +function positiveInteger(value) { + const parsed = Number(value); + return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : null; +} + +export function resolveDestinationConcurrency({ + isLocalProvider = false, + configuredConcurrency = null, +} = {}) { + const providerCeiling = isLocalProvider + ? LOCAL_DESTINATION_CONCURRENCY + : DEFAULT_DESTINATION_CONCURRENCY; + const requested = positiveInteger(configuredConcurrency); + return requested ? Math.min(providerCeiling, requested) : providerCeiling; +} + +export function isLongFormDestination(channel) { + return LONG_FORM_DESTINATIONS.includes(String(channel || "").trim().toLowerCase()); +} export async function mapWithConcurrency(items, worker, concurrency = DEFAULT_DESTINATION_CONCURRENCY, { signal = null } = {}) { const values = Array.isArray(items) ? items : []; @@ -24,3 +53,56 @@ export async function mapWithConcurrency(items, worker, concurrency = DEFAULT_DE await Promise.all(Array.from({ length: limit }, () => runLane())); return results; } + +export async function mapDestinationsWithPolicy( + items, + worker, + { + isLocalProvider = false, + configuredConcurrency = null, + signal = null, + channelOf = (item) => item, + } = {}, +) { + const values = Array.isArray(items) ? items : []; + if (!values.length) return []; + if (typeof worker !== "function") throw new TypeError("mapDestinationsWithPolicy requires a worker function."); + + const indexed = values.map((value, index) => ({ + value, + index, + channel: String(channelOf(value) || "").trim().toLowerCase(), + })); + const shortForm = indexed.filter((item) => !isLongFormDestination(item.channel)); + const longForm = indexed.filter((item) => isLongFormDestination(item.channel)); + const output = new Array(values.length); + + const shortConcurrency = resolveDestinationConcurrency({ + isLocalProvider, + configuredConcurrency, + }); + + const runIndexed = async (item) => { + const result = await worker(item.value, item.index); + output[item.index] = result; + return result; + }; + + await mapWithConcurrency( + shortForm, + runIndexed, + shortConcurrency, + { signal }, + ); + + if (!signal?.aborted) { + await mapWithConcurrency( + longForm, + runIndexed, + LONG_FORM_DESTINATION_CONCURRENCY, + { signal }, + ); + } + + return output; +} diff --git a/frontend/tests/generationQualityContracts.test.mjs b/frontend/tests/generationQualityContracts.test.mjs index b1c9c604..be6e2bf4 100644 --- a/frontend/tests/generationQualityContracts.test.mjs +++ b/frontend/tests/generationQualityContracts.test.mjs @@ -14,7 +14,11 @@ import { assessCrossChannelDuplicates, duplicateRevisionTargets, } from "../lib/ai/crossChannelQuality.mjs"; -import { mapWithConcurrency } from "../lib/ai/generationConcurrency.mjs"; +import { + mapWithConcurrency, + mapDestinationsWithPolicy, + resolveDestinationConcurrency, +} from "../lib/ai/generationConcurrency.mjs"; import { createLinkedAbort } from "../lib/ai/requestAbort.mjs"; import { PROVIDER_ERROR_CODES, @@ -135,7 +139,7 @@ test("strategy quality distinguishes complete, needs_review, and failed with sta test("strategy gate occurs before any destination generation lane starts", async () => { const source = await readFile(new URL("../lib/ai/generateStudioPackage.js", import.meta.url), "utf8"); const gate = source.indexOf("strategyQuality.status !== STRATEGY_QUALITY_STATES.COMPLETE"); - const destinationStart = source.indexOf("mapWithConcurrency(channels"); + const destinationStart = source.indexOf("mapDestinationsWithPolicy(channels"); assert.ok(gate >= 0, "strategy quality gate must exist"); assert.ok(destinationStart >= 0, "destination orchestration must exist"); assert.ok(gate < destinationStart, "strategy must be accepted before destination calls begin"); @@ -403,7 +407,7 @@ test("generation cancellation is propagated through API, providers, UI, and camp ]; assert.match(routeSource, /signal: request\.signal/); - assert.match(packageSource, /\{ signal: executionConfig\.signal \}/); + assert.match(packageSource, /signal: executionConfig\.signal/); assert.match(pageSource, /Cancel generation/); assert.match(pageSource, /MARK_CHANNELS_CANCELLED/); @@ -445,3 +449,61 @@ test("provider cancellation keeps a distinct safe error class", () => { assert.equal(normalized.recoveryAction, "retry_destination"); assert.equal(normalized.httpStatus, 499); }); + + +test("destination scheduling respects provider ceilings and isolates long-form work", async () => { + assert.equal(resolveDestinationConcurrency({ isLocalProvider: false }), 2); + assert.equal(resolveDestinationConcurrency({ isLocalProvider: false, configuredConcurrency: 1 }), 1); + assert.equal(resolveDestinationConcurrency({ isLocalProvider: false, configuredConcurrency: 9 }), 2); + assert.equal(resolveDestinationConcurrency({ isLocalProvider: true }), 1); + assert.equal(resolveDestinationConcurrency({ isLocalProvider: true, configuredConcurrency: 4 }), 1); + + const active = new Set(); + let peak = 0; + const starts = []; + const result = await mapDestinationsWithPolicy( + ["linkedin", "x", "blog", "newsletter"], + async (channel) => { + active.add(channel); + peak = Math.max(peak, active.size); + starts.push({ channel, active: [...active] }); + await new Promise((resolve) => setTimeout(resolve, 5)); + active.delete(channel); + return channel.toUpperCase(); + }, + { isLocalProvider: false }, + ); + + assert.ok(peak <= 2, `hosted peak concurrency was ${peak}`); + const firstLongForm = starts.findIndex((item) => ["blog", "newsletter"].includes(item.channel)); + assert.ok(firstLongForm >= 0); + assert.ok( + starts.slice(0, firstLongForm).every((item) => !["blog", "newsletter"].includes(item.channel)), + "long-form work should start only after the short-form phase completes", + ); + assert.deepEqual(result, ["LINKEDIN", "X", "BLOG", "NEWSLETTER"]); +}); + +test("local provider destination scheduling stays sequential", async () => { + let active = 0; + let peak = 0; + await mapDestinationsWithPolicy( + ["linkedin", "x", "blog"], + async () => { + active += 1; + peak = Math.max(peak, active); + await new Promise((resolve) => setTimeout(resolve, 3)); + active -= 1; + }, + { isLocalProvider: true, configuredConcurrency: 4 }, + ); + assert.equal(peak, 1); +}); + +test("orchestration uses the provider-aware lane policy for initial and duplicate repair work", async () => { + const source = await readFile(new URL("../lib/ai/generateStudioPackage.js", import.meta.url), "utf8"); + assert.match(source, /mapDestinationsWithPolicy\(channels/); + assert.match(source, /isLocalProvider: localProvider/); + assert.match(source, /mapDestinationsWithPolicy\(duplicateTargets/); + assert.match(source, /channelOf: \(target\) => target\.channel/); +});