Schedule generation by provider capacity and content length

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,15 +499,19 @@ 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,
       generationInputs,
       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.");

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;
+}

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/);
+});